< Summary

Information
Class: dotnet_etcd.WatchManager
Assembly: dotnet-etcd
File(s): /home/runner/work/dotnet-etcd/dotnet-etcd/dotnet-etcd/watchclient/WatchManager.cs
Line coverage
88%
Covered lines: 477
Uncovered lines: 64
Coverable lines: 541
Total lines: 972
Line coverage: 88.1%
Branch coverage
74%
Covered branches: 82
Total branches: 110
Branch coverage: 74.5%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.cctor()100%11100%
.ctor(...)100%22100%
WatchAsync()16.66%7669.81%
WrappedCallback()100%1010100%
Dispatch(...)50%2273.91%
AckTimeoutFor(...)66.66%66100%
SafeCancel(...)100%1162.5%
TrackResumeRevision(...)80.76%262694.11%
Watch(...)100%11100%
Watch(...)100%11100%
Callback()75%4480%
WatchRange(...)75%4493.33%
Watch(...)75%4493.33%
WatchRange(...)75%4493.54%
WatchAsync()100%11100%
WatchRangeAsync()100%11100%
WatchAsync()100%11100%
WatchRangeAsync()100%11100%
WatchRange(...)100%11100%
CancelWatch(...)66.66%66100%
Dispose()87.5%8886.95%
GetRangeEnd(...)50%4472.72%
GetServerWatchId(...)50%4477.77%
EnsureWatchStream(...)100%22100%
HandleConnectionFailure()100%2280.32%
get_WatchId()100%11100%
get_CancellationTokenSource()100%11100%
get_Request()100%11100%
get_Callback()100%11100%
get_NextRevision()100%11100%

File(s)

/home/runner/work/dotnet-etcd/dotnet-etcd/dotnet-etcd/watchclient/WatchManager.cs

