diff --git a/src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs b/src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs index 366f252b0e..f9111ded64 100644 --- a/src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs +++ b/src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs @@ -433,10 +433,47 @@ void Resubscribe(string reason) sub.Dispose(); return; } - // HeartBeatEvent is [SystemMessage] — PostPipeline accepts - // null AccessContext without warning. No identity stamp - // needed; receiver doesn't gate on principal. - h.Post(new HeartBeatEvent(), o => o.WithTarget(owner)); + // HeartBeatEvent is [SystemMessage] — PostPipeline accepts null AccessContext + // without warning. No identity stamp needed; receiver doesn't gate on principal. + // + // 🚨 OBSERVE the delivery — do NOT fire-and-forget. The heartbeat has no ack, so a + // HEALTHY owner never responds and the Observe simply times out (the normal case, + // ignored below). But a terminal NotFound DeliveryFailure means the owner ADDRESS + // NO LONGER EXISTS: a recycled/deactivated grain REACTIVATES on the heartbeat post + // and acks, so only a PERMANENTLY-GONE owner NotFounds — e.g. a one-shot + // {Partition}/_Activity/import-{fingerprint} lock whose dedicated off-router import + // hub was disposed when the import completed (the node persists as history, so the + // change feed never fires a Created/Deleted pulse to drive resubscribe). When the + // owner is gone we must STOP: tear the keep-alive down. Fire-and-forget left that + // dead owner heart-beaten every interval forever → "[ROUTE] NotFound" per partition + // for the life of the silo — the recurring import-activity NotFound storm that pegs + // the CPU and trips the liveness probe. (Pinned by + // HeartbeatStopsWhenOwnerDiesAfterSubscribeTest.) + // Post the heartbeat (still keeps the grain alive) and OBSERVE its delivery. The + // post is async-queued and the Observe registers synchronously right after, before + // any response can land — no race. A healthy owner has no ack, so the Observe just + // times out (the normal case, ignored). A terminal NotFound means the owner ADDRESS + // is gone — only a permanently-gone owner NotFounds (a recycled grain reactivates on + // the post and acks) — so STOP: tear the keep-alive down. + var hbDelivery = h.Post(new HeartBeatEvent(), o => o.WithTarget(owner)); + if (hbDelivery is not null) + h.Observe(hbDelivery) + .Take(1) + // A healthy owner sends no heartbeat ack, so the observe never emits — switch + // to Empty (COMPLETES) instead of the throwing Timeout overload: the normal + // case must not route through the error path every interval (exceptions-as- + // control-flow is a real CPU/alloc cost given how many heartbeats run). A + // terminal NotFound still arrives FAST as an OnError before this fires. + .Timeout(heartbeatInterval, Observable.Empty()) + .Subscribe( + _ => { }, // no ack to consume + ex => + { + // Owner address gone (a recycled grain reactivates on the post and + // acks — only a permanently-gone owner NotFounds): STOP heart-beating. + if (ex is DeliveryFailureException { Failure.ErrorType: ErrorType.NotFound }) + keepAlive.Dispose(); + }); }); // On keepAlive (not directly on reduced): a terminal NotFound from the initial // SubscribeRequest disposes keepAlive → stops this heartbeat. Normal teardown diff --git a/src/MeshWeaver.Data/Serialization/SyncStreamOptions.cs b/src/MeshWeaver.Data/Serialization/SyncStreamOptions.cs index f8ef13aef4..0db6d70569 100644 --- a/src/MeshWeaver.Data/Serialization/SyncStreamOptions.cs +++ b/src/MeshWeaver.Data/Serialization/SyncStreamOptions.cs @@ -6,11 +6,14 @@ namespace MeshWeaver.Data.Serialization; public class SyncStreamOptions { /// - /// Interval between HeartBeatEvents posted to the owner hub. Doubles as the resubscribe - /// detection window: when the owner is gone (recycled / idle / crashed), the next - /// heartbeat returns DeliveryFailure and the subscriber re-issues SubscribeRequest to - /// pick up a fresh snapshot from the new grain. Default: 45 seconds. - /// Tests use a much shorter interval to verify the resubscribe path without waiting. + /// Interval between HeartBeatEvents posted to the owner hub to keep its grain alive. The + /// heartbeat also detects a PERMANENTLY-gone owner: a recycled/idle grain REACTIVATES on the + /// heartbeat post and acks, so only a gone owner (e.g. a one-shot _Activity/import-* lock + /// whose dedicated import hub was disposed) returns a terminal NotFound — which tears the + /// keep-alive down so we never heartbeat a dead owner in an endless NotFound loop. Recycled-grain + /// RE-SUBSCRIPTION (picking up a fresh snapshot after a restart) is driven by the mesh change feed, + /// not the heartbeat. Default: 45 seconds. Tests use a much shorter interval to exercise the + /// teardown/resubscribe paths without waiting. /// public TimeSpan HeartbeatInterval { get; set; } = TimeSpan.FromSeconds(45); diff --git a/test/MeshWeaver.Data.Test/HeartbeatStopsWhenOwnerDiesAfterSubscribeTest.cs b/test/MeshWeaver.Data.Test/HeartbeatStopsWhenOwnerDiesAfterSubscribeTest.cs new file mode 100644 index 0000000000..8880b1af53 --- /dev/null +++ b/test/MeshWeaver.Data.Test/HeartbeatStopsWhenOwnerDiesAfterSubscribeTest.cs @@ -0,0 +1,108 @@ +using System; +using System.Linq; +using System.Reactive.Linq; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.Data.Serialization; +using MeshWeaver.Fixture; +using MeshWeaver.Messaging; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace MeshWeaver.Data.Test; + +/// +/// Pins the OTHER half of "we must not heart-beat a dead owner forever" — the half that the +/// recurring {Partition}/_Activity/import-{fingerprint} storm slips through. +/// covers the owner that NEVER existed (the +/// initial SubscribeRequest NotFounds → keep-alive torn down). This covers the owner that +/// existed at subscribe time and then died — exactly a one-shot import-activity lock: the +/// importer writes it via GetMeshNodeStream(activityPath).Update (a cached, heart-beated sync +/// stream), the dedicated off-router import hub is disposed when the import completes, and the node +/// persists as history so there is no Created/Deleted change-feed event to trigger the +/// change-feed resubscribe. +/// +/// Once heartbeats became fire-and-forget — with the change feed as the SOLE recycled-grain +/// detector — a subscription whose owner dies without a change-feed pulse keeps posting +/// to the missing owner every interval forever → [ROUTE] NotFound +/// per partition, per interval, for the life of the silo. The fix: the heartbeat observes its +/// delivery and, on a terminal NotFound , tears the keep-alive down +/// (after a resubscribe attempt for a genuinely recycled grain). +/// +/// The test makes it deterministic: the owner serves a real type so the subscribe SUCCEEDS and +/// the heartbeat runs; then a flag flips the owner "dead" so every further heartbeat comes back +/// NotFound — with NO change-feed event. After death the heartbeat count must FREEZE. Without the fix +/// it climbs one-per-interval — a deterministic RED. +/// +public class HeartbeatStopsWhenOwnerDiesAfterSubscribeTest(ITestOutputHelper output) : HubTestBase(output) +{ + // Far shorter than the 2s observation window, so a still-alive heartbeat would tick ~10×. + private static readonly TimeSpan ShortHeartbeat = TimeSpan.FromMilliseconds(200); + + private int _heartbeatCount; + private volatile bool _ownerDead; + + /// Stand-in for the import-activity content; keyed by . + public record GhostData(string Id); + + protected override MessageHubConfiguration ConfigureHost(MessageHubConfiguration configuration) + => base.ConfigureHost(configuration) + // Count every heartbeat. Once "dead", reply NotFound to it — the shape the router + // returns when the owner grain/hub is gone — WITHOUT emitting any change-feed event, + // so only the heartbeat-delivery path can tear the keep-alive down. + .WithHandler((hub, delivery) => + { + Interlocked.Increment(ref _heartbeatCount); + if (_ownerDead) + hub.Post( + new DeliveryFailure(delivery) { ErrorType = ErrorType.NotFound, Message = "owner gone" }, + o => o.ResponseFor(delivery)); + return delivery.Processed(); + }) + // A REAL data source so the initial SubscribeRequest SUCCEEDS and the heartbeat starts. + .AddData(data => data.AddSource(src => src + .WithType(t => t.WithInitialData( + _ => Observable.Return(new[] { new GhostData("import-fingerprint") }.AsEnumerable()))))); + + protected override MessageHubConfiguration ConfigureClient(MessageHubConfiguration configuration) + => base.ConfigureClient(configuration) + .WithServices(services => + services.Configure(o => o.HeartbeatInterval = ShortHeartbeat)) + .AddData(data => data.AddHubSource(CreateHostAddress(), ds => ds.WithType())); + + [HubFact] + public async Task OwnerDiesAfterSubscribe_NoChangeFeedEvent_HeartbeatTearsDown() + { + GetHost(); // activate the owner hub + var client = GetClient(); + var workspace = client.ServiceProvider.GetRequiredService(); + + // 1. Subscribe SUCCEEDS — the Initial arrives, so the keep-alive heartbeat is now ticking + // against a LIVE owner (distinct from the never-existed case). + await workspace.GetObservable("import-fingerprint") + .Where(d => d is not null) + .Should().Within(15.Seconds()) + .Match(d => d!.Id == "import-fingerprint"); + + // 2. Confirm the heartbeat is genuinely flowing to the live owner before we kill it. + await Observable.Interval(50.Milliseconds()) + .Select(_ => Volatile.Read(ref _heartbeatCount)) + .Should().Within(10.Seconds()) + .Match(c => c >= 2); + + // 3. The owner DIES — every further heartbeat NotFounds, with NO change-feed event (the + // _Activity/import-* lock persists as history). Only the heartbeat-delivery path can save us. + var countAtDeath = Volatile.Read(ref _heartbeatCount); + _ownerDead = true; + + // 4. Sanctioned fixed wait ("confirm nothing happens"): several heartbeat intervals later the + // count must have FROZEN — the terminal NotFound tore the keep-alive down. Without the fix + // the zombie timer keeps posting and this climbs ~one-per-interval. + Thread.Sleep(TimeSpan.FromSeconds(2)); + var countAfter = Volatile.Read(ref _heartbeatCount); + + (countAfter - countAtDeath).Should().BeLessThanOrEqualTo(1, + "once a heartbeat comes back NotFound the keep-alive must tear down — a one-shot " + + "{Partition}/_Activity/import-* lock must never be heart-beaten in an endless NotFound loop"); + } +}