cntryl · smiggleworth · Aug 25, 2026 · Aug 25, 2026
diff --git a/README.md b/README.md
@@ -85,6 +85,18 @@ equivalent domain method) and return typed handles implementing both
 a bounded buffer; a slow consumer terminates with
 `SubscriptionBackpressureException` without terminating sibling handles.
 
+Callback delivery uses an independent bounded queue configured with
+`AsyncHandlerOptions.QueueCapacity` (default `1024`). Every subscription handle
+exposes `Completion`: normal unsubscribe completes it, while callback queue
+overflow faults it with `AsyncHandlerOverflowException`, terminates the local
+registration, and is surfaced by async enumeration as the same typed failure.
+RPC worker saturation continues to use broker-visible protocol backpressure.
+
+Schedule backend unavailability and broker saturation use the distinct coded
+error `FitzErrorCodes.ScheduleBackendError` (`7010`). `Retryability` classifies
+it as retryable subject to operation safety; it is not reported as malformed
+cron or parse input.
+
 ## Unreleased preview migration
 
 This preview intentionally breaks the earlier callback subscription surface.

diff --git a/src/Abstractions/Domains/Kv/KvSubscription.cs b/src/Abstractions/Domains/Kv/KvSubscription.cs
@@ -4,13 +4,13 @@ namespace Cntryl.Fitz.Abstractions.Domains.Kv;
 
 public sealed class KvSubscription : SubscriptionHandle<KvNotification>
 {
-    public KvSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe)
-        : this(pattern, EmptyNotifications(), unsubscribe)
+    public KvSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe, Task? completion = null)
+        : this(pattern, EmptyNotifications(), unsubscribe, completion)
     {
     }
 
-    public KvSubscription(string pattern, IAsyncEnumerable<KvNotification> notifications, Func<CancellationToken, ValueTask> unsubscribe)
-        : base(pattern, notifications, unsubscribe)
+    public KvSubscription(string pattern, IAsyncEnumerable<KvNotification> notifications, Func<CancellationToken, ValueTask> unsubscribe, Task? completion = null)
+        : base(pattern, notifications, unsubscribe, completion)
     {
     }
 }
diff --git a/src/Abstractions/Domains/Lease/LeaseSubscription.cs b/src/Abstractions/Domains/Lease/LeaseSubscription.cs
@@ -4,15 +4,16 @@ namespace Cntryl.Fitz.Abstractions.Domains.Lease;
 
 public sealed class LeaseSubscription : SubscriptionHandle<LeaseChangeEvent>
 {
-    public LeaseSubscription(string route, Func<CancellationToken, ValueTask> unsubscribe)
-        : this(route, EmptyNotifications(), unsubscribe)
+    public LeaseSubscription(string route, Func<CancellationToken, ValueTask> unsubscribe, Task? completion = null)
+        : this(route, EmptyNotifications(), unsubscribe, completion)
     {
     }
 
     public LeaseSubscription(
         string route, IAsyncEnumerable<LeaseChangeEvent> notifications,
-        Func<CancellationToken, ValueTask> unsubscribe)
-        : base(route, notifications, unsubscribe)
+        Func<CancellationToken, ValueTask> unsubscribe,
+        Task? completion = null)
+        : base(route, notifications, unsubscribe, completion)
     {
         Route = route;
     }

diff --git a/src/Abstractions/Domains/Notice/NoticeSubscription.cs b/src/Abstractions/Domains/Notice/NoticeSubscription.cs
@@ -4,15 +4,16 @@ namespace Cntryl.Fitz.Abstractions.Domains.Notice;
 
 public sealed class NoticeSubscription : SubscriptionHandle<NoticeMessage>
 {
-    public NoticeSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe)
-        : this(pattern, EmptyNotifications(), unsubscribe)
+    public NoticeSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe, Task? completion = null)
+        : this(pattern, EmptyNotifications(), unsubscribe, completion)
     {
     }
 
     public NoticeSubscription(
         string pattern, IAsyncEnumerable<NoticeMessage> notifications,
-        Func<CancellationToken, ValueTask> unsubscribe)
-        : base(pattern, notifications, unsubscribe)
+        Func<CancellationToken, ValueTask> unsubscribe,
+        Task? completion = null)
+        : base(pattern, notifications, unsubscribe, completion)
     {
     }
 }
