fix: reject all sends when the connection is closed

Agoric · May 2, 2020 · 61b0975 · 61b0975
1 parent 9f65899
commit 61b0975
Showing 1 changed file with 29 additions and 3 deletions.
diff --git a/packages/SwingSet/src/vats/network/network.js b/packages/SwingSet/src/vats/network/network.js
@@ -2,6 +2,7 @@
 import makeStore from '@agoric/store';
 import rawHarden from '@agoric/harden';
 import { E as defaultE } from '@agoric/eventual-send';
+import { producePromise } from '@agoric/produce-promise';
 import { toBytes } from './bytes';
 
 const harden = /** @type {<T>(x: T) => T} */ (rawHarden);
@@ -13,6 +14,11 @@ export const ENDPOINT_SEPARATOR = '/';
  * @typedef {import('@agoric/store').Store<T,U>} Store
  */
 
+/**
+ * @template T,U
+ * @typedef {import('@agoric/produce-promise').PromiseRecord<T, U>} PromiseRecord
+ */
+
 /**
  * @typedef {import('./bytes').Bytes} Bytes
  * @typedef {import('./bytes').Data} Data
@@ -108,6 +114,10 @@ export const makeConnection = (
   E = defaultE,
 ) => {
   let closed;
+  /**
+   * @type {Set<PromiseRecord<Bytes,any>>}
+   */
+  const pendingAcks = new Set();
   /**
    * @type {Connection}
    */
@@ -124,6 +134,10 @@ export const makeConnection = (
       }
       current.delete(connection);
       closed = Error('Connection closed');
+      for (const ackDeferred of [...pendingAcks.values()]) {
+        pendingAcks.delete(ackDeferred);
+        ackDeferred.reject(closed);
+      }
       await E(handler)
         .onClose(connection, undefined, handler)
         .catch(rethrowUnlessMissing);
@@ -134,10 +148,22 @@ export const makeConnection = (
         throw closed;
       }
       const bytes = toBytes(data);
-      const ack = await E(handler)
+      const ackDeferred = producePromise();
+      pendingAcks.add(ackDeferred);
+      E(handler)
         .onReceive(connection, bytes, handler)
-        .catch(err => rethrowUnlessMissing(err) || '');
-      return toBytes(ack);
+        .catch(err => rethrowUnlessMissing(err) || '')
+        .then(
+          ack => {
+            pendingAcks.delete(ackDeferred);
+            ackDeferred.resolve(toBytes(ack));
+          },
+          err => {
+            pendingAcks.delete(ackDeferred);
+            ackDeferred.reject(err);
+          },
+        );
+      return ackDeferred.promise;
     },
   });