fix(settings): сохранять результат операций базы при вторичных сбоях
This commit is contained in:
1 parent
738c98b9d3
commit
92b65a30ce
15 files changed
+159
-32
No files matched your search
@@ -5,7 +5,9 @@ import os from "node:os"
|
||||
import path from "node:path"
|
||||
import { executeSystemDatabaseRestore, validateSystemDatabaseRestoreArchive, type SystemDatabaseRestoreOutcome } from "./system-database-restore-execution.js"
|
||||
import { env } from "../config.js"
|
||||
import { withAdvisoryLock } from "../db/index.js"
|
||||
import { pool } from "../db/index.js"
|
||||
import { withOperationLock } from "../db/operation-lock.js"
|
||||
import { executeDatabaseOperation, type DatabaseOperationWarnings } from "./system-database-operation.js"
|
||||
import { refreshScheduler, stopScheduler } from "./scheduler.js"
|
||||
import {
|
||||
startTrafficFlowListener,
|
||||
@@ -34,7 +36,7 @@ function spawnCollected(command: string, args: string[], opts?: { stdin?: Buffer
|
||||
child.on("error", reject)
|
||||
child.on("close", (code) => {
|
||||
if (code !== 0) {
|
||||
reject(new Error(err.trim() || `${command} завершился с кодом ${code}`))
|
||||
reject(new Error(err.trim().replaceAll(env.DATABASE_URL, "[DATABASE_URL]") || `${command} завершился с кодом ${code}`))
|
||||
return
|
||||
}
|
||||
resolve(Buffer.concat(chunks))
|
||||
@@ -44,20 +46,16 @@ function spawnCollected(command: string, args: string[], opts?: { stdin?: Buffer
|
||||
})
|
||||
}
|
||||
|
||||
async function withDatabaseOperation<T>(fn: () => Promise<T>): Promise<T> {
|
||||
return await withAdvisoryLock(BACKUP_LOCK_KEY, async () => {
|
||||
await stopTrafficFlowListener()
|
||||
stopScheduler()
|
||||
try {
|
||||
return await fn()
|
||||
} finally {
|
||||
await startTrafficFlowListener()
|
||||
await refreshScheduler()
|
||||
}
|
||||
})
|
||||
async function withDatabaseOperation<T extends object>(fn: () => Promise<T>): Promise<T & DatabaseOperationWarnings> {
|
||||
return withOperationLock(() => pool.connect(), BACKUP_LOCK_KEY, () => executeDatabaseOperation(fn, {
|
||||
stopListener: stopTrafficFlowListener,
|
||||
stopScheduler,
|
||||
startListener: startTrafficFlowListener,
|
||||
startScheduler: refreshScheduler,
|
||||
}), "Операция с базой приложения уже выполняется")
|
||||
}
|
||||
|
||||
export async function exportSystemDatabaseBackup(): Promise<{ filename: string; buffer: Buffer }> {
|
||||
export async function exportSystemDatabaseBackup(): Promise<{ filename: string; buffer: Buffer } & DatabaseOperationWarnings> {
|
||||
return await withDatabaseOperation(async () => {
|
||||
const buffer = await spawnCollected("pg_dump", ["-Fc", env.DATABASE_URL])
|
||||
return {
|
||||
@@ -67,7 +65,7 @@ export async function exportSystemDatabaseBackup(): Promise<{ filename: string;
|
||||
})
|
||||
}
|
||||
|
||||
export async function restoreSystemDatabaseBackup(buffer: Buffer): Promise<SystemDatabaseRestoreOutcome> {
|
||||
export async function restoreSystemDatabaseBackup(buffer: Buffer): Promise<SystemDatabaseRestoreOutcome & DatabaseOperationWarnings> {
|
||||
validateSystemDatabaseRestoreArchive(buffer)
|
||||
return await withDatabaseOperation(() => executeSystemDatabaseRestore(buffer, env.DATABASE_URL, {
|
||||
allocate: async () => {
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
import assert from "node:assert/strict"
|
||||
import {test} from "node:test"
|
||||
import {executeDatabaseOperation} from "./system-database-operation.js"
|
||||
function fixture(fail: string[] = []) {
|
||||
const calls:string[]=[]
|
||||
const step=(name:string)=>async()=>{calls.push(name);if(fail.includes(name))throw Error(name+" failed")}
|
||||
return {calls,deps:{stopListener:step("stop listener"),stopScheduler:step("stop scheduler"),startListener:step("start listener"),startScheduler:step("start scheduler")},run:async()=>{await step("command")();return {cleanupError:"cleanup warning"}}}
|
||||
}
|
||||
test("confirmed command survives independent restart failures",async()=>{
|
||||
for(const fail of [[],["start listener"],["start scheduler"],["start listener","start scheduler"]]){
|
||||
const f=fixture(fail);const result=await executeDatabaseOperation(f.run,f.deps)
|
||||
assert.equal(result.cleanupError,"cleanup warning");assert.equal(!!result.listenerError,fail.includes("start listener"));assert.equal(!!result.schedulerError,fail.includes("start scheduler"))
|
||||
assert.deepEqual(f.calls,["stop listener","stop scheduler","command","start listener","start scheduler"])
|
||||
}
|
||||
})
|
||||
test("core failure survives restart failures without command retry",async()=>{
|
||||
const f=fixture(["command","start listener","start scheduler"])
|
||||
await assert.rejects(executeDatabaseOperation(f.run,f.deps),/command failed/)
|
||||
assert.equal(f.calls.filter(s=>s==="command").length,1);assert.deepEqual(f.calls.slice(-2),["start listener","start scheduler"])
|
||||
})
|
||||
test("stop failure prevents command and still attempts both restarts",async()=>{
|
||||
for(const step of ["stop listener","stop scheduler"]){const f=fixture([step,"start listener"]);await assert.rejects(executeDatabaseOperation(f.run,f.deps),new RegExp(step+" failed"));assert.equal(f.calls.includes("command"),false);assert.deepEqual(f.calls.slice(-2),["start listener","start scheduler"])}
|
||||
})
|
||||
@@ -0,0 +1,23 @@
|
||||
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
|
||||
}
|
||||
/** Restart both consumers independently; secondary errors cannot erase a completed command. */
|
||||
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 {
|
||||
try { await deps.startListener() } catch { warnings.listenerError = "Не удалось перезапустить приём NetFlow" }
|
||||
try { await deps.startScheduler() } catch { warnings.schedulerError = "Не удалось перезапустить планировщик" }
|
||||
}
|
||||
if (failed) throw failure
|
||||
return { ...result!, ...warnings }
|
||||
}
|
||||
Reference in new issue
Block a user