diff --git a/src/Abstractions/Domains/Queue/QueueSubscription.cs b/src/Abstractions/Domains/Queue/QueueSubscription.cs
@@ -4,15 +4,16 @@ namespace Cntryl.Fitz.Abstractions.Domains.Queue;
 
 public sealed class QueueSubscription : SubscriptionHandle<QueueAvailabilityEvent>
 {
-    public QueueSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe)
-        : this(pattern, EmptyNotifications(), unsubscribe)
+    public QueueSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe, Task? completion = null)
+        : this(pattern, EmptyNotifications(), unsubscribe, completion)
     {
     }
 
     public QueueSubscription(
         string pattern, IAsyncEnumerable<QueueAvailabilityEvent> notifications,
-        Func<CancellationToken, ValueTask> unsubscribe)
-        : base(pattern, notifications, unsubscribe)
+        Func<CancellationToken, ValueTask> unsubscribe,
+        Task? completion = null)
+        : base(pattern, notifications, unsubscribe, completion)
     {
     }
 }
diff --git a/src/Abstractions/Domains/Schedule/ScheduleSubscription.cs b/src/Abstractions/Domains/Schedule/ScheduleSubscription.cs
@@ -4,15 +4,16 @@ namespace Cntryl.Fitz.Abstractions.Domains.Schedule;
 
 public sealed class ScheduleSubscription : SubscriptionHandle<ScheduleNotification>
 {
-    public ScheduleSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe)
-        : this(pattern, EmptyNotifications(), unsubscribe)
+    public ScheduleSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe, Task? completion = null)
+        : this(pattern, EmptyNotifications(), unsubscribe, completion)
     {
     }
 
     public ScheduleSubscription(
         string pattern, IAsyncEnumerable<ScheduleNotification> notifications,
-        Func<CancellationToken, ValueTask> unsubscribe)
-        : base(pattern, notifications, unsubscribe)
+        Func<CancellationToken, ValueTask> unsubscribe,
+        Task? completion = null)
+        : base(pattern, notifications, unsubscribe, completion)
     {
     }
 }
diff --git a/src/Abstractions/Domains/Stream/StreamSubscription.cs b/src/Abstractions/Domains/Stream/StreamSubscription.cs
@@ -4,15 +4,16 @@ namespace Cntryl.Fitz.Abstractions.Domains.Stream;
 
 public sealed class StreamSubscription : SubscriptionHandle<StreamCommitEvent>
 {
-    public StreamSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe)
-        : this(pattern, EmptyNotifications(), unsubscribe)
+    public StreamSubscription(string pattern, Func<CancellationToken, ValueTask> unsubscribe, Task? completion = null)
+        : this(pattern, EmptyNotifications(), unsubscribe, completion)
     {
     }
 
     public StreamSubscription(
         string pattern, IAsyncEnumerable<StreamCommitEvent> notifications,
-        Func<CancellationToken, ValueTask> unsubscribe)
-        : base(pattern, notifications, unsubscribe)
+        Func<CancellationToken, ValueTask> unsubscribe,
+        Task? completion = null)
+        : base(pattern, notifications, unsubscribe, completion)
     {
     }
 }
diff --git a/src/Abstractions/FitzErrorCodes.cs b/src/Abstractions/FitzErrorCodes.cs
@@ -21,4 +21,5 @@ public static class FitzErrorCodes
     public const uint RpcSubscriptionLimit = 6013;
     public const uint ScheduleInvalidSubscriptionPattern = 7006;
     public const uint ScheduleSubscriptionLimit = 7007;
+    public const uint ScheduleBackendError = 7010;
 }
diff --git a/src/Abstractions/Runtime/SubscriptionHandle.cs b/src/Abstractions/Runtime/SubscriptionHandle.cs
@@ -5,18 +5,35 @@ namespace Cntryl.Fitz.Runtime;
 public abstract class SubscriptionHandle : IAsyncDisposable
 {
     private readonly Func<CancellationToken, ValueTask> _unsubscribe;
+    private readonly TaskCompletionSource? _ownedCompletion;
     private int _unsubscribed;
 
     protected SubscriptionHandle(
         string pattern,
-        Func<CancellationToken, ValueTask> unsubscribe)
+        Func<CancellationToken, ValueTask> unsubscribe,
+        Task? completion = null)
     {
         Pattern = pattern;
         _unsubscribe = unsubscribe;
+        if (completion is null)
+        {
+            _ownedCompletion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+            Completion = _ownedCompletion.Task;
+        }
+        else
+        {
+            Completion = completion;
+        }
     }
 
     public string Pattern { get; }
 
