Add exclusive map ownership and capacity-aware autoscaling creation
This commit is contained in:
@@ -0,0 +1,32 @@
|
||||
import { db } from '../db/index.js';
|
||||
import { planScaling } from './planner.js';
|
||||
import { scalingSnapshot } from './snapshot.js';
|
||||
import { liveScaling, type ScalingPolicy, type ScalingDecision } from '../../autoscaling.js';
|
||||
import { deploymentStage } from '../provisioning/history.js';
|
||||
|
||||
export interface ScalingIntent { policy_id: string; revision: number; decision: string; operation_id: string }
|
||||
export function scalingIntent(operationId: string): ScalingIntent | undefined {
|
||||
return db().prepare('SELECT policy_id,revision,decision,operation_id FROM autoscaling_decisions WHERE operation_id=? AND resolved=0').get(operationId) as ScalingIntent | undefined;
|
||||
}
|
||||
/** Rechecked before placement, after asynchronous preparation, and before dispatch. */
|
||||
export function scalingAdmission(operationId: string): string | null {
|
||||
const intent = scalingIntent(operationId);
|
||||
if (!intent) return null;
|
||||
const row = db().prepare('SELECT * FROM autoscaling_policies WHERE id=?').get(intent.policy_id) as { id: string; settings: string; mode: ScalingPolicy['mode']; revision: number; paused_reason: string | null; owns_map: number } | undefined;
|
||||
let reason: string | null = null;
|
||||
const paused = (db().prepare("SELECT value FROM settings WHERE key='autoscaling_paused'").get() as { value: string } | undefined)?.value === '1';
|
||||
if (!row || !liveScaling(row.mode) || !row.owns_map || row.revision !== intent.revision || row.paused_reason || paused) reason = 'Autoscaling policy is paused or changed.';
|
||||
else {
|
||||
const policy: ScalingPolicy = { ...JSON.parse(row.settings), id: row.id, mode: row.mode, revision: row.revision, nextActionAt: 0, pausedReason: null, updatedAt: '' };
|
||||
const requested = JSON.parse(intent.decision) as ScalingDecision;
|
||||
const snapshot = scalingSnapshot(policy, Date.now(), operationId);
|
||||
if (requested.action === 'create') {
|
||||
snapshot.reservedNames = snapshot.reservedNames.filter(n => n !== requested.server);
|
||||
snapshot.blockers = snapshot.blockers.filter(b => b !== `Unresolved native instance ${requested.server}.`);
|
||||
}
|
||||
const current = planScaling(policy, snapshot, Date.now());
|
||||
if (current.action !== requested.action || current.server !== requested.server) reason = `Autoscaling intent no longer eligible: ${current.reason}`;
|
||||
}
|
||||
if (reason) deploymentStage(operationId, reason, { blocker: { code: 'autoscaling', message: reason, automatic: true } });
|
||||
return reason;
|
||||
}
|
||||
@@ -3,6 +3,9 @@ import { db } from '../db/index.js';
|
||||
import { liveScaling, scalingSettings, type ScalingPolicy, type ScalingPreview, type ScalingReport, type ScalingView } from '../../autoscaling.js';
|
||||
import { emptyObservation, planScaling } from './planner.js';
|
||||
import { scalingSnapshot } from './snapshot.js';
|
||||
import { nativeLaunchSelection, queueNativeCreationIntent } from '../provisioning/native.js';
|
||||
import { cancelCreation } from '../provisioning/cancel-creation.js';
|
||||
import { scalingAdmission } from './admission.js';
|
||||
|
||||
export class ScalingError extends Error { constructor(message: string, public status = 400) { super(message); } }
|
||||
interface PolicyRow { id: string; revision: number; settings: string; mode: ScalingPolicy['mode']; next_action_at: number; paused_reason: string | null; updated_at: string }
|
||||
@@ -19,7 +22,15 @@ export function previewScaling(input: unknown, id?: string, revision?: number):
|
||||
const policy: ScalingPolicy = { ...settings, id: existing?.id ?? 'preview', revision: existing?.revision ?? 0,
|
||||
nextActionAt: existing?.nextActionAt ?? 0, pausedReason: null, updatedAt: existing?.updatedAt ?? '' };
|
||||
const snapshot = scalingSnapshot(policy);
|
||||
const blockers = liveScaling(policy.mode) ? ['Live autoscaling is not available yet.'] : [];
|
||||
const blockers: string[] = [];
|
||||
if (policy.mode === 'create-destroy') blockers.push('Automatic removal is not available yet.');
|
||||
if (liveScaling(policy.mode)) {
|
||||
if (snapshot.keyedNames.length < policy.maximum) blockers.push(`Save keys for at least ${policy.maximum} lobby identities before activating this policy.`);
|
||||
blockers.push(...snapshot.blockers);
|
||||
for (const member of snapshot.members) if (!['occupied', 'empty'].includes(member.state)) blockers.push(`${member.name}: ${member.blocker ?? member.state}. Settle this work before activation.`);
|
||||
if (db().prepare('SELECT 1 FROM autoscaling_decisions WHERE policy_id=? AND resolved=0').get(policy.id)) blockers.push('Settle the current autoscaling operation before editing a live policy.');
|
||||
try { nativeLaunchSelection({ automatic: true }, policy.game); } catch (error) { blockers.push((error as Error).message); }
|
||||
}
|
||||
const decision = planScaling(policy, snapshot, Date.now());
|
||||
const token = createHash('sha256').update(JSON.stringify({ settings, revision: policy.revision, globalPaused: scalingGloballyPaused(),
|
||||
members: snapshot.members.map(m => [m.name, m.identity, m.state]), keys: snapshot.keyedNames.sort(), reservations: snapshot.reservedNames.sort(), blockers: snapshot.blockers, decision })).digest('hex');
|
||||
@@ -37,21 +48,56 @@ export function saveScaling(input: unknown, id?: string, revision?: number, toke
|
||||
ON CONFLICT(id) DO UPDATE SET settings=excluded.settings,mode=excluded.mode,revision=autoscaling_policies.revision+1,paused_reason=NULL,updated_at=excluded.updated_at`).run(nextId, policy.game, policy.mapSlug, JSON.stringify(settings), policy.mode, at);
|
||||
// An edit starts a new observation interval; earlier inactivity is not authorization.
|
||||
db().prepare('DELETE FROM autoscaling_members WHERE policy_id=?').run(nextId);
|
||||
db().prepare('UPDATE autoscaling_policies SET owns_map=CASE WHEN ? THEN 1 ELSE owns_map END WHERE id=?').run(liveScaling(policy.mode) ? 1 : 0, nextId);
|
||||
if (liveScaling(policy.mode)) for (const member of preview.snapshot.members) db().prepare('INSERT INTO autoscaling_members(policy_id,server,identity,observation,adopted) VALUES(?,?,?,?,1)')
|
||||
.run(nextId, member.name, member.identity, JSON.stringify({ identity: member.identity, observedAt: member.observedAt, emptySince: null, evaluatedAt: Date.now() }));
|
||||
reconcileScaling(getScalingPolicy(nextId)!);
|
||||
return getScalingPolicy(nextId)!;
|
||||
})();
|
||||
}
|
||||
|
||||
export function setScalingGlobalPause(paused: boolean): void {
|
||||
db().prepare("INSERT INTO settings(key,value) VALUES('autoscaling_paused',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value").run(paused ? '1' : '0');
|
||||
db().transaction(() => {
|
||||
db().prepare("INSERT INTO settings(key,value) VALUES('autoscaling_paused',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value").run(paused ? '1' : '0');
|
||||
if (paused) for (const policy of listScalingPolicies()) reconcileScaling(policy);
|
||||
})();
|
||||
}
|
||||
|
||||
export async function runAutoscaling(): Promise<void> {
|
||||
function reconcileScaling(policy: ScalingPolicy): void {
|
||||
const intents = db().prepare(`SELECT a.id,a.revision,d.id AS operation_id,d.action,d.state,d.submitted_at,n.operation_id AS command_id
|
||||
FROM autoscaling_decisions a JOIN server_deployments d ON d.id=a.operation_id
|
||||
LEFT JOIN native_provisioning n ON n.deployment_id=d.id WHERE a.policy_id=? AND a.resolved=0`).all(policy.id) as
|
||||
{ id: string; revision: number; operation_id: string; action: string; state: string; submitted_at: string | null; command_id: string | null }[];
|
||||
for (const intent of intents) {
|
||||
if (intent.action === 'create' && !intent.command_id && !intent.submitted_at
|
||||
&& (scalingAdmission(intent.operation_id) || intent.state === 'cancelled')) cancelCreation(intent.operation_id, 'autoscaling-withdrawal', false);
|
||||
const current = db().prepare('SELECT state,server_deleted_at FROM server_deployments WHERE id=?').get(intent.operation_id) as { state: string; server_deleted_at: string | null };
|
||||
const terminal = ['done','warning','failed','cancelled'].includes(current.state);
|
||||
const command = intent.command_id ? db().prepare('SELECT state FROM ec_host_operations WHERE id=?').get(intent.command_id) as { state: string } | undefined : undefined;
|
||||
const receiptPending = current.state === 'cancelled' && !current.server_deleted_at && command?.state === 'completed'
|
||||
&& !!db().prepare('SELECT 1 FROM native_provisioning WHERE deployment_id=? AND applied_at IS NULL').get(intent.operation_id);
|
||||
const settled = (!intent.command_id || !!command && ['completed','failed'].includes(command.state)) && !receiptPending;
|
||||
if ((current.state === 'failed' || current.state === 'interrupted' || command?.state === 'unknown') && liveScaling(policy.mode)) {
|
||||
db().prepare('UPDATE autoscaling_policies SET paused_reason=?,updated_at=? WHERE id=?').run(`Inspect failed or uncertain Provisioning operation ${intent.operation_id} before resuming.`, new Date().toISOString(), policy.id);
|
||||
}
|
||||
if (terminal && settled) db().prepare('UPDATE autoscaling_decisions SET resolved=1 WHERE id=?').run(intent.id);
|
||||
}
|
||||
if (!liveScaling(policy.mode) && !db().prepare('SELECT 1 FROM autoscaling_decisions WHERE policy_id=? AND resolved=0').get(policy.id)) {
|
||||
db().prepare('UPDATE autoscaling_policies SET owns_map=0 WHERE id=?').run(policy.id);
|
||||
}
|
||||
}
|
||||
|
||||
export async function runAutoscaling(): Promise<boolean> {
|
||||
const now = Date.now();
|
||||
for (const policy of listScalingPolicies()) {
|
||||
let submitted = false;
|
||||
for (const initial of listScalingPolicies()) {
|
||||
reconcileScaling(initial);
|
||||
const policy = getScalingPolicy(initial.id)!;
|
||||
if (policy.mode === 'off') continue;
|
||||
db().transaction(() => {
|
||||
const snapshot = scalingSnapshot(policy, now);
|
||||
if (scalingGloballyPaused()) snapshot.blockers.unshift('Autoscaling is globally paused.');
|
||||
if (db().prepare('SELECT 1 FROM autoscaling_decisions WHERE policy_id=? AND resolved=0').get(policy.id)) snapshot.blockers.unshift('Waiting for the current autoscaling operation to settle.');
|
||||
const decision = planScaling(policy, snapshot, now);
|
||||
for (const member of snapshot.members) {
|
||||
const observation = emptyObservation(member, { identity: member.identity, observedAt: member.observedAt, emptySince: member.emptySince, evaluatedAt: now }, now);
|
||||
@@ -62,13 +108,34 @@ export async function runAutoscaling(): Promise<void> {
|
||||
const evidence = JSON.stringify({ members: snapshot.members, reservations: snapshot.reservedNames, keyCount: snapshot.keyedNames.length, blockers: snapshot.blockers });
|
||||
const fingerprint = createHash('sha256').update(JSON.stringify([policy.revision, decision, snapshot.members.map(m => [m.name, m.identity, m.state])])).digest('hex');
|
||||
const last = db().prepare('SELECT id,fingerprint FROM autoscaling_decisions WHERE policy_id=? ORDER BY evaluated_at DESC,rowid DESC LIMIT 1').get(policy.id) as { id: string; fingerprint: string } | undefined;
|
||||
if (last?.fingerprint === fingerprint) db().prepare('UPDATE autoscaling_decisions SET last_seen_at=?,evidence=? WHERE id=?').run(now, evidence, last.id);
|
||||
const execute = liveScaling(policy.mode) && decision.action === 'create';
|
||||
const decisionId = randomUUID();
|
||||
if (!execute && last?.fingerprint === fingerprint) db().prepare('UPDATE autoscaling_decisions SET last_seen_at=?,evidence=? WHERE id=?').run(now, evidence, last.id);
|
||||
else db().prepare('INSERT INTO autoscaling_decisions(id,policy_id,revision,evaluated_at,last_seen_at,fingerprint,decision,evidence) VALUES(?,?,?,?,?,?,?,?)')
|
||||
.run(randomUUID(), policy.id, policy.revision, now, now, fingerprint, JSON.stringify(decision), evidence);
|
||||
.run(decisionId, policy.id, policy.revision, now, now, fingerprint, JSON.stringify(decision), evidence);
|
||||
if (execute) {
|
||||
db().prepare('UPDATE autoscaling_decisions SET resolved=0 WHERE id=?').run(decisionId);
|
||||
try {
|
||||
db().transaction(() => {
|
||||
const selection = nativeLaunchSelection({ automatic: true }, policy.game);
|
||||
const result = queueNativeCreationIntent({ server: decision.server!, game: policy.game, mapSlug: policy.mapSlug, selection,
|
||||
createdBy: `autoscaling:${policy.id}`, autoscalingDecisionId: decisionId });
|
||||
if (!result.ok) throw new Error(result.message);
|
||||
db().prepare('UPDATE autoscaling_decisions SET operation_id=? WHERE id=?').run(result.deploymentId, decisionId);
|
||||
db().prepare("INSERT INTO provisioning_sources(operation_id,kind,source_id,server) VALUES(?,'autoscaling',?,?)").run(result.deploymentId, policy.id, decision.server);
|
||||
db().prepare('UPDATE autoscaling_policies SET next_action_at=? WHERE id=?').run(now + policy.cooldownMinutes * 60_000, policy.id);
|
||||
submitted = true;
|
||||
})();
|
||||
} catch (error) {
|
||||
db().prepare('UPDATE autoscaling_decisions SET resolved=1 WHERE id=?').run(decisionId);
|
||||
db().prepare('UPDATE autoscaling_policies SET paused_reason=? WHERE id=?').run(`Creation could not be queued: ${(error as Error).message}`, policy.id);
|
||||
}
|
||||
}
|
||||
db().prepare(`INSERT OR IGNORE INTO autoscaling_samples(policy_id,sampled_at,mode,occupied,empty,unknown,deficit) VALUES(?,?,?,?,?,?,?)`)
|
||||
.run(policy.id, Math.floor(now / 15_000) * 15_000, policy.mode, decision.occupied, decision.empty, decision.unknown, Math.max(0, policy.spares - decision.empty));
|
||||
})();
|
||||
}
|
||||
return submitted;
|
||||
}
|
||||
|
||||
export function pruneAutoscaling(now = Date.now()): void {
|
||||
@@ -93,5 +160,5 @@ export function scalingReport(): ScalingReport {
|
||||
COALESCE(SUM(CASE WHEN unknown>0 THEN MIN(duration,15000) ELSE 0 END)/60000.0,0) AS unknownMinutes FROM intervals WHERE duration<=30000`).get(policy.id) as ScalingView['totals'];
|
||||
return { policy, snapshot, decision, token: '', blockers: [], history, lastEvaluatedAt: history[0]?.lastSeenAt ?? null, totals };
|
||||
});
|
||||
return { policies, globalPaused: scalingGloballyPaused(), liveAvailable: false, removalAvailable: false };
|
||||
return { policies, globalPaused: scalingGloballyPaused(), liveAvailable: true, removalAvailable: false };
|
||||
}
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
import { db } from '../db/index.js';
|
||||
import { getServer } from '../fleet/servers.js';
|
||||
import { mapIdentity } from './snapshot.js';
|
||||
|
||||
export function scalingOwner(game: string, mapSlug: string): string | null {
|
||||
return (db().prepare('SELECT id FROM autoscaling_policies WHERE game=? AND map_slug=? AND owns_map=1').get(game, mapSlug) as { id: string } | undefined)?.id ?? null;
|
||||
}
|
||||
export function scalingOwnershipBlocker(server: string): string | null {
|
||||
const identity = mapIdentity(getServer(server) ?? { name: server });
|
||||
const owner = identity && scalingOwner(identity.game, identity.mapSlug);
|
||||
return owner ? `Autoscaling manages this map. Pause its policy and settle pending work before changing the lineup: /autoscaling#${owner}` : null;
|
||||
}
|
||||
export function assertScalingOwnership(game: string, mapSlug: string, decisionId?: string): void {
|
||||
const owner = scalingOwner(game, mapSlug);
|
||||
if (!owner) { if (decisionId) throw new Error('Autoscaling no longer owns this map.'); return; }
|
||||
if (decisionId && db().prepare('SELECT 1 FROM autoscaling_decisions WHERE id=? AND policy_id=? AND resolved=0 AND operation_id IS NULL').get(decisionId, owner)) return;
|
||||
throw new Error(`Autoscaling manages this map. Pause its policy and settle pending work before changing the lineup: /autoscaling#${owner}`);
|
||||
}
|
||||
/** Operator cancellation is a policy pause, not a fresh scaling signal. */
|
||||
export function pauseScalingForOperation(operationId: string): void {
|
||||
const row = db().prepare('SELECT policy_id FROM autoscaling_decisions WHERE operation_id=?').get(operationId) as { policy_id: string } | undefined;
|
||||
if (!row) return;
|
||||
db().prepare("UPDATE autoscaling_policies SET mode='off',revision=revision+1,paused_reason='Paused after operator cancellation.',updated_at=? WHERE id=?").run(new Date().toISOString(), row.policy_id);
|
||||
db().prepare('UPDATE autoscaling_authorizations SET revoked_at=? WHERE policy_id=? AND claimed_at IS NULL').run(Date.now(), row.policy_id);
|
||||
}
|
||||
Reference in New Issue
Block a user