Merge pull request #55 from leapfrogtechnology/sequential-run

Support both sequential and parallel execution strategies for running synchronization

Author: mesaugat

examples/node-app-mssql/sync-db.yml

@@ -1,4 +1,5 @@
 basePath: src/sql
+execution: sequential
 sql:
   # Create objects in dbo schema
   - function/dbo/sum.sql

src/SyncDb.ts

@@ -2,6 +2,7 @@ import { Command, flags } from '@oclif/command';
 
 import { log } from './logger';
 import { handleFlags } from './cli';
+import { getElapsedTime } from './util/ts';
 import SyncResult from './domain/SyncResult';
 import SyncParams from './domain/SyncParams';
 import { printError, printLine } from './util/io';
@@ -35,11 +36,11 @@ class SyncDb extends Command {
       ...userParams,
       // Individual success handler
       onSuccess: (context: ExecutionContext) =>
-        printLine(`  [✓] ${context.connectionId} - Successful (${context.timeElapsed.toFixed(2)}s)`),
+        printLine(`  [✓] ${context.connectionId} - Successful (${context.timeElapsed}s)`),
 
       // Individual error handler
       onFailed: (context: ExecutionContext) =>
-        printLine(`  [✖] ${context.connectionId} - Failed (${context.timeElapsed.toFixed(2)}s)`)
+        printLine(`  [✖] ${context.connectionId} - Failed (${context.timeElapsed}s)`)
     };
   }
 