+    /// <summary>
+    /// Completes after normal unsubscribe and faults if local notification
+    /// delivery terminates, including async-handler queue overflow.
+    /// </summary>
+    public Task Completion { get; }
+
     public ValueTask UnsubscribeAsync(CancellationToken cancellationToken = default)
     {
         return UnsubscribeCoreAsync(cancellationToken);
@@ -35,7 +52,16 @@ private async ValueTask UnsubscribeCoreAsync(CancellationToken cancellationToken
             return;
         }
 
-        await _unsubscribe(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            await _unsubscribe(cancellationToken).ConfigureAwait(false);
+            _ownedCompletion?.TrySetResult();
+        }
+        catch (Exception exception)
+        {
+            _ownedCompletion?.TrySetException(exception);
+            throw;
+        }
     }
 }
 
@@ -46,8 +72,9 @@ public abstract class SubscriptionHandle<T> : SubscriptionHandle, IAsyncEnumerab
     protected SubscriptionHandle(
         string pattern,
         IAsyncEnumerable<T> notifications,
-        Func<CancellationToken, ValueTask> unsubscribe)
-        : base(pattern, unsubscribe)
+        Func<CancellationToken, ValueTask> unsubscribe,
+        Task? completion = null)
+        : base(pattern, unsubscribe, completion)
     {
         _notifications = notifications;
     }
@@ -80,3 +107,38 @@ public SubscriptionBackpressureException(string message, Exception innerExceptio
     {
     }
 }
+
+public sealed class AsyncHandlerOverflowException : Exception
+{
+    public const string ErrorCode = "ASYNC_HANDLER_OVERFLOW";
+
+    public AsyncHandlerOverflowException()
+        : this("The async handler queue overflowed.")
+    {
+    }
+
+    public AsyncHandlerOverflowException(string message)
+        : base(message)
+    {
+        Domain = "unknown";
+        Subscription = "unknown";
+    }
+
+    public AsyncHandlerOverflowException(string message, Exception innerException)
+        : base(message, innerException)
+    {
+        Domain = "unknown";
+        Subscription = "unknown";
+    }
+
+    public AsyncHandlerOverflowException(string domain, string subscription)
+        : base($"The async handler queue overflowed for {domain} subscription '{subscription}'.")
+    {
+        Domain = domain;
+        Subscription = subscription;
+    }
+
+    public string Code { get; } = ErrorCode;
+    public string Domain { get; }
+    public string Subscription { get; }
+}
diff --git a/src/Core/AsyncHandlerOptions.cs b/src/Core/AsyncHandlerOptions.cs
@@ -2,5 +2,6 @@ namespace Cntryl.Fitz;
 
 public sealed record AsyncHandlerOptions(
     int? MaxConcurrency = null,
-    TimeSpan? Timeout = null
+    TimeSpan? Timeout = null,
+    int QueueCapacity = 1024
 );
diff --git a/src/Core/Connection/FitzConnection.cs b/src/Core/Connection/FitzConnection.cs
@@ -58,7 +58,7 @@ public FitzConnection(ClientConfig config, Func<ITransport> transportFactory)
         _asyncHandlerDispatcher = new AsyncHandlerDispatcher(
             _config.ResolvedAsyncHandlers.MaxConcurrency,
             ResolveAsyncHandlerTimeout(),
-            Math.Min(Math.Max(_config.ResolvedMaxRequestQueueSize, 0), 1024),
+            Math.Max(_config.ResolvedAsyncHandlers.QueueCapacity, 0),
             OnAsyncHandlerError,
             onMetricsChanged: OnAsyncHandlerMetricsChanged,
             onSaturated: OnAsyncHandlerSaturated);

diff --git a/src/Core/Domains/Kv/KvClient.cs b/src/Core/Domains/Kv/KvClient.cs
@@ -103,11 +103,12 @@ public async Task<KvSubscription> SubscribeAsync(
             buffer.Write(notification);
             return ValueTask.CompletedTask;
         }, cancellationToken).ConfigureAwait(false);
