ar-io · djwhitt · Jan 9, 2025 · Jan 8, 2025 · Jan 9, 2025
diff --git a/scripts/import-data/utils.ts b/scripts/import-data/utils.ts
@@ -34,9 +34,9 @@ export const fetchWithRetry = async (
       if (response.ok) {
         return response;
       }
-      if (response.status === 429) {
+      if (response.status === 429 || response.status === 503) {
         console.warn(
-          `Import queue is full! Waiting 30 seconds before retrying...`,
+          `${await response.text()}. Waiting 30 seconds before retrying...`,
         );
         await new Promise((resolve) => setTimeout(resolve, 30000));
         continue;
@@ -88,8 +88,14 @@ export const getFilesInRange = async ({
       .filter((file) => path.extname(file) === '.json')
       .filter((file) => {
         const match = file.match(/^\d+/);
-        const number = match ? parseInt(match[0], 10) : null;
-        return number !== null && number >= min && number <= max;
+        if (!match) return false;
+        const number = parseInt(match[0], 10);
+        return number >= min && number <= max;
+      })
+      .sort((a, b) => {
+        const numA = parseInt(a.match(/^\d+/)?.[0] ?? '0', 10);
+        const numB = parseInt(b.match(/^\d+/)?.[0] ?? '0', 10);
+        return numA - numB;
       });
 
     return filesInRange;

diff --git a/src/routes/ar-io.ts b/src/routes/ar-io.ts
@@ -20,12 +20,12 @@ import createPrometheusMiddleware from 'express-prometheus-middleware';
 
 import * as config from '../config.js';
 import * as system from '../system.js';
-import * as events from '../events.js';
 import * as metrics from '../metrics.js';
 import { release } from '../version.js';
 import { db, signatureStore } from '../system.js';
 import log from '../log.js';
 import { ParquetExporter } from '../workers/parquet-exporter.js';
+import { NormalizedDataItem, PartialJsonTransaction } from '../types.js';
 
 export const arIoRouter = Router();
 export let parquetExporter: ParquetExporter | null = null;
@@ -215,10 +215,16 @@ arIoRouter.post(
         return;
       }
 
-      system.eventEmitter.emit(events.ANS104_BUNDLE_QUEUED, {
-        id,
-        root_tx_id: id,
-      });
+      const queuedBundle = await system.queueBundle(
+        { id, root_tx_id: id } as NormalizedDataItem | PartialJsonTransaction,
+        true,
+        true,
+      );
+
+      if (queuedBundle.error !== undefined) {
+        res.status(503).send(queuedBundle.error);
+        return;
+      }
 
       res.json({ message: 'Bundle queued' });
     } catch (error: any) {

diff --git a/src/system.ts b/src/system.ts
@@ -479,20 +479,23 @@ export const bundleDataImporter = new BundleDataImporter({
 metrics.registerQueueLengthGauge('bundleDataImporter', {
   length: () => bundleDataImporter.queueDepth(),
 });
-
-async function queueBundle(
+export type QueueBundleResponse = {
+  status: 'skipped' | 'queued' | 'error';
+  error?: string;
+};
+export async function queueBundle(
   item: NormalizedDataItem | PartialJsonTransaction,
   isPrioritized = false,
   bypassFilter = false,
-) {
+): Promise<QueueBundleResponse> {
   try {
     if ('root_tx_id' in item && item.root_tx_id === null) {
       log.debug('Skipping download of optimistically indexed data item', {
         id: item.id,
         rootTxId: item.root_tx_id,
         parentId: item.parent_id,
       });
-      return;
+      return { status: 'skipped' };
     }
 
     await db.saveBundle({
@@ -530,22 +533,18 @@ async function queueBundle(
         skippedAt: currentUnixTimestamp(),
       });
     }
+
+    return { status: 'queued' };
   } catch (error: any) {
     log.error('Error saving or queueing bundle', {
       message: error.message,
       stack: error.stack,
     });
+
+    return { status: 'error', error: 'Error queueing bundle' };
   }
 }
 
-// Queue bundles from the queue-bundle route
-eventEmitter.on(
-  events.ANS104_BUNDLE_QUEUED,
-  async (item: NormalizedDataItem | PartialJsonTransaction) => {
-    await queueBundle(item, true, true);
-  },
-);
-
 // Queue L1 bundles
 eventEmitter.on(
   events.ANS104_TX_INDEXED,