Bound MSB apply wait horizon

src/config/config.js

@@ -1,6 +1,8 @@
 import b4a from "b4a";
 import path from "path";
 
+const DEFAULT_MAX_MSB_SIGNED_LENGTH_FUTURE_DELTA = 1_000_000;
+
 export class Config {
     #options;
 
@@ -47,6 +49,15 @@ export class Config {
         }
         this.maxMsbSignedLength = maxMsbSignedLength;
 
+        const maxMsbSignedLengthFutureDelta =
+            this.#select("maxMsbSignedLengthFutureDelta", options, defaults) ??
+            DEFAULT_MAX_MSB_SIGNED_LENGTH_FUTURE_DELTA;
+        if (!Number.isSafeInteger(maxMsbSignedLengthFutureDelta) ||
+            maxMsbSignedLengthFutureDelta < 0) {
+            throw new Error("Peer: maxMsbSignedLengthFutureDelta must be a non-negative safe integer.");
+        }
+        this.maxMsbSignedLengthFutureDelta = maxMsbSignedLengthFutureDelta;
+
         const maxMsbApplyOperationBytes = this.#select("maxMsbApplyOperationBytes", options, defaults);
         if (!Number.isSafeInteger(maxMsbApplyOperationBytes)) {
             throw new Error("Peer: maxMsbApplyOperationBytes must be a safe integer.");

src/config/env.js

@@ -15,6 +15,7 @@ const configData = {
         txPoolMaxSize: 1_000,
         maxTxDelay: 60,
         maxMsbSignedLength: 1_000_000_000,
+        maxMsbSignedLengthFutureDelta: 1_000_000,
         maxMsbApplyOperationBytes: 1024 * 1024,
         enableInteractiveMode: true,
         enableBackgroundTasks: true,
@@ -39,6 +40,7 @@ const configData = {
         txPoolMaxSize: 1_000,
         maxTxDelay: 60,
         maxMsbSignedLength: 1_000_000_000,
+        maxMsbSignedLengthFutureDelta: 1_000_000,
         maxMsbApplyOperationBytes: 1024 * 1024,
         enableInteractiveMode: true,
         enableBackgroundTasks: true,
@@ -63,6 +65,7 @@ const configData = {
         txPoolMaxSize: 1_000,
         maxTxDelay: 60,
         maxMsbSignedLength: 1_000_000_000,
+        maxMsbSignedLengthFutureDelta: 1_000_000,
         maxMsbApplyOperationBytes: 1024 * 1024,
         enableInteractiveMode: true,
         enableBackgroundTasks: false,

src/msbClient.js

@@ -97,11 +97,35 @@ export class MsbClient extends ReadyResource {
         return await this.#msb.network.tryConnect(pubKeyHex, role);
     }
 
-    async waitForSignedLengthAtLeast(targetSignedLength) {
+    async waitForSignedLengthAtLeast(targetSignedLength, { pollMs = 1_000 } = {}) {
         const core = this.#msb.state?.base?.view?.core ?? null;
         if (!core) throw new Error('MSB view core not available.');
+        if (!Number.isSafeInteger(targetSignedLength) || targetSignedLength < 0) {
+            throw new Error('Invalid MSB signed length target.');
+        }
+        if (!Number.isSafeInteger(pollMs) || pollMs < 1) {
+            throw new Error('Invalid MSB signed length wait poll interval.');
+        }
         while (core.signedLength < targetSignedLength) {
-            await new Promise((resolve) => core.once('append', resolve));
+            await new Promise((resolve) => {
+                const onAppend = () => {
+                    cleanup();
+                    resolve();
+                };
+                const cleanup = () => {
+                    clearTimeout(timer);
+                    if (typeof core.off === 'function') {
+                        core.off('append', onAppend);
+                    } else if (typeof core.removeListener === 'function') {
+                        core.removeListener('append', onAppend);
+                    }
+                };
+                const timer = setTimeout(() => {
+                    cleanup();
+                    resolve();
+                }, pollMs);
+                core.once('append', onAppend);
+            });
         }
     }
 

src/operations/tx/index.js

@@ -36,6 +36,11 @@ export class TxOperation {
         if(false === this.#validator.validate(op)) return;
         // Stall guard: don't allow a writer to pin apply waiting on an absurd MSB height
         if (op.value.msbsl > this.#config.maxMsbSignedLength) return;
+        const localMsbSignedLength = this.#msbClient.getSignedLength();
+        if (localMsbSignedLength > 0 &&
+            op.value.msbsl > localMsbSignedLength + this.#config.maxMsbSignedLengthFutureDelta) {
+            return;
+        }
         // Wait for local MSB view to reach the referenced signed length
         await this.#msbClient.waitForSignedLengthAtLeast(op.value.msbsl);
         // Fetch MSB apply-op at msbsl by tx key (op.key = tx hash)

tests/unit/applyGuards.test.js

@@ -164,6 +164,65 @@ test('apply: tx msbsl stall guard skips waiting', async (t) => {
     });
 });
 
+test('apply: tx msbsl relative future guard skips waiting', async (t) => {
+    await withTempDir(async ({ storesDirectory }) => {
+        const msbBootstrapBuf = b4a.alloc(32).fill(7);
+        const msb = makeMsbStub({
+            msbBootstrapBuf,
+            signedLength: 100,
+            async getEntry() {
+                return null;
+            },
+        });
+
+        msb.state.base.view.core.once = () => {
+            throw new Error('apply should not wait for a far-future msbsl');
+        };
+
+        const storeName = 'peer-relative-stall-guard';
+        const wallet = await prepareWallet(storesDirectory, storeName);
+        const config = createConfig(ENV.DEVELOPMENT, {
+            storesDirectory,
+            storeName,
+            maxMsbSignedLength: 1_000_000_000,
+            maxMsbSignedLengthFutureDelta: 10,
+        });
+        const peer = new Peer({
+            config,
+            msb,
+            protocol: TestProtocol,
+            contract: TestContract,
+            wallet,
+        });
+
+        try {
+            await peer.ready();
+
+            const txHashHex = makeHex32(1);
+            const op = {
+                type: 'tx',
+                key: txHashHex,
+                value: {
+                    dispatch: { type: 'ping', value: { msg: 'hi' } },
+                    msbsl: 111,
+                    ipk: makeHex32(2),
+                    wp: makeHex32(3),
+                },
+            };
+
+            const timeout = new Promise((_, reject) =>
+                setTimeout(() => reject(new Error('append timed out (possible apply stall)')), 2000)
+            );
+            await Promise.race([peer.base.append(op), timeout]);
+
+            const txl = await peer.bee.get('txl');
+            t.is(txl, null, 'tx should not be indexed when msbsl exceeds the local future window');
+        } finally {
+            await closePeer(peer);
+        }
+    });
+});
+
 test('apply: tx MSB payload size guard blocks otherwise-valid tx', async (t) => {
     await withTempDir(async ({ storesDirectory }) => {
         const msbBootstrapBuf = b4a.alloc(32).fill(7);