+        buffer.ObserveCompletion(registration.Completion);
         return new KvSubscription(pattern, buffer.ReadAllAsync(CancellationToken.None), async token =>
         {
             buffer.Complete();
             await registration.UnsubscribeAsync(token).ConfigureAwait(false);
-        });
+        }, registration.Completion);
     }
 
     internal async Task<KvSubscription> SubscribeAsync(
@@ -127,8 +128,12 @@ internal async Task<KvSubscription> SubscribeAsync(
             SingleReader = true,
             SingleWriter = false
         });
-        SubscriptionRegistration<KvNotification>? registration = new(channel);
         var handleId = Interlocked.Increment(ref _nextHandleId);
+        SubscriptionRegistration<KvNotification>? registration = new(
+            channel,
+            "kv",
+            pattern,
+            token => UnsubscribeAsync(pattern, handleId, token));
         await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
         try
         {
@@ -152,8 +157,12 @@ internal async Task<KvSubscription> SubscribeAsync(
                 state.Writers[handleId] = registration;
             }
             SubscriptionPump.Start(registration, handler, _dispatchAsyncHandler);
+            var completion = registration.Completion;
             registration = null;
-            return new KvSubscription(pattern, token => UnsubscribeAsync(pattern, handleId, token));
+            return new KvSubscription(
+                pattern,
+                token => UnsubscribeAsync(pattern, handleId, token),
+                completion);
         }
         finally
         {

diff --git a/src/Core/Domains/Lease/LeaseClient.cs b/src/Core/Domains/Lease/LeaseClient.cs
@@ -464,11 +464,12 @@ public async Task<LeaseSubscription> SubscribeAsync(
             buffer.Write(notification);
             return ValueTask.CompletedTask;
         }, ct).ConfigureAwait(false);
+        buffer.ObserveCompletion(registration.Completion);
         return new LeaseSubscription(route, buffer.ReadAllAsync(CancellationToken.None), async token =>
         {
             buffer.Complete();
             await registration.UnsubscribeAsync(token).ConfigureAwait(false);
-        });
+        }, registration.Completion);
     }
 
     internal async Task<LeaseSubscription> SubscribeAsync(
@@ -500,14 +501,18 @@ internal async Task<LeaseSubscription> SubscribeAsync(
 
         try
         {
-            registration = new SubscriptionRegistration<LeaseChangeEvent>(channel);
+            registration = new SubscriptionRegistration<LeaseChangeEvent>(
+                channel,
+                "lease",
+                route,
+                token => UnsubscribeAsync(route, handleId, token));
             await _subscriptionGate.WaitAsync(ct).ConfigureAwait(false);
             gateAcquired = true;
 
             if (_subscriptionsByRoute.TryGetValue(route, out var existingSubscription))
             {
                 existingSubscription.Registrations[handleId] = registration;
-                var existingHandle = CreateSubscription(route, handleId);
+                var existingHandle = CreateSubscription(route, handleId, registration.Completion);
                 SubscriptionPump.Start(registration, handler, _dispatchAsyncHandler);
                 registration = null;
                 return existingHandle;
@@ -519,7 +524,7 @@ internal async Task<LeaseSubscription> SubscribeAsync(
             _subscriptionsByRoute[route] = subscription;
             _routesBySubscriptionId[subscriptionId] = route;
 
-            var handle = CreateSubscription(route, handleId);
+            var handle = CreateSubscription(route, handleId, registration.Completion);
             SubscriptionPump.Start(registration, handler, _dispatchAsyncHandler);
             registration = null;
             return handle;
@@ -535,11 +540,12 @@ internal async Task<LeaseSubscription> SubscribeAsync(
         }
     }
 
-    private LeaseSubscription CreateSubscription(string route, long handleId)
+    private LeaseSubscription CreateSubscription(string route, long handleId, Task completion)
     {
         return new LeaseSubscription(
             route,
-            cancellationToken => UnsubscribeAsync(route, handleId, cancellationToken));
+            cancellationToken => UnsubscribeAsync(route, handleId, cancellationToken),
+            completion);
     }
 
     private async Task<ulong> SubscribeWireAsync(string route, CancellationToken ct)