feat!: Support publishing and consuming chain storage stream cells

.github/workflows/test-scripts.yml

@@ -0,0 +1,14 @@
+name: Run scripts tests
+
+on:
+  pull_request:
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  run-scripts-tests:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - run: scripts/test/test.sh

golang/cosmos/x/vstorage/keeper/keeper.go

@@ -2,6 +2,8 @@ package keeper
 
 import (
 	"bytes"
+	"encoding/json"
+	"strconv"
 	"strings"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
@@ -11,6 +13,16 @@ import (
 	"github.com/Agoric/agoric-sdk/golang/cosmos/x/vstorage/types"
 )
 
+// StreamCell is an envelope representing a sequence of values written at a path in a single block.
+// It is persisted to storage as a { "blockHeight": "<digits>", "values": ["...", ...] } JSON text
+// that off-chain consumers rely upon.
+// Many of those consumers *also* rely upon the strings of "values" being valid JSON text
+// (cf. scripts/get-flattened-publication.sh), but we do not enforce that in this package.
+type StreamCell struct {
+	BlockHeight string   `json:"blockHeight"`
+	Values      []string `json:"values"`
+}
+
 // Keeper maintains the link to data storage and exposes getter/setter methods
 // for the various parts of the state machine
 type Keeper struct {
@@ -127,6 +139,30 @@ func (k Keeper) SetStorageAndNotify(ctx sdk.Context, path, value string) {
 	)
 }
 
+func (k Keeper) AppendStorageValueAndNotify(ctx sdk.Context, path, value string) error {
+	blockHeight := strconv.FormatInt(ctx.BlockHeight(), 10)
+
+	// Preserve correctly-formatted data within the current block,
+	// otherwise initialize a blank cell.
+	currentData := k.GetData(ctx, path)
+	var cell StreamCell
+	_ = json.Unmarshal([]byte(currentData), &cell)
+	if cell.BlockHeight != blockHeight {
+		cell = StreamCell{BlockHeight: blockHeight, Values: make([]string, 0, 1)}
+	}
+
+	// Append the new value.
+	cell.Values = append(cell.Values, value)
+
+	// Perform the write.
+	bz, err := json.Marshal(cell)
+	if err != nil {
+		return err
+	}
+	k.SetStorageAndNotify(ctx, path, string(bz))
+	return nil
+}
+
 func componentsToPath(components []string) string {
 	return strings.Join(components, types.PathSeparator)
 }

golang/cosmos/x/vstorage/vstorage.go

@@ -61,7 +61,15 @@ func (sh vstorageHandler) Receive(cctx *vm.ControllerContext, str string) (ret s
 		keeper.SetStorageAndNotify(cctx.Context, msg.Path, msg.Value)
 		return "true", nil
 
+	case "append":
+		err = keeper.AppendStorageValueAndNotify(cctx.Context, msg.Path, msg.Value)
+		if err != nil {
+			return "", err
+		}
+		return "true", nil
+
 	case "get":
+		// Note that "get" does not (currently) unwrap a StreamCell.
 		value := keeper.GetData(cctx.Context, msg.Path)
 		if value == "" {
 			return "null", nil

packages/casting/src/follower-cosmjs.js

@@ -305,7 +305,7 @@ export const makeCosmjsFollower = (
      * @param {import('./types').CastingChange} allegedChange
      */
     const tryQueryAndUpdate = async allegedChange => {
-      const committer = prepareUpdateInOrder();
+      let committer = prepareUpdateInOrder();
 
       // Make an unproven query if we have no alleged value.
       const { values: allegedValues, blockHeight: allegedBlockHeight } =
@@ -334,18 +334,32 @@ export const makeCosmjsFollower = (
         }
       }
       lastBuf = buf;
-      const data = decode(buf);
-      if (!unserializer) {
-        /** @type {T} */
-        const value = data;
-        committer.commit({ value });
-        return;
+      let streamCell = decode(buf);
+      // Upgrade a naked value to a JSON stream cell if necessary.
+      if (
+        streamCell.blockHeight === undefined ||
+        streamCell.values === undefined
+      ) {
+        streamCell = { values: [JSON.stringify(streamCell)] };
       }
-      const value = await E(unserializer).unserialize(data);
-      if (!committer.isValid()) {
-        return;
+      for (let i = 0; i < streamCell.values.length; i += 1) {
+        const data = JSON.parse(streamCell.values[i]);
+        const isLast = i + 1 === streamCell.values.length;
+        const value = /** @type {T} */ (
+          unserializer
+            ? // eslint-disable-next-line no-await-in-loop,@jessie.js/no-nested-await
+              await E(unserializer).unserialize(data)
+            : data
+        );
+        // QUESTION: How would reach a point where this `isValid()` fails,
+        // and what is the proper handling?
+        if (!unserializer || committer.isValid()) {
+          committer.commit({ value });
+          if (!isLast) {
+            committer = prepareUpdateInOrder();
+          }
+        }
       }
-      committer.commit({ value });
     };
 
     const changeFollower = E(leader).watchCasting(castingSpecP);
@@ -362,6 +376,7 @@ export const makeCosmjsFollower = (
           return;
         }
         harden(allegedChange);
+        // eslint-disable-next-line @jessie.js/no-nested-await
         await queryAndUpdateOnce(allegedChange);
       }
     };

packages/casting/test/fake-rpc-server.js

@@ -64,10 +64,19 @@ const fakeStatusResult = {
   },
 };
 
-export const startFakeServer = (t, fakeValues, marshaller = makeMarshal()) => {
+/** @typedef {Partial<import('ava').ExecutionContext<{cleanups: Array<() => void>}>> & {context}} FakeServerTestContext */
+/**
+ * @param {FakeServerTestContext} t
+ * @param {Array<{any}>} fakeValues
+ * @param {object} [options]
+ * @param {Marshaller} [options.marshaller]
+ * @param {number} [options.batchSize] count of stream-cell results per response, or 0/absent to return lone naked values
+ */
+export const startFakeServer = (t, fakeValues, options = {}) => {
   const { log = console.log } = t;
   lastPort += 1;
   const PORT = lastPort;
+  const { marshaller = makeMarshal(), batchSize = 0 } = options;
   return new Promise(resolve => {
     log('starting http server on port', PORT);
     const app = express();
@@ -97,6 +106,8 @@ export const startFakeServer = (t, fakeValues, marshaller = makeMarshal()) => {
       buf.set(ascii, dataPrefix.length);
       return toBase64(buf);
     };
+    let blockHeight = 74863;
+    let responseValueBase64;
     app.post('/tendermint-rpc', (req, res) => {
       log('received', req.path, req.body, req.params);
       const reply = result => {
@@ -114,10 +125,23 @@ export const startFakeServer = (t, fakeValues, marshaller = makeMarshal()) => {
           break;
         }
         case 'abci_query': {
-          const value =
-            fakeValues.length === 0
-              ? null
-              : encode(marshaller.serialize(fakeValues.shift()));
+          blockHeight += 2;
+          const values = fakeValues.splice(0, Math.max(1, batchSize));
+          if (values.length > 0) {
+            if (batchSize > 0) {
+              // Return a JSON stream cell.
+              const serializedValues = values.map(val =>
+                JSON.stringify(marshaller.serialize(val)),
+              );
+              responseValueBase64 = encode({
+                blockHeight: String(blockHeight - 1),
+                values: serializedValues,
+              });
+            } else {
+              // Return a single naked value.
+              responseValueBase64 = encode(marshaller.serialize(values[0]));
+            }
+          }
           const result = {
             response: {
               code: 0,
@@ -127,9 +151,9 @@ export const startFakeServer = (t, fakeValues, marshaller = makeMarshal()) => {
               key: Buffer.from(
                 'swingset/data:mailbox.agoric1foobarbaz',
               ).toString('base64'),
-              value,
+              value: responseValueBase64,
               proofOps: null,
-              height: '74863',
+              height: String(blockHeight),
               codespace: '',
             },
           };
@@ -226,10 +250,12 @@ export const develop = async () => {
       unserialize({ body: jsonMarshalled, slots: [] }),
     ),
   );
-  const mockT = {
-    log: console.log,
-    context: { cleanups: [] },
-  };
+  const mockT = /** @type {FakeServerTestContext} */ (
+    /** @type {unknown} */ ({
+      log: console.log,
+      context: { cleanups: [] },
+    })
+  );
   const PORT = await startFakeServer(mockT, [...fakeValues]);
   console.log(
     `Try this in another terminal:

packages/casting/test/test-mvp.js

@@ -12,39 +12,60 @@ import {
 import { delay } from '../src/defaults.js';
 import { startFakeServer } from './fake-rpc-server.js';
 
-test('happy path', async t => {
-  const expected = ['latest', 'later', 'done'];
-  t.plan(expected.length);
-  const PORT = await t.context.startServer(t, [...expected]);
-  /** @type {import('../src/types.js').LeaderOptions} */
-  const lo = {
-    retryCallback: null, // fail fast, no retries
-    keepPolling: () => delay(200).then(() => true), // poll really quickly
-    jitter: null, // no jitter
-  };
-  /** @type {import('../src/types.js').FollowerOptions} */
-  const so = {
-    proof: 'none',
-  };
+// TODO: Replace with test.macro({title, exec}).
+const testHappyPath = (label, ...input) => {
+  // eslint-disable-next-line no-shadow
+  const title = label => `happy path ${label}`;
+  const makeExec =
+    ({ fakeValues, options }) =>
+    async t => {
+      const expected = fakeValues;
+      t.plan(expected.length);
+      const PORT = await t.context.startFakeServer(t, [...expected], options);
+      /** @type {import('../src/types.js').LeaderOptions} */
+      const lo = {
+        retryCallback: null, // fail fast, no retries
+        keepPolling: () => delay(200).then(() => true), // poll really quickly
+        jitter: null, // no jitter
+      };
+      /** @type {import('../src/types.js').FollowerOptions} */
+      const so = {
+        proof: 'none',
+      };
 
-  // The rest of this test is taken almost verbatim from the README.md, with
-  // some minor modifications (testLeaderOptions and deepEqual).
-  const leader = makeLeader(`http://localhost:${PORT}/network-config`, lo);
-  const castingSpec = makeCastingSpec(':mailbox.agoric1foobarbaz');
-  const follower = await makeFollower(castingSpec, leader, so);
-  for await (const { value } of iterateLatest(follower)) {
-    t.log(`here's a mailbox value`, value);
+      // The rest of this test is taken almost verbatim from the README.md, with
+      // some minor modifications (testLeaderOptions and deepEqual).
+      const leader = makeLeader(`http://localhost:${PORT}/network-config`, lo);
+      const castingSpec = makeCastingSpec(':mailbox.agoric1foobarbaz');
+      const follower = await makeFollower(castingSpec, leader, so);
+      for await (const { value } of iterateLatest(follower)) {
+        t.log(`here's a mailbox value`, value);
 
-    // The rest here is to drive the test.
-    t.deepEqual(value, expected.shift());
-    if (expected.length === 0) {
-      break;
-    }
-  }
+        // The rest here is to drive the test.
+        t.deepEqual(value, expected.shift());
+        if (expected.length === 0) {
+          break;
+        }
+      }
+    };
+  test(title(label), makeExec(...input));
+};
+
+testHappyPath('naked values', {
+  fakeValues: ['latest', 'later', 'done'],
+  options: {},
+});
+testHappyPath('batchSize=1', {
+  fakeValues: ['latest', 'later', 'done'],
+  options: { batchSize: 1 },
+});
+testHappyPath('batchSize=2', {
+  fakeValues: ['latest', 'later', 'done'],
+  options: { batchSize: 2 },
 });
 
 test('bad network config', async t => {
-  const PORT = await t.context.startServer(t, []);
+  const PORT = await t.context.startFakeServer(t, []);
   await t.throwsAsync(
     () =>
       makeLeader(`http://localhost:${PORT}/bad-network-config`, {
@@ -58,7 +79,7 @@ test('bad network config', async t => {
 });
 
 test('missing rpc server', async t => {
-  const PORT = await t.context.startServer(t, []);
+  const PORT = await t.context.startFakeServer(t, []);
   await t.throwsAsync(
     () =>
       makeLeader(`http://localhost:${PORT}/missing-network-config`, {
@@ -83,7 +104,7 @@ test('unrecognized proof', async t => {
 
 test.before(t => {
   t.context.cleanups = [];
-  t.context.startServer = startFakeServer;
+  t.context.startFakeServer = startFakeServer;
 });
 
 test.after(t => {

packages/notifier/src/types.js

@@ -238,7 +238,7 @@
  * @property {(data: string) => void} setValue publishes some data
  * @property {() => ERef<VStorageKey>} getStoreKey get the
  * externally-reachable store key for this storage item
- * @property {(subPath: string) => StorageNode} makeChildNode
+ * @property {(subPath: string, options?: {sequence?: boolean}) => StorageNode} makeChildNode
  */
 
 /**

packages/vats/src/core/chain-behaviors.js

@@ -312,6 +312,7 @@ export const makeChainStorage = async ({
     bridgeManager,
     BRIDGE_ID.STORAGE,
     ROOT_PATH,
+    { sequence: true },
   );
   chainStorageP.resolve(rootNodeP);
 };