From b96a309e2942533b906cb93a3acad1f6de289488 Mon Sep 17 00:00:00 2001 From: Jeff Repanich Date: Tue, 25 Aug 2026 10:15:44 -0400 Subject: [PATCH] fix: align client failure handling --- README.md | 8 +- docs/PUBLIC_CONTRACT.md | 19 ++++ fitz/errors.go | 13 ++- fitz/errors_test.go | 15 ++++ fitz/kv.go | 7 ++ fitz/lease.go | 7 ++ fitz/managed_iterator.go | 37 +++++--- fitz/managed_iterator_test.go | 20 ++++- fitz/notice.go | 9 ++ fitz/queue.go | 13 ++- fitz/schedule.go | 13 ++- fitz/stream.go | 13 ++- fitz/types.go | 9 ++ internal/core/client/client.go | 9 ++ internal/core/client/client_test.go | 14 +++ .../core/connection/async_launcher_test.go | 6 +- internal/core/connection/connection.go | 78 +++++++++++++---- internal/core/connection/connection_test.go | 26 ++++++ internal/core/errors/errors.go | 19 ++++ internal/core/errors/errors_registry_test.go | 1 + internal/core/subscriptions/registry.go | 87 +++++++++++++++++-- internal/core/subscriptions/registry_test.go | 20 +++++ internal/domains/kv/watch.go | 24 +++-- internal/domains/lease/lease.go | 48 ++++++---- internal/domains/notice/notice.go | 35 +++++--- internal/domains/notice/notice_test.go | 41 +++++++++ internal/domains/queue/queue.go | 35 +++++--- internal/domains/schedule/schedule.go | 35 +++++--- internal/domains/stream/stream.go | 43 ++++++--- internal/domains/stream/stream_test.go | 33 +++++++ 30 files changed, 623 insertions(+), 114 deletions(-) create mode 100644 fitz/errors_test.go diff --git a/README.md b/README.md index c2bc7a1..ff2b7ef 100644 --- a/README.md +++ b/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. diff --git a/docs/PUBLIC_CONTRACT.md b/docs/PUBLIC_CONTRACT.md index c500130..a97bbf7 100644 --- a/docs/PUBLIC_CONTRACT.md +++ b/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 diff --git a/fitz/errors.go b/fitz/errors.go index c1d1dc4..979ba44 100644 --- a/fitz/errors.go +++ b/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 diff --git a/fitz/errors_test.go b/fitz/errors_test.go new file mode 100644 index 0000000..c752153 --- /dev/null +++ b/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)) +} diff --git a/fitz/kv.go b/fitz/kv.go index 701b9a6..ee45afe 100644 --- a/fitz/kv.go +++ b/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 } diff --git a/fitz/lease.go b/fitz/lease.go index cf039c1..c7f90ae 100644 --- a/fitz/lease.go +++ b/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 diff --git a/fitz/managed_iterator.go b/fitz/managed_iterator.go index 77dcfe8..e73e0cf 100644 --- a/fitz/managed_iterator.go +++ b/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(): + } + }() } diff --git a/fitz/managed_iterator_test.go b/fitz/managed_iterator_test.go index 64088fa..07d1aa6 100644 --- a/fitz/managed_iterator_test.go +++ b/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) +} diff --git a/fitz/notice.go b/fitz/notice.go index 3f8f0c2..710ecd3 100644 --- a/fitz/notice.go +++ b/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) diff --git a/fitz/queue.go b/fitz/queue.go index 7ec2e96..f4ef1a6 100644 --- a/fitz/queue.go +++ b/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) diff --git a/fitz/schedule.go b/fitz/schedule.go index 46bf012..c685685 100644 --- a/fitz/schedule.go +++ b/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 }) } diff --git a/fitz/stream.go b/fitz/stream.go index 67ecb9b..2a65eab 100644 --- a/fitz/stream.go +++ b/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...) diff --git a/fitz/types.go b/fitz/types.go index 4a81d92..1128846 100644 --- a/fitz/types.go +++ b/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 diff --git a/internal/core/client/client.go b/internal/core/client/client.go index 0920a17..fdbb9a8 100644 --- a/internal/core/client/client.go +++ b/internal/core/client/client.go @@ -111,6 +111,7 @@ type Config struct { MaxRequestQueueSize int AsyncHandlerTimeout time.Duration AsyncHandlerMaxConcurrency int + AsyncHandlerQueueCapacity int // Reconnection ReconnectEnabled bool @@ -152,6 +153,7 @@ func defaultConfig() *Config { MaxRequestQueueSize: 1024, AsyncHandlerTimeout: 30 * time.Second, AsyncHandlerMaxConcurrency: 256, + AsyncHandlerQueueCapacity: 1024, ReconnectEnabled: true, ReconnectBackoff: 250 * time.Millisecond, ReconnectMaxDelay: 5 * time.Second, @@ -213,6 +215,12 @@ func WithAsyncHandlerMaxConcurrency(limit int) Option { return func(c *Config) { c.AsyncHandlerMaxConcurrency = limit } } +// WithAsyncHandlerQueueCapacity sets the number of detached callback jobs that +// may wait for an execution slot. A value <= 0 uses the default capacity. +func WithAsyncHandlerQueueCapacity(capacity int) Option { + return func(c *Config) { c.AsyncHandlerQueueCapacity = capacity } +} + // WithReconnect enables/disables automatic reconnection. func WithReconnect(enabled bool, backoff time.Duration, maxAttempts int) Option { return func(c *Config) { @@ -669,6 +677,7 @@ func (c *Client) dialConnection(ctx context.Context, transportType TransportType MaxRequestQueueSize: c.config.MaxRequestQueueSize, AsyncHandlerTimeout: c.config.AsyncHandlerTimeout, AsyncHandlerMaxConcurrency: c.config.AsyncHandlerMaxConcurrency, + AsyncHandlerQueueCapacity: c.config.AsyncHandlerQueueCapacity, RetryEnabled: c.config.RetryEnabled, RetryConfigured: true, RetryMaxAttempts: c.config.RetryMaxAttempts, diff --git a/internal/core/client/client_test.go b/internal/core/client/client_test.go index 9a673ef..680aff0 100644 --- a/internal/core/client/client_test.go +++ b/internal/core/client/client_test.go @@ -438,6 +438,20 @@ func TestShouldUseDefaultAsyncHandlerMaxConcurrencyGivenNoOptionWhenNewClientCre assert.Equal(t, 256, c.config.AsyncHandlerMaxConcurrency) } +func TestShouldUseIndependentDefaultAsyncHandlerQueueCapacityGivenNoOptionWhenNewClientCreated(t *testing.T) { + c := NewClientWithOptions("localhost:4091", nil, WithMaxRequestQueueSize(7), WithAsyncHandlerMaxConcurrency(3)) + + require.NotNil(t, c.config) + assert.Equal(t, 1024, c.config.AsyncHandlerQueueCapacity) +} + +func TestShouldApplyAsyncHandlerQueueCapacityOptionGivenOverrideWhenNewClientWithOptionsCalled(t *testing.T) { + c := NewClientWithOptions("localhost:4091", nil, WithAsyncHandlerQueueCapacity(19)) + + require.NotNil(t, c.config) + assert.Equal(t, 19, c.config.AsyncHandlerQueueCapacity) +} + func TestShouldUseDefaultMaxInFlightRequestsGivenNoOptionWhenNewClientCreated(t *testing.T) { // Act c := NewClient("localhost:4091", nil) diff --git a/internal/core/connection/async_launcher_test.go b/internal/core/connection/async_launcher_test.go index fe5ae3d..8f1028a 100644 --- a/internal/core/connection/async_launcher_test.go +++ b/internal/core/connection/async_launcher_test.go @@ -14,7 +14,7 @@ import ( ) func TestShouldRejectAsyncHandlerLaunchGivenQueueFullWhenLaunchAsyncHandlerCalled(t *testing.T) { - conn := New(testkit.NewMockTransport(), Config{AsyncHandlerMaxConcurrency: 1}) + conn := New(testkit.NewMockTransport(), Config{AsyncHandlerMaxConcurrency: 1, AsyncHandlerQueueCapacity: 1}) t.Cleanup(func() { _ = conn.Close() }) @@ -67,7 +67,7 @@ func TestShouldRejectAsyncHandlerLaunchGivenQueueFullWhenLaunchAsyncHandlerCalle } func TestShouldExpireQueuedAsyncHandlerGivenTimeoutBeforeWorkerStartsWhenLaunchAsyncHandlerCalled(t *testing.T) { - conn := New(testkit.NewMockTransport(), Config{AsyncHandlerMaxConcurrency: 1}) + conn := New(testkit.NewMockTransport(), Config{AsyncHandlerMaxConcurrency: 1, AsyncHandlerQueueCapacity: 1}) t.Cleanup(func() { _ = conn.Close() }) @@ -111,6 +111,7 @@ func TestShouldEndQueuedAsyncHandlerSpanGivenShutdownWhenJobDrained(t *testing.T conn := New(testkit.NewMockTransport(), Config{ AsyncHandlerMaxConcurrency: 1, + AsyncHandlerQueueCapacity: 1, Tracer: tp.Tracer("fitz-go-async-test"), }) @@ -166,6 +167,7 @@ func TestShouldCancelReceivedAsyncHandlerJobGivenShutdownBeforeSlotAcquired(t *t conn := New(testkit.NewMockTransport(), Config{ AsyncHandlerMaxConcurrency: 1, + AsyncHandlerQueueCapacity: 1, Tracer: tp.Tracer("fitz-go-async-test"), }) t.Cleanup(func() { diff --git a/internal/core/connection/connection.go b/internal/core/connection/connection.go index e6e5626..ce77a95 100644 --- a/internal/core/connection/connection.go +++ b/internal/core/connection/connection.go @@ -101,6 +101,7 @@ type Connection struct { asyncSlotAcquireFailures metric.Int64Counter asyncHandlersActive metric.Int64UpDownCounter asyncSlotOccupancyMs metric.Int64Histogram + asyncHandlersSaturated metric.Int64Counter // Observability (optional) logger *slog.Logger @@ -126,6 +127,7 @@ type Config struct { MaxRequestQueueSize int // Default 1024 waiters beyond MaxInFlightRequests AsyncHandlerTimeout time.Duration // Default 30s for detached async handler spans AsyncHandlerMaxConcurrency int // Default 256 concurrent async handlers + AsyncHandlerQueueCapacity int // Default 1024 queued async handlers ReconnectEnabled bool ReconnectBackoff time.Duration RetryEnabled bool @@ -147,20 +149,21 @@ type Config struct { // DefaultConfig returns default configuration. func DefaultConfig() Config { return Config{ - AuthSettleDelay: 500 * time.Millisecond, - ReadTimeout: 30 * time.Second, - WriteTimeout: 10 * time.Second, - MaxInFlightRequests: 256, - MaxRequestQueueSize: 1024, - RetryEnabled: true, - RetryConfigured: true, - RetryMaxAttempts: 3, - RetryBackoff: 100 * time.Millisecond, - RetryMaxBackoff: time.Second, - HeartbeatEnabled: true, - HeartbeatConfigured: true, - HeartbeatInterval: 10 * time.Second, - HeartbeatTimeout: 30 * time.Second, + AuthSettleDelay: 500 * time.Millisecond, + ReadTimeout: 30 * time.Second, + WriteTimeout: 10 * time.Second, + MaxInFlightRequests: 256, + MaxRequestQueueSize: 1024, + AsyncHandlerQueueCapacity: 1024, + RetryEnabled: true, + RetryConfigured: true, + RetryMaxAttempts: 3, + RetryBackoff: 100 * time.Millisecond, + RetryMaxBackoff: time.Second, + HeartbeatEnabled: true, + HeartbeatConfigured: true, + HeartbeatInterval: 10 * time.Second, + HeartbeatTimeout: 30 * time.Second, } } @@ -201,6 +204,9 @@ func New(trans transport.Transport, cfg Config) *Connection { if cfg.AsyncHandlerMaxConcurrency <= 0 { cfg.AsyncHandlerMaxConcurrency = 256 } + if cfg.AsyncHandlerQueueCapacity <= 0 { + cfg.AsyncHandlerQueueCapacity = 1024 + } if !cfg.RetryConfigured { cfg.RetryEnabled = true } @@ -237,7 +243,7 @@ func New(trans transport.Transport, cfg Config) *Connection { requestSem: make(chan struct{}, cfg.MaxInFlightRequests), oneWaySem: make(chan struct{}, cfg.MaxInFlightRequests), asyncHandlerSem: make(chan struct{}, cfg.AsyncHandlerMaxConcurrency), - asyncHandlerJobs: make(chan asyncHandlerJob, cfg.AsyncHandlerMaxConcurrency), + asyncHandlerJobs: make(chan asyncHandlerJob, cfg.AsyncHandlerQueueCapacity), token: cfg.Token, authConfirmed: make(chan struct{}), mux: NewMultiplexer(), @@ -304,6 +310,33 @@ func (c *Connection) initAsyncHandlerMetrics() { ); err == nil { c.asyncSlotOccupancyMs = occupancy } + if saturated, err := c.meter.Int64Counter( + "fitz.async_handlers.saturated", + metric.WithDescription("Count of callback jobs rejected because the async handler queue is full"), + metric.WithUnit("{failure}"), + ); err == nil { + c.asyncHandlersSaturated = saturated + } + if active, err := c.meter.Int64ObservableGauge( + "fitz.async_handlers.active", + metric.WithDescription("Current number of executing async handlers"), + metric.WithUnit("{handler}"), + ); err == nil { + _, _ = c.meter.RegisterCallback(func(_ context.Context, observer metric.Observer) error { + observer.ObserveInt64(active, int64(len(c.asyncHandlerSem))) + return nil + }, active) + } + if queued, err := c.meter.Int64ObservableGauge( + "fitz.async_handlers.queued", + metric.WithDescription("Current number of callback jobs waiting in the async handler queue"), + metric.WithUnit("{handler}"), + ); err == nil { + _, _ = c.meter.RegisterCallback(func(_ context.Context, observer metric.Observer) error { + observer.ObserveInt64(queued, int64(len(c.asyncHandlerJobs))) + return nil + }, queued) + } if hist, err := c.meter.Int64Histogram( "fitz.request.duration", @@ -1114,6 +1147,15 @@ func (c *Connection) AsyncHandlerMaxConcurrency() int { return c.cfg.AsyncHandlerMaxConcurrency } +// AsyncHandlerQueueCapacity returns the configured number of callback jobs +// that may wait for an execution slot. +func (c *Connection) AsyncHandlerQueueCapacity() int { + if c == nil || c.cfg.AsyncHandlerQueueCapacity <= 0 { + return 1024 + } + return c.cfg.AsyncHandlerQueueCapacity +} + // MaxInFlightRequests returns the configured maximum number of concurrently // admitted outbound request operations. func (c *Connection) MaxInFlightRequests() int { @@ -1296,6 +1338,9 @@ func (c *Connection) LaunchAsyncHandler(parent context.Context, spanName string, return true default: queueErr := errors.New("async handler queue full") + if c.asyncHandlersSaturated != nil { + c.asyncHandlersSaturated.Add(parent, 1) + } span.RecordError(queueErr) span.SetStatus(codes.Error, queueErr.Error()) cancel() @@ -1601,7 +1646,8 @@ func IsTransientRetryable(err error) bool { coreerrors.RpcTimeout, coreerrors.RpcWorkerNotFound, coreerrors.RpcBackpressure, - coreerrors.RpcRouteNotRegistered: + coreerrors.RpcRouteNotRegistered, + coreerrors.ScheduleBackendError: return true } } diff --git a/internal/core/connection/connection_test.go b/internal/core/connection/connection_test.go index 9110131..00937c1 100644 --- a/internal/core/connection/connection_test.go +++ b/internal/core/connection/connection_test.go @@ -156,6 +156,26 @@ func TestShouldReturnConfiguredAsyncHandlerMaxConcurrencyGivenConfigWhenNewCalle assert.Equal(t, 8, conn.AsyncHandlerMaxConcurrency()) } +func TestShouldReturnIndependentDefaultAsyncHandlerQueueCapacityGivenUnsetConfigWhenNewCalled(t *testing.T) { + cfg := connection.DefaultConfig() + cfg.AsyncHandlerQueueCapacity = 0 + cfg.AsyncHandlerMaxConcurrency = 3 + cfg.MaxRequestQueueSize = 7 + + conn := connection.New(&testkit.MockTransport{}, cfg) + + assert.Equal(t, 1024, conn.AsyncHandlerQueueCapacity()) +} + +func TestShouldReturnConfiguredAsyncHandlerQueueCapacityGivenConfigWhenNewCalled(t *testing.T) { + cfg := connection.DefaultConfig() + cfg.AsyncHandlerQueueCapacity = 19 + + conn := connection.New(&testkit.MockTransport{}, cfg) + + assert.Equal(t, 19, conn.AsyncHandlerQueueCapacity()) +} + func TestShouldUseConfiguredMeterGivenConfigWhenNewCalled(t *testing.T) { // Arrange transport := &testkit.MockTransport{} @@ -631,6 +651,12 @@ func TestShouldClassifyRetryableGivenIsolationConflictWhenRetryPolicyEvaluated(t assert.True(t, connection.IsTransientRetryable(err)) } +func TestShouldClassifyRetryableGivenScheduleBackendErrorWhenRetryPolicyEvaluated(t *testing.T) { + err := coreerrors.NewDomainError(coreerrors.ScheduleBackendError, "backend busy") + + assert.True(t, connection.IsTransientRetryable(err)) +} + func TestShouldClassifyFatalGivenInvalidModeWhenRetryPolicyEvaluated(t *testing.T) { err := coreerrors.NewDomainError(coreerrors.KvInvalidMode, "invalid mode") diff --git a/internal/core/errors/errors.go b/internal/core/errors/errors.go index 9fbb908..8763ed4 100644 --- a/internal/core/errors/errors.go +++ b/internal/core/errors/errors.go @@ -8,9 +8,25 @@ package errors import ( + "errors" "fmt" ) +var ErrAsyncHandlerOverflow = errors.New("async handler queue overflow") + +type AsyncHandlerOverflowError struct { + Domain string + SubscriptionID uint64 +} + +func (e *AsyncHandlerOverflowError) Error() string { + return fmt.Sprintf("%s subscription %d: %s", e.Domain, e.SubscriptionID, ErrAsyncHandlerOverflow) +} + +func (e *AsyncHandlerOverflowError) Unwrap() error { + return ErrAsyncHandlerOverflow +} + // Error code ranges by domain (from Fitz server) const ( // KV Domain (1000-1099) @@ -78,6 +94,7 @@ const ( ScheduleInvalidSubscription = 7006 ScheduleSubscriptionLimit = 7007 ScheduleInvalidDeliveryMode = 7008 + ScheduleBackendError = 7010 ) // IsBackpressure returns true if the error code indicates backpressure @@ -207,6 +224,8 @@ func (e ErrorCode) String() string { return "schedule_subscription_limit" case ScheduleInvalidDeliveryMode: return "schedule_invalid_delivery_mode" + case ScheduleBackendError: + return "schedule_backend_error" default: return fmt.Sprintf("unknown_error_%d", e) diff --git a/internal/core/errors/errors_registry_test.go b/internal/core/errors/errors_registry_test.go index 200a720..b5b4d67 100644 --- a/internal/core/errors/errors_registry_test.go +++ b/internal/core/errors/errors_registry_test.go @@ -69,6 +69,7 @@ func TestErrorCodeRegistry(t *testing.T) { {"ScheduleInvalidSubscription", ScheduleInvalidSubscription, 7006}, {"ScheduleSubscriptionLimit", ScheduleSubscriptionLimit, 7007}, {"ScheduleInvalidDeliveryMode", ScheduleInvalidDeliveryMode, 7008}, + {"ScheduleBackendError", ScheduleBackendError, 7010}, } for _, tc := range cases { diff --git a/internal/core/subscriptions/registry.go b/internal/core/subscriptions/registry.go index c6cafe9..80a0746 100644 --- a/internal/core/subscriptions/registry.go +++ b/internal/core/subscriptions/registry.go @@ -23,6 +23,42 @@ type Registry[H any] struct { restoreBySubIDScratch map[uint64]*entry[H] } +// Completion reports the terminal state of a local subscription handler. +// The channel yields exactly one result: nil after an explicit unsubscribe, +// or the terminal error that caused the handler to be removed. +type Completion struct { + once sync.Once + done chan error +} + +func NewCompletion() *Completion { + return &Completion{done: make(chan error, 1)} +} + +func (c *Completion) Done() <-chan error { + if c == nil { + return nil + } + return c.done +} + +func (c *Completion) Complete(err error) { + if c == nil { + return + } + c.once.Do(func() { + c.done <- err + close(c.done) + }) +} + +type Registration[H any] struct { + Pattern string + HandlerID uint64 + Handler H + Completion *Completion +} + type pendingSubscribe struct { done chan struct{} } @@ -30,7 +66,7 @@ type pendingSubscribe struct { type entry[H any] struct { pattern string subID uint64 - handlers map[uint64]H + handlers map[uint64]Registration[H] } type restoreEntry struct { @@ -58,7 +94,9 @@ func (r *Registry[H]) Subscribe(pattern string, handler H, wireSubscribe func(st if existing, ok := r.byPattern[pattern]; ok { handlerID := r.nextHandler() - existing.handlers[handlerID] = handler + existing.handlers[handlerID] = Registration[H]{ + Pattern: pattern, HandlerID: handlerID, Handler: handler, Completion: NewCompletion(), + } r.mu.Unlock() return existing.subID, handlerID, nil } @@ -93,8 +131,10 @@ func (r *Registry[H]) Subscribe(pattern string, handler H, wireSubscribe func(st registered := &entry[H]{ pattern: pattern, subID: subID, - handlers: map[uint64]H{ - handlerID: handler, + handlers: map[uint64]Registration[H]{ + handlerID: { + Pattern: pattern, HandlerID: handlerID, Handler: handler, Completion: NewCompletion(), + }, }, } r.byPattern[pattern] = registered @@ -116,6 +156,11 @@ func (r *Registry[H]) Unsubscribe(pattern string, handlerID uint64) bool { return false } + handler, exists := registered.handlers[handlerID] + if !exists { + return false + } + handler.Completion.Complete(nil) delete(registered.handlers, handlerID) if len(registered.handlers) != 0 { return false @@ -136,12 +181,42 @@ func (r *Registry[H]) Handlers(subID uint64) []H { } handlers := make([]H, 0, len(registered.handlers)) - for _, handler := range registered.handlers { - handlers = append(handlers, handler) + for _, registration := range registered.handlers { + handlers = append(handlers, registration.Handler) } return handlers } +func (r *Registry[H]) Registrations(subID uint64) []Registration[H] { + r.mu.Lock() + defer r.mu.Unlock() + + registered, ok := r.bySubID[subID] + if !ok { + return nil + } + + registrations := make([]Registration[H], 0, len(registered.handlers)) + for _, registration := range registered.handlers { + registrations = append(registrations, registration) + } + return registrations +} + +func (r *Registry[H]) Completion(pattern string, handlerID uint64) *Completion { + r.mu.Lock() + defer r.mu.Unlock() + registered, ok := r.byPattern[pattern] + if !ok { + return nil + } + registration, ok := registered.handlers[handlerID] + if !ok { + return nil + } + return registration.Completion +} + func (r *Registry[H]) Restore(wireSubscribe func(string) (uint64, error), wireUnsubscribe func(string, uint64) error) error { r.mu.Lock() locked := true diff --git a/internal/core/subscriptions/registry_test.go b/internal/core/subscriptions/registry_test.go index 567ae2d..3367e8c 100644 --- a/internal/core/subscriptions/registry_test.go +++ b/internal/core/subscriptions/registry_test.go @@ -54,6 +54,26 @@ func TestShouldSendWireUnsubscribeOnlyForLastHandlerWhenUnsubscribeCalled(t *tes assert.Empty(t, registry.Handlers(subID)) } +func TestShouldCompleteLocalHandlerGivenUnsubscribeWhenMultipleHandlersShareWireSubscription(t *testing.T) { + registry := NewRegistry[string]() + _, firstHandlerID, err := registry.Subscribe("stream://realm/area/resource", "first", func(string) (uint64, error) { return 7, nil }) + require.NoError(t, err) + _, _, err = registry.Subscribe("stream://realm/area/resource", "second", func(string) (uint64, error) { return 7, nil }) + require.NoError(t, err) + completion := registry.Completion("stream://realm/area/resource", firstHandlerID) + + assert.False(t, registry.Unsubscribe("stream://realm/area/resource", firstHandlerID)) + assert.NoError(t, <-completion.Done()) +} + +func TestShouldReturnNoCompletionGivenUnknownHandler(t *testing.T) { + registry := NewRegistry[string]() + _, handlerID, err := registry.Subscribe("stream://realm/area/resource", "first", func(string) (uint64, error) { return 7, nil }) + require.NoError(t, err) + + assert.Nil(t, registry.Completion("stream://realm/area/resource", handlerID+1)) +} + func TestShouldPreserveHandlersGivenReconnectWhenRestoreCalled(t *testing.T) { registry := NewRegistry[string]() _, firstHandlerID, err := registry.Subscribe("queue://realm/area/resource", "first", func(string) (uint64, error) { diff --git a/internal/domains/kv/watch.go b/internal/domains/kv/watch.go index 9727b10..4b2bae2 100644 --- a/internal/domains/kv/watch.go +++ b/internal/domains/kv/watch.go @@ -9,6 +9,8 @@ import ( "github.com/cntryl/fitz-go/v2/internal/core/connection" "github.com/cntryl/fitz-go/v2/internal/core/encoding" + coreerrors "github.com/cntryl/fitz-go/v2/internal/core/errors" + "github.com/cntryl/fitz-go/v2/internal/core/subscriptions" "github.com/cntryl/fitz-go/v2/internal/core/types" "github.com/cntryl/fitz-go/v2/internal/protocol" "go.opentelemetry.io/otel/attribute" @@ -24,9 +26,10 @@ type ChangeNotification struct { type ChangeHandler func(context.Context, ChangeNotification) error type Subscription struct { - handlerID uint64 - pattern string - client *client + handlerID uint64 + pattern string + client *client + completion *subscriptions.Completion } func (s *Subscription) Unsubscribe() { @@ -35,6 +38,13 @@ func (s *Subscription) Unsubscribe() { } } +func (s *Subscription) Completion() <-chan error { + if s == nil || s.completion == nil { + return nil + } + return s.completion.Done() +} + // Subscribe registers a handler for exact KV routes matched by pattern. // Patterns use whole-segment * and ** wildcards, and notifications carry the concrete route. func (c *client) Subscribe(ctx context.Context, pattern string, handler ChangeHandler) (*Subscription, error) { @@ -58,7 +68,7 @@ func (c *client) Subscribe(ctx context.Context, pattern string, handler ChangeHa span.SetStatus(codes.Error, err.Error()) return nil, err } - return &Subscription{handlerID: handlerID, pattern: pattern, client: c}, nil + return &Subscription{handlerID: handlerID, pattern: pattern, client: c, completion: c.subscriptions.Completion(pattern, handlerID)}, nil } func (c *client) initNotifyHandler() { @@ -75,7 +85,8 @@ func (c *client) handleNotify(subID uint64, route string, payload []byte) { } notification := ChangeNotification{Route: route, MutationCount: binary.BigEndian.Uint64(payload)} conn := c.currentConnection() - for _, handler := range c.subscriptions.Handlers(subID) { + for _, registration := range c.subscriptions.Registrations(subID) { + handler := registration.Handler if !conn.LaunchAsyncHandler(conn.LifecycleContext(), "fitz.kv.handler", conn.AsyncHandlerTimeout(), func(handlerCtx context.Context, span trace.Span) { if err := handler(handlerCtx, notification); err != nil { span.RecordError(err) @@ -85,7 +96,8 @@ func (c *client) handleNotify(subID uint64, route string, payload []byte) { attribute.Int64("fitz.subscription_id", int64(subID)), attribute.String("fitz.route", route), )) { - return + registration.Completion.Complete(&coreerrors.AsyncHandlerOverflowError{Domain: "kv", SubscriptionID: subID}) + go c.unsubscribe(&Subscription{handlerID: registration.HandlerID, pattern: registration.Pattern, client: c, completion: registration.Completion}) } } } diff --git a/internal/domains/lease/lease.go b/internal/domains/lease/lease.go index e122f99..bcde83f 100644 --- a/internal/domains/lease/lease.go +++ b/internal/domains/lease/lease.go @@ -14,6 +14,7 @@ import ( "github.com/cntryl/fitz-go/v2/internal/core/connection" coreerrors "github.com/cntryl/fitz-go/v2/internal/core/errors" "github.com/cntryl/fitz-go/v2/internal/core/reconnect" + "github.com/cntryl/fitz-go/v2/internal/core/subscriptions" "github.com/cntryl/fitz-go/v2/internal/core/types" "github.com/cntryl/fitz-go/v2/internal/protocol" "go.opentelemetry.io/otel/attribute" @@ -148,10 +149,11 @@ type ChangeHandler func(ctx context.Context, notif ChangeNotification) error // Subscription represents an active lease change subscription. // Call Unsubscribe to stop receiving and release the subscription. type Subscription struct { - subID uint64 - route string - client *client - handler ChangeHandler + subID uint64 + route string + client *client + handler ChangeHandler + completion *subscriptions.Completion } // Unsubscribe removes this subscription. @@ -161,6 +163,13 @@ func (s *Subscription) Unsubscribe() { } } +func (s *Subscription) Completion() <-chan error { + if s == nil || s.completion == nil { + return nil + } + return s.completion.Done() +} + // Client is the Lease domain client interface. type Client interface { // Acquire attempts to acquire a lease on the given route. @@ -515,6 +524,8 @@ func (c *client) handleNotify(subID uint64, route string, payload []byte) { attribute.Int64("fitz.subscription_id", int64(subID)), attribute.String("fitz.route", route), )) { + sub.completion.Complete(&coreerrors.AsyncHandlerOverflowError{Domain: "lease", SubscriptionID: subID}) + go c.unsubscribe(sub) if log := c.conn.Logger(); log != nil { log.Warn("lease notify handler dropped", "route", route, "sub_id", subID, "reason", "async handler queue full") } @@ -548,8 +559,13 @@ func (c *client) Subscribe(ctx context.Context, route string, handler ChangeHand // unsubscribe removes a subscription. func (c *client) unsubscribe(sub *Subscription) { c.mu.Lock() + if _, exists := c.subscriptions[sub.subID]; !exists { + c.mu.Unlock() + return + } delete(c.subscriptions, sub.subID) c.mu.Unlock() + sub.completion.Complete(nil) c.conn.AddSubscriptions(-1) // Send UNSUBSCRIBE to server (best-effort, ignore errors). @@ -576,13 +592,13 @@ func (c *client) RestoreSubscriptions(ctx context.Context) error { c.mu.RLock() snapshot := make([]*Subscription, 0, len(c.subscriptions)) for _, sub := range c.subscriptions { - snapshot = append(snapshot, &Subscription{route: sub.route, handler: sub.handler, client: c}) + snapshot = append(snapshot, &Subscription{route: sub.route, handler: sub.handler, client: c, completion: sub.completion}) } c.mu.RUnlock() restored := make(map[uint64]*Subscription, len(snapshot)) for _, sub := range snapshot { - restoredSub, err := c.restoreSubscribe(ctx, sub.route, sub.handler) + restoredSub, err := c.restoreSubscribe(ctx, sub.route, sub.handler, sub.completion) if err != nil { for _, restoredSub := range restored { c.rollbackRestoredSubscription(restoredSub) @@ -598,7 +614,7 @@ func (c *client) RestoreSubscriptions(ctx context.Context) error { return nil } -func (c *client) restoreSubscribe(ctx context.Context, route string, handler ChangeHandler) (*Subscription, error) { +func (c *client) restoreSubscribe(ctx context.Context, route string, handler ChangeHandler, completion *subscriptions.Completion) (*Subscription, error) { resp, err := c.conn.SendRequestWithWriter(ctx, protocol.MessageTypeLeaseSubscribe, subscribePayloadWriter(route)) if err != nil { return nil, fmt.Errorf("SUBSCRIBE request failed: %w", err) @@ -622,10 +638,11 @@ func (c *client) restoreSubscribe(ctx context.Context, route string, handler Cha c.conn.AddSubscriptions(1) return &Subscription{ - subID: subID, - route: route, - client: c, - handler: handler, + subID: subID, + route: route, + client: c, + handler: handler, + completion: completion, }, nil } @@ -669,10 +686,11 @@ func (c *client) subscribe(ctx context.Context, route string, handler ChangeHand c.conn.AddSubscriptions(1) sub := &Subscription{ - subID: subID, - route: route, - client: c, - handler: handler, + subID: subID, + route: route, + client: c, + handler: handler, + completion: subscriptions.NewCompletion(), } c.mu.Lock() c.subscriptions[subID] = sub diff --git a/internal/domains/notice/notice.go b/internal/domains/notice/notice.go index 973d4de..ca1e077 100644 --- a/internal/domains/notice/notice.go +++ b/internal/domains/notice/notice.go @@ -10,6 +10,7 @@ import ( "sync/atomic" "github.com/cntryl/fitz-go/v2/internal/core/connection" + coreerrors "github.com/cntryl/fitz-go/v2/internal/core/errors" "github.com/cntryl/fitz-go/v2/internal/core/reconnect" "github.com/cntryl/fitz-go/v2/internal/core/subscriptions" "github.com/cntryl/fitz-go/v2/internal/core/types" @@ -31,10 +32,11 @@ type NoticeHandler func(ctx context.Context, msg NoticeMsg) error // Subscription represents an active notice subscription. // Call Unsubscribe to stop receiving and release the subscription. type Subscription struct { - subID uint64 - handlerID uint64 - route string - client *client + subID uint64 + handlerID uint64 + route string + client *client + completion *subscriptions.Completion } // Unsubscribe removes this subscription. @@ -44,6 +46,13 @@ func (s *Subscription) Unsubscribe() { } } +func (s *Subscription) Completion() <-chan error { + if s == nil || s.completion == nil { + return nil + } + return s.completion.Done() +} + // Client is the Notice domain client interface. type Client interface { // Publish sends a notification to a route (fire-and-forget). @@ -93,13 +102,14 @@ func (c *client) initNotifyHandler() { // handleNotify is called by the mux when a NOTIFY (504) frame arrives. func (c *client) handleNotify(subID uint64, route string, payload []byte) { - handlers := c.subscriptions.Handlers(subID) - if len(handlers) == 0 { + registrations := c.subscriptions.Registrations(subID) + if len(registrations) == 0 { return } lifecycleCtx := c.currentConn().LifecycleContext() - for _, handler := range handlers { + for _, registration := range registrations { + handler := registration.Handler msg := NoticeMsg{ Route: route, Body: append([]byte(nil), payload...), @@ -116,6 +126,8 @@ func (c *client) handleNotify(subID uint64, route string, payload []byte) { attribute.Int64("fitz.subscription_id", int64(subID)), attribute.String("fitz.route", route), )) { + registration.Completion.Complete(&coreerrors.AsyncHandlerOverflowError{Domain: "notice", SubscriptionID: subID}) + go c.unsubscribe(&Subscription{subID: subID, handlerID: registration.HandlerID, route: registration.Pattern, client: c, completion: registration.Completion}) if log := c.currentConn().Logger(); log != nil { log.Warn("notice handler dropped", "route", route, "sub_id", subID, "reason", "async handler queue full") } @@ -168,10 +180,11 @@ func (c *client) Subscribe(ctx context.Context, pattern string, handler NoticeHa return nil, err } return &Subscription{ - subID: subID, - handlerID: handlerID, - route: pattern, - client: c, + subID: subID, + handlerID: handlerID, + route: pattern, + client: c, + completion: c.subscriptions.Completion(pattern, handlerID), }, nil } diff --git a/internal/domains/notice/notice_test.go b/internal/domains/notice/notice_test.go index bdbf666..db5a05f 100644 --- a/internal/domains/notice/notice_test.go +++ b/internal/domains/notice/notice_test.go @@ -12,10 +12,51 @@ import ( coreerrors "github.com/cntryl/fitz-go/v2/internal/core/errors" "github.com/cntryl/fitz-go/v2/internal/core/subscriptions" "github.com/cntryl/fitz-go/v2/internal/protocol" + "github.com/cntryl/fitz-go/v2/internal/testkit" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +func TestShouldTerminateSubscriptionWithTypedErrorGivenAsyncHandlerQueueOverflow(t *testing.T) { + conn := connection.New(testkit.NewMockTransport(), connection.Config{ + AsyncHandlerMaxConcurrency: 1, + AsyncHandlerQueueCapacity: 1, + }) + t.Cleanup(func() { _ = conn.Close() }) + c := NewClient(conn).(*client) + started := make(chan struct{}) + release := make(chan struct{}) + var calls sync.Once + subID, handlerID, err := c.subscriptions.Subscribe("notice://realm/area/resource", func(context.Context, NoticeMsg) error { + calls.Do(func() { close(started) }) + <-release + return nil + }, func(string) (uint64, error) { return 42, nil }) + require.NoError(t, err) + completion := c.subscriptions.Completion("notice://realm/area/resource", handlerID) + + c.handleNotify(subID, "notice://realm/area/resource", []byte("first")) + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("first notice handler did not start") + } + c.handleNotify(subID, "notice://realm/area/resource", []byte("queued")) + c.handleNotify(subID, "notice://realm/area/resource", []byte("overflow")) + + select { + case terminalErr := <-completion.Done(): + require.ErrorIs(t, terminalErr, coreerrors.ErrAsyncHandlerOverflow) + var overflow *coreerrors.AsyncHandlerOverflowError + require.ErrorAs(t, terminalErr, &overflow) + assert.Equal(t, "notice", overflow.Domain) + assert.Equal(t, uint64(42), overflow.SubscriptionID) + case <-time.After(time.Second): + t.Fatal("subscription completion did not report queue overflow") + } + close(release) +} + type scriptedRestoreTransport struct { mu sync.Mutex written [][]byte diff --git a/internal/domains/queue/queue.go b/internal/domains/queue/queue.go index 0630f1c..0e85f45 100644 --- a/internal/domains/queue/queue.go +++ b/internal/domains/queue/queue.go @@ -12,6 +12,7 @@ import ( "sync/atomic" "github.com/cntryl/fitz-go/v2/internal/core/connection" + coreerrors "github.com/cntryl/fitz-go/v2/internal/core/errors" "github.com/cntryl/fitz-go/v2/internal/core/reconnect" "github.com/cntryl/fitz-go/v2/internal/core/subscriptions" "github.com/cntryl/fitz-go/v2/internal/core/types" @@ -79,10 +80,11 @@ func WithWaitSeconds(waitSeconds uint64) ReserveOption { // Subscription represents an active queue availability subscription. // Call Unsubscribe to stop receiving and release the subscription. type Subscription struct { - subID uint64 - handlerID uint64 - pattern string - client *client + subID uint64 + handlerID uint64 + pattern string + client *client + completion *subscriptions.Completion } // Unsubscribe removes this subscription. @@ -92,6 +94,13 @@ func (s *Subscription) Unsubscribe() { } } +func (s *Subscription) Completion() <-chan error { + if s == nil || s.completion == nil { + return nil + } + return s.completion.Done() +} + // Extend extends the lease on this queue item. func (q *QueueItem) Extend(ctx context.Context, leaseSecs uint64) error { ctx, span := q.conn.Tracer().Start(ctx, "fitz.queue.Extend", trace.WithAttributes( @@ -399,8 +408,8 @@ func (c *client) handleNotify(subID uint64, route string, payload []byte) { if len(payload) != 24 { return } - handlers := c.subscriptions.Handlers(subID) - if len(handlers) == 0 { + registrations := c.subscriptions.Registrations(subID) + if len(registrations) == 0 { return } @@ -412,7 +421,8 @@ func (c *client) handleNotify(subID uint64, route string, payload []byte) { } lifecycleCtx := c.conn.LifecycleContext() - for _, handler := range handlers { + for _, registration := range registrations { + handler := registration.Handler if !c.conn.LaunchAsyncHandler(lifecycleCtx, "fitz.queue.handler", c.conn.AsyncHandlerTimeout(), func(handlerCtx context.Context, span trace.Span) { if err := handler(handlerCtx, notif); err != nil { span.RecordError(err) @@ -425,6 +435,8 @@ func (c *client) handleNotify(subID uint64, route string, payload []byte) { attribute.Int64("fitz.subscription_id", int64(subID)), attribute.String("fitz.route", route), )) { + registration.Completion.Complete(&coreerrors.AsyncHandlerOverflowError{Domain: "queue", SubscriptionID: subID}) + go c.unsubscribe(&Subscription{subID: subID, handlerID: registration.HandlerID, pattern: registration.Pattern, client: c, completion: registration.Completion}) if log := c.conn.Logger(); log != nil { log.Warn("queue notify handler dropped", "route", route, "sub_id", subID, "reason", "async handler queue full") } @@ -456,10 +468,11 @@ func (c *client) Subscribe(ctx context.Context, pattern string, handler Availabi return nil, err } return &Subscription{ - subID: subID, - handlerID: handlerID, - pattern: pattern, - client: c, + subID: subID, + handlerID: handlerID, + pattern: pattern, + client: c, + completion: c.subscriptions.Completion(pattern, handlerID), }, nil } diff --git a/internal/domains/schedule/schedule.go b/internal/domains/schedule/schedule.go index 1217069..fe888b5 100644 --- a/internal/domains/schedule/schedule.go +++ b/internal/domains/schedule/schedule.go @@ -12,6 +12,7 @@ import ( "sync/atomic" "github.com/cntryl/fitz-go/v2/internal/core/connection" + coreerrors "github.com/cntryl/fitz-go/v2/internal/core/errors" "github.com/cntryl/fitz-go/v2/internal/core/reconnect" "github.com/cntryl/fitz-go/v2/internal/core/subscriptions" "github.com/cntryl/fitz-go/v2/internal/core/types" @@ -54,10 +55,11 @@ type ScheduleHandler func(ctx context.Context, n Notification) error // Subscription represents an active subscription to schedule fire notifications. // Call Unsubscribe to stop receiving notifications. type Subscription struct { - subID uint64 - handlerID uint64 - pattern string - client *client + subID uint64 + handlerID uint64 + pattern string + client *client + completion *subscriptions.Completion } // Unsubscribe stops receiving schedule fire notifications for this subscription. @@ -67,6 +69,13 @@ func (s *Subscription) Unsubscribe() { } } +func (s *Subscription) Completion() <-chan error { + if s == nil || s.completion == nil { + return nil + } + return s.completion.Done() +} + // Client is the Schedule domain client interface. type Client interface { // Create creates a cron-based schedule at the given route (upsert per spec). Returns the schedule route (identity). @@ -109,13 +118,14 @@ func (c *client) initScheduleNotifyHandler() { func (c *client) handleScheduleNotify(subID uint64, route string, payload []byte) { conn := c.currentConn() - handlers := c.subscriptions.Handlers(subID) - if len(handlers) == 0 { + registrations := c.subscriptions.Registrations(subID) + if len(registrations) == 0 { return } lifecycleCtx := conn.LifecycleContext() - for _, handler := range handlers { + for _, registration := range registrations { + handler := registration.Handler msg := Notification{ Route: route, Payload: append([]byte(nil), payload...), @@ -132,6 +142,8 @@ func (c *client) handleScheduleNotify(subID uint64, route string, payload []byte attribute.Int64("fitz.subscription_id", int64(subID)), attribute.String("fitz.route", route), )) { + registration.Completion.Complete(&coreerrors.AsyncHandlerOverflowError{Domain: "schedule", SubscriptionID: subID}) + go c.unsubscribe(&Subscription{subID: subID, handlerID: registration.HandlerID, pattern: registration.Pattern, client: c, completion: registration.Completion}) if log := conn.Logger(); log != nil { log.Warn("schedule notify handler dropped", "sub_id", subID, "reason", "async handler queue full") } @@ -394,10 +406,11 @@ func (c *client) Subscribe(ctx context.Context, pattern string, handler Schedule return nil, err } return &Subscription{ - subID: subID, - handlerID: handlerID, - pattern: pattern, - client: c, + subID: subID, + handlerID: handlerID, + pattern: pattern, + client: c, + completion: c.subscriptions.Completion(pattern, handlerID), }, nil } diff --git a/internal/domains/stream/stream.go b/internal/domains/stream/stream.go index cb0ce82..691a683 100644 --- a/internal/domains/stream/stream.go +++ b/internal/domains/stream/stream.go @@ -12,6 +12,7 @@ import ( "sync/atomic" "github.com/cntryl/fitz-go/v2/internal/core/connection" + coreerrors "github.com/cntryl/fitz-go/v2/internal/core/errors" "github.com/cntryl/fitz-go/v2/internal/core/iter" "github.com/cntryl/fitz-go/v2/internal/core/reconnect" "github.com/cntryl/fitz-go/v2/internal/core/subscriptions" @@ -42,6 +43,10 @@ func parsePlainStreamResponse(payload []byte) (bool, []byte, error) { return false, nil, errors.New(message) } +func parseStreamReadResponse(payload []byte) (bool, []byte, error) { + return connection.ParseStandardResponse(payload) +} + // Record represents a single stream record. type Record struct { Route string @@ -125,10 +130,11 @@ type CommitHandler func(context.Context, CommitNotification) error // Subscription represents a stream subscription. type Subscription struct { - subID uint64 - handlerID uint64 - pattern string - client *client + subID uint64 + handlerID uint64 + pattern string + client *client + completion *subscriptions.Completion } // Unsubscribe removes the subscription. @@ -138,6 +144,13 @@ func (sub *Subscription) Unsubscribe() { } } +func (sub *Subscription) Completion() <-chan error { + if sub == nil || sub.completion == nil { + return nil + } + return sub.completion.Done() +} + // StreamSession is a write session for appending to a stream. // Obtained from Begin; use Append, then Commit or Rollback. // Expected offset (OCC) is provided on each Append call and tracked by the session/server. @@ -406,7 +419,7 @@ func (c *client) ReadPage(ctx context.Context, route string, fromOffset uint64, return fmt.Errorf("read request failed: %w", err) } - success, remaining, err := parsePlainStreamResponse(resp) + success, remaining, err := parseStreamReadResponse(resp) if err != nil { return fmt.Errorf("read failed: %w", mapStreamError(err)) } @@ -455,7 +468,7 @@ func (c *client) Peek(ctx context.Context, route string) (*Record, error) { return fmt.Errorf("peek request failed: %w", err) } - success, remaining, err := connection.ParseStandardResponse(resp) + success, remaining, err := parsePlainStreamResponse(resp) if err != nil { return fmt.Errorf("peek failed: %w", mapStreamError(err)) } @@ -885,8 +898,8 @@ func (c *client) initNotifyHandler() { // handleNotify is called by the mux when a NOTIFY (609) frame arrives. func (c *client) handleNotify(subID uint64, route string, payload []byte) { - handlers := c.subscriptions.Handlers(subID) - if len(handlers) == 0 { + registrations := c.subscriptions.Registrations(subID) + if len(registrations) == 0 { return } @@ -917,7 +930,8 @@ func (c *client) handleNotify(subID uint64, route string, payload []byte) { } lifecycleCtx := c.conn.LifecycleContext() - for _, handler := range handlers { + for _, registration := range registrations { + handler := registration.Handler if !c.conn.LaunchAsyncHandler(lifecycleCtx, "fitz.stream.handler", c.conn.AsyncHandlerTimeout(), func(handlerCtx context.Context, span trace.Span) { if err := handler(handlerCtx, notif); err != nil { span.RecordError(err) @@ -930,6 +944,8 @@ func (c *client) handleNotify(subID uint64, route string, payload []byte) { attribute.Int64("fitz.subscription_id", int64(subID)), attribute.String("fitz.route", route), )) { + registration.Completion.Complete(&coreerrors.AsyncHandlerOverflowError{Domain: "stream", SubscriptionID: subID}) + go c.unsubscribe(&Subscription{subID: subID, handlerID: registration.HandlerID, pattern: registration.Pattern, client: c, completion: registration.Completion}) if log := c.conn.Logger(); log != nil { log.Warn("stream notify handler dropped", "route", route, "sub_id", subID, "reason", "async handler queue full") } @@ -961,10 +977,11 @@ func (c *client) Subscribe(ctx context.Context, pattern string, handler CommitHa return nil, err } return &Subscription{ - subID: subID, - handlerID: handlerID, - pattern: pattern, - client: c, + subID: subID, + handlerID: handlerID, + pattern: pattern, + client: c, + completion: c.subscriptions.Completion(pattern, handlerID), }, nil } diff --git a/internal/domains/stream/stream_test.go b/internal/domains/stream/stream_test.go index d87e6fb..a2fbd7f 100644 --- a/internal/domains/stream/stream_test.go +++ b/internal/domains/stream/stream_test.go @@ -2,6 +2,7 @@ package stream import ( "context" + "encoding/binary" "errors" "io" "sync" @@ -13,6 +14,38 @@ import ( "github.com/stretchr/testify/require" ) +func TestShouldDecodePlainErrorMessageGivenLastResponseWhenParsePlainStreamResponseCalled(t *testing.T) { + message := "last failed without a numeric READ code" + payload := make([]byte, 5+len(message)) + payload[0] = 1 + binary.BigEndian.PutUint32(payload[1:5], uint32(len(message))) + copy(payload[5:], message) + + success, remaining, err := parsePlainStreamResponse(payload) + + assert.False(t, success) + assert.Nil(t, remaining) + assert.EqualError(t, err, message) +} + +func TestShouldPreserveCodedErrorGivenReadResponseWhenParseStreamReadResponseCalled(t *testing.T) { + message := "stream read failed" + payload := make([]byte, 9+len(message)) + payload[0] = 1 + binary.BigEndian.PutUint32(payload[1:5], uint32(coreerrors.StreamResourceNotFound)) + binary.BigEndian.PutUint32(payload[5:9], uint32(len(message))) + copy(payload[9:], message) + + success, remaining, err := parseStreamReadResponse(payload) + + assert.False(t, success) + assert.Nil(t, remaining) + var domainErr *coreerrors.DomainError + require.ErrorAs(t, err, &domainErr) + assert.Equal(t, coreerrors.ErrorCode(coreerrors.StreamResourceNotFound), domainErr.Code) + assert.Equal(t, message, domainErr.Message) +} + type staleStreamTransport struct { closed chan struct{} closeOnce sync.Once