From 2b72baefb973affc1662db7137ffa6acd1e37196 Mon Sep 17 00:00:00 2001 From: Shubham Ranjan Date: Sat, 4 Jul 2026 23:36:48 +0530 Subject: [PATCH] fix(watch): resume watch from last observed revision on reconnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Watch_ShouldRecover_AfterServerRestart was flaky (nightly beast: 5-20% failure) — Assert.NotEmpty failed because no events arrived after a server restart. Root cause: on a broken stream, WatchManager.HandleConnectionFailure re-registered each watch with the ORIGINAL request, whose StartRevision is 0 ('watch from now'), and never tracked the last observed revision. So after a reconnect the watch only saw events created strictly after re-registration completed; anything written during the reconnect gap was permanently lost. This is a real data-loss bug (same family as issue #283), not just a test-timing issue — etcd watches are revision-based and the official clientv3 resumes from the last observed revision (nextRev). Fix (mirrors clientv3): - Track NextRevision per watch: after an event batch, lastEvent.ModRevision + 1; for a created/progress notification, header.Revision + 1; on a compaction cancel, the CompactRevision. Monotonic. - On reconnect, set CreateRequest.StartRevision = NextRevision so gap events are replayed. Tests: - Unit (deterministic, no docker): reconnect resumes from lastRev+1, from a progress notification's header+1, and from CompactRevision after a compaction cancel. - Integration: WatchResilienceTests now polls etcd readiness instead of a fixed 15s sleep. Verified: the restart test run 30x locally -> 30/30 green (was 5-20% flaky). --- .../Integration/WatchResilienceTests.cs | 31 +++-- .../Unit/WatchManagerCoverageTests.cs | 115 ++++++++++++++++++ dotnet-etcd/watchclient/WatchManager.cs | 60 +++++++++ 3 files changed, 197 insertions(+), 9 deletions(-) diff --git a/dotnet-etcd.Tests/Integration/WatchResilienceTests.cs b/dotnet-etcd.Tests/Integration/WatchResilienceTests.cs index 1924942..6443f1e 100644 --- a/dotnet-etcd.Tests/Integration/WatchResilienceTests.cs +++ b/dotnet-etcd.Tests/Integration/WatchResilienceTests.cs @@ -117,19 +117,32 @@ public async Task Watch_ShouldRecover_AfterServerRestart() Console.WriteLine($"Restarting {ContainerName}..."); RunDockerCommand($"restart {ContainerName}"); - // 4. Wait for restart to complete - Console.WriteLine("Waiting for restart..."); - await Task.Delay(15000); - - // 6. Put post-reconnect, retrying until an event is received or timeout. - // The watch reconnects with StartRevision=0 (watch from now), so if the put races - // ahead of the re-registration the event is permanently missed. Retrying every 2s - // ensures at least one put lands after the watch has been re-established. + // 4. Poll until etcd is serving again instead of a fixed sleep (restart time varies, + // especially on loaded CI runners). + Console.WriteLine("Waiting for etcd to become ready..."); + var readySw = Stopwatch.StartNew(); + while (readySw.Elapsed.TotalSeconds < 60) + { + try + { + await _client.GetAsync("watch-resilience-health-probe"); + break; // etcd responded + } + catch + { + await Task.Delay(1000); + } + } + + // 5. Put post-reconnect, retrying until an event is received or timeout. + // The watch resumes from the last observed revision + 1, so even if a put races ahead + // of the watch re-registration the event is replayed rather than missed. The retry + // loop also absorbs the brief window where etcd is up but the stream is mid-reconnect. Console.WriteLine("Putting value after reconnect (retrying until event received)..."); var sw = Stopwatch.StartNew(); while (sw.Elapsed.TotalSeconds < 30 && events.Count == 0) { - try + try { await _client.PutAsync(testKey, "recovered-restart"); } diff --git a/dotnet-etcd.Tests/Unit/WatchManagerCoverageTests.cs b/dotnet-etcd.Tests/Unit/WatchManagerCoverageTests.cs index 956e102..0cd245e 100644 --- a/dotnet-etcd.Tests/Unit/WatchManagerCoverageTests.cs +++ b/dotnet-etcd.Tests/Unit/WatchManagerCoverageTests.cs @@ -69,6 +69,121 @@ public void Constructor_WithNullFactory_ThrowsArgumentNullException() Assert.Throws(() => new WatchManager(null!)); } + // --------------------------------------------------------------------- + // Reconnect / revision resume (issue: events lost across a reconnect) + // --------------------------------------------------------------------- + + private static WatchResponse EventAtRevision(long watchId, long revision, string key = "k", string value = "v") + { + WatchResponse response = EventResponse(watchId, key, value); + response.Header = new ResponseHeader { Revision = revision }; + response.Events[0].Kv.ModRevision = revision; + return response; + } + + [Fact] + public void Reconnect_ResumesWatchFromLastObservedRevisionPlusOne() + { + var fakes = new Queue>(); + var fake1 = new FakeDuplexStreamingCall(); + var fake2 = new FakeDuplexStreamingCall(); + fakes.Enqueue(fake1); + fakes.Enqueue(fake2); + // Return the next fake for the initial stream and the reconnect; keep returning fake2 after. + using var manager = new WatchManager((_, _, _) => fakes.Count > 1 ? fakes.Dequeue() : fakes.Peek()); + + long observedRevision = 0; + manager.Watch(KeyRequest("k"), (WatchResponse r) => + { + if (r.Header != null && r.Header.Revision > 0) + { + observedRevision = r.Header.Revision; + } + }); + + // Server delivers an event at revision 10; the manager must remember it. + fake1.Enqueue(EventAtRevision(FirstWatchId, 10)); + Assert.True(WaitFor(() => observedRevision == 10), "initial event was not delivered"); + + // Break the stream -> HandleConnectionFailure reconnects on fake2. + fake1.Responses.ThrowOnMoveNext = new RpcException(new Status(StatusCode.Unavailable, "connection dropped")); + fake1.Enqueue(new WatchResponse()); // unblock the receive loop so it re-enters MoveNext and throws + + // The watch must be re-registered on the new stream resuming from revision 11 (10 + 1), + // otherwise events written during the reconnect gap are permanently lost. + Assert.True( + WaitFor(() => fake2.Requests.Written.Any(r => r.CreateRequest != null), 5000), + "watch was not re-registered on the reconnected stream"); + + WatchRequest reRegistered = fake2.Requests.Written.First(r => r.CreateRequest != null); + Assert.Equal(11, reRegistered.CreateRequest.StartRevision); + } + + [Fact] + public void Reconnect_AfterProgressNotification_ResumesFromHeaderRevisionPlusOne() + { + var fakes = new Queue>(); + var fake1 = new FakeDuplexStreamingCall(); + var fake2 = new FakeDuplexStreamingCall(); + fakes.Enqueue(fake1); + fakes.Enqueue(fake2); + using var manager = new WatchManager((_, _, _) => fakes.Count > 1 ? fakes.Dequeue() : fakes.Peek()); + + long observedRevision = 0; + manager.Watch(KeyRequest("k"), (WatchResponse r) => + { + if (r.Header != null && r.Header.Revision > observedRevision) + { + observedRevision = r.Header.Revision; + } + }); + + // A progress notification carries the current revision but no events. + fake1.Enqueue(new WatchResponse { WatchId = FirstWatchId, Header = new ResponseHeader { Revision = 20 } }); + Assert.True(WaitFor(() => observedRevision == 20), "progress notification was not delivered"); + + fake1.Responses.ThrowOnMoveNext = new RpcException(new Status(StatusCode.Unavailable, "connection dropped")); + fake1.Enqueue(new WatchResponse()); + + Assert.True( + WaitFor(() => fake2.Requests.Written.Any(r => r.CreateRequest != null), 5000), + "watch was not re-registered on the reconnected stream"); + Assert.Equal(21, fake2.Requests.Written.First(r => r.CreateRequest != null).CreateRequest.StartRevision); + } + + [Fact] + public void Reconnect_AfterCompactionCancel_ResumesFromCompactRevision() + { + var fakes = new Queue>(); + var fake1 = new FakeDuplexStreamingCall(); + var fake2 = new FakeDuplexStreamingCall(); + fakes.Enqueue(fake1); + fakes.Enqueue(fake2); + using var manager = new WatchManager((_, _, _) => fakes.Count > 1 ? fakes.Dequeue() : fakes.Peek()); + + bool canceledSeen = false; + manager.Watch(KeyRequest("k"), (WatchResponse r) => + { + if (r.Canceled) + { + canceledSeen = true; + } + }); + + // The server cancels the watch because our resume point was compacted away. + fake1.Enqueue(new WatchResponse { WatchId = FirstWatchId, Canceled = true, CompactRevision = 50 }); + Assert.True(WaitFor(() => canceledSeen), "cancel response was not delivered"); + + fake1.Responses.ThrowOnMoveNext = new RpcException(new Status(StatusCode.Unavailable, "connection dropped")); + fake1.Enqueue(new WatchResponse()); + + Assert.True( + WaitFor(() => fake2.Requests.Written.Any(r => r.CreateRequest != null), 5000), + "watch was not re-registered on the reconnected stream"); + // Cannot resume from before the compaction revision; resume from CompactRevision. + Assert.Equal(50, fake2.Requests.Written.First(r => r.CreateRequest != null).CreateRequest.StartRevision); + } + // --------------------------------------------------------------------- // WatchAsync(WatchRequest, Action) - core path // --------------------------------------------------------------------- diff --git a/dotnet-etcd/watchclient/WatchManager.cs b/dotnet-etcd/watchclient/WatchManager.cs index fd602a4..96ebb5b 100644 --- a/dotnet-etcd/watchclient/WatchManager.cs +++ b/dotnet-etcd/watchclient/WatchManager.cs @@ -99,10 +99,55 @@ void WrappedCallback(WatchResponse response) _watchIdMapping[response.WatchId] = watchId; } + // Track the revision to resume from on reconnect so no events are missed in the gap. + TrackResumeRevision(watchId, response); + callback(response); } } + /// + /// Advances the stored resume revision for a watch based on a received response, mirroring the + /// etcd clientv3 "nextRev" logic: after an event batch resume from lastEvent.ModRevision + 1; + /// for a created/progress notification with no events, advance to the header revision. A + /// compaction cancel resets the resume point to the compact revision. Only ever moves forward. + /// + private void TrackResumeRevision(long clientWatchId, WatchResponse response) + { + if (!_watches.TryGetValue(clientWatchId, out WatchCancellation? watch)) + { + return; + } + + long candidate = watch.NextRevision; + + if (response.Canceled && response.CompactRevision > 0) + { + // Our resume point was compacted away; the earliest we can resume from is CompactRevision. + candidate = response.CompactRevision; + } + else if (response.Events != null && response.Events.Count > 0) + { + long lastModRevision = response.Events[^1].Kv.ModRevision; + if (lastModRevision + 1 > candidate) + { + candidate = lastModRevision + 1; + } + } + else if (response.Header != null && response.Header.Revision > 0) + { + // Created response or progress notification (no events): everything up to the header + // revision has been observed, so the next start revision is header.Revision + 1. + long boundary = response.Header.Revision + 1; + if (boundary > candidate) + { + candidate = boundary; + } + } + + watch.NextRevision = candidate; + } + /// /// Creates a new watch request /// @@ -674,6 +719,14 @@ private void HandleConnectionFailure() foreach (var watch in _watches.Values) { + // Resume from the revision after the last observed event so events written while + // the stream was down are replayed instead of lost. Falls back to "from now" + // (StartRevision 0) only when nothing has been observed yet. + if (watch.NextRevision > 0 && watch.Request.CreateRequest != null) + { + watch.Request.CreateRequest.StartRevision = watch.NextRevision; + } + await _watchStream!.CreateWatchAsync(watch.Request, watch.Callback).ConfigureAwait(false); } } @@ -692,5 +745,12 @@ private class WatchCancellation public required CancellationTokenSource CancellationTokenSource { get; set; } public required WatchRequest Request { get; set; } public required Action Callback { get; set; } + + /// + /// The revision to resume this watch from if the stream is re-established. Tracks the + /// revision after the last event/notification observed, so a reconnect does not miss + /// events written during the gap (mirrors the etcd clientv3 nextRev behavior). + /// + public long NextRevision { get; set; } } }