fix(watch): resume from last observed revision on reconnect (fixes nightly beast flake)

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");
                 }

dotnet-etcd.Tests/Unit/WatchManagerCoverageTests.cs

@@ -69,6 +69,121 @@ public void Constructor_WithNullFactory_ThrowsArgumentNullException()
         Assert.Throws<ArgumentNullException>(() => 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<FakeDuplexStreamingCall<WatchRequest, WatchResponse>>();
+        var fake1 = new FakeDuplexStreamingCall<WatchRequest, WatchResponse>();
+        var fake2 = new FakeDuplexStreamingCall<WatchRequest, WatchResponse>();
+        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<FakeDuplexStreamingCall<WatchRequest, WatchResponse>>();
+        var fake1 = new FakeDuplexStreamingCall<WatchRequest, WatchResponse>();
+        var fake2 = new FakeDuplexStreamingCall<WatchRequest, WatchResponse>();
+        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<FakeDuplexStreamingCall<WatchRequest, WatchResponse>>();
+        var fake1 = new FakeDuplexStreamingCall<WatchRequest, WatchResponse>();
+        var fake2 = new FakeDuplexStreamingCall<WatchRequest, WatchResponse>();
+        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<WatchResponse>) - core path
     // ---------------------------------------------------------------------

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);
         }
     }
 
+    /// <summary>
+    ///     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.
+    /// </summary>
+    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;
+    }
+
     /// <summary>
     ///     Creates a new watch request
     /// </summary>
@@ -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<WatchResponse> Callback { get; set; }
+
+        /// <summary>
+        ///     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).
+        /// </summary>
+        public long NextRevision { get; set; }
     }
 }