dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(comm): queue failover re-seed behind in-flight fan-out; stamp stream generation on gRPC errors (plan R2-02 T11)

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-13 10:18:16 -04:00
parent bb8be55382
commit c29e588905
2 changed files with 92 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs
+48 -4
View File
@@ -99,6 +99,21 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
/// <summary>Ordered buffer of live deltas that arrived while a fan-out was in flight. Actor-thread only.</summary>
private readonly List<AlarmStateChanged> _buffer = new();
/// <summary>
/// A failover re-seed was requested while a fan-out was already in flight; it must run
/// right after the in-flight one completes rather than being silently dropped (N7.1) —
/// the in-flight snapshot's read-time predates the stream death. Actor-thread only.
/// </summary>
private bool _reseedQueued;
/// <summary>
/// Monotonic stream generation stamped on each opened gRPC stream and echoed back on its
/// error callback: a late error raced out of a previous (cancelled) stream carries a stale
/// generation and is ignored so it never burns retry budget or double-flips (N7.2).
/// Actor-thread only.
/// </summary>
private int _streamGeneration;
private const int BufferWarnThreshold = 10_000;
private bool _bufferWarned;
@@ -182,6 +197,15 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
// gRPC stream error — flip node + reconnect + re-seed.
Receive<GrpcAlarmStreamError>(msg =>
{
// Ignore a late error raced out of a previous (cancelled) stream — the
// RpcException(Cancelled) filter at SiteStreamGrpcClient.cs covers the normal
// path, but a genuine socket fault can beat the cancel (N7.2).
if (msg.Generation != _streamGeneration)
{
_log.Debug("Ignoring stale gRPC error from stream generation {0} (current {1})",
msg.Generation, _streamGeneration);
return;
}
_log.Warning("Site-alarm gRPC stream error for {0}: {1}", _siteIdentifier, msg.Exception.Message);
HandleGrpcError();
});
@@ -266,7 +290,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
if (_stopped) return;
if (_fanoutInFlight)
{
// Already seeding/reconciling — don't stack a second fan-out.
// A failover re-seed requested mid-fan-out must run right after the in-flight
// one — its snapshot read-time predates the stream death, so skipping it would
// serve stale up to the next 60s reconcile (N7.1). An initial-seed collision
// never queues (there is only ever one).
if (!isInitial) _reseedQueued = true;
return;
}
@@ -324,6 +352,13 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
Timers.Cancel(PublishTimerKey);
_publishPending = false;
Publish();
// A failover re-seed requested while this fan-out was in flight runs now (N7.1).
if (_reseedQueued)
{
_reseedQueued = false;
StartFanout(isInitial: false);
}
}
private void OnSeedFailed(SeedFailed msg)
@@ -347,6 +382,13 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
_publishPending = false;
Publish();
}
// A failover re-seed requested while this fan-out was in flight runs now (N7.1).
if (_reseedQueued)
{
_reseedQueued = false;
StartFanout(isInitial: false);
}
}
/// <summary>
@@ -459,6 +501,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
Timers.StartSingleTimer(StabilityTimerKey, new GrpcAlarmStreamStable(), StabilityWindow);
var generation = ++_streamGeneration;
var client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint);
var self = Self;
var ct = _grpcCts.Token;
@@ -468,7 +511,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
await client.SubscribeSiteAsync(
_correlationId,
alarm => self.Tell(alarm),
ex => self.Tell(new GrpcAlarmStreamError(ex)),
ex => self.Tell(new GrpcAlarmStreamError(ex, generation)),
ct);
}, ct);
}
@@ -564,8 +607,9 @@ internal sealed record RunReconcile;
/// <summary>Internal: coalesced-publish tick — flush the dirty cache to viewers once (N6).</summary>
internal sealed record PublishCoalesced;
/// <summary>Internal: site-alarm gRPC stream error occurred.</summary>
internal sealed record GrpcAlarmStreamError(Exception Exception);
/// <summary>Internal: site-alarm gRPC stream error occurred, stamped with the stream
/// generation it came from so a late error from a cancelled stream can be ignored (N7.2).</summary>
internal sealed record GrpcAlarmStreamError(Exception Exception, int Generation);
/// <summary>Internal: reconnect the site-alarm gRPC stream (flip node).</summary>
internal sealed record ReconnectAlarmStream;
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs
+44
View File
@@ -397,6 +397,50 @@ public class SiteAlarmAggregatorActorTests : TestKit
AwaitCondition(() => TotalSubs() > before, TimeSpan.FromSeconds(3));
}
// ── R2 T11: queued re-seed after reconnect + stream-generation stamp (N7.1/N7.2) ──
[Fact]
public void ReseedRequestedDuringInFlightFanout_IsQueued_NotDropped()
{
var (actor, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.Zero);
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
seed.CompleteNext(); // initial seed done (CallCount 1)
AwaitAssert(() => Assert.Equal(1, sink.Count));
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
actor.Tell(new RunReconcile()); // reconcile fan-out now in flight (CallCount 2)
AwaitAssert(() => Assert.Equal(2, seed.CallCount));
// Stream error while the reconcile is in flight → the failover re-seed must not be
// silently skipped. Pre-fix: StartFanout no-ops and CallCount stays 2 until the next
// 60s reconcile tick.
factory.ClientFor(GrpcNodeA).Subs.Last().OnError(new Exception("stream fault"));
seed.CompleteNext(); // finish the in-flight reconcile
AwaitAssert(() => Assert.Equal(3, seed.CallCount)); // queued re-seed ran immediately after
}
[Fact]
public void LateErrorFromAPreviousStreamGeneration_IsIgnored()
{
var (_, seed, _, factory) = CreateActor(publishCoalesce: TimeSpan.Zero);
AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
seed.CompleteNext();
AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count;
var firstSub = factory.ClientFor(GrpcNodeA).Subs.Single();
firstSub.OnError(new Exception("real fault")); // gen 1 error → flip, reconnect
AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); // gen 2 stream open
AwaitAssert(() => Assert.Equal(2, TotalSubs()));
firstSub.OnError(new Exception("late zombie fault")); // stale gen-1 error races in
// Pre-fix this burns retry budget and opens a THIRD stream; post-fix it is ignored.
Thread.Sleep(400);
Assert.Equal(2, TotalSubs());
}
// ── R2 T10: live-delta publish coalescing (N6) ──
[Fact]