42 lines
2.0 KiB
TypeScript
42 lines
2.0 KiB
TypeScript
export type DatabaseOperationWarnings = { listenerError?: string; schedulerError?: string }
|
|
interface Dependencies {
|
|
stopListener: () => Promise<void> | void
|
|
stopScheduler: () => Promise<void> | void
|
|
startListener: () => Promise<void> | void
|
|
startScheduler: () => Promise<void> | void
|
|
beforeRestart?: () => Promise<void> | void
|
|
}
|
|
/** Process-local latch; durable restart recovery needs a separate control journal. */
|
|
export function createDatabaseSchemaReadiness() {
|
|
let ready = true
|
|
return {
|
|
beginReplacement() { ready = false },
|
|
confirmSchema() { ready = true },
|
|
assertReady() { if (!ready) throw new Error("Схема базы после восстановления не подтверждена") },
|
|
}
|
|
}
|
|
/** Restart both consumers independently only after the common readiness guard. */
|
|
export async function executeDatabaseOperation<T extends object>(run: () => Promise<T>, deps: Dependencies): Promise<T & DatabaseOperationWarnings> {
|
|
let result: T | undefined, failed = false, failure: unknown
|
|
const warnings: DatabaseOperationWarnings = {}
|
|
try {
|
|
await deps.stopListener()
|
|
await deps.stopScheduler()
|
|
result = await run()
|
|
} catch (error) { failed = true; failure = error }
|
|
finally {
|
|
let restartAllowed = true
|
|
try { await deps.beforeRestart?.() } catch {
|
|
restartAllowed = false
|
|
warnings.listenerError = "Приём NetFlow оставлен остановленным: схема базы не подтверждена"
|
|
warnings.schedulerError = "Планировщик оставлен остановленным: схема базы не подтверждена"
|
|
}
|
|
if (restartAllowed) {
|
|
try { await deps.startListener() } catch { warnings.listenerError = "Не удалось перезапустить приём NetFlow" }
|
|
try { await deps.startScheduler() } catch { warnings.schedulerError = "Не удалось перезапустить планировщик" }
|
|
}
|
|
}
|
|
if (failed) throw failure
|
|
return { ...result!, ...warnings }
|
|
}
|