fix: align client failure and saturation handling

README.md

@@ -63,6 +63,9 @@ Use one control plane for request lifetime: `context.Context`.
 
 - RPC calls use context deadlines/cancellation only.
 - KV/Schedule/Notice/Queue/Lease/Stream subscription handlers return `error`.
+- Callback subscriptions expose `Completion()`: it yields `nil` after normal
+  unsubscribe, or a typed `*fitz.AsyncHandlerOverflowError` when the local
+  callback queue saturates and that local subscription is terminated.
 - Streaming iterators should be closed when no longer needed.
 - Clients validate route shape locally: scheme, segment count, empty segments,
   and method-specific wildcard placement. Route existence, permissions,
@@ -89,7 +92,10 @@ Reconnect guarantees:
 
 Production defaults also include heartbeat (`10s` interval, `30s` timeout),
 safe automatic retries for replayable reads, and a bounded outbound request
-queue of `1024`. See [docs/PUBLIC_CONTRACT.md](docs/PUBLIC_CONTRACT.md).
+queue of `1024`. Detached callbacks also use an independent queue capacity of
+`1024`, configurable with `fitz.WithAsyncHandlerQueueCapacity`; changing it does
+not change request admission or callback concurrency. See
+[docs/PUBLIC_CONTRACT.md](docs/PUBLIC_CONTRACT.md).
 
 The broker-backed test suite verifies those guarantees through a live disconnect proxy rather than by closing one client and creating another.
 

docs/PUBLIC_CONTRACT.md

@@ -48,6 +48,25 @@ visibility through `EnqueueWithOptions(..., WithQueueEnqueueDelaySeconds(n))`.
   and Queue `Enqueue` only after an explicit retryable broker rejection.
 - The outbound request queue is bounded. When saturated, operations fail with
   `ErrRequestQueueFull`.
+- Detached subscription callbacks use a separate bounded queue. Configure it
+  with `WithAsyncHandlerQueueCapacity` (default `1024`) and configure executing
+  callbacks independently with `WithAsyncHandlerMaxConcurrency` (default
+  `256`). The receive loop never waits for queue space.
+- Every KV, Notice, Queue, Lease, Schedule, and Stream callback subscription
+  exposes `Completion() <-chan error`. Explicit unsubscribe yields `nil`.
+  Saturation yields `*AsyncHandlerOverflowError`, terminates that local
+  registration, and attempts a best-effort wire unsubscribe when it was the
+  final local registration. Queue/Stream polling iterators and Schedule push
+  iterators surface the same terminal error through `Iterator.Err()`.
+- RPC workers retain their protocol-level backpressure response on saturation;
+  they are not converted to subscription completion errors.
+- Async-handler saturation increments `fitz.async_handlers.saturated`; active
+  and queued work are observable through `fitz.async_handlers.active` and
+  `fitz.async_handlers.queued`.
+- Schedule backend unavailability and broker saturation use
+  `ErrCodeScheduleBackendError` (`7010`). `IsRetryable` classifies that code as
+  retryable subject to operation safety. It is distinct from cron and parse
+  errors and is never mapped to malformed schedule input.
 
 ## Handles And Wake Helpers
 

fitz/errors.go

@@ -80,6 +80,7 @@ const (
 	ErrCodeScheduleInvalidSubscription = uint32(coreerrors.ScheduleInvalidSubscription)
 	ErrCodeScheduleSubscriptionLimit   = uint32(coreerrors.ScheduleSubscriptionLimit)
 	ErrCodeScheduleInvalidDeliveryMode = uint32(coreerrors.ScheduleInvalidDeliveryMode)
+	ErrCodeScheduleBackendError        = uint32(coreerrors.ScheduleBackendError)
 )
 
 // DomainError is a server-returned error carrying a numeric code and message.
@@ -96,6 +97,14 @@ type DomainError = coreerrors.DomainError
 // from server-returned domain errors.
 type TransportError = transport.TransportError
 
+// AsyncHandlerOverflowError terminates a callback subscription when its
+// configured local async-handler queue cannot accept another notification.
+type AsyncHandlerOverflowError = coreerrors.AsyncHandlerOverflowError
+
+// ErrAsyncHandlerOverflow supports errors.Is checks for callback-subscription
+// queue saturation.
+var ErrAsyncHandlerOverflow = coreerrors.ErrAsyncHandlerOverflow
+
 // IsRetryable reports whether err indicates a transient, retryable condition.
 // The following server-signaled situations are considered retryable:
 //   - KV isolation conflict (concurrent transaction collision)      [1004]
