enh: healthyguard

packages/daemon/src/server.ts

@@ -56,23 +56,35 @@ global.EventSource = require('eventsource')
   {
     info(`Migrating mothership`)
     await (
-      await pbService.spawn({
+      await pbService.spawn(
+        {
           command: 'migrate',
           isMothership: true,
           version: DAEMON_PB_SEMVER,
           slug: PUBLIC_APP_DB,
-      })
+          onUnexpectedStop: () => {
+            error(`migrate had an unexpected stop. Check it out`)
+          },
+        },
+        { logger }
+      )
     ).exited
     info(`Migrating done`)
   }
   info(`Serving`)
-  const { url } = await pbService.spawn({
+  const { url } = await pbService.spawn(
+    {
       command: 'serve',
       isMothership: true,
       version: DAEMON_PB_SEMVER,
       slug: PUBLIC_APP_DB,
       port: DAEMON_PB_PORT_BASE,
-  })
+      onUnexpectedStop: () => {
+        error(`migrate had an unexpected stop. Check it out`)
+      },
+    },
+    { logger }
+  )
 
   /**
    * Launch services

packages/daemon/src/services/InstanceService/InstanceService.ts

@@ -43,7 +43,7 @@ type InstanceApi = {
   status: () => InstanceApiStatus
   internalUrl: () => string
   startRequest: () => () => void
-  shutdown: () => Promise<void>
+  shutdown: (reason?: Error) => Promise<void>
 }
 
 export type InstanceServiceConfig = SingletonBaseConfig & {
@@ -116,16 +116,30 @@ export const instanceService = mkSingleton(
         )
       }
 
+      /*
+      Initialize shutdown manager
+      */
+      const shutdownManager = createCleanupManager()
+      shutdownManager.add(async () => {
+        dbg(`Deleting from cache`)
+        delete instanceApis[id]
+        dbg(`There are ${values(instanceApis).length} still in cache`)
+      }, CLEANUP_PRIORITY_LAST) // Make this the very last thing that happens
+      shutdownManager.add(async () => {
+        dbg(`Shutting down`)
+        status = InstanceApiStatus.ShuttingDown
+      })
+
       info(`Starting`)
       let status = InstanceApiStatus.Starting
       let internalUrl = ''
       let startRequest: InstanceApi['startRequest'] = () => {
         throw new Error(`Not ready yet`)
       }
-      let shutdown: InstanceApi['shutdown'] = () => {
-        throw new Error(`Not ready yet`)
-      }
 
