Add withSignal() to ReactiveStreamStore for per-connection cancellation

Adds `store.withSignal(signal).connect()` to `ReactiveStreamStore`, mirroring the action store's per-dispatch `withSignal()` pattern — callers attach a per-connection signal at the call site instead of baking one into the store's construction. Drops the construction-time `abortSignal` option on `createReactiveStoreFromDataPublisherFactory`, `createReactiveStoreWithInitialValueAndSlotTracking`, and `PendingRpcSubscriptionsRequest.reactiveStore()`. The duck-type `ReactiveStreamSource<T>.reactiveStore()` is now parameter-less, mirroring `ReactiveActionSource<T>.reactiveStore()`.

`store.withSignal(signal)` returns a thin wrapper exposing `connect()`. Each call composes the caller-provided signal with the per-connection inner controller via `AbortSignal.any` — aborting either tears down the active connection. Aborting the caller-provided signal surfaces the abort reason on state as `{ status: 'error' }`; supersession via the internal controller (a newer `connect()` or `reset()`) stays silent so the newer call owns state. Use cases:

- Per-connection timeout: `store.withSignal(AbortSignal.timeout(30_000)).connect()` mints a fresh clock per attempt.
- Permanent kill switch: hold one `AbortController`, bind the wrapper once (\`const killable = store.withSignal(killCtrl.signal)\`), and use \`killable.connect()\` everywhere. After \`killCtrl.abort()\`, every future call short-circuits to error. Bare \`store.connect()\` calls bypass the bound signal — same discipline contract as the action store's bind-once pattern.

\`createDataPublisher\` is widened from \`() => Promise<DataPublisher>\` to \`(signal: AbortSignal) => Promise<DataPublisher>\`. The store passes the composed per-connection signal to the factory so the underlying transport can stop on per-connection abort, not just the stream-store's listeners. Existing no-arg factories still satisfy the new shape — TypeScript allows fewer parameters than the declared type.

Internal: the per-connection inner controller plus the optional caller signal are composed with \`AbortSignal.any\`, replacing the manually-scoped abort forwarder and the prior \`outerController\` that bridged the construction-time \`abortSignal\` to inner connections. Same observable behaviour for the supersede / reset paths; new behaviour for the caller-signal path (surfaces as error rather than silently disconnecting).

Tests in \`@solana/subscribable\`, \`@solana/rpc-subscriptions-spec\`, and \`@solana/kit\` updated to exercise the new \`withSignal\` API. The "abort signal" describe blocks (which tested the deprecated construction-time signal) are replaced with focused \`withSignal()\` blocks covering the per-connection timeout, kill-switch, and supersede-doesn't-touch-caller-signal cases.

Author: mcintyre94

.changeset/blue-webs-work.md

@@ -0,0 +1,23 @@
+---
+'@solana/subscribable': major
+'@solana/rpc-subscriptions-spec': major
+'@solana/kit': major
+---
+
+Add `withSignal()` to `ReactiveStreamStore` for per-connection cancellation, replacing the construction-time `abortSignal` option. Mirrors the action store's per-dispatch `withSignal()` pattern — callers attach a per-connection signal at the call site instead of baking one into the store.
+
+```ts
+const store = createReactiveStoreFromDataPublisherFactory({
+    createDataPublisher: signal => transport({ signal, ...plan }),
+    dataChannelName: 'notification',
+    errorChannelName: 'error',
+});
+// Per-connection timeout — fresh clock per attempt:
+store.withSignal(AbortSignal.timeout(30_000)).connect();
+```
+
+`store.withSignal(signal)` returns a thin wrapper exposing `connect()` that composes the caller-provided signal with the per-connection inner controller via `AbortSignal.any`. Aborting the caller's signal surfaces the abort reason on state as `{ status: 'error' }`; supersession via the internal controller (a newer `connect()` or `reset()`) stays silent so the newer call owns state. The "permanent kill switch" pattern is expressible by binding once: `const killable = store.withSignal(killCtrl.signal); killable.connect();`. After `killCtrl.abort()`, every `killable.connect()` short-circuits to error.
+
+`createDataPublisher` is widened from `() => Promise<DataPublisher>` to `(signal: AbortSignal) => Promise<DataPublisher>`. The store passes the composed per-connection signal to the factory so the underlying transport can stop on per-connection abort, not just the stream-store's listeners. Existing no-arg factories still satisfy the new shape — TypeScript allows fewer parameters than the declared type.
+
+The construction-time `abortSignal` option on `createReactiveStoreFromDataPublisherFactory`, `createReactiveStoreWithInitialValueAndSlotTracking`, and `PendingRpcSubscriptionsRequest.reactiveStore()` is removed. Callers wanting a long-lived kill switch use the bind-once `withSignal` pattern. `ReactiveStreamSource<T>.reactiveStore()` is now parameter-less (mirrors `ReactiveActionSource<T>.reactiveStore()`).

packages/kit/src/__tests__/create-reactive-store-with-initial-value-and-slot-tracking-test.ts

@@ -15,12 +15,6 @@ import type { ReactiveState, ReactiveStreamStore } from '@solana/subscribable';
  * @see {@link createReactiveStoreWithInitialValueAndSlotTracking}
  */
 export type CreateReactiveStoreWithInitialValueAndSlotTrackingConfig<TRpcValue, TSubscriptionValue, TItem> = Readonly<{
-    /**
-     * Triggering this abort signal will cancel any in-flight RPC request and subscription, and
-     * permanently disconnect the store from further updates. Subsequent
-     * {@link ReactiveStreamStore.connect | `connect()`} calls become no-ops.
-     */
-    abortSignal: AbortSignal;
     /**
      * A pending RPC request whose response will be used to set the store's initial state.
      * The response must be a {@link SolanaRpcResponse} so that its slot can be compared with
@@ -73,8 +67,10 @@ const IDLE_STATE: ReactiveState<never> = Object.freeze({
  *   subscription with a fresh inner abort signal.
  * - {@link ReactiveStreamStore.reset | `reset()`} aborts the current connection and returns the
  *   store to `idle`, clearing `data` and `error`.
- * - Triggering the caller's `abortSignal` disconnects the store permanently; subsequent `connect()`
- *   calls are no-ops.
+ * - Attach a caller-provided cancellation source via
+ *   {@link ReactiveStreamStore.withSignal | `withSignal()`} — `store.withSignal(signal).connect()`
+ *   composes the signal with the per-connection controller. Aborting the caller's signal
+ *   transitions the store to `error` with that abort reason.
  *
  * @param config
  *
@@ -92,7 +88,6 @@ const IDLE_STATE: ReactiveState<never> = Object.freeze({
  * const myAddress = address('FnHyam9w4NZoWR6mKN1CuGBritdsEWZQa4Z4oawLZGxa');
  *
  * const balanceStore = createReactiveStoreWithInitialValueAndSlotTracking({
- *     abortSignal: AbortSignal.timeout(60_000),
  *     rpcRequest: rpc.getBalance(myAddress, { commitment: 'confirmed' }),
  *     rpcValueMapper: lamports => lamports,
  *     rpcSubscriptionRequest: rpcSubscriptions.accountNotifications(myAddress),
@@ -108,13 +103,12 @@ const IDLE_STATE: ReactiveState<never> = Object.freeze({
  *         console.log(`Balance at slot ${state.data.context.slot}:`, state.data.value);
  *     }
  * });
- * balanceStore.connect();
+ * balanceStore.withSignal(AbortSignal.timeout(60_000)).connect();
  * ```
  *
  * @see {@link ReactiveStreamStore}
  */
 export function createReactiveStoreWithInitialValueAndSlotTracking<TRpcValue, TSubscriptionValue, TItem>({
-    abortSignal,
     rpcRequest,
     rpcValueMapper,
     rpcSubscriptionRequest,
@@ -127,9 +121,6 @@ export function createReactiveStoreWithInitialValueAndSlotTracking<TRpcValue, TS
     let currentInnerController: AbortController | undefined;
     const subscribers = new Set<() => void>();
 
-    const outerController = new AbortController();
-    abortSignal.addEventListener('abort', () => outerController.abort(abortSignal.reason));
-
     function notify() {
         subscribers.forEach(cb => cb());
     }
@@ -146,10 +137,14 @@ export function createReactiveStoreWithInitialValueAndSlotTracking<TRpcValue, TS
         notify();
     }
 
-    function performConnect() {
-        if (outerController.signal.aborted) return;
+    function performConnect(callerSignal: AbortSignal | undefined) {
         // Abort any currently active connection before starting a fresh one.
         currentInnerController?.abort();
+        // If the caller's signal is already aborted, surface as error and bail.
+        if (callerSignal?.aborted) {
+            setState({ data: currentState.data, error: callerSignal.reason, status: 'error' });
+            return;
+        }
         // Transition based on whether we have prior data/error to preserve.
         if (currentState.status === 'idle') {
             setState({ data: undefined, error: undefined, status: 'loading' });
@@ -159,12 +154,25 @@ export function createReactiveStoreWithInitialValueAndSlotTracking<TRpcValue, TS
 
         const innerController = new AbortController();
         currentInnerController = innerController;
-        const forwardAbort = () => innerController.abort(outerController.signal.reason);
-        outerController.signal.addEventListener('abort', forwardAbort, { signal: innerController.signal });
         const innerSignal = innerController.signal;
+        const signal = callerSignal ? AbortSignal.any([innerSignal, callerSignal]) : innerSignal;
+        // Caller's signal aborting (not just supersede via the inner controller) transitions the
+        // store to error with the caller's abort reason. Scoped to the inner signal so the
+        // listener is removed on reconnect / reset.
+        if (callerSignal) {
+            callerSignal.addEventListener(
+                'abort',
+                () => {
+                    if (innerSignal.aborted) return;
+                    setState({ data: currentState.data, error: callerSignal.reason, status: 'error' });
+                    innerController.abort(callerSignal.reason);
+                },
+                { signal: innerSignal },
+            );
+        }
 
         function handleError(err: unknown) {
-            if (innerSignal.aborted) return;
+            if (signal.aborted) return;
             if (currentState.status === 'error') return;
             setState({ data: currentState.data, error: err, status: 'error' });
             innerController.abort(err);
@@ -175,9 +183,9 @@ export function createReactiveStoreWithInitialValueAndSlotTracking<TRpcValue, TS
         }
 
         rpcRequest
-            .send({ abortSignal: innerSignal })
+            .send({ abortSignal: signal })
             .then(({ context: { slot }, value }) => {
-                if (innerSignal.aborted) return;
+                if (signal.aborted) return;
                 // `lastUpdateSlot` persists across retries so the store never regresses. If the
                 // retried RPC returns a slot older than one we've already seen, we wait for the
                 // subscription to deliver something newer before leaving `retrying`.
@@ -188,13 +196,13 @@ export function createReactiveStoreWithInitialValueAndSlotTracking<TRpcValue, TS
             .catch(handleError);
 
         rpcSubscriptionRequest
-            .subscribe({ abortSignal: innerSignal })
+            .subscribe({ abortSignal: signal })
             .then(async notifications => {
                 for await (const {
                     context: { slot },
                     value,
                 } of notifications) {
-                    if (innerSignal.aborted) return;
+                    if (signal.aborted) return;
                     if (slot < lastUpdateSlot) continue;
                     lastUpdateSlot = slot;
                     handleValue({ context: { slot }, value: rpcSubscriptionValueMapper(value) });
@@ -212,7 +220,9 @@ export function createReactiveStoreWithInitialValueAndSlotTracking<TRpcValue, TS
     }
 
     return {
-        connect: performConnect,
+        connect(): void {
+            performConnect(undefined);
+        },
         getError(): unknown {
             return currentState.error;
         },
@@ -225,13 +235,20 @@ export function createReactiveStoreWithInitialValueAndSlotTracking<TRpcValue, TS
         reset: performReset,
         retry(): void {
             if (currentState.status !== 'error') return;
-            performConnect();
+            performConnect(undefined);
         },
         subscribe(callback: () => void): () => void {
             subscribers.add(callback);
             return () => {
                 subscribers.delete(callback);
             };
         },
+        withSignal(signal: AbortSignal) {
+            return {
+                connect(): void {
+                    performConnect(signal);
+                },
+            };
+        },
     };
 }

packages/kit/src/create-reactive-store-with-initial-value-and-slot-tracking.ts

@@ -66,28 +66,26 @@ describe('PendingRpcSubscriptionsRequest.reactiveStore()', () => {
     });
 
     it('returns a store that starts in `idle` status and does not call the transport before connect()', () => {
-        const store = rpcSubscriptions
-            .thingNotifications()
-            .reactiveStore({ abortSignal: new AbortController().signal });
+        const store = rpcSubscriptions.thingNotifications().reactiveStore();
         expect(store.getUnifiedState()).toStrictEqual({
             data: undefined,
             error: undefined,
             status: 'idle',
         });
         expect(mockTransport).not.toHaveBeenCalled();
     });
-    it('passes the abort signal to the transport on connect()', async () => {
+    it('passes the per-connection signal to the transport on withSignal().connect()', async () => {
         expect.assertions(1);
         const abortController = new AbortController();
-        const store = rpcSubscriptions.thingNotifications().reactiveStore({ abortSignal: abortController.signal });
-        store.connect();
+        const store = rpcSubscriptions.thingNotifications().reactiveStore();
+        store.withSignal(abortController.signal).connect();
         await flushMicrotasks();
-        expect(mockTransport).toHaveBeenCalledWith(expect.objectContaining({ signal: abortController.signal }));
+        const transportSignal = mockTransport.mock.calls[0][0].signal;
+        // The transport receives a composed signal — aborting the caller's source aborts it.
+        expect(transportSignal.aborted).toBe(false);
     });
     it('transitions to `loading` after connect() before the transport resolves', () => {
-        const store = rpcSubscriptions
-            .thingNotifications()
-            .reactiveStore({ abortSignal: new AbortController().signal });
+        const store = rpcSubscriptions.thingNotifications().reactiveStore();
         store.connect();
         expect(store.getUnifiedState()).toStrictEqual({
             data: undefined,
@@ -97,9 +95,7 @@ describe('PendingRpcSubscriptionsRequest.reactiveStore()', () => {
     });
     it('reflects incoming notifications after connect()', async () => {
         expect.assertions(1);
-        const store = rpcSubscriptions
-            .thingNotifications()
-            .reactiveStore({ abortSignal: new AbortController().signal });
+        const store = rpcSubscriptions.thingNotifications().reactiveStore();
         store.connect();
         await flushMicrotasks();
         publish('notification', { value: 42 });
@@ -111,9 +107,7 @@ describe('PendingRpcSubscriptionsRequest.reactiveStore()', () => {
     });
     it('calls store subscribers when a notification arrives', async () => {
         expect.assertions(1);
-        const store = rpcSubscriptions
-            .thingNotifications()
-            .reactiveStore({ abortSignal: new AbortController().signal });
+        const store = rpcSubscriptions.thingNotifications().reactiveStore();
         store.connect();
         await flushMicrotasks();
         const subscriber = jest.fn();
@@ -123,9 +117,7 @@ describe('PendingRpcSubscriptionsRequest.reactiveStore()', () => {
     });
     it('surfaces errors via getUnifiedState()', async () => {
         expect.assertions(1);
-        const store = rpcSubscriptions
-            .thingNotifications()
-            .reactiveStore({ abortSignal: new AbortController().signal });
+        const store = rpcSubscriptions.thingNotifications().reactiveStore();
         store.connect();
         await flushMicrotasks();
         const error = new Error('o no');
@@ -138,21 +130,19 @@ describe('PendingRpcSubscriptionsRequest.reactiveStore()', () => {
     });
     it('re-invokes the transport on connect() after an error', async () => {
         expect.assertions(1);
-        const store = rpcSubscriptions
-            .thingNotifications()
-            .reactiveStore({ abortSignal: new AbortController().signal });
+        const store = rpcSubscriptions.thingNotifications().reactiveStore();
         store.connect();
         await flushMicrotasks();
         publish('error', new Error('stream died'));
         store.connect();
         await flushMicrotasks();
         expect(mockTransport).toHaveBeenCalledTimes(2);
     });
-    it('aborts the signal forwarded to the data publisher listeners when the caller aborts', async () => {
+    it('aborts the signal forwarded to the data publisher listeners when the caller signal aborts', async () => {
         expect.assertions(2);
         const abortController = new AbortController();
-        const store = rpcSubscriptions.thingNotifications().reactiveStore({ abortSignal: abortController.signal });
-        store.connect();
+        const store = rpcSubscriptions.thingNotifications().reactiveStore();
+        store.withSignal(abortController.signal).connect();
         await flushMicrotasks();
         const onCall = mockOn.mock.calls.find(([channel]: [string]) => channel === 'notification');
         const listenerSignal = (onCall![2] as { signal: AbortSignal }).signal;
@@ -162,9 +152,7 @@ describe('PendingRpcSubscriptionsRequest.reactiveStore()', () => {
     });
     it('returns the same getUnifiedState() snapshot across consecutive calls when state has not changed', async () => {
         expect.assertions(2);
-        const store = rpcSubscriptions
-            .thingNotifications()
-            .reactiveStore({ abortSignal: new AbortController().signal });
+        const store = rpcSubscriptions.thingNotifications().reactiveStore();
         store.connect();
         await flushMicrotasks();
         expect(store.getUnifiedState()).toBe(store.getUnifiedState());

packages/rpc-subscriptions-spec/src/__tests__/rpc-subscriptions-test.ts

@@ -9,32 +9,35 @@ import { ReactiveStreamStore } from '@solana/subscribable';
  * {@link PendingRpcSubscriptionsRequest | PendingRpcSubscriptionsRequest<TNotification>} will
  * trigger the subscription and return a promise for an async iterable that vends `TNotifications`.
  *
- * Calling the {@link PendingRpcSubscriptionsRequest.reactiveStore | `reactiveStore(options)`}
- * method will return a {@link ReactiveStreamStore} compatible with `useSyncExternalStore`, Svelte
+ * Calling the {@link PendingRpcSubscriptionsRequest.reactiveStore | `reactiveStore()`} method
+ * will return a {@link ReactiveStreamStore} compatible with `useSyncExternalStore`, Svelte
  * stores, and other reactive primitives. The returned store is in `status: 'idle'`; the caller
  * is responsible for invoking {@link ReactiveStreamStore.connect | `connect()`} to open the
- * underlying stream.
+ * underlying stream. Attach a caller-provided cancellation source via
+ * {@link ReactiveStreamStore.withSignal | `withSignal()`}.
  */
 export type PendingRpcSubscriptionsRequest<TNotification> = {
     /**
      * Synchronously returns a {@link ReactiveStreamStore} that holds the latest notification.
      * Compatible with `useSyncExternalStore` and other reactive primitives that expect a
      * `{ subscribe, getUnifiedState }` contract. The returned store is in `status: 'idle'` — call
      * {@link ReactiveStreamStore.connect | `connect()`} to open the subscription; a follow-up
-     * `connect()` after an error reopens it.
+     * `connect()` after an error reopens it. Attach a caller-provided cancellation source via
+     * {@link ReactiveStreamStore.withSignal | `withSignal()`}.
      *
      * @example
      * ```ts
-     * const store = rpc.accountNotifications(address).reactiveStore({ abortSignal });
-     * store.connect();
+     * const store = rpc.accountNotifications(address).reactiveStore();
+     * // Per-connection timeout — fresh clock per attempt:
+     * store.withSignal(AbortSignal.timeout(30_000)).connect();
      * // React — the unified snapshot has stable identity per update.
      * const state = useSyncExternalStore(store.subscribe, store.getUnifiedState);
      * if (state.status === 'error') return <ErrorMessage error={state.error} onRetry={store.connect} />;
      * if (state.status === 'loading' || state.status === 'idle') return <Spinner />;
      * return <View data={state.data} />;
      * ```
      */
-    reactiveStore(options: RpcSubscribeOptions): ReactiveStreamStore<TNotification>;
+    reactiveStore(): ReactiveStreamStore<TNotification>;
     /**
      * Triggers the subscription and returns a promise for an async iterable of notifications.
      * Use `for await...of` to consume notifications as they arrive. Abort the signal to

packages/rpc-subscriptions-spec/src/rpc-subscriptions-request.ts

@@ -82,11 +82,10 @@ function createPendingRpcSubscription<TNotification>(
     subscriptionsPlan: RpcSubscriptionsPlan<TNotification>,
 ): PendingRpcSubscriptionsRequest<TNotification> {
     return {
-        reactiveStore({ abortSignal }: RpcSubscribeOptions) {
+        reactiveStore() {
             return createReactiveStoreFromDataPublisherFactory<TNotification>({
-                abortSignal,
-                createDataPublisher() {
-                    return transport({ signal: abortSignal, ...subscriptionsPlan });
+                createDataPublisher(signal) {
+                    return transport({ signal, ...subscriptionsPlan });
                 },
                 dataChannelName: 'notification',
                 errorChannelName: 'error',