add selfTest function to test all the process table by table

try catch all sensitive function and write logs if it happens

Author: dgrelaud

lib/index.js

@@ -4,6 +4,7 @@ const crypto = require('crypto');
 const hlc = require('./hlc.js');
 const Debugger = require('debug');
 const EventEmitter = require('node:events');
+const assert = require('node:assert');
 const { performance } = require('node:perf_hooks');
 
 const LAST_PATCH_AT_TIMESTAMP = 0;
@@ -177,7 +178,8 @@ const SQLiteOnSteroid = (db, myPeerId = null, options) => {
       deleteOldPatches      : (timestamp) => deleteOldPatches.run(timestamp),
       // directUpsert          : (patch) => directUpsert.run(_directUpsertSQLParamsFn(patch)),
       debounceFromTimestamp : Number.MAX_SAFE_INTEGER,
-      debounceTimer         : null
+      debounceTimer         : null,
+      knownColumns          : _columns
     };
     return {
       applyPatchesSQL       : _applyPatchesSQL,
@@ -495,51 +497,105 @@ const SQLiteOnSteroid = (db, myPeerId = null, options) => {
       tab   : tableName,
       delta : _tableStatement.cleanRow(rowPatch)
     };
-    _tableStatement.savePatch(_patch); // TODO insert by batch top optimize a bit
-    lastSequenceId++; // only increment lastSequenceId if patch is saved. TODO add tests. Make it cyclic, increment time when it switch
-    lastPatchAtTimestamp = _patch.at;
-    _tableStatement?.applyPatches?.(_patch.at); // It is not executed if patch is stored only in pending_patches table (happens when a ping stat message is generated internally)
+    try {
+      _tableStatement.savePatch(_patch); // TODO insert by batch top optimize a bit
+      lastSequenceId++; // only increment lastSequenceId if patch is saved. TODO add tests. Make it cyclic, increment time when it switch
+      lastPatchAtTimestamp = _patch.at;
+      _tableStatement?.applyPatches?.(_patch.at); // It is not executed if patch is stored only in pending_patches table (happens when a ping stat message is generated internally)
+    }
+    catch (e) {
+      console.error('Error processing patch %o ', _patch, e.message);
+      return callback(e);
+    }
     debugWrite('--> all peers %o', _patch);
     _broadcast(_patch);
     callback?.(null, _generateSessionToken(myPeerId, _patch.seq));
   }
 