#LineLine coverage
 1#nullable enable
 2using System;
 3using System.Collections.Concurrent;
 4using System.Collections.Generic;
 5using System.Text;
 6using System.Threading;
 7using System.Threading.Tasks;
 8using dotnet_etcd.interfaces;
 9using Etcdserverpb;
 10using Google.Protobuf;
 11using Grpc.Core;
 12using Mvccpb;
 13
 14namespace dotnet_etcd;
 15
 16/// <summary>
 17///     Manages watch streams and provides a way to cancel watches
 18/// </summary>
 19public class WatchManager : IWatchManager
 20{
 21    /// <summary>
 22    ///     How long to wait for etcd to acknowledge a watch before giving up. Generous, because the
 23    ///     stream may have to reconnect (e.g. the server is restarting) before the create is accepted.
 24    /// </summary>
 225    private static readonly TimeSpan CreateWatchTimeout = TimeSpan.FromSeconds(30);
 26
 36027    private readonly object _lockObject = new();
 36028    private readonly ConcurrentDictionary<long, WatchCancellation> _watches = new();
 36029    private readonly ConcurrentDictionary<long, long> _watchIdMapping = new();
 30
 31    /// <summary>Watches whose create request is still awaiting the server's Created acknowledgement.</summary>
 36032    private readonly ConcurrentDictionary<long, TaskCompletionSource<WatchResponse>> _pendingCreates = new();
 33
 34    /// <summary>
 35    ///     Tail of the callback chain. Every response is appended to this single chain so user
 36    ///     callbacks run off the receive loop but remain serialized and ordered, as they were when the
 37    ///     loop invoked them inline. Note a slow callback queues responses without bound — callbacks
 38    ///     must not block indefinitely.
 39    /// </summary>
 36040    private readonly object _dispatchLock = new();
 41
 36042    private Task _dispatchChain = Task.CompletedTask;
 43
 44    private readonly
 45        Func<Metadata?, DateTime?, CancellationToken, IAsyncDuplexStreamingCall<WatchRequest, WatchResponse>>
 46        _watchStreamFactory;
 47
 48    private bool _disposed;
 36049    private long _nextWatchId = 1;
 50    private Watcher? _watchStream;
 51
 52    /// <summary>
 53    ///     Creates a new WatchManager
 54    /// </summary>
 55    /// <param name="watchStreamFactory">A factory function that creates a new watch stream</param>
 36056    public WatchManager(
 36057        Func<Metadata?, DateTime?, CancellationToken, IAsyncDuplexStreamingCall<WatchRequest, WatchResponse>>
 72058            watchStreamFactory) => _watchStreamFactory =
 36059        watchStreamFactory ?? throw new ArgumentNullException(nameof(watchStreamFactory));
 60
 61    /// <summary>
 62    ///     Creates a new watch request
 63    /// </summary>
 64    /// <param name="request">The watch requests to create</param>
 65    /// <param name="callback">The callback to invoke when a watch event is received</param>
 66    /// <param name="headers">The initial metadata to send with the call</param>
 67    /// <param name="deadline">An optional deadline for the call</param>
 68    /// <param name="cancellationToken">An optional token for canceling the call</param>
 69    /// <returns>A watch ID that can be used to cancel the watch</returns>
 70    public async Task<long> WatchAsync(WatchRequest request, Action<WatchResponse> callback, Metadata? headers = null,
 71        DateTime? deadline = null, CancellationToken cancellationToken = default)
 7272    {
 7273        ObjectDisposedException.ThrowIf(_disposed, this);
 74
 75        // Create a new watch stream if needed
 7176        EnsureWatchStream(headers, deadline, cancellationToken);
 77
 78        // Generate a new watch ID
 7179        long watchId = Interlocked.Increment(ref _nextWatchId);
 80
 81        // Create a cancellation token source that can be used to cancel the watch
 7182        CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 83
 84        // Work on our own copy: the watch id is stamped onto the request and the reconnect path
 85        // rewrites StartRevision on it, and neither should mutate the caller's object (which callers
 86        // may reuse across watches).
 7187        request = request.Clone();
 7188        request.CreateRequest.WatchId = watchId;
 89
 90        // Create a watch cancellation object
 7191        WatchCancellation watchCancellation = new()
 7192        {
 7193            WatchId = watchId,
 7194            CancellationTokenSource = cts,
 7195            Request = request,
 7196            Callback = WrappedCallback,
 7197
 7198            // Seed the resume point with the revision the caller asked to start from (etcd clientv3
 7199            // does the same: `nextRev := w.initReq.rev`). Without this, a watch created with an
 71100            // explicit StartRevision would look identical to a "from now" watch, and the Created ack —
 71101            // which carries the *current* cluster revision — would advance the resume point past the
 71102            // backlog the caller asked to replay.
 71103            NextRevision = request.CreateRequest.StartRevision
 71104        };
 105
 106        // Completed when etcd acknowledges the watch. Continuations MUST run asynchronously: the
 107        // callback below is invoked from the stream's receive loop, so resuming the awaiting caller
 108        // inline would run its code on that loop and stall event delivery for every watch on the
 109        // stream (and deadlock outright if the caller then blocks).
 71110        TaskCompletionSource<WatchResponse> acknowledged = new(TaskCreationOptions.RunContinuationsAsynchronously);
 111
 112        // Register the watch BEFORE writing the create request: the server can answer while WriteAsync
 113        // is still in flight, and both TrackResumeRevision and the reconnect loop ignore watches that
 114        // are not in _watches yet.
 71115        _watches[watchId] = watchCancellation;
 71116        _pendingCreates[watchId] = acknowledged;
 117
 118        try
 71119        {
 71120            await _watchStream!.CreateWatchAsync(request, WrappedCallback).ConfigureAwait(false);
 121
 122            // Wait until etcd has actually registered the watch. Creating a watch is asynchronous on
 123            // the server, so until the Created ack arrives the watch does not exist: a caller that
 124            // wrote a key as soon as Watch() returned could have the write applied first and never
 125            // see the event. If the stream dies while we wait, HandleConnectionFailure re-sends the
 126            // create on the new stream and its ack completes this same task.
 71127            TimeSpan ackTimeout = AckTimeoutFor(deadline);
 128
 71129            WatchResponse ack = await acknowledged.Task
 71130                .WaitAsync(ackTimeout, cts.Token)
 71131                .ConfigureAwait(false);
 132
 71133            if (ack.Canceled)
 0134            {
 0135                throw new RpcException(new Status(StatusCode.FailedPrecondition,
 0136                    $"etcd rejected the watch: {ack.CancelReason}"));
 137            }
 71138        }
 0139        catch (Exception ex)
 0140        {
 141            // The watch was never established. Cancel first so any response that shows up late is
 142            // ignored, then drop the entry so the reconnect loop won't try to re-register a watch the
 143            // caller believes failed.
 0144            _watches.TryRemove(watchId, out _);
 0145            SafeCancel(cts);
 0146            cts.Dispose();
 147
 148            // The server may nonetheless have registered the watch (e.g. we timed out waiting for an
 149            // ack that was merely slow). Best-effort tear it down rather than leak a server-side
 150            // watcher that streams into a callback nobody listens to.
 0151            if (_watchStream != null)
 0152            {
 0153                _ = _watchStream.CancelWatchAsync(watchId).ContinueWith(
 0154                    t => _ = t.Exception, TaskScheduler.Default);
 0155            }
 156
 0157            throw ex is TimeoutException
 0158                ? new RpcException(new Status(StatusCode.DeadlineExceeded,
 0159                    $"etcd did not acknowledge the watch within {AckTimeoutFor(deadline).TotalSeconds:0}s"))
 0160                : ex;
 161        }
 162        finally
 71163        {
 71164            _pendingCreates.TryRemove(watchId, out _);
 71165        }
 166
 71167        return watchId;
 168
 169        // Create a wrapper callback that checks if the watch has been canceled
 170        void WrappedCallback(WatchResponse response)
 146171        {
 172            // Complete the create before the cancellation guard below: a watch that is cancelled or
 173            // disposed while its create is still in flight must still release the awaiting caller
 174            // rather than leave it blocked until the timeout.
 146175            if ((response.Created || response.Canceled) &&
 146176                _pendingCreates.TryGetValue(watchId, out TaskCompletionSource<WatchResponse>? pending))
 71177            {
 71178                pending.TrySetResult(response);
 71179            }
 180
 146181            if (cts.IsCancellationRequested)
 2182            {
 2183                return;
 184            }
 185
 186            // If this is a creation response, update our watch ID mapping
 144187            if (response.Created)
 78188            {
 189                // Map the server-assigned watch ID to our client-generated watch ID
 78190                _watchIdMapping[response.WatchId] = watchId;
 78191            }
 192
 193            // Track the revision to resume from on reconnect so no events are missed in the gap.
 144194            TrackResumeRevision(watchId, response);
 195
 144196            Dispatch(watchId, response, callback, cts);
 146197        }
 71198    }
 199
 200    /// <summary>
 201    ///     Hands a response to the user's callback off the stream's receive loop.
 202    ///     <para>
 203    ///         The receive loop invokes callbacks inline, so running user code on it would let one slow
 204    ///         or blocking callback stall event delivery for every watch on the stream — and a callback
 205    ///         that starts another watch would deadlock outright, because the loop it is blocking is the
 206    ///         only thing that could deliver the new watch's acknowledgement.
 207    ///     </para>
 208    ///     <para>
 209    ///         Responses are appended to a single chain, so callbacks stay serialized and in order
 210    ///         exactly as they were when they ran on the receive loop. That matters: overloads such as
 211    ///         WatchRange(string[] paths, method) hand the SAME delegate to several watches, and those
 212    ///         callers are entitled to assume it is never entered concurrently.
 213    ///     </para>
 214    /// </summary>
 215    private void Dispatch(long watchId, WatchResponse response, Action<WatchResponse> callback,
 216        CancellationTokenSource cts)
 144217    {
 144218        lock (_dispatchLock)
 144219        {
 144220            _dispatchChain = _dispatchChain.ContinueWith(_ =>
 144221            {
 144222                if (cts.IsCancellationRequested)
 0223                {
 0224                    return;
 144225                }
 144226
 144227                try
 144228                {
 144229                    callback(response);
 144230                }
 0231                catch (Exception ex)
 0232                {
 144233                    // A throwing user callback must not fault the chain and silently stop delivery of
 144234                    // every subsequent event for this watch.
 0235                    Console.Error.WriteLine($"Watch callback for watch {watchId} threw: {ex}");
 0236                }
 288237            }, CancellationToken.None, TaskContinuationOptions.None, TaskScheduler.Default);
 144238        }
 144239    }
 240
 241    /// <summary>
 242    ///     How long to wait for the create acknowledgement. Honours the caller's deadline when one is
 243    ///     given, so a caller that asked for a short deadline is not held for the full default.
 244    /// </summary>
 245    private static TimeSpan AckTimeoutFor(DateTime? deadline)
 71246    {
 71247        if (deadline == null)
 70248        {
 70249            return CreateWatchTimeout;
 250        }
 251
 1252        TimeSpan remaining = deadline.Value.ToUniversalTime() - DateTime.UtcNow;
 1253        return remaining < TimeSpan.Zero ? TimeSpan.Zero
 1254            : remaining < CreateWatchTimeout ? remaining
 1255            : CreateWatchTimeout;
 71256    }
 257
 258    /// <summary>
 259    ///     Cancels a token source that another owner may already have disposed. Cancel() throws on a
 260    ///     disposed source (Dispose() itself is idempotent), and both the create-failure path and
 261    ///     CancelWatch/Dispose can reach the same source.
 262    /// </summary>
 263    private static void SafeCancel(CancellationTokenSource cts)
 33264    {
 265        try
 33266        {
 33267            cts.Cancel();
 33268        }
 0269        catch (ObjectDisposedException)
 0270        {
 271            // Already torn down by the other owner; nothing to do.
 0272        }
 33273    }
 274
 275    /// <summary>
 276    ///     Advances the stored resume revision for a watch based on a received response, mirroring the
 277    ///     etcd clientv3 "nextRev" logic: after an event batch resume from lastEvent.ModRevision + 1;
 278    ///     for a created/progress notification with no events, advance to the header revision. A
 279    ///     compaction cancel resets the resume point to the compact revision. Only ever moves forward.
 280    /// </summary>
 281    private void TrackResumeRevision(long clientWatchId, WatchResponse response)
 144282    {
 144283        if (!_watches.TryGetValue(clientWatchId, out WatchCancellation? watch))
 0284        {
 0285            return;
 286        }
 287
 144288        long candidate = watch.NextRevision;
 289
 144290        if (response.Canceled && response.CompactRevision > 0)
 1291        {
 292            // Our resume point was compacted away; the earliest we can resume from is CompactRevision.
 1293            candidate = response.CompactRevision;
 1294        }
 143295        else if (response.Created)
 78296        {
 297            // A Created ack carries the CURRENT cluster revision, which says nothing about what this
 298            // watch has observed. Only use it to seed a brand new "watch from now" watch: for a watch
 299            // being re-registered after a reconnect (StartRevision > 0) the server is about to replay
 300            // the backlog from that revision, and advancing to the ack's header would skip straight
 301            // past it — losing exactly the events the resume exists to recover.
 78302            if (candidate == 0 && response.Header != null && response.Header.Revision > 0)
 66303            {
 66304                candidate = response.Header.Revision + 1;
 66305            }
 78306        }
 65307        else if (response.Events != null && response.Events.Count > 0)
 64308        {
 64309            long lastModRevision = response.Events[^1].Kv.ModRevision;
 64310            if (lastModRevision + 1 > candidate)
 20311            {
 20312                candidate = lastModRevision + 1;
 20313            }
 64314        }
 1315        else if (response.Header != null && response.Header.Revision > 0)
 1316        {
 317            // Progress notification (no events): everything up to the header revision has been
 318            // observed, so the next start revision is header.Revision + 1.
 1319            long boundary = response.Header.Revision + 1;
 1320            if (boundary > candidate)
 1321            {
 1322                candidate = boundary;
 1323            }
 1324        }
 325
 144326        watch.NextRevision = candidate;
 144327    }
 328
 329    /// <summary>
 330    ///     Creates a new watch request
 331    /// </summary>
 332    /// <param name="request">The watch request to create</param>
 333    /// <param name="callback">The callback to invoke when a watch event is received</param>
 334    /// <param name="headers">The initial metadata to send with the call</param>
 335    /// <param name="deadline">An optional deadline for the call</param>
 336    /// <param name="cancellationToken">An optional token for canceling the call</param>
 337    /// <returns>A watch ID that can be used to cancel the watch</returns>
 338    public long Watch(WatchRequest request, Action<WatchResponse> callback, Metadata? headers = null,
 339        DateTime? deadline = null, CancellationToken cancellationToken = default)
 38340    {
 341        // Run the async method synchronously. GetAwaiter().GetResult() rather than Wait()+Result so a
 342        // failure surfaces as the RpcException the async overloads throw, not wrapped in an
 343        // AggregateException that a `catch (RpcException)` would miss.
 76344        Task<long> task = Task.Run(() => WatchAsync(request, callback, headers, deadline, cancellationToken), cancellati
 38345        return task.GetAwaiter().GetResult();
 38346    }
 347
 348    /// <summary>
 349    ///     Watches a specific key
 350    /// </summary>
 351    /// <param name="key">Key to watch</param>
 352    /// <param name="action">Action to be executed when watch event is triggered</param>
 353    /// <param name="headers">The initial metadata to send with the call</param>
 354    /// <param name="deadline">An optional deadline for the call</param>
 355    /// <param name="cancellationToken">An optional token for canceling the call</param>
 356    /// <returns>Watch ID</returns>
 357    public long Watch(string key, Action<WatchEvent> action, Metadata? headers = null, DateTime? deadline = null,
 358        CancellationToken cancellationToken = default)
 2359    {
 2360        ObjectDisposedException.ThrowIf(_disposed, this);
 361
 362        // Create a watch request for the key
 1363        WatchRequest request = new()
 1364        {
 1365            CreateRequest = new WatchCreateRequest
 1366            {
 1367                Key = ByteString.CopyFromUtf8(key), ProgressNotify = true, PrevKv = true
 1368            }
 1369        };
 370
 371        // Call the Watch method with the request
 1372        return Watch(request, Callback, headers, deadline, cancellationToken);
 373
 374        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 375        void Callback(WatchResponse response)
 2376        {
 2377            if (response.Events == null)
 0378            {
 0379                return;
 380            }
 381
 8382            foreach (Event evt in response.Events)
 1383            {
 1384                WatchEvent watchEvent = new() { Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Ty
 1385                action(watchEvent);
 1386            }
 2387        }
 1388    }
 389
 390    /// <summary>
 391    ///     Watches a range of keys with a prefix
 392    /// </summary>
 393    /// <param name="prefixKey">Prefix key to watch</param>
 394    /// <param name="action">Action to be executed when watch event is triggered</param>
 395    /// <param name="headers">The initial metadata to send with the call</param>
 396    /// <param name="deadline">An optional deadline for the call</param>
 397    /// <param name="cancellationToken">An optional token for canceling the call</param>
 398    /// <returns>Watch ID</returns>
 399    public long WatchRange(string prefixKey, Action<WatchEvent> action, Metadata? headers = null,
 400        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1401    {
 1402        ObjectDisposedException.ThrowIf(_disposed, this);
 403
 404        // Create a watch request for the range
 1405        WatchRequest request = new()
 1406        {
 1407            CreateRequest = new WatchCreateRequest
 1408            {
 1409                Key = ByteString.CopyFromUtf8(prefixKey),
 1410                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(prefixKey)),
 1411                ProgressNotify = true,
 1412                PrevKv = true
 1413            }
 1414        };
 415
 416        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1417        Action<WatchResponse> callback = response =>
 2418        {
 2419            if (response.Events == null)
 0420            {
 0421                return;
 1422            }
 1423
 8424            foreach (Event evt in response.Events)
 1425            {
 1426                WatchEvent watchEvent = new()
 1427                {
 1428                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1429                };
 1430                action(watchEvent);
 1431            }
 3432        };
 433
 434        // Call the Watch method with the request
 1435        return Watch(request, callback, headers, deadline, cancellationToken);
 1436    }
 437
 438    /// <summary>
 439    ///     Watches a specific key with start revision
 440    /// </summary>
 441    /// <param name="key">Key to watch</param>
 442    /// <param name="startRevision">Start revision</param>
 443    /// <param name="action">Action to be executed when watch event is triggered</param>
 444    /// <param name="headers">The initial metadata to send with the call</param>
 445    /// <param name="deadline">An optional deadline for the call</param>
 446    /// <param name="cancellationToken">An optional token for canceling the call</param>
 447    /// <returns>Watch ID</returns>
 448    public long Watch(string key, long startRevision, Action<WatchEvent> action, Metadata? headers = null,
 449        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1450    {
 1451        ObjectDisposedException.ThrowIf(_disposed, this);
 452
 453        // Create a watch request for the key with start revision
 1454        WatchRequest request = new()
 1455        {
 1456            CreateRequest = new WatchCreateRequest
 1457            {
 1458                Key = ByteString.CopyFromUtf8(key),
 1459                StartRevision = startRevision,
 1460                ProgressNotify = true,
 1461                PrevKv = true
 1462            }
 1463        };
 464
 465        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1466        Action<WatchResponse> callback = response =>
 2467        {
 2468            if (response.Events == null)
 0469            {
 0470                return;
 1471            }
 1472
 8473            foreach (Event evt in response.Events)
 1474            {
 1475                WatchEvent watchEvent = new()
 1476                {
 1477                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1478                };
 1479                action(watchEvent);
 1480            }
 3481        };
 482
 483        // Call the Watch method with the request
 1484        return Watch(request, callback, headers, deadline, cancellationToken);
 1485    }
 486
 487    /// <summary>
 488    ///     Watches a range of keys with a prefix and start revision
 489    /// </summary>
 490    /// <param name="prefixKey">Prefix key to watch</param>
 491    /// <param name="startRevision">Start revision</param>
 492    /// <param name="action">Action to be executed when watch event is triggered</param>
 493    /// <param name="headers">The initial metadata to send with the call</param>
 494    /// <param name="deadline">An optional deadline for the call</param>
 495    /// <param name="cancellationToken">An optional token for canceling the call</param>
 496    /// <returns>Watch ID</returns>
 497    public long WatchRange(string prefixKey, long startRevision, Action<WatchEvent> action, Metadata? headers = null,
 498        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1499    {
 1500        ObjectDisposedException.ThrowIf(_disposed, this);
 501
 502        // Create a watch request for the range with start revision
 1503        WatchRequest request = new()
 1504        {
 1505            CreateRequest = new WatchCreateRequest
 1506            {
 1507                Key = ByteString.CopyFromUtf8(prefixKey),
 1508                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(prefixKey)),
 1509                StartRevision = startRevision,
 1510                ProgressNotify = true,
 1511                PrevKv = true
 1512            }
 1513        };
 514
 515        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1516        Action<WatchResponse> callback = response =>
 2517        {
 2518            if (response.Events == null)
 0519            {
 0520                return;
 1521            }
 1522
 8523            foreach (Event evt in response.Events)
 1524            {
 1525                WatchEvent watchEvent = new()
 1526                {
 1527                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1528                };
 1529                action(watchEvent);
 1530            }
 3531        };
 532
 533        // Call the Watch method with the request
 1534        return Watch(request, callback, headers, deadline, cancellationToken);
 1535    }
 536
 537    /// <summary>
 538    ///     Watches a specific key asynchronously
 539    /// </summary>
 540    /// <param name="key">Key to watch</param>
 541    /// <param name="action">Action to be executed when watch event is triggered</param>
 542    /// <param name="headers">The initial metadata to send with the call</param>
 543    /// <param name="deadline">An optional deadline for the call</param>
 544    /// <param name="cancellationToken">An optional token for canceling the call</param>
 545    /// <returns>Watch ID</returns>
 546    public async Task<long> WatchAsync(string key, Action<WatchEvent> action, Metadata? headers = null,
 547        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1548    {
 1549        ObjectDisposedException.ThrowIf(_disposed, this);
 550
 551        // Create a watch request for the key
 1552        WatchRequest request = new()
 1553        {
 1554            CreateRequest = new WatchCreateRequest
 1555            {
 1556                Key = ByteString.CopyFromUtf8(key), ProgressNotify = true, PrevKv = true
 1557            }
 1558        };
 559
 560        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1561        Action<WatchResponse> callback = response =>
 2562        {
 2563            if (response.Events == null)
 0564            {
 0565                return;
 1566            }
 1567
 8568            foreach (Event evt in response.Events)
 1569            {
 1570                WatchEvent watchEvent = new()
 1571                {
 1572                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1573                };
 1574                action(watchEvent);
 1575            }
 3576        };
 577
 578        // Call the WatchAsync method with the request
 1579        return await WatchAsync(request, callback, headers, deadline, cancellationToken).ConfigureAwait(false);
 1580    }
 581
 582    /// <summary>
 583    ///     Watches a range of keys with a prefix asynchronously
 584    /// </summary>
 585    /// <param name="prefixKey">Prefix key to watch</param>
 586    /// <param name="action">Action to be executed when watch event is triggered</param>
 587    /// <param name="headers">The initial metadata to send with the call</param>
 588    /// <param name="deadline">An optional deadline for the call</param>
 589    /// <param name="cancellationToken">An optional token for canceling the call</param>
 590    /// <returns>Watch ID</returns>
 591    public async Task<long> WatchRangeAsync(string prefixKey, Action<WatchEvent> action, Metadata? headers = null,
 592        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1593    {
 1594        ObjectDisposedException.ThrowIf(_disposed, this);
 595
 596        // Create a watch request for the range
 1597        WatchRequest request = new()
 1598        {
 1599            CreateRequest = new WatchCreateRequest
 1600            {
 1601                Key = ByteString.CopyFromUtf8(prefixKey),
 1602                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(prefixKey)),
 1603                ProgressNotify = true,
 1604                PrevKv = true
 1605            }
 1606        };
 607
 608        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1609        Action<WatchResponse> callback = response =>
 2610        {
 2611            if (response.Events == null)
 0612            {
 0613                return;
 1614            }
 1615
 8616            foreach (Event evt in response.Events)
 1617            {
 1618                WatchEvent watchEvent = new()
 1619                {
 1620                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1621                };
 1622                action(watchEvent);
 1623            }
 3624        };
 625
 626        // Call the WatchAsync method with the request
 1627        return await WatchAsync(request, callback, headers, deadline, cancellationToken).ConfigureAwait(false);
 1628    }
 629
 630    /// <summary>
 631    ///     Watches a specific key with start revision asynchronously
 632    /// </summary>
 633    /// <param name="key">Key to watch</param>
 634    /// <param name="startRevision">Start revision</param>
 635    /// <param name="action">Action to be executed when watch event is triggered</param>
 636    /// <param name="headers">The initial metadata to send with the call</param>
 637    /// <param name="deadline">An optional deadline for the call</param>
 638    /// <param name="cancellationToken">An optional token for canceling the call</param>
 639    /// <returns>Watch ID</returns>
 640    public async Task<long> WatchAsync(string key, long startRevision, Action<WatchEvent> action,
 641        Metadata? headers = null, DateTime? deadline = null, CancellationToken cancellationToken = default)
 1642    {
 1643        ObjectDisposedException.ThrowIf(_disposed, this);
 644
 645        // Create a watch request for the key with start revision
 1646        WatchRequest request = new()
 1647        {
 1648            CreateRequest = new WatchCreateRequest
 1649            {
 1650                Key = ByteString.CopyFromUtf8(key),
 1651                StartRevision = startRevision,
 1652                ProgressNotify = true,
 1653                PrevKv = true
 1654            }
 1655        };
 656
 657        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1658        Action<WatchResponse> callback = response =>
 2659        {
 2660            if (response.Events == null)
 0661            {
 0662                return;
 1663            }
 1664
 8665            foreach (Event evt in response.Events)
 1666            {
 1667                WatchEvent watchEvent = new()
 1668                {
 1669                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1670                };
 1671                action(watchEvent);
 1672            }
 3673        };
 674
 675        // Call the WatchAsync method with the request
 1676        return await WatchAsync(request, callback, headers, deadline, cancellationToken).ConfigureAwait(false);
 1677    }
 678
 679    /// <summary>
 680    ///     Watches a range of keys with a prefix and start revision asynchronously
 681    /// </summary>
 682    /// <param name="prefixKey">Prefix key to watch</param>
 683    /// <param name="startRevision">Start revision</param>
 684    /// <param name="action">Action to be executed when watch event is triggered</param>
 685    /// <param name="headers">The initial metadata to send with the call</param>
 686    /// <param name="deadline">An optional deadline for the call</param>
 687    /// <param name="cancellationToken">An optional token for canceling the call</param>
 688    /// <returns>Watch ID</returns>
 689    public async Task<long> WatchRangeAsync(string prefixKey, long startRevision, Action<WatchEvent> action,
 690        Metadata? headers = null, DateTime? deadline = null, CancellationToken cancellationToken = default)
 1691    {
 1692        ObjectDisposedException.ThrowIf(_disposed, this);
 693
 694        // Create a watch request for the range with start revision
 1695        WatchRequest request = new()
 1696        {
 1697            CreateRequest = new WatchCreateRequest
 1698            {
 1699                Key = ByteString.CopyFromUtf8(prefixKey),
 1700                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(prefixKey)),
 1701                StartRevision = startRevision,
 1702                ProgressNotify = true,
 1703                PrevKv = true
 1704            }
 1705        };
 706
 707        // Create a wrapper callback that converts the WatchResponse to a WatchEvent
 1708        Action<WatchResponse> callback = response =>
 2709        {
 2710            if (response.Events == null)
 0711            {
 0712                return;
 1713            }
 1714
 8715            foreach (Event evt in response.Events)
 1716            {
 1717                WatchEvent watchEvent = new()
 1718                {
 1719                    Key = evt.Kv.Key.ToStringUtf8(), Value = evt.Kv.Value.ToStringUtf8(), Type = evt.Type
 1720                };
 1721                action(watchEvent);
 1722            }
 3723        };
 724
 725        // Call the WatchAsync method with the request
 1726        return await WatchAsync(request, callback, headers, deadline, cancellationToken).ConfigureAwait(false);
 1727    }
 728
 729    /// <summary>
 730    ///     Watches a key range
 731    /// </summary>
 732    /// <param name="path">The path to watch</param>
 733    /// <param name="callback">The callback to invoke when a watch event is received</param>
 734    /// <param name="headers">The initial metadata to send with the call</param>
 735    /// <param name="deadline">An optional deadline for the call</param>
 736    /// <param name="cancellationToken">An optional token for canceling the call</param>
 737    /// <returns>A watch ID that can be used to cancel the watch</returns>
 738    public long WatchRange(string path, Action<WatchResponse> callback, Metadata? headers = null,
 739        DateTime? deadline = null, CancellationToken cancellationToken = default)
 1740    {
 1741        ObjectDisposedException.ThrowIf(_disposed, this);
 742
 743        // Create a watch request for the range
 1744        WatchRequest request = new()
 1745        {
 1746            CreateRequest = new WatchCreateRequest
 1747            {
 1748                Key = ByteString.CopyFromUtf8(path),
 1749                RangeEnd = ByteString.CopyFromUtf8(GetRangeEnd(path)),
 1750                ProgressNotify = true,
 1751                PrevKv = true
 1752            }
 1753        };
 754
 755        // Call the Watch method with the request
 1756        return Watch(request, callback, headers, deadline, cancellationToken);
 1757    }
 758
 759    /// <summary>
 760    ///     Cancels a watch request
 761    /// </summary>
 762    /// <param name="watchId">The ID of the watch to cancel</param>
 763    public void CancelWatch(long watchId)
 2764    {
 2765        ObjectDisposedException.ThrowIf(_disposed, this);
 766
 2767        if (!_watches.TryRemove(watchId, out WatchCancellation? watchCancellation))
 1768        {
 1769            return;
 770        }
 771
 772        // Cancel the watch
 1773        SafeCancel(watchCancellation.CancellationTokenSource);
 774
 775        // Find the server watch ID that corresponds to our client watch ID
 1776        long serverWatchId = GetServerWatchId(watchId);
 777
 778        // Cancel the watch on the server if we found a mapping
 1779        if (serverWatchId != -1 && _watchStream != null)
 1780        {
 1781            _watchIdMapping.TryRemove(serverWatchId, out _);
 1782            _watchStream.CancelWatchAsync(serverWatchId).ContinueWith(_ =>
 1783            {
 1784                // Ignore exceptions
 2785            });
 1786        }
 787
 788        // Dispose the cancellation token source
 1789        watchCancellation.CancellationTokenSource.Dispose();
 2790    }
 791
 792    /// <summary>
 793    ///     Disposes the watch manager
 794    /// </summary>
 795    public void Dispose()
 109796    {
 109797        if (_disposed)
 1798        {
 1799            return;
 800        }
 801
 108802        _disposed = true;
 803
 804        // Release anyone still waiting for a watch to be acknowledged, so disposing the manager can
 805        // never leave a caller blocked until the create timeout.
 324806        foreach (TaskCompletionSource<WatchResponse> pending in _pendingCreates.Values)
 0807        {
 0808            pending.TrySetException(new ObjectDisposedException(nameof(WatchManager)));
 0809        }
 810
 108811        _pendingCreates.Clear();
 812
 813        // Cancel all watches
 388814        foreach (WatchCancellation watchCancellation in _watches.Values)
 32815        {
 32816            SafeCancel(watchCancellation.CancellationTokenSource);
 32817            watchCancellation.CancellationTokenSource.Dispose();
 32818        }
 819
 108820        _watches.Clear();
 821
 822        // Dispose the watch stream
 108823        if (_watchStream != null)
 31824        {
 31825            _watchStream.Dispose();
 31826            _watchStream = null;
 31827        }
 828
 108829        GC.SuppressFinalize(this);
 109830    }
 831
 832    private static string GetRangeEnd(string path)
 5833    {
 834        // Calculate the range end for the given path
 835        // This is the same logic used in EtcdClient.GetRangeEnd
 5836        byte[] bytes = Encoding.UTF8.GetBytes(path);
 10837        for (int i = bytes.Length - 1; i >= 0; i--)
 5838        {
 5839            if (bytes[i] >= 0xff)
 0840            {
 0841                continue;
 842            }
 843
 5844            bytes[i]++;
 5845            return Encoding.UTF8.GetString(bytes, 0, i + 1);
 846        }
 847
 0848        return string.Empty;
 5849    }
 850
 851    /// <summary>
 852    ///     Gets the server watch ID for a client watch ID
 853    /// </summary>
 854    /// <param name="clientWatchId">The client watch ID</param>
 855    /// <returns>The server watch ID, or -1 if not found</returns>
 856    private long GetServerWatchId(long clientWatchId)
 1857    {
 4858        foreach (KeyValuePair<long, long> kvp in _watchIdMapping)
 1859        {
 1860            if (kvp.Value == clientWatchId)
 1861            {
 1862                return kvp.Key;
 863            }
 0864        }
 865
 0866        return -1;
 1867    }
 868
 869    /// <param name="cancellationToken">An optional token for canceling the call</param>
 870    private void EnsureWatchStream(Metadata? headers, DateTime? deadline, CancellationToken cancellationToken)
 80871    {
 80872        lock (_lockObject)
 80873        {
 80874            if (_watchStream != null)
 14875            {
 14876                return;
 877            }
 878
 879            // Create a new watch stream
 66880            IAsyncDuplexStreamingCall<WatchRequest, WatchResponse> watchStreamCall =
 66881                _watchStreamFactory(headers, deadline, cancellationToken);
 66882            _watchStream = new Watcher(watchStreamCall, HandleConnectionFailure);
 66883        }
 80884    }
 885
 886    private void HandleConnectionFailure()
 9887    {
 888        Watcher? abandoned;
 889
 9890        lock (_lockObject)
 9891        {
 9892            abandoned = _watchStream;
 9893            _watchStream = null;
 9894            _watchIdMapping.Clear();
 9895        }
 896
 897        // Tear the old stream down. Leaving it undisposed keeps its receive loop, gRPC call and
 898        // callbacks alive: if that stream is in fact still healthy, etcd goes on delivering the same
 899        // events on it as well as on the replacement, and every event is handed to the callback twice.
 9900        abandoned?.Dispose();
 901
 902        // Must run async to avoid blocking the caller (which might be the dead stream loop)
 9903        Task.Run(async () =>
 9904        {
 9905            try
 9906            {
 9907                // Wait small delay to allow network to stabilize
 9908                await Task.Delay(500);
 9909
 9910                lock (_lockObject)
 9911                {
 9912                   if (_disposed) return;
 9913                   EnsureWatchStream(null, null, default);
 9914                }
 9915
 9916                bool anyFailed = false;
 9917
 45918                foreach (var watch in _watches.Values)
 9919                {
 9920                    // Resume from the revision after the last observed event so events written while
 9921                    // the stream was down are replayed instead of lost. A watch whose create was never
 9922                    // acknowledged has nothing to resume from, but nothing can have been missed either:
 9923                    // its Watch() call has not returned yet, so the caller cannot have written anything.
 9924                    if (watch.NextRevision > 0 && watch.Request.CreateRequest != null)
 8925                    {
 8926                        watch.Request.CreateRequest.StartRevision = watch.NextRevision;
 8927                    }
 9928
 9929                    try
 9930                    {
 9931                        await _watchStream!.CreateWatchAsync(watch.Request, watch.Callback).ConfigureAwait(false);
 9932                    }
 0933                    catch (Exception ex)
 0934                    {
 9935                        // Keep going: one watch failing to re-register must not strand the others. But
 9936                        // remember it failed — swallowing it here would otherwise disable the retry
 9937                        // below and leave that watch silently dead for the life of the client.
 0938                        anyFailed = true;
 0939                        Console.Error.WriteLine($"Failed to re-register watch {watch.WatchId}: {ex.Message}");
 0940                    }
 9941                }
 9942
 9943                if (anyFailed)
 0944                {
 0945                    throw new InvalidOperationException("one or more watches could not be re-registered");
 9946                }
 9947            }
 0948            catch (Exception ex)
 0949            {
 0950                Console.Error.WriteLine($"Watch reconnection failed: {ex.Message}");
 9951                // Retry in 5s
 0952                _ = Task.Delay(5000).ContinueWith(_ => HandleConnectionFailure());
 0953            }
 18954        });
 9955    }
 956
 957    private class WatchCancellation
 958    {
 71959        public long WatchId { get; set; }
 137960        public required CancellationTokenSource CancellationTokenSource { get; set; }
 96961        public required WatchRequest Request { get; set; }
 80962        public required Action<WatchResponse> Callback { get; set; }
 963
 964
 965        /// <summary>
 966        ///     The revision to resume this watch from if the stream is re-established. Tracks the
 967        ///     revision after the last event/notification observed, so a reconnect does not miss
 968        ///     events written during the gap (mirrors the etcd clientv3 nextRev behavior).
 969        /// </summary>
 376970        public long NextRevision { get; set; }
 971    }
 972}
Generated by: ReportGenerator 5.5.10.0
07/13/2026 - 17:48:55
GitHub | reportgenerator.io

Methods/Properties

.cctor()
.ctor(System.Func`4<Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken,dotnet_etcd.interfaces.IAsyncDuplexStreamingCall`2<Etcdserverpb.WatchRequest,Etcdserverpb.WatchResponse>>)
WatchAsync()
WrappedCallback()
Dispatch(System.Int64,Etcdserverpb.WatchResponse,System.Action`1<Etcdserverpb.WatchResponse>,System.Threading.CancellationTokenSource)
AckTimeoutFor(System.Nullable`1<System.DateTime>)
SafeCancel(System.Threading.CancellationTokenSource)
TrackResumeRevision(System.Int64,Etcdserverpb.WatchResponse)
Watch(Etcdserverpb.WatchRequest,System.Action`1<Etcdserverpb.WatchResponse>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
Watch(System.String,System.Action`1<dotnet_etcd.WatchEvent>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
Callback()
WatchRange(System.String,System.Action`1<dotnet_etcd.WatchEvent>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
Watch(System.String,System.Int64,System.Action`1<dotnet_etcd.WatchEvent>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
WatchRange(System.String,System.Int64,System.Action`1<dotnet_etcd.WatchEvent>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
WatchAsync()
WatchRangeAsync()
WatchAsync()
WatchRangeAsync()
WatchRange(System.String,System.Action`1<Etcdserverpb.WatchResponse>,Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
CancelWatch(System.Int64)
Dispose()
GetRangeEnd(System.String)
GetServerWatchId(System.Int64)
EnsureWatchStream(Grpc.Core.Metadata,System.Nullable`1<System.DateTime>,System.Threading.CancellationToken)
HandleConnectionFailure()
get_WatchId()
get_CancellationTokenSource()
get_Request()
get_Callback()
get_NextRevision()