@@ -61,11 +62,20 @@ class SyncDb extends Command {
       const config = await loadConfig();
       const connections = await resolveConnections();
       const { synchronize } = await import('./api');
+      const timeStart = process.hrtime();
 
       await printLine('Synchronizing...\n');
 
       const results = await synchronize(config, connections, params);
-      const { totalCount, failedCount } = await this.processResults(results);
+      const { totalCount, failedCount, successfulCount } = await this.processResults(results);
+
+      if (successfulCount > 0) {
+        // Display output.
+        await printLine(
+          `Synchronization complete for ${successfulCount} / ${totalCount} connection(s). ` +
+            `(${getElapsedTime(timeStart)}s)`
+        );
+      }
 
       // If all completed successfully, exit gracefully.
       if (failedCount === 0) {
@@ -101,11 +111,6 @@ class SyncDb extends Command {
 
     await printLine();
 
-    if (successfulCount > 0) {
-      // Display output.
-      await printLine(`Synchronization successful for ${successfulCount} / ${totalCount} connection(s).`);
-    }
-
     // If there are errors, display all of them.
     if (!allComplete) {
       await printLine(`Synchronization failed for ${failedCount} connection(s):\n`);

src/api.ts

@@ -7,11 +7,14 @@ import SyncParams from './domain/SyncParams';
 import SyncConfig from './domain/SyncConfig';
 import SyncResult from './domain/SyncResult';
 import { DEFAULT_SYNC_PARAMS } from './constants';
-import { synchronizeDatabase } from './services/sync';
 import ConnectionConfig from './domain/ConnectionConfig';
 import ConnectionReference from './domain/ConnectionReference';
 import { isKnexInstance, getConfig, createInstance } from './util/db';
 
+// Services
+import { synchronizeDatabase } from './services/sync';
+import { executeProcesses } from './services/execution';
+
 /**
  * Synchronize all the configured database connections.
  *
@@ -31,7 +34,7 @@ export async function synchronize(
   const connections = mapToConnectionReferences(connectionList);
   const params = mergeDeepRight(DEFAULT_SYNC_PARAMS, options || {});
   const isCLI = process.env.SYNC_DB_CLI === 'true';
-  const promises = connections.map(({ connection, id: connectionId }) =>
+  const processes = connections.map(({ connection, id: connectionId }) => () =>
     synchronizeDatabase(connection, {
       isCLI,
       config,
@@ -40,7 +43,9 @@ export async function synchronize(
     })
   );
 
-  const results = await Promise.all(promises);
+  // Explicitly suppressing the `| Error` type since
+  // all errors are already caught inside synchronizeDatabase().
+  const results = (await executeProcesses(processes, config)) as SyncResult[];
 
   log('Synchronization completed.');
 

src/constants.ts

@@ -16,6 +16,7 @@ export const CONNECTIONS_FILENAME = 'connections.sync-db.json';
 export const INJECTED_CONFIG_TABLE = '__sync_db_injected_config';
 export const DEFAULT_CONFIG: SyncConfig = {
   basePath: path.resolve(process.cwd(), 'src/sql'),
+  execution: 'parallel',
   sql: [],
   hooks: {
     pre_sync: [],

src/domain/SyncConfig.ts

@@ -5,6 +5,7 @@ import Mapping from './Mapping';
  */
 interface SyncConfig {
   basePath: string;
+  execution: 'parallel' | 'sequential';
   sql: string[];
   hooks: {
     pre_sync: string[];

src/services/execution.ts

@@ -0,0 +1,25 @@
+import { log } from '../logger';
+import SyncConfig from '../domain/SyncConfig';
+import { Promiser, runSequentially } from '../util/promise';
+
+/**
+ * Execute a list of processes according to the configuration.
+ *
+ * @param {Promiser<T>[]} processes
+ * @param {SyncConfig} config
+ * @returns {Promise<T[]>}
+ */
+export function executeProcesses<T>(processes: Promiser<T>[], config: SyncConfig): Promise<(T | Error)[]> {
+  log(`Execution Strategy: ${config.execution}`);
+
+  switch (config.execution) {
+    case 'sequential':
+      return runSequentially(processes);
+
+    case 'parallel':
+      return Promise.all(processes.map(fn => fn()));
+
+    default:
+      throw new Error(`Execution strategy should be "sequential" or "parallel" found: "${config.execution}".`);
+  }
+}

src/services/sqlRunner.ts

@@ -109,14 +109,14 @@ export function getDropStatement(type: string, fqon: string): string {
  * @param {Knex} trx
  * @param {SqlCode[]} files
  * @param {string} connectionId
- * @returns {Promise<void>}
+ * @returns {Promise<any[]>}
  */
-export function runSequentially(trx: Knex, files: SqlCode[], connectionId: string): Promise<void> {
+export function runSequentially(trx: Knex, files: SqlCode[], connectionId: string): Promise<any[]> {
   const log = dbLogger(connectionId);
   const promises = files.map(file => {
     log(`Running ${file.name}`);
 
-    return trx.raw(file.sql);
+    return () => trx.raw(file.sql);
   });
 
   return promise.runSequentially(promises);

src/services/sync.ts

@@ -2,7 +2,7 @@ import * as Knex from 'knex';
 
 import { dbLogger } from '../logger';
 import * as sqlRunner from './sqlRunner';
-import { NS_PER_SEC } from '../constants';
+import { getElapsedTime } from '../util/ts';
 import SyncResult from '../domain/SyncResult';
 import SyncContext from '../domain/SyncContext';
 import * as configInjection from './configInjection';
@@ -109,10 +109,9 @@ export async function synchronizeDatabase(connection: Knex, context: SyncContext
     result.error = e;
   }
 
-  const timeDiff = process.hrtime(timeStart);
-  const timeElapsed = Number(timeDiff[0]) + Number(timeDiff[1] / NS_PER_SEC);
+  const timeElapsed = getElapsedTime(timeStart);
 
-  log(`Execution completed in ${timeDiff[0]} s, ${timeDiff[1]} ns`);
+  log(`Execution completed in ${timeElapsed} s`);
 
   // If it's a CLI environment, invoke the handler.
   if (context.isCLI) {

src/util/promise.ts

@@ -1,11 +1,45 @@
+import { promisify } from 'util';
+
 /**
- * Run each of the promise sequentially.
+ * Promiser - A function that returns a promise.
+ */
+export type Promiser<T> = () => PromiseLike<T>;
+
+/**
+ * Resolve a promise after a timeout.
+ */
+export const timeout = promisify(setTimeout);
+
+/**
+ * Run each of the promise sequentially and return their results in the same order.
  *
- * @param {PromiseLike<T>[]} promises
- * @returns {Promise<void>}
+ * @param {Promiser<T>[]} promisers
+ * @param {boolean} [failCascade=true]
+ * @returns {(Promise<(T | Error)[]>)}
  */
-export async function runSequentially<T>(promises: PromiseLike<T>[]): Promise<void> {
-  for (const promise of promises) {
-    await promise;
+export async function runSequentially<T>(
+  promisers: Promiser<T>[],
+  failCascade: boolean = true
+): Promise<(T | Error)[]> {
+  const result: (T | Error)[] = [];
+
+  for (const promiser of promisers) {
+    try {
+      const value = await promiser();
+
+      result.push(value);
+    } catch (err) {
+      // If failCascade = true,
+      // any error (promise rejection) will be cascaded thus halting the process.
+      if (failCascade) {
+        throw err;
+      }
+
+      // If failCascade = false,
+      // the failed promise will be resolved with the rejected error as a value.
+      result.push(err instanceof Error ? err : new Error(err));
+    }
   }
+
+  return result;
 }

src/util/ts.ts

@@ -0,0 +1,19 @@
+import { NS_PER_SEC } from '../constants';
+
+/**
+ * Calculate elapsed time from the given start time (process.hrtime()).
+ *
+ * @param {[number, number]} timeStart
+ * @param {(false | number)} fixed
+ * @returns {(number | string)}
+ */
+export function getElapsedTime(timeStart: [number, number], fixed: false | number = 2): number {
+  const timeDiff = process.hrtime(timeStart);
+  const timeElapsed = Number(timeDiff[0]) + Number(timeDiff[1] / NS_PER_SEC);
+
+  if (fixed === false) {
+    return timeElapsed;
+  }
+
+  return Number(timeElapsed.toFixed(fixed));
+}