+  /**
+   * Test if the database is correctly initialized and if the rowPatch is correctly stored/retrieved in/from the database.
+   *
+   * The extracted patch from the database should be the same as the inserted rowPatch.
+   *
+   * This function cannot be used in production because it creates a real patch.
+   * But it can be used (table by table) in the tests to verify that everything is working as expected (BigInt are converted to strings, blob ...)
+   *
+   * Same usage as `upsert`
+   *
+   * @param {*} tableName
+   * @param {*} rowPatch
+   * @param {*} callback(err) returns an error if the Replic-sqlite is not correctly initialized
+   */
+  function selfTest (tableName, rowPatch, callback) {
+    upsert(tableName, rowPatch, (err) => {
+      if (err) {
+        return callback(err);
+      }
+      const _missingPatch = globalStatements.getPatchFromColumn(myPeerId, 0, Number.MAX_SAFE_INTEGER);
+      const _patchRetrieved = JSON.parse(_missingPatch[0].patch);
+      const _knownColumns = tableStatements[tableName].knownColumns;
+      const _knownColumnsSorted = [..._knownColumns].sort();
+      try {
+        assert.deepStrictEqual(Object.keys(_patchRetrieved.delta).sort(), _knownColumnsSorted);
+        assert.deepStrictEqual(Object.keys(rowPatch).sort(), _knownColumnsSorted);
+      }
+      catch (e) {
+        e.message += '\nReplic-sqlite: Please check that all columns are tested (no missing columns in the patch).';
+        return callback(e);
+      }
+      try {
+        assert.deepStrictEqual(_patchRetrieved.delta, rowPatch);
+      }
+      catch (e) {
+        e.message += `\nReplic-sqlite: Please check that "prepareStatementHook" and "${tableName}_patches" table are correctly set to convert table columns for insert/extract operations (e.g., BigInt → string, BLOB → string), etc.\n`;
+        return callback(e);
+      }
+      callback(null);
+    });
+  }
+
   function _onPatchReceivedFromPeers (patch) {
-    if (parseInt(patch.peer, 10) === myPeerId) {
-      debug('Received patch from myself. Ignore it.');
-      return;
-    }
-    if (patch.ver !== dbVersion) {
-      // if version mismatch, save it in pending_patches table for later processing
-      tableStatements[PENDING_PATCHES_TABLE_NAME].savePatch(patch);
+    try {
+      if (parseInt(patch.peer, 10) === myPeerId) {
+        debug('Received patch from myself. Ignore it.');
+        return;
+      }
+      if (patch.ver !== dbVersion) {
+        // if version mismatch, save it in pending_patches table for later processing
+        tableStatements[PENDING_PATCHES_TABLE_NAME].savePatch(patch);
+        _detectMissingSequenceIds(patch);
+        return;
+      }
+      let _tableStatement = tableStatements[patch.tab];
+      if (!_tableStatement) {
+        // should never happen since we only send patches to peers with the same dbVersion (manage above)
+        console.warn(`Table ${patch.tab} not found when receiving patch from peers. Ignore patch.`);
+        return;
+      }
+      _tableStatement.savePatch(patch);
       _detectMissingSequenceIds(patch);
-      return;
+
+      // update real table, if it is not a ping message generated internally (stored only in pending_patches) with the short and reserved table name '_'
+      if (_tableStatement.applyPatches) {
+        // All the patches will be applied in the next event loop
+        if (_tableStatement.debounceFromTimestamp === Number.MAX_SAFE_INTEGER) {
+          _tableStatement.debounceFromTimestamp--; // make sure the function is called once in the next event loop (security)
+          setImmediate(() => {
+            _tableStatement.applyPatches(_tableStatement.debounceFromTimestamp);
+            _tableStatement.debounceFromTimestamp = Number.MAX_SAFE_INTEGER;
+          });
+        }
+        if (patch.at < _tableStatement.debounceFromTimestamp) {
+          // store the timestamp of the oldest patch to apply for the debouncer
+          // TODO tets this, verify that it applies the patches really from this timestamp
+          _tableStatement.debounceFromTimestamp = patch.at;
+        }
+      }
     }
-    let _tableStatement = tableStatements[patch.tab];
-    if (!_tableStatement) {
-      // should never happen since we only send patches to peers with the same dbVersion (manage above)
-      console.warn(`Table ${patch.tab} not found when receiving patch from peers. Ignore patch.`);
+    catch (e) {
+      console.error('Error processing patch %o ', patch, e.message);
       return;
     }
-    _tableStatement.savePatch(patch);
-    _detectMissingSequenceIds(patch);
-
-    // update real table, if it is not a ping message generated internally (stored only in pending_patches) with the short and reserved table name '_'
-    if (_tableStatement.applyPatches) {
-      // All the patches will be applied in the next event loop
-      if (_tableStatement.debounceFromTimestamp === Number.MAX_SAFE_INTEGER) {
-        _tableStatement.debounceFromTimestamp--; // make sure the function is called once in the next event loop (security)
-        setImmediate(() => {
-          _tableStatement.applyPatches(_tableStatement.debounceFromTimestamp);
-          _tableStatement.debounceFromTimestamp = Number.MAX_SAFE_INTEGER;
-        });
-      }
-      if (patch.at < _tableStatement.debounceFromTimestamp) {
-        // store the timestamp of the oldest patch to apply for the debouncer
-        // TODO tets this, verify that it applies the patches really from this timestamp
-        _tableStatement.debounceFromTimestamp = patch.at;
-      }
-    }
   }
 
   function addRemotePeer (remotePeerId, socket, connectionInfo) {
@@ -656,7 +712,14 @@ const SQLiteOnSteroid = (db, myPeerId = null, options) => {
   function _onRequestForMissingPatchFromPeers (msg) {
     // TODO retourner une erreur si la demande est trop vieilel (patch table nettoyée)
     // TODO limiter   msg.maxSeq  (max paquet de 100, 1000 ?)
-    const _missingPatch = globalStatements.getPatchFromColumn(msg.peer, msg.minSeq, msg.maxSeq);
+    let _missingPatch = [];
+    try {
+      _missingPatch = globalStatements.getPatchFromColumn(msg.peer, msg.minSeq, msg.maxSeq);
+    }
+    catch (e) {
+      console.error('Error processing request for missing patch %o ', msg, e.message);
+      return;
+    }
     if (peerSockets[msg.forPeer]) {
       if (Array.isArray(_missingPatch)) {
         for (const _patch of _missingPatch) {
@@ -917,6 +980,7 @@ const SQLiteOnSteroid = (db, myPeerId = null, options) => {
     metrics,
     backoff,
     readYourWrite,
+    selfTest,
     _parseSessionToken,
     _generateSessionToken,
     _onPatchReceivedFromPeers,

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "replic-sqlite",
-  "version": "0.2.10",
+  "version": "0.2.11",
   "description": "Simple, Fast, Multi-writer, Convergent, Replicated SQLite with 800 LOC only",
   "main": "lib/index.js",
   "directories": {