103 lines
6.2 KiB
TypeScript
103 lines
6.2 KiB
TypeScript
import { AsyncLocalStorage } from "node:async_hooks"
|
|
import { pool } from "../db/index.js"
|
|
import { assertOperationConfigurationLockScope, borrowOperationConfigurationLockClient, captureOperationConfigurationLockHandoff } from "./operation-config-lock-lease.js"
|
|
|
|
type LockCompletion = { active: boolean; readonly completed: Promise<void> }
|
|
type ServerLockCompletion = LockCompletion & { ids: ReadonlySet<number>; readonly parent: ServerLockCompletion | undefined }
|
|
const held = new AsyncLocalStorage<ServerLockCompletion>()
|
|
const applicationLease = new AsyncLocalStorage<LockCompletion>()
|
|
const collectorLease = new AsyncLocalStorage<LockCompletion>()
|
|
/** Read-only scheduler handoff; never exposes a client or admits arbitrary work. */
|
|
export function captureConfigurationLockHandoff(): Promise<void> | undefined {
|
|
const completions: Promise<void>[] = []
|
|
const application = applicationLease.getStore()
|
|
if (application) {
|
|
if (!application.active) throw new Error("Configuration lock scope has completed")
|
|
completions.push(application.completed)
|
|
}
|
|
const collector = collectorLease.getStore()
|
|
if (collector) {
|
|
if (!collector.active) throw new Error("Collector lock scope has completed")
|
|
completions.push(collector.completed)
|
|
}
|
|
for (let server = held.getStore(); server; server = server.parent) {
|
|
if (!server.active) throw new Error("Configuration lock scope has completed")
|
|
completions.push(server.completed)
|
|
}
|
|
return completions.length ? Promise.all(completions).then(() => {}) : undefined
|
|
}
|
|
export function assertNoConfigurationLockContext(): void {
|
|
if (held.getStore() || applicationLease.getStore() || collectorLease.getStore()) throw new Error("Scheduler bootstrap inside configuration lock context")
|
|
}
|
|
export async function withApplicationMutationLock<T>(work: () => Promise<T>): Promise<T> {
|
|
assertOperationConfigurationLockScope()
|
|
if (applicationLease.getStore()?.active) return work()
|
|
if (collectorLease.getStore()?.active) throw Object.assign(new Error("Изменение конфигурации внутри общего сбора запрещено"), { statusCode: 409 })
|
|
let complete!: () => void
|
|
const completed = new Promise<void>(resolve => { complete = resolve })
|
|
const client = borrowOperationConfigurationLockClient() ?? await pool.connect(), context = { active: true, completed }
|
|
let acquired = false, unsafe = false
|
|
try {
|
|
acquired = (await client.query("SELECT pg_try_advisory_lock(hashtextextended('mm-config:application',0)) AS acquired")).rows[0].acquired
|
|
if (!acquired) throw Object.assign(new Error("Глобальная конфигурация изменяется другой операцией"), { statusCode: 409 })
|
|
return await applicationLease.run(context, work)
|
|
} finally {
|
|
context.active = false
|
|
if (acquired) try { await client.query("SELECT pg_advisory_unlock(hashtextextended('mm-config:application',0))") } catch { unsafe = true }
|
|
try { client.release(unsafe) } finally { complete() }
|
|
}
|
|
}
|
|
/** Collectors coexist with each other but exclude configuration mutation for
|
|
* the complete native/sample/status operation. This lock grants no authority.
|
|
* Verified operations retain their original exact exclusive client protocol. */
|
|
export async function withApplicationCollectorLock<T>(work: () => Promise<T>): Promise<T> {
|
|
assertOperationConfigurationLockScope()
|
|
if (applicationLease.getStore()?.active || captureOperationConfigurationLockHandoff()) return withApplicationMutationLock(work)
|
|
if (collectorLease.getStore()?.active) return work()
|
|
let complete!: () => void
|
|
const completed = new Promise<void>(resolve => { complete = resolve })
|
|
const client = await pool.connect(), context = { active: true, completed }
|
|
let acquired = false, unsafe = false
|
|
try {
|
|
acquired = (await client.query("SELECT pg_try_advisory_lock_shared(hashtextextended('mm-config:application',0)) AS acquired")).rows[0].acquired
|
|
if (!acquired) throw Object.assign(new Error("Глобальная конфигурация изменяется другой операцией"), { statusCode: 409 })
|
|
return await collectorLease.run(context, work)
|
|
} finally {
|
|
context.active = false
|
|
if (acquired) try { await client.query("SELECT pg_advisory_unlock_shared(hashtextextended('mm-config:application',0))") } catch { unsafe = true }
|
|
try { client.release(unsafe) } finally { complete() }
|
|
}
|
|
}
|
|
/** Shared by UI, MCP and background tasks; sorted acquisition avoids cross-router deadlocks. */
|
|
export async function withConfigMutationLock<T>(serverIds: number[], work: () => Promise<T>): Promise<T> {
|
|
assertOperationConfigurationLockScope()
|
|
const parent = held.getStore()
|
|
const inherited = parent?.active ? parent.ids : new Set<number>()
|
|
const requested = [...new Set(serverIds)].sort((a, b) => a - b).filter(id => !inherited.has(id))
|
|
if (!requested.length) return work()
|
|
// Never extend an existing multi-router lease out of order.
|
|
if (inherited.size && requested[0] < Math.max(...inherited)) throw new Error("Некорректный порядок блокировок серверов")
|
|
const client = borrowOperationConfigurationLockClient() ?? await pool.connect()
|
|
const acquired: number[] = []
|
|
let unsafe = false
|
|
let complete!: () => void
|
|
const completed = new Promise<void>(resolve => { complete = resolve })
|
|
const context = { ids: new Set([...inherited, ...requested]), active: true, completed, parent }
|
|
try {
|
|
for (const id of requested) {
|
|
const { rows: [row] } = await client.query<{ acquired: boolean }>("SELECT pg_try_advisory_lock(hashtextextended($1,0)) AS acquired", ["mm-config:" + id])
|
|
if (!row.acquired) throw Object.assign(new Error("Конфигурация сервера изменяется другой операцией"), { statusCode: 409 })
|
|
acquired.push(id)
|
|
}
|
|
return await held.run(context, work)
|
|
} finally {
|
|
// Fire-and-forget children may inherit AsyncLocalStorage after work resolves.
|
|
// They must acquire a new lease instead of trusting an already released one.
|
|
context.active = false
|
|
for (const id of acquired.reverse()) {
|
|
try { await client.query("SELECT pg_advisory_unlock(hashtextextended($1,0))", ["mm-config:" + id]) } catch { unsafe = true }
|
|
}
|
|
try { client.release(unsafe) } finally { complete() }
|
|
}
|
|
}
|