demux - npm Package Compare versions

Comparing version 3.1.4-f6522ed.0 to 3.1.4-fd42365.0

dist/AbstractActionHandler.d.ts

 import * as Logger from "bunyan";
-import { Block, BlockMeta, HandlerVersion, IndexState, VersionedAction } from "./interfaces";
+import { Block, HandlerVersion, IndexState, NextBlock, VersionedAction } from "./interfaces";
 /**
@@ -15,2 +15,3 @@ * Takes `block`s output from implementations of `AbstractActionReader` and processes their actions through the
     protected log: Logger;
+    private deferredEffects;
     private handlerVersionMap;
@@ -25,3 +26,3 @@ /**
      */
-    handleBlock(block: Block, blockMeta: BlockMeta, isReplay: boolean): Promise<number | null>;
+    handleBlock(nextBlock: NextBlock, isReplay: boolean): Promise<number | null>;
     /**
@@ -62,7 +63,7 @@ * Updates the `lastProcessedBlockNumber` and `lastProcessedBlockHash` meta state, coinciding with the block
      */
-    protected applyUpdaters(state: any, block: Block, isReplay: boolean, context: any): Promise<VersionedAction[]>;
+    protected applyUpdaters(state: any, block: Block, context: any, isReplay: boolean): Promise<VersionedAction[]>;
     /**
      * Process versioned actions against asynchronous side effects.
      */
-    protected runEffects(versionedActions: VersionedAction[], block: Block, context: any): void;
+    protected runEffects(versionedActions: VersionedAction[], context: any, nextBlock: NextBlock): void;
     /**
@@ -77,3 +78,7 @@ * Will run when a rollback block number is passed to handleActions. Implement this method to
      */
-    protected handleActions(state: any, block: Block, context: any, isReplay: boolean): Promise<void>;
+    protected handleActions(state: any, context: any, nextBlock: NextBlock, isReplay: boolean): Promise<void>;
+    private range;
+    private runOrDeferEffect;
+    private runDeferredEffects;
+    private getNextDeferredBlockNumber;
     private initHandlerVersions;
@@ -80,0 +85,0 @@ private refreshIndexState;

dist/AbstractActionHandler.js

@@ -35,2 +35,3 @@ "use strict";
         this.handlerVersionName = "v1";
+        this.deferredEffects = {};
         this.handlerVersionMap = {};
@@ -43,7 +44,8 @@ this.initHandlerVersions(handlerVersions);
      */
