add restart modules

commands/station.js

@@ -6,6 +6,7 @@ import * as bacalhau from '../lib/bacalhau.js'
 import fs from 'node:fs/promises'
 import { metrics } from '../lib/metrics.js'
 import { paths } from '../lib/paths.js'
+import pRetry from 'p-retry'
 
 const { FIL_WALLET_ADDRESS } = process.env
 
@@ -50,7 +51,7 @@ export const station = async ({ json, experimental }) => {
   })
 
   const modules = [
-    zinniaRuntime.start({
+    pRetry(() => zinniaRuntime.start({
       FIL_WALLET_ADDRESS,
       STATE_ROOT: join(paths.moduleState, 'zinnia'),
       CACHE_ROOT: join(paths.moduleCache, 'zinnia'),
@@ -64,7 +65,7 @@ export const station = async ({ json, experimental }) => {
         })
       },
       onMetrics: m => metrics.submit('zinnia', m)
-    })
+    }), { retries: 1000 })
   ]
 
   if (experimental) {

lib/zinnia.js

@@ -4,6 +4,7 @@ import Sentry from '@sentry/node'
 import { installBinaryModule, downloadSourceFiles, getBinaryModuleExecutable } from './modules.js'
 import { moduleBinaries } from './paths.js'
 import os from 'node:os'
+import { once } from 'node:events'
 
 const ZINNIA_DIST_TAG = 'v0.13.0'
 const ZINNIA_MODULES = [
@@ -57,6 +58,12 @@ export async function start ({
     }
   })
 
+  childProcess.on('exit', (code, signal) => {
+    const reason = signal ? `via signal ${signal}` : `with code: ${code}`
+    const msg = `Zinnia exited ${reason}`
+    onActivity({ type: 'info', message: msg })
+  })
+
   const readyPromise = new Promise((resolve, reject) => {
     childProcess.stdout.setEncoding('utf-8')
     childProcess.stdout.on('data', data => {
@@ -72,29 +79,29 @@ export async function start ({
     childProcess.catch(reject)
   })
 
-  childProcess.on('close', code => {
-    console.error(`Zinnia closed all stdio with code ${code ?? '<no code>'}`)
-    childProcess.stderr.removeAllListeners()
-    childProcess.stdout.removeAllListeners()
-    Sentry.captureException('Zinnia exited')
-  })
-
-  childProcess.on('exit', (code, signal) => {
-    const reason = signal ? `via signal ${signal}` : `with code: ${code}`
-    const msg = `Zinnia exited ${reason}`
-    onActivity({ type: 'info', message: msg })
-  })
-
-  try {
-    await Promise.race([
-      readyPromise,
-      timers.setTimeout(500)
-    ])
-  } catch (err) {
-    const errorMsg = err instanceof Error ? err.message : '' + err
-    const message = `Cannot start Zinnia: ${errorMsg}`
-    onActivity({ type: 'error', message })
-  }
+  await Promise.all([
+    (async () => {
+      try {
+        await Promise.race([
+          readyPromise,
+          timers.setTimeout(500)
+        ])
+      } catch (err) {
+        const errorMsg = err instanceof Error ? err.message : '' + err
+        const message = `Cannot start Zinnia: ${errorMsg}`
+        onActivity({ type: 'error', message })
+        throw err
+      }
+    })(),
+    (async () => {
+      const [code] = await once(childProcess, 'close')
+      console.error(`Zinnia closed all stdio with code ${code ?? '<no code>'}`)
+      childProcess.stderr.removeAllListeners()
+      childProcess.stdout.removeAllListeners()
+      Sentry.captureException('Zinnia exited')
+      throw new Error('Zinnia exited')
+    })()
+  ])
 }
 
 function handleEvents ({ onActivity, onMetrics, text }) {

package.json

@@ -46,6 +46,7 @@
     "@sentry/node": "^7.41.0",
     "execa": "^5.1.1",
     "gunzip-maybe": "^1.4.2",
+    "p-retry": "^5.1.2",
     "tar-fs": "^3.0.3",
     "undici": "^5.20.0",
     "unzip-stream": "^0.3.1",