Add configurable server destruction concurrency
This commit is contained in:
@@ -570,83 +570,105 @@ function betaRemovalEvidence(server: string, message: string): ProvisioningBlock
|
||||
return { code: 'active-beta', message, automatic: true, href: beta ? `/betas/${beta.id}` : null, label: 'Blocking beta' };
|
||||
}
|
||||
|
||||
interface DestructionJob { id: string; server: string; stage: string; server_deleted_at: string | null }
|
||||
|
||||
async function runDestructions(): Promise<void> {
|
||||
const queued = db().prepare(`SELECT id, server, stage, server_deleted_at FROM server_deployments WHERE action = 'destroy' AND state = 'waiting'
|
||||
AND (server_deleted_at IS NULL OR stage = ?) ORDER BY started_at`).all(REMOVING_SERVER) as { id: string; server: string; stage: string; server_deleted_at: string | null }[];
|
||||
for (const operation of queued) {
|
||||
if (creating || destroying.has(operation.server)) continue;
|
||||
if (!db().prepare("SELECT id FROM server_deployments WHERE id = ? AND state = 'waiting'").get(operation.id)) continue;
|
||||
destroying.add(operation.server);
|
||||
try {
|
||||
if (operation.stage === REMOVING_SERVER) {
|
||||
if (operation.server_deleted_at && getServer(operation.server)) {
|
||||
finishDeployment(operation.id, { ok: false, message: 'A replacement server is registered; local cleanup cannot safely resume.' });
|
||||
continue;
|
||||
}
|
||||
const newer = db().prepare(`SELECT 1 FROM server_deployments WHERE server = ? AND action = 'create'
|
||||
AND rowid > (SELECT rowid FROM server_deployments WHERE id = ?)`).get(operation.server, operation.id);
|
||||
if (newer) {
|
||||
finishDeployment(operation.id, { ok: false, message: 'A newer creation exists; local cleanup cannot safely resume.' });
|
||||
continue;
|
||||
}
|
||||
if (!claimRemoval(operation.id)) continue;
|
||||
finishDeployment(operation.id, await cleanupDestroyedServer(operation.server, operation.id));
|
||||
continue;
|
||||
AND (server_deleted_at IS NULL OR stage = ?) ORDER BY started_at, rowid`).all(REMOVING_SERVER) as DestructionJob[];
|
||||
const active = new Set<Promise<void>>();
|
||||
let next = 0;
|
||||
try {
|
||||
while (next < queued.length || active.size) {
|
||||
// Recheck the saved limit before each admission. Lowering it never
|
||||
// interrupts an active removal; its slot lasts through local cleanup.
|
||||
while (next < queued.length && !creating && destroying.size < fleetSettings().destroyConcurrency
|
||||
&& active.size < fleetSettings().destroyConcurrency) {
|
||||
const task = processDestruction(queued[next++]).finally(() => { active.delete(task); });
|
||||
active.add(task);
|
||||
}
|
||||
if (!getServer(operation.server)) {
|
||||
if (hasActiveCreation(operation.server)) {
|
||||
const creatingOperation = db().prepare(`SELECT id FROM server_deployments WHERE server = ? AND server_deleted_at IS NULL
|
||||
AND action IN ('create','rebuild') AND state IN ('queued','running','awaiting-health') LIMIT 1`).get(operation.server) as { id: string } | undefined;
|
||||
deploymentStage(operation.id, 'Waiting for server creation to finish', { phase: 'eligibility', checks: [],
|
||||
blocker: { code: 'creation-pending', message: 'Waiting for server creation to finish', automatic: true,
|
||||
href: creatingOperation ? `/provisioning/${encodeURIComponent(creatingOperation.id)}` : null, label: 'Creation operation' } });
|
||||
continue;
|
||||
}
|
||||
finishDeployment(operation.id, { ok: false, message: 'Server is no longer registered' });
|
||||
continue;
|
||||
}
|
||||
const blocker = activeBetaRemovalBlocker(operation.server);
|
||||
if (blocker) { deploymentStage(operation.id, blocker, { phase: 'eligibility', checks: [], blocker: betaRemovalEvidence(operation.server, blocker) }); continue; }
|
||||
// A fresh successful zero is required; cached occupancy and timeouts are not evidence of emptiness.
|
||||
const evidence = await probePlayerEvidence(operation.server);
|
||||
let players = evidence.players;
|
||||
let occupancy: NonNullable<ProgressUpdate['checks']>[number] = { key: 'occupancy', label: 'Player occupancy',
|
||||
result: players === 0 ? 'pass' : players === null ? 'unknown' : 'waiting', source: evidence.source,
|
||||
checkedAt: evidence.observedAt, receivedAt: evidence.receivedAt, expected: 0, actual: players };
|
||||
if (players === null) {
|
||||
const health = await komodoInspectHealthy(operation.server);
|
||||
if (['stopped','exited','created','dead'].includes(health.state?.toLowerCase() ?? '')) {
|
||||
players = 0;
|
||||
occupancy = { key: 'stopped', label: 'Stopped container permits removal', result: 'pass', source: 'komodo', actual: health.state ?? null };
|
||||
}
|
||||
}
|
||||
if (!db().prepare("SELECT id FROM server_deployments WHERE id = ? AND state = 'waiting'").get(operation.id)) continue;
|
||||
// Beta ownership may have changed while the network probes were in flight.
|
||||
const latestBlocker = activeBetaRemovalBlocker(operation.server);
|
||||
if (latestBlocker) { deploymentStage(operation.id, latestBlocker, { phase: 'eligibility', checks: [occupancy], blocker: betaRemovalEvidence(operation.server, latestBlocker) }); continue; }
|
||||
if (players !== 0) {
|
||||
const message = players === null
|
||||
? 'Waiting for player count: server could not be reached'
|
||||
: 'Waiting for server to empty: ' + players + ' player(s) online';
|
||||
deploymentStage(operation.id, message, { phase: 'eligibility', checks: [occupancy],
|
||||
blocker: { code: players === null ? 'players-unknown' : 'players-present', message, automatic: true } });
|
||||
continue;
|
||||
}
|
||||
if (!claimRemoval(operation.id)) continue;
|
||||
deploymentStage(operation.id, 'Server is empty (0 players verified)', { phase: 'eligibility', checks: [occupancy], blocker: null,
|
||||
checkpoint: { key: 'eligibility', label: 'Removal eligibility verified' } });
|
||||
const result = await destroyServerUnlocked(operation.server, operation.id);
|
||||
finishDeployment(operation.id, result);
|
||||
} catch (err) {
|
||||
const row = db().prepare('SELECT state FROM server_deployments WHERE id = ?').get(operation.id) as { state: string };
|
||||
if (row.state === 'waiting') deploymentStage(operation.id, 'Waiting for player count: ' + (err as Error).message, {
|
||||
phase: 'eligibility', checks: [], blocker: { code: 'players-unknown', message: (err as Error).message, automatic: true }
|
||||
});
|
||||
else if (row.state !== 'cancelled') finishDeployment(operation.id, { ok: false, message: (err as Error).message });
|
||||
} finally { destroying.delete(operation.server); }
|
||||
if (!active.size) break; // Creation or rebuilding holds the remaining capacity.
|
||||
await Promise.race(active);
|
||||
}
|
||||
} finally {
|
||||
// Keep the shared pass locked until every admitted removal has settled,
|
||||
// including when an unexpected persistence error escapes a task.
|
||||
await Promise.allSettled(active);
|
||||
}
|
||||
}
|
||||
|
||||
async function processDestruction(operation: DestructionJob): Promise<void> {
|
||||
if (creating || destroying.has(operation.server)) return;
|
||||
if (!db().prepare("SELECT id FROM server_deployments WHERE id = ? AND state = 'waiting'").get(operation.id)) return;
|
||||
destroying.add(operation.server);
|
||||
try {
|
||||
if (operation.stage === REMOVING_SERVER) {
|
||||
if (operation.server_deleted_at && getServer(operation.server)) {
|
||||
finishDeployment(operation.id, { ok: false, message: 'A replacement server is registered; local cleanup cannot safely resume.' });
|
||||
return;
|
||||
}
|
||||
const newer = db().prepare(`SELECT 1 FROM server_deployments WHERE server = ? AND action = 'create'
|
||||
AND rowid > (SELECT rowid FROM server_deployments WHERE id = ?)`).get(operation.server, operation.id);
|
||||
if (newer) {
|
||||
finishDeployment(operation.id, { ok: false, message: 'A newer creation exists; local cleanup cannot safely resume.' });
|
||||
return;
|
||||
}
|
||||
if (!claimRemoval(operation.id)) return;
|
||||
finishDeployment(operation.id, await cleanupDestroyedServer(operation.server, operation.id));
|
||||
return;
|
||||
}
|
||||
if (!getServer(operation.server)) {
|
||||
if (hasActiveCreation(operation.server)) {
|
||||
const creatingOperation = db().prepare(`SELECT id FROM server_deployments WHERE server = ? AND server_deleted_at IS NULL
|
||||
AND action IN ('create','rebuild') AND state IN ('queued','running','awaiting-health') LIMIT 1`).get(operation.server) as { id: string } | undefined;
|
||||
deploymentStage(operation.id, 'Waiting for server creation to finish', { phase: 'eligibility', checks: [],
|
||||
blocker: { code: 'creation-pending', message: 'Waiting for server creation to finish', automatic: true,
|
||||
href: creatingOperation ? `/provisioning/${encodeURIComponent(creatingOperation.id)}` : null, label: 'Creation operation' } });
|
||||
return;
|
||||
}
|
||||
finishDeployment(operation.id, { ok: false, message: 'Server is no longer registered' });
|
||||
return;
|
||||
}
|
||||
const blocker = activeBetaRemovalBlocker(operation.server);
|
||||
if (blocker) { deploymentStage(operation.id, blocker, { phase: 'eligibility', checks: [], blocker: betaRemovalEvidence(operation.server, blocker) }); return; }
|
||||
// A fresh successful zero is required; cached occupancy and timeouts are not evidence of emptiness.
|
||||
const evidence = await probePlayerEvidence(operation.server);
|
||||
let players = evidence.players;
|
||||
let occupancy: NonNullable<ProgressUpdate['checks']>[number] = { key: 'occupancy', label: 'Player occupancy',
|
||||
result: players === 0 ? 'pass' : players === null ? 'unknown' : 'waiting', source: evidence.source,
|
||||
checkedAt: evidence.observedAt, receivedAt: evidence.receivedAt, expected: 0, actual: players };
|
||||
if (players === null) {
|
||||
const health = await komodoInspectHealthy(operation.server);
|
||||
if (['stopped','exited','created','dead'].includes(health.state?.toLowerCase() ?? '')) {
|
||||
players = 0;
|
||||
occupancy = { key: 'stopped', label: 'Stopped container permits removal', result: 'pass', source: 'komodo', actual: health.state ?? null };
|
||||
}
|
||||
}
|
||||
if (!db().prepare("SELECT id FROM server_deployments WHERE id = ? AND state = 'waiting'").get(operation.id)) return;
|
||||
// Beta ownership may have changed while the network probes were in flight.
|
||||
const latestBlocker = activeBetaRemovalBlocker(operation.server);
|
||||
if (latestBlocker) { deploymentStage(operation.id, latestBlocker, { phase: 'eligibility', checks: [occupancy], blocker: betaRemovalEvidence(operation.server, latestBlocker) }); return; }
|
||||
if (players !== 0) {
|
||||
const message = players === null
|
||||
? 'Waiting for player count: server could not be reached'
|
||||
: 'Waiting for server to empty: ' + players + ' player(s) online';
|
||||
deploymentStage(operation.id, message, { phase: 'eligibility', checks: [occupancy],
|
||||
blocker: { code: players === null ? 'players-unknown' : 'players-present', message, automatic: true } });
|
||||
return;
|
||||
}
|
||||
if (!claimRemoval(operation.id)) return;
|
||||
deploymentStage(operation.id, 'Server is empty (0 players verified)', { phase: 'eligibility', checks: [occupancy], blocker: null,
|
||||
checkpoint: { key: 'eligibility', label: 'Removal eligibility verified' } });
|
||||
const result = await destroyServerUnlocked(operation.server, operation.id);
|
||||
finishDeployment(operation.id, result);
|
||||
} catch (err) {
|
||||
const row = db().prepare('SELECT state FROM server_deployments WHERE id = ?').get(operation.id) as { state: string };
|
||||
if (row.state === 'waiting') deploymentStage(operation.id, 'Waiting for player count: ' + (err as Error).message, {
|
||||
phase: 'eligibility', checks: [], blocker: { code: 'players-unknown', message: (err as Error).message, automatic: true }
|
||||
});
|
||||
else if (row.state !== 'cancelled') finishDeployment(operation.id, { ok: false, message: (err as Error).message });
|
||||
} finally { destroying.delete(operation.server); }
|
||||
}
|
||||
|
||||
async function destroyServerUnlocked(name: string, operation: string): Promise<DestroyServerResult> {
|
||||
const blocker = activeBetaRemovalBlocker(name);
|
||||
if (blocker) return { ok: false, kind: 'conflict', message: blocker, warnings: [] };
|
||||
|
||||
Reference in New Issue
Block a user