diff --git a/pool.go b/pool.go index 2004a52..168527a 100644 --- a/pool.go +++ b/pool.go @@ -28,12 +28,14 @@ type PoolEventKind string const ( EventConnected PoolEventKind = "connected" EventDisconnected PoolEventKind = "disconnected" + EventDialFailed PoolEventKind = "dial_failed" ) type PoolEvent struct { ID string Kind PoolEventKind At time.Time + Err error } type PoolStats struct { diff --git a/worker.go b/worker.go index 060250c..dd854f7 100644 --- a/worker.go +++ b/worker.go @@ -311,8 +311,15 @@ func (w *DefaultWorker) spawnDialer( w.logger.Debug("session: dialing") } + onDialError := func(err error) { + if dialCtx.Err() != nil { + return + } + pool.Events <- PoolEvent{ID: w.id, Kind: EventDialFailed, Err: err, At: time.Now()} + } + go func() { - conn, err := connect(w.id, dialCtx, pool, w.handler) + conn, err := connect(w.id, dialCtx, pool, w.handler, onDialError) if err != nil { return @@ -333,13 +340,14 @@ func connect( ctx context.Context, pool PoolPlugin, handler slog.Handler, + onDialError func(error), ) (*transport.Connection, error) { cc := pool.ConnectionConfig conn, err := transport.NewConnection(ctx, id, &cc, handler) if err != nil { return nil, err } - return conn, conn.Connect(ctx, nil) + return conn, conn.Connect(ctx, onDialError) } // ---------------------------/ diff --git a/worker_test.go b/worker_test.go index fc3d1ab..9524e9a 100644 --- a/worker_test.go +++ b/worker_test.go @@ -103,12 +103,12 @@ func TestWorkerSession(t *testing.T) { honeybeetest.Never(t, func() bool { select { - case <-events: - return true + case e := <-events: + return e.Kind == EventConnected || e.Kind == EventDisconnected default: return false } - }, "expected no events when dial fails") + }, "expected no connected/disconnected events when dial fails") // worker goroutine is still running assert.False(t, func() bool { @@ -639,6 +639,161 @@ func TestWorkerSession(t *testing.T) { } }, "expected wg to drain after parent cancel") }) + + t.Run("EventDialFailed emitted with correct ID, non-nil Err, and non-zero At when dialer fails", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + w := makeWorker(t, ctx, cancel) + _, events, pool := makeWorkerContext(t) + + cc, _ := transport.NewConnectionConfig(transport.WithRetryDisabled()) + pool.ConnectionConfig = *cc + dialErr := errors.New("connection refused") + pool.ConnectionConfig.Dialer = &honeybeetest.MockDialer{ + DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { + return nil, nil, dialErr + }, + } + + var wg sync.WaitGroup + wg.Go(func() { w.Start(pool) }) + + honeybeetest.Eventually(t, func() bool { + select { + case e := <-events: + return e.Kind == EventDialFailed && + e.ID == w.id && + e.Err != nil && + !e.At.IsZero() + default: + return false + } + }, "expected EventDialFailed") + }) + + t.Run("no EventDialFailed when first dial succeeds", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + w := makeWorker(t, ctx, cancel) + _, events, pool := makeWorkerContext(t) + mockSocket := honeybeetest.NewMockSocket() + pool.ConnectionConfig.Dialer = mockDialer(mockSocket) + + var wg sync.WaitGroup + wg.Go(func() { w.Start(pool) }) + + honeybeetest.Eventually(t, func() bool { + select { + case e := <-events: + return e.Kind == EventConnected + default: + return false + } + }, "expected EventConnected") + + // drain any remaining events and assert none are EventDialFailed + honeybeetest.Never(t, func() bool { + select { + case e := <-events: + return e.Kind == EventDialFailed + default: + return false + } + }, "expected no EventDialFailed when first dial succeeds") + }) + + t.Run("no EventDialFailed when worker is stopped mid-dial", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + w := makeWorker(t, ctx, cancel) + _, events, pool := makeWorkerContext(t) + + cc, _ := transport.NewConnectionConfig(transport.WithRetryDisabled()) + pool.ConnectionConfig = *cc + pool.ConnectionConfig.Dialer = &honeybeetest.MockDialer{ + DialContextFunc: func(dialCtx context.Context, _ string, _ http.Header) (types.Socket, *http.Response, error) { + <-dialCtx.Done() + return nil, nil, dialCtx.Err() + }, + } + + var wg sync.WaitGroup + wg.Go(func() { w.Start(pool) }) + + w.Stop() + + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + honeybeetest.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, "expected worker to exit after Stop") + + honeybeetest.Never(t, func() bool { + select { + case e := <-events: + return e.Kind == EventDialFailed + default: + return false + } + }, "expected no EventDialFailed when stopped mid-dial") + }) + + t.Run("no EventDialFailed when keepalive replaces an in-progress dial", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + config, _ := NewWorkerConfig( + WithReconnectDelay(0), + WithKeepaliveTimeout(20*time.Millisecond), + ) + w := &DefaultWorker{ + ctx: ctx, + cancel: cancel, + id: "wss://test", + config: config, + sendHeartbeat: make(chan struct{}), + processedCount: &atomic.Uint64{}, + outgoingCount: &atomic.Uint64{}, + restartCount: &atomic.Uint64{}, + } + _, events, pool := makeWorkerContext(t) + + var dialCount atomic.Uint64 + cc, _ := transport.NewConnectionConfig(transport.WithRetryDisabled()) + pool.ConnectionConfig = *cc + pool.ConnectionConfig.Dialer = &honeybeetest.MockDialer{ + DialContextFunc: func(dialCtx context.Context, _ string, _ http.Header) (types.Socket, *http.Response, error) { + dialCount.Add(1) + <-dialCtx.Done() + return nil, nil, dialCtx.Err() + }, + } + + var wg sync.WaitGroup + wg.Go(func() { w.Start(pool) }) + + // keepalive fires after 20ms, cancelling the first dial and spawning a second + honeybeetest.Eventually(t, func() bool { + return dialCount.Load() >= 2 + }, "expected at least two dial attempts after keepalive fired") + + honeybeetest.Never(t, func() bool { + select { + case e := <-events: + return e.Kind == EventDialFailed + default: + return false + } + }, "expected no EventDialFailed when keepalive replaces dial") + }) } func TestWorkerSend(t *testing.T) {