Bound MSB waits and update protobuf lock

package.json

@@ -91,6 +91,7 @@
     "process": "npm:bare-node-process",
     "protomux": "^3.10.1",
     "protomux-wakeup": "^2.4.0",
+    "rache": "1.0.0",
     "readline": "npm:bare-node-readline",
     "ready-resource": "1.1.2",
     "repl": "npm:bare-node-repl",

src/config/config.js

@@ -1,6 +1,9 @@
 import b4a from "b4a";
 import path from "path";
 
+const DEFAULT_HYPERBEE_CACHE_MAX_ENTRIES = 65_536;
+const DEFAULT_MAX_MSB_SIGNED_LENGTH_FUTURE_DELTA = 1_000_000;
+
 export class Config {
     #options;
 
@@ -41,12 +44,29 @@ export class Config {
         }
         this.maxTxDelay = maxTxDelay;
 
+        const hyperbeeCacheMaxEntries =
+            this.#select("hyperbeeCacheMaxEntries", options, defaults) ??
+            DEFAULT_HYPERBEE_CACHE_MAX_ENTRIES;
+        if (!Number.isSafeInteger(hyperbeeCacheMaxEntries) || hyperbeeCacheMaxEntries < 1) {
+            throw new Error("Peer: hyperbeeCacheMaxEntries must be a positive safe integer.");
+        }
+        this.hyperbeeCacheMaxEntries = hyperbeeCacheMaxEntries;
+
         const maxMsbSignedLength = this.#select("maxMsbSignedLength", options, defaults);
         if (!Number.isSafeInteger(maxMsbSignedLength)) {
             throw new Error("Peer: maxMsbSignedLength must be a safe integer.");
         }
         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

@@ -14,7 +14,9 @@ const configData = {
         storeName: "mainnet",
         txPoolMaxSize: 1_000,
         maxTxDelay: 60,
+        hyperbeeCacheMaxEntries: 65_536,
         maxMsbSignedLength: 1_000_000_000,
+        maxMsbSignedLengthFutureDelta: 1_000_000,
         maxMsbApplyOperationBytes: 1024 * 1024,
         enableInteractiveMode: true,
         enableBackgroundTasks: true,
@@ -38,7 +40,9 @@ const configData = {
         storeName: "testnet",
         txPoolMaxSize: 1_000,
         maxTxDelay: 60,
+        hyperbeeCacheMaxEntries: 65_536,
         maxMsbSignedLength: 1_000_000_000,
+        maxMsbSignedLengthFutureDelta: 1_000_000,
         maxMsbApplyOperationBytes: 1024 * 1024,
         enableInteractiveMode: true,
         enableBackgroundTasks: true,
@@ -62,7 +66,9 @@ const configData = {
         storeName: "peer",
         txPoolMaxSize: 1_000,
         maxTxDelay: 60,
+        hyperbeeCacheMaxEntries: 65_536,
         maxMsbSignedLength: 1_000_000_000,
+        maxMsbSignedLengthFutureDelta: 1_000_000,
         maxMsbApplyOperationBytes: 1024 * 1024,
         enableInteractiveMode: true,
         enableBackgroundTasks: false,

src/index.js

@@ -5,6 +5,7 @@ import ReadyResource from 'ready-resource';
 import b4a from 'b4a';
 import Hyperbee from 'hyperbee';
 import Corestore from 'corestore';
+import Rache from 'rache';
 import w from 'protomux-wakeup';
 const wakeup = new w();
 import Protomux from 'protomux'
@@ -13,7 +14,7 @@ import { MsbClient } from './msbClient.js';
 import { handlerFor } from './operations/index.js';
 import TransactionPool from './transaction/transactionPool.js';
 import { TransactionObserver } from './tasks/transactionObserver.js';
-import { Updater } from './tasks/updater.js';
+import { Updater, installNonNullAutobaseAck, installSignedAutobaseStore, installSignedCorestoreStorageFactory } from './tasks/updater.js';
 export { ensureTextCodecs } from './textCodec.js';
 export {default as Protocol} from "./artifacts/protocol.js";
 export {default as Contract} from "./artifacts/contract.js";
@@ -32,7 +33,9 @@ export class Peer extends ReadyResource {
         this.config = config;
 
         this.keyPair = null;
-        this.store = new Corestore(this.config.fullStoresDirectory);
+        this.store = new Corestore(this.config.fullStoresDirectory, {
+            globalCache: new Rache({ maxSize: this.config.hyperbeeCacheMaxEntries })
+        });
         this.msbClient = new MsbClient(msb);
         this.swarm = null;
         this.base = null;
@@ -75,10 +78,15 @@ export class Peer extends ReadyResource {
     }
 
     async _boot() {
+        installSignedCorestoreStorageFactory(this.store)
+        installSignedAutobaseStore(this.store, () => this.base?.local?.keyPair ?? null)
         this.base = new Autobase(this.store, this.config.bootstrap, {
-            ackInterval : 1000,
+            // MAYHEM PATCH: disable Autobase's implicit null ACK; updater appends
+            // a signed no-op instead so Pear/Bare never encodes a null head.
+            ackInterval : 0,
             valueEncoding: 'json',
             open: store => {
+                installSignedAutobaseStore(store)
                 this.bee = new Hyperbee(store.get('view'), {
                     extension: false,
                     keyEncoding: 'utf-8',
@@ -108,6 +116,7 @@ export class Peer extends ReadyResource {
                 await batch.close();
             }
         })
+        installNonNullAutobaseAck(this.base)
         this.base.on('warning', (e) => console.log(e))
     }
 

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)