fix: Connect sequencing issue with Scribe connections

Author: iridiankin

packages/prophet/AuthorityProphet/AuthorityPartitionConnection.js

@@ -29,7 +29,7 @@ export default class AuthorityPartitionConnection extends PartitionConnection {
 
   _doConnect (/* options: ConnectOptions, onError: Function */) {}
 
-  async narrateEventLog (options: ?NarrateOptions = {}): Promise<any> {
+  narrateEventLog (options: ?NarrateOptions = {}): Promise<any> {
     return !options ? undefined : {};
   }
 

packages/prophet/FalseProphet/FalseProphetPartitionConnection.js

@@ -6,7 +6,7 @@ import ValaaReference from "~/raem/ValaaReference";
 import type { VRef } from "~/raem/ValaaReference"; // eslint-disable-line no-duplicate-imports
 
 import PartitionConnection from "~/prophet/api/PartitionConnection";
-import { NarrateOptions, ChronicleOptions, ChronicleRequest } from "~/prophet/api/types";
+import { ChronicleOptions, ChronicleRequest } from "~/prophet/api/types";
 import { initializeAspects, obtainAspect, tryAspect } from "~/prophet/tools/EventAspects";
 import EVENT_VERSION from "~/prophet/tools/EVENT_VERSION";
 

packages/prophet/Scribe/ScribePartitionConnection.js

@@ -88,29 +88,8 @@ export default class ScribePartitionConnection extends PartitionConnection {
           this._prophet._persistedMediaLookup[mediaRawId] = info;
         }
       },
-      () => (options.narrateOptions && this.narrateEventLog(options.narrateOptions)),
-      (narration) => {
-        if (!narration) return narration;
-        const actionCount = Object.values(narration).reduce(
-            (s, log) => s + (Array.isArray(log) ? log.length : 0),
-            options.eventIdBegin || 0);
-        if (!actionCount && (options.newPartition === false)) {
-          throw new Error(`No events found when connecting to an expected existing partition '${
-            this.getPartitionURI().toString()}'`);
-        } else if (actionCount && (options.newPartition === true)) {
-          throw new Error(`Existing events found when trying to create a new partition '${
-            this.getPartitionURI().toString()}'`);
-        }
-        if ((options.requireLatestMediaContents !== false)
-            && (narration.mediaRetrievalStatus || { latestFailures: [] }).latestFailures.length) {
-          // FIXME(iridian): This error temporarily demoted to log error
-          this.outputErrorEvent(new Error(`Failed to connect to partition: encountered ${
-                narration.mediaRetrievalStatus.latestFailures.length
-              } latest media content retrieval failures (and ${
-              ""}options.requireLatestMediaContents does not equal false).`));
-        }
-        return narration;
-      },
+      () => ((options.narrateOptions !== false)
+          && this.narrateEventLog(options.narrateOptions)),
     ], onError);
   }
 

packages/prophet/Scribe/_eventOps.js

