Retry SQL query on successful FTS5 recovery

ts/sql/Interface.ts

@@ -1141,7 +1141,7 @@ export type ServerWritableDirectInterface = WritableInterface & {
     allStickers: ReadonlyArray<string>
   ) => Array<string>;
 
-  runCorruptionChecks: () => void;
+  runCorruptionChecks: () => boolean;
 };
 
 export type ServerWritableInterface =

ts/sql/Server.ts

@@ -6142,41 +6142,54 @@ function getAllBadgeImageFileLocalPaths(db: ReadableDB): Set<string> {
   return new Set(localPaths);
 }
 
-function runCorruptionChecks(db: ReadableDB): void {
-  let writable: WritableDB;
+function runCorruptionChecks(db: WritableDB, isRetrying = false): boolean {
+  let ok = true;
+
   try {
-    writable = toUnsafeWritableDB(db, 'integrity check');
-  } catch (error) {
-    logger.error(
-      'runCorruptionChecks: not running the check, no writable instance',
-      Errors.toLogFormat(error)
-    );
-    return;
-  }
-  try {
-    const result = writable.pragma('integrity_check');
+    const result = db.pragma('integrity_check');
     if (result.length === 1 && result.at(0)?.integrity_check === 'ok') {
       logger.info('runCorruptionChecks: general integrity is ok');
     } else {
       logger.error('runCorruptionChecks: general integrity is not ok', result);
+      ok = false;
     }
   } catch (error) {
     logger.error(
       'runCorruptionChecks: general integrity check error',
       Errors.toLogFormat(error)
     );
+    ok = false;
   }
   try {
-    writable.exec(
-      "INSERT INTO messages_fts(messages_fts) VALUES('integrity-check')"
-    );
+    db.exec("INSERT INTO messages_fts(messages_fts) VALUES('integrity-check')");
     logger.info('runCorruptionChecks: FTS5 integrity ok');
   } catch (error) {
     logger.error(
       'runCorruptionChecks: FTS5 integrity check error.',
       Errors.toLogFormat(error)
     );
+    ok = false;
+
+    if (!isRetrying) {
+      try {
+        db.exec("INSERT INTO messages_fts(messages_fts) VALUES('rebuild');");
+
+        logger.info('runCorruptionChecks: FTS5 index rebuilt');
+      } catch (rebuildError) {
+        logger.error(
+          'runCorruptionChecks: FTS5 recovery failed',
+          Errors.toLogFormat(rebuildError)
+        );
+        return false;
       }
+
+      // Successfully recovered, try again.
+      logger.info('runCorruptionChecks: retrying');
+      return runCorruptionChecks(db, true);
+    }
+  }
+
+  return ok;
 }
 
 type StoryDistributionForDatabase = Readonly<

ts/sql/mainWorker.ts

@@ -20,28 +20,12 @@ if (!parentPort) {
 const port = parentPort;
 
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
-function respond(seq: number, error: Error | undefined, response?: any) {
-  let errorKind: SqliteErrorKind | undefined;
-  if (error !== undefined) {
-    errorKind = parseSqliteError(error);
-
-    if (errorKind === SqliteErrorKind.Corrupted && db != null) {
-      DataWriter.runCorruptionChecks(db);
-    }
-  }
-
+function respond(seq: number, response?: any) {
   const wrappedResponse: WrappedWorkerResponse = {
     type: 'response',
     seq,
-    error:
-      error == null
-        ? undefined
-        : {
-            name: error.name,
-            message: error.message,
-            stack: error.stack,
-          },
-    errorKind,
+    error: undefined,
+    errorKind: undefined,
     response,
   };
   port.postMessage(wrappedResponse);
@@ -84,7 +68,10 @@ let db: WritableDB | undefined;
 let isPrimary = false;
 let isRemoved = false;
 
-port.on('message', ({ seq, request }: WrappedWorkerRequest) => {
+const onMessage = (
+  { seq, request }: WrappedWorkerRequest,
+  isRetrying = false
+): void => {
   try {
     if (request.type === 'init') {
       isPrimary = request.isPrimary;
@@ -95,13 +82,13 @@ port.on('message', ({ seq, request }: WrappedWorkerRequest) => {
         logger,
       });
 
-      respond(seq, undefined, undefined);
+      respond(seq, undefined);
       return;
     }
 
     // 'close' is sent on shutdown, but we already removed the database.
     if (isRemoved && request.type === 'close') {
-      respond(seq, undefined, undefined);
+      respond(seq, undefined);
       process.exit(0);
       return;
     }
@@ -127,7 +114,7 @@ port.on('message', ({ seq, request }: WrappedWorkerRequest) => {
 
       isRemoved = true;
 
-      respond(seq, undefined, undefined);
+      respond(seq, undefined);
       return;
     }
 
@@ -143,7 +130,7 @@ port.on('message', ({ seq, request }: WrappedWorkerRequest) => {
       }
       db = undefined;
 
-      respond(seq, undefined, undefined);
+      respond(seq, undefined);
       process.exit(0);
       return;
     }
@@ -161,11 +148,39 @@ port.on('message', ({ seq, request }: WrappedWorkerRequest) => {
       const result = method(db, ...request.args);
       const end = performance.now();
 
-      respond(seq, undefined, { result, duration: end - start });
+      respond(seq, { result, duration: end - start });
     } else {
       throw new Error('Unexpected request type');
     }
   } catch (error) {
-    respond(seq, error, undefined);
+    const errorKind = parseSqliteError(error);
+
+    if (errorKind === SqliteErrorKind.Corrupted && db != null) {
+      const wasRecovered = DataWriter.runCorruptionChecks(db);
+      if (
+        wasRecovered &&
+        !isRetrying &&
+        // Don't retry 'init'/'close'/'removeDB' automatically and notify user
+        // about the database error (even on successful recovery).
+        (request.type === 'sqlCall:read' || request.type === 'sqlCall:write')
+      ) {
+        logger.error(`Retrying request: ${request.type}`);
+        return onMessage({ seq, request }, true);
       }
-});
+    }
+
+    const wrappedResponse: WrappedWorkerResponse = {
+      type: 'response',
+      seq,
+      error: {
+        name: error.name,
+        message: error.message,
+        stack: error.stack,
+      },
+      errorKind,
+      response: undefined,
+    };
+    port.postMessage(wrappedResponse);
+  }
+};
+port.on('message', (message: WrappedWorkerRequest) => onMessage(message));