diff --git a/src/events.ts b/src/events.ts index 3167430..1e9848f 100644 --- a/src/events.ts +++ b/src/events.ts @@ -111,7 +111,15 @@ export function subscribeToStream( } } catch (err) { // Swallow polling errors; the subscription continues - console.warn('[conduit-sdk] event polling error:', err); + const error = err instanceof Error ? err : new Error(String(err)); + console.warn('[conduit-sdk] event polling error:', error); + + // A consumer error handler must not stop future polling. + try { + handlers.onError?.(error); + } catch (handlerError) { + console.warn('[conduit-sdk] event polling onError handler error:', handlerError); + } } if (!stopped) timer = setTimeout(poll, pollInterval); diff --git a/src/tests/events-subscribe.test.ts b/src/tests/events-subscribe.test.ts index 4427883..9af8829 100644 --- a/src/tests/events-subscribe.test.ts +++ b/src/tests/events-subscribe.test.ts @@ -122,6 +122,58 @@ describe('subscribeToStream', () => { warn.mockRestore(); }); + it('surfaces polling errors through onError', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const pollingError = new Error('rpc unavailable'); + const onError = vi.fn(); + mockGetEvents.mockRejectedValueOnce(pollingError).mockResolvedValue({ events: [] }); + const { subscribeToStream } = await import('../events.js'); + + const sub = subscribeToStream('http://localhost:8000', 'CSTREAM', { + onError, + pollInterval: 1000, + }); + await vi.waitFor(() => expect(onError).toHaveBeenCalledWith(pollingError)); + + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(mockGetEvents).toHaveBeenCalledTimes(2)); + + sub.unsubscribe(); + warn.mockRestore(); + }); + + it('normalizes non-Error polling failures before calling onError', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const onError = vi.fn(); + mockGetEvents.mockRejectedValueOnce('rpc unavailable'); + const { subscribeToStream } = await import('../events.js'); + + const sub = subscribeToStream('http://localhost:8000', 'CSTREAM', { onError }); + await vi.waitFor(() => expect(onError).toHaveBeenCalledTimes(1)); + + expect(onError.mock.calls[0]?.[0]).toEqual(new Error('rpc unavailable')); + sub.unsubscribe(); + warn.mockRestore(); + }); + + it('keeps polling when onError itself throws', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + mockGetEvents.mockRejectedValueOnce(new Error('rpc unavailable')).mockResolvedValue({ events: [] }); + const { subscribeToStream } = await import('../events.js'); + + const sub = subscribeToStream('http://localhost:8000', 'CSTREAM', { + onError: () => { throw new Error('consumer handler failed'); }, + pollInterval: 1000, + }); + await vi.waitFor(() => expect(warn).toHaveBeenCalledTimes(2)); + + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(mockGetEvents).toHaveBeenCalledTimes(2)); + + sub.unsubscribe(); + warn.mockRestore(); + }); + it('unsubscribe stops further polling', async () => { mockGetEvents.mockResolvedValue({ events: [] }); const { subscribeToStream } = await import('../events.js'); diff --git a/src/types/index.ts b/src/types/index.ts index 13b6e53..d55d509 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -132,6 +132,7 @@ export interface StreamEventHandlers { onResume?: (e: ResumeEvent) => void; onTopUp?: (e: TopUpEvent) => void; onClawback?: (e: ClawbackEvent) => void; + /** Called when an event polling request fails. Polling continues afterward. */ onError?: (error: Error) => void; /** Polling interval in ms; default 5000 */ pollInterval?: number;