From cbaa4fab0981b7e84a442227c7fc02906e038580 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Fri, 24 Jul 2026 11:15:46 +0200 Subject: [PATCH] Fix cross-hub patch ack acking on the wrong emission: write-identity gate + cold-store defer Residual acked-write-loss behind TwoSiloRecycleConvergenceTest (main run 30079395006, shard 2 - recurred AFTER the round-1 pending-save flush fix c43cc20ee; identical signature to run 30068597014: fast Success ack on the post-recycle write, store frozen at the pre-recycle version, WaitForPersistedBeyond timeout). Root cause (ApplyMeshNodePatchInTurn, MeshWeaver.Data/DataExtensions.cs): 1. The ack watcher was `stream.Skip(1).Take(1)` on the pathless per-node reduced stream - emission COUNTING, the exact pattern PR #584 banned ("write-echo = own stamped Version + target identity, never emission counting"). On a COLD activation (idle-recycle -> reactivate-on-write) the MeshNode init gate opens inside BuildInstanceCollection - on the storage-read emission, BEFORE the loaded collection commits to the primary stream - releasing the held PatchDataRequest into a window where the store is still empty. The first post-subscribe emissions are then the empty-store marker and the initial LOAD echo (the PRE-patch node): Skip(1).Take(1) took the load echo as "committed", IPostCommitFlush.Flush()ed the STALE state and posted PatchDataResponse SUCCESS, while the merge turn no-op'd AckOnce(false, NotFound) against the empty store (suppressed by the AckOnce Interlocked guard). A success-acked write that never existed anywhere - round 1's dispose-flush had nothing to save because nothing was ever committed in-RAM. 2. The merge turn insta-rejected NotFound on an empty store - wrong for a node that EXISTS in storage with its load in flight (the cold window above). Fix: - Write-identity ack: the merge lambda stamps (entityId, minted Version) Volatile/Interlocked BEFORE ApplyChanges; the ack watcher fires only on an emission that provably CONTAINS the write (ChangeContainsStampedWrite: same id, Version at-or-past the stamp). Load echoes, sibling-satellite churn, null drop-markers, and a concurrent writer's earlier commit can no longer ack this patch - restoring the read-after-write contract the flush-chained ack exists for. - Cold-store defer: on an empty collection the merge re-arms ONCE when the primary store first carries data (bounded 10s -> NotFound), so the in-flight initial load is awaited instead of raced. A genuinely absent node still NotFounds - now against the LOADED store. - Cold own-node resolution with satellites present (Count > 1): resolve the target by Path == hub.Address.Path instead of failing the Count==1 shortcut. Deterministic repro: PatchAckWriteIdentityTest scripts the exact interleaving (pre-stamp load echo -> late load echo -> satellite churn -> commit) through the production gate and pins the counterfactual: the old Skip(1).Take(1) shape selects the LOAD ECHO on that sequence. Scenario guard: ColdReactivationPatchReadYourWriteTest drives dispose -> immediate cross-hub Update -> read-your-write through the real pipeline. Verification (all on this worktree, real runs with restore): PatchAckWriteIdentityTest 6/6; ColdReactivationPatchReadYourWriteTest + DisposalPendingSaveFlushTest 2/2; TwoSiloRecycleConvergenceTest 1/1; CrossHubPatchAtomicityTest 1/1; MeshWeaver.Data.Test 287/287; VersionHistoryTest 5/5; WorkspaceCacheEvictionTest 5/5; MeshNodeVersionSyncTest 3/3; CI-exact `dotnet restore` + `dotnet build --no-restore -c Release -p:CIRun=true -warnaserror` clean (0 warnings, 0 errors). Note: the generic (non-MeshNode) deferred patch path retains Skip(1).Take(1); MeshNode never routes there when a primary stream exists, and non-MeshNode reduced streams have no cold load-echo shape - left unchanged deliberately. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012wP3c7DLHutUczcBWgs4h8 --- src/MeshWeaver.Data/DataExtensions.cs | 145 ++++++++++++++++-- .../PatchAckWriteIdentityTest.cs | 138 +++++++++++++++++ .../ColdReactivationPatchReadYourWriteTest.cs | 85 ++++++++++ 3 files changed, 359 insertions(+), 9 deletions(-) create mode 100644 test/MeshWeaver.Data.Test/PatchAckWriteIdentityTest.cs create mode 100644 test/MeshWeaver.Hosting.Monolith.Test/ColdReactivationPatchReadYourWriteTest.cs diff --git a/src/MeshWeaver.Data/DataExtensions.cs b/src/MeshWeaver.Data/DataExtensions.cs index 982116a2b6..464464256f 100644 --- a/src/MeshWeaver.Data/DataExtensions.cs +++ b/src/MeshWeaver.Data/DataExtensions.cs @@ -1250,14 +1250,46 @@ void AckOnce(bool success, MeshNodeError? error = null) hub.Post(resp, o => o.ResponseFor(request)); } - // OFF-TURN ack β€” IDENTICAL shape to the deferred path: wait for the reduced stream's - // post-commit emission (off the action-block turn), durably Flush (persist + publish the - // cache-eviction feed event), then ack. Subscribed BEFORE the write so the tick is never - // missed. A no-op/NotFound write never emits β†’ AckOnce already fired inside the lambda. + // 🚨 Write-identity stamps β€” set by the merge lambda AT COMMIT TIME (the same + // happens-before pattern as MeshNodeStreamHandle.UpdateOwn's echo detection): + // the ack watcher below fires ONLY for an emission that provably CONTAINS this + // patch (same entity id, Version at-or-past the minted stamp). Until the merge + // has stamped, nothing can ack success. + long stampedVersion = -1; + string? stampedId = null; + + // OFF-TURN ack β€” wait for a post-commit emission CONTAINING THIS WRITE (off the + // action-block turn), durably Flush (persist + publish the cache-eviction feed + // event), then ack. Subscribed BEFORE the write so the tick is never missed. + // A no-op/NotFound write never stamps β†’ the gate never opens β†’ AckOnce fires + // from the lambda's failure path instead. + // + // 🚨 NEVER `.Skip(1).Take(1)` here β€” that is emission-COUNTING, and on the + // pathless per-node reduced stream "the next emission" is not necessarily this + // write: on a COLD activation (idle-recycle β†’ reactivate-on-write) the init + // gate opens BEFORE the initial collection commits to the primary stream, so + // the first post-subscribe emission is the initial LOAD echo (the PRE-patch + // node), and on a busy owner it can be sibling-satellite churn. The old + // Skip(1).Take(1) took that load echo as "committed", Flush()ed the STALE + // pre-patch node and posted PatchDataResponse SUCCESS while the merge turn + // no-op'd NotFound against the still-empty store (suppressed by the AckOnce + // guard) β€” a success-acked write that never existed anywhere. That is the + // acked-write-loss behind TwoSiloRecycleConvergenceTest on main runs + // 30068597014 / 30079395006 (post-recycle store frozen at the pre-recycle + // version despite a fast Success ack). Write-echo detection is identity-based, + // never emission-count-based (PR #584 rule). var postSub = stream - .Skip(1) + .Where(c => ChangeContainsStampedWrite( + c.Value, + System.Threading.Volatile.Read(ref stampedId), + System.Threading.Interlocked.Read(ref stampedVersion), + idKey, versionKey, jsonOpts)) .Take(1) - .Timeout(TimeSpan.FromSeconds(5)) + // Bounded: covers the one-shot cold-store defer below (10s) + flush. On + // expiry the AckOnce guard means this NACK only fires if no terminal was + // posted yet (e.g. commit emission lost in a teardown) β€” the caller's + // retry machinery takes over; never a silent hang. + .Timeout(TimeSpan.FromSeconds(20)) .Subscribe( committed => { @@ -1287,8 +1319,9 @@ void AckOnce(bool success, MeshNodeError? error = null) } // ATOMIC read-merge-write on the primary action-block turn β€” a PURE, bounded transform. - // No ack here (the off-turn flush above acks on success), no IO, no nested subscribe. - primary.Update( + // No ack here (the off-turn identity-gated flush above acks on success), no IO, no + // nested subscribe. `deferred` guards the ONE-SHOT cold-store re-arm below. + void RunMergeTurn(bool deferred) => primary.Update( store => { try @@ -1307,8 +1340,56 @@ void AckOnce(bool success, MeshNodeError? error = null) entityId = only.Key?.ToString(); liveEntity = only.Value; } + // Cold with satellites already loaded (Count > 1): the Count==1 shortcut + // can't pick β€” resolve the OWN node by Path == hub.Address.Path (segments + // only; ToString() on hosted hubs appends "~" and never matches). + if (liveEntity is null && collection is { Instances.Count: > 1 }) + { + var ownPath = hub.Address.Path; + var pathKey = jsonOpts.PropertyNamingPolicy?.ConvertName("Path") ?? "Path"; + foreach (var kvp in collection.Instances) + { + var candidate = System.Text.Json.JsonSerializer + .SerializeToNode(kvp.Value, kvp.Value.GetType(), jsonOpts) + as System.Text.Json.Nodes.JsonObject; + if ((candidate?[pathKey] ?? candidate?["Path"] ?? candidate?["path"]) + is System.Text.Json.Nodes.JsonValue pv + && pv.TryGetValue(out var candidatePath) + && string.Equals(candidatePath, ownPath, StringComparison.OrdinalIgnoreCase)) + { + entityId = kvp.Key?.ToString(); + liveEntity = kvp.Value; + break; + } + } + } if (liveEntity is null || string.IsNullOrEmpty(entityId)) { + // 🚨 COLD-ACTIVATION WINDOW β€” defer, never insta-NotFound. The MeshNode + // init gate opens inside BuildInstanceCollection (on the storage-read + // emission), which RELEASES this held PatchDataRequest BEFORE the loaded + // collection has committed to the primary stream. The old code no-op'd + // AckOnce(false, NotFound) here against the not-yet-loaded store β€” for a + // node that EXISTS in storage β€” and raced the load echo into the old + // counting ack (see postSub comment). Re-arm ONCE when the store first + // carries data (the in-flight initial load), bounded; a genuinely absent + // node then NotFounds on the deferred attempt against the LOADED store. + if (!deferred && (collection is null || collection.Instances.Count == 0)) + { + var deferSub = primary + .Where(ci => ci.Value?.Collections.GetValueOrDefault(collectionName) + is { Instances.Count: > 0 }) + .Take(1) + .Timeout(TimeSpan.FromSeconds(10)) + .Subscribe( + _ => RunMergeTurn(deferred: true), + _ => AckOnce(false, new MeshNodeError( + MeshNodeErrorCode.NotFound, hubPath, + "Target MeshNode not found for patch apply " + + "(store did not initialize within the bound)"))); + hub.RegisterForDisposal(deferSub); + return null; // no write this turn β€” the deferred attempt commits + } AckOnce(false, new MeshNodeError( MeshNodeErrorCode.NotFound, hubPath, "Target MeshNode not found for patch apply")); @@ -1329,12 +1410,19 @@ as System.Text.Json.Nodes.JsonObject request.Message, jsonOpts, mergeGuardLogger, hubPath); // The OWNER mints the monotonic Version on apply (same rule as the deferred path). // 🚨 #325: forward-only from the node's own version β€” see NextMeshNodeVersion. - currentNode[versionKey] = NextMeshNodeVersion(currentNode, versionKey, version); + var minted = NextMeshNodeVersion(currentNode, versionKey, version); + currentNode[versionKey] = minted; var merged = System.Text.Json.JsonSerializer .Deserialize(currentNode.ToJsonString(jsonOpts), jsonOpts); if (merged is null) throw new System.Text.Json.JsonException("Merged value deserialised to null"); + // Stamp BEFORE the commit (id first, then the version that opens the + // gate) β€” the ack watcher only accepts an emission at-or-past this + // write, so a load echo / sibling emission can never ack it. + System.Threading.Volatile.Write(ref stampedId, entityId); + System.Threading.Interlocked.Exchange(ref stampedVersion, minted); + var newStore = s.Update(collectionName, c => c.Update(entityId, merged)); return primary.ApplyChanges(new EntityStoreAndUpdates( newStore, @@ -1348,6 +1436,45 @@ [new EntityUpdate(collectionName, entityId, merged) { OldValue = liveEntity }], } }, ex => AckOnce(false, ClassifyPatchException(ex, hubPath))); + + RunMergeTurn(deferred: false); + } + + /// + /// 🚨 Write-identity echo gate for the cross-hub MeshNode patch ack + /// (): true only for an emission that provably + /// CONTAINS the stamped write β€” same entity id AND Version at-or-past the version the + /// merge lambda minted. Emission-COUNTING (Skip(1).Take(1)) is forbidden here: + /// on a cold activation the reduced stream's first post-subscribe emission is the + /// initial LOAD echo (pre-patch state), and on a busy owner it can be sibling-satellite + /// churn β€” acking on either flushed stale state and reported success for a write that + /// never landed (TwoSiloRecycleConvergenceTest, runs 30068597014 / 30079395006). + /// Internal for the deterministic pin in MeshWeaver.Data.Test. + /// + internal static bool ChangeContainsStampedWrite( + object? value, + string? stampedId, + long stampedVersion, + string idKey, + string versionKey, + System.Text.Json.JsonSerializerOptions jsonOpts) + { + if (stampedVersion < 0 || string.IsNullOrEmpty(stampedId) || value is null) + return false; + var obj = System.Text.Json.JsonSerializer + .SerializeToNode(value, value.GetType(), jsonOpts) as System.Text.Json.Nodes.JsonObject; + if (obj is null) + return false; + var idNode = obj[idKey] ?? obj["Id"] ?? obj["id"]; + if (idNode is not System.Text.Json.Nodes.JsonValue idValue + || !idValue.TryGetValue(out var id) + || !string.Equals(id, stampedId, StringComparison.Ordinal)) + return false; + long ver = 0; + if (obj[versionKey] is System.Text.Json.Nodes.JsonValue verValue + && verValue.TryGetValue(out var parsed)) + ver = parsed; + return ver >= stampedVersion; } private static Type? WalkBaseForGeneric(Type type, Type genericDef) diff --git a/test/MeshWeaver.Data.Test/PatchAckWriteIdentityTest.cs b/test/MeshWeaver.Data.Test/PatchAckWriteIdentityTest.cs new file mode 100644 index 0000000000..4405684908 --- /dev/null +++ b/test/MeshWeaver.Data.Test/PatchAckWriteIdentityTest.cs @@ -0,0 +1,138 @@ +using System; +using System.Collections.Generic; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Text.Json; +using Xunit; + +namespace MeshWeaver.Data.Test; + +/// +/// 🚨 Deterministic pin of the cross-hub MeshNode patch-ack WRITE-IDENTITY gate +/// (DataExtensions.ChangeContainsStampedWrite, consumed by +/// ApplyMeshNodePatchInTurn's ack watcher) β€” the residual acked-write-loss +/// behind TwoSiloRecycleConvergenceTest on main runs 30068597014 and +/// 30079395006 (the second one AFTER the round-1 pending-save flush fix). +/// +/// The pinned interleaving. A post-recycle write reactivates the owner +/// per-node hub COLD. The MeshNode init gate opens inside +/// BuildInstanceCollection β€” on the storage-read emission, BEFORE the loaded +/// collection commits to the primary stream β€” releasing the held +/// PatchDataRequest into a window where the store is still empty. The old ack +/// watcher was stream.Skip(1).Take(1): emission COUNTING. The first +/// post-subscribe emission in that window is the initial LOAD echo β€” the PRE-patch +/// node β€” so the watcher flushed the stale state to storage and posted +/// PatchDataResponse SUCCESS, while the merge turn no-op'd NotFound against +/// the empty store (suppressed by the AckOnce guard). Result: a success-acked write +/// that never existed anywhere β€” the store stays frozen at the pre-recycle version +/// and the test's WaitForPersistedBeyond times out. +/// +/// These tests script that exact emission sequence and assert the production +/// identity gate (a) never fires before the merge stamps, (b) rejects the load echo +/// and sibling-satellite churn, and (c) fires exactly on the emission that contains +/// the stamped write β€” plus the counterfactual: the old counting shape selects the +/// load echo on the same sequence. Write-echo detection is identity-based, never +/// emission-count-based (PR #584 rule). +/// +public class PatchAckWriteIdentityTest +{ + private static readonly JsonSerializerOptions Options = new(); + + /// Local stand-in with the serialized shape the gate inspects + /// (PascalCase Id/Version like a MeshNode with default options). + private sealed record Node(string Id, long Version, string? ApiKey = null); + + private const string IdKey = "Id"; + private const string VersionKey = "Version"; + + private static readonly Node LoadEchoV12 = new("Anthropic", 12, "sk-v6"); + private static readonly Node SiblingSatellite = new("_Activity-1", 99); + private static readonly Node CommitV13 = new("Anthropic", 13, "sk-post-recycle"); + + private static bool Gate(object? value, string? stampedId, long stampedVersion) + => DataExtensions.ChangeContainsStampedWrite( + value, stampedId, stampedVersion, IdKey, VersionKey, Options); + + [Fact] + public void BeforeTheMergeStamps_NothingAcks() + { + // stampedVersion = -1 / stampedId = null until the merge lambda commits. + Gate(LoadEchoV12, null, -1).Should().BeFalse( + "no emission may ack a patch whose merge has not stamped yet β€” the cold " + + "activation's load echo arrives exactly in this window"); + Gate(CommitV13, null, -1).Should().BeFalse(); + } + + [Fact] + public void LoadEcho_AtPreWriteVersion_DoesNotAck() + { + Gate(LoadEchoV12, "Anthropic", 13).Should().BeFalse( + "the initial LOAD echo carries the PRE-patch state (Version 12 < stamped 13) β€” " + + "acking on it is the acked-write-loss of runs 30068597014 / 30079395006"); + } + + [Fact] + public void SiblingSatelliteChurn_DoesNotAck() + { + Gate(SiblingSatellite, "Anthropic", 13).Should().BeFalse( + "the pathless per-node reduced stream surfaces sibling satellites " + + "(Source/Release/_Activity) β€” a foreign id must never ack this write"); + } + + [Fact] + public void EmissionContainingTheStampedWrite_Acks() + { + Gate(CommitV13, "Anthropic", 13).Should().BeTrue( + "the commit emission carries the stamped id at the minted version"); + // A later state that already CONTAINS the write (version advanced past the + // stamp by a subsequent writer) also satisfies read-your-write. + Gate(new Node("Anthropic", 14, "sk-later"), "Anthropic", 13).Should().BeTrue(); + } + + [Fact] + public void ProductionGate_SelectsTheCommit_NeverTheLoadEcho() + { + // The full scripted interleaving through the production Where(...).Take(1) + // shape: replay (pre-subscribe current), load echo, sibling churn, commit. + var stamped = (Id: (string?)null, Version: -1L); + var source = new Subject(); + Node? acked = null; + using var sub = source + .Where(v => Gate(v, stamped.Id, stamped.Version)) + .Take(1) + .Subscribe(v => acked = v); + + source.OnNext(LoadEchoV12); // pre-stamp load echo β€” must not ack + stamped = ("Anthropic", 13); // merge lambda stamps at commit time + source.OnNext(LoadEchoV12); // late load echo β€” still pre-write state + source.OnNext(SiblingSatellite); // satellite churn β€” foreign id + acked.Should().BeNull("nothing so far contains the stamped write"); + + source.OnNext(CommitV13); // the commit β€” contains the write + acked.Should().Be(CommitV13); + } + + [Fact] + public void CounterfactualOldCountingShape_TakesTheLoadEcho() + { + // The OLD ack watcher was `stream.Skip(1).Take(1)` β€” emission counting. On + // the cold-activation sequence (replay, then load echo, then commit) it + // selects the LOAD ECHO as "the committed value": the flush then persisted + // the stale pre-patch node and acked SUCCESS for a write that never landed. + // This test documents the defective selection the identity gate exists to + // prevent; it goes RED if anyone reintroduces counting semantics into a + // sequence-equivalent shape. + var source = new Subject(); + Node? taken = null; + using var sub = source.Skip(1).Take(1).Subscribe(v => taken = v); + + source.OnNext(LoadEchoV12); // 1st: the replayed current (skipped) + source.OnNext(LoadEchoV12); // 2nd: the initial-load echo β€” counting takes THIS + source.OnNext(CommitV13); // the real commit arrives too late + + taken.Should().Be(LoadEchoV12, + "counting semantics mistake the cold-activation load echo for the commit β€” " + + "the exact acked-write-loss interleaving (do not 'fix' this assertion; it " + + "documents why the ack gate must be write-identity-based)"); + } +} diff --git a/test/MeshWeaver.Hosting.Monolith.Test/ColdReactivationPatchReadYourWriteTest.cs b/test/MeshWeaver.Hosting.Monolith.Test/ColdReactivationPatchReadYourWriteTest.cs new file mode 100644 index 0000000000..21dcedceea --- /dev/null +++ b/test/MeshWeaver.Hosting.Monolith.Test/ColdReactivationPatchReadYourWriteTest.cs @@ -0,0 +1,85 @@ +using System; +using System.Reactive.Linq; +using System.Reactive.Threading.Tasks; +using System.Threading.Tasks; +using MeshWeaver.Data; +using MeshWeaver.Hosting.Monolith.TestBase; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Services; +using MeshWeaver.Messaging; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace MeshWeaver.Hosting.Monolith.Test; + +/// +/// Scenario guard for the recycleβ†’reactivate-on-write flow behind +/// TwoSiloRecycleConvergenceTest (main runs 30068597014 / 30079395006): a +/// cross-hub stream.Update that lands on a COLD per-node hub (disposed β†’ +/// reactivated by the write itself) must give read-your-write β€” a successful Update +/// completion implies the write is applied and reaches durable storage; it must +/// never be silently dropped by the cold-store window (merge turn racing the initial +/// load) while the ack watcher mistakes the load echo for the commit. +/// +/// The deterministic pin of the ack-identity mechanism is +/// MeshWeaver.Data.Test.PatchAckWriteIdentityTest; this test drives the same +/// window through the REAL pipeline: dispose the per-node hub, then immediately +/// write through the cross-hub cache path (PatchDataRequest β†’ cold reactivation β†’ +/// init-gate release β†’ merge/load race) and require the store to advance. +/// +public class ColdReactivationPatchReadYourWriteTest(ITestOutputHelper output) : MonolithMeshTestBase(output) +{ + [Fact(Timeout = 60_000)] + public async Task UpdateAgainstDisposedHub_ReactivatesAndPersists() + { + var ct = TestContext.Current.CancellationToken; + var path = $"{TestPartition}/cold-patch-node"; + await NodeFactory.CreateNode( + new MeshNode("cold-patch-node", TestPartition) { Name = "initial", NodeType = "Markdown" }) + .Should().Emit(); + + // Warm the per-node hub and wait until the CREATE is durable. + await Mesh.Observe(new GetDataRequest(new MeshNodeReference()), o => o.WithTarget(new Address(path))) + .Should().Emit(); + var storage = Mesh.ServiceProvider.GetRequiredService(); + await Observable.Interval(TimeSpan.FromMilliseconds(50)).StartWith(0L) + .SelectMany(_ => storage.Read(path, Mesh.JsonSerializerOptions)) + .Where(n => n is not null) + .FirstAsync().Timeout(10.Seconds()).ToTask(ct); + + // Recycle: dispose the per-node hub β€” the next write must reactivate it COLD. + var nodeHub = Mesh.GetHostedHub(new Address(path), HostedHubCreation.Never); + nodeHub.Should().NotBeNull(); + nodeHub!.Dispose(); + Output.WriteLine($"[recycle] disposed per-node hub for {path}"); + + // Post-recycle write through the cross-hub cache path (PatchDataRequest to a + // cold hub β€” the init gate releases the patch into the merge/load race). + // Resilient on the reactive tick, mirroring the TwoSilo test: the very first + // attempt can race the disposing hub's ShuttingDown NACK window. + var marker = $"post-recycle-{Guid.NewGuid():N}"[..24]; + var workspace = Mesh.GetWorkspace(); + await Observable.Interval(TimeSpan.FromMilliseconds(500)).StartWith(0L) + .SelectMany(_ => workspace.GetMeshNodeStream(path) + .Update(n => n with { Name = marker }) + .Take(1) + .Select(_ => true) + .Catch(_ => Observable.Return(false))) + .Where(ok => ok) + .FirstAsync().Timeout(30.Seconds()).ToTask(ct); + Output.WriteLine($"[write] update acked with marker {marker}"); + + // Read-your-write: a successful cross-hub Update against the reactivated hub + // must reach durable storage β€” the store may NEVER stay frozen at the + // pre-recycle state (the acked-write-loss signature: WaitForPersistedBeyond + // timing out at the pre-recycle version). + var persisted = await Observable.Interval(TimeSpan.FromMilliseconds(50)).StartWith(0L) + .SelectMany(_ => storage.Read(path, Mesh.JsonSerializerOptions)) + .Where(n => n is not null && n.Name == marker) + .FirstAsync().Timeout(15.Seconds()).ToTask(ct); + persisted!.Name.Should().Be(marker, + "a write acknowledged against a cold-reactivated per-node hub must be applied " + + "and durable β€” never dropped by the merge/load race with the ack watcher " + + "taking the load echo for the commit (runs 30068597014 / 30079395006)"); + } +}