Properly propagate errors from worker threads in data services

evm/evm-data-service/src/main.ts

@@ -106,5 +106,8 @@ runProgram(async () => {
     })
 
     log.info(`listening on port ${service.port}`)
-    return waitForInterruption(service)
+    await Promise.race([
+        waitForInterruption(service),
+        service.run
+    ])
 }, err => log.fatal(err))

hyperliquid/hyperliquid-fills-data-service/src/main.ts

@@ -60,5 +60,8 @@ runProgram(async () => {
     })
 
     log.info(`listening on port ${service.port}`)
-    return waitForInterruption(service)
+    await Promise.race([
+        waitForInterruption(service),
+        service.run
+    ])
 }, err => log.fatal(err))

hyperliquid/hyperliquid-replica-cmds-data-service/src/main.ts

@@ -60,5 +60,8 @@ runProgram(async () => {
     })
 
     log.info(`listening on port ${service.port}`)
-    return waitForInterruption(service)
+    await Promise.race([
+        waitForInterruption(service),
+        service.run
+    ])
 }, err => log.fatal(err))

solana/solana-data-service/src/main.ts

@@ -79,8 +79,8 @@ runProgram(async () => {
 
     log.info(`listening on port ${service.port}`)
 
-    return new Promise((resolve, reject) => {
-        waitForInterruption(service).then(resolve, reject)
-        service.started.catch(reject)
-    })
+    await Promise.race([
+        waitForInterruption(service),
+        service.run
+    ])
 }, err => log.fatal(err))

util/util-internal-data-service/src/data-service.ts

@@ -187,9 +187,9 @@ export class DataService {
                 }
             } else {
                 if (!this.firstBlockIngested) {
-                    return this.firstBlockIngestedFuture.reject(
-                        err ?? new Error('data ingestion unexpectedly terminated')
-                    )
+                    let error = err ?? new Error('data ingestion unexpectedly terminated')
+                    this.firstBlockIngestedFuture.reject(error)
+                    throw error
                 }
 
                 let head = this.chain.getHeader()

util/util-internal-data-service/src/index.ts

@@ -23,17 +23,19 @@ export interface DataServiceOptions {
 }
 
 
-export async function runDataService(args: DataServiceOptions): Promise<ListeningServer & {started: Promise<void>}> {
+export async function runDataService(args: DataServiceOptions): Promise<ListeningServer & {started: Promise<void>, run: Promise<void>}> {
     let service = new DataService(args.source, args.blockCacheSize ?? 1000)
     let app = createHttpApp(service)
 
     await service.init()
     let server = await app.listen(args.port ?? 3000)
 
-    service.run().then()
+    let runPromise = service.run()
+    runPromise.catch(() => {})
 
     return {
         started: service.started(),
+        run: runPromise,
         port: server.port,
         close(): Promise<void> {
             service.stop()