mux into one UpdateRecord feed

packages/smart-wallet/src/smartWallet.js

@@ -37,14 +37,20 @@ import {
  *   for each brand.
  */
 
-/** @typedef {Array<BrandDescriptor>} BrandsRecord */
-
 /**
- * @typedef {Array<{ currentAmount: Amount }>} BalancesRecord
+ * @typedef {{ latestOfferStatus: import('./offers.js').OfferStatus } |
+ * { latestBalance: { currentAmount: Amount } } |
+ * { latestBrand: BrandDescriptor }
+ * } UpdateRecord Record of an update to the state of this wallet.
+ *
+ * Client is responsible for coalescing updates into a current state. See `coalesceUpdates` utility.
  *
- * Array to support batching updates. The currentAmount key exists to allow disambiguation of what purse
- * the amount is for. We currently only support one purse per brand so the brand in the currentAmount
- * suffices, but this structure allows for forward-compatibility with multiple purses per brand.
+ * The reason for this burden on the client is that transferring the full state is untenable
+ * (because it would grow monotonically).
+ *
+ * `latestBalance` amount is nested for forward-compatibility with supporting
+ * more than one purse per brand. An additional key will be needed to
+ * disambiguate. For now the brand in the amount suffices.
  */
 
 // TODO remove petname? what will UI show then? look up in agoricNames?
@@ -116,33 +122,16 @@ export const makeSmartWallet = async (
 
   // #region publishing
   // NB: state size must not grow monotonically
-  // wallets subscribe to this node over RPC so provide whatever they need.
+  // This is the node that UIs subscribe to for everything they need.
   // e.g. agoric follow :published.wallet.agoric1nqxg4pye30n3trct0hf7dclcwfxz8au84hr3ht
-  // TODO publish "latest" style keys: `latestOfferStatus`, `latestIssuer`, `latestBalance`.
   const myWalletStorageNode = E(storageNode).makeChildNode(address);
 
-  /** @type {PublishKit<BrandsRecord>} */
-  const brandsPublishKit = makeStoredPublishKit(
-    E(myWalletStorageNode).makeChildNode('brands'),
-    marshaller,
-  );
-
-  const publishBrands = async () => {
-    const brands = Array.from(brandDescriptors.values());
-    brandsPublishKit.publisher.publish(brands);
-  };
-
-  /** @type {PublishKit<import('./offers.js').OfferStatus>} */
-  const offerPublishKit = makeStoredPublishKit(
-    E(myWalletStorageNode).makeChildNode('offers'),
+  /** @type {StoredPublishKit<UpdateRecord>} */
+  const updatePublishKit = makeStoredPublishKit(
+    myWalletStorageNode,
     marshaller,
   );
 
-  /** @type {StoredPublishKit<BalancesRecord>} */
-  const balancesPublishKit = makeStoredPublishKit(
-    E(myWalletStorageNode).makeChildNode('balances'),
-    marshaller,
-  );
   /**
    * @param {Purse} purse
    * @param {Amount} balance
@@ -154,11 +143,9 @@ export const makeSmartWallet = async (
     } else {
       purseBalances.set(purse, balance);
     }
-    balancesPublishKit.publisher.publish(
-      Array.from(purseBalances.values()).map(b => ({
-        currentAmount: b,
-      })),
-    );
+    updatePublishKit.publisher.publish({
+      latestBalance: { currentAmount: balance },
+    });
   };
 
   // #endregion
@@ -196,7 +183,8 @@ export const makeSmartWallet = async (
     // relevant purse. when it's time to make an offer, you know how to make
     // payments. REMEMBER when doing that, need to handle every exception to
     // put the money back in the purse if anything fails.
-    brandDescriptors.init(desc.brand, { ...desc, displayInfo });
+    const fullDesc = { ...desc, displayInfo };
+    brandDescriptors.init(desc.brand, fullDesc);
     brandPurses.init(desc.brand, purse);
 
     // publish purse's balance and changes
@@ -214,7 +202,8 @@ export const makeSmartWallet = async (
       },
     });
 
-    publishBrands();
+    console.log('DEBUG WALLET PUBLISHING BRAND', fullDesc);
+    updatePublishKit.publisher.publish({ latestBrand: fullDesc });
   };
 
   // Ensure a purse for each issuer
@@ -287,7 +276,8 @@ export const makeSmartWallet = async (
     E.get(marshaller).unserialize,
     invitationFromSpec,
     sufficientPayments,
-    offerPublishKit.publisher.publish,
+    offerStatus =>
+      updatePublishKit.publisher.publish({ latestOfferStatus: offerStatus }),
     payouts => depositPaymentsIntoPurses(payouts, brandPurses.get),
     (offerId, invitationMakers) =>
       offerToInvitationMakers.init(offerId, invitationMakers),
@@ -306,17 +296,7 @@ export const makeSmartWallet = async (
     getDepositFacet: () => depositFacet,
     getOffersFacet: () => offersFacet,
 
-    getDataFeeds: () => ({
-      balances: {
-        subscriber: balancesPublishKit.subscriber,
-      },
-      brands: {
-        subscriber: brandsPublishKit.subscriber,
-      },
-      offers: {
-        subscriber: offerPublishKit.subscriber,
-      },
-    }),
+    getUpdatesSubscriber: () => updatePublishKit.subscriber,
   });
 };
 harden(makeSmartWallet);

packages/smart-wallet/src/utils.js

@@ -1,6 +1,7 @@
 // @ts-check
 
 import { deeplyFulfilledObject, objectMap } from '@agoric/internal';
+import { observeIteration, subscribeEach } from '@agoric/notifier';
 import { E } from '@endo/far';
 
 /**
@@ -41,3 +42,37 @@ export const depositPaymentsIntoPurses = async (
   return deeplyFulfilledObject(amountPKeywordRecord);
 };
 harden(depositPaymentsIntoPurses);
+
+/**
+ * Coalesce updates from a wallet UpdateRecord publication feed. Note that local
+ * state may not reflect the wallet's state if the initial updates are missed.
+ *
+ * If this proves to be a problem we can add an option to this or a related
+ * utility to reset state from RPC.
+ *
+ * @param {ERef<StoredSubscriber<import('./smartWallet').UpdateRecord>>} updates
+ */
+export const coalesceUpdates = updates => {
+  /** @type {Map<Brand, import('./smartWallet').BrandDescriptor>} */
+  const brands = new Map();
+  /** @type {{ [id: number]: import('./offers').OfferStatus}} */
+  const offerStatuses = {};
+  /** @type {Map<Brand, Amount>} */
+  const balances = new Map();
+  observeIteration(subscribeEach(updates), {
+    updateState: updateRecord => {
+      // XXX use discriminated union
+      if ('latestBalance' in updateRecord) {
+        const { currentAmount } = updateRecord.latestBalance;
+        balances.set(currentAmount.brand, currentAmount);
+      } else if ('latestOfferStatus' in updateRecord) {
+        const status = updateRecord.latestOfferStatus;
+        offerStatuses[status.id] = status;
+      } else if ('latestBrand' in updateRecord) {
+        const descriptor = updateRecord.latestBrand;
+        brands.set(descriptor.brand, descriptor);
+      }
+    },
+  });
+  return { brands, offerStatuses, balances };
+};

packages/smart-wallet/test/test-psm-integration.js

@@ -13,6 +13,7 @@ import { eventLoopIteration } from '@agoric/zoe/tools/eventLoopIteration.js';
 import { E } from '@endo/far';
 import { makeDefaultTestContext } from './contexts.js';
 import { currentState } from './supports.js';
+import { coalesceUpdates } from '../src/utils.js';
 
 /**
  * @type {import('ava').TestFn<Awaited<ReturnType<makeDefaultTestContext>>
@@ -48,16 +49,17 @@ test.before(async t => {
   t.context = await makeDefaultTestContext(t, makePsmTestSpace);
 });
 /**
- * @param {import('../src/smartWallet.js').BalancesRecord} balancesRecord
+ * @param {Awaited<ReturnType<typeof coalesceUpdates>>} state
  * @param {Brand<'nat'>} brand
  */
-const purseBalance = (balancesRecord, brand) => {
-  const match = balancesRecord.find(b => b.currentAmount.brand === brand);
+const purseBalance = (state, brand) => {
+  const balances = Array.from(state.balances.values());
+  const match = balances.find(b => b.brand === brand);
   if (!match) {
-    console.debug('pursesRecord', ...balancesRecord);
+    console.debug('balances', ...balances);
     assert.fail(`${brand} not found in record`);
   }
-  return match.currentAmount.value;
+  return match.value;
 };
 
 test('null swap', async t => {
@@ -67,8 +69,8 @@ test('null swap', async t => {
   const minted = await E(agoricNames).lookup('brand', 'IST');
 
   const wallet = await t.context.simpleProvideWallet('agoric1nullswap');
+  const computedState = coalesceUpdates(E(wallet).getUpdatesSubscriber());
   const offersFacet = wallet.getOffersFacet();
-  const feeds = await E(wallet).getDataFeeds();
 
   const psmInstance = await E(agoricNames).lookup('instance', 'psm');
 
@@ -96,11 +98,10 @@ test('null swap', async t => {
   await eventLoopIteration();
 
   offersFacet.executeOffer(offerCapData);
-  {
-    const balancesRecord = await currentState(feeds.balances.subscriber);
-    t.is(purseBalance(balancesRecord, anchor.brand), 0n);
-    t.is(purseBalance(balancesRecord, minted), 0n);
-  }
+  await eventLoopIteration();
+
+  t.is(purseBalance(computedState, anchor.brand), 0n);
+  t.is(purseBalance(computedState, minted), 0n);
 
   // success if nothing threw
   t.pass();
@@ -119,12 +120,13 @@ test('want stable', async t => {
   const stableBrand = await E(agoricNames).lookup('brand', Stable.symbol);
 
   const wallet = await t.context.simpleProvideWallet('agoric1wantstable');
+  const computedState = coalesceUpdates(E(wallet).getUpdatesSubscriber());
+
   const offersFacet = wallet.getOffersFacet();
-  const feeds = await E(wallet).getDataFeeds();
-  {
-    const pursesRecord = await currentState(feeds.balances.subscriber);
-    t.is(purseBalance(pursesRecord, anchor.brand), 0n);
-  }
+  // let promises settle to notify brands and create purses
+  await eventLoopIteration();
+
+  t.is(purseBalance(computedState, anchor.brand), 0n);
 
   t.log('Fund the wallet');
   assert(anchor.mint);
@@ -155,11 +157,9 @@ test('want stable', async t => {
 
   t.log('Execute the swap');
   offersFacet.executeOffer(offerCapData);
-  {
-    const balancesRecord = await currentState(feeds.balances.subscriber);
-    t.is(purseBalance(balancesRecord, anchor.brand), 0n);
-    t.is(purseBalance(balancesRecord, stableBrand), swapSize - 1n);
-  }
+  await eventLoopIteration();
+  t.is(purseBalance(computedState, anchor.brand), 0n);
+  t.is(purseBalance(computedState, stableBrand), swapSize - 1n);
 });
 
 test('govern offerFilter', async t => {
@@ -172,8 +172,8 @@ test('govern offerFilter', async t => {
   } = await E.get(t.context.consume);
 
   const wallet = await t.context.simpleProvideWallet(committeeAddress);
+  const computedState = coalesceUpdates(E(wallet).getUpdatesSubscriber());
   const offersFacet = wallet.getOffersFacet();
-  const offersSubscriber = wallet.getDataFeeds().offers.subscriber;
 
   const psmInstance = await E(agoricNames).lookup('instance', 'psm');
 
@@ -254,7 +254,9 @@ test('govern offerFilter', async t => {
   }
 
   t.log('Make sure vote happened');
-  t.like(await currentState(offersSubscriber), {
+  await eventLoopIteration();
+  const status = computedState.offerStatuses[44];
+  t.like(status, {
     id: 44,
     state: 'paid',
     result: { chosen: { strings: ['wantStable'] }, shares: 1n },

packages/smart-wallet/test/test-walletFactory.js

@@ -30,10 +30,10 @@ test.before(async t => {
 
 test('bridge', async t => {
   const smartWallet = await t.context.simpleProvideWallet(mockAddress1);
-  const feeds = await E(smartWallet).getDataFeeds();
-  t.truthy(feeds.offers.subscriber);
+  const updates = await E(smartWallet).getUpdatesSubscriber();
+  t.truthy(updates);
 
-  const offersHead = () => currentState(feeds.offers.subscriber);
+  const lastUpdate = () => currentState(updates);
 
   const ctx = makeImportContext();
 
@@ -63,25 +63,25 @@ test('bridge', async t => {
   });
   t.is(res, undefined);
 
-  t.deepEqual(await offersHead(), {
-    ...offerSpec,
-    state: 'error',
+  t.deepEqual(await lastUpdate(), {
+    latestOfferStatus: {
+      ...offerSpec,
+      state: 'error',
+    },
   });
 });
 
 test('notifiers', async t => {
   async function checkAddress(address) {
     const smartWallet = await t.context.simpleProvideWallet(address);
 
-    const feeds = await E(smartWallet).getDataFeeds();
+    const updates = await E(smartWallet).getUpdatesSubscriber();
 
-    for (const [key, feed] of Object.entries(feeds)) {
-      t.is(
-        // @ts-expect-error faulty typedef
-        await subscriptionKey(feed.subscriber),
-        `mockChainStorageRoot.wallet.${address}.${key}`,
-      );
-    }
+    t.is(
+      // @ts-expect-error faulty typedef
+      await subscriptionKey(updates),
+      `mockChainStorageRoot.wallet.${address}`,
+    );
   }
 
   await Promise.all(