add DISABLE_READ_THROUGH_DATA_CACHE

src/config.ts

@@ -263,7 +263,7 @@ export const BUNDLE_REPAIR_RETRY_INTERVAL_SECONDS = +env.varOrDefault(
 );
 
 export const BUNDLE_REPAIR_RETRY_BATCH_SIZE = +env.varOrDefault(
-  'BUNDLE_REPAIR_RETRY_INTERVAL_SECONDS',
+  'BUNDLE_REPAIR_RETRY_BATCH_SIZE',
   '1000',
 );
 
@@ -562,6 +562,9 @@ export const GET_DATA_CIRCUIT_BREAKER_TIMEOUT_MS = +env.varOrDefault(
   '500',
 );
 
+export const DISABLE_READ_THROUGH_DATA_CACHE =
+  env.varOrDefault('DISABLE_READ_THROUGH_DATA_CACHE', 'false') === 'true';
+
 //
 // AO
 //

src/database/sql/bundles/repair.sql

@@ -22,23 +22,21 @@ FROM (
   ORDER BY b.import_attempt_count, b.last_queued_at ASC
   LIMIT @limit
 )
-ORDER BY RANDOM()
 
 -- updateFullyIndexedAt
-UPDATE bundles
-SET
-  first_fully_indexed_at = IFNULL(first_fully_indexed_at, @fully_indexed_at),
-  last_fully_indexed_at = @fully_indexed_at
-WHERE matched_data_item_count IS NOT NULL
-  AND matched_data_item_count > 0
-  AND EXISTS (
-    SELECT 1
-    FROM bundle_data_items bdi
-    WHERE bdi.parent_id = bundles.id
-      AND bdi.filter_id = bundles.index_filter_id
-    GROUP BY bdi.parent_id
-    HAVING COUNT(*) = bundles.matched_data_item_count
-  ) AND last_fully_indexed_at IS NULL
+  UPDATE bundles
+  SET
+    first_fully_indexed_at = IFNULL(first_fully_indexed_at, @fully_indexed_at),
+    last_fully_indexed_at = @fully_indexed_at
+  WHERE matched_data_item_count IS NOT NULL
+    AND matched_data_item_count > 0
+    AND (
+      SELECT COUNT(*)
+      FROM bundle_data_items bdi
+      WHERE bdi.parent_id = bundles.id
+        AND bdi.filter_id = bundles.index_filter_id
+    ) = bundles.matched_data_item_count
+    AND last_fully_indexed_at IS NULL;
 
 -- updateForFilterChange
 UPDATE bundles

src/database/standalone-sqlite.ts

@@ -810,9 +810,18 @@ export class StandaloneSqliteDatabaseWorker {
   }
 
   getFailedBundleIds(limit: number) {
+    const reprocessCutoff = currentUnixTimestamp() - BUNDLE_REPROCESS_WAIT_SECS;
+
     const rows = this.stmts.bundles.selectFailedBundleIds.all({
       limit,
-      reprocess_cutoff: currentUnixTimestamp() - BUNDLE_REPROCESS_WAIT_SECS,
+      reprocess_cutoff: reprocessCutoff,
+    });
+
+    this.log.debug('getFailedBundleIds', {
+      reprocessCutoff,
+      BUNDLE_REPROCESS_WAIT_SECS,
+      limit,
+      rowsLength: rows.length,
     });
 
     return rows.map((row): string => toB64Url(row.id));

src/system.ts

@@ -412,27 +412,39 @@ const contiguousDataStore =
         baseDir: 'data/contiguous',
       });
 
-export const onDemandContiguousDataSource = new ReadThroughDataCache({
-  log,
-  dataSource: new SequentialDataSource({
-    log,
-    dataSources: onDemandDataSources,
-  }),
-  dataStore: contiguousDataStore,
-  contiguousDataIndex,
-  dataContentAttributeImporter,
-});
+export const onDemandContiguousDataSource =
+  config.DISABLE_READ_THROUGH_DATA_CACHE
+    ? new SequentialDataSource({
+        log,
+        dataSources: onDemandDataSources,
+      })
+    : new ReadThroughDataCache({
+        log,
+        dataSource: new SequentialDataSource({
+          log,
+          dataSources: onDemandDataSources,
+        }),
+        dataStore: contiguousDataStore,
+        contiguousDataIndex,
+        dataContentAttributeImporter,
+      });
 
-export const backgroundContiguousDataSource = new ReadThroughDataCache({
-  log,
-  dataSource: new SequentialDataSource({
-    log,
-    dataSources: backgroundDataSources,
-  }),
-  dataStore: contiguousDataStore,
-  contiguousDataIndex,
-  dataContentAttributeImporter,
-});
+export const backgroundContiguousDataSource =
+  config.DISABLE_READ_THROUGH_DATA_CACHE
+    ? new SequentialDataSource({
+        log,
+        dataSources: backgroundDataSources,
+      })
+    : new ReadThroughDataCache({
+        log,
+        dataSource: new SequentialDataSource({
+          log,
+          dataSources: backgroundDataSources,
+        }),
+        dataStore: contiguousDataStore,
+        contiguousDataIndex,
+        dataContentAttributeImporter,
+      });
 
 export const dataItemIndexer = new DataItemIndexer({
   log,

src/workers/bundle-repair-worker.ts

@@ -105,6 +105,13 @@ export class BundleRepairWorker {
       const bundleIds = await this.bundleIndex.getFailedBundleIds(
         config.BUNDLE_REPAIR_RETRY_BATCH_SIZE,
       );
+
+      this.log.debug('Bundles to retry', {
+        idsLength: bundleIds.length,
+        bundleIds,
+        batchSize: config.BUNDLE_REPAIR_RETRY_BATCH_SIZE,
+      });
+
       for (const bundleId of bundleIds) {
         this.log.info('Retrying failed bundle', { bundleId });
         await this.txFetcher.queueTxId({ txId: bundleId });