-    handleBlock(block, blockMeta, isReplay) {
+    handleBlock(nextBlock, isReplay) {
         return __awaiter(this, void 0, void 0, function* () {
+            const { block, blockMeta } = nextBlock;
             const { blockInfo } = block;
-            const { isRollback, isFirstBlock } = blockMeta;
-            if (isRollback || (isReplay && isFirstBlock)) {
+            const { isRollback, isEarliestBlock } = blockMeta;
+            if (isRollback || (isReplay && isEarliestBlock)) {
                 const rollbackBlockNumber = blockInfo.blockNumber - 1;
@@ -65,7 +67,7 @@ const rollbackCount = this.lastProcessedBlockNumber - rollbackBlockNumber;
             // If it's the first block but we've already processed blocks, seek to next block
-            if (isFirstBlock && this.lastProcessedBlockHash) {
+            if (isEarliestBlock && this.lastProcessedBlockHash) {
                 return nextBlockNeeded;
             }
             // Only check if this is the block we need if it's not the first block
-            if (!isFirstBlock) {
+            if (!isEarliestBlock) {
                 if (blockInfo.blockNumber !== nextBlockNeeded) {
@@ -80,3 +82,3 @@ return nextBlockNeeded;
             const handleWithArgs = (state, context = {}) => __awaiter(this, void 0, void 0, function* () {
-                yield this.handleActions(state, block, context, isReplay);
+                yield this.handleActions(state, context, nextBlock, isReplay);
             });
@@ -106,3 +108,3 @@ yield this.handleWithState(handleWithArgs);
      */
-    applyUpdaters(state, block, isReplay, context) {
+    applyUpdaters(state, block, context, isReplay) {
         return __awaiter(this, void 0, void 0, function* () {
@@ -141,8 +143,8 @@ const versionedActions = [];
      */
-    runEffects(versionedActions, block, context) {
+    runEffects(versionedActions, context, nextBlock) {
+        this.runDeferredEffects(nextBlock.lastIrreversibleBlockNumber);
         for (const { action, handlerVersionName } of versionedActions) {
             for (const effect of this.handlerVersionMap[handlerVersionName].effects) {
                 if (this.matchActionType(action.type, effect.actionType)) {
-                    const { payload } = action;
-                    effect.run(payload, block, context);
+                    this.runOrDeferEffect(effect, action.payload, nextBlock, context);
                 }
@@ -155,8 +157,9 @@ }
      */
-    handleActions(state, block, context, isReplay) {
+    handleActions(state, context, nextBlock, isReplay) {
         return __awaiter(this, void 0, void 0, function* () {
+            const { block } = nextBlock;
             const { blockInfo } = block;
-            const versionedActions = yield this.applyUpdaters(state, block, isReplay, context);
+            const versionedActions = yield this.applyUpdaters(state, block, context, isReplay);
             if (!isReplay) {
-                this.runEffects(versionedActions, block, context);
+                this.runEffects(versionedActions, context, nextBlock);
             }
@@ -168,2 +171,39 @@ yield this.updateIndexState(state, block, isReplay, this.handlerVersionName, context);
     }
+    range(start, end) {
+        return Array(end - start).fill(0).map((_, i) => i + start);
+    }
+    runOrDeferEffect(effect, payload, nextBlock, context) {
+        const { block, lastIrreversibleBlockNumber } = nextBlock;
+        const shouldRunImmediately = (!effect.deferUntilIrreversible || block.blockInfo.blockNumber <= lastIrreversibleBlockNumber);
+        if (shouldRunImmediately) {
+            effect.run(payload, block, context);
+        }
+        else if (!this.deferredEffects[block.blockInfo.blockNumber]) {
+            this.deferredEffects[block.blockInfo.blockNumber] = [() => effect.run(payload, block, context)];
+        }
+        else {
+            this.deferredEffects[block.blockInfo.blockNumber].push(() => effect.run(payload, block, context));
+        }
+    }
+    runDeferredEffects(lastIrreversibleBlockNumber) {
+        const nextDeferredBlockNumber = this.getNextDeferredBlockNumber();
+        if (!nextDeferredBlockNumber) {
+            return;
+        }
+        for (const blockNumber of this.range(nextDeferredBlockNumber, lastIrreversibleBlockNumber + 1)) {
+            if (this.deferredEffects[blockNumber]) {
+                for (const deferredEffect of this.deferredEffects[blockNumber]) {
+                    deferredEffect();
+                }
+                delete this.deferredEffects[blockNumber];
+            }
+        }
+    }
+    getNextDeferredBlockNumber() {
+        const blockNumbers = Object.keys(this.deferredEffects).map((num) => parseInt(num, 10));
+        if (blockNumbers.length === 0) {
+            return 0;
+        }
+        return Math.min(...blockNumbers);
+    }
     initHandlerVersions(handlerVersions) {
@@ -170,0 +210,0 @@ if (handlerVersions.length === 0) {

dist/AbstractActionReader.d.ts

 import * as Logger from "bunyan";
-import { ActionReaderConfig, Block, BlockMeta } from "./interfaces";
+import { ActionReaderOptions, Block, NextBlock } from "./interfaces";
 /**
@@ -10,11 +10,9 @@ * Reads blocks from a blockchain, outputting normalized `Block` objects.
     currentBlockNumber: number;
-    currentBlockMeta: BlockMeta | null;
     protected onlyIrreversible: boolean;
-    protected maxHistoryLength: number;
-    protected currentBlockData: Block | null;
+    protected currentBlockData: Block;
     protected lastIrreversibleBlockNumber: number;
     protected blockHistory: Block[];
     protected log: Logger;
-    private isFirstRun;
-    constructor(options?: ActionReaderConfig);
+    private initialized;
+    constructor(options?: ActionReaderOptions);
     /**
@@ -41,3 +39,3 @@ * Loads the number of the latest block.
      */
-    nextBlock(): Promise<[Block, BlockMeta]>;
+    getNextBlock(): Promise<NextBlock>;
     /**
@@ -57,3 +55,8 @@ * Changes the state of the `AbstractActionReader` instance to have just processed the block at the given block
     protected resolveFork(): Promise<void>;
+    private initBlockState;
     private getLatestNeededBlockNumber;
+    private acceptBlock;
+    private range;
+    private pruneHistory;
+    private reloadHistory;
     private addPreviousBlockToHistory;
@@ -60,0 +63,0 @@ private logForkDetected;

dist/AbstractActionReader.js

@@ -19,2 +19,11 @@ "use strict";
 const Logger = __importStar(require("bunyan"));
+const defaultBlock = {
+    blockInfo: {
+        blockNumber: 0,
+        blockHash: "",
+        previousBlockHash: "",
+        timestamp: new Date(0),
+    },
+    actions: [],
+};
 /**
@@ -26,12 +35,10 @@ * Reads blocks from a blockchain, outputting normalized `Block` objects.
         this.headBlockNumber = 0;
-        this.currentBlockMeta = null;
-        this.currentBlockData = null;
+        this.currentBlockData = defaultBlock;
         this.lastIrreversibleBlockNumber = 0;
         this.blockHistory = [];
-        this.isFirstRun = true;
-        const optionsWithDefaults = Object.assign({ startAtBlock: 1, onlyIrreversible: false, maxHistoryLength: 600 }, options);
+        this.initialized = false;
+        const optionsWithDefaults = Object.assign({ startAtBlock: 1, onlyIrreversible: false }, options);
         this.startAtBlock = optionsWithDefaults.startAtBlock;
         this.currentBlockNumber = optionsWithDefaults.startAtBlock - 1;
         this.onlyIrreversible = optionsWithDefaults.onlyIrreversible;
-        this.maxHistoryLength = optionsWithDefaults.maxHistoryLength;
         this.log = Logger.createLogger({ name: "demux" });
@@ -46,42 +53,32 @@ }
      */
-    nextBlock() {
+    getNextBlock() {
         return __awaiter(this, void 0, void 0, function* () {
-            let blockData = null;
-            let isRollback = false;
-            let isNewBlock = false;
-            // If we're on the head block, refresh current head block
-            if (this.currentBlockNumber === this.headBlockNumber || !this.headBlockNumber) {
+            const blockMeta = {
+                isRollback: false,
+                isNewBlock: false,
+                isEarliestBlock: false,
+            };
+            // TODO: Should this only be called when updating headBlockNumber?
+            this.lastIrreversibleBlockNumber = yield this.getLastIrreversibleBlockNumber();
+            if (!this.initialized) {
+                yield this.initBlockState();
+            }
+            if (this.currentBlockNumber === this.headBlockNumber) {
                 this.headBlockNumber = yield this.getLatestNeededBlockNumber();
             }
-            // If currentBlockNumber is negative, it means we wrap to the end of the chain (most recent blocks)
-            if (this.currentBlockNumber < 0 && this.isFirstRun) {
-                this.currentBlockNumber = this.headBlockNumber + this.currentBlockNumber;
-                this.startAtBlock = this.currentBlockNumber + 1;
-            }
-            else if (this.isFirstRun) {
-                this.isFirstRun = false;
-            }
-            // If we're now behind one or more new blocks, process them
             if (this.currentBlockNumber < this.headBlockNumber) {
                 const unvalidatedBlockData = yield this.getBlock(this.currentBlockNumber + 1);
-                const expectedHash = this.currentBlockData !== null ? this.currentBlockData.blockInfo.blockHash : "INVALID";
-                const actualHash = unvalidatedBlockData.blockInfo.previousBlockHash;
-                // Continue if the new block is on the same chain as our history, or if we've just started
-                if (expectedHash === actualHash || this.blockHistory.length === 0) {
-                    blockData = unvalidatedBlockData; // Block is now validated
-                    if (this.currentBlockData) {
-                        this.blockHistory.push(this.currentBlockData); // No longer current, belongs on history
-                    }
-                    this.blockHistory.splice(0, this.blockHistory.length - this.maxHistoryLength); // Trim history
-                    this.currentBlockData = blockData; // Replaced with the real current block
-                    isNewBlock = true;
-                    this.currentBlockNumber = this.currentBlockData.blockInfo.blockNumber;
+                const expectedHash = this.currentBlockData.blockInfo.blockHash;
+                const actualHash = this.currentBlockNumber ?
+                    unvalidatedBlockData.blockInfo.previousBlockHash :
+                    defaultBlock.blockInfo.blockHash;
+                if (expectedHash === actualHash) {
+                    this.acceptBlock(unvalidatedBlockData);
+                    blockMeta.isNewBlock = true;
                 }
                 else {
-                    // Since the new block did not match our history, we can assume our history is wrong
-                    // and need to roll back
                     this.logForkDetected(unvalidatedBlockData, expectedHash, actualHash);
                     yield this.resolveFork();
-                    isNewBlock = true;
-                    isRollback = true; // Signal action handler that we must roll back
+                    blockMeta.isNewBlock = true;
+                    blockMeta.isRollback = true;
                     // Reset for safety, as new fork could have less blocks than the previous fork
@@ -91,13 +88,8 @@ this.headBlockNumber = yield this.getLatestNeededBlockNumber();
             }
-            // Let handler know if this is the earliest block we'll send
-            const isFirstBlock = this.currentBlockNumber === this.startAtBlock;
-            if (this.currentBlockData === null) {
-                throw Error("currentBlockData must not be null.");
-            }
-            this.currentBlockMeta = {
-                isRollback,
-                isFirstBlock,
-                isNewBlock,
+            blockMeta.isEarliestBlock = this.currentBlockNumber === this.startAtBlock;
+            return {
+                block: this.currentBlockData,
+                blockMeta,
+                lastIrreversibleBlockNumber: this.lastIrreversibleBlockNumber,
             };
-            return [this.currentBlockData, this.currentBlockMeta];
         });
@@ -114,37 +106,11 @@ }
         return __awaiter(this, void 0, void 0, function* () {
-            // Clear current block data
-            this.currentBlockData = null;
-            this.headBlockNumber = 0;
+            this.headBlockNumber = yield this.getLatestNeededBlockNumber();
             if (blockNumber < this.startAtBlock) {
-                throw Error("Cannot seek to block before configured startAtBlock.");
+                throw new Error("Cannot seek to block before configured `startAtBlock` number.");
             }
-            // If we're going back to the first block, we don't want to get the preceding block
-            if (blockNumber === 1) {
-                this.blockHistory = [];
-                this.currentBlockNumber = 0;
-                return;
+            if (blockNumber > this.headBlockNumber) {
+                throw new Error(`Cannot seek to block number ${blockNumber} as it does not exist yet.`);
             }
-            // Check if block exists in history
-            let toDelete = -1;
-            for (let i = this.blockHistory.length - 1; i >= 0; i--) {
-                if (this.blockHistory[i].blockInfo.blockNumber === blockNumber) {
-                    break;
-                }
-                else {
-                    toDelete += 1;
-                }
-            }
-            if (toDelete >= 0) {
-                this.blockHistory.splice(toDelete);
-                this.currentBlockData = this.blockHistory.pop() || null;
-            }
-            // Load current block
             this.currentBlockNumber = blockNumber - 1;
-            if (!this.currentBlockData) {
-                this.currentBlockData = yield this.getBlock(this.currentBlockNumber);
-            }
-            // Fetch block if there is no history
-            if (this.blockHistory.length === 0) {
-                yield this.addPreviousBlockToHistory(false);
-            }
+            yield this.reloadHistory();
         });
@@ -159,5 +125,2 @@ }
         return __awaiter(this, void 0, void 0, function* () {
-            if (this.currentBlockData === null) {
-                throw Error("`currentBlockData` must not be null when initiating fork resolution.");
-            }
             if (this.blockHistory.length === 0) {
@@ -174,20 +137,31 @@ yield this.addPreviousBlockToHistory();
                 this.currentBlockData = yield this.getBlock(this.currentBlockData.blockInfo.blockNumber);
-                if (this.currentBlockData !== null) {
-                    const { blockInfo: currentBlockInfo } = this.currentBlockData;
-                    const { blockInfo: previousBlockInfo } = previousBlockData;
-                    if (currentBlockInfo.previousBlockHash === previousBlockInfo.blockHash) {
-                        this.logForkResolved(currentBlockInfo, previousBlockInfo);
-                        break;
-                    }
-                    this.logForkMismatch(currentBlockInfo, previousBlockInfo);
+                const { blockInfo: currentBlockInfo } = this.currentBlockData;
+                const { blockInfo: previousBlockInfo } = previousBlockData;
+                if (currentBlockInfo.previousBlockHash === previousBlockInfo.blockHash) {
+                    this.logForkResolved(currentBlockInfo, previousBlockInfo);
+                    break;
                 }
+                this.logForkMismatch(currentBlockInfo, previousBlockInfo);
                 this.currentBlockData = previousBlockData;
                 this.blockHistory.pop();
             }
+            if (this.blockHistory.length === 0) {
+                yield this.addPreviousBlockToHistory();
+            }
             this.currentBlockNumber = this.blockHistory[this.blockHistory.length - 1].blockInfo.blockNumber + 1;
         });
     }
+    initBlockState() {
+        return __awaiter(this, void 0, void 0, function* () {
+            this.headBlockNumber = yield this.getLatestNeededBlockNumber();
+            if (this.currentBlockNumber < 0) {
+                this.currentBlockNumber = this.headBlockNumber + this.currentBlockNumber;
+                this.startAtBlock = this.currentBlockNumber + 1;
+            }
+            yield this.reloadHistory();
+            this.initialized = true;
+        });
+    }
     getLatestNeededBlockNumber() {
         return __awaiter(this, void 0, void 0, function* () {
-            this.lastIrreversibleBlockNumber = yield this.getLastIrreversibleBlockNumber();
             if (this.onlyIrreversible) {
@@ -201,8 +175,75 @@ return this.lastIrreversibleBlockNumber;
     }
+    acceptBlock(blockData) {
+        this.blockHistory.push(this.currentBlockData);
+        this.pruneHistory();
+        this.currentBlockData = blockData;
+        this.currentBlockNumber = this.currentBlockData.blockInfo.blockNumber;
+    }
+    range(start, end) {
+        if (start > end) {
+            return [];
+        }
+        return Array(end - start).fill(0).map((_, i) => i + start);
+    }
+    pruneHistory() {
+        let toDelete = 0;
+        for (const block of this.blockHistory) {
+            if (block.blockInfo.blockNumber < this.lastIrreversibleBlockNumber) {
+                toDelete += 1;
+            }
+            else {
+                break;
+            }
+        }
+        if (toDelete === this.blockHistory.length) {
+            this.blockHistory = [this.blockHistory[this.blockHistory.length - 1]];
+            return;
+        }
+        this.blockHistory.splice(0, toDelete);
+    }
+    reloadHistory(maxTries = 10) {
+        return __awaiter(this, void 0, void 0, function* () {
+            if (this.currentBlockNumber === 0) {
+                this.blockHistory = [];
+                this.currentBlockData = defaultBlock;
+                return;
+            }
+            if (this.currentBlockNumber === 1) {
+                this.blockHistory = [defaultBlock];
+                this.currentBlockData = yield this.getBlock(1);
+                return;
+            }
+            let historyRange = this.range(this.lastIrreversibleBlockNumber, this.currentBlockNumber + 1);
+            if (historyRange.length <= 1) {
+                historyRange = [this.currentBlockNumber - 1, this.currentBlockNumber];
+            }
+            let microForked = true;
+            let tryCount = 0;
+            while (microForked) {
+                microForked = false;
+                this.blockHistory = [];
+                for (const blockNumber of historyRange) {
+                    const historyBlock = yield this.getBlock(blockNumber);
+                    if (this.blockHistory.length === 0) {
+                        this.blockHistory.push(historyBlock);
+                        continue;
+                    }
+                    const latestHistoryBlockHash = this.blockHistory[this.blockHistory.length - 1].blockInfo.blockHash;
+                    if (latestHistoryBlockHash !== historyBlock.blockInfo.previousBlockHash) {
+                        microForked = true;
+                        break;
+                    }
+                    this.blockHistory.push(historyBlock);
+                }
+                tryCount += 1;
+                if (tryCount === maxTries) {
+                    throw new Error("Could not reload history.");
+                }
+            }
+            this.currentBlockData = this.blockHistory.pop();
+        });
+    }
     addPreviousBlockToHistory(checkIrreversiblility = true) {
         return __awaiter(this, void 0, void 0, function* () {
-            if (!this.currentBlockData) {
-                throw Error("`currentBlockData` must not be null when initiating fork resolution.");
-            }
-            if (this.currentBlockData.blockInfo.blockNumber <= this.lastIrreversibleBlockNumber && checkIrreversiblility) {
+            if (this.currentBlockData.blockInfo.blockNumber < this.lastIrreversibleBlockNumber && checkIrreversiblility) {
                 throw new Error("Last irreversible block has been passed without resolving fork");
@@ -209,0 +250,0 @@ }

dist/BaseActionWatcher.js

@@ -57,13 +57,10 @@ "use strict";
             while (!headBlockNumber || this.actionReader.currentBlockNumber < headBlockNumber) {
-                const [blockData, blockMeta] = yield this.actionReader.nextBlock();
-                if (!blockMeta.isNewBlock) {
+                const nextBlock = yield this.actionReader.getNextBlock();
+                if (!nextBlock.blockMeta.isNewBlock) {
                     break;
                 }
-                let seekBlockNum = null;
-                if (blockData) {
-                    seekBlockNum = yield this.actionHandler.handleBlock(blockData, blockMeta, isReplay);
+                const nextBlockNumberNeeded = yield this.actionHandler.handleBlock(nextBlock, isReplay);
+                if (nextBlockNumberNeeded) {
+                    yield this.actionReader.seekToBlock(nextBlockNumberNeeded - 1);
                 }
-                if (seekBlockNum) {
-                    yield this.actionReader.seekToBlock(seekBlockNum - 1);
-                }
                 headBlockNumber = this.actionReader.headBlockNumber;
@@ -70,0 +67,0 @@ }

dist/interfaces.d.ts

@@ -1,2 +0,2 @@
-export interface ActionReaderConfig {
+export interface ActionReaderOptions {
     /**
@@ -27,3 +27,3 @@ * For positive values, this sets the first block that this will start at. For negative
     isRollback: boolean;
-    isFirstBlock: boolean;
+    isEarliestBlock: boolean;
     isNewBlock: boolean;
@@ -43,2 +43,7 @@ }
 }
+export interface NextBlock {
+    block: Block;
+    blockMeta: BlockMeta;
+    lastIrreversibleBlockNumber: number;
+}
 export interface Action {
@@ -59,2 +64,3 @@ type: string;
     run: StatelessActionCallback;
+    deferUntilIrreversible?: boolean;
     onRollback?: StatelessActionCallback;
@@ -71,1 +77,5 @@ }
 }
+export declare type CurriedEffectRun = (() => void | Promise<void>);
+export interface DeferredEffects {
+    [key: number]: CurriedEffectRun[];
+}

package.json

 {
   "name": "demux",
-  "version": "3.1.4-f6522ed.0",
+  "version": "3.1.4-fd42365.0",
   "author": {
@@ -5,0 +5,0 @@ "name": "Julien Heller",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Major refactor

Supply chain risk

Package has recently undergone a major refactor. It may be unstable or indicate significant internal changes. Use caution when updating to versions that include significant changes.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

302107

increased by0.97%

Lines of code

856

increased by12.63%

Number of medium supply chain risk alerts

0

decreased by-100%

No dependency changes