Merge pull request #5487 from Agoric/mfig-chain-streams

feat(casting): `@agoric/casting` MVP

.github/workflows/deployment-test.yml

@@ -37,7 +37,7 @@ jobs:
           cache-key: trusty
       - uses: ./.github/actions/restore-node
         with:
-          node-version: 14.x
+          node-version: 16.x
 
       # Select a branch on loadgen to test against by adding text to the body of the
       # pull request. For example: #loadgen-branch: user-123-update-foo
@@ -105,7 +105,7 @@ jobs:
         env:
           NETWORK_NAME: chaintest
       - uses: actions/upload-artifact@v2
-        if: failure()
+        if: always()
         with:
           name: deployment-test-results-${{ env.NOW }}
           path: /usr/src/agoric-sdk/chaintest/results

.github/workflows/test-all-packages.yml

@@ -142,6 +142,8 @@ jobs:
       # END-TEST-BOILERPLATE
       - name: yarn test (cosmos)
         run: cd golang/cosmos && yarn ${{ steps.vars.outputs.test }}
+      - name: yarn test (casting)
+        run: cd packages/casting && yarn ${{ steps.vars.outputs.test }}
       - name: yarn test (run-protocol)
         run: cd packages/run-protocol && yarn ${{ steps.vars.outputs.test }}
       - name: yarn test (pegasus)

golang/cosmos/x/swingset/keeper/keeper.go

@@ -259,3 +259,11 @@ func (k Keeper) SetMailbox(ctx sdk.Context, peer string, mailbox string) {
 	path := StoragePathMailbox + "." + peer
 	k.vstorageKeeper.LegacySetStorageAndNotify(ctx, path, mailbox)
 }
+
+func (k Keeper) PathToEncodedKey(path string) []byte {
+	return k.vstorageKeeper.PathToEncodedKey(path)
+}
+
+func (k Keeper) GetStoreName() string {
+	return k.vstorageKeeper.GetStoreName()
+}

golang/cosmos/x/vstorage/vstorage.go

@@ -20,6 +20,11 @@ type vstorageMessage struct {
 	Value  string `json:"value"`
 }
 
+type vstorageStoreKey struct {
+	StoreName   string `json:"storeName"`
+	StoreSubkey string `json:"storeSubkey"`
+}
+
 func NewStorageHandler(keeper Keeper) vstorageHandler {
 	return vstorageHandler{keeper: keeper}
 }
@@ -67,6 +72,17 @@ func (sh vstorageHandler) Receive(cctx *vm.ControllerContext, str string) (ret s
 		}
 		return string(bz), nil
 
