diff --git a/dotnet-etcd.Tests/Integration/WatchResilienceTests.cs b/dotnet-etcd.Tests/Integration/WatchResilienceTests.cs index bbc78e9..6b2da09 100644 --- a/dotnet-etcd.Tests/Integration/WatchResilienceTests.cs +++ b/dotnet-etcd.Tests/Integration/WatchResilienceTests.cs @@ -1,105 +1,141 @@ using System; +using System.Collections.Concurrent; +using System.Collections.Generic; using System.Diagnostics; +using System.Linq; using System.Threading.Tasks; using Xunit; using dotnet_etcd; -using Etcdserverpb; -using System.Collections.Generic; namespace dotnet_etcd.Tests.Integration { // Use the isolated collection for resilience tests to prevent parallel execution interference - [Collection("WatchResilienceTests")] + [Collection("WatchResilienceTests")] [Trait("Category", "Integration")] public class WatchResilienceTests : IDisposable { private readonly EtcdClient _client; private readonly string _testKeyPrefix = "watch-resilience-"; - + // Dedicated single-node etcd (port 2409) so pausing/restarting the server here never // disrupts the shared etcd1 cluster used by the other integration tests. private const string EtcdUrl = "http://localhost:2409"; private const string ContainerName = "etcd-resilience"; - public WatchResilienceTests() - { - _client = new EtcdClient(EtcdUrl); - } + // Watch delivery is asynchronous and its latency is unbounded in practice: on a loaded CI + // runner the receive loop can be starved for far longer than any sleep we would care to + // hardcode. Every wait below therefore polls up to this budget and returns as soon as the + // condition holds, so the test is fast when things are fast and only slow when it must be. + private static readonly TimeSpan EventTimeout = TimeSpan.FromSeconds(20); + + // Diagnostics captured from the watch stream. If this test ever fails again, the assertion + // message carries the Created ack and the revisions, which is what distinguishes "the event + // was never delivered" from "the watch was never registered". + private readonly List _trace = []; + private readonly Stopwatch _sw = Stopwatch.StartNew(); + + public WatchResilienceTests() => _client = new EtcdClient(EtcdUrl); public void Dispose() { _client.Dispose(); - // Ensure container is unpaused in case test failed mid-way + // Ensure container is unpaused in case the test failed mid-way. RunDockerCommand($"unpause {ContainerName}"); - // Ensure container is running (if restart failed or left it in weird state) - // But 'unpause' handles pause. 'restart' handles restart. - // If restart was pending, it should be fine. } [Fact] public async Task Watch_ShouldRecover_AfterNetworkPause() { - var testKey = $"{_testKeyPrefix}pause"; - var events = new List(); - - // 1. Start Watch - _client.Watch(testKey, (response) => - { - foreach (var evt in response.Events) - { - events.Add(new WatchEvent - { - Type = evt.Type, - Key = evt.Kv.Key.ToStringUtf8(), - Value = evt.Kv.Value.ToStringUtf8() - }); - } - }); + string testKey = $"{_testKeyPrefix}pause"; + ConcurrentQueue events = StartWatch(testKey); - // 2. Initial Put - await _client.PutAsync(testKey, "initial"); - await Task.Delay(1000); - Assert.NotEmpty(events); + // 1. Establish the watch by observing the first event. + long putRev = (await _client.PutAsync(testKey, "initial")).Header.Revision; + Trace($"PUT initial rev={putRev}"); + await WaitForEvents(events, 1, "initial watch establishment"); events.Clear(); - // 3. Simulate Network Disconnect (Pause Container) - Console.WriteLine($"Pausing {ContainerName}..."); + // 2. Simulate a network partition by pausing the container. + Trace($"pausing {ContainerName}"); RunDockerCommand($"pause {ContainerName}"); - // 4. Wait for > 10s (simulating standard timeout/keepalive expiry) - Console.WriteLine("Waiting 12s..."); + // Stay paused past the keepalive timeout so the client really sees the connection die. await Task.Delay(12000); - // 5. Resume Network - Console.WriteLine($"Unpausing {ContainerName}..."); + Trace($"unpausing {ContainerName}"); RunDockerCommand($"unpause {ContainerName}"); - - // Allow client to attempt reconnection - await Task.Delay(3000); - - // 6. Put post-disconnect - Console.WriteLine("Putting value after reconnect..."); - await _client.PutAsync(testKey, "recovered-pause"); - - // 7. Verification - await Task.Delay(2000); - - Assert.Single(events); - Assert.Equal("recovered-pause", events[0].Value); + + // 3. Write through the recovered connection. The put is retried because the client may + // still be re-establishing the stream, and the watch resumes from the revision after + // the last one it saw, so the event is delivered even if it lands mid-reconnect. + await PutUntilSucceeds(testKey, "recovered-pause"); + + // 4. The watch must deliver the post-partition event. + await WaitForEvents(events, 1, "event after network pause"); + Assert.Equal("recovered-pause", events.First().Value); } [Fact] public async Task Watch_ShouldRecover_AfterServerRestart() { - var testKey = $"{_testKeyPrefix}restart"; - var events = new List(); + string testKey = $"{_testKeyPrefix}restart"; + ConcurrentQueue events = StartWatch(testKey); + + // 1. Establish the watch by observing the first event. + long putRev = (await _client.PutAsync(testKey, "initial")).Header.Revision; + Trace($"PUT initial rev={putRev}"); + await WaitForEvents(events, 1, "initial watch establishment"); + events.Clear(); + + // 2. Restart the server, which forces the stream to drop. + Trace($"restarting {ContainerName}"); + RunDockerCommand($"restart {ContainerName}"); + + // 3. Wait for etcd to serve again (restart time varies wildly on loaded CI runners). + await WaitUntil(async () => + { + try + { + await _client.GetAsync("watch-resilience-health-probe"); + return true; + } + catch + { + return false; + } + }, TimeSpan.FromSeconds(60), "etcd to become ready after restart"); + + // 4. Write through the recovered connection and require the watch to deliver it. + await PutUntilSucceeds(testKey, "recovered-restart"); + await WaitForEvents(events, 1, "event after server restart"); + Assert.Equal("recovered-restart", events.First().Value); + } + + /// + /// Registers the watch and returns the (thread-safe) collection its events land in. The + /// callback runs on the gRPC receive loop, so the collection must not be a plain List that + /// the test thread reads concurrently. + /// + private ConcurrentQueue StartWatch(string testKey) + { + ConcurrentQueue events = new(); - // 1. Start Watch - _client.Watch(testKey, (response) => + _client.Watch(testKey, response => { - foreach (var evt in response.Events) + if (response.Created) { - events.Add(new WatchEvent + Trace($"CREATED watchId={response.WatchId} rev={response.Header?.Revision}"); + } + + if (response.Events == null) + { + return; + } + + foreach (Mvccpb.Event evt in response.Events) + { + Trace($"EVENT {evt.Type} rev={evt.Kv.ModRevision} value={evt.Kv.Value.ToStringUtf8()}"); + events.Enqueue(new WatchEvent { Type = evt.Type, Key = evt.Kv.Key.ToStringUtf8(), @@ -108,60 +144,82 @@ public async Task Watch_ShouldRecover_AfterServerRestart() } }); - // 2. Initial Put - await _client.PutAsync(testKey, "initial"); - await Task.Delay(1000); - Assert.NotEmpty(events); - events.Clear(); - - // 3. Simulate Server Restart (Forces Connection Drop) - Console.WriteLine($"Restarting {ContainerName}..."); - RunDockerCommand($"restart {ContainerName}"); + Trace("Watch() returned"); + return events; + } - // 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) + /// + /// Puts until the write is accepted — etcd may still be coming back up — so that a + /// transient write failure is not misreported as the watch losing the event. + /// + private async Task PutUntilSucceeds(string key, string value) + { + Exception? last = null; + await WaitUntil(async () => { try { - await _client.GetAsync("watch-resilience-health-probe"); - break; // etcd responded + long rev = (await _client.PutAsync(key, value)).Header.Revision; + Trace($"PUT {value} rev={rev}"); + return true; } - catch + catch (Exception ex) { - await Task.Delay(1000); + last = ex; + return false; } - } + }, TimeSpan.FromSeconds(30), $"put '{value}' to succeed (last error: {last?.Message})"); + } + + private async Task WaitForEvents(ConcurrentQueue events, int expected, string what) => + await WaitUntil(() => Task.FromResult(events.Count >= expected), EventTimeout, + $"{what}: expected >= {expected} watch event(s), got {events.Count}"); - // 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) + /// + /// Polls until the condition holds, failing with the captured watch-stream trace so a + /// timeout says why it timed out rather than just "collection was empty". + /// + private async Task WaitUntil(Func> condition, TimeSpan timeout, string what) + { + Stopwatch sw = Stopwatch.StartNew(); + while (sw.Elapsed < timeout) { - try + if (await condition()) { - await _client.PutAsync(testKey, "recovered-restart"); + return; } - catch - { - // etcd may still be starting up; keep retrying - } - await Task.Delay(2000); + + await Task.Delay(200); + } + + if (await condition()) + { + return; + } + + Assert.Fail($"Timed out after {timeout.TotalSeconds:0}s waiting for {what}.\nWatch stream trace:\n " + + string.Join("\n ", GetTrace())); + } + + private void Trace(string message) + { + lock (_trace) + { + _trace.Add($"[{_sw.ElapsedMilliseconds,6}ms] {message}"); } + } - // 7. Verification - Assert.NotEmpty(events); - Assert.Equal("recovered-restart", events[0].Value); + private string[] GetTrace() + { + lock (_trace) + { + return _trace.Count == 0 ? ["(no watch responses received at all)"] : [.. _trace]; + } } private void RunDockerCommand(string args) { - var psi = new ProcessStartInfo + ProcessStartInfo psi = new() { FileName = "docker", Arguments = args, @@ -171,17 +229,18 @@ private void RunDockerCommand(string args) CreateNoWindow = true }; - using var process = Process.Start(psi); + using Process? process = Process.Start(psi); if (process == null) { - Console.WriteLine("Failed to start docker process"); - return; + Console.WriteLine("Failed to start docker process"); + return; } + process.WaitForExit(); if (process.ExitCode != 0) { - var error = process.StandardError.ReadToEnd(); - // Don't throw for unpause in Dispose if already unpaused + string error = process.StandardError.ReadToEnd(); + // Don't complain when Dispose unpauses a container that was never paused. if (!args.Contains("unpause") || !error.Contains("is not paused")) { Console.WriteLine($"Docker command failed: {error}"); diff --git a/dotnet-etcd.Tests/Unit/Mocks/FakeDuplexStreamingCall.cs b/dotnet-etcd.Tests/Unit/Mocks/FakeDuplexStreamingCall.cs index 3cfb57b..3b62cec 100644 --- a/dotnet-etcd.Tests/Unit/Mocks/FakeDuplexStreamingCall.cs +++ b/dotnet-etcd.Tests/Unit/Mocks/FakeDuplexStreamingCall.cs @@ -1,6 +1,7 @@ using System.Collections.Concurrent; using System.Threading.Channels; using dotnet_etcd.interfaces; +using Etcdserverpb; using Grpc.Core; namespace dotnet_etcd.Tests.Unit.Mocks; @@ -19,11 +20,20 @@ public class RecordingClientStreamWriter : IClientStreamWriter /// True once has been called. public bool IsCompleted { get; private set; } + /// + /// Invoked synchronously from inside , before it returns. Lets a + /// test model a server that responds while the write is still in flight — the window in which + /// a real etcd can deliver the Created response before the client has finished registering the + /// watch. + /// + public Action? OnWrite { get; set; } + public WriteOptions? WriteOptions { get; set; } public Task WriteAsync(T message) { _written.Enqueue(message); + OnWrite?.Invoke(message); return Task.CompletedTask; } @@ -31,6 +41,7 @@ public Task WriteAsync(T message, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); _written.Enqueue(message); + OnWrite?.Invoke(message); return Task.CompletedTask; } @@ -86,15 +97,43 @@ public async Task MoveNext(CancellationToken cancellationToken) /// public class FakeDuplexStreamingCall : IAsyncDuplexStreamingCall { + private long _autoAckRevision; + + public FakeDuplexStreamingCall() => Requests.OnWrite = AutoAck; + public RecordingClientStreamWriter Requests { get; } = new(); public ChannelStreamReader Responses { get; } = new(); + /// + /// Mirrors a real etcd, which answers every WatchCreateRequest with a Created response echoing + /// the client-supplied watch id. On by default so that a watch stream behaves like a server; + /// turn it off to drive the acknowledgement (or withhold it) by hand. + /// + public bool AutoAckWatchCreate { get; set; } = true; + /// True once has been called. public bool IsDisposed { get; private set; } public IClientStreamWriter RequestStream => Requests; public IAsyncStreamReader ResponseStream => Responses; + private void AutoAck(TRequest request) + { + if (!AutoAckWatchCreate || + request is not WatchRequest { CreateRequest: not null } watchRequest || + this is not FakeDuplexStreamingCall watchStream) + { + return; + } + + watchStream.Enqueue(new WatchResponse + { + Created = true, + WatchId = watchRequest.CreateRequest.WatchId, + Header = new ResponseHeader { Revision = Interlocked.Increment(ref _autoAckRevision) } + }); + } + public Task GetHeadersAsync() => Task.FromResult(new Metadata()); public Status GetStatus() => Status.DefaultSuccess; diff --git a/dotnet-etcd.Tests/Unit/WatchManagerCoverageTests.cs b/dotnet-etcd.Tests/Unit/WatchManagerCoverageTests.cs index 0cd245e..7aab320 100644 --- a/dotnet-etcd.Tests/Unit/WatchManagerCoverageTests.cs +++ b/dotnet-etcd.Tests/Unit/WatchManagerCoverageTests.cs @@ -81,6 +81,239 @@ private static WatchResponse EventAtRevision(long watchId, long revision, string return response; } + [Fact] + public void Watch_CalledFromInsideAWatchCallback_DoesNotDeadlockTheStream() + { + using WatchManager manager = CreateManager(out FakeDuplexStreamingCall fake); + + long nested = 0; + using ManualResetEventSlim done = new(false); + + manager.Watch(KeyRequest("a"), (WatchResponse r) => + { + if (r.Events.Count == 0 || Interlocked.Read(ref nested) != 0) + { + return; + } + + // Reacting to an event by starting another watch is an ordinary thing to do. It must not + // wedge the stream: the callback runs on the receive loop, and a blocking watch create can + // only be acknowledged BY that loop. + Interlocked.Exchange(ref nested, manager.Watch(KeyRequest("b"), (WatchResponse _) => { })); + done.Set(); + }); + + fake.Enqueue(EventResponse(FirstWatchId, "a", "v")); + + Assert.True(done.Wait(15000), + "a watch created from inside a watch callback deadlocked the receive loop"); + Assert.NotEqual(0, Interlocked.Read(ref nested)); + } + + [Fact] + public void Reconnect_WatchWithExplicitStartRevision_ResumesFromItRatherThanTheAckHeader() + { + var fakes = new Queue>(); + var fake1 = new FakeDuplexStreamingCall { AutoAckWatchCreate = false }; + var fake2 = new FakeDuplexStreamingCall { AutoAckWatchCreate = false }; + fakes.Enqueue(fake1); + fakes.Enqueue(fake2); + using var manager = new WatchManager((_, _, _) => fakes.Count > 1 ? fakes.Dequeue() : fakes.Peek()); + + // The "resume from a checkpoint" pattern: the caller asks to replay everything since rev 11. + WatchRequest request = KeyRequest("k"); + request.CreateRequest.StartRevision = 11; + + Task watch = manager.WatchAsync(request, (WatchResponse _) => { }); + + // etcd acks at the CURRENT cluster revision (99) and will now replay 11..99. + fake1.Enqueue(new WatchResponse + { + Created = true, + WatchId = FirstWatchId, + Header = new ResponseHeader { Revision = 99 } + }); + Assert.True(WaitFor(() => watch.IsCompleted, 5000), "watch was never acknowledged"); + + // Stream dies before any of the backlog is replayed. The caller asked for 11 and has seen + // nothing, so the resume must still be 11 — not 100, which would skip the whole backlog. + fake1.Responses.ThrowOnMoveNext = new RpcException(new Status(StatusCode.Unavailable, "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(11, fake2.Requests.Written.First(r => r.CreateRequest != null).CreateRequest.StartRevision); + } + + [Fact] + public void Reconnect_ReplayCreateAck_DoesNotSkipPastEventsStillToBeReplayed() + { + var fakes = new Queue>(); + var fake1 = new FakeDuplexStreamingCall(); + // The reconnect stream acks by hand so we can give the ack a header revision far ahead of the + // watch's resume point, which is what a real etcd does: the ack carries the CURRENT cluster + // revision, not the watch's position. + var fake2 = new FakeDuplexStreamingCall { AutoAckWatchCreate = false }; + var fake3 = new FakeDuplexStreamingCall { AutoAckWatchCreate = false }; + fakes.Enqueue(fake1); + fakes.Enqueue(fake2); + fakes.Enqueue(fake3); + using var manager = new WatchManager((_, _, _) => fakes.Count > 1 ? fakes.Dequeue() : fakes.Peek()); + + long observed = 0; + manager.Watch(KeyRequest("k"), (WatchResponse r) => + { + if (r.Events.Count > 0) + { + observed = r.Events[^1].Kv.ModRevision; + } + }); + + fake1.Enqueue(EventAtRevision(FirstWatchId, 10)); + Assert.True(WaitFor(() => observed == 10), "initial event was not delivered"); + + // First disconnect: the watch must resume from 11. + fake1.Responses.ThrowOnMoveNext = new RpcException(new Status(StatusCode.Unavailable, "dropped")); + fake1.Enqueue(new WatchResponse()); + Assert.True(WaitFor(() => fake2.Requests.Written.Any(r => r.CreateRequest != null), 5000), + "watch was not re-registered after the first disconnect"); + Assert.Equal(11, fake2.Requests.Written.First(r => r.CreateRequest != null).CreateRequest.StartRevision); + + // etcd acks the replay create at the current cluster revision (99) and will now replay 11..99. + // Nothing has been replayed yet. + fake2.Enqueue(new WatchResponse + { + Created = true, + WatchId = FirstWatchId, + Header = new ResponseHeader { Revision = 99 } + }); + Assert.True(WaitFor(() => fake2.Requests.Written.Any()), "create was not written to the new stream"); + Thread.Sleep(200); // let the ack be processed + + // Second disconnect before any replayed event arrives. The watch still has not seen 11..99, so + // it must STILL resume from 11. Treating the ack's header as "observed" would resume at 100 and + // silently skip every one of those events. + fake2.Responses.ThrowOnMoveNext = new RpcException(new Status(StatusCode.Unavailable, "dropped")); + fake2.Enqueue(new WatchResponse()); + Assert.True(WaitFor(() => fake3.Requests.Written.Any(r => r.CreateRequest != null), 5000), + "watch was not re-registered after the second disconnect"); + Assert.Equal(11, fake3.Requests.Written.First(r => r.CreateRequest != null).CreateRequest.StartRevision); + } + + [Fact] + public async Task WatchAsync_DoesNotComplete_UntilServerAcknowledgesTheWatch() + { + var fake = new FakeDuplexStreamingCall { AutoAckWatchCreate = false }; + using var manager = new WatchManager((_, _, _) => fake); + + Task watch = manager.WatchAsync(KeyRequest("k"), (WatchResponse _) => { }); + + // etcd registers a watch asynchronously: until the Created ack comes back the watch does not + // exist server-side. If WatchAsync completed here, a caller doing the natural + // `await WatchAsync(...); await PutAsync(...);` could have its put applied before the watch + // existed, and the event would never be delivered — silently, and forever. + await Task.Delay(300); + Assert.False(watch.IsCompleted, + "WatchAsync completed before etcd acknowledged the watch; a write issued now could be missed"); + + fake.Enqueue(new WatchResponse + { + Created = true, + WatchId = FirstWatchId, + Header = new ResponseHeader { Revision = 7 } + }); + + Assert.Equal(FirstWatchId, await watch); + } + + [Fact] + public async Task WatchAsync_WhenStreamDiesBeforeAck_RetriesCreateAndCompletesOnceAcknowledged() + { + var fakes = new Queue>(); + var fake1 = new FakeDuplexStreamingCall { AutoAckWatchCreate = false }; + var fake2 = new FakeDuplexStreamingCall { AutoAckWatchCreate = false }; + fakes.Enqueue(fake1); + fakes.Enqueue(fake2); + using var manager = new WatchManager((_, _, _) => fakes.Count > 1 ? fakes.Dequeue() : fakes.Peek()); + + Task watch = manager.WatchAsync(KeyRequest("k"), (WatchResponse _) => { }); + + // The stream dies before the server ever acknowledged the watch. This is what happens when a + // watch is opened against an etcd that is still coming back up. + fake1.Responses.ThrowOnMoveNext = new RpcException(new Status(StatusCode.Unavailable, "connection dropped")); + fake1.Enqueue(new WatchResponse()); + + // The create must be re-sent on the new stream rather than the watch being silently dropped. + Assert.True( + WaitFor(() => fake2.Requests.Written.Any(r => r.CreateRequest != null), 5000), + "watch create was not retried on the reconnected stream"); + Assert.False(watch.IsCompleted, "WatchAsync completed even though the watch was never acknowledged"); + + fake2.Enqueue(new WatchResponse + { + Created = true, + WatchId = FirstWatchId, + Header = new ResponseHeader { Revision = 9 } + }); + + Assert.Equal(FirstWatchId, await watch); + } + + [Fact] + public void Reconnect_ResumesFromCreatedRevision_WhenCreatedArrivesBeforeRegistrationCompletes() + { + 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()); + + using var createdDelivered = new ManualResetEventSlim(false); + + // Model the real server: the Created response comes back while the create request write is + // still in flight. Holding WriteAsync open until the callback has run guarantees the response + // is dispatched before the manager finishes registering the watch — the exact window in which + // the created revision used to be dropped. + fake1.Requests.OnWrite = request => + { + if (request.CreateRequest == null) + { + return; + } + + fake1.Enqueue(new WatchResponse + { + Created = true, + WatchId = request.CreateRequest.WatchId, + Header = new ResponseHeader { Revision = 42 } + }); + + createdDelivered.Wait(5000); + }; + + manager.Watch(KeyRequest("k"), (WatchResponse r) => + { + if (r.Created) + { + createdDelivered.Set(); + } + }); + + // No events are ever delivered — only the Created ack. Break the stream. + 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 + + Assert.True( + WaitFor(() => fake2.Requests.Written.Any(r => r.CreateRequest != null), 5000), + "watch was not re-registered on the reconnected stream"); + + // The watch observed everything up to revision 42, so it must resume at 43. Resuming at 0 + // ("from now") would silently drop every event written during the outage. + WatchRequest reRegistered = fake2.Requests.Written.First(r => r.CreateRequest != null); + Assert.Equal(43, reRegistered.CreateRequest.StartRevision); + } + [Fact] public void Reconnect_ResumesWatchFromLastObservedRevisionPlusOne() { diff --git a/dotnet-etcd.Tests/beast.sh b/dotnet-etcd.Tests/beast.sh index af6c822..bfb60ae 100755 --- a/dotnet-etcd.Tests/beast.sh +++ b/dotnet-etcd.Tests/beast.sh @@ -68,7 +68,13 @@ failed_iterations=() for i in $(seq 1 "$ITERATIONS"); do log="$LOG_DIR/run-$i.log" - if dotnet test "$PROJECT" -c Debug --no-build "${FILTER_ARG[@]}" >"$log" 2>&1; then + # verbosity=detailed so the tests' own Console output lands in the log. Without it a failing + # iteration gives only the assertion message, which is not enough to diagnose a flake after the + # fact from a CI artifact. + # ${arr[@]+"${arr[@]}"} because `set -u` treats an empty array as unbound on bash 3.2 (macOS), + # which is exactly the -f All case. + if dotnet test "$PROJECT" -c Debug --no-build ${FILTER_ARG[@]+"${FILTER_ARG[@]}"} \ + --logger "console;verbosity=detailed" >"$log" 2>&1; then passes=$((passes + 1)) echo " [$i/$ITERATIONS] PASS" else diff --git a/dotnet-etcd.Tests/start-etcd.sh b/dotnet-etcd.Tests/start-etcd.sh index c5e3f52..8140dad 100755 --- a/dotnet-etcd.Tests/start-etcd.sh +++ b/dotnet-etcd.Tests/start-etcd.sh @@ -5,6 +5,7 @@ set -euo pipefail # - etcd1 / etcd2 / etcd3 : 3-node cluster (ports 2379 / 22379 / 32379) # - etcd-ssl : TLS-enabled node (port 2389) # - etcd-authttl : short auth-token-ttl node for token-renewal tests (port 2399) +# - etcd-resilience : dedicated node the watch resilience tests pause/restart (port 2409) cd "$(dirname "$0")" # The fixture reads cluster-type.txt to pick its connection string. The cluster is reachable on @@ -40,9 +41,13 @@ wait_for_health() { return 1 } -# The 3-node cluster (most integration tests) and the short-TTL auth node (token-renewal test). +# The 3-node cluster (most integration tests), the short-TTL auth node (token-renewal test) and the +# resilience node. etcd-resilience must be gated too: WatchResilienceTests opens a watch against it +# as its very first action, so if it is still starting the watch is registered against a server that +# is not serving yet and the test fails on an empty event list. wait_for_health etcd1 --endpoints=http://etcd1:2379,http://etcd2:2379,http://etcd3:2379 wait_for_health etcd-authttl +wait_for_health etcd-resilience echo "$CLUSTER_TYPE" > cluster-type.txt @@ -52,3 +57,4 @@ echo " - http://localhost:22379 (etcd2)" echo " - http://localhost:32379 (etcd3)" echo " - https://localhost:2389 (etcd-ssl)" echo " - http://localhost:2399 (etcd-authttl, auth-token-ttl=2s)" +echo " - http://localhost:2409 (etcd-resilience, paused/restarted by watch resilience tests)" diff --git a/dotnet-etcd/watchclient/WatchManager.cs b/dotnet-etcd/watchclient/WatchManager.cs index 96ebb5b..1d3b3c9 100644 --- a/dotnet-etcd/watchclient/WatchManager.cs +++ b/dotnet-etcd/watchclient/WatchManager.cs @@ -18,10 +18,29 @@ namespace dotnet_etcd; /// public class WatchManager : IWatchManager { + /// + /// How long to wait for etcd to acknowledge a watch before giving up. Generous, because the + /// stream may have to reconnect (e.g. the server is restarting) before the create is accepted. + /// + private static readonly TimeSpan CreateWatchTimeout = TimeSpan.FromSeconds(30); + private readonly object _lockObject = new(); private readonly ConcurrentDictionary _watches = new(); private readonly ConcurrentDictionary _watchIdMapping = new(); + /// Watches whose create request is still awaiting the server's Created acknowledgement. + private readonly ConcurrentDictionary> _pendingCreates = new(); + + /// + /// Tail of the callback chain. Every response is appended to this single chain so user + /// callbacks run off the receive loop but remain serialized and ordered, as they were when the + /// loop invoked them inline. Note a slow callback queues responses without bound — callbacks + /// must not block indefinitely. + /// + private readonly object _dispatchLock = new(); + + private Task _dispatchChain = Task.CompletedTask; + private readonly Func> _watchStreamFactory; @@ -62,31 +81,103 @@ public async Task WatchAsync(WatchRequest request, Action c // Create a cancellation token source that can be used to cancel the watch CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + // Work on our own copy: the watch id is stamped onto the request and the reconnect path + // rewrites StartRevision on it, and neither should mutate the caller's object (which callers + // may reuse across watches). + request = request.Clone(); + request.CreateRequest.WatchId = watchId; + // Create a watch cancellation object - WatchCancellation watchCancellation = new() - { - WatchId = watchId, + WatchCancellation watchCancellation = new() + { + WatchId = watchId, CancellationTokenSource = cts, Request = request, - Callback = WrappedCallback + Callback = WrappedCallback, + + // Seed the resume point with the revision the caller asked to start from (etcd clientv3 + // does the same: `nextRev := w.initReq.rev`). Without this, a watch created with an + // explicit StartRevision would look identical to a "from now" watch, and the Created ack — + // which carries the *current* cluster revision — would advance the resume point past the + // backlog the caller asked to replay. + NextRevision = request.CreateRequest.StartRevision }; - request.CreateRequest.WatchId = watchId; - // Create the watch - await _watchStream!.CreateWatchAsync(request, WrappedCallback).ConfigureAwait(false); + // Completed when etcd acknowledges the watch. Continuations MUST run asynchronously: the + // callback below is invoked from the stream's receive loop, so resuming the awaiting caller + // inline would run its code on that loop and stall event delivery for every watch on the + // stream (and deadlock outright if the caller then blocks). + TaskCompletionSource acknowledged = new(TaskCreationOptions.RunContinuationsAsynchronously); - // Add the watch cancellation to the dictionary + // Register the watch BEFORE writing the create request: the server can answer while WriteAsync + // is still in flight, and both TrackResumeRevision and the reconnect loop ignore watches that + // are not in _watches yet. _watches[watchId] = watchCancellation; + _pendingCreates[watchId] = acknowledged; + + try + { + await _watchStream!.CreateWatchAsync(request, WrappedCallback).ConfigureAwait(false); + + // Wait until etcd has actually registered the watch. Creating a watch is asynchronous on + // the server, so until the Created ack arrives the watch does not exist: a caller that + // wrote a key as soon as Watch() returned could have the write applied first and never + // see the event. If the stream dies while we wait, HandleConnectionFailure re-sends the + // create on the new stream and its ack completes this same task. + TimeSpan ackTimeout = AckTimeoutFor(deadline); + + WatchResponse ack = await acknowledged.Task + .WaitAsync(ackTimeout, cts.Token) + .ConfigureAwait(false); + + if (ack.Canceled) + { + throw new RpcException(new Status(StatusCode.FailedPrecondition, + $"etcd rejected the watch: {ack.CancelReason}")); + } + } + catch (Exception ex) + { + // The watch was never established. Cancel first so any response that shows up late is + // ignored, then drop the entry so the reconnect loop won't try to re-register a watch the + // caller believes failed. + _watches.TryRemove(watchId, out _); + SafeCancel(cts); + cts.Dispose(); + + // The server may nonetheless have registered the watch (e.g. we timed out waiting for an + // ack that was merely slow). Best-effort tear it down rather than leak a server-side + // watcher that streams into a callback nobody listens to. + if (_watchStream != null) + { + _ = _watchStream.CancelWatchAsync(watchId).ContinueWith( + t => _ = t.Exception, TaskScheduler.Default); + } - // Since we don't get a server watch ID from CreateWatchAsync, we can't map it - // The server will assign a watch ID and include it in the watch response - // Our wrappedCallback will handle this mapping when it receives the response + throw ex is TimeoutException + ? new RpcException(new Status(StatusCode.DeadlineExceeded, + $"etcd did not acknowledge the watch within {AckTimeoutFor(deadline).TotalSeconds:0}s")) + : ex; + } + finally + { + _pendingCreates.TryRemove(watchId, out _); + } return watchId; // Create a wrapper callback that checks if the watch has been canceled void WrappedCallback(WatchResponse response) { + // Complete the create before the cancellation guard below: a watch that is cancelled or + // disposed while its create is still in flight must still release the awaiting caller + // rather than leave it blocked until the timeout. + if ((response.Created || response.Canceled) && + _pendingCreates.TryGetValue(watchId, out TaskCompletionSource? pending)) + { + pending.TrySetResult(response); + } + if (cts.IsCancellationRequested) { return; @@ -102,7 +193,82 @@ void WrappedCallback(WatchResponse response) // Track the revision to resume from on reconnect so no events are missed in the gap. TrackResumeRevision(watchId, response); - callback(response); + Dispatch(watchId, response, callback, cts); + } + } + + /// + /// Hands a response to the user's callback off the stream's receive loop. + /// + /// The receive loop invokes callbacks inline, so running user code on it would let one slow + /// or blocking callback stall event delivery for every watch on the stream — and a callback + /// that starts another watch would deadlock outright, because the loop it is blocking is the + /// only thing that could deliver the new watch's acknowledgement. + /// + /// + /// Responses are appended to a single chain, so callbacks stay serialized and in order + /// exactly as they were when they ran on the receive loop. That matters: overloads such as + /// WatchRange(string[] paths, method) hand the SAME delegate to several watches, and those + /// callers are entitled to assume it is never entered concurrently. + /// + /// + private void Dispatch(long watchId, WatchResponse response, Action callback, + CancellationTokenSource cts) + { + lock (_dispatchLock) + { + _dispatchChain = _dispatchChain.ContinueWith(_ => + { + if (cts.IsCancellationRequested) + { + return; + } + + try + { + callback(response); + } + catch (Exception ex) + { + // A throwing user callback must not fault the chain and silently stop delivery of + // every subsequent event for this watch. + Console.Error.WriteLine($"Watch callback for watch {watchId} threw: {ex}"); + } + }, CancellationToken.None, TaskContinuationOptions.None, TaskScheduler.Default); + } + } + + /// + /// How long to wait for the create acknowledgement. Honours the caller's deadline when one is + /// given, so a caller that asked for a short deadline is not held for the full default. + /// + private static TimeSpan AckTimeoutFor(DateTime? deadline) + { + if (deadline == null) + { + return CreateWatchTimeout; + } + + TimeSpan remaining = deadline.Value.ToUniversalTime() - DateTime.UtcNow; + return remaining < TimeSpan.Zero ? TimeSpan.Zero + : remaining < CreateWatchTimeout ? remaining + : CreateWatchTimeout; + } + + /// + /// Cancels a token source that another owner may already have disposed. Cancel() throws on a + /// disposed source (Dispose() itself is idempotent), and both the create-failure path and + /// CancelWatch/Dispose can reach the same source. + /// + private static void SafeCancel(CancellationTokenSource cts) + { + try + { + cts.Cancel(); + } + catch (ObjectDisposedException) + { + // Already torn down by the other owner; nothing to do. } } @@ -126,6 +292,18 @@ private void TrackResumeRevision(long clientWatchId, WatchResponse response) // Our resume point was compacted away; the earliest we can resume from is CompactRevision. candidate = response.CompactRevision; } + else if (response.Created) + { + // A Created ack carries the CURRENT cluster revision, which says nothing about what this + // watch has observed. Only use it to seed a brand new "watch from now" watch: for a watch + // being re-registered after a reconnect (StartRevision > 0) the server is about to replay + // the backlog from that revision, and advancing to the ack's header would skip straight + // past it — losing exactly the events the resume exists to recover. + if (candidate == 0 && response.Header != null && response.Header.Revision > 0) + { + candidate = response.Header.Revision + 1; + } + } else if (response.Events != null && response.Events.Count > 0) { long lastModRevision = response.Events[^1].Kv.ModRevision; @@ -136,8 +314,8 @@ private void TrackResumeRevision(long clientWatchId, WatchResponse response) } 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. + // 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) { @@ -160,10 +338,11 @@ private void TrackResumeRevision(long clientWatchId, WatchResponse response) public long Watch(WatchRequest request, Action callback, Metadata? headers = null, DateTime? deadline = null, CancellationToken cancellationToken = default) { - // Run the async method synchronously + // Run the async method synchronously. GetAwaiter().GetResult() rather than Wait()+Result so a + // failure surfaces as the RpcException the async overloads throw, not wrapped in an + // AggregateException that a `catch (RpcException)` would miss. Task task = Task.Run(() => WatchAsync(request, callback, headers, deadline, cancellationToken), cancellationToken); - task.Wait(cancellationToken); - return task.Result; + return task.GetAwaiter().GetResult(); } /// @@ -591,7 +770,7 @@ public void CancelWatch(long watchId) } // Cancel the watch - watchCancellation.CancellationTokenSource.Cancel(); + SafeCancel(watchCancellation.CancellationTokenSource); // Find the server watch ID that corresponds to our client watch ID long serverWatchId = GetServerWatchId(watchId); @@ -622,10 +801,19 @@ public void Dispose() _disposed = true; + // Release anyone still waiting for a watch to be acknowledged, so disposing the manager can + // never leave a caller blocked until the create timeout. + foreach (TaskCompletionSource pending in _pendingCreates.Values) + { + pending.TrySetException(new ObjectDisposedException(nameof(WatchManager))); + } + + _pendingCreates.Clear(); + // Cancel all watches foreach (WatchCancellation watchCancellation in _watches.Values) { - watchCancellation.CancellationTokenSource.Cancel(); + SafeCancel(watchCancellation.CancellationTokenSource); watchCancellation.CancellationTokenSource.Dispose(); } @@ -697,12 +885,20 @@ private void EnsureWatchStream(Metadata? headers, DateTime? deadline, Cancellati private void HandleConnectionFailure() { + Watcher? abandoned; + lock (_lockObject) { + abandoned = _watchStream; _watchStream = null; _watchIdMapping.Clear(); } + // Tear the old stream down. Leaving it undisposed keeps its receive loop, gRPC call and + // callbacks alive: if that stream is in fact still healthy, etcd goes on delivering the same + // events on it as well as on the replacement, and every event is handed to the callback twice. + abandoned?.Dispose(); + // Must run async to avoid blocking the caller (which might be the dead stream loop) Task.Run(async () => { @@ -717,17 +913,36 @@ private void HandleConnectionFailure() EnsureWatchStream(null, null, default); } + bool anyFailed = false; + 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. + // the stream was down are replayed instead of lost. A watch whose create was never + // acknowledged has nothing to resume from, but nothing can have been missed either: + // its Watch() call has not returned yet, so the caller cannot have written anything. if (watch.NextRevision > 0 && watch.Request.CreateRequest != null) { watch.Request.CreateRequest.StartRevision = watch.NextRevision; } - await _watchStream!.CreateWatchAsync(watch.Request, watch.Callback).ConfigureAwait(false); + try + { + await _watchStream!.CreateWatchAsync(watch.Request, watch.Callback).ConfigureAwait(false); + } + catch (Exception ex) + { + // Keep going: one watch failing to re-register must not strand the others. But + // remember it failed — swallowing it here would otherwise disable the retry + // below and leave that watch silently dead for the life of the client. + anyFailed = true; + Console.Error.WriteLine($"Failed to re-register watch {watch.WatchId}: {ex.Message}"); + } + } + + if (anyFailed) + { + throw new InvalidOperationException("one or more watches could not be re-registered"); } } catch (Exception ex) @@ -746,6 +961,7 @@ private class WatchCancellation 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 diff --git a/dotnet-etcd/watchclient/WatchStream.cs b/dotnet-etcd/watchclient/WatchStream.cs index e125c90..b6a5e6e 100644 --- a/dotnet-etcd/watchclient/WatchStream.cs +++ b/dotnet-etcd/watchclient/WatchStream.cs @@ -17,6 +17,14 @@ public class Watcher : IWatcher private readonly ConcurrentDictionary> _callbacks = new(); private readonly CancellationTokenSource _cts = new(); + /// + /// gRPC allows only one pending write per stream ("Only one write can be pending at a time"). + /// Creates and cancels are issued from user threads and from the reconnect loop, so the writes + /// must be serialized. The lock covers the write only — never a wait for a server response, or + /// a create awaiting its acknowledgement would block every cancel and reconnect behind it. + /// + private readonly SemaphoreSlim _writeLock = new(1, 1); + private readonly IAsyncDuplexStreamingCall _streamingCall; private readonly Action? _onConnectionFailure; @@ -46,10 +54,22 @@ public async Task CreateWatchAsync(WatchRequest request, Action c ArgumentNullException.ThrowIfNull(callback); - _callbacks[request.CreateRequest.WatchId] = callback; + long watchId = request.CreateRequest.WatchId; - // Send the watch request - await _streamingCall.RequestStream.WriteAsync(request); + // Register before writing: the server can answer before WriteAsync returns. + _callbacks[watchId] = callback; + + try + { + await WriteAsync(request).ConfigureAwait(false); + } + catch + { + // The create never reached the server; don't leave a callback behind for a watch that + // does not exist. + _callbacks.TryRemove(watchId, out _); + throw; + } } /// @@ -62,12 +82,25 @@ public async Task CancelWatchAsync(long watchId) // Send a cancel request WatchRequest request = new() { CancelRequest = new WatchCancelRequest { WatchId = watchId } }; - await _streamingCall.RequestStream.WriteAsync(request); + await WriteAsync(request).ConfigureAwait(false); // Remove the callback _callbacks.TryRemove(watchId, out _); } + private async Task WriteAsync(WatchRequest request) + { + await _writeLock.WaitAsync(_cts.Token).ConfigureAwait(false); + try + { + await _streamingCall.RequestStream.WriteAsync(request).ConfigureAwait(false); + } + finally + { + _writeLock.Release(); + } + } + private async Task ProcessWatchResponses() { try @@ -123,6 +156,11 @@ public void Dispose() _cts.Cancel(); _streamingCall.Dispose(); + // Deliberately not disposing _writeLock/_cts: a write may be in flight, and disposing them + // underneath it would surface as an ObjectDisposedException from inside the semaphore instead + // of the stream's own cancellation. Neither holds an unmanaged resource here, so letting the + // GC reclaim them is safe. + GC.SuppressFinalize(this); } }