feat: surface subscribeToStream polling errors

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);

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');

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;