+      /*
+      Initialize API
+      */
       const api: InstanceApi = {
         status: () => {
           return status
@@ -146,43 +160,38 @@ export const instanceService = mkSingleton(
           }
           return startRequest()
         },
-        shutdown: async () => {
+        shutdown: async (reason) => {
-          if (status !== InstanceApiStatus.Healthy) {
+          if (reason) {
-            throw new Error(
+            error(`Panic shutdown for ${reason}`)
-              `Attempt to shut down an instance request when instance is not in a healthy state.`
+          } else {
-            )
+            dbg(`Graceful shutdown`)
           }
-          return shutdown()
+          if (api.status() === InstanceApiStatus.ShuttingDown) {
+            throw new Error(`Already shutting down`)
+          }
+          return shutdownManager.shutdown()
         },
       }
+      const _safeShutdown = async (reason?: Error) =>
+        api.status() === InstanceApiStatus.ShuttingDown || api.shutdown(reason)
       instanceApis[id] = api
 
-      /*
+      const healthyGuard = () => {
-        Initialize shutdown manager
+        if (api.status() !== InstanceApiStatus.ShuttingDown) return
-        */
+        throw new Error(`Instance is shutting down. Aborting.`)
-      const shutdownManager = createCleanupManager()
+      }
-      shutdownManager.add(async () => {
-        dbg(`Deleting from cache`)
-        delete instanceApis[id]
-        dbg(`There are ${values(instanceApis).length} still in cache`)
-      }, CLEANUP_PRIORITY_LAST) // Make this the very last thing that happens
-      shutdownManager.add(async () => {
-        dbg(`Shutting down`)
-        status = InstanceApiStatus.ShuttingDown
-      })
-      shutdown = () => shutdownManager.shutdown()
 
       /*
         Create serialized client communication functions to prevent race conditions
         */
       const clientLimiter = new Bottleneck({ maxConcurrent: 1 })
-      const _updateInstanceStatus = clientLimiter.wrap(
+      const updateInstanceStatus = clientLimiter.wrap(
         client.updateInstanceStatus
       )
-      const _updateInstance = clientLimiter.wrap(client.updateInstance)
+      const updateInstance = clientLimiter.wrap(client.updateInstance)
-      const _createInvocation = clientLimiter.wrap(client.createInvocation)
+      const createInvocation = clientLimiter.wrap(client.createInvocation)
-      const _pingInvocation = clientLimiter.wrap(client.pingInvocation)
+      const pingInvocation = clientLimiter.wrap(client.pingInvocation)
-      const _finalizeInvocation = clientLimiter.wrap(client.finalizeInvocation)
+      const finalizeInvocation = clientLimiter.wrap(client.finalizeInvocation)
 
       /*
       Handle async setup
@@ -194,29 +203,34 @@ export const instanceService = mkSingleton(
         Obtain empty port
         */
         dbg(`Obtaining port`)
-        await _updateInstanceStatus(instance.id, InstanceStatus.Port)
+        healthyGuard()
+        await updateInstanceStatus(instance.id, InstanceStatus.Port)
+        healthyGuard()
         const [newPort, releasePort] = await getNextPort()
         shutdownManager.add(() => {
           dbg(`Releasing port`)
           releasePort()
         }, CLEANUP_PRIORITY_LAST)
-
         systemInstanceLogger.breadcrumb(`port:${newPort}`)
         dbg(`Found port`)
 
         /*
         Create the user instance logger
         */
+        healthyGuard()
         const userInstanceLogger = await instanceLoggerService().get(
           instance.id,
-          { parentLogger: systemInstanceLogger }
+          {
+            parentLogger: systemInstanceLogger,
+          }
         )
-        const _writeUserLog = serialAsyncExecutionGuard(
+
+        const writeUserLog = serialAsyncExecutionGuard(
           userInstanceLogger.write,
           () => `${instance.id}:userLog`
         )
         shutdownManager.add(() =>
-          _writeUserLog(`Shutting down instance`).catch(error)
+          writeUserLog(`Shutting down instance`).catch(error)
         )
 
         /*
@@ -224,12 +238,14 @@ export const instanceService = mkSingleton(
         */
         dbg(`Starting instance`)
         dbg(`Set instance status: starting`)
-        await _updateInstanceStatus(instance.id, InstanceStatus.Starting)
+        healthyGuard()
+        await updateInstanceStatus(instance.id, InstanceStatus.Starting)
         shutdownManager.add(async () => {
           dbg(`Set instance status: idle`)
-          await _updateInstanceStatus(id, InstanceStatus.Idle).catch(error)
+          await updateInstanceStatus(id, InstanceStatus.Idle).catch(error)
         })
-        await _writeUserLog(`Starting instance`)
+        healthyGuard()
+        await writeUserLog(`Starting instance`)
 
         /*
         Spawn the child process
@@ -241,33 +257,37 @@ export const instanceService = mkSingleton(
               slug: instance.id,
               port: newPort,
               version,
-              onUnexpectedStop: async (code, stdout, stderr) => {
+              onUnexpectedStop: (code, stdout, stderr) => {
                 warn(
                   `PocketBase processes exited unexpectedly with ${code}. Putting in maintenance mode.`
                 )
                 warn(stdout)
                 warn(stderr)
                 shutdownManager.add(async () => {
-                  await _updateInstance(instance.id, {
+                  await updateInstance(instance.id, {
                     maintenance: true,
                   })
-                  await _writeUserLog(
+                  await writeUserLog(
                     `Putting instance in maintenance mode because it shut down with return code ${code}. `,
                     StreamNames.Error
                   )
                   await Promise.all(
                     stdout.map((data) =>
-                      _writeUserLog(data, StreamNames.Error).catch(error)
+                      writeUserLog(data, StreamNames.Error).catch(error)
                     )
                   )
                   await Promise.all(
                     stderr.map((data) =>
-                      _writeUserLog(data, StreamNames.Error).catch(error)
+                      writeUserLog(data, StreamNames.Error).catch(error)
                     )
                   )
                 })
                 setImmediate(() => {
-                  api.shutdown().catch(error)
+                  _safeShutdown(
+                    new Error(
+                      `PocketBase processes exited unexpectedly with ${code}. Putting in maintenance mode.`
+                    )
+                  ).catch(error)
                 })
               },
             })
@@ -292,14 +312,16 @@ export const instanceService = mkSingleton(
         /*
         Create the invocation record
         */
-        const invocation = await _createInvocation(instance, pid)
+        healthyGuard()
+        const invocation = await createInvocation(instance, pid)
         shutdownManager.add(async () => {
-          await _finalizeInvocation(invocation).catch(error)
+          await finalizeInvocation(invocation).catch(error)
         })
 
         /**
          * Deno worker
          */
+        healthyGuard()
         const denoApi = await (async () => {
           const workerPath = join(
             DAEMON_PB_DATA_DIR,
@@ -310,7 +332,9 @@ export const instanceService = mkSingleton(
           dbg(`Checking ${workerPath} for a worker entry point`)
           if (existsSync(workerPath)) {
             dbg(`Found worker ${workerPath}`)
-            await _writeUserLog(`Starting worker`)
+            healthyGuard()
+            await writeUserLog(`Starting worker`)
+            healthyGuard()
             const api = await createDenoProcess({
               path: workerPath,
               port: newPort,
@@ -323,7 +347,7 @@ export const instanceService = mkSingleton(
           }
         })()
         shutdownManager.add(async () => {
-          await _writeUserLog(`Shutting down worker`).catch(error)
+          await writeUserLog(`Shutting down worker`).catch(error)
           await denoApi?.shutdown().catch(error)
         })
 
@@ -355,7 +379,8 @@ export const instanceService = mkSingleton(
                 lastRequest + DAEMON_PB_IDLE_TTL < now()
               ) {
                 dbg(`idle for ${DAEMON_PB_IDLE_TTL}, shutting down`)
-                await shutdown()
+                healthyGuard()
+                await _safeShutdown().catch(error)
                 return false
               } else {
                 raw(`${openRequestCount} requests remain open`)
@@ -369,7 +394,7 @@ export const instanceService = mkSingleton(
         {
           tm.repeat(
             () =>
-              _pingInvocation(invocation)
+              pingInvocation(invocation)
                 .then(() => true)
                 .catch((e) => {
                   warn(`_pingInvocation failed with ${e}`)
@@ -381,10 +406,11 @@ export const instanceService = mkSingleton(
 
         dbg(`${internalUrl} is running`)
         status = InstanceApiStatus.Healthy
-        await _updateInstanceStatus(instance.id, InstanceStatus.Running)
+        healthyGuard()
+        await updateInstanceStatus(instance.id, InstanceStatus.Running)
       })().catch((e) => {
         warn(`Instance failed to start with ${e}`)
-        shutdown().catch(e)
+        _safeShutdown(e).catch(error)
       })
 
       return api

packages/daemon/src/services/PocketBaseService.ts

@@ -6,11 +6,7 @@ import {
   smartFetch,
   tryFetch,
 } from '$util'
-import {
+import { createCleanupManager, createTimerManager } from '@pockethost/common'
-  createCleanupManager,
-  createTimerManager,
-  safeCatch,
-} from '@pockethost/common'
 import {
   mkSingleton,
   SingletonBaseConfig,
@@ -23,6 +19,7 @@ import { type } from 'os'
 import { join } from 'path'
 import { gte, maxSatisfying, rsort } from 'semver'
 import { AsyncReturnType } from 'type-fest'
+import { AsyncContext } from '../util/AsyncContext'
 
 export type PocketbaseCommand = 'serve' | 'migrate'
 export type SpawnConfig = {
@@ -31,7 +28,7 @@ export type SpawnConfig = {
   version?: string
   port?: number
   isMothership?: boolean
-  onUnexpectedStop?: (
+  onUnexpectedStop: (
     code: number | null,
     stdout: string[],
     stderr: string[]
@@ -149,17 +146,13 @@ export const createPocketbaseService = async (
     }
   }
 
-  const _spawn = safeCatch(
+  const _spawn = async (cfg: SpawnConfig, context?: AsyncContext) => {
-    `spawnInstance`,
+    const logger = (context?.logger || _serviceLogger).create('spawn')
-    _serviceLogger,
+    const { dbg, warn, error } = logger
-    async (cfg: SpawnConfig) => {
     const _cfg: Required<SpawnConfig> = {
       version: maxVersion,
       port: await getPort(),
       isMothership: false,
-        onUnexpectedStop: (code) => {
-          dbg(`Unexpected stop default handler. Exit code: ${code}`)
-        },
       ...cfg,
     }
     const { version, command, slug, port, onUnexpectedStop, isMothership } =
@@ -236,7 +229,10 @@ export const createPocketbaseService = async (
 
     const url = mkInternalUrl(port)
     if (command === 'serve') {
-        await tryFetch(_serviceLogger)(url, async () => isRunning)
+      await tryFetch(url, {
+        preflight: async () => isRunning,
+        logger: _serviceLogger,
+      })
     }
     const api: PocketbaseProcess = {
       url,
@@ -275,7 +271,6 @@ export const createPocketbaseService = async (
     }
     return api
   }
-  )
 
   const shutdown = () => {
     dbg(`Shutting down pocketbaseService`)

packages/daemon/src/services/clientService/InstanceMIxin.ts

@@ -10,6 +10,7 @@ import {
 import { reduce } from '@s-libs/micro-dash'
 import Bottleneck from 'bottleneck'
 import { endOfMonth, startOfMonth } from 'date-fns'
+import { AsyncContext } from '../../util/AsyncContext'
 import { MixinContext } from './PbClient'
 
 export type InstanceApi = ReturnType<typeof createInstanceMixin>
@@ -59,13 +60,11 @@ export const createInstanceMixin = (context: MixinContext) => {
         })
   )
 
-  const getInstanceById = safeCatch(
+  const getInstanceById = async (
-    `getInstanceById`,
+    instanceId: InstanceId,
-    logger,
+    context?: AsyncContext
-    async (
+  ): Promise<[InstanceFields, UserFields] | []> =>
-      instanceId: InstanceId
+    client
-    ): Promise<[InstanceFields, UserFields] | []> => {
-      return client
       .collection(INSTANCE_COLLECTION)
       .getOne<InstanceFields>(instanceId, {
         $autoCancel: false,
@@ -81,8 +80,6 @@ export const createInstanceMixin = (context: MixinContext) => {
             return [instance, user]
           })
       })
-    }
-  )
 
   const updateInstance = safeCatch(
     `updateInstance`,

packages/daemon/src/util/AsyncContext.ts

@@ -0,0 +1,5 @@
+import { Logger } from '@pockethost/common'
+
+export type AsyncContext = {
+  logger: Logger
+}

packages/daemon/src/util/tryFetch.ts

@@ -1,12 +1,23 @@
-import { Logger, safeCatch } from '@pockethost/common'
+import { logger as defaultLogger } from '@pockethost/common'
+import fetch from 'node-fetch'
+import { AsyncContext } from './AsyncContext'
 
-const TRYFETCH_RETRY_MS = 50
+export const TRYFETCH_RETRY_MS = 50
-export const tryFetch = (logger: Logger) =>
+
-  safeCatch(
+export type Config = Required<AsyncContext> & {
-    `tryFetch`,
+  preflight: () => Promise<boolean>
-    logger,
+  retryMs: number
-    (url: string, preflight?: () => Promise<boolean>) => {
+}
-      const { dbg } = logger.create('tryFetch')
+
+export const tryFetch = async (url: string, config?: Partial<Config>) => {
+  const { logger, preflight, retryMs }: Config = {
+    logger: defaultLogger(),
+    preflight: async () => true,
+    retryMs: TRYFETCH_RETRY_MS,
+    ...config,
+  }
+  const _logger = logger.create(`tryFetch`)
+  const { dbg } = _logger
   return new Promise<void>((resolve, reject) => {
     const tryFetch = async () => {
       if (preflight) {
@@ -29,12 +40,11 @@ export const tryFetch = (logger: Logger) =>
         resolve()
       } catch (e) {
         dbg(
-              `Could not fetch ${url}, trying again in ${TRYFETCH_RETRY_MS}ms. Raw error ${e}`
+          `Could not fetch ${url}, trying again in ${retryMs}ms. Raw error ${e}`
         )
-            setTimeout(tryFetch, TRYFETCH_RETRY_MS)
+        setTimeout(tryFetch, retryMs)
       }
     }
     tryFetch()
   })
-    }
+}
-  )