Remove await when(...) in Network and IBC vats (#8964)

* fixup: durable vatUpgradeInfo

* test(boot): override bridgeHandlers in makeSwingsetTestKit

* test(vats-restart): object from ibc v1 usable after upgrade

* test(boot): mock IBC bridge

* test: mock client, server contracts

* test(boot): multi-step network/ibc upgrade testing (WIP)

 - use contracts in net/ibc upgrade test

* chore(network): include localAddress in "Already accepted" msg

* test(boot): Add pause functionality to the mock IBC server

For testing upgrades while an operation is pending (i.e., before it is `dequeue`d)

* chore: avoid `Error`s in state

Serializing them causes them to appear on the console,
which is very misleading

* feat(network): add watchers

* fixup! add watchers

* fixup! add watchers

* Fix Inbound async

* fixup! add watchers

* fix: watch after merge

* fix peg tests

* fix: adding types

* fix: test lint fixes

* fix: add watchers to ibc

* return ack watcher

* fix: return ackWatcher

* fix: bridge type

* move to singleton pattern

* fix! move watchers to singleton

* fix: move watchers to kit

* fix: Move network watchers to kits

* feat: Add vow await all

* fix: lint:

* fix: adding types

* Update packages/boot/test/bootstrapTests/test-net-ibc-upgrade.ts

Co-authored-by: Richard Gibson <richard.gibson@gmail.com>

* fix: all -> allVows

* Apply suggestions from code review

Co-authored-by: Michael FIG <mfig@agoric.com>

* fix: add allVow tests

* fix: lint

* fix: add more test cases

---------

Co-authored-by: Dan Connolly <connolly@agoric.com>
Co-authored-by: Richard Gibson <richard.gibson@gmail.com>
Co-authored-by: Michael FIG <mfig@agoric.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

packages/boot/package.json

@@ -51,6 +51,7 @@
     "@agoric/governance": "^0.10.3",
     "@agoric/store": "^0.9.2",
     "@agoric/swingset-liveslots": "^0.10.2",
+    "@endo/patterns": "^1.2.0",
     "ava": "^5.3.0",
     "c8": "^9.1.0",
     "tsx": "3.12.8"

packages/boot/test/bootstrapTests/ibcBridgeMock.js

@@ -0,0 +1,60 @@
+// @ts-check
+import {
+  makePinnedHistoryTopic,
+  prepareDurablePublishKit,
+  subscribeEach,
+} from '@agoric/notifier';
+import { M } from '@endo/patterns';
+import { makeScalarBigMapStore } from '@agoric/vat-data';
+import { makeDurableZone } from '@agoric/zone/durable.js';
+
+// TODO: factor out overlap with packages/vats/test/test-network.js
+export const makeBridge = (
+  t,
+  baggage = makeScalarBigMapStore('baggage', {
+    keyShape: M.string(),
+    durable: true,
+  }),
+  zone = makeDurableZone(baggage),
+) => {
+  const makeDurablePublishKit = prepareDurablePublishKit(
+    baggage,
+    'DurablePublishKit',
+  );
+
+  const { subscriber, publisher } = makeDurablePublishKit();
+
+  const pinnedHistoryTopic = makePinnedHistoryTopic(subscriber);
+  const events = subscribeEach(pinnedHistoryTopic)[Symbol.asyncIterator]();
+
+  let hndlr;
+  /** @type {import('@agoric/vats/src/types.js').ScopedBridgeManager} */
+  const bridgeHandler = zone.exo('IBC Bridge Manager', undefined, {
+    toBridge: async obj => {
+      const { method, type, ...params } = obj;
+      publisher.publish([method, params]);
+      t.is(type, 'IBC_METHOD');
+      if (method === 'sendPacket') {
+        const { packet } = params;
+        return { ...packet, sequence: '39' };
+      }
+      return undefined;
+    },
+    fromBridge: async obj => {
+      if (!hndlr) throw Error('no handler!');
+      // EV must be late-bound; it is not availble at bridge construction time.
+      const { EV } = t.context.runutils;
+      await EV(hndlr).fromBridge(obj);
+    },
+    initHandler: h => {
+      if (hndlr) throw Error('already init');
+      hndlr = h;
+    },
+    setHandler: h => {
+      if (!hndlr) throw Error('must init first');
+      hndlr = h;
+    },
+  });
+
+  return { bridgeHandler, events };
+};

packages/boot/test/bootstrapTests/ibcClientMock.js

@@ -0,0 +1,46 @@
+/** @file Mock IBC Server */
+// @ts-check
+import { Far } from '@endo/far';
+import { V as E } from '@agoric/vat-data/vow.js';
+
+/**
+ * @param {ZCF} zcf
+ * @param {{
+ *   address: string,
+ *   networkVat: any
+ * }} privateArgs
+ * @param {import("@agoric/vat-data").Baggage} _baggage
+ */
+export const start = async (zcf, privateArgs, _baggage) => {
+  const { address, networkVat } = privateArgs;
+  const myPort = await E(networkVat).bind(address);
+
+  const { log } = console;
+  let connP;
+  let ackP;
+
+  const creatorFacet = Far('CF', {
+    connect: remote => {
+      log('connect', remote);
+      // don't return the promise.
+      // We want to test a promise that lasts across cranks.
+      connP = E(myPort).connect(
+        remote,
+
+        // TODO: handler
+      );
+    },
+    send: data => {
+      log('send', data);
+      assert(connP, 'must connect first');
+      ackP = E(connP).send(data);
+    },
+    getAck: () => E.when(ackP),
+    close: () => E(connP).close(),
+    getLocalAddress: async () => {
+      return E(myPort).getLocalAddress();
+    },
+  });
+
+  return { creatorFacet };
+};

packages/boot/test/bootstrapTests/ibcServerMock.js

@@ -0,0 +1,88 @@
+/** @file Mock IBC Server */
+// @ts-check
+import { Far } from '@endo/far';
+import { makePromiseKit } from '@endo/promise-kit';
+import { V as E } from '@agoric/vat-data/vow.js';
+
+const { quote: q, Fail } = assert;
+const { log } = console;
+
+/**
+ *
+ * @param {ZCF} zcf
+ * @param {{
+ *   address: string,
+ *   networkVat: any
+ * }} privateArgs
+ * @param {import("@agoric/vat-data").Baggage} _baggage
+ */
+export const start = async (zcf, privateArgs, _baggage) => {
+  const { address, networkVat } = privateArgs;
+
+  const boundPort = await E(networkVat).bind(address);
+
+  /** @type {Array<[label: string, resolve: (value: any) => void, reject: (reason: any) => void]>} */
+  const queue = [];
+
+  /** @type {ListenHandler} */
+  const listener = Far('L', {
+    async onAccept(_port, _localAddr, _remoteAddr, _listenHandler) {
+      const ch = Far('CH', {
+        async onReceive(_c, packetBytes) {
+          log('Receiving Data', packetBytes);
+          assert.typeof(packetBytes, 'string');
+          const { promise, resolve, reject } = makePromiseKit();
+          queue.push([
+            'onReceive',
+            () => resolve(`got ${packetBytes}`),
+            reject,
+          ]);
+          return promise;
+        },
+        async onOpen(_c, localAddr, remoteAddr, _connectionHandler) {
+          log('onOpen', { localAddr, remoteAddr });
+          const { promise, resolve, reject } = makePromiseKit();
+          queue.push(['onOpen', resolve, reject]);
+          return promise;
+        },
+      });
+      const { promise, resolve, reject } = makePromiseKit();
+      queue.push(['onAccept', () => resolve(ch), reject]);
+      return promise;
+    },
+    async onListen(port, _listenHandler) {
+      console.debug(`listening on echo port: ${port}`);
+      const { promise, resolve, reject } = makePromiseKit();
+      queue.push(['onListen', resolve, reject]);
+      return promise;
+    },
+  });
+
+  const creatorFacet = Far('CF', {
+    /**
+     * Assert that the next pending operation has the expected label and release it.
+     *
+     * @param {string} expectedLabel
+     * @returns {Promise<void>}
+     */
+    dequeue: async expectedLabel => {
+      queue.length > 0 ||
+        Fail`got empty queue when expecting ${q(expectedLabel)}`;
+      const [label, resolve, reject] = /** @type {any} */ (queue.shift());
+      if (label === expectedLabel) {
+        resolve();
+        return;
+      }
+      reject(Error(`expecting to dequeue ${expectedLabel} but saw ${label}`));
+      Fail`expecting to dequeue ${q(expectedLabel)} but saw ${q(label)}`;
+    },
+    listen: async () => {
+      await E(boundPort).addListener(listener);
+    },
+    getLocalAddress: async () => {
+      return E(boundPort).getLocalAddress();
+    },
+  });
+
+  return { creatorFacet };
+};