josh/electric-cherry
1
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases Wiki Activity

Add safe manual instance rebalancing

Browse Source
This commit is contained in:
josh
2026-10-08 12:37:56 -04:00
parent 26de43639f
commit 012cd51a66
45 changed files with 2006 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/rebalancing/admission.ts
+26
View File
@@ -0,0 +1,26 @@
import { db } from '../db/index.js';
import type { RebalanceCandidate, RebalanceSettings } from '../../rebalancing.js';
import { rebalanceMoveBlocker } from './planner.js';
import { rebalanceSnapshot } from './snapshot.js';
/** A delivered migration must finish even after the operator stops its parent run. */
export function rebalanceAdmission(operationId: string): string | null {
const operation = db().prepare('SELECT triggered_by FROM server_deployments WHERE id=?').get(operationId) as { triggered_by: string } | undefined;
if (!operation?.triggered_by.startsWith('rebalance:')) return null;
const migration = db().prepare('SELECT remove_operation_id,removed_at FROM ec_host_migrations WHERE deployment_id=?')
.get(operationId) as { remove_operation_id: string | null; removed_at: string | null } | undefined;
if (migration?.remove_operation_id || migration?.removed_at) return null;
const row = db().prepare(`SELECT m.candidate,r.settings,r.state FROM rebalancing_moves m
JOIN rebalancing_runs r ON r.id=m.run_id WHERE m.operation_id=? AND m.state='moving'`).get(operationId) as
{ candidate: string; settings: string; state: string } | undefined;
if (!row || row.state !== 'running') return 'Rebalance run is no longer requesting this move.';
const settings = JSON.parse(row.settings) as RebalanceSettings;
return rebalanceMoveBlocker(rebalanceSnapshot(settings, operationId), JSON.parse(row.candidate) as RebalanceCandidate);
}
/** Cancelling an individual uncommitted move also withdraws the rest of that run. */
export function requestRebalanceStopForOperation(operationId: string): void {
db().prepare(`UPDATE rebalancing_runs SET state='stopping',updated_at=?,message='Stopping after the current move.'
WHERE state='running' AND id IN (SELECT run_id FROM rebalancing_moves WHERE operation_id=?)`)
.run(new Date().toISOString(), operationId);
}
src/lib/server/rebalancing/index.ts
+214
View File
@@ -0,0 +1,214 @@
import { randomUUID } from 'node:crypto';
import { db } from '../db/index.js';
import { getHost } from '../hosts/index.js';
import { requestInstanceMove } from '../provisioning/migrations.js';
import { cancelProvisioningJob } from '../provisioning/cancellation.js';
import { rebalanceSettings, type RebalanceCandidate, type RebalanceMove, type RebalancePreview,
type RebalanceReport, type RebalanceRun } from '../../rebalancing.js';
import { planRebalance, rebalanceMoveBlocker } from './planner.js';
import { rebalanceHosts, rebalanceSnapshot } from './snapshot.js';
import { rebalanceAdmission } from './admission.js';
export class RebalanceError extends Error { constructor(message: string, public status = 400) { super(message); } }
interface RunRow { id: string; settings: string; state: RebalanceRun['state']; created_at: string; updated_at: string; message: string }
interface MoveRow { id: string; candidate: string; state: RebalanceMove['state']; operation_id: string | null; message: string; updated_at: string }
const atNow = () => new Date().toISOString();
const movesFor = (id: string): RebalanceMove[] => (db().prepare('SELECT * FROM rebalancing_moves WHERE run_id=? ORDER BY ordinal').all(id) as MoveRow[])
.map(row => ({ ...JSON.parse(row.candidate), id: row.id, state: row.state, operationId: row.operation_id, message: row.message, updatedAt: row.updated_at }));
function present(row: RunRow): RebalanceRun {
return { id: row.id, settings: JSON.parse(row.settings), state: row.state, createdAt: row.created_at,
updatedAt: row.updated_at, message: row.message, moves: movesFor(row.id) };
}
export function getRebalanceRun(id: string): RebalanceRun | null {
const row = db().prepare('SELECT * FROM rebalancing_runs WHERE id=?').get(id) as RunRow | undefined;
return row ? present(row) : null;
}
export function rebalanceReport(): RebalanceReport {
return { hosts: rebalanceHosts(), runs: (db().prepare(`SELECT * FROM rebalancing_runs
ORDER BY state IN ('running','stopping') DESC,created_at DESC,rowid DESC LIMIT 20`).all() as RunRow[]).map(present) };
}
/** Preview has no resource reservations or host side effects. */
export function previewRebalance(input: unknown): RebalancePreview {
const settings = rebalanceSettings(input), now = Date.now();
const plan = planRebalance(rebalanceSnapshot(settings), settings);
const preview: RebalancePreview = { ...plan, settings, id: randomUUID(), createdAt: new Date(now).toISOString(),
expiresAt: new Date(now + 5 * 60_000).toISOString() };
db().prepare('INSERT INTO rebalancing_previews(id,preview,expires_at) VALUES(?,?,?)').run(preview.id, JSON.stringify(preview), preview.expiresAt);
return preview;
}
function hasOutstandingRun(): boolean {
return !!db().prepare(`SELECT 1 FROM rebalancing_runs WHERE state IN ('running','stopping')
UNION ALL SELECT 1 FROM rebalancing_moves m JOIN server_deployments d ON d.id=m.operation_id
WHERE d.state IN ('queued','waiting','running','awaiting-health','interrupted')
UNION ALL SELECT 1 FROM rebalancing_moves m JOIN native_provisioning n ON n.deployment_id=m.operation_id
LEFT JOIN ec_host_migrations migration ON migration.deployment_id=m.operation_id
JOIN ec_host_operations o ON o.id IN (n.operation_id,n.stage_operation_id,migration.remove_operation_id)
WHERE o.state NOT IN ('completed','failed') LIMIT 1`).get();
}
const moveSignature = (moves: RebalanceCandidate[]) => JSON.stringify(moves.map(move =>
[move.instanceId, move.generation, move.sourceHostId, move.targetHostId]));
export function startRebalance(previewId: string): RebalanceRun {
return db().transaction(() => {
const row = db().prepare('SELECT preview,expires_at,run_id FROM rebalancing_previews WHERE id=?').get(previewId) as
{ preview: string; expires_at: string; run_id: string | null } | undefined;
if (!row) throw new RebalanceError('Preview not found. Preview the distribution again.', 404);
// A retried request returns the original bounded run, even after its preview expires.
if (row.run_id) return getRebalanceRun(row.run_id)!;
if (Date.parse(row.expires_at) <= Date.now()) throw new RebalanceError('Preview expired. Preview the distribution again.', 409);
if (hasOutstandingRun()) throw new RebalanceError('Finish or stop the current rebalance and settle its operations before starting another.', 409);
const preview = JSON.parse(row.preview) as RebalancePreview;
const fresh = planRebalance(rebalanceSnapshot(preview.settings), preview.settings);
if (!fresh.moves.length || moveSignature(fresh.moves) !== moveSignature(preview.moves)) {
throw new RebalanceError('The fleet changed. Preview the distribution again before starting.', 409);
}
const id = randomUUID(), at = atNow();
db().prepare(`INSERT INTO rebalancing_runs(id,settings,state,created_at,updated_at,message)
VALUES(?,?,'running',?,?,'Waiting to begin the first move.')`).run(id, JSON.stringify(preview.settings), at, at);
for (const [ordinal, move] of fresh.moves.entries()) db().prepare(`INSERT INTO rebalancing_moves
(id,run_id,ordinal,server,instance_id,generation,source_host_id,target_host_id,candidate,state,message,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,'planned','Waiting for earlier moves.',?)`)
.run(randomUUID(), id, ordinal, move.server, move.instanceId, move.generation, move.sourceHostId, move.targetHostId, JSON.stringify(move), at);
db().prepare('UPDATE rebalancing_previews SET run_id=? WHERE id=?').run(id, previewId);
return getRebalanceRun(id)!;
})();
}
function updateRun(id: string, state: RebalanceRun['state'], message: string): void {
db().prepare('UPDATE rebalancing_runs SET state=?,message=?,updated_at=? WHERE id=? AND (state!=? OR message!=?)')
.run(state, message, atNow(), id, state, message);
}
function updateMove(id: string, state: RebalanceMove['state'], message: string): void {
db().prepare('UPDATE rebalancing_moves SET state=?,message=?,updated_at=? WHERE id=? AND (state!=? OR message!=?)')
.run(state, message, atNow(), id, state, message);
}
function skipRemaining(id: string, reason: string): void {
db().prepare("UPDATE rebalancing_moves SET state='skipped',message=?,updated_at=? WHERE run_id=? AND state='planned'").run(reason, atNow(), id);
}
function committed(operationId: string): boolean {
return !!db().prepare('SELECT 1 FROM ec_host_migrations WHERE deployment_id=? AND (remove_operation_id IS NOT NULL OR removed_at IS NOT NULL)').get(operationId);
}
export function stopRebalance(id: string): RebalanceRun {
return db().transaction(() => {
const run = getRebalanceRun(id);
if (!run) throw new RebalanceError('Rebalance run not found.', 404);
if (!['running', 'stopping'].includes(run.state)) return run;
updateRun(id, 'stopping', 'Stopping after the current move.');
skipRemaining(id, 'Stopped before this move began.');
for (const move of run.moves.filter(move => move.state === 'moving')) {
if (move.operationId && !committed(move.operationId)
&& cancelProvisioningJob(move.operationId, 'Rebalance stopped before source removal.')) {
updateMove(move.id, 'skipped', 'Stopped before source removal.');
}
}
if (!movesFor(id).some(move => move.state === 'moving')) updateRun(id, 'stopped', 'Stopped. Completed moves are retained.');
return getRebalanceRun(id)!;
})();
}
function refreshedAfterMove(move: RebalanceMove): boolean {
return [move.sourceHostId, move.targetHostId].every(id => {
const heartbeat = getHost(id)?.lastHeartbeat;
return !!heartbeat && Date.parse(heartbeat) > Date.parse(move.updatedAt);
});
}
/** Only a single approved move is admitted; Provisioning owns every host command. */
function reconcileRun(run: RebalanceRun): boolean {
let active = run.moves.find(move => move.state === 'moving');
if (active) {
const operation = active.operationId && db().prepare(`SELECT d.state,d.stage,d.error,n.verified_at
FROM server_deployments d LEFT JOIN native_provisioning n ON n.deployment_id=d.id WHERE d.id=?`).get(active.operationId) as
{ state: string; stage: string; error: string | null; verified_at: string | null } | undefined;
if (!operation) {
updateMove(active.id, 'failed', 'Provisioning operation is unavailable. Inspect the instance before starting another run.');
skipRemaining(run.id, 'Stopped after a move could not be reconciled.');
updateRun(run.id, 'failed', 'A move could not be reconciled.'); return false;
}
if (operation.state === 'cancelled') {
updateMove(active.id, 'skipped', operation.stage);
if (run.state === 'running') updateRun(run.id, 'stopping', 'An operation was cancelled; stopping the remaining moves.');
} else if (['done', 'warning'].includes(operation.state) && operation.verified_at) {
updateMove(active.id, 'done', 'Healthy on the destination host.');
} else if (['failed', 'warning', 'interrupted'].includes(operation.state)) {
updateMove(active.id, 'failed', operation.error ?? operation.stage);
skipRemaining(run.id, 'Stopped after a move needs attention.');
updateRun(run.id, 'failed', 'A move needs attention. Use its Provisioning operation to recover.'); return false;
} else {
if (active.operationId && !committed(active.operationId)) {
const reason = run.state === 'stopping' ? 'Rebalance stopped before source removal.' : rebalanceAdmission(active.operationId);
if (reason && cancelProvisioningJob(active.operationId, reason, { preserveRebalanceRun: true })) {
updateMove(active.id, 'skipped', reason);
} else {
updateMove(active.id, 'moving', operation.stage);
updateRun(run.id, run.state, operation.stage); return false;
}
} else {
updateMove(active.id, 'moving', operation.stage);
updateRun(run.id, run.state, run.state === 'stopping' ? 'Finishing the current move before stopping.' : operation.stage); return false;
}
}
run = getRebalanceRun(run.id)!;
active = undefined;
}
if (run.state === 'stopping') {
skipRemaining(run.id, 'Stopped before this move began.');
updateRun(run.id, 'stopped', 'Stopped. Completed moves are retained.'); return false;
}
if (!run.moves.some(move => move.state === 'planned')) {
const done = run.moves.filter(move => move.state === 'done').length;
const skipped = run.moves.filter(move => move.state === 'skipped').length;
updateRun(run.id, 'done', `Finished: ${done} moved${skipped ? `, ${skipped} skipped` : ''}.`); return false;
}
const latest = run.moves.filter(move => move.state === 'done').sort((a, b) => b.updatedAt.localeCompare(a.updatedAt))[0];
if (latest && !refreshedAfterMove(latest)) {
updateRun(run.id, 'running', 'Waiting for fresh source and destination measurements after the last move.'); return false;
}
const snapshot = rebalanceSnapshot(run.settings);
for (const move of run.moves.filter(move => move.state === 'planned')) {
const reason = rebalanceMoveBlocker(snapshot, move);
if (reason) { updateMove(move.id, 'skipped', reason); continue; }
// This transaction includes the durable move-to-operation link. A restarted
// worker sees either both records or neither and cannot repeat admission.
let operationId: string;
try { operationId = requestInstanceMove(move.server, move.targetHostId, { triggeredBy: `rebalance:${run.id}` }); }
catch (error) { updateMove(move.id, 'skipped', (error as Error).message); continue; }
db().prepare("UPDATE rebalancing_moves SET state='moving',operation_id=?,message='Waiting for safe migration admission.',updated_at=? WHERE id=?")
.run(operationId, atNow(), move.id);
updateRun(run.id, 'running', `Moving ${move.server} to ${move.targetHostName}.`);
return true;
}
const done = run.moves.filter(move => move.state === 'done').length;
updateRun(run.id, 'done', `Finished: ${done} moved. Remaining moves no longer qualify.`);
return false;
}
export async function runRebalancing(): Promise<boolean> {
return db().transaction(() => {
// Recovery is explicit in Provisioning. Reflect a recovered move, but
// never restart the other moves from the failed parent run.
const recovered = db().prepare(`SELECT m.id,m.run_id FROM rebalancing_moves m
JOIN rebalancing_runs r ON r.id=m.run_id JOIN server_deployments d ON d.id=m.operation_id
JOIN native_provisioning n ON n.deployment_id=d.id
WHERE r.state='failed' AND m.state='failed' AND d.state IN ('done','warning') AND n.verified_at IS NOT NULL`)
.all() as { id: string; run_id: string }[];
for (const move of recovered) {
updateMove(move.id, 'done', 'Recovered through Provisioning; healthy on the destination host.');
updateRun(move.run_id, 'stopped', 'The failed move was recovered. Remaining moves stay stopped.');
}
const row = db().prepare("SELECT * FROM rebalancing_runs WHERE state IN ('running','stopping') ORDER BY created_at LIMIT 1").get() as RunRow | undefined;
if (!row) return false;
return reconcileRun(present(row));
})();
}
export function pruneRebalancing(): void {
const now = atNow(), cutoff = new Date(Date.now() - 31 * 86_400_000).toISOString();
db().prepare('DELETE FROM rebalancing_previews WHERE expires_at<? AND run_id IS NULL').run(now);
db().prepare(`DELETE FROM rebalancing_runs WHERE state IN ('done','stopped') AND updated_at<?
AND NOT EXISTS (SELECT 1 FROM rebalancing_moves m JOIN server_deployments d ON d.id=m.operation_id
WHERE m.run_id=rebalancing_runs.id AND d.state NOT IN ('done','warning','cancelled'))`).run(cutoff);
}
src/lib/server/rebalancing/planner.ts
+96
View File
@@ -0,0 +1,96 @@
import type { RebalanceCandidate, RebalancePlan, RebalanceSettings } from '../../rebalancing.js';
/** Difference in the pair's peak dominant utilization, in percentage points. */
export const REBALANCE_MIN_IMPROVEMENT = 5;
export const REBALANCE_COOLDOWN_MS = 30 * 60_000;
export interface Resources { cpuCores: number; memoryBytes: number }
export interface RebalancingHost {
id: string; name: string; cpuCount: number; memoryTotal: number;
used: Resources; running: Resources; startup: Resources;
instances: number; ports: number; reason: string | null;
}
export interface RebalancingInstance {
server: string; instanceId: string; generation: number; sourceHostId: string;
game: string; relief: Resources; demand: Resources;
releasesPort?: boolean;
destinations: Record<string, string | null>;
}
export interface RebalancingSnapshot {
hosts: RebalancingHost[]; instances: RebalancingInstance[];
exclusions: RebalancePlan['exclusions']; warnings: string[];
}
const pressure = (host: RebalancingHost, used = host.used) => Math.max(used.cpuCores / host.cpuCount, used.memoryBytes / host.memoryTotal);
const add = (a: Resources, b: Resources): Resources => ({ cpuCores: a.cpuCores + b.cpuCores, memoryBytes: a.memoryBytes + b.memoryBytes });
const subtract = (a: Resources, b: Resources): Resources => ({ cpuCores: Math.max(0, a.cpuCores - b.cpuCores), memoryBytes: Math.max(0, a.memoryBytes - b.memoryBytes) });
const maximum = (a: Resources, b: Resources): Resources => ({ cpuCores: Math.max(a.cpuCores, b.cpuCores), memoryBytes: Math.max(a.memoryBytes, b.memoryBytes) });
function evaluate(source: RebalancingHost, target: RebalancingHost, instance: RebalancingInstance): { reason: string | null; improvement: number; demand: Resources } {
const demand = maximum(instance.demand, target.running);
const startup = add(target.used, maximum(demand, target.startup));
const improvement = 100 * (Math.max(pressure(source), pressure(target))
- Math.max(pressure(source, subtract(source.used, instance.relief)), pressure(target, add(target.used, demand))));
let reason: string | null = source.reason ?? target.reason;
if (!reason && source.id === target.id) reason = 'Source and destination must be different hosts.';
if (!reason && !(target.id in instance.destinations)) reason = 'Destination compatibility has not been verified.';
if (!reason) reason = instance.destinations[target.id] ?? null;
if (!reason && target.ports < 1) reason = 'No available UDP/TCP game port.';
if (!reason && startup.cpuCores > target.cpuCount * 0.95) reason = 'Destination has insufficient startup CPU capacity.';
if (!reason && startup.memoryBytes > target.memoryTotal - 500_000_000) reason = 'Destination has insufficient startup memory capacity.';
if (!reason && (!Number.isFinite(improvement) || improvement + 1e-8 < REBALANCE_MIN_IMPROVEMENT)) reason = `Move would not reduce the pair’s peak utilization by at least ${REBALANCE_MIN_IMPROVEMENT} percentage points.`;
return { reason, improvement, demand };
}
/** Validate the approved pair against fresh evidence without replacing it with a different move. */
export function rebalanceMoveBlocker(snapshot: RebalancingSnapshot, move: Pick<RebalanceCandidate, 'server' | 'instanceId' | 'generation' | 'sourceHostId' | 'targetHostId'>): string | null {
const instance = snapshot.instances.find(item => item.server === move.server && item.instanceId === move.instanceId
&& item.generation === move.generation && item.sourceHostId === move.sourceHostId);
if (!instance) return snapshot.exclusions.find(item => item.server === move.server)?.reason ?? 'The instance or its generation has changed.';
const source = snapshot.hosts.find(host => host.id === move.sourceHostId), target = snapshot.hosts.find(host => host.id === move.targetHostId);
if (!source || !target) return 'The selected source or destination is unavailable.';
return evaluate(source, target, instance).reason;
}
/** Greedy largest improvements minimize churn; simulated relief never changes live capacity. */
export function planRebalance(snapshot: RebalancingSnapshot, settings: RebalanceSettings): RebalancePlan {
const hosts = snapshot.hosts.filter(host => settings.hostIds.includes(host.id)).map(host => ({ ...host, used: { ...host.used } }));
const before = new Map(hosts.map(host => [host.id, { ...host, used: { ...host.used } }]));
const instances = snapshot.instances.filter(instance => settings.hostIds.includes(instance.sourceHostId)
&& (settings.game === 'all' || settings.game === instance.game));
const moved = new Set<string>(), moves: RebalanceCandidate[] = [];
const limit = Math.max(1, Math.min(20, Math.floor(settings.maxMoves) || 1));
while (moves.length < limit) {
let best: { instance: RebalancingInstance; source: RebalancingHost; target: RebalancingHost; improvement: number; demand: Resources } | undefined;
for (const instance of [...instances].sort((a, b) => a.server.localeCompare(b.server))) {
if (moved.has(instance.instanceId)) continue;
const source = hosts.find(host => host.id === instance.sourceHostId);
if (!source) continue;
for (const target of [...hosts].sort((a, b) => a.id.localeCompare(b.id))) {
const result = evaluate(source, target, instance);
if (!result.reason && (!best || result.improvement > best.improvement + 1e-8)) best = { instance, source, target, ...result };
}
}
if (!best) break;
const { instance, source, target, improvement, demand } = best;
moves.push({ server: instance.server, instanceId: instance.instanceId, generation: instance.generation,
sourceHostId: source.id, sourceHostName: source.name, targetHostId: target.id, targetHostName: target.name, improvement });
source.used = subtract(source.used, instance.relief); source.instances--;
if (instance.releasesPort) source.ports++;
target.used = add(target.used, demand); target.instances++; target.ports--;
moved.add(instance.instanceId);
}
const exclusions = [...snapshot.exclusions];
for (const instance of instances.filter(item => !moved.has(item.instanceId))) {
const source = hosts.find(host => host.id === instance.sourceHostId)!;
const results = hosts.filter(host => host.id !== source.id).map(target => ({ target, ...evaluate(source, target, instance) }));
const fits = results.some(result => !result.reason);
const reason = fits && moves.length === limit ? 'Move limit reached.' : results.length
? results.map(result => `${result.target.name}: ${result.reason ?? 'A different move provided more improvement.'}`).join(' ')
: 'No other participating host is available.';
exclusions.push({ server: instance.server, reason });
}
const projection = (host: RebalancingHost) => ({ cpuPercent: host.cpuCount > 0 ? host.used.cpuCores / host.cpuCount * 100 : 0,
memoryPercent: host.memoryTotal > 0 ? host.used.memoryBytes / host.memoryTotal * 100 : 0, instances: host.instances });
return { hosts: hosts.map(host => ({ id: host.id, name: host.name, before: projection(before.get(host.id)!), after: projection(host) })),
moves, exclusions, warnings: [...snapshot.warnings] };
}
src/lib/server/rebalancing/schema.ts
+29
View File
@@ -0,0 +1,29 @@
import type Database from 'better-sqlite3';
export function ensureRebalancingSchema(handle: Database.Database): void {
handle.exec(`
CREATE TABLE IF NOT EXISTS rebalancing_runs (
id TEXT PRIMARY KEY, settings TEXT NOT NULL,
state TEXT NOT NULL CHECK(state IN ('running','stopping','done','stopped','failed')),
created_at TEXT NOT NULL, updated_at TEXT NOT NULL, message TEXT NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS rebalancing_one_run ON rebalancing_runs((1))
WHERE state IN ('running','stopping');
CREATE TABLE IF NOT EXISTS rebalancing_moves (
id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES rebalancing_runs(id) ON DELETE CASCADE,
ordinal INTEGER NOT NULL, server TEXT NOT NULL, instance_id TEXT NOT NULL, generation INTEGER NOT NULL,
source_host_id TEXT NOT NULL, target_host_id TEXT NOT NULL,
candidate TEXT NOT NULL, state TEXT NOT NULL CHECK(state IN ('planned','moving','done','skipped','failed')),
operation_id TEXT UNIQUE REFERENCES server_deployments(id) ON DELETE SET NULL,
message TEXT NOT NULL, updated_at TEXT NOT NULL, UNIQUE(run_id,ordinal)
);
CREATE INDEX IF NOT EXISTS rebalancing_move_run ON rebalancing_moves(run_id,state,ordinal);
CREATE INDEX IF NOT EXISTS rebalancing_move_instance ON rebalancing_moves(instance_id,state,updated_at);
CREATE TABLE IF NOT EXISTS rebalancing_previews (
id TEXT PRIMARY KEY, preview TEXT NOT NULL, expires_at TEXT NOT NULL,
run_id TEXT REFERENCES rebalancing_runs(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS rebalancing_preview_expiry ON rebalancing_previews(expires_at);
CREATE INDEX IF NOT EXISTS rebalancing_preview_run ON rebalancing_previews(run_id);
`);
}
src/lib/server/rebalancing/snapshot.ts
+144
View File
@@ -0,0 +1,144 @@
import { db } from '../db/index.js';
import { listHosts, HEARTBEAT_MAX_AGE_MS, requireReadyHost } from '../hosts/index.js';
import { capacityCalibration } from '../hosts/calibration.js';
import { assertReadyLibrary } from '../game-files/index.js';
import { listServers } from '../fleet/servers.js';
import { getServerKey } from '../fleet/server-keys.js';
import { observationReader, serverObservation } from '../fleet/status.js';
import { latestGameSubmission, telemetryReport } from '../fleet/status-evidence.js';
import { mapIdentity, protectedWork } from '../autoscaling/snapshot.js';
import { captureMoveRecipe } from '../provisioning/move-recipes.js';
import { freshStatusTimestamp } from '../../server-freshness.js';
import type { Host } from '../../hosts.js';
import type { NativeInstance } from '../backends/native.js';
import type { RebalanceHostOption, RebalanceSettings } from '../../rebalancing.js';
import { REBALANCE_COOLDOWN_MS, type RebalancingSnapshot, type Resources } from './planner.js';
const HISTORY_MS = 5 * 60_000;
const nonnegative = (value: unknown): value is number => typeof value === 'number' && Number.isFinite(value) && value >= 0;
function hostBlocker(host: Host, now: number): string | null {
if (host.mode !== 'active' || host.decommission) return 'An active host is required.';
const stamp = Date.parse(host.lastHeartbeat ?? '');
if (!host.connected || !Number.isFinite(stamp) || stamp > now || now - stamp > HEARTBEAT_MAX_AGE_MS) return 'Waiting for a fresh host report.';
if (host.readiness !== 'ready') return host.readinessReasons.join(' ') || 'The host is not ready.';
const { cpuCount, cpuUsagePercent, memoryTotal, memoryAvailable } = host.inventory;
if (!nonnegative(cpuCount) || !cpuCount || !nonnegative(cpuUsagePercent) || cpuUsagePercent > 100
|| !nonnegative(memoryTotal) || !memoryTotal || !nonnegative(memoryAvailable) || memoryAvailable > memoryTotal) return 'Waiting for valid CPU and memory measurements.';
return null;
}
export function rebalanceHosts(): RebalanceHostOption[] {
const now = Date.now();
return listHosts().map(host => { const reason = hostBlocker(host, now); return { id: host.id, name: host.name, eligible: !reason, reason }; });
}
/** Database evidence only: no probes, writes, reservations, or source capacity credits. */
export function rebalanceSnapshot(settings: RebalanceSettings, exceptOperationId = ''): RebalancingSnapshot {
const handle = db(), now = Date.now(), at = new Date(now).toISOString(), since = new Date(now - HISTORY_MS).toISOString();
const hosts = listHosts().filter(host => settings.hostIds.includes(host.id));
const natives = handle.prepare('SELECT * FROM native_instances WHERE retired_at IS NULL').all() as NativeInstance[];
const operations = handle.prepare(`SELECT id,server,stage FROM server_deployments WHERE server_deleted_at IS NULL
AND state IN ('queued','waiting','running','awaiting-health','interrupted') AND id!=?`).all(exceptOperationId) as { id: string; server: string; stage: string }[];
const ownServer = exceptOperationId ? (handle.prepare('SELECT server FROM server_deployments WHERE id=?').get(exceptOperationId) as { server: string } | undefined)?.server : undefined;
const observe = observationReader();
const snapshot: RebalancingSnapshot = { hosts: [], instances: [], exclusions: [], warnings: [] };
for (const id of settings.hostIds) if (!hosts.some(host => host.id === id)) snapshot.warnings.push(`Participating host ${id} is unavailable.`);
for (const host of hosts) {
const reason = hostBlocker(host, now), { running, startup } = capacityCalibration(host.id);
if (reason) snapshot.warnings.push(`${host.name}: ${reason}`);
const inventory = host.inventory;
const valid = !reason || nonnegative(inventory.cpuCount) && inventory.cpuCount > 0 && nonnegative(inventory.cpuUsagePercent)
&& nonnegative(inventory.memoryTotal) && inventory.memoryTotal > 0 && nonnegative(inventory.memoryAvailable) && inventory.memoryAvailable <= inventory.memoryTotal;
const history = handle.prepare(`SELECT AVG(cpu_usage_percent) AS cpu,AVG(memory_available) AS memory FROM ec_host_resource_samples
WHERE host_id=? AND sampled_at>=? AND sampled_at<=? AND cpu_count=? AND memory_total=?`)
.get(host.id, since, at, inventory.cpuCount ?? 0, inventory.memoryTotal ?? 0) as { cpu: number | null; memory: number | null };
const used: Resources = { cpuCores: valid ? inventory.cpuCount! * Math.max(inventory.cpuUsagePercent!, history.cpu ?? 0) / 100 : 0,
memoryBytes: valid ? inventory.memoryTotal! - Math.min(inventory.memoryAvailable!, history.memory ?? inventory.memoryAvailable!) : 0 };
const hostInstances = natives.filter(instance => instance.host_id === host.id), measured = new Map(inventory.instanceResources?.map(sample => [sample.instanceId, sample]));
const reserved = new Set<string>();
const reserve = (id: string, allowance: Resources) => {
if (reserved.has(id)) return;
reserved.add(id); used.cpuCores += allowance.cpuCores; used.memoryBytes += allowance.memoryBytes;
};
for (const instance of hostInstances) {
const sample = measured.get(instance.id);
if (sample?.generation === instance.generation) continue; // Already in the host measurement.
const latest = handle.prepare(`SELECT d.state,d.submitted_at,n.operation_id,n.stage_operation_id FROM native_provisioning n
JOIN server_deployments d ON d.id=n.deployment_id WHERE n.instance_id=? AND n.generation=? AND d.id!=?
AND d.action IN ('create','rebuild') AND d.server_deleted_at IS NULL ORDER BY d.rowid DESC LIMIT 1`)
.get(instance.id, instance.generation, exceptOperationId) as { state: string; submitted_at: string | null; operation_id: string | null; stage_operation_id: string | null } | undefined;
const accounted = handle.prepare(`SELECT 1 FROM native_provisioning n JOIN server_deployments d ON d.id=n.deployment_id
WHERE n.instance_id=? AND n.generation=? AND d.action IN ('create','rebuild') AND d.state IN ('done','warning') AND d.finished_at<?`)
.get(instance.id, instance.generation, host.lastHeartbeat ?? '');
if (accounted) continue;
reserve(instance.id, latest?.state === 'queued' && !latest.submitted_at && !latest.operation_id && !latest.stage_operation_id ? running : startup);
}
const incoming = handle.prepare(`SELECT m.instance_id,m.removed_at,i.generation FROM ec_host_migrations m
JOIN server_deployments d ON d.id=m.deployment_id JOIN native_instances i ON i.id=m.instance_id
WHERE m.target_host_id=? AND m.deployment_id!=? AND d.server_deleted_at IS NULL AND i.retired_at IS NULL
AND d.state NOT IN ('done','warning','cancelled')`).all(host.id, exceptOperationId) as { instance_id: string; removed_at: string | null; generation: number }[];
for (const migration of incoming) {
const generation = migration.generation + (migration.removed_at ? 0 : 1);
if (measured.get(migration.instance_id)?.generation !== generation) reserve(migration.instance_id, startup);
}
const ports = handle.prepare('SELECT port,bind_address FROM ec_host_ports WHERE host_id=?').all(host.id) as { port: number; bind_address: string }[];
const usedPorts = new Set(ports.filter(port => port.port >= host.portMin && port.port <= host.portMax
&& (port.bind_address === host.bindAddress || ['0.0.0.0', '::'].includes(port.bind_address) || ['0.0.0.0', '::'].includes(host.bindAddress))).map(port => port.port));
snapshot.hosts.push({ id: host.id, name: host.name, cpuCount: valid ? inventory.cpuCount! : 0, memoryTotal: valid ? inventory.memoryTotal! : 0,
used, running, startup, reason, instances: hostInstances.length + incoming.filter(row => !hostInstances.some(instance => instance.id === row.instance_id)).length,
ports: Math.max(0, host.portMax - host.portMin + 1 - usedPorts.size) });
}
const destinations = new Map<string, Record<string, string | null>>();
for (const row of listServers()) {
const identity = mapIdentity(row);
if (!row.host_id || !settings.hostIds.includes(row.host_id) || settings.game !== 'all' && identity?.game !== settings.game) continue;
const source = hosts.find(host => host.id === row.host_id), native = natives.find(instance => instance.id === row.native_instance_id);
// The admitted rebuild itself must not hide the pre-removal empty evidence.
const observation = row.name === ownServer ? serverObservation(row, { provisioning: false, settling: false,
telemetry: telemetryReport(row.name), submission: latestGameSubmission(row.name) }) : observe(row);
let reason = protectedWork(row.name, exceptOperationId);
if (!reason && (!identity || !native || !row.map_slug)) reason = 'A native instance with a known map is required.';
if (!reason && (!source || hostBlocker(source, now))) reason = source ? hostBlocker(source, now) : 'Source host is unavailable.';
const operation = operations.find(item => item.server === row.name);
if (!reason && operation) reason = `Provisioning ${operation.id}: ${operation.stage}`;
if (!reason && native && handle.prepare(`SELECT 1 FROM ec_host_operations WHERE instance_id=? AND state IN ('pending','accepted','running','unknown')
AND kind IN ('instance.create','instance.rebuild','instance.start','instance.stop','instance.restart','instance.remove','instance.cancel-create')`).get(native.id)) reason = 'An instance lifecycle command is unsettled.';
if (!reason && (row.desired_state === 'stopped' || ['stopped','exited','created','dead','missing'].includes(row.container_state ?? ''))) reason = 'Stopped instances are not eligible.';
if (!reason && (row.last_players ?? 0) > 0 && freshStatusTimestamp(row.last_seen_at, now)
&& (!observation.telemetry_after || Date.parse(row.last_seen_at!) > Date.parse(observation.telemetry_after))) reason = 'Players are present according to a current RCON report.';
if (!reason && (observation.health !== 'online' || observation.last_players === null || !freshStatusTimestamp(observation.last_seen_at, now))) reason = 'Waiting for healthy, fresh occupancy evidence.';
if (!reason && observation.last_players !== 0) reason = 'Players are present.';
if (!reason && !getServerKey(row.name)) reason = 'Save a server key before moving this instance.';
if (!reason && handle.prepare(`SELECT 1 FROM rebalancing_moves WHERE instance_id=? AND state='done' AND updated_at>=?`)
.get(native!.id, new Date(now - REBALANCE_COOLDOWN_MS).toISOString())) reason = 'This instance was rebalanced less than 30 minutes ago.';
const sample = source?.inventory.instanceResources?.find(item => item.instanceId === native?.id && item.generation === native.generation);
if (!reason && (!sample || source?.inventory.instanceResourcesError || !nonnegative(sample.cpuCores) || !nonnegative(sample.memoryBytes) || !sample.memoryBytes)) reason = 'Waiting for resource measurements for the current instance generation.';
if (!reason && sample!.uptimeSeconds < 300) reason = 'Waiting for this instance to finish its first five minutes of startup.';
if (!reason) { try { captureMoveRecipe(row.name); } catch (error) { reason = (error as Error).message; } }
if (reason) { snapshot.exclusions.push({ server: row.name, reason }); continue; }
const key = JSON.stringify([native!.library_id, native!.image]);
if (!destinations.has(key)) {
const eligibility: Record<string, string | null> = {};
for (const host of hosts) {
try { requireReadyHost(host.id, { image: native!.image, requiredBytes: 1024 ** 3 }); assertReadyLibrary(host.id, native!.library_id); eligibility[host.id] = null; }
catch (error) { eligibility[host.id] = (error as Error).message; }
}
destinations.set(key, eligibility);
}
const history = handle.prepare(`SELECT AVG(cpu_cores) AS cpu,AVG(memory_bytes) AS memory FROM ec_instance_resource_samples
WHERE host_id=? AND instance_id=? AND generation=? AND phase='running' AND bucket>=? AND bucket<=?`)
.get(source!.id, native!.id, native!.generation, since, at) as { cpu: number | null; memory: number | null };
const bindConflicts = (bind: string) => bind === source!.bindAddress || ['0.0.0.0', '::'].includes(bind) || ['0.0.0.0', '::'].includes(source!.bindAddress);
const portOwners = handle.prepare('SELECT instance_id,bind_address FROM ec_host_ports WHERE host_id=? AND port=?')
.all(source!.id, native!.port) as { instance_id: string; bind_address: string }[];
snapshot.instances.push({ server: row.name, instanceId: native!.id, generation: native!.generation, sourceHostId: source!.id, game: identity!.game,
releasesPort: native!.port >= source!.portMin && native!.port <= source!.portMax && bindConflicts(native!.bind_address)
&& portOwners.some(owner => owner.instance_id === native!.id && bindConflicts(owner.bind_address))
&& !portOwners.some(owner => owner.instance_id !== native!.id && bindConflicts(owner.bind_address)),
relief: { cpuCores: Math.min(sample!.cpuCores, history.cpu ?? sample!.cpuCores), memoryBytes: Math.min(sample!.memoryBytes, history.memory ?? sample!.memoryBytes) },
demand: { cpuCores: Math.max(sample!.cpuCores, history.cpu ?? sample!.cpuCores), memoryBytes: Math.max(sample!.memoryBytes, history.memory ?? sample!.memoryBytes) },
destinations: destinations.get(key)! });
}
return snapshot;
}