184 lines
14 KiB
TypeScript
184 lines
14 KiB
TypeScript
import { concurrencyBlocker, preparingJobs, committedRebalanceStartup } from './concurrency.js';
|
|
import { db } from '../db/index.js';
|
|
import { provisioningCancelled } from './cancellation.js';
|
|
import { nativeInstance, nativeRequest, runtimePasswordForServer, type NativeInstance } from '../backends/native.js';
|
|
import { getHost, getHostOperationStatus, queueHostOperation, reserveHostPort } from '../hosts/index.js';
|
|
import { choosePlacement } from '../hosts/capacity.js';
|
|
import { scheduleAvailabilityRefresh } from '../hosts/availability.js';
|
|
import { activeBetaRemovalBlocker, getServer } from '../fleet/servers.js';
|
|
import { getServerKey } from '../fleet/server-keys.js';
|
|
import { recipeForDeployment, recipeStorage, recipeSummaryForDeployment, type LaunchRecipePlan } from '../fleet/launch-recipes.js';
|
|
import { captureMoveRecipe, moveRecipeMatches, moveRecipeStorage } from './move-recipes.js';
|
|
import { rebalanceAdmission } from '../rebalancing/admission.js';
|
|
import { probePlayerEvidence } from '../fleet/status-evidence.js';
|
|
import { beginDeployment, deploymentStage } from './history.js';
|
|
import { fleetSettings } from '../settings.js';
|
|
import type { NativePlacement } from '../../hosts.js';
|
|
|
|
interface Migration {
|
|
deployment_id: string; source_host_id: string; instance_id: string;
|
|
target_host_id: string | null; target_port: number | null; target_bind: string | null;
|
|
remove_operation_id: string | null; removed_at: string | null;
|
|
}
|
|
interface Job { id: string; server: string; action: string; state: string; stage: string }
|
|
type Recipe = LaunchRecipePlan & { native: NativePlacement };
|
|
export { captureMoveRecipe } from './move-recipes.js';
|
|
|
|
export function isRebalanceMigration(id: string): boolean {
|
|
return !!db().prepare("SELECT 1 FROM server_deployments WHERE id=? AND triggered_by LIKE 'rebalance:%'").get(id);
|
|
}
|
|
|
|
/** Queue a move to one explicit host; its source remains active throughout. */
|
|
export function requestInstanceMove(serverName: string, targetHostId: string,
|
|
options: { triggeredBy: string; recipe?: LaunchRecipePlan }): string {
|
|
return db().transaction(() => {
|
|
const instance = nativeInstance(serverName);
|
|
const server = getServer(serverName);
|
|
if (!instance || !server?.map_slug) throw new Error('The native instance is unavailable.');
|
|
if (instance.host_id === targetHostId) throw new Error('Choose a different destination host.');
|
|
if (getHost(instance.host_id)?.mode !== 'active') throw new Error('The source host must be active.');
|
|
if (!options.triggeredBy.startsWith('rebalance:')) throw new Error('An instance move requires a rebalance run.');
|
|
const blocker = activeBetaRemovalBlocker(serverName);
|
|
if (blocker) throw new Error(blocker);
|
|
if (!getServerKey(serverName)) throw new Error('Save a server key before moving this instance.');
|
|
if (db().prepare(`SELECT 1 FROM server_deployments WHERE server=? AND server_deleted_at IS NULL
|
|
AND state IN ('queued','waiting','running','awaiting-health','interrupted')`).get(serverName)) {
|
|
throw new Error('Another provisioning operation is active for this instance.');
|
|
}
|
|
const recipe = options.recipe ?? captureMoveRecipe(serverName);
|
|
if (recipe.game !== server.game || recipe.mapSlug !== server.map_slug || recipe.image !== instance.image
|
|
|| !moveRecipeMatches(serverName, recipe)) throw new Error('The deployed files changed; generate a fresh rebalance preview.');
|
|
const password = runtimePasswordForServer(serverName);
|
|
if (!password) throw new Error('The deployed runtime credentials are unavailable.');
|
|
choosePlacement({ hostId: targetHostId, excludeHostIds: [instance.host_id], libraryId: instance.library_id,
|
|
image: instance.image, thorough: true, dispatch: true });
|
|
const id = beginDeployment(serverName, 'rebuild', options.triggeredBy);
|
|
db().prepare("UPDATE server_deployments SET state='waiting' WHERE id=?").run(id);
|
|
db().prepare('INSERT INTO server_creation_jobs(deployment_id,server,port,game_mode,map_slug,expected_map) VALUES(?,?,NULL,?,?,?)')
|
|
.run(id, serverName, 'zombie', recipe.mapSlug, recipe.expectedMap);
|
|
const stored = moveRecipeStorage({ ...recipe, native: { libraryId: instance.library_id, image: instance.image, hostId: targetHostId } });
|
|
db().prepare('INSERT INTO server_launch_recipes(deployment_id,recipe,recipe_summary) VALUES(?,?,?)').run(id, stored.recipe, stored.recipeSummary);
|
|
db().prepare('INSERT INTO ec_host_migrations(deployment_id,source_host_id,instance_id,runtime_password) VALUES(?,?,?,?)')
|
|
.run(id, instance.host_id, instance.id, password);
|
|
deploymentStage(id, 'Rebalance queued; waiting to confirm an empty lobby and destination capacity', { phase: 'eligibility', recipeStatus: 'queued' });
|
|
return id;
|
|
})();
|
|
}
|
|
|
|
export function hostMigration(id: string): Migration | undefined {
|
|
return db().prepare('SELECT * FROM ec_host_migrations WHERE deployment_id=?').get(id) as Migration | undefined;
|
|
}
|
|
|
|
/** Reserve a destination, remove the source once, then hand creation to the normal health-verified rebuild queue. */
|
|
export async function processMigrationRemoval(job: Job, instance: NativeInstance, migration: Migration,
|
|
eligibility: () => string | null): Promise<void> {
|
|
if (provisioningCancelled(job.id)) return;
|
|
const hold = (message: string) => { if (!provisioningCancelled(job.id)) deploymentStage(job.id, message, { blocker: { code: 'native-host', message, automatic: true } }); };
|
|
if (migration.remove_operation_id) {
|
|
const operation = getHostOperationStatus(migration.remove_operation_id);
|
|
if (operation?.state === 'failed' && Date.now() - Date.parse(operation.updatedAt) >= 60_000) {
|
|
// A failed receipt certifies no mutation began. Unknown results keep
|
|
// their reservation and original operation ID until reconciled.
|
|
db().transaction(() => {
|
|
db().prepare('DELETE FROM ec_host_ports WHERE host_id=? AND instance_id=?').run(migration.target_host_id, instance.id);
|
|
db().prepare(`UPDATE ec_host_migrations SET target_host_id=NULL,target_port=NULL,target_bind=NULL,
|
|
remove_operation_id=NULL WHERE deployment_id=?`).run(job.id);
|
|
db().prepare("UPDATE server_deployments SET state='waiting',submitted_at=NULL WHERE id=?").run(job.id);
|
|
scheduleAvailabilityRefresh();
|
|
})();
|
|
hold('Source removal failed before mutation; checking destination capacity again before retrying.');
|
|
return;
|
|
}
|
|
if (operation?.state !== 'completed') {
|
|
hold(operation?.error ?? 'Waiting for confirmed source removal. The original agent operation will reconcile.');
|
|
return;
|
|
}
|
|
db().transaction(() => {
|
|
if (hostMigration(job.id)?.removed_at) return;
|
|
const current = nativeInstance(job.server);
|
|
if (current?.id !== instance.id || current.generation !== instance.generation || current.host_id !== migration.source_host_id) throw new Error('Migration source changed.');
|
|
const target = getHost(migration.target_host_id!);
|
|
if (!target) throw new Error('Migration destination is unavailable.');
|
|
const rebalance = isRebalanceMigration(job.id);
|
|
const recipe = (rebalance ? { format: 1, ...recipeSummaryForDeployment(job.id)!.recipe }
|
|
: recipeForDeployment(job.id)!.recipe) as Recipe;
|
|
recipe.native = { libraryId: instance.library_id, image: instance.image, hostId: target.id,
|
|
bindAddress: migration.target_bind!, publicAddress: target.publicAddress };
|
|
const stored = rebalance ? moveRecipeStorage(recipe) : recipeStorage(recipe);
|
|
db().prepare('UPDATE server_launch_recipes SET recipe=?,recipe_summary=? WHERE deployment_id=?').run(stored.recipe, stored.recipeSummary, job.id);
|
|
db().prepare('DELETE FROM ec_host_ports WHERE host_id=? AND instance_id=?').run(instance.host_id, instance.id);
|
|
db().prepare(`UPDATE native_instances SET host_id=?,generation=generation+1,bind_address=?,port=?,public_address=?,public_port=NULL WHERE id=?`)
|
|
.run(target.id, migration.target_bind, migration.target_port, target.publicAddress, instance.id);
|
|
db().prepare('UPDATE native_provisioning SET generation=generation+1 WHERE deployment_id=?').run(job.id);
|
|
db().prepare('UPDATE server_creation_jobs SET port=? WHERE deployment_id=?').run(migration.target_port, job.id);
|
|
db().prepare(`UPDATE servers SET game_port=?,container_state='starting',container_checked_at=NULL,
|
|
last_players=NULL,last_roster=NULL,last_seen_at=NULL,last_error=NULL,rcon_failures=0,active_since=NULL,inactive_since=NULL,
|
|
game_started_at=NULL,game_ended_at=NULL WHERE name=?`).run(migration.target_port, job.server);
|
|
db().prepare('UPDATE ec_host_migrations SET removed_at=? WHERE deployment_id=?').run(operation.updatedAt, job.id);
|
|
db().prepare("UPDATE server_deployments SET state='waiting',submitted_at=NULL WHERE id=?").run(job.id);
|
|
deploymentStage(job.id, `Source removed; rebuilding on ${target.name}`, { phase: 'startup-queue', blocker: null });
|
|
scheduleAvailabilityRefresh();
|
|
})();
|
|
return;
|
|
}
|
|
if (!getHost(instance.host_id)?.connected) { hold('Waiting for the source host to reconnect.'); return; }
|
|
const reason = rebalanceAdmission(job.id) ?? eligibility() ?? activeBetaRemovalBlocker(job.server);
|
|
if (reason) { hold(reason); return; }
|
|
if (!getServerKey(job.server)) { hold('Save a server key before migrating this instance.'); return; }
|
|
const rebalance = isRebalanceMigration(job.id);
|
|
const recipe = rebalance ? recipeSummaryForDeployment(job.id)!.recipe : null;
|
|
if (rebalance && !moveRecipeMatches(job.server, { format: 1, ...recipe! })) {
|
|
hold('The deployed files changed; stop this run and generate a fresh rebalance preview.'); return;
|
|
}
|
|
const options = { libraryId: instance.library_id, image: instance.image, thorough: true, dispatch: true,
|
|
excludeHostIds: [instance.host_id], ...(rebalance ? { hostId: recipe!.native!.hostId } : {}) };
|
|
// No source mutation is submitted unless another host can take this instance.
|
|
choosePlacement(options);
|
|
const health = await nativeRequest<{ state: string; healthy?: boolean }>(instance, 'instance.inspect');
|
|
if (provisioningCancelled(job.id)) return;
|
|
if (rebalance && (health.state !== 'running' || health.healthy !== true)) {
|
|
hold('The source lobby is no longer healthy and running. Waiting for current health evidence.'); return;
|
|
}
|
|
if (!['stopped','exited','created','dead','missing'].includes(health.state)) {
|
|
const occupancy = await probePlayerEvidence(job.server);
|
|
if (provisioningCancelled(job.id)) return;
|
|
if (occupancy.players !== 0) { hold(occupancy.players === null ? 'Waiting for a fresh local player count.' : `Waiting for ${occupancy.players} player(s) to leave.`); return; }
|
|
}
|
|
db().transaction(() => {
|
|
const state = db().prepare('SELECT state FROM server_deployments WHERE id=? AND server_deleted_at IS NULL').get(job.id) as { state: string } | undefined;
|
|
if (!state || !['waiting','queued','interrupted'].includes(state.state) || hostMigration(job.id)?.remove_operation_id) return;
|
|
const current = nativeInstance(job.server);
|
|
if (current?.id !== instance.id || current.generation !== instance.generation
|
|
|| getHost(instance.host_id)?.mode !== (rebalance ? 'active' : 'draining')) return;
|
|
const blocker = rebalanceAdmission(job.id) ?? eligibility() ?? activeBetaRemovalBlocker(job.server);
|
|
if (blocker) { hold(blocker); return; }
|
|
if (rebalance && !moveRecipeMatches(job.server, { format: 1, ...recipe! })) {
|
|
hold('The deployed files changed; stop this run and generate a fresh rebalance preview.'); return;
|
|
}
|
|
if (rebalance) {
|
|
const startups = db().prepare(`SELECT COUNT(*) AS n FROM (
|
|
SELECT id FROM server_deployments WHERE id!=? AND server_deleted_at IS NULL
|
|
AND action IN ('create','rebuild') AND (state IN ('running','awaiting-health') OR ${committedRebalanceStartup})
|
|
UNION SELECT value FROM json_each(?) WHERE value!=?)`).get(job.id, JSON.stringify([...preparingJobs]), job.id) as { n: number };
|
|
if (startups.n >= fleetSettings().createConcurrency) { hold('Waiting for startup capacity before moving this lobby.'); return; }
|
|
}
|
|
const removing = db().prepare("SELECT COUNT(*) AS n FROM ec_host_operations WHERE kind='instance.remove' AND state IN ('pending','accepted','running','unknown')").get() as { n: number };
|
|
if (removing.n >= fleetSettings().destroyConcurrency) { hold('Waiting for destruction capacity.'); return; }
|
|
const { host, port } = choosePlacement(options);
|
|
const capacityBlocker = concurrencyBlocker(job.id, 'rebuild', instance.host_id, true) ?? concurrencyBlocker(job.id, 'rebuild', host.id);
|
|
if (capacityBlocker) { hold(capacityBlocker); return; }
|
|
reserveHostPort(host.id, instance.id, { port, bindAddress: host.bindAddress, protocol: 'udp' });
|
|
reserveHostPort(host.id, instance.id, { port, bindAddress: host.bindAddress, protocol: 'tcp' });
|
|
const attempt = (db().prepare(`SELECT COUNT(*) AS n FROM ec_host_operations
|
|
WHERE host_id=? AND instance_id=? AND kind='instance.remove' AND idempotency_key LIKE ?`)
|
|
.get(instance.host_id, instance.id, `${job.id}:migrate-remove:%`) as { n: number }).n;
|
|
const operation = queueHostOperation(instance.host_id, 'instance.remove', {}, {
|
|
instanceId: instance.id, generation: instance.generation, idempotencyKey: `${job.id}:migrate-remove:${attempt}`
|
|
});
|
|
db().prepare('UPDATE ec_host_migrations SET target_host_id=?,target_port=?,target_bind=?,remove_operation_id=? WHERE deployment_id=?')
|
|
.run(host.id, port, host.bindAddress, operation.id, job.id);
|
|
db().prepare("UPDATE server_deployments SET state='running',submitted_at=? WHERE id=?").run(new Date().toISOString(), job.id);
|
|
deploymentStage(job.id, `Capacity reserved on ${host.name}; waiting for source removal`, { phase: 'container-action', blocker: null });
|
|
})();
|
|
}
|