fix(sqlite): flush data items using bundle db writer PE-6872

Prior to this commit, data items were flushed via the core DB writer.
Under sufficiently heavy bunde import load this lead to contention with
data item writes. This change adjusts the flushing logic to use the
bundle DB writer. This avoids concurrent bundle DB write attempts and
eliminates the contention.

package.json

@@ -8,7 +8,7 @@
     "url": "https://github.com/ar-io/ar-io-node"
   },
   "dependencies": {
-    "@ar.io/sdk": "^2.0.0",
+    "@ar.io/sdk": "^2.2.5",
     "@aws-lite/client": "^0.21.7",
     "@aws-lite/s3": "^0.1.21",
     "@clickhouse/client": "^1.3.0",

src/database/standalone-sqlite.ts

@@ -941,17 +941,25 @@ export class StandaloneSqliteDatabaseWorker {
       const endHeight = block.height - MAX_FORK_DEPTH;
 
       this.saveCoreStableDataFn(endHeight);
-      this.saveBundlesStableDataFn(endHeight);
 
       this.deleteCoreStaleNewDataFn(
         endHeight,
         maxStableBlockTimestamp - NEW_TX_CLEANUP_WAIT_SECS,
       );
-      this.deleteBundlesStaleNewDataFn(
-        endHeight,
-        maxStableBlockTimestamp - NEW_DATA_ITEM_CLEANUP_WAIT_SECS,
-      );
+
+      return { endHeight, maxStableBlockTimestamp };
     }
+
+    return {};
+  }
+
+  flushStableDataItems(endHeight: number, maxStableBlockTimestamp: number) {
+    this.saveBundlesStableDataFn(endHeight);
+
+    this.deleteBundlesStaleNewDataFn(
+      endHeight,
+      maxStableBlockTimestamp - NEW_DATA_ITEM_CLEANUP_WAIT_SECS,
+    );
   }
 
   getDataAttributes(id: string) {
@@ -2370,7 +2378,7 @@ const WORKER_POOL_NAMES: Array<WorkerPoolName> = [
   'bundles',
 ];
 
-type WorkerMethodName = keyof StandaloneSqliteDatabase;
+type WorkerMethodName = keyof StandaloneSqliteDatabaseWorker;
 
 type WorkerRoleName = 'read' | 'write';
 const WORKER_ROLE_NAMES: Array<WorkerRoleName> = ['read', 'write'];
@@ -2742,16 +2750,22 @@ export class StandaloneSqliteDatabase
     return this.queueWrite('bundles', 'saveBundle', [bundle]);
   }
 
-  saveBlockAndTxs(
+  async saveBlockAndTxs(
     block: PartialJsonBlock,
     txs: PartialJsonTransaction[],
     missingTxIds: string[],
   ): Promise<void> {
-    return this.queueWrite('core', 'saveBlockAndTxs', [
-      block,
-      txs,
-      missingTxIds,
-    ]);
+    const { endHeight, maxStableBlockTimestamp } = await this.queueWrite(
+      'core',
+      'saveBlockAndTxs',
+      [block, txs, missingTxIds],
+    );
+    if (maxStableBlockTimestamp !== undefined) {
+      await this.queueWrite('bundles', 'flushStableDataItems', [
+        endHeight,
+        maxStableBlockTimestamp,
+      ]);
+    }
   }
 
   async getDataAttributes(
@@ -3068,9 +3082,18 @@ if (!isMainThread) {
           parentPort?.postMessage(null);
           break;
         case 'saveBlockAndTxs':
-          const [block, txs, missingTxIds] = args;
-          worker.saveBlockAndTxs(block, txs, missingTxIds);
-          parentPort?.postMessage(null);
+          {
+            const [block, txs, missingTxIds] = args;
+            const ret = worker.saveBlockAndTxs(block, txs, missingTxIds);
+            parentPort?.postMessage(ret);
+          }
+          break;
+        case 'flushStableDataItems':
+          {
+            const [endHeight, maxStableBlockTimestamp] = args;
+            worker.flushStableDataItems(endHeight, maxStableBlockTimestamp);
+            parentPort?.postMessage(null);
+          }
           break;
         case 'getDataAttributes':
           const dataAttributes = worker.getDataAttributes(args[0]);