+	case "getStoreKey":
+		value := vstorageStoreKey{
+			StoreName:   keeper.GetStoreName(),
+			StoreSubkey: string(keeper.PathToEncodedKey(msg.Path)),
+		}
+		bz, err := json.Marshal(value)
+		if err != nil {
+			return "", err
+		}
+		return string(bz), nil
+
 	case "has":
 		value := keeper.GetData(cctx.Context, msg.Path)
 		if value == "" {

packages/agoric-cli/package.json

@@ -29,11 +29,13 @@
   "dependencies": {
     "@agoric/access-token": "^0.4.18",
     "@agoric/assert": "^0.4.0",
+    "@agoric/casting": "^0.1.0",
     "@agoric/nat": "^4.1.0",
     "@endo/bundle-source": "^2.2.0",
     "@endo/captp": "^2.0.7",
     "@endo/compartment-mapper": "^0.7.5",
     "@endo/init": "^0.5.41",
+    "@endo/marshal": "^0.6.7",
     "@endo/promise-kit": "^0.2.41",
     "@iarna/toml": "^2.2.3",
     "anylogger": "^0.21.0",

packages/agoric-cli/src/entrypoint.js

@@ -4,6 +4,7 @@
 
 import '@endo/init/pre.js';
 import 'esm';
+import '@agoric/casting/node-fetch-shim.js';
 import '@endo/init';
 
 import path from 'path';

packages/agoric-cli/src/follow.js

@@ -0,0 +1,121 @@
+// @ts-check
+import process from 'process';
+import { Far } from '@endo/marshal';
+import { decodeToJustin } from '@endo/marshal/src/marshal-justin.js';
+
+import {
+  delay,
+  iterateLatest,
+  makeFollower,
+  makeLeader,
+  makeCastingSpec,
+} from '@agoric/casting';
+
+export default async function followerMain(progname, rawArgs, powers, opts) {
+  const { anylogger } = powers;
+  const console = anylogger('agoric:follower');
+
+  const {
+    integrity,
+    output,
+    bootstrap = 'http://localhost:26657',
+    verbose,
+    sleep,
+  } = opts;
+
+  /** @type {import('@agoric/casting').FollowerOptions} */
+  const followerOptions = {
+    integrity,
+  };
+
+  /** @type {(buf: any) => any} */
+  let formatOutput;
+  switch (output) {
+    case 'justinlines':
+    case 'justin': {
+      followerOptions.unserializer = null;
+      const pretty = !output.endsWith('lines');
+      formatOutput = ({ body }) => {
+        const encoded = JSON.parse(body);
+        return decodeToJustin(encoded, pretty);
+      };
+      break;
+    }
+    case 'jsonlines':
+    case 'json': {
+      const spaces = output.endsWith('lines') ? undefined : 2;
+      const bigintToStringReplacer = (_, arg) => {
+        if (typeof arg === 'bigint') {
+          return `${arg}`;
+        }
+        return arg;
+      };
+      formatOutput = obj => JSON.stringify(obj, bigintToStringReplacer, spaces);
+      break;
+    }
+    case 'hex': {
+      // Dump as hex strings.
+      followerOptions.decode = buf => buf;
+      followerOptions.unserializer = null;
+      formatOutput = buf =>
+        buf.reduce((acc, b) => acc + b.toString(16).padStart(2, '0'), '');
+      break;
+    }
+    case 'text': {
+      followerOptions.decode = buf => new TextDecoder().decode(buf);
+      followerOptions.unserializer = null;
+      formatOutput = buf => buf;
+      break;
+    }
+    default: {
+      console.error(`Unknown output format: ${output}`);
+      return 1;
+    }
+  }
+
+  if (integrity !== 'none') {
+    followerOptions.crasher = Far('follower crasher', {
+      crash: (...args) => {
+        console.error(...args);
+        console.warn(`You are running with '--integrity=${integrity}'`);
+        console.warn(
+          `If you trust your RPC nodes, you can turn off proofs with '--integrity=none'`,
+        );
+        process.exit(1);
+      },
+    });
+  }
+
+  // TODO: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/
+  /** @type {import('@agoric/casting').LeaderOptions} */
+  const leaderOptions = {
+    retryCallback: (e, _attempt) => {
+      verbose && console.warn('Retrying due to:', e);
+      return delay(1000 + Math.random() * 1000);
+    },
+    keepPolling: async () => {
+      let toSleep = sleep * 1000;
+      if (toSleep <= 0) {
+        toSleep = (5 + Math.random()) * 1000;
+      }
+      await delay(toSleep);
+      return true;
+    },
+  };
+
+  const [_cmd, ...specs] = rawArgs;
+
+  verbose && console.warn('Creating leader for', bootstrap);
+  const leader = makeLeader(bootstrap, leaderOptions);
+  await Promise.all(
+    specs.map(async spec => {
+      verbose && console.warn('Following', spec);
+      const castingSpec = makeCastingSpec(spec);
+      const follower = makeFollower(leader, castingSpec, followerOptions);
+      for await (const { value } of iterateLatest(follower)) {
+        process.stdout.write(`${formatOutput(value)}\n`);
+      }
+    }),
+  );
+  return 0;
+}

packages/agoric-cli/src/main.js

@@ -8,6 +8,7 @@ import initMain from './init.js';
 import installMain from './install.js';
 import setDefaultsMain from './set-defaults.js';
 import startMain from './start.js';
+import followMain from './follow.js';
 import walletMain from './open.js';
 
 const DEFAULT_DAPP_TEMPLATE = 'dapp-fungible-faucet';
@@ -160,6 +161,58 @@ const main = async (progname, rawArgs, powers) => {
       return subMain(installMain, ['install', forceSdkVersion], opts);
     });
 
+  program
+    .command('follow <path-spec...>')
+    .description('follow an Agoric Casting leader')
+    .option(
+      '--integrity <strict | optimistic | none>',
+      'set integrity mode',
+      value => {
+        assert(
+          ['strict', 'optimistic', 'none'].includes(value),
+          X`--integrity must be one of 'strict', 'optimistic', or 'none'`,
+          TypeError,
+        );
+        return value;
+      },
+      'optimistic',
+    )
+    .option(
+      '--sleep <seconds>',
+      'sleep <seconds> between polling (may be fractional)',
+      value => {
+        const num = Number(value);
+        assert.equal(`${num}`, value, X`--sleep must be a number`, TypeError);
+        return num;
+      },
+      0,
+    )
+    .option(
+      '-o, --output <format>',
+      'value output format',
+      value => {
+        assert(
+          [
+            'hex',
+            'justin',
+            'justinlines',
+            'json',
+            'jsonlines',
+            'text',
+          ].includes(value),
+          X`--output must be one of 'hex', 'justin', 'justinlines', 'json', 'jsonlines', or 'text'`,
+          TypeError,
+        );
+        return value;
+      },
+      'justin',
+    )
+    .option('-B, --bootstrap <config>', 'network bootstrap configuration')
+    .action(async (pathSpecs, cmd) => {
+      const opts = { ...program.opts(), ...cmd.opts() };
+      return subMain(followMain, ['follow', ...pathSpecs], opts);
+    });
+
   const addRunOptions = cmd =>
     cmd
       .option(

packages/agoric-cli/src/sdk-package-names.js

@@ -4,6 +4,7 @@
 export default [
   "@agoric/access-token",
   "@agoric/assert",
+  "@agoric/casting",
   "@agoric/cosmic-swingset",
   "@agoric/cosmos",
   "@agoric/deploy-script-support",

packages/casting/README.md

@@ -0,0 +1,76 @@
+# Agoric Casting
+
+This [Agoric](https://agoric.com) Casting package follows ocap broadcasts in a
+flexible, future-proof way.
+
+TL;DR: You can run `yarn demo`, or to follow a mailbox castingSpec do:
+```sh
+npx agoric follow -Bhttp://devnet.agoric.net/network-config :mailbox.agoric1foobarbaz -otext
+```
+
+An example of following an on-chain mailbox in code (using this package) is:
+
+```js
+// First, obtain a Hardened JS environment via Endo.
+import '@endo/init/pre-remoting.js'; // needed only for the next line
+import '@agoric/castingSpec/node-fetch-shim.js'; // needed for Node.js
+import '@endo/init';
+
+import {
+  iterateLatest,
+  makeFollower,
+  makeLeader,
+  makeCastingSpec,
+} from '@agoric/casting';
+
+// Iterate over a mailbox follower on the devnet.
+const leader = makeLeader('https://devnet.agoric.net/network-config');
+const castingSpec = makeCastingSpec(':mailbox.agoric1foobarbaz');
+const follower = makeFollower(leader, castingSpec);
+for await (const { value } of iterateLatest(follower)) {
+  console.log(`here's a mailbox value`, value);
+}
+```
+
+## Follower options
+
+The `followerOpts` argument in `makeFollower(leader, key, followerOpts)` provides an optional bag of options:
+- the `integrity` option, which has three possibilities:
+  - `'strict'` - release data only after proving it was validated (may incur waits for one block's data to be validated in the next block),
+  - `'optimistic'` (default) - release data immediately, but may crash the follower in the future if an already-released value could not be proven,
+  - `'none'` - release data immediately without validation
+- the `decode` option is a function to translate `buf: Uint8Array` into `data: string`
+  - (default) - interpret buf as a utf-8 string, then `JSON.parse` it
+- the `unserializer` option can be
+  - (default) - release unserialized objects using `@agoric/marshal`'s `makeMarshal()`
+  - `null` - don't additionally unserialize data before releasing it
+  - any unserializer object supporting `E(unserializer).unserialize(data)`
+- the `crasher` option can be
+  - `null` (default) follower failures only propagate an exception/rejection
+  - any crasher object supporting `E(crasher).crash(reason)`
+
+## Behind the scenes
+
+- the network config contains enough information to obtain Tendermint RPC nodes
+  for a given Agoric network.  You can use `makeLeaderFromRpcAddresses` directly
+  if you want to avoid fetching a network-config.
+- each follower uses periodic CosmJS state polling (every X milliseconds) which
+  can be refreshed more expediently via a Tendermint subscription to the
+  corresponding `state_change` event
+- published (string) values are automatically unmarshalled, but without object references.   a custom `marshaller` for your application.
+- the `iterateRecent` adapter transforms a follower into a local async iterator
+  that produces only the last queried value (with no history reconstruction)
+
+## Status
+
+This package currently depends on:
+- Hardened Javascript
+- `@agoric/notify` async iterable adapters to implement `iterateLatest`
+- `@endo/marshal` for default object unserialization
+- [CosmJS](https://github.com/cosmos/cosmjs) for proof verification, although [it does not yet support light client tracking of the validator set](https://github.com/cosmos/cosmjs/issues/492)
+- a bespoke follower of [WebSocket Tendermint events](https://docs.tendermint.com/master/tendermint-core/subscription.html#legacy-streaming-api)
+
+Short-term goals:
+- integrate the new [SharedSubscription API](https://github.com/Agoric/agoric-sdk/pull/5418#discussion_r886253328) from the [Agoric/agoric-sdk#5418 `makePublisherKit` PR](https://github.com/Agoric/agoric-sdk/pull/5418)
+- support `iterateEach` with the [lossless forward iteration algorithm](https://github.com/Agoric/agoric-sdk/blob/mfig-vstream/golang/cosmos/x/vstream/spec/01_concepts.md#forward-iteration-lossless-history) via the [Agoric/agoric-sdk#5466 `x/vstream` PR](https://github.com/Agoric/agoric-sdk/pull/5466)
+- upgrade to the [Tendermint event log API](https://docs.tendermint.com/master/tendermint-core/subscription.html#event-log-api) when Tendermint v0.36 is supported by the Agoric chain

packages/casting/jsconfig.json

@@ -0,0 +1,24 @@
+// This file can contain .js-specific Typescript compiler config.
+{
+  "compilerOptions": {
+    "target": "esnext",
+    "module": "esnext",
+
+    "noEmit": true,
+/*
+    // The following flags are for creating .d.ts files:
+    "noEmit": false,
+    "declaration": true,
+    "emitDeclarationOnly": true,
+*/
+    "downlevelIteration": true,
+    "strictNullChecks": true,
+    "moduleResolution": "node",
+  },
+  "include": [
+    "*.js",
+    "public/**/*.js",
+    "src/**/*.js",
+    "test/**/*.js",
+  ],
+}

packages/casting/node-fetch-shim.js

@@ -0,0 +1,4 @@
+/* global globalThis */
+import fetch from 'node-fetch';
+
+globalThis.fetch = fetch;