fix: implement epochs and make tolerant of restarts

packages/SwingSet/package.json

@@ -29,6 +29,7 @@
     "@agoric/assert": "^0.0.9",
     "@agoric/babel-parser": "^7.6.4",
     "@agoric/bundle-source": "^1.1.7",
+    "@agoric/captp": "^1.4.0",
     "@agoric/eventual-send": "^0.10.0",
     "@agoric/import-bundle": "^0.0.9",
     "@agoric/install-ses": "^0.3.0",

packages/SwingSet/src/devices/plugin-src.js

@@ -11,13 +11,15 @@ export function buildRootDeviceNode(tools) {
   const senders = {};
   // Take a shallow copy so that these are not frozen.
   const connectedMods = restart ? [...restart.connectedMods] : [];
+  const nextEpochs = restart ? [...restart.nextEpochs] : [];
   const connectedState = restart ? [...restart.connectedState] : [];
 
   function saveState() {
     setDeviceState(
       harden({
         registeredReceiver,
         // Take a shallow copy so that these are not frozen.
+        nextEpochs: [...nextEpochs],
         connectedMods: [...connectedMods],
         connectedState: [...connectedState],
       }),
@@ -38,13 +40,15 @@ export function buildRootDeviceNode(tools) {
       // Allocate this module first.
       if (connectedMods[index] === undefined) {
         connectedMods[index] = mod;
-        saveState();
       }
       if (connectedMods[index] !== mod) {
         throw TypeError(
           `Index ${index} is already allocated to ${connectedMods[index]}, not ${mod}`,
         );
       }
+      const epoch = nextEpochs[index] || 0;
+      nextEpochs[index] = epoch + 1;
+      saveState();
 
       const modNS = endowments.require(mod);
       const receiver = obj => {
@@ -69,7 +73,7 @@ export function buildRootDeviceNode(tools) {
       );
 
       // Establish a CapTP connection.
-      const { dispatch } = makeCapTP(mod, receiver, bootstrap);
+      const { dispatch } = makeCapTP(mod, receiver, bootstrap, { epoch });
 
       // Save the dispatch function for later.
       senders[index] = dispatch;
@@ -82,28 +86,35 @@ export function buildRootDeviceNode(tools) {
 
   function send(index, obj) {
     const mod = connectedMods[index];
-    console.info('send', index, obj, mod);
+    // console.info('send', index, obj, mod);
     if (!mod) {
       throw TypeError(`No module associated with ${index}`);
     }
     let sender = senders[index];
     if (!sender) {
-      // Lazily create a sender.
-      console.info('Destroying', index);
-      SO(registeredReceiver).abort(index);
+      // Lazily create a fresh sender.
       connect(mod, index);
       sender = senders[index];
     }
     // Now actually send.
     sender(obj);
   }
 
+  endowments.registerResetter(() => {
+    connectedMods.forEach((mod, index) => {
+      if (mod) {
+        // console.info('Startup resetting', index, mod, nextEpochs[index]);
+        SO(registeredReceiver).reset(index, nextEpochs[index]);
+      }
+    });
+  });
+
   return harden({
     connect,
     send,
     registerReceiver(receiver) {
       if (registeredReceiver) {
-        throw Error(`registerd receiver already set`);
+        throw Error(`registered receiver already set`);
       }
       registeredReceiver = receiver;
       saveState();

packages/SwingSet/src/devices/plugin.js

@@ -1,10 +1,24 @@
 export function buildPlugin(pluginRequire, queueThunkForKernel) {
   const srcPath = require.resolve('./plugin-src');
+  let resetter;
+
+  function reset() {
+    const init = resetter;
+    if (init) {
+      resetter = undefined;
+      init();
+    }
+  }
+
+  function registerResetter(init) {
+    resetter = init;
+  }
 
   // srcPath and endowments are provided to buildRootDeviceNode() for use
   // during configuration.
   return {
     srcPath,
-    endowments: { require: pluginRequire, queueThunkForKernel },
+    endowments: { require: pluginRequire, queueThunkForKernel, registerResetter },
+    reset,
   };
 }

packages/captp/lib/captp.js

@@ -24,6 +24,7 @@ export { E };
  * @property {(err: any) => void} onReject
  * @property {typeof defaultRemotable} Remotable
  * @property {typeof defaultMakeMarshal} makeMarshal
+ * @property {number} epoch
  */
 /**
  * Create a CapTP connection.
@@ -38,6 +39,7 @@ export function makeCapTP(ourId, rawSend, bootstrapObj = undefined, opts = {}) {
     onReject = err => console.error('CapTP', ourId, 'exception:', err),
     Remotable = defaultRemotable,
     makeMarshal = defaultMakeMarshal,
+    epoch = undefined,
   } = opts;
 
   const disconnectReason = id =>
@@ -179,6 +181,7 @@ export function makeCapTP(ourId, rawSend, bootstrapObj = undefined, opts = {}) {
         const [questionID, pr] = makeQuestion();
         send({
           type: 'CTP_CALL',
+          epoch,
           questionID,
           target,
           method: serialize(harden([prop])),
@@ -193,6 +196,7 @@ export function makeCapTP(ourId, rawSend, bootstrapObj = undefined, opts = {}) {
         const [questionID, pr] = makeQuestion();
         send({
           type: 'CTP_CALL',
+          epoch,
           questionID,
           target,
           method: serialize(harden([prop, args])),
@@ -259,6 +263,7 @@ export function makeCapTP(ourId, rawSend, bootstrapObj = undefined, opts = {}) {
       answers.set(questionID, bootstrap);
       return send({
         type: 'CTP_RETURN',
+        epoch,
         answerID: questionID,
         result: serialize(bootstrap),
       });
@@ -296,13 +301,15 @@ export function makeCapTP(ourId, rawSend, bootstrapObj = undefined, opts = {}) {
         .then(res =>
           send({
             type: 'CTP_RETURN',
+            epoch,
             answerID: questionID,
             result: serialize(harden(res)),
           }),
         )
         .catch(rej =>
           send({
             type: 'CTP_RETURN',
+            epoch,
             answerID: questionID,
             exception: serialize(harden(rej)),
           }),
@@ -357,6 +364,7 @@ export function makeCapTP(ourId, rawSend, bootstrapObj = undefined, opts = {}) {
     const [questionID, pr] = makeQuestion();
     send({
       type: 'CTP_BOOTSTRAP',
+      epoch,
       questionID,
     });
     return harden(pr.p);
@@ -383,7 +391,7 @@ export function makeCapTP(ourId, rawSend, bootstrapObj = undefined, opts = {}) {
 
   // Abort a connection.
   const abort = (reason = undefined) => {
-    dispatch({ type: 'CTP_DISCONNECT', reason });
+    dispatch({ type: 'CTP_DISCONNECT', epoch, reason });
   };
 
   return harden({ abort, dispatch, getBootstrap, serialize, unserialize });

packages/cosmic-swingset/lib/ag-solo/fake-chain.js

@@ -27,12 +27,13 @@ async function makeMapStorage(file) {
     await fs.promises.writeFile(file, json);
   };
 
+  let obj = {};
   try {
     content = await fs.promises.readFile(file);
+    obj = JSON.parse(content);
   } catch (e) {
     return map;
   }
-  const obj = JSON.parse(content);
   Object.entries(obj).forEach(([k, v]) => map.set(k, importMailbox(v)));
 
   return map;

packages/cosmic-swingset/lib/ag-solo/hello.js

@@ -9,7 +9,7 @@ export const bootPlugin = ({ getState, setState }) => {
       state = { count: state.count + 1 };
       setState(state);
       await fs.writeFile(dataFile, `Said hello ${state.count} times\n`);
-      return `Hello, ${name}`;
+      return `Hello, ${name}!`;
     },
   });
 };

packages/cosmic-swingset/lib/ag-solo/start.js

@@ -223,6 +223,10 @@ async function buildSwingset(
       intervalMillis = interval;
       setTimeout(queuedMoveTimeForward, intervalMillis);
     },
+    resetOutdatedState: withInputQueue(() => {
+      plugin.reset();
+      return processKernel();
+    }),
   };
 }
 
@@ -259,6 +263,7 @@ export default async function start(basedir, argv) {
     deliverInboundCommand,
     deliverOutbound,
     startTimer,
+    resetOutdatedState,
   } = d;
 
   let hostport;
@@ -313,6 +318,7 @@ export default async function start(basedir, argv) {
 
   // Start timer here!
   startTimer(1200);
+  resetOutdatedState();
 
   // Remove wallet traces.
   await unlink('html/wallet').catch(_ => {});