@@ -106,6 +115,7 @@ type TransportError = transport.TransportError
 //   - RPC worker not found (route may not yet be registered)        [6002]
 //   - RPC backpressure                                              [6003]
 //   - RPC route not registered (transient routing gap)              [6004]
+//   - Schedule backend unavailable or saturated                    [7010]
 func IsRetryable(err error) bool {
 	var de *coreerrors.DomainError
 	if !errors.As(err, &de) {
@@ -119,7 +129,8 @@ func IsRetryable(err error) bool {
 		ErrCodeRpcTimeout,
 		ErrCodeRpcWorkerNotFound,
 		ErrCodeRpcBackpressure,
-		ErrCodeRpcRouteNotRegistered:
+		ErrCodeRpcRouteNotRegistered,
+		ErrCodeScheduleBackendError:
 		return true
 	}
 	return false

fitz/errors_test.go

@@ -0,0 +1,15 @@
+package fitz
+
+import (
+	"testing"
+
+	coreerrors "github.com/cntryl/fitz-go/v2/internal/core/errors"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestShouldClassifyScheduleBackendErrorAsRetryable(t *testing.T) {
+	err := coreerrors.NewDomainError(ErrCodeScheduleBackendError, "backend busy")
+
+	assert.Equal(t, uint32(7010), ErrCodeScheduleBackendError)
+	assert.True(t, IsRetryable(err))
+}

fitz/kv.go

@@ -82,6 +82,13 @@ func (s *KVSubscription) Unsubscribe() {
 	}
 }
 
+func (s *KVSubscription) Completion() <-chan error {
+	if s == nil || s.inner == nil {
+		return nil
+	}
+	return s.inner.Completion()
+}
+
 type kvClient struct {
 	inner internalkv.Client
 }

fitz/lease.go

@@ -98,6 +98,13 @@ func (s *LeaseSubscription) Unsubscribe() {
 	}
 }
 
+func (s *LeaseSubscription) Completion() <-chan error {
+	if s == nil || s.inner == nil {
+		return nil
+	}
+	return s.inner.Completion()
+}
+
 type LeaseInfo struct {
 	Held             bool
 	OwnerID          string

fitz/managed_iterator.go

@@ -15,16 +15,17 @@ type managedPollResult[T any] struct {
 
 func startManagedPollingIterator[T any](
 	ctx context.Context,
-	subscribeWake func(context.Context, func()) (func(), error),
+	subscribeWake func(context.Context, func()) (func(), <-chan error, error),
 	poll func(context.Context) (managedPollResult[T], error),
 ) (Iterator[T], error) {
-	helperCtx, cancel := context.WithCancel(ctx)
+	helperCtx, cancel := context.WithCancelCause(ctx)
 	gate := NewWakeGate()
-	unsubscribe, err := subscribeWake(helperCtx, func() { gate.Wake() })
+	unsubscribe, completion, err := subscribeWake(helperCtx, func() { gate.Wake() })
 	if err != nil {
-		cancel()
+		cancel(err)
 		return nil, err
 	}
+	monitorSubscriptionCompletion(helperCtx, cancel, completion)
 
 	values := make(chan T)
 	errors := make(chan error, 1)
@@ -59,20 +60,20 @@ func startManagedPollingIterator[T any](
 		}
 	}()
 
-	return coreiter.NewChannelIterator[T](values, errors, cancel), nil
+	return coreiter.NewChannelIterator[T](values, errors, func() { cancel(context.Canceled) }), nil
 }
 
 func startManagedPushIterator[T any](
 	ctx context.Context,
 	capacity int,
-	subscribe func(context.Context, func(T) error) (func(), error),
+	subscribe func(context.Context, func(T) error) (func(), <-chan error, error),
 ) (Iterator[T], error) {
-	helperCtx, cancel := context.WithCancel(ctx)
+	helperCtx, cancel := context.WithCancelCause(ctx)
 	values := make(chan T, capacity)
 	errors := make(chan error, 1)
 	var deliveryMu sync.RWMutex
 	closed := false
-	unsubscribe, err := subscribe(helperCtx, func(value T) error {
+	unsubscribe, completion, err := subscribe(helperCtx, func(value T) error {
 		deliveryMu.RLock()
 		defer deliveryMu.RUnlock()
 		if closed {
@@ -86,9 +87,10 @@ func startManagedPushIterator[T any](
 		}
 	})
 	if err != nil {
-		cancel()
+		cancel(err)
 		return nil, err
 	}
+	monitorSubscriptionCompletion(helperCtx, cancel, completion)
 	go func() {
 		<-helperCtx.Done()
 		unsubscribe()
@@ -99,5 +101,20 @@ func startManagedPushIterator[T any](
 		close(errors)
 		deliveryMu.Unlock()
 	}()
-	return coreiter.NewChannelIterator[T](values, errors, cancel), nil
+	return coreiter.NewChannelIterator[T](values, errors, func() { cancel(context.Canceled) }), nil
+}
+
+func monitorSubscriptionCompletion(ctx context.Context, cancel context.CancelCauseFunc, completion <-chan error) {
+	if completion == nil {
+		return
+	}
+	go func() {
+		select {
+		case err := <-completion:
+			if err != nil {
+				cancel(err)
+			}
+		case <-ctx.Done():
+		}
+	}()
 }

fitz/managed_iterator_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"testing"
 
+	coreerrors "github.com/cntryl/fitz-go/v2/internal/core/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -12,12 +13,12 @@ func TestShouldRejectLateCallbackWithoutPanicAfterManagedPushIteratorCloses(t *t
 	unsubscribing := make(chan struct{})
 	releaseUnsubscribe := make(chan struct{})
 	iterator, err := startManagedPushIterator(context.Background(), 1,
-		func(_ context.Context, callback func(int) error) (func(), error) {
+		func(_ context.Context, callback func(int) error) (func(), <-chan error, error) {
 			handler = callback
 			return func() {
 				close(unsubscribing)
 				<-releaseUnsubscribe
-			}, nil
+			}, nil, nil
 		})
 	require.NoError(t, err)
 
@@ -27,3 +28,18 @@ func TestShouldRejectLateCallbackWithoutPanicAfterManagedPushIteratorCloses(t *t
 	require.False(t, iterator.Next()) // waits until shutdown closes the value channel
 	require.Error(t, handler(1))
 }
+
+func TestShouldSurfaceSubscriptionOverflowAsManagedIteratorError(t *testing.T) {
+	completion := make(chan error, 1)
+	iterator, err := startManagedPushIterator(context.Background(), 1,
+		func(_ context.Context, _ func(int) error) (func(), <-chan error, error) {
+			return func() {}, completion, nil
+		})
+	require.NoError(t, err)
+	overflow := &coreerrors.AsyncHandlerOverflowError{Domain: "schedule", SubscriptionID: 7}
+	completion <- overflow
+	close(completion)
+
+	require.False(t, iterator.Next())
+	require.ErrorIs(t, iterator.Err(), coreerrors.ErrAsyncHandlerOverflow)
+}

fitz/notice.go

@@ -30,6 +30,15 @@ func (s *NoticeSubscription) Unsubscribe() {
 	}
 }
 
+// Completion yields nil after unsubscribe or a typed terminal error when local
+// callback delivery can no longer continue.
+func (s *NoticeSubscription) Completion() <-chan error {
+	if s == nil || s.inner == nil {
+		return nil
+	}
+	return s.inner.Completion()
+}
+
 type NoticeClient interface {
 	Publish(ctx context.Context, route string, body []byte) error
 	Subscribe(ctx context.Context, pattern string, handler NoticeHandler) (*NoticeSubscription, error)

fitz/queue.go

@@ -59,6 +59,13 @@ func (s *QueueSubscription) Unsubscribe() {
 	}
 }
 
+func (s *QueueSubscription) Completion() <-chan error {
+	if s == nil || s.inner == nil {
+		return nil
+	}
+	return s.inner.Completion()
+}
+
 type QueueClient interface {
 	Enqueue(ctx context.Context, route string, body []byte) (uint64, error)
 	EnqueueWithOptions(ctx context.Context, route string, body []byte, opts ...QueueEnqueueOption) (uint64, error)
@@ -170,12 +177,12 @@ func (c *queueClient) ReserveWhenAvailable(ctx context.Context, route string, le
 		batchSize = 1
 	}
 	return startManagedPollingIterator(ctx,
-		func(helperCtx context.Context, wake func()) (func(), error) {
+		func(helperCtx context.Context, wake func()) (func(), <-chan error, error) {
 			subscription, err := c.Subscribe(helperCtx, route, func(context.Context, QueueAvailabilityNotification) error { wake(); return nil })
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
-			return subscription.Unsubscribe, nil
+			return subscription.Unsubscribe, subscription.Completion(), nil
 		},
 		func(helperCtx context.Context) (managedPollResult[[]*QueueItem], error) {
 			reserved, err := c.Reserve(helperCtx, route, leaseSecs, batchSize)

fitz/schedule.go

@@ -43,6 +43,13 @@ func (s *ScheduleSubscription) Unsubscribe() {
 	}
 }
 
+func (s *ScheduleSubscription) Completion() <-chan error {
+	if s == nil || s.inner == nil {
+		return nil
+	}
+	return s.inner.Completion()
+}
+
 type ScheduleClient interface {
 	Create(ctx context.Context, route string, cronExpr string, deliveryMode ScheduleDeliveryMode, payload []byte) (id string, err error)
 	Cancel(ctx context.Context, route string) error
@@ -111,14 +118,14 @@ func (c *scheduleClient) Subscribe(ctx context.Context, pattern string, handler
 // WaitForNotifications returns an iterator of schedule fire notifications.
 func (c *scheduleClient) WaitForNotifications(ctx context.Context, route string) (Iterator[ScheduleNotification], error) {
 	return startManagedPushIterator(ctx, 16,
-		func(helperCtx context.Context, emit func(ScheduleNotification) error) (func(), error) {
+		func(helperCtx context.Context, emit func(ScheduleNotification) error) (func(), <-chan error, error) {
 			subscription, err := c.Subscribe(helperCtx, route, func(_ context.Context, notification ScheduleNotification) error {
 				return emit(notification)
 			})
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
-			return subscription.Unsubscribe, nil
+			return subscription.Unsubscribe, subscription.Completion(), nil
 		})
 }
 

fitz/stream.go

@@ -157,6 +157,13 @@ func (s *StreamSubscription) Unsubscribe() {
 	}
 }
 
+func (s *StreamSubscription) Completion() <-chan error {
+	if s == nil || s.inner == nil {
+		return nil
+	}
+	return s.inner.Completion()
+}
+
 type StreamSession interface {
 	Append(ctx context.Context, expectedOffset uint64, body []byte, opts ...StreamAppendOption) (offset uint64, err error)
 	Commit(ctx context.Context, mode StreamCommitMode) error
@@ -260,12 +267,12 @@ func (c *streamClient) ReadWhenCommitted(ctx context.Context, route string, from
 	offset := fromOffset
 	var cursorFingerprint, capturedWatermark *uint64
 	return startManagedPollingIterator(ctx,
-		func(helperCtx context.Context, wake func()) (func(), error) {
+		func(helperCtx context.Context, wake func()) (func(), <-chan error, error) {
 			subscription, err := c.Subscribe(helperCtx, route, func(context.Context, StreamCommitNotification) error { wake(); return nil })
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
-			return subscription.Unsubscribe, nil
+			return subscription.Unsubscribe, subscription.Completion(), nil
 		},
 		func(helperCtx context.Context) (managedPollResult[[]StreamRecord], error) {
 			pollOptions := append([]StreamReadOption{}, opts...)

fitz/types.go

@@ -135,6 +135,15 @@ func WithAsyncHandlerMaxConcurrency(limit int) Option {
 	}
 }
 
+// WithAsyncHandlerQueueCapacity sets how many detached subscription callback
+// jobs may wait for an execution slot. It is independent of request admission
+// and WithAsyncHandlerMaxConcurrency. The default is 1024.
+func WithAsyncHandlerQueueCapacity(capacity int) Option {
+	return func(cfg *clientConfig) {
+		cfg.coreOptions = append(cfg.coreOptions, coreclient.WithAsyncHandlerQueueCapacity(capacity))
+	}
+}
+
 // WithReconnect controls the automatic reconnect behavior. When enabled, the
 // client will attempt to re-establish the transport connection using exponential
 // backoff starting at backoff, doubling up to the ceiling configured by