@@ -56,7 +56,8 @@ export async function _narrateEventLog (connection: ScribePartitionConnection,
       }));
 
   if ((options.fullNarrate !== true)
-      && ((ret.scribeEventLog || []).length || (ret.scribeCommandQueue || []).length)) {
+      && (options.newPartition
+          || (ret.scribeEventLog || []).length || (ret.scribeCommandQueue || []).length)) {
     connection.logEvent(2, () => [
       "Initiated async upstream narration, local narration results:", ret,
     ]);

packages/prophet/api/PartitionConnection.js

@@ -10,7 +10,8 @@ import { ConnectOptions, MediaInfo, NarrateOptions, ChronicleOptions, ChronicleR
 import Follower from "~/prophet/api/Follower";
 
 import Logger from "~/tools/Logger";
-import { dumpObject, invariantifyArray, invariantifyObject, thenChainEagerly } from "~/tools";
+import { dumpObject, invariantifyArray, invariantifyObject, isPromise, thenChainEagerly }
+    from "~/tools";
 
 /**
  * Interface for sending commands to upstream and registering for downstream truth updates
@@ -123,8 +124,29 @@ export default class PartitionConnection extends Follower {
         "\n\tBegun connecting with options", ...dumpObject(options), ...dumpObject(this)
       ]);
       return (this._activeConnection = thenChainEagerly(
-          this._doConnect(options, onError),
+          this._doConnect(Object.create(options), onError),
           (connectResults) => {
+            if (options.narrateOptions !== false) {
+                const actionCount = Object.values(connectResults).reduce(
+                  (s, log) => s + (Array.isArray(log) ? log.length : 0),
+                  options.eventIdBegin || 0);
+              if (!actionCount && (options.newPartition === false)) {
+                throw new Error(`No events found when connecting to an existing partition '${
+                  this.getPartitionURI().toString()}'`);
+              } else if (actionCount && (options.newPartition === true)) {
+                throw new Error(`Existing events found when trying to create a new partition '${
+                  this.getPartitionURI().toString()}'`);
+              }
+              if ((options.requireLatestMediaContents !== false)
+                  && (connectResults.mediaRetrievalStatus
+                      || { latestFailures: [] }).latestFailures.length) {
+                // FIXME(iridian): This error temporarily demoted to log error
+                this.outputErrorEvent(new Error(`Failed to connect to partition: encountered ${
+                  connectResults.mediaRetrievalStatus.latestFailures.length
+                    } latest media content retrieval failures (and ${
+                    ""}options.requireLatestMediaContents does not equal false).`));
+              }
+            }
             this.warnEvent(1, () => [
               "\n\tDone connecting with results:", connectResults,
               "\n\tstatus:", this.getStatus(),
@@ -135,20 +157,24 @@ export default class PartitionConnection extends Follower {
       ));
     } catch (error) { return onError(error); }
     function errorOnConnect (wrapper, error) {
-      throw this.wrapErrorEvent(error, wrapper, "\n\toptions:", ...dumpObject(options));
+      throw this.wrapErrorEvent(error, wrapper,
+          "\n\toptions:", ...dumpObject(options));
     }
   }
 
   _doConnect (options: ConnectOptions, onError: Function) {
-    if (!this._prophet._upstream) throw new Error("Cannot connect: upstream missing");
+    if (!this._prophet._upstream) {
+      throw new Error("Cannot connect using default _doConnect with no upstream");
+    }
     options.receiveTruths = this.getReceiveTruths(options.receiveTruths);
     options.receiveCommands = this.getReceiveCommands(options.receiveCommands);
-    const postponeNarrateOptions = options.narrateOptions;
+    const postponedNarrateOptions = options.narrateOptions;
     options.narrateOptions = false;
     this.setUpstreamConnection(this._prophet._upstream.acquirePartitionConnection(
         this.getPartitionURI(), options));
     return thenChainEagerly(this._upstreamConnection.getActiveConnection(),
-        () => (postponeNarrateOptions && this.narrateEventLog(postponeNarrateOptions)),
+        () => ((postponedNarrateOptions !== false)
+            && this.narrateEventLog(postponedNarrateOptions)),
         onError);
   }
 
@@ -171,6 +197,7 @@ export default class PartitionConnection extends Follower {
   }
 
   setUpstreamConnection (connection: PartitionConnection) {
+    if (isPromise(connection)) throw new Error("setUpstreamConnection must not be a promise");
     this._upstreamConnection = connection;
   }
 
@@ -196,7 +223,8 @@ export default class PartitionConnection extends Follower {
             this.getName()}`);
       }
       if (this._activeConnection) return this._activeConnection;
-      throw new Error(`Connection not being activated`);
+      throw new Error(
+          `Cannot get an active connection promise from connection which is not being activated`);
     } catch (error) {
       throw this.wrapErrorEvent(error, new Error(`getActiveConnection(${
           requireSynchronous ? "sync" : "async"})`));

packages/prophet/api/Prophet.js

@@ -130,19 +130,18 @@ export default class Prophet extends LogEventGenerator {
         throw new Error(
             "Can't create new partition connection with options.newConnection === false");
       }
-      connection = this._createPartitionConnection(partitionURI, options);
+      connection = this._createPartitionConnection(partitionURI, { ...options, connect: false });
       if (!connection) return undefined;
       connection.addReference();
       this._connections[String(partitionURI)] = connection;
-      if (options.connect !== false) {
-        connection.connect(options); // Initiates the connection but doesn't wait for it to complete.
-      }
+      if (options.connect !== false) connection.connect(options); // Initiate connect but dont wait.
       return connection;
     } catch (error) {
       throw this.wrapErrorEvent(error,
           new Error(`acquirePartitionConnection(${String(partitionURI)})`),
           "\n\toptions:", ...dumpObject(options));
     }
+
       /*
       if (options.newPartition || options.synchronous) {
         if (options.synchronous) return connection;

packages/prophet/test/ProphetTestHarness.js

@@ -131,16 +131,18 @@ export default class ProphetTestHarness extends ScriptTestHarness {
     this.testPartitionURI = options.testPartitionURI
         || (options.testAuthorityURI && createPartitionURI(this.testAuthorityURI, "test_partition"))
         || testPartitionURI;
+    const activeTestPartitionConnection = this.prophet.acquirePartitionConnection(
+        this.testPartitionURI, { newPartition: true }).getActiveConnection();
 
-    this.testPartitionConnection = thenChainEagerly(
-        this.prophet.acquirePartitionConnection(this.testPartitionURI, { newPartition: true })
-        .getActiveConnection(), [
-          (connection) => Promise.all([
-            connection,
-            this.chronicleEvent(createdTestPartitionEntity, { isTruth: true }).getPremiereStory(),
-          ]),
-          ([connection]) => (this.testPartitionConnection = connection),
-        ]);
+    this.testPartitionConnection = thenChainEagerly(activeTestPartitionConnection, [
+      (connection) => {
+        const testPartitionStory =
+            this.chronicleEvent(createdTestPartitionEntity, { isTruth: true })
+            .getPremiereStory();
+        return Promise.all([connection, testPartitionStory]);
+      },
+      ([connection]) => (this.testPartitionConnection = connection),
+    ]);
   }
 
   chronicleEvents (events: EventBase[], ...rest: any) {

packages/raem/VALK/builtinSteppers.js

@@ -817,7 +817,7 @@ function _advanceCapture (valker, thisArgument, vakon, callScope, capturingValke
               : !advanceError ? `call/releaseTransaction (valk caller with active valker)`
               : `call/releaseTransaction({ abort: true }) (valk caller with active valker)`,
           ...((transactionError && advanceError)
-              ? ["\n\t\tabort-cause:", ...dumpObject(advanceError)] : []),
+              ? ["\n\t\tadvance abort cause:", ...dumpObject(advanceError)] : []),
           "\n\tthis:", ...dumpObject(thisArgument),
           "\n\tcallee vakon:", ...dumpKuery(vakon),
           "\n\tscope:", ...dumpObject(callScope),
@@ -846,32 +846,34 @@ function _runCapture (valker, thisArgument, vakon, callScope, capturingValker: V
     advanceError = error;
   }
 
-  let releaseError;
+  let transactionError;
   try {
     if (!advanceError) {
       transaction.releaseTransaction();
       return ret;
     }
     transaction.releaseTransaction({ abort: true });
   } catch (error) {
-    releaseError = error;
+    transactionError = error;
   }
   let opName;
-  if (!advanceError) {
+  if (transactionError) {
+    const prefix = `call/releaseTransaction ${advanceError ? "({ abort: true })" : "()"}`;
     opName = (actualValker !== valker.rootDiscourse)
-        ? `call/releaseTransaction (non-valk caller in active transactional callback ${
-            ""}context)`
-        : `call/releaseTransaction (non-valk caller as outermost context)`;
+        ? `${prefix} (non-valk caller in active transactional callback context)`
+        : `${prefix} (non-valk caller as outermost context)`;
   } else if (actualValker !== valker.rootDiscourse) {
     opName = `call/run (non-valk caller in active transactional callback context)`;
   } else {
     opName = `call/run (non-valk caller as outermost context)`;
-    const connectingMissingPartitions = tryConnectToMissingPartitionsAndThen(error,
+    const connectingMissingPartitions = tryConnectToMissingPartitionsAndThen(advanceError,
         () => _runCapture(valker, thisArgument, vakon, callScope, capturingValker));
     if (connectingMissingPartitions) return connectingMissingPartitions;
   }
   throw capturingValker.addVALKRuntimeErrorStackFrame(
-      actualValker.wrapErrorEvent(advanceError || releaseError, opName,
+      actualValker.wrapErrorEvent(transactionError || advanceError, opName,
+          ...((transactionError && advanceError)
+              ? ["\n\t\tadvance abort cause:", ...dumpObject(advanceError)] : []),
           "\n\ttransaction:", ...dumpObject(transaction),
           "\n\tthis:", ...dumpObject(thisArgument),
           "\n\tcallee vakon:", ...dumpKuery(vakon),