Add a function createReactiveStoreFromRpcAndSubscription (#1528)

* Add a function `createReactiveStoreFromRpcAndSubscription`

This creates a reactive store that combines an RPC and RPC Subscription,
that can both map to the same data type. It exposes the latest value
from either source, by slot comparison.

* Update balance in react-app to use reactive store

* Do not update state after signal is aborted

* Update lockfile

* Rename to createReactiveStoreWithInitialValueAndSlotTracking

Author: mcintyre94

.changeset/giant-falcons-invite.md

@@ -0,0 +1,5 @@
+---
+'@solana/kit': minor
+---
+
+Add `createReactiveStoreWithInitialValueAndSlotTracking()`, a helper that combines an initial RPC fetch with an ongoing subscription into a single `ReactiveStore`. Uses slot-based comparison to ensure only the most recent value is kept, regardless of arrival order. Compatible with `useSyncExternalStore`, Svelte stores, and other reactive primitives.

examples/react-app/src/functions/balance.ts

@@ -1,18 +1,21 @@
-import { AccountNotificationsApi, Address, GetBalanceApi, Lamports, Rpc, RpcSubscriptions } from '@solana/kit';
+import {
+    AccountNotificationsApi,
+    Address,
+    createReactiveStoreWithInitialValueAndSlotTracking,
+    GetBalanceApi,
+    Lamports,
+    Rpc,
+    RpcSubscriptions,
+} from '@solana/kit';
 import { SWRSubscription } from 'swr/subscription';
 
-const EXPLICIT_ABORT_TOKEN = Symbol();
-
 /**
  * This is an example of a strategy to fetch some account data and to keep it up to date over time.
  * It's implemented as an SWR subscription function (https://swr.vercel.app/docs/subscription) but
  * the approach is generalizable.
  *
- *     1. Fetch the current account state and publish it to the consumer
- *     2. Subscribe to account data notifications and publish them to the consumer
- *
- * At all points in time, check that the update you received -- no matter from where -- is from a
- * higher slot (ie. is newer) than the last one you published to the consumer.
+ * It uses {@link createReactiveStoreWithInitialValueAndSlotTracking} to combine an initial RPC fetch with an
+ * ongoing subscription, using slot-based comparison to ensure only the latest value is published.
  */
 export function balanceSubscribe(
     rpc: Rpc<GetBalanceApi>,
@@ -21,53 +24,22 @@ export function balanceSubscribe(
 ) {
     const [{ address }, { next }] = subscriptionArgs;
     const abortController = new AbortController();
-    // Keep track of the slot of the last-published update.
-    let lastUpdateSlot = -1n;
-    // Fetch the current balance of this account.
-    rpc.getBalance(address, { commitment: 'confirmed' })
-        .send({ abortSignal: abortController.signal })
-        .then(({ context: { slot }, value: lamports }) => {
-            if (slot < lastUpdateSlot) {
-                // The last-published update (ie. from the subscription) is newer than this one.
-                return;
-            }
-            lastUpdateSlot = slot;
-            next(null /* err */, lamports /* data */);
-        })
-        .catch(e => {
-            if (e !== EXPLICIT_ABORT_TOKEN) {
-                next(e /* err */);
-            }
-        });
-    // Subscribe for updates to that balance.
-    rpcSubscriptions
-        .accountNotifications(address)
-        .subscribe({ abortSignal: abortController.signal })
-        .then(async accountInfoNotifications => {
-            try {
-                for await (const {
-                    context: { slot },
-                    value: { lamports },
-                } of accountInfoNotifications) {
-                    if (slot < lastUpdateSlot) {
-                        // The last-published update (ie. from the initial fetch) is newer than this
-                        // one.
-                        continue;
-                    }
-                    lastUpdateSlot = slot;
-                    next(null /* err */, lamports /* data */);
-                }
-            } catch (e) {
-                next(e /* err */);
-            }
-        })
-        .catch(e => {
-            if (e !== EXPLICIT_ABORT_TOKEN) {
-                next(e /* err */);
-            }
-        });
-    // Return a cleanup callback that aborts the RPC call/subscription.
+    const store = createReactiveStoreWithInitialValueAndSlotTracking({
+        abortSignal: abortController.signal,
+        rpcRequest: rpc.getBalance(address, { commitment: 'confirmed' }),
+        rpcSubscriptionRequest: rpcSubscriptions.accountNotifications(address),
+        rpcSubscriptionValueMapper: ({ lamports }) => lamports,
+        rpcValueMapper: lamports => lamports,
+    });
+    store.subscribe(() => {
+        const error = store.getError();
+        if (error) {
+            next(error as Error);
+        } else {
+            next(null, store.getState());
+        }
+    });
     return () => {
-        abortController.abort(EXPLICIT_ABORT_TOKEN);
+        abortController.abort();
     };
 }

packages/kit/README.md

@@ -38,6 +38,39 @@ await airdrop({
 
 > [!NOTE] This only works on test clusters.
 
+### `createReactiveStoreWithInitialValueAndSlotTracking(config)`
+
+Creates a `ReactiveStore` that combines an initial RPC fetch with an ongoing subscription to keep its state up to date. Uses slot-based comparison to ensure only the most recent value is kept, regardless of whether it came from the RPC response or a subscription notification.
+
+The returned store is compatible with React's `useSyncExternalStore`, Svelte stores, Solid's `from()`, and any other reactive primitive that expects a `{ subscribe, getState }` contract.
+
+```ts
+import {
+    address,
+    createReactiveStoreWithInitialValueAndSlotTracking,
+    createSolanaRpc,
+    createSolanaRpcSubscriptions,
+} from '@solana/kit';
+
+const rpc = createSolanaRpc('http://127.0.0.1:8899');
+const rpcSubscriptions = createSolanaRpcSubscriptions('ws://127.0.0.1:8900');
+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),
+    rpcSubscriptionValueMapper: ({ lamports }) => lamports,
+});
+
+const unsubscribe = balanceStore.subscribe(() => {
+    const error = balanceStore.getError();
+    if (error) console.error('Error:', error);
+    else console.log('Balance:', balanceStore.getState());
+});
+```
+
 ### `decompileTransactionMessageFetchingLookupTables(compiledTransactionMessage, rpc, config)`
 
 Returns a `TransactionMessage` from a `CompiledTransactionMessage`. If any of the accounts in the compiled message require an address lookup table to find their address, this function will use the supplied RPC instance to fetch the contents of the address lookup table from the network.

packages/kit/package.json

@@ -119,6 +119,7 @@
         "@solana/rpc-parsed-types": "workspace:*",
         "@solana/rpc-spec-types": "workspace:*",
         "@solana/rpc-subscriptions": "workspace:*",
+        "@solana/subscribable": "workspace:*",
         "@solana/rpc-types": "workspace:*",
         "@solana/signers": "workspace:*",
         "@solana/sysvars": "workspace:*",