Add reactiveStore to PendingRpcRequest

.changeset/bold-drinks-strive.md

@@ -0,0 +1,14 @@
+---
+'@solana/rpc-spec': minor
+'@solana/kit': minor
+---
+
+Add a `reactiveStore()` method to `PendingRpcRequest`. It fires the request on construction and synchronously returns a `ReactiveActionStore` that holds the request's `idle`/`running`/`success`/`error` lifecycle state. Compatible with `useSyncExternalStore`, Svelte stores, and other reactive primitives. Call `dispatch()` to re-fire the request (e.g. after an error), or `reset()` to abort the in-flight call and return to idle.
+
+```ts
+const store = rpc.getAccountInfo(address).reactiveStore();
+const state = useSyncExternalStore(store.subscribe, store.getState);
+if (state.status === 'error') return <ErrorMessage error={state.error} onRetry={store.dispatch} />;
+if (state.status === 'running' && !state.data) return <Spinner />;
+return <View data={state.data!} />;
+```

CLAUDE.md

@@ -50,6 +50,8 @@ Four private "impl" packages (`@solana/crypto-impl`, `@solana/text-encoding-impl
 - **Lint/prettier**: Also run through Jest runners (`jest-runner-eslint`, `jest-runner-prettier`).
 - **Commands**: `pnpm test` runs all unit tests. `pnpm lint` runs lint checks. `pnpm style:fix` auto-fixes formatting.
 - **`expect.assertions`**: Only use `expect.assertions(n)` in **async** tests (where you need to guarantee the expected number of assertions ran). Synchronous tests do not need it.
+- **Flushing async state**: When a test needs to wait for queued microtasks or promise chains to settle, prefer `jest.useFakeTimers()` + `await jest.runAllTimersAsync()` over hand-rolled `flushMicrotasks` helpers that `await Promise.resolve()` in a loop. The loop count is fragile and breaks as soon as an extra `.then` is introduced. When enabling fake timers in a scoped `beforeEach` (i.e. not at the top of the file), pair it with an `afterEach(() => { jest.useRealTimers(); })` so subsequent describes don't inherit fake timers.
+- **Placeholder mocks**: When a test mock must satisfy an interface but a particular method shouldn't be called in that test, make the stub throw/reject rather than using a bare `jest.fn()` that silently returns `undefined`. For sync methods use `jest.fn().mockImplementation(() => { throw new Error('not implemented'); })`; for async methods use `jest.fn().mockRejectedValue(new Error('not implemented'))`. An accidental call then fails the test loudly instead of producing `undefined` and a confusing downstream assertion error.
 
 ## Error System
 

packages/accounts/src/__tests__/__setup__.ts

@@ -20,7 +20,10 @@ export function getMockRpc(
 ): Rpc<GetAccountInfoApi | GetMultipleAccountsApi> & { getAccountInfo: jest.Mock; getMultipleAccounts: jest.Mock } {
     const wrapInPendingResponse = <T>(value: T): PendingRpcRequest<SolanaRpcResponse<T>> => {
         const send = jest.fn().mockResolvedValue({ context: { slot: 0n }, value });
-        return { send };
+        const reactiveStore = jest.fn().mockImplementation(() => {
+            throw new Error('not implemented');
+        });
+        return { reactiveStore, send };
     };
 
     const getAccountInfo = jest

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

@@ -13,7 +13,12 @@ function createMockRpcRequest(): {
 } {
     const { promise, resolve, reject } = Promise.withResolvers<SolanaRpcResponse<TestValue>>();
     return {
-        mockRequest: { send: jest.fn().mockReturnValue(promise) },
+        mockRequest: {
+            reactiveStore: jest.fn().mockImplementation(() => {
+                throw new Error('not implemented');
+            }),
+            send: jest.fn().mockReturnValue(promise),
+        },
         reject,
         resolve,
     };

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

@@ -13,7 +13,12 @@ function createMockRpcRequest(): {
 } {
     const { promise, resolve, reject } = Promise.withResolvers<SolanaRpcResponse<TestValue>>();
     return {
-        mockRequest: { send: jest.fn().mockReturnValue(promise) },
+        mockRequest: {
+            reactiveStore: jest.fn().mockImplementation(() => {
+                throw new Error('not implemented');
+            }),
+            send: jest.fn().mockReturnValue(promise),
+        },
         reject,
         resolve,
     };
@@ -628,6 +633,9 @@ describe('createReactiveStoreWithInitialValueAndSlotTracking', () => {
                 pushNotification(notification: SolanaRpcResponse<TestValue>): void;
             }[] = [];
             const rpcRequest: PendingRpcRequest<SolanaRpcResponse<TestValue>> = {
+                reactiveStore: jest.fn().mockImplementation(() => {
+                    throw new Error('not implemented');
+                }),
                 send: jest.fn().mockImplementation(() => {
                     const { promise, resolve, reject } = Promise.withResolvers<SolanaRpcResponse<TestValue>>();
                     rpcInstances.push({ reject, resolve });

packages/rpc-graphql/src/loaders/__tests__/account-loader-test.ts

@@ -419,6 +419,9 @@ describe('account loader', () => {
                 // First we should see `getMultipleAccounts` used for the first two layers
                 rpc.getMultipleAccounts
                     .mockImplementationOnce(() => ({
+                        reactiveStore: jest.fn().mockImplementation(() => {
+                            throw new Error('not implemented');
+                        }),
                         send: () =>
                             Promise.resolve(
                                 getMultipleAccountsMockResponse([
@@ -442,6 +445,9 @@ describe('account loader', () => {
                             ),
                     }))
                     .mockImplementationOnce(() => ({
+                        reactiveStore: jest.fn().mockImplementation(() => {
+                            throw new Error('not implemented');
+                        }),
                         send: () =>
                             Promise.resolve(
                                 getMultipleAccountsMockResponse([
@@ -464,6 +470,9 @@ describe('account loader', () => {
                 // Then we should see `getAccountInfo` used for the single
                 // account in the last layer
                 rpc.getAccountInfo.mockReturnValue({
+                    reactiveStore: jest.fn().mockImplementation(() => {
+                        throw new Error('not implemented');
+                    }),
                     send: jest.fn().mockResolvedValueOnce({
                         context: {
                             slot: 0,

packages/rpc-spec/README.md

@@ -34,6 +34,16 @@ Pending requests are the result of calling a supported method on a `Rpc` object.
 
 Calling the `send(options)` method on a `PendingRpcRequest<TResponse>` will trigger the request and return a promise for `TResponse`.
 
+Calling the `reactiveStore()` method fires the request immediately and synchronously returns a [`ReactiveActionStore`](https://github.com/anza-xyz/kit/tree/main/packages/subscribable) compatible with `useSyncExternalStore`, Svelte stores, and other reactive primitives. The store starts in `status: 'running'`, transitions to `success` or `error` when the request settles, and can be re-fired via `dispatch()` or cancelled via `reset()`.
+
+```ts
+const store = rpc.getAccountInfo(address).reactiveStore();
+const state = useSyncExternalStore(store.subscribe, store.getState);
+if (state.status === 'error') return <ErrorMessage error={state.error} onRetry={store.dispatch} />;
+if (state.status === 'running' && !state.data) return <Spinner />;
+return <View data={state.data!} />;
+```
+
 ### `Rpc<TRpcMethods, TRpcTransport>`
 
 An object that exposes all of the functions described by `TRpcMethods`. Calling each method returns a `PendingRpcRequest<TResponse>` where `TResponse` is that method's response type.

packages/rpc-spec/package.json

@@ -75,7 +75,8 @@
     ],
     "dependencies": {
         "@solana/errors": "workspace:*",
-        "@solana/rpc-spec-types": "workspace:*"
+        "@solana/rpc-spec-types": "workspace:*",
+        "@solana/subscribable": "workspace:*"
     },
     "peerDependencies": {
         "typescript": ">=5.4.0"

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

@@ -80,6 +80,114 @@ describe('JSON-RPC 2.0', () => {
             expect(rpc).not.toHaveProperty('then');
         });
     });
+    describe('when calling reactiveStore() on a pending request', () => {
+        let execute: jest.Mock;
+        let rpc: Rpc<TestRpcMethods>;
+        beforeEach(() => {
+            jest.useFakeTimers();
+            execute = jest.fn(
+                () =>
+                    new Promise(() => {
+                        /* never resolve */
+                    }),
+            );
+            rpc = createRpc({
+                api: new Proxy({} as RpcApi<TestRpcMethods>, {
+                    get() {
+                        return (..._params: unknown[]): RpcPlan<unknown> => ({ execute });
+                    },
+                }),
+                transport: makeHttpRequest,
+            });
+        });
+        afterEach(() => {
+            jest.useRealTimers();
+        });
+        it('fires the request on creation with a non-aborted signal', () => {
+            rpc.someMethod(123).reactiveStore();
+            expect(execute).toHaveBeenCalledTimes(1);
+            const { signal } = execute.mock.calls[0][0];
+            expect(signal).toBeInstanceOf(AbortSignal);
+            expect(signal.aborted).toBe(false);
+        });
+        it('forwards the transport to the plan on creation', () => {
+            rpc.someMethod(123).reactiveStore();
+            expect(execute).toHaveBeenCalledWith(expect.objectContaining({ transport: makeHttpRequest }));
+        });
+        it('returns a store synchronously in the `running` status', () => {
+            const store = rpc.someMethod(123).reactiveStore();
+            expect(store.getState()).toStrictEqual({
+                data: undefined,
+                error: undefined,
+                status: 'running',
+            });
+        });
+        it('transitions to `success` with resolved data once the plan resolves', async () => {
+            expect.assertions(1);
+            const { promise, resolve } = Promise.withResolvers<number>();
+            execute.mockReturnValueOnce(promise);
+            const store = rpc.someMethod(123).reactiveStore();
+            resolve(42);
+            await jest.runAllTimersAsync();
+            expect(store.getState()).toStrictEqual({
+                data: 42,
+                error: undefined,
+                status: 'success',
+            });
+        });
+        it('transitions to `error` when the plan rejects', async () => {
+            expect.assertions(1);
+            const { promise, reject } = Promise.withResolvers<number>();
+            execute.mockReturnValueOnce(promise);
+            const store = rpc.someMethod(123).reactiveStore();
+            const error = new Error('o no');
+            reject(error);
+            await jest.runAllTimersAsync();
+            expect(store.getState()).toStrictEqual({
+                data: undefined,
+                error,
+                status: 'error',
+            });
+        });
+        it('notifies subscribers when state changes', async () => {
+            expect.assertions(2);
+            const { promise, resolve } = Promise.withResolvers<number>();
+            execute.mockReturnValueOnce(promise);
+            const store = rpc.someMethod(123).reactiveStore();
+            const subscriberA = jest.fn();
+            const subscriberB = jest.fn();
+            store.subscribe(subscriberA);
+            store.subscribe(subscriberB);
+            resolve(42);
+            await jest.runAllTimersAsync();
+            expect(subscriberA).toHaveBeenCalledTimes(1);
+            expect(subscriberB).toHaveBeenCalledTimes(1);
+        });
+        it('re-fires the plan when dispatch() is called', async () => {
+            expect.assertions(1);
+            // request 1: rejects
+            execute.mockRejectedValueOnce(new Error('o no'));
+            const store = rpc.someMethod(123).reactiveStore();
+            await jest.runAllTimersAsync();
+            // request 2: resolves
+            execute.mockResolvedValueOnce(42);
+            store.dispatch();
+            await jest.runAllTimersAsync();
+            expect(execute).toHaveBeenCalledTimes(2);
+        });
+        it('aborts the in-flight signal and returns to idle when reset() is called', () => {
+            const store = rpc.someMethod(123).reactiveStore();
+            const { signal } = execute.mock.calls[0][0];
+            expect(signal.aborted).toBe(false);
+            store.reset();
+            expect(signal.aborted).toBe(true);
+            expect(store.getState()).toStrictEqual({
+                data: undefined,
+                error: undefined,
+                status: 'idle',
+            });
+        });
+    });
     describe('when calling a method having a concrete implementation', () => {
         let rpc: Rpc<TestRpcMethods>;
         beforeEach(() => {

packages/rpc-spec/src/rpc.ts

@@ -1,5 +1,6 @@
 import { SOLANA_ERROR__RPC__API_PLAN_MISSING_FOR_RPC_METHOD, SolanaError } from '@solana/errors';
 import { Callable, Flatten, OverloadImplementations, UnionToIntersection } from '@solana/rpc-spec-types';
+import { createReactiveActionStore, ReactiveActionStore } from '@solana/subscribable';
 
 import { RpcApi, RpcPlan } from './rpc-api';
 import { RpcTransport } from './rpc-transport';
@@ -26,8 +27,29 @@ export type Rpc<TRpcMethods> = {
  * Calling the {@link PendingRpcRequest.send | `send(options)`} method on a
  * {@link PendingRpcRequest | PendingRpcRequest<TResponse>} will trigger the request and return a
  * promise for `TResponse`.
+ *
+ * Calling the {@link PendingRpcRequest.reactiveStore | `reactiveStore()`} method will fire the
+ * request and return a {@link ReactiveActionStore} compatible with `useSyncExternalStore`, Svelte
+ * stores, and other reactive primitives.
  */
 export type PendingRpcRequest<TResponse> = {
+    /**
+     * Synchronously returns a {@link ReactiveActionStore} that fires the request on construction
+     * and holds its lifecycle state. Compatible with `useSyncExternalStore` and other reactive
+     * primitives that expect a `{ subscribe, getState }` contract. Call `dispatch()` to re-fire the
+     * request (for example after an error), or `reset()` to abort the in-flight call and return to
+     * `status: 'idle'`.
+     *
+     * @example
+     * ```ts
+     * const store = rpc.getAccountInfo(address).reactiveStore();
+     * const state = useSyncExternalStore(store.subscribe, store.getState);
+     * if (state.status === 'error') return <ErrorMessage error={state.error} onRetry={store.dispatch} />;
+     * if (state.status === 'running' && !state.data) return <Spinner />;
+     * return <View data={state.data!} />;
+     * ```
+     */
+    reactiveStore(): ReactiveActionStore<[], TResponse>;
     send(options?: RpcSendOptions): Promise<TResponse>;
 };
 
@@ -96,6 +118,11 @@ function createPendingRpcRequest<TRpcMethods, TRpcTransport extends RpcTransport
     plan: RpcPlan<TResponse>,
 ): PendingRpcRequest<TResponse> {
     return {
+        reactiveStore(): ReactiveActionStore<[], TResponse> {
+            const store = createReactiveActionStore<[], TResponse>(signal => plan.execute({ signal, transport }));
+            store.dispatch();
+            return store;
+        },
         async send(options?: RpcSendOptions): Promise<TResponse> {
             return await plan.execute({ signal: options?.abortSignal, transport });
         },

packages/rpc-spec/tsconfig.json

@@ -1,7 +1,7 @@
 {
     "$schema": "https://json.schemastore.org/tsconfig",
     "compilerOptions": {
-        "lib": ["DOM", "ES2022.Error"]
+        "lib": ["DOM", "ES2022.Error", "ES2024.Promise"]
     },
     "display": "@solana/rpc-spec",
     "extends": "../tsconfig/base.json",

packages/transaction-confirmation/src/__tests__/confirmation-strategy-blockheight-test.ts

@@ -27,8 +27,10 @@ describe('createBlockHeightExceedencePromiseFactory', () => {
         });
         const rpcSubscriptions = {
             slotNotifications: () => ({
-                reactive: jest.fn(),
-                reactiveStore: jest.fn(),
+                reactive: jest.fn().mockRejectedValue(new Error('not implemented')),
+                reactiveStore: jest.fn().mockImplementation(() => {
+                    throw new Error('not implemented');
+                }),
                 subscribe: createSubscriptionIterable,
             }),
         };

packages/transaction-confirmation/src/__tests__/confirmation-strategy-nonce-test.ts

@@ -38,15 +38,20 @@ describe('createNonceInvalidationPromiseFactory', () => {
         getAccountInfoMock = jest.fn().mockReturnValue(FOREVER_PROMISE);
         const rpc = {
             getAccountInfo: () => ({
+                reactiveStore: jest.fn().mockImplementation(() => {
+                    throw new Error('not implemented');
+                }),
                 send: getAccountInfoMock,
             }),
         };
         createSubscriptionIterable = jest.fn().mockResolvedValue({
             [Symbol.asyncIterator]: accountNotificationGenerator,
         });
         createPendingSubscription = jest.fn().mockReturnValue({
-            reactive: jest.fn(),
-            reactiveStore: jest.fn(),
+            reactive: jest.fn().mockRejectedValue(new Error('not implemented')),
+            reactiveStore: jest.fn().mockImplementation(() => {
+                throw new Error('not implemented');
+            }),
             subscribe: createSubscriptionIterable,
         });
         const rpcSubscriptions = {

packages/transaction-confirmation/src/__tests__/confirmation-strategy-signature-test.ts

@@ -21,15 +21,20 @@ describe('createSignatureConfirmationPromiseFactory', () => {
         getSignatureStatusesMock = jest.fn().mockReturnValue(FOREVER_PROMISE);
         const rpc = {
             getSignatureStatuses: () => ({
+                reactiveStore: jest.fn().mockImplementation(() => {
+                    throw new Error('not implemented');
+                }),
                 send: getSignatureStatusesMock,
             }),
         };
         createSubscriptionIterable = jest.fn().mockResolvedValue({
             [Symbol.asyncIterator]: signatureNotificationGenerator,
         });
         createPendingSubscription = jest.fn().mockReturnValue({
-            reactive: jest.fn(),
-            reactiveStore: jest.fn(),
+            reactive: jest.fn().mockRejectedValue(new Error('not implemented')),
+            reactiveStore: jest.fn().mockImplementation(() => {
+                throw new Error('not implemented');
+            }),
             subscribe: createSubscriptionIterable,
         });
         const rpcSubscriptions = {