Handle errors in LMDB and Pino (#527)

Author: satyakigh

src/datastore/LMDB.ts

@@ -80,7 +80,7 @@ export class LMDBStoreFactory implements DataStoreFactory {
                 noSubdir: config.noSubdir,
                 overlappingSync: config.overlappingSync,
             },
-            `Initialized LMDB ${Version} with stores: ${toString(storeNames)} and ${formatNumber(this.totalBytes() / (1024 * 1024), 4)} MB`,
+            `Initialized LMDB ${Version} with stores: ${toString(storeNames)} and ${formatNumber(stats(this.env).totalSize / (1024 * 1024), 4)} MB`,
         );
     }
 
@@ -106,31 +106,48 @@ export class LMDBStoreFactory implements DataStoreFactory {
         if (this.closed) return;
         const msg = extractErrorMessage(error);
 
-        if (msg.includes('MDB_BAD_RSLOT') || msg.includes("doesn't match env pid")) {
-            this.recoverFromFork();
-        } else {
-            this.recoverFromError();
+        try {
+            if (msg.includes('MDB_BAD_RSLOT') || msg.includes("doesn't match env pid")) {
+                this.recoverFromFork();
+            } else {
+                this.recoverFromError();
+            }
+        } catch (recoveryError) {
+            this.log.error(recoveryError, 'LMDB recovery failed, disabling database');
+            this.telemetry.count('recovery.failed', 1);
         }
     }
 
     private ensureValidEnv(): void {
         if (process.pid !== this.openPid) {
             this.telemetry.count('process.forked', 1);
             this.log.warn({ oldPid: this.openPid, newPid: process.pid }, 'Process fork detected, reopening LMDB');
-            this.reopenEnv();
 
-            // Update all stores with new handles
-            for (const store of this.storeNames) {
-                this.stores.get(store)?.updateStore(createDB(this.env, store));
+            try {
+                this.reopenEnv();
+
+                // Update all stores with new handles
+                for (const store of this.storeNames) {
+                    this.stores.get(store)?.updateStore(createDB(this.env, store));
+                }
+            } catch (e) {
+                this.log.error(e, 'Failed to reopen LMDB after fork');
+                this.deleteAndRecreate();
             }
         }
     }
 
     private recoverFromFork(): void {
         this.telemetry.count('forked.recover', 1);
         this.log.warn({ oldPid: this.openPid, newPid: process.pid }, 'Process fork detected, reopening LMDB');
-        this.reopenEnv();
-        this.recreateStores();
+
+        try {
+            this.reopenEnv();
+            this.recreateStores();
+        } catch {
+            this.log.warn('Fork recovery failed, deleting and recreating');
+            this.deleteAndRecreate();
+        }
     }
 
     private recoverFromError(): void {
@@ -143,9 +160,18 @@ export class LMDBStoreFactory implements DataStoreFactory {
             this.log.info('Successfully recovered by reopening LMDB');
         } catch {
             this.log.warn('Reopen failed, deleting database');
+            this.deleteAndRecreate();
+        }
+    }
+
+    private deleteAndRecreate(): void {
+        try {
             this.deleteVersionDir();
             this.reopenEnv();
             this.recreateStores();
+        } catch (e) {
+            this.log.error(e, 'Failed to recreate LMDB after deletion');
+            this.telemetry.count('recovery.failed', 1);
         }
     }
 
@@ -209,8 +235,6 @@ export class LMDBStoreFactory implements DataStoreFactory {
         if (this.closed) return;
 
         try {
-            const totalBytes = this.totalBytes();
-
             const envStat = stats(this.env);
             this.telemetry.histogram('version', VersionNumber);
             this.telemetry.histogram('env.size.bytes', envStat.totalSize, { unit: 'By' });
@@ -219,10 +243,12 @@ export class LMDBStoreFactory implements DataStoreFactory {
             });
             this.telemetry.histogram('env.entries', envStat.entries);
 
+            let totalBytes = envStat.totalSize;
             for (const [name, store] of this.stores.entries()) {
                 const stat = store.stats();
                 this.telemetry.histogram(`store.${name}.size.bytes`, stat.totalSize, { unit: 'By' });
                 this.telemetry.histogram(`store.${name}.entries`, stat.entries);
+                totalBytes += stat.totalSize;
             }
 
             this.telemetry.histogram('total.usage', 100 * (totalBytes / TotalMaxDbSize), { unit: '%' });
@@ -231,17 +257,6 @@ export class LMDBStoreFactory implements DataStoreFactory {
             this.handleError(e);
         }
     }
-
-    private totalBytes() {
-        let totalBytes = 0;
-        totalBytes += stats(this.env).totalSize;
-
-        for (const store of this.stores.values()) {
-            totalBytes += store.stats().totalSize;
-        }
-
-        return totalBytes;
-    }
 }
 
 const VersionNumber = 5;

src/telemetry/LoggerFactory.ts

@@ -155,6 +155,14 @@ export class LoggerFactory implements Closeable {
         clearInterval(this.interval);
     }
 
+    /**
+     * Returns true if the error is a known pino stream race condition that
+     * occurs when the ThreadStream worker exits while async log calls are in flight.
+     */
+    static isPinoStreamError(error: unknown): boolean {
+        return error instanceof TypeError && error.message.includes('pino.msgPrefix');
+    }
+
     static getLogger(clazz: string | Function): Logger {
         if (LoggerFactory._instance === undefined) {
             throw new Error('Logger not configured');

src/telemetry/TelemetryService.ts

@@ -152,11 +152,13 @@ export class TelemetryService implements Closeable {
 
     private registerErrorHandlers(telemetry: ScopedTelemetry): void {
         process.on('unhandledRejection', (reason, _promise) => {
+            if (LoggerFactory.isPinoStreamError(reason)) return;
             telemetry.error('process.promise.unhandled', reason, undefined, { captureErrorAttributes: true });
             void this.metricsReader?.forceFlush();
         });
 
         process.on('uncaughtException', (error, origin) => {
+            if (LoggerFactory.isPinoStreamError(error)) return;
             telemetry.error('process.exception.uncaught', error, origin, { captureErrorAttributes: true });
             void this.metricsReader?.forceFlush();
         });