2284 lines
95 KiB
TypeScript
2284 lines
95 KiB
TypeScript
import { assertPublicFileTarget, isPrivateConfigPath } from '../fleet/private-config.js';
|
||
import { assertScalingOwnership } from '../autoscaling/ownership.js';
|
||
import { removalAuthorization } from '../autoscaling/authorization.js';
|
||
import { scalingIntent } from '../autoscaling/admission.js';
|
||
import { putFileBlob, getFileBlob, blobSize } from '../db/file-blobs.js';
|
||
import {
|
||
concurrencyBlocker,
|
||
preparingJobs,
|
||
committedRebalanceStartup,
|
||
lifecycleHolders
|
||
} from './concurrency.js';
|
||
import { isIP } from 'node:net';
|
||
import { setImmediate as yieldToController } from 'node:timers/promises';
|
||
import { choosePlacement, availableHostPort, type PlacementOptions } from '../hosts/capacity.js';
|
||
import { randomBytes, randomUUID } from 'node:crypto';
|
||
import { bankCredential } from '../auth/bank.js';
|
||
import { posix } from 'node:path';
|
||
import { db } from '../db/index.js';
|
||
import { unsettledCancelledCommand } from './cancellation.js';
|
||
import { config, relDestForKind, serverDir } from '../config.js';
|
||
import {
|
||
runtimePasswordForServer,
|
||
nativeInstance,
|
||
nativeRequest,
|
||
unplacedNativeCreation,
|
||
type NativeInstance
|
||
} from '../backends/native.js';
|
||
import {
|
||
getHost,
|
||
listHosts,
|
||
getHostOperationStatus,
|
||
queueHostOperation,
|
||
queueHostOperationSerialized,
|
||
releaseHostPorts,
|
||
reserveHostPort,
|
||
type HostOperationStatus
|
||
} from '../hosts/index.js';
|
||
import { newestReadyLibraryId, assertReadyLibrary } from '../game-files/index.js';
|
||
import {
|
||
blueprintRecipe,
|
||
captureRecipePlan,
|
||
recipeForDeployment,
|
||
recipeSummaryForDeployment,
|
||
recipeStorage,
|
||
summarizeRecipe,
|
||
type LaunchRecipePlan
|
||
} from '../fleet/launch-recipes.js';
|
||
import {
|
||
prepareNativeLaunch,
|
||
evaluateNativePlacement,
|
||
type PreparedNativeLaunch
|
||
} from './launch-preparation.js';
|
||
import {
|
||
getServerKey,
|
||
reserveServerKey,
|
||
releaseServerKey,
|
||
availableServerKeyCount
|
||
} from '../fleet/server-keys.js';
|
||
import { activeBetaRemovalBlocker, getServer } from '../fleet/servers.js';
|
||
import { applyActivationFiles } from './activation-files.js';
|
||
import { nextLobbyNumber, type CreateServerInput, type CreateServerResult } from './instances.js';
|
||
import {
|
||
beginDeployment,
|
||
deploymentStage,
|
||
finishDeployment,
|
||
REMOVING_SERVER,
|
||
cancelRemoval
|
||
} from './history.js';
|
||
import { recordProgress, type CheckInput } from './progress.js';
|
||
import { trackActivity } from '../pipeline/activity.js';
|
||
import { fleetSettings } from '../settings.js';
|
||
import { resolveRecoveredHealthOperation } from '../maintenance/recovery.js';
|
||
import { probeServerStatus, probePlayerEvidence } from '../fleet/status-evidence.js';
|
||
import { monitoringIdentity } from '../fleet/health-observations.js';
|
||
import { handoffVerifiedHealth } from './health-handoff.js';
|
||
import { checkRelease, freshRelease, probeVersion } from '../integrations/plutonium.js';
|
||
import { recordScript, removeScript } from '../fleet/inventory.js';
|
||
import { sha256Buffer } from '../fleet/files.js';
|
||
import { captureBetaBeforeImage, type StagedFileChange } from '../fleet/staged-files.js';
|
||
import type { LaunchPreview } from '../../deployment-plan.js';
|
||
import type { ProvisioningAction } from '../../provisioning.js';
|
||
import { hostAgentUpdatesEnabled } from '../../hosts.js';
|
||
import { cancelCreation, finishCreationCancellation } from './cancel-creation.js';
|
||
import { reconcileHostDrains } from './host-drain.js';
|
||
import { hostMigration, isRebalanceMigration, processMigrationRemoval } from './migrations.js';
|
||
import { safetyOverride, safetyOverrideCapability } from './safety-override.js';
|
||
import { runtimeLaunchSql } from '../fleet/runtime-config.js';
|
||
|
||
export type NativeSelection = import('../../hosts.js').NativePlacement;
|
||
type NativeRecipe = LaunchRecipePlan & { native?: NativeSelection };
|
||
interface Job {
|
||
selection?: string | null;
|
||
id: string;
|
||
server: string;
|
||
action: ProvisioningAction;
|
||
state: string;
|
||
submitted_at: string | null;
|
||
stage: string;
|
||
instance_id: string;
|
||
generation: number;
|
||
stage_operation_id: string | null;
|
||
operation_id: string | null;
|
||
applied_at: string | null;
|
||
recovery_step: 'restart' | 'rebuild' | null;
|
||
}
|
||
interface FilePlan {
|
||
id: number;
|
||
path: string;
|
||
sha256: string | null;
|
||
checkBefore: boolean;
|
||
expectedBeforeSha: string | null;
|
||
}
|
||
const unhealthyRecoveryCapability = 'instance.unhealthy-recovery';
|
||
|
||
/** Validate a launch request without selecting a host or consuming capacity. */
|
||
export function nativeLaunchSelection(
|
||
input: Record<string, unknown>,
|
||
game: string
|
||
): NativeSelection {
|
||
if (input.hostId !== undefined && typeof input.hostId !== 'string')
|
||
throw new Error('Invalid host.');
|
||
const libraryId = newestReadyLibraryId(game);
|
||
if (!libraryId)
|
||
throw new Error(`Upload ${game.toUpperCase()} game files before launching a server.`);
|
||
if (input.libraryId !== undefined && input.libraryId !== '' && input.libraryId !== libraryId) {
|
||
throw new Error(
|
||
`The current ${game.toUpperCase()} game files changed. Refresh the launch preview.`
|
||
);
|
||
}
|
||
const image =
|
||
input.image === undefined || input.image === '' ? config.defaultRuntimeImage : input.image;
|
||
if (typeof image !== 'string' || !/^[a-zA-Z0-9][a-zA-Z0-9._/:@-]{0,499}$/.test(image))
|
||
throw new Error('Invalid runtime image.');
|
||
const automatic = input.automatic === true || !input.hostId;
|
||
if (!automatic) {
|
||
const host = getHost(input.hostId as string);
|
||
if (!host) throw new Error('The selected host was not found.');
|
||
if (host.decommission)
|
||
throw new Error('The selected host is being removed. Choose another host.');
|
||
}
|
||
// Only the controller-configured default may be prepared automatically.
|
||
// Custom runtimes must be reported as admin-prepared by the host agent; a
|
||
// launch request must never authorize a registry download.
|
||
if (
|
||
image !== config.defaultRuntimeImage &&
|
||
!listHosts().some(
|
||
(host) =>
|
||
!host.decommission &&
|
||
(automatic ? host.mode === 'active' : host.id === input.hostId) &&
|
||
host.inventory.preparedRuntimeImages?.includes(image)
|
||
)
|
||
) {
|
||
throw new Error(
|
||
'Custom runtime images must be prepared by an admin from the host page before launching.'
|
||
);
|
||
}
|
||
const rawNetwork = input.automatic === true ? (input.network ?? {}) : input;
|
||
if (!rawNetwork || typeof rawNetwork !== 'object' || Array.isArray(rawNetwork))
|
||
throw new Error('Invalid network overrides.');
|
||
const network = rawNetwork as NativeSelection;
|
||
const bindAddress = network.bindAddress || undefined;
|
||
if (bindAddress !== undefined && (typeof bindAddress !== 'string' || !isIP(bindAddress)))
|
||
throw new Error('Invalid bind address.');
|
||
if (
|
||
network.publicAddress !== undefined &&
|
||
(typeof network.publicAddress !== 'string' || network.publicAddress.length > 253)
|
||
)
|
||
throw new Error('Invalid public address.');
|
||
if (
|
||
network.publicPort !== undefined &&
|
||
(!Number.isInteger(network.publicPort) ||
|
||
Number(network.publicPort) < 1 ||
|
||
Number(network.publicPort) > 65535)
|
||
)
|
||
throw new Error('Invalid public port.');
|
||
return {
|
||
hostId: automatic ? undefined : (input.hostId as string),
|
||
libraryId,
|
||
image,
|
||
automatic,
|
||
bindAddress,
|
||
publicAddress: network.publicAddress,
|
||
publicPort: network.publicPort,
|
||
network: {
|
||
bindAddress,
|
||
publicAddress: network.publicAddress,
|
||
publicPort: network.publicPort
|
||
}
|
||
};
|
||
}
|
||
|
||
export function nativeSelection(
|
||
input: Record<string, unknown>,
|
||
game: string,
|
||
options: Pick<PlacementOptions, 'planned'> = {}
|
||
): NativeSelection {
|
||
const selection = nativeLaunchSelection(input, game);
|
||
const { host } = choosePlacement({
|
||
hostId: selection.hostId,
|
||
libraryId: selection.libraryId,
|
||
image: selection.image!,
|
||
bindAddress: selection.bindAddress,
|
||
...options
|
||
});
|
||
return {
|
||
...selection,
|
||
hostId: host.id,
|
||
bindAddress: selection.bindAddress ?? host.bindAddress,
|
||
publicAddress: selection.publicAddress ?? host.publicAddress
|
||
};
|
||
}
|
||
|
||
export function previewNativeCreation(input: Record<string, unknown>): LaunchPreview {
|
||
let recipe: NativeRecipe;
|
||
let blueprintName: string | null = null;
|
||
if (typeof input.deploymentId === 'string') {
|
||
const previous = recipeForDeployment(input.deploymentId);
|
||
if (!previous) throw new Error('The recorded deployment is unavailable.');
|
||
recipe = captureRecipePlan(previous.recipe.game, previous.recipe.mapSlug);
|
||
blueprintName = previous.blueprintName;
|
||
} else if (typeof input.blueprintId === 'string')
|
||
({ recipe, blueprintName } = blueprintRecipe(input.blueprintId));
|
||
else {
|
||
if ((input.game !== 't5' && input.game !== 't6') || typeof input.mapSlug !== 'string')
|
||
throw new Error('Choose a supported game and map.');
|
||
recipe = captureRecipePlan(input.game, input.mapSlug);
|
||
}
|
||
const selection = nativeSelection(input, recipe.game);
|
||
recipe.native = selection;
|
||
recipe.image = selection.image!;
|
||
const base = `${recipe.game}_zombie_${recipe.mapSlug}`;
|
||
const lobby = nextLobbyNumber(base);
|
||
const name = lobby === null ? null : `${base}_${lobby}`;
|
||
const port = nextNativePort(selection);
|
||
const id = randomUUID();
|
||
const expiresAt = new Date(Date.now() + 30 * 60_000).toISOString();
|
||
db()
|
||
.prepare(
|
||
'INSERT INTO server_launch_previews(id,expires_at,server,port,recipe,blueprint_name) VALUES(?,?,?,?,?,?)'
|
||
)
|
||
.run(id, expiresAt, name, port, JSON.stringify(recipe), blueprintName);
|
||
const checks: LaunchPreview['checks'] = [
|
||
{
|
||
label: 'Host capacity',
|
||
state: 'pass',
|
||
detail: selection.automatic
|
||
? 'Capacity available. The least loaded eligible host is selected when provisioning starts.'
|
||
: `${getHost(selection.hostId!)!.name}: capacity available.`
|
||
},
|
||
{
|
||
label: 'Game files',
|
||
state: 'pass',
|
||
detail: `The latest ${recipe.game.toUpperCase()} game files are verified on an eligible host.`
|
||
},
|
||
{
|
||
label: 'Lobby',
|
||
state: name ? 'pass' : 'blocked',
|
||
detail: name ?? 'All lobby IDs for this map are reserved.'
|
||
},
|
||
{
|
||
label: 'Host port',
|
||
state: port ? 'pass' : 'blocked',
|
||
detail: port
|
||
? selection.automatic
|
||
? 'UDP/TCP capacity available; port assigned at placement.'
|
||
: `${selection.bindAddress}:${port} / UDP and TCP`
|
||
: 'No available UDP/TCP port on this host.'
|
||
},
|
||
{ label: 'Configuration', state: 'pass', detail: recipe.configSource }
|
||
];
|
||
return {
|
||
id,
|
||
expiresAt,
|
||
name,
|
||
port,
|
||
hasSavedKey: !!name && availableServerKeyCount(recipe.game) > 0,
|
||
recipe: summarizeRecipe(recipe),
|
||
blueprintName,
|
||
checks,
|
||
ready: checks.every((check) => check.state !== 'blocked')
|
||
};
|
||
}
|
||
|
||
export function nextNativePort(selection: NativeSelection, taken: number[] = []): number | null {
|
||
const host = selection.hostId ? getHost(selection.hostId) : null;
|
||
return host
|
||
? availableHostPort(host, {
|
||
bindAddress: selection.bindAddress,
|
||
planned: taken.map((port) => ({ hostId: host.id, port }))
|
||
})
|
||
: null;
|
||
}
|
||
|
||
export function queueNativeCreation(input: CreateServerInput): CreateServerResult {
|
||
try {
|
||
return db().transaction(() => {
|
||
const preview = input.previewId
|
||
? (db()
|
||
.prepare('SELECT * FROM server_launch_previews WHERE id=?')
|
||
.get(input.previewId) as
|
||
| {
|
||
id: string;
|
||
recipe: string;
|
||
server: string;
|
||
port: number | null;
|
||
expires_at: string;
|
||
deployment_id: string | null;
|
||
blueprint_name: string | null;
|
||
}
|
||
| undefined)
|
||
: undefined;
|
||
if (input.previewId && !preview) throw new Error('Review a fresh launch preview.');
|
||
if (preview?.deployment_id)
|
||
return {
|
||
ok: true as const,
|
||
name: preview.server,
|
||
port: preview.port,
|
||
deploymentId: preview.deployment_id
|
||
};
|
||
if (preview && preview.expires_at <= new Date().toISOString())
|
||
throw new Error('Launch preview expired.');
|
||
const saved = preview ? (JSON.parse(preview.recipe) as NativeRecipe) : undefined;
|
||
const game = input.game ?? 't5';
|
||
if (
|
||
input.gameMode !== 'zombie' ||
|
||
(saved && (saved.game !== game || saved.mapSlug !== input.mapSlug))
|
||
)
|
||
throw new Error('Selection changed. Review a fresh launch preview.');
|
||
const selection = nativeLaunchSelection(
|
||
(saved?.native ?? input) as unknown as Record<string, unknown>,
|
||
game
|
||
);
|
||
const lobby = nextLobbyNumber(`${game}_zombie_${input.mapSlug}`);
|
||
if (lobby === null) throw new Error('All lobby IDs are reserved.');
|
||
const name = `${game}_zombie_${input.mapSlug}_${lobby}`;
|
||
if (
|
||
(input.expectedName && input.expectedName !== name) ||
|
||
(preview && preview.server !== name)
|
||
)
|
||
throw new Error('Available lobby changed. Review a fresh preview.');
|
||
const result = queueNativeCreationIntent({
|
||
server: name,
|
||
game,
|
||
mapSlug: input.mapSlug,
|
||
selection,
|
||
plutoKey: input.plutoKey,
|
||
createdBy: input.createdBy,
|
||
blueprintName: preview?.blueprint_name,
|
||
recipe: saved ?? captureRecipePlan(game, input.mapSlug)
|
||
});
|
||
if (result.ok && preview)
|
||
db()
|
||
.prepare(
|
||
'UPDATE server_launch_previews SET deployment_id=?,port=NULL WHERE id=?'
|
||
)
|
||
.run(result.deploymentId, preview.id);
|
||
return result;
|
||
})();
|
||
} catch (error) {
|
||
return { ok: false, kind: 'bad-request', message: (error as Error).message };
|
||
}
|
||
}
|
||
|
||
/** Persist the reviewed logical identity; all host/port/resource decisions belong to provisioning. */
|
||
export function queueNativeCreationIntent(input: {
|
||
server: string;
|
||
game: 't5' | 't6';
|
||
mapSlug: string;
|
||
selection: NativeSelection;
|
||
plutoKey?: string;
|
||
createdBy?: string;
|
||
blueprintName?: string | null;
|
||
recipe?: LaunchRecipePlan;
|
||
autoscalingDecisionId?: string;
|
||
}): CreateServerResult {
|
||
try {
|
||
return db().transaction(() => {
|
||
assertScalingOwnership(input.game, input.mapSlug, input.autoscalingDecisionId);
|
||
const expected = `${input.game}_zombie_${input.mapSlug}_${nextLobbyNumber(`${input.game}_zombie_${input.mapSlug}`)}`;
|
||
const reserved = db()
|
||
.prepare(
|
||
`SELECT 1 FROM server_creation_jobs j JOIN server_deployments d ON d.id=j.deployment_id WHERE j.server=?
|
||
AND d.server_deleted_at IS NULL AND (d.state IN ('queued','waiting','running','awaiting-health','interrupted') OR (d.state='cancelled' AND d.submitted_at IS NOT NULL))`
|
||
)
|
||
.get(input.server);
|
||
if (
|
||
(!input.autoscalingDecisionId && input.server !== expected) ||
|
||
nativeInstance(input.server) ||
|
||
getServer(input.server) ||
|
||
reserved
|
||
)
|
||
throw new Error('Available lobby changed. Review a fresh preview.');
|
||
reserveServerKey(input.server, input.game, input.plutoKey);
|
||
const recipe: NativeRecipe = {
|
||
...(input.recipe ?? captureRecipePlan(input.game, input.mapSlug)),
|
||
native: input.selection,
|
||
image: input.selection.image!
|
||
};
|
||
const id = beginDeployment(input.server, 'create', input.createdBy ?? 'ui');
|
||
db().prepare("UPDATE server_deployments SET state='queued' 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, input.server, 'zombie', input.mapSlug, recipe.expectedMap);
|
||
db()
|
||
.prepare('INSERT INTO native_creation_intents(deployment_id,selection) VALUES(?,?)')
|
||
.run(id, JSON.stringify(input.selection));
|
||
const stored = recipeStorage(recipe);
|
||
db()
|
||
.prepare(
|
||
'INSERT INTO server_launch_recipes(deployment_id,recipe,recipe_summary,blueprint_name) VALUES(?,?,?,?)'
|
||
)
|
||
.run(id, stored.recipe, stored.recipeSummary, input.blueprintName ?? null);
|
||
deploymentStage(id, 'Awaiting placement', {
|
||
phase: 'eligibility',
|
||
recipeStatus: 'queued'
|
||
});
|
||
return { ok: true as const, name: input.server, port: null, deploymentId: id };
|
||
})();
|
||
} catch (error) {
|
||
return { ok: false, kind: 'bad-request', message: (error as Error).message };
|
||
}
|
||
}
|
||
|
||
export function bindNativeOperation(id: string, server: string): void {
|
||
const instance = nativeInstance(server);
|
||
if (instance)
|
||
db()
|
||
.prepare(
|
||
'INSERT OR IGNORE INTO native_provisioning(deployment_id,instance_id,generation) VALUES(?,?,?)'
|
||
)
|
||
.run(id, instance.id, instance.generation);
|
||
}
|
||
|
||
/** Cancelling an unsubmitted launch releases only its local reservation. */
|
||
export function cancelQueuedNativeCreation(server: string, triggeredBy: string): string | null {
|
||
if (getServer(server)) return null;
|
||
const unplaced = unplacedNativeCreation(server);
|
||
if (unplaced) return cancelCreation(unplaced, triggeredBy);
|
||
const instance = nativeInstance(server);
|
||
if (!instance) return null;
|
||
return db().transaction(() => {
|
||
const creation = db()
|
||
.prepare(
|
||
`SELECT d.id FROM server_deployments d JOIN native_provisioning n ON n.deployment_id=d.id
|
||
WHERE n.instance_id=? AND d.action='create' AND d.state='queued' AND n.operation_id IS NULL`
|
||
)
|
||
.get(instance.id) as { id: string } | undefined;
|
||
if (!creation) return null;
|
||
return cancelCreation(creation.id, triggeredBy);
|
||
})();
|
||
}
|
||
|
||
export function queueNativeRebuild(name: string, triggeredBy: string): CreateServerResult {
|
||
try {
|
||
const instance = nativeInstance(name)!;
|
||
const server = getServer(name);
|
||
const blocker = activeBetaRemovalBlocker(name);
|
||
if (blocker) throw new Error(blocker);
|
||
if (!server?.map_slug || !getServerKey(name))
|
||
throw new Error('A saved map and server key are required before rebuilding.');
|
||
assertReadyLibrary(instance.host_id, instance.library_id);
|
||
const recipe: NativeRecipe = {
|
||
...captureRecipePlan(server.game as 't5' | 't6', server.map_slug),
|
||
image: instance.image,
|
||
native: {
|
||
hostId: instance.host_id,
|
||
libraryId: instance.library_id,
|
||
image: instance.image,
|
||
bindAddress: instance.bind_address
|
||
}
|
||
};
|
||
return db().transaction(() => {
|
||
const id = beginDeployment(name, 'rebuild', 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(?,?,?,?,?,?)'
|
||
)
|
||
.run(id, name, instance.port, 'zombie', recipe.mapSlug, recipe.expectedMap);
|
||
const stored = recipeStorage(recipe);
|
||
db()
|
||
.prepare(
|
||
'INSERT INTO server_launch_recipes(deployment_id,recipe,recipe_summary) VALUES(?,?,?)'
|
||
)
|
||
.run(id, stored.recipe, stored.recipeSummary);
|
||
deploymentStage(id, 'Waiting to rebuild when empty', {
|
||
phase: 'eligibility',
|
||
recipeStatus: 'queued'
|
||
});
|
||
return { ok: true as const, name, port: instance.port, deploymentId: id };
|
||
})();
|
||
} catch (error) {
|
||
return { ok: false, kind: 'bad-request', message: (error as Error).message };
|
||
}
|
||
}
|
||
|
||
function filesFor(job: Job): FilePlan[] {
|
||
const rows = db()
|
||
.prepare(
|
||
`SELECT f.id,f.kind,f.rel_path,f.content_hash,f.check_before,f.expected_before_sha
|
||
FROM staged_file_changes f JOIN provisioning_sources s
|
||
ON s.kind=f.source_kind AND s.source_id=f.source_id AND s.server=f.server
|
||
WHERE s.operation_id=? AND f.server=? AND f.state='pending' ORDER BY f.id`
|
||
)
|
||
.all(job.id, job.server) as Array<
|
||
Pick<
|
||
StagedFileChange,
|
||
'id' | 'kind' | 'rel_path' | 'check_before' | 'expected_before_sha'
|
||
> & { content_hash: string | null }
|
||
>;
|
||
for (const row of rows) assertPublicFileTarget(job.server, row.kind, row.rel_path);
|
||
return rows.map((row) => ({
|
||
id: row.id,
|
||
path: posix.join(
|
||
relDestForKind(row.kind, job.server.startsWith('t6_') ? 't6' : 't5'),
|
||
row.rel_path
|
||
),
|
||
sha256: row.content_hash,
|
||
checkBefore: !!row.check_before,
|
||
expectedBeforeSha: row.expected_before_sha
|
||
}));
|
||
}
|
||
|
||
function fileMetadata({
|
||
contentBase64,
|
||
...file
|
||
}: FilePlan & { contentBase64?: string | null }): FilePlan {
|
||
return file;
|
||
}
|
||
|
||
function storedFiles(job: Job): FilePlan[] {
|
||
const row = db()
|
||
.prepare('SELECT files_json FROM native_provisioning WHERE deployment_id=?')
|
||
.get(job.id) as { files_json: string | null } | undefined;
|
||
const files = (JSON.parse(row?.files_json ?? '[]') as FilePlan[]).map(fileMetadata);
|
||
if (files.some((file) => isPrivateConfigPath(file.path)))
|
||
throw new Error(
|
||
'Private bank configuration cannot be captured or changed by a file campaign.'
|
||
);
|
||
return files;
|
||
}
|
||
|
||
/** Agent receipts are the authoritative before images; do not read a controller mount. */
|
||
function applyReceipt(job: Job, result: Record<string, unknown>) {
|
||
if (job.applied_at || job.recovery_step) return;
|
||
const plans = storedFiles(job);
|
||
if (!plans.length) return;
|
||
const before = result.before as
|
||
| { id?: number; path: string; contentBase64: string | null; contentSha256?: string }[]
|
||
| undefined;
|
||
db().transaction(() => {
|
||
for (const plan of plans) {
|
||
const snapshot = before?.find((item) => item.id === plan.id || item.path === plan.path);
|
||
if (!snapshot) throw new Error('Agent activation receipt is missing a before image.');
|
||
const row = db()
|
||
.prepare(
|
||
`SELECT *, ${blobSize('content_hash')} AS size FROM staged_file_changes WHERE id=?`
|
||
)
|
||
.get(plan.id) as Omit<StagedFileChange, 'content'> & {
|
||
content_hash: string | null;
|
||
size: number;
|
||
};
|
||
if (!row || row.state === 'applied') continue;
|
||
const bytes = snapshot.contentSha256
|
||
? getFileBlob(snapshot.contentSha256)
|
||
: snapshot.contentBase64 === null
|
||
? null
|
||
: Buffer.from(snapshot.contentBase64, 'base64');
|
||
if (row.source_kind === 'beta') {
|
||
captureBetaBeforeImage(row, bytes);
|
||
db()
|
||
.prepare(
|
||
'UPDATE beta_servers SET deployed_at=?,updated_at=? WHERE beta_id=? AND server=?'
|
||
)
|
||
.run(
|
||
new Date().toISOString(),
|
||
new Date().toISOString(),
|
||
row.source_id,
|
||
job.server
|
||
);
|
||
}
|
||
if (row.deployment_file_id && !row.preserve_before_image) {
|
||
db()
|
||
.prepare(
|
||
'UPDATE deployment_files SET verb=?,content_before_hash=?,sha256_before=?,snapshot_pending=0 WHERE id=?'
|
||
)
|
||
.run(
|
||
row.content_hash === null
|
||
? 'removed'
|
||
: bytes === null
|
||
? 'added'
|
||
: 'updated',
|
||
putFileBlob(bytes),
|
||
bytes === null ? null : sha256Buffer(bytes),
|
||
row.deployment_file_id
|
||
);
|
||
if (row.source_kind === 'campaign')
|
||
db()
|
||
.prepare(
|
||
'INSERT INTO campaign_events (deployment_id, message) VALUES (?, ?)'
|
||
)
|
||
.run(
|
||
row.source_id,
|
||
`${job.server}: captured rollback snapshot for ${row.rel_path}`
|
||
);
|
||
}
|
||
if (row.content_hash === null) removeScript(job.server, row.rel_path);
|
||
else
|
||
recordScript(
|
||
job.server,
|
||
row.rel_path,
|
||
row.script_version_id &&
|
||
db()
|
||
.prepare('SELECT 1 FROM script_versions WHERE id=?')
|
||
.get(row.script_version_id)
|
||
? row.script_version_id
|
||
: null,
|
||
row.content_hash,
|
||
row.size,
|
||
row.inventory_source
|
||
);
|
||
db()
|
||
.prepare("UPDATE staged_file_changes SET state='applied',applied_at=? WHERE id=?")
|
||
.run(new Date().toISOString(), row.id);
|
||
}
|
||
db()
|
||
.prepare('UPDATE native_provisioning SET applied_at=? WHERE deployment_id=?')
|
||
.run(new Date().toISOString(), job.id);
|
||
})();
|
||
}
|
||
|
||
/** Consume a conclusive partial failure without treating removal as a healthy rebuild. */
|
||
function applyIncompleteRebuildReceipt(job: Job, operation: HostOperationStatus): void {
|
||
const result = operation.result as {
|
||
reconciled?: unknown;
|
||
rebuildFailure?: {
|
||
phase?: unknown;
|
||
generation?: unknown;
|
||
nextGeneration?: unknown;
|
||
oldContainerMissing?: unknown;
|
||
replacementContainerMissing?: unknown;
|
||
replacementFilesRemoved?: unknown;
|
||
} | null;
|
||
} | null;
|
||
if (!result || !Object.prototype.hasOwnProperty.call(result, 'rebuildFailure')) return;
|
||
const failure = result.rebuildFailure;
|
||
if (
|
||
operation.state !== 'failed' ||
|
||
operation.kind !== 'instance.rebuild' ||
|
||
result.reconciled !== true ||
|
||
!failure ||
|
||
failure.phase !== 'disrupting' ||
|
||
failure.generation !== job.generation ||
|
||
failure.nextGeneration !== job.generation + 1 ||
|
||
failure.oldContainerMissing !== true ||
|
||
failure.replacementContainerMissing !== true ||
|
||
failure.replacementFilesRemoved !== true
|
||
) {
|
||
throw new Error(
|
||
'The interrupted rebuild receipt does not confirm safe replacement cleanup. Inspect the original host command.'
|
||
);
|
||
}
|
||
db().transaction(() => {
|
||
const binding = db()
|
||
.prepare(
|
||
'SELECT operation_id,applied_at FROM native_provisioning WHERE deployment_id=?'
|
||
)
|
||
.get(job.id) as { operation_id: string | null; applied_at: string | null } | undefined;
|
||
if (binding?.operation_id !== operation.id)
|
||
throw new Error('The interrupted rebuild command no longer belongs to this operation.');
|
||
if (binding.applied_at) return;
|
||
const requested = db()
|
||
.prepare(
|
||
`SELECT COALESCE(json_extract(COALESCE(p.payload,o.payload),'$.nextGeneration'),
|
||
json_extract(COALESCE(p.payload,o.payload),'$.newGeneration')) AS nextGeneration FROM ec_host_operations o
|
||
LEFT JOIN ec_host_operation_payloads p ON p.operation_id=o.id WHERE o.id=?`
|
||
)
|
||
.get(operation.id) as { nextGeneration: number | null } | undefined;
|
||
if (requested?.nextGeneration !== failure.nextGeneration) {
|
||
throw new Error(
|
||
'The interrupted rebuild receipt does not match its requested replacement generation.'
|
||
);
|
||
}
|
||
const instance = nativeInstance(job.server);
|
||
if (
|
||
!instance ||
|
||
instance.id !== job.instance_id ||
|
||
instance.generation !== job.generation ||
|
||
operation.instanceId !== instance.id ||
|
||
operation.hostId !== instance.host_id ||
|
||
operation.generation !== instance.generation
|
||
) {
|
||
throw new Error(
|
||
'The interrupted rebuild belongs to a different instance or generation. Inspect its host command.'
|
||
);
|
||
}
|
||
db()
|
||
.prepare(
|
||
"UPDATE servers SET container_state='missing',container_checked_at=?,last_players=NULL,last_roster=NULL WHERE name=?"
|
||
)
|
||
.run(operation.updatedAt, job.server);
|
||
db()
|
||
.prepare('UPDATE native_provisioning SET applied_at=? WHERE deployment_id=?')
|
||
.run(operation.updatedAt, job.id);
|
||
recordProgress(job.id, {
|
||
checks: [
|
||
{
|
||
key: 'container',
|
||
label: 'Container state after interrupted rebuild',
|
||
source: 'agent',
|
||
result: 'fail',
|
||
actual: 'missing',
|
||
checkedAt: operation.updatedAt,
|
||
message:
|
||
'The old container is absent and replacement creation never started. Partial replacement files were removed.'
|
||
}
|
||
],
|
||
event: {
|
||
kind: 'reconciled',
|
||
message:
|
||
'Interrupted rebuild settled: old container removed; replacement creation failed.',
|
||
details: {
|
||
operationId: operation.id,
|
||
generation: job.generation,
|
||
nextGeneration: failure.nextGeneration
|
||
}
|
||
}
|
||
});
|
||
})();
|
||
}
|
||
|
||
/** Retain confirmed file snapshots and instance identity even after observation is cancelled. */
|
||
export function reconcileNativeProvisioningReceipt(id: string): void {
|
||
const job = db()
|
||
.prepare(
|
||
`SELECT d.*,n.instance_id,n.generation,n.operation_id,n.applied_at,n.recovery_step
|
||
FROM server_deployments d JOIN native_provisioning n ON n.deployment_id=d.id
|
||
WHERE d.id=? AND d.server_deleted_at IS NULL`
|
||
)
|
||
.get(id) as Job | undefined;
|
||
if (!job) return;
|
||
const migration = hostMigration(id);
|
||
if (job.state === 'cancelled' && migration?.remove_operation_id && !migration.removed_at) {
|
||
const removal = getHostOperationStatus(migration.remove_operation_id);
|
||
if (removal?.state === 'failed') {
|
||
db()
|
||
.prepare('DELETE FROM ec_host_ports WHERE host_id=? AND instance_id=?')
|
||
.run(migration.target_host_id, migration.instance_id);
|
||
} else if (removal?.state === 'completed') {
|
||
const original = nativeInstance(job.server);
|
||
if (
|
||
original?.id === migration.instance_id &&
|
||
original.host_id === migration.source_host_id &&
|
||
original.generation === job.generation
|
||
) {
|
||
db()
|
||
.prepare('UPDATE native_instances SET retired_at=? WHERE id=?')
|
||
.run(removal.updatedAt, original.id);
|
||
releaseHostPorts(original.id);
|
||
db().prepare('DELETE FROM servers WHERE name=?').run(job.server);
|
||
releaseServerKey(job.server);
|
||
db()
|
||
.prepare('UPDATE ec_host_migrations SET removed_at=? WHERE deployment_id=?')
|
||
.run(removal.updatedAt, id);
|
||
recordProgress(id, {
|
||
event: {
|
||
kind: 'cleanup',
|
||
message:
|
||
'Cancelled migration: source removal completed; replacement creation was not started.'
|
||
}
|
||
});
|
||
}
|
||
}
|
||
}
|
||
if (!job.operation_id) return;
|
||
const operation = getHostOperationStatus(job.operation_id);
|
||
if (operation?.state === 'failed') {
|
||
applyIncompleteRebuildReceipt(job, operation);
|
||
for (const file of storedFiles(job)) {
|
||
db()
|
||
.prepare("UPDATE staged_file_changes SET state=? WHERE id=? AND state='applying'")
|
||
.run(job.state === 'cancelled' ? 'cancelled' : 'pending', file.id);
|
||
}
|
||
return;
|
||
}
|
||
if (operation?.state !== 'completed') return;
|
||
const instance = nativeInstance(job.server);
|
||
if (!instance || instance.id !== job.instance_id) return;
|
||
if (
|
||
job.state === 'cancelled' &&
|
||
job.action === 'create' &&
|
||
db().prepare('SELECT 1 FROM native_creation_cancellations WHERE creation_id=?').get(id)
|
||
) {
|
||
// Cleanup owns both generations of a cancelled recovery rebuild. Do not
|
||
// advance the registry underneath its already-bound removal command.
|
||
db()
|
||
.prepare(
|
||
'UPDATE native_provisioning SET applied_at=COALESCE(applied_at,?) WHERE deployment_id=?'
|
||
)
|
||
.run(operation.updatedAt, id);
|
||
return;
|
||
}
|
||
if (job.action === 'destroy') {
|
||
if (instance.generation !== job.generation) return;
|
||
db()
|
||
.prepare('UPDATE native_instances SET retired_at=? WHERE id=?')
|
||
.run(operation.updatedAt, instance.id);
|
||
releaseHostPorts(instance.id);
|
||
db().prepare('DELETE FROM servers WHERE name=?').run(job.server);
|
||
releaseServerKey(job.server);
|
||
finishCreationCancellation(job.id);
|
||
return;
|
||
}
|
||
if (operation.kind === 'instance.rebuild' && instance.generation === job.generation) {
|
||
db()
|
||
.prepare('UPDATE native_instances SET generation=generation+1 WHERE id=?')
|
||
.run(instance.id);
|
||
// Replacements queued while the delivered rebuild was settling bound the
|
||
// previous generation. Only untouched, newer jobs may follow this receipt.
|
||
if (job.state === 'cancelled')
|
||
db()
|
||
.prepare(
|
||
`UPDATE native_provisioning SET generation=?
|
||
WHERE instance_id=? AND generation=? AND operation_id IS NULL AND stage_operation_id IS NULL
|
||
AND deployment_id IN (SELECT newer.id FROM server_deployments newer
|
||
WHERE newer.rowid>(SELECT rowid FROM server_deployments WHERE id=?)
|
||
AND newer.server_deleted_at IS NULL AND newer.state IN ('queued','waiting'))`
|
||
)
|
||
.run(job.generation + 1, instance.id, job.generation, id);
|
||
} else if (
|
||
instance.generation !== job.generation &&
|
||
!(operation.kind === 'instance.rebuild' && instance.generation === job.generation + 1)
|
||
)
|
||
return;
|
||
applyReceipt(job, (operation.result ?? {}) as Record<string, unknown>);
|
||
db()
|
||
.prepare(
|
||
'UPDATE native_provisioning SET applied_at=COALESCE(applied_at,?) WHERE deployment_id=?'
|
||
)
|
||
.run(operation.updatedAt, id);
|
||
}
|
||
|
||
function hold(job: Job, message: string) {
|
||
if (!stillActive(job)) return;
|
||
deploymentStage(job.id, message, {
|
||
blocker: { code: 'native-host', message, automatic: true }
|
||
});
|
||
}
|
||
function stillActive(job: Job): boolean {
|
||
return !!db()
|
||
.prepare(
|
||
`SELECT 1 FROM server_deployments WHERE id=? AND server_deleted_at IS NULL
|
||
AND state IN ('queued','waiting','running','awaiting-health','interrupted')`
|
||
)
|
||
.get(job.id);
|
||
}
|
||
function complete(job: Job) {
|
||
db().transaction(() => {
|
||
db()
|
||
.prepare('UPDATE native_provisioning SET verified_at=? WHERE deployment_id=?')
|
||
.run(new Date().toISOString(), job.id);
|
||
db()
|
||
.prepare('UPDATE restart_rollout SET healthy_at=?,error=NULL WHERE operation_id=?')
|
||
.run(new Date().toISOString(), job.id);
|
||
if (job.recovery_step)
|
||
recordProgress(job.id, {
|
||
recovery: {
|
||
state: 'released',
|
||
operationId: job.id,
|
||
message: 'Container and game health verified after provisioning recovery.'
|
||
}
|
||
});
|
||
finishDeployment(job.id, { ok: true });
|
||
})();
|
||
}
|
||
|
||
/** Continue the same launch with at most one restart and one replacement. */
|
||
function queueUnhealthyRecovery(job: Job, instance: NativeInstance): void {
|
||
db().transaction(() => {
|
||
if (!stillActive(job)) return;
|
||
const live = nativeInstance(job.server);
|
||
if (live?.id !== instance.id || live.generation !== instance.generation) return;
|
||
if (job.recovery_step === 'rebuild') {
|
||
const message =
|
||
'Container is still unhealthy after recovery restart and rebuild. Inspect the server before retrying.';
|
||
db()
|
||
.prepare('UPDATE restart_rollout SET error=? WHERE operation_id=?')
|
||
.run(message, job.id);
|
||
finishDeployment(job.id, { ok: false, message });
|
||
return;
|
||
}
|
||
const host = getHost(instance.host_id);
|
||
if (!host?.connected || host.mode === 'disabled') {
|
||
hold(job, 'Host is unavailable; waiting for reconnection.');
|
||
return;
|
||
}
|
||
if (!host.inventory.capabilities?.includes(unhealthyRecoveryCapability)) {
|
||
hold(
|
||
job,
|
||
'Update this host’s agent to recover unhealthy containers during provisioning.'
|
||
);
|
||
return;
|
||
}
|
||
const step = job.recovery_step === 'restart' ? 'rebuild' : 'restart';
|
||
const settings = fleetSettings();
|
||
const slots = db()
|
||
.prepare(
|
||
`SELECT i.host_id FROM restart_rollout r LEFT JOIN native_instances i
|
||
ON i.server=r.server AND i.retired_at IS NULL WHERE r.healthy_at IS NULL AND (r.operation_id IS NULL OR r.operation_id!=?)`
|
||
)
|
||
.all(job.id) as { host_id: string | null }[];
|
||
if (settings.restartConcurrency > 0 && slots.length >= settings.restartConcurrency) {
|
||
hold(job, 'Waiting for restart recovery capacity.');
|
||
return;
|
||
}
|
||
if (
|
||
settings.restartConcurrencyPerHost > 0 &&
|
||
slots.filter((slot) => slot.host_id === instance.host_id).length >=
|
||
settings.restartConcurrencyPerHost
|
||
) {
|
||
hold(job, 'Waiting for restart recovery capacity on this host.');
|
||
return;
|
||
}
|
||
let payload: Record<string, unknown> = { unhealthyRecovery: true };
|
||
if (step === 'rebuild') {
|
||
const beta = activeBetaRemovalBlocker(
|
||
job.server,
|
||
job.action === 'rebuild' ? job.id : undefined
|
||
);
|
||
if (beta) {
|
||
hold(job, beta);
|
||
return;
|
||
}
|
||
if (job.action !== 'rebuild' && settings.rebuildConcurrency > 0) {
|
||
if (
|
||
lifecycleHolders().filter(
|
||
(holder) => holder.id !== job.id && holder.action === 'rebuild'
|
||
).length >= settings.rebuildConcurrency
|
||
) {
|
||
hold(job, 'Waiting for rebuild capacity.');
|
||
return;
|
||
}
|
||
}
|
||
assertReadyLibrary(instance.host_id, instance.library_id);
|
||
// Recover the files and credentials selected for this launch, even
|
||
// if new versions have been promoted while readiness was pending.
|
||
const launch = db()
|
||
.prepare(
|
||
`SELECT COALESCE(p.payload,o.payload) AS payload FROM ec_host_operations o
|
||
JOIN native_instances i ON ${runtimeLaunchSql}
|
||
LEFT JOIN ec_host_operation_payloads p ON p.operation_id=o.id
|
||
WHERE i.id=? AND o.state='completed' AND o.history_expired_at IS NULL ORDER BY o.rowid DESC LIMIT 1`
|
||
)
|
||
.get(instance.id) as { payload: string } | undefined;
|
||
if (!launch) {
|
||
hold(job, 'Waiting for the original launch payload before recovery rebuild.');
|
||
return;
|
||
}
|
||
payload = {
|
||
...JSON.parse(launch.payload),
|
||
nextGeneration: instance.generation + 1,
|
||
unhealthyRecovery: true
|
||
};
|
||
delete payload.safetyOverride;
|
||
delete payload.stageId;
|
||
}
|
||
const operation = queueHostOperationSerialized(
|
||
instance.host_id,
|
||
`instance.${step}`,
|
||
Buffer.from(JSON.stringify(payload)),
|
||
{
|
||
instanceId: instance.id,
|
||
generation: instance.generation,
|
||
idempotencyKey: `${job.id}:unhealthy:${step}`
|
||
},
|
||
{ name: instance.server, unhealthyRecovery: true }
|
||
);
|
||
const at = new Date().toISOString();
|
||
db()
|
||
.prepare(
|
||
'UPDATE native_provisioning SET operation_id=?,generation=?,recovery_step=?,applied_at=NULL WHERE deployment_id=?'
|
||
)
|
||
.run(operation.id, instance.generation, step, job.id);
|
||
db()
|
||
.prepare(
|
||
"UPDATE server_deployments SET state='running',submitted_at=?,error=NULL,finished_at=NULL WHERE id=?"
|
||
)
|
||
.run(at, job.id);
|
||
db()
|
||
.prepare(
|
||
`INSERT INTO restart_rollout(server,requested_at,operation_id) VALUES(?,?,?) ON CONFLICT(server)
|
||
DO UPDATE SET requested_at=excluded.requested_at,operation_id=excluded.operation_id,healthy_at=NULL,error=NULL`
|
||
)
|
||
.run(job.server, at, job.id);
|
||
const message = `Container unhealthy; ${step === 'restart' ? 'forcing a restart' : 'forcing a rebuild after restart recovery failed'}.`;
|
||
deploymentStage(job.id, message, {
|
||
phase: 'container-action',
|
||
blocker: null,
|
||
checks: [],
|
||
command: {
|
||
state: 'attempted',
|
||
message: `Recovery ${step} queued as agent operation ${operation.id}.`
|
||
},
|
||
recovery: {
|
||
state: 'held',
|
||
operationId: job.id,
|
||
message: 'Recovery runs within this provisioning operation.'
|
||
},
|
||
event: {
|
||
kind: 'unhealthy-recovery',
|
||
message,
|
||
details: { step, operationId: operation.id, generation: instance.generation }
|
||
}
|
||
});
|
||
})();
|
||
}
|
||
|
||
export async function processNativeProvisioning(
|
||
eligibility: (operation: {
|
||
id: string;
|
||
server: string;
|
||
action: string;
|
||
state: string;
|
||
stage: string;
|
||
}) => string | null,
|
||
options: { wait?: boolean } = {}
|
||
): Promise<void> {
|
||
const cancelled = db()
|
||
.prepare(
|
||
`SELECT d.id FROM server_deployments d
|
||
JOIN native_provisioning n ON n.deployment_id=d.id
|
||
LEFT JOIN ec_host_operations o ON o.id=n.operation_id
|
||
LEFT JOIN ec_host_migrations m ON m.deployment_id=d.id
|
||
LEFT JOIN ec_host_operations removal ON removal.id=m.remove_operation_id
|
||
WHERE d.state='cancelled' AND d.server_deleted_at IS NULL AND (
|
||
(o.state='completed' AND (n.applied_at IS NULL OR d.action='destroy'))
|
||
OR (o.state='failed' AND o.kind='instance.rebuild' AND json_type(o.result,'$.rebuildFailure') IS NOT NULL AND n.applied_at IS NULL)
|
||
OR (o.state='failed' AND EXISTS (SELECT 1 FROM staged_file_changes f JOIN provisioning_sources s
|
||
ON s.kind=f.source_kind AND s.source_id=f.source_id AND s.server=f.server
|
||
WHERE s.operation_id=d.id AND f.state='applying'))
|
||
OR (m.removed_at IS NULL AND (removal.state='completed' OR (removal.state='failed'
|
||
AND EXISTS (SELECT 1 FROM ec_host_ports p WHERE p.host_id=m.target_host_id AND p.instance_id=m.instance_id))))
|
||
)`
|
||
)
|
||
.all() as { id: string }[];
|
||
for (const row of cancelled) {
|
||
try {
|
||
reconcileNativeProvisioningReceipt(row.id);
|
||
} catch (error) {
|
||
deploymentStage(
|
||
row.id,
|
||
'Cancelled; waiting to reconcile the delivered command: ' + (error as Error).message
|
||
);
|
||
}
|
||
}
|
||
reconcileHostDrains();
|
||
const jobs = db()
|
||
.prepare(
|
||
`SELECT d.*,n.instance_id,n.generation,n.stage_operation_id,n.operation_id,n.applied_at,n.recovery_step,q.selection
|
||
FROM server_deployments d LEFT JOIN native_provisioning n ON n.deployment_id=d.id
|
||
LEFT JOIN native_creation_intents q ON q.deployment_id=d.id
|
||
WHERE (n.deployment_id IS NOT NULL OR q.deployment_id IS NOT NULL) AND d.server_deleted_at IS NULL AND d.state IN ('queued','waiting','running','awaiting-health','interrupted') ORDER BY d.started_at,d.rowid`
|
||
)
|
||
.all() as Job[];
|
||
// One host's observation must not prevent another job being admitted.
|
||
// Per-job ownership also covers overlapping direct calls and worker passes.
|
||
const work = jobs.map((job) => {
|
||
const pending = activeJobs.get(job.id);
|
||
if (pending) return pending;
|
||
const work = trackActivity(
|
||
`provisioning:${job.id}`,
|
||
`Evaluating ${job.action} for ${job.server}`,
|
||
() => processJob(job, eligibility)
|
||
)
|
||
.catch((error) => {
|
||
if (stillActive(job)) hold(job, (error as Error).message);
|
||
})
|
||
.finally(() => {
|
||
activeJobs.delete(job.id);
|
||
preparingJobs.delete(job.id);
|
||
});
|
||
activeJobs.set(job.id, work);
|
||
return work;
|
||
});
|
||
if (options.wait !== false) {
|
||
await Promise.all(work);
|
||
reconcileHostDrains();
|
||
}
|
||
}
|
||
|
||
const activeJobs = new Map<string, Promise<void>>();
|
||
// Preparation owns startup capacity until its command is committed or discarded.
|
||
|
||
function otherPreparations(id: string): number {
|
||
if (!preparingJobs.size) return 0;
|
||
const reserved = new Set(
|
||
(
|
||
db()
|
||
.prepare(
|
||
`SELECT id FROM server_deployments WHERE id IN (SELECT value FROM json_each(?))
|
||
AND ${committedRebalanceStartup}`
|
||
)
|
||
.all(JSON.stringify([...preparingJobs])) as { id: string }[]
|
||
).map((row) => row.id)
|
||
);
|
||
return [...preparingJobs].filter((other) => other !== id && !reserved.has(other)).length;
|
||
}
|
||
|
||
let launchCommit: Promise<void> = Promise.resolve();
|
||
function commitPreparedLaunch(commit: () => void): Promise<void> {
|
||
// Worker replies can arrive together. Persist one launch per event-loop turn
|
||
// so a batch of large SQLite writes cannot hold off requests and health timers.
|
||
const next = launchCommit.then(async () => {
|
||
await yieldToController();
|
||
commit();
|
||
});
|
||
launchCommit = next.catch(() => {});
|
||
return next;
|
||
}
|
||
|
||
function sameLaunchFiles(first: LaunchRecipePlan, second: LaunchRecipePlan): boolean {
|
||
const selected = (recipe: LaunchRecipePlan) => ({
|
||
game: recipe.game,
|
||
mapSlug: recipe.mapSlug,
|
||
expectedMap: recipe.expectedMap,
|
||
configFile: recipe.configFile,
|
||
configSource: recipe.configSource,
|
||
files: recipe.files
|
||
});
|
||
return JSON.stringify(selected(first)) === JSON.stringify(selected(second));
|
||
}
|
||
|
||
/** Capacity history and host ranking run in the background reader, one job at a time. */
|
||
async function placeQueuedCreation(
|
||
job: Job,
|
||
instance: NativeInstance | null,
|
||
eligibility: (job: Job) => string | null
|
||
): Promise<NativeInstance | null> {
|
||
const recipe = recipeForDeployment(job.id)!.recipe as NativeRecipe;
|
||
const prior: NativeSelection = job.selection ? JSON.parse(job.selection) : recipe.native!;
|
||
const network = prior.automatic ? (prior.network ?? {}) : prior;
|
||
const options: PlacementOptions = {
|
||
hostId: prior.automatic ? undefined : prior.hostId,
|
||
libraryId: prior.libraryId,
|
||
image: prior.image ?? recipe.image,
|
||
bindAddress: network.bindAddress,
|
||
thorough: true,
|
||
dispatch: true,
|
||
excludeInstanceId: instance?.id
|
||
};
|
||
const decision = await evaluateNativePlacement(options);
|
||
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?.state !== 'queued') return null;
|
||
if (eligibility(job)) return null;
|
||
const live = nativeInstance(job.server);
|
||
if (
|
||
instance
|
||
? live?.id !== instance.id || live.generation !== instance.generation
|
||
: !!live || !!getServer(job.server)
|
||
)
|
||
return null;
|
||
const active = (
|
||
db()
|
||
.prepare(
|
||
`SELECT count(*) AS n FROM server_deployments WHERE id!=? AND server_deleted_at IS NULL
|
||
AND action IN ('create','rebuild') AND state IN ('running','awaiting-health')`
|
||
)
|
||
.get(job.id) as { n: number }
|
||
).n;
|
||
if (active + otherPreparations(job.id) >= fleetSettings().createConcurrency) {
|
||
hold(job, 'Waiting for startup capacity.');
|
||
return null;
|
||
}
|
||
const capacityBlocker = concurrencyBlocker(job.id, job.action, decision.hostId);
|
||
if (capacityBlocker) {
|
||
hold(job, capacityBlocker);
|
||
return null;
|
||
}
|
||
if (getHost(decision.hostId)?.lastHeartbeat !== decision.lastHeartbeat) {
|
||
hold(job, 'Host measurements changed; waiting for another placement check.');
|
||
return null;
|
||
}
|
||
// A short current-state check protects against reservations made while the
|
||
// reader ranked hosts. It never scans resource history or contacts agents.
|
||
const { host, port } = choosePlacement({
|
||
...options,
|
||
hostId: decision.hostId,
|
||
thorough: false
|
||
});
|
||
const bind = network.bindAddress ?? host.bindAddress;
|
||
const instanceId = instance?.id ?? randomUUID();
|
||
if (instance) releaseHostPorts(instanceId);
|
||
reserveHostPort(host.id, instanceId, { port, bindAddress: bind, protocol: 'udp' });
|
||
reserveHostPort(host.id, instanceId, { port, bindAddress: bind, protocol: 'tcp' });
|
||
if (instance)
|
||
db()
|
||
.prepare(
|
||
`UPDATE native_instances SET host_id=?,bind_address=?,port=?,public_address=?,public_port=? WHERE id=?`
|
||
)
|
||
.run(
|
||
host.id,
|
||
bind,
|
||
port,
|
||
network.publicAddress ?? host.publicAddress,
|
||
network.publicPort ?? null,
|
||
instanceId
|
||
);
|
||
else {
|
||
db()
|
||
.prepare(
|
||
`INSERT INTO native_instances(id,server,host_id,library_id,image,bind_address,port,public_address,public_port,created_at)
|
||
VALUES(?,?,?,?,?,?,?,?,?,?)`
|
||
)
|
||
.run(
|
||
instanceId,
|
||
job.server,
|
||
host.id,
|
||
prior.libraryId,
|
||
options.image,
|
||
bind,
|
||
port,
|
||
network.publicAddress ?? host.publicAddress,
|
||
network.publicPort ?? null,
|
||
new Date().toISOString()
|
||
);
|
||
db()
|
||
.prepare(
|
||
'INSERT INTO native_provisioning(deployment_id,instance_id,generation) VALUES(?,?,1)'
|
||
)
|
||
.run(job.id, instanceId);
|
||
}
|
||
db()
|
||
.prepare('UPDATE server_creation_jobs SET port=? WHERE deployment_id=?')
|
||
.run(port, job.id);
|
||
recipe.native = {
|
||
...prior,
|
||
hostId: host.id,
|
||
bindAddress: bind,
|
||
publicAddress: network.publicAddress ?? host.publicAddress,
|
||
publicPort: network.publicPort
|
||
};
|
||
const stored = recipeStorage(recipe);
|
||
db()
|
||
.prepare(
|
||
'UPDATE server_launch_recipes SET recipe=?,recipe_summary=? WHERE deployment_id=?'
|
||
)
|
||
.run(stored.recipe, stored.recipeSummary, job.id);
|
||
db()
|
||
.prepare('UPDATE server_launch_previews SET port=? WHERE deployment_id=?')
|
||
.run(port, job.id);
|
||
deploymentStage(
|
||
job.id,
|
||
`Placed on ${host.name} after CPU, memory, history and port checks`,
|
||
{ phase: 'eligibility' }
|
||
);
|
||
preparingJobs.add(job.id);
|
||
return nativeInstance(job.server)!;
|
||
})();
|
||
}
|
||
|
||
async function processJob(job: Job, eligibility: (job: Job) => string | null) {
|
||
if (!stillActive(job)) return;
|
||
const unsettled = unsettledCancelledCommand(job.server, job.id);
|
||
if (unsettled) {
|
||
const pendingReceipt = unsettled.state === 'completed' || unsettled.state === 'failed';
|
||
const message =
|
||
`Waiting for the host command from cancelled operation ${unsettled.deploymentId} ${pendingReceipt ? 'to be reconciled' : 'to settle'}. Agent state: ${unsettled.state}.` +
|
||
(unsettled.error ? ` Last error: ${unsettled.error.slice(0, 500)}` : '');
|
||
deploymentStage(job.id, message, {
|
||
blocker: {
|
||
code: 'cancelled-command',
|
||
message,
|
||
automatic: true,
|
||
href: `/provisioning/${encodeURIComponent(unsettled.deploymentId)}`,
|
||
label: 'View cancelled operation'
|
||
}
|
||
});
|
||
return;
|
||
}
|
||
let instance = nativeInstance(job.server);
|
||
if (
|
||
job.action === 'create' &&
|
||
!job.operation_id &&
|
||
!job.stage_operation_id &&
|
||
!job.submitted_at
|
||
) {
|
||
// A resumed bound job owns its original identity. An unplaced intent
|
||
// cannot adopt an instance or registered server that appeared later.
|
||
if (
|
||
job.instance_id
|
||
? !instance ||
|
||
instance.id !== job.instance_id ||
|
||
instance.generation !== job.generation
|
||
: !!instance || !!getServer(job.server)
|
||
) {
|
||
finishDeployment(job.id, {
|
||
ok: false,
|
||
message: 'Original instance was retired or replaced; no command was sent.'
|
||
});
|
||
return;
|
||
}
|
||
const settings = fleetSettings();
|
||
const active = db()
|
||
.prepare(
|
||
`SELECT COUNT(*) AS n FROM server_deployments WHERE id!=? AND server_deleted_at IS NULL
|
||
AND action IN ('create','rebuild') AND (state IN ('running','awaiting-health') OR ${committedRebalanceStartup})`
|
||
)
|
||
.get(job.id) as { n: number };
|
||
if (active.n + otherPreparations(job.id) >= settings.createConcurrency) {
|
||
hold(job, 'Waiting for startup capacity.');
|
||
return;
|
||
}
|
||
const capacityBlocker = concurrencyBlocker(job.id, job.action);
|
||
if (capacityBlocker) {
|
||
hold(job, capacityBlocker);
|
||
return;
|
||
}
|
||
if (eligibility(job)) return;
|
||
instance = await placeQueuedCreation(job, instance, eligibility);
|
||
if (!instance) return;
|
||
job.instance_id = instance.id;
|
||
job.generation = instance.generation;
|
||
}
|
||
const migration = hostMigration(job.id);
|
||
const rebuilding =
|
||
job.recovery_step === 'rebuild' ||
|
||
(!job.recovery_step && job.action === 'rebuild' && !migration);
|
||
if (
|
||
!instance ||
|
||
instance.id !== job.instance_id ||
|
||
(instance.generation !== job.generation &&
|
||
!(rebuilding && instance.generation === job.generation + 1))
|
||
) {
|
||
finishDeployment(job.id, {
|
||
ok: false,
|
||
message: 'Original instance was retired or replaced; no command was sent.'
|
||
});
|
||
return;
|
||
}
|
||
if (migration && !migration.removed_at) {
|
||
await processMigrationRemoval(job, instance, migration, () => eligibility(job));
|
||
return;
|
||
}
|
||
const host = getHost(instance.host_id);
|
||
if (!host?.connected || host.mode === 'disabled') {
|
||
db()
|
||
.prepare(
|
||
"UPDATE servers SET container_state='unknown',last_players=NULL,last_roster=NULL WHERE name=?"
|
||
)
|
||
.run(job.server);
|
||
hold(
|
||
job,
|
||
'Host is unavailable; existing server state is unknown. Waiting for reconnection.'
|
||
);
|
||
return;
|
||
}
|
||
if (job.operation_id) {
|
||
const operation = getHostOperationStatus(job.operation_id);
|
||
const recoveryCleanup =
|
||
operation?.kind === 'instance.cancel-create' &&
|
||
!!db()
|
||
.prepare(
|
||
`SELECT 1 FROM ec_host_operations
|
||
WHERE id=? AND json_extract(payload,'$.recoveryRebuildOperationId') IS NOT NULL`
|
||
)
|
||
.get(operation.id);
|
||
if (
|
||
operation?.kind === 'instance.cancel-create' &&
|
||
!['completed', 'failed'].includes(operation.state) &&
|
||
(!host.inventory.capabilities?.includes('instance.cancel-create') ||
|
||
(recoveryCleanup &&
|
||
!host.inventory.capabilities?.includes(unhealthyRecoveryCapability)))
|
||
) {
|
||
hold(
|
||
job,
|
||
hostAgentUpdatesEnabled(host.inventory)
|
||
? 'Waiting for the host agent update. New command delivery is paused so its current batch can finish and the automatic updater can run. Cleanup resumes after the updated agent connects.'
|
||
: 'Update the host agent to support creation cancellation: rerun the current installer on this host. Automatic updates are not enabled. New command delivery is paused; cleanup and port release await the updated agent.'
|
||
);
|
||
return;
|
||
}
|
||
if (!operation || operation.state === 'unknown') {
|
||
hold(
|
||
job,
|
||
'Agent result is unknown. Reconciling the original operation; no replacement or repeated command will be sent.'
|
||
);
|
||
return;
|
||
}
|
||
if (operation.state === 'failed') {
|
||
applyIncompleteRebuildReceipt(job, operation);
|
||
if (
|
||
job.action === 'destroy' &&
|
||
scalingIntent(job.id) &&
|
||
(operation.result as { autoscalingInvalidated?: boolean } | null)
|
||
?.autoscalingInvalidated === true
|
||
) {
|
||
db().transaction(() => {
|
||
db()
|
||
.prepare(
|
||
"UPDATE server_deployments SET state='cancelled',stage='Automatic removal withdrawn; new empty interval required',finished_at=?,error=NULL WHERE id=?"
|
||
)
|
||
.run(new Date().toISOString(), job.id);
|
||
recordProgress(job.id, {
|
||
blocker: null,
|
||
event: {
|
||
kind: 'outcome',
|
||
message:
|
||
'Automatic removal was invalidated before mutation. A new decision and empty interval are required.',
|
||
details: { state: 'cancelled' }
|
||
}
|
||
});
|
||
})();
|
||
return;
|
||
}
|
||
if (migration && !job.recovery_step) {
|
||
if (Date.now() - Date.parse(operation.updatedAt) >= 60_000) {
|
||
db().transaction(() => {
|
||
db()
|
||
.prepare('DELETE FROM restart_rollout WHERE operation_id=?')
|
||
.run(job.id);
|
||
db()
|
||
.prepare(
|
||
'UPDATE native_provisioning SET 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);
|
||
})();
|
||
}
|
||
hold(
|
||
job,
|
||
`${operation.error ?? 'Destination creation failed before mutation.'} Retrying after fresh capacity checks.`
|
||
);
|
||
return;
|
||
}
|
||
// Failed commands either never mutated or have a conclusively
|
||
// reconciled partial rebuild above. Ambiguous outcomes retain capacity.
|
||
db().transaction(() => {
|
||
db().prepare('DELETE FROM restart_rollout WHERE operation_id=?').run(job.id);
|
||
const files = storedFiles(job);
|
||
for (const file of files)
|
||
db()
|
||
.prepare(
|
||
"UPDATE staged_file_changes SET state='pending' WHERE id=? AND state='applying'"
|
||
)
|
||
.run(file.id);
|
||
finishDeployment(job.id, {
|
||
ok: false,
|
||
message: operation.error ?? 'Agent operation failed before mutation.'
|
||
});
|
||
})();
|
||
return;
|
||
}
|
||
if (operation.state !== 'completed') {
|
||
const blocked = operation.result as { message?: string } | null;
|
||
hold(job, blocked?.message ?? `Agent operation ${operation.state}`);
|
||
return;
|
||
}
|
||
if (job.action === 'destroy') {
|
||
db().transaction(() => {
|
||
deploymentStage(job.id, REMOVING_SERVER, {
|
||
phase: 'cleanup',
|
||
checkpoint: {
|
||
key: 'container-action',
|
||
label: 'Agent confirmed instance and container removal',
|
||
at: operation.updatedAt
|
||
}
|
||
});
|
||
db()
|
||
.prepare('UPDATE native_instances SET retired_at=? WHERE id=?')
|
||
.run(new Date().toISOString(), instance!.id);
|
||
releaseHostPorts(instance!.id);
|
||
db().prepare('DELETE FROM servers WHERE name=?').run(job.server);
|
||
releaseServerKey(job.server);
|
||
recordProgress(job.id, {
|
||
checkpoint: {
|
||
key: 'cleanup',
|
||
label: 'Instance retired and port reservations released'
|
||
}
|
||
});
|
||
complete(job);
|
||
finishCreationCancellation(job.id);
|
||
})();
|
||
return;
|
||
}
|
||
if (operation.kind === 'instance.rebuild' && instance.generation === job.generation) {
|
||
db()
|
||
.prepare('UPDATE native_instances SET generation=? WHERE id=? AND generation=?')
|
||
.run(job.generation + 1, instance.id, job.generation);
|
||
instance = nativeInstance(job.server)!;
|
||
}
|
||
applyReceipt(job, (operation.result ?? {}) as Record<string, unknown>);
|
||
db()
|
||
.prepare(
|
||
'UPDATE native_provisioning SET applied_at=COALESCE(applied_at,?) WHERE deployment_id=?'
|
||
)
|
||
.run(operation.updatedAt, job.id);
|
||
if (job.state !== 'awaiting-health') {
|
||
db()
|
||
.prepare(
|
||
"UPDATE server_deployments SET state='awaiting-health',error=NULL,finished_at=NULL WHERE id=?"
|
||
)
|
||
.run(job.id);
|
||
const action = job.recovery_step ? `recovery ${job.recovery_step}` : job.action;
|
||
recordProgress(job.id, {
|
||
command: {
|
||
state: 'accepted',
|
||
message: `Agent confirmed ${action}; verifying the resulting instance.`
|
||
},
|
||
checkpoint: {
|
||
key: job.recovery_step ? `recovery-${job.recovery_step}` : 'container-action',
|
||
label: `Agent confirmed ${action}`,
|
||
at: operation.updatedAt
|
||
}
|
||
});
|
||
}
|
||
const checks: CheckInput[] = [];
|
||
const check = (
|
||
value: CheckInput,
|
||
phase: import('../../provisioning-detail.js').ProvisioningPhase
|
||
) => {
|
||
checks.push({ ...value, checkedAt: new Date().toISOString() });
|
||
recordProgress(job.id, {
|
||
checks,
|
||
...(value.result === 'pass'
|
||
? { checkpoint: { key: phase, label: value.label } }
|
||
: {})
|
||
});
|
||
};
|
||
deploymentStage(job.id, 'Checking container state with host agent', {
|
||
phase: job.action === 'stop' ? 'verify-stopped' : 'verify-container',
|
||
blocker: null,
|
||
checks
|
||
});
|
||
const verificationIdentity = monitoringIdentity(job.server);
|
||
const health = await nativeRequest<{ healthy: boolean; state: string }>(
|
||
instance,
|
||
'instance.inspect'
|
||
);
|
||
const containerObservedAt = new Date().toISOString();
|
||
if (!stillActive(job)) return;
|
||
if (job.action === 'stop') {
|
||
const stopped = ['stopped', 'exited', 'created', 'dead'].includes(health.state);
|
||
check(
|
||
{
|
||
key: 'stopped',
|
||
label: 'Stopped container verified',
|
||
source: 'agent',
|
||
result: stopped ? 'pass' : 'waiting',
|
||
actual: health.state
|
||
},
|
||
'verify-stopped'
|
||
);
|
||
if (!stopped) {
|
||
hold(job, 'Waiting for the agent to verify a stopped container.');
|
||
return;
|
||
}
|
||
db().transaction(() => {
|
||
db()
|
||
.prepare(
|
||
'UPDATE servers SET container_state=?,last_players=NULL,last_roster=NULL WHERE name=?'
|
||
)
|
||
.run(health.state, job.server);
|
||
db()
|
||
.prepare(
|
||
`UPDATE leaderboard_matches SET status='interrupted', ended_at=?
|
||
WHERE server=? AND status='active' AND server_deleted_at IS NULL`
|
||
)
|
||
.run(containerObservedAt, job.server);
|
||
complete(job);
|
||
})();
|
||
return;
|
||
}
|
||
check(
|
||
{
|
||
key: 'container',
|
||
label: 'Container readiness verified',
|
||
source: 'agent',
|
||
result: health.healthy ? 'pass' : 'waiting',
|
||
actual: health.state
|
||
},
|
||
'verify-container'
|
||
);
|
||
if (!health.healthy) {
|
||
if (
|
||
health.state === 'unhealthy' &&
|
||
(job.action === 'create' || job.action === 'rebuild')
|
||
)
|
||
queueUnhealthyRecovery(job, instance);
|
||
else hold(job, `Waiting for container readiness: ${health.state}`);
|
||
return;
|
||
}
|
||
deploymentStage(job.id, 'Checking game status', { phase: 'verify-game', blocker: null });
|
||
const status = await probeServerStatus(job.server, operation.updatedAt);
|
||
if (!stillActive(job)) return;
|
||
check(
|
||
{
|
||
key: 'game',
|
||
label: 'Fresh game status verified',
|
||
source: status.ok ? status.source : null,
|
||
result: status.ok ? 'pass' : 'unknown',
|
||
actual: status.ok ? status.players : null,
|
||
message: status.ok ? null : status.reason
|
||
},
|
||
'verify-game'
|
||
);
|
||
if (!status.ok) {
|
||
hold(job, `Waiting for game status: ${status.reason}`);
|
||
return;
|
||
}
|
||
if (job.action === 'create' || job.action === 'rebuild') {
|
||
deploymentStage(job.id, 'Checking selected engine map', {
|
||
phase: 'verify-map',
|
||
blocker: null
|
||
});
|
||
const recipe = db()
|
||
.prepare(
|
||
'SELECT expected_map AS expectedMap FROM server_creation_jobs WHERE deployment_id=?'
|
||
)
|
||
.get(job.id) as { expectedMap: string };
|
||
check(
|
||
{
|
||
key: 'map',
|
||
label: 'Selected engine map verified',
|
||
source: status.source,
|
||
result: status.map === recipe.expectedMap ? 'pass' : 'waiting',
|
||
expected: recipe.expectedMap,
|
||
actual: status.map
|
||
},
|
||
'verify-map'
|
||
);
|
||
if (status.map !== recipe.expectedMap) {
|
||
hold(
|
||
job,
|
||
`Waiting for selected map ${recipe.expectedMap}; observed ${status.map ?? 'unknown'}.`
|
||
);
|
||
return;
|
||
}
|
||
deploymentStage(job.id, 'Checking production Plutonium release', {
|
||
phase: 'verify-plutonium',
|
||
blocker: null
|
||
});
|
||
await checkRelease();
|
||
if (!stillActive(job)) return;
|
||
const target = freshRelease();
|
||
checks.push({
|
||
key: 'plutonium-release',
|
||
label: 'Production Plutonium release',
|
||
source: 'manifest',
|
||
result: target === null ? 'unknown' : 'pass',
|
||
actual: target
|
||
});
|
||
recordProgress(job.id, { checks });
|
||
deploymentStage(job.id, 'Checking running Plutonium revision', {
|
||
phase: 'verify-plutonium',
|
||
blocker: null
|
||
});
|
||
const revision = await probeVersion(job.server, { operationId: job.id });
|
||
if (!stillActive(job)) return;
|
||
check(
|
||
{
|
||
key: 'plutonium-version',
|
||
label: 'Plutonium revision verified',
|
||
source: 'rcon',
|
||
expected: target,
|
||
actual: revision,
|
||
result:
|
||
revision === null || target === null
|
||
? 'unknown'
|
||
: revision < target
|
||
? 'fail'
|
||
: 'pass'
|
||
},
|
||
'verify-plutonium'
|
||
);
|
||
if (revision === null || target === null) {
|
||
hold(job, 'Waiting to verify the Plutonium runtime revision.');
|
||
return;
|
||
}
|
||
if (revision < target) {
|
||
finishDeployment(job.id, {
|
||
ok: false,
|
||
message: `Running Plutonium r${revision} is older than production r${target}.`
|
||
});
|
||
return;
|
||
}
|
||
}
|
||
const completed = db().transaction(() => {
|
||
if (
|
||
!stillActive(job) ||
|
||
!handoffVerifiedHealth(
|
||
job.server,
|
||
verificationIdentity,
|
||
{ state: health.state, observedAt: containerObservedAt },
|
||
status
|
||
)
|
||
)
|
||
return false;
|
||
recordProgress(job.id, { blocker: null, checks });
|
||
complete(job);
|
||
return true;
|
||
})();
|
||
if (!completed)
|
||
hold(job, 'Waiting for current health evidence before completing verification.');
|
||
return;
|
||
}
|
||
// A queued activation from a previous controller release may still say restart.
|
||
// Upgrade only intent that has never submitted a host command.
|
||
if (
|
||
job.action === 'restart' &&
|
||
job.state === 'waiting' &&
|
||
!job.submitted_at &&
|
||
filesFor(job).length
|
||
) {
|
||
db().prepare("UPDATE server_deployments SET action='rebuild' WHERE id=?").run(job.id);
|
||
job.action = 'rebuild';
|
||
}
|
||
const reason = eligibility(job);
|
||
if (reason) {
|
||
hold(job, reason);
|
||
return;
|
||
}
|
||
const overrideSafety = !!safetyOverride(job.id);
|
||
if (overrideSafety && !host.inventory.capabilities?.includes(safetyOverrideCapability)) {
|
||
hold(job, 'Update this host’s agent before using admin override.');
|
||
return;
|
||
}
|
||
if (job.action === 'destroy' || job.action === 'rebuild') {
|
||
const beta = activeBetaRemovalBlocker(
|
||
job.server,
|
||
job.action === 'rebuild' ? job.id : undefined
|
||
);
|
||
if (beta) {
|
||
hold(job, beta);
|
||
return;
|
||
}
|
||
}
|
||
const files = job.stage_operation_id ? storedFiles(job) : filesFor(job);
|
||
if (
|
||
job.action === 'rebuild' &&
|
||
files.length &&
|
||
!host.inventory.capabilities?.includes('instance.rebuild-files')
|
||
) {
|
||
hold(job, 'Update the host agent to support beta and campaign file rebuilds.');
|
||
return;
|
||
}
|
||
if (job.action === 'rebuild' && !recipeSummaryForDeployment(job.id)) {
|
||
const server = getServer(job.server);
|
||
if (!server?.map_slug) throw new Error('A saved map is required before rebuilding.');
|
||
const recipe: NativeRecipe = {
|
||
...captureRecipePlan(server.game as 't5' | 't6', server.map_slug),
|
||
image: instance.image,
|
||
native: {
|
||
hostId: instance.host_id,
|
||
libraryId: instance.library_id,
|
||
image: instance.image,
|
||
bindAddress: instance.bind_address
|
||
}
|
||
};
|
||
const stored = recipeStorage(recipe);
|
||
db().transaction(() => {
|
||
db()
|
||
.prepare(
|
||
'INSERT INTO server_creation_jobs(deployment_id,server,port,game_mode,map_slug,expected_map) VALUES(?,?,?,?,?,?)'
|
||
)
|
||
.run(
|
||
job.id,
|
||
job.server,
|
||
instance!.port,
|
||
'zombie',
|
||
recipe.mapSlug,
|
||
recipe.expectedMap
|
||
);
|
||
db()
|
||
.prepare(
|
||
'INSERT INTO server_launch_recipes(deployment_id,recipe,recipe_summary) VALUES(?,?,?)'
|
||
)
|
||
.run(job.id, stored.recipe, stored.recipeSummary);
|
||
})();
|
||
}
|
||
if (files.length && !job.stage_operation_id) {
|
||
db().transaction(() => {
|
||
const stage = queueHostOperationSerialized(
|
||
instance.host_id,
|
||
'files.stage',
|
||
Buffer.from(
|
||
JSON.stringify({
|
||
stageId: job.id,
|
||
files: files.map((file) => ({
|
||
...file,
|
||
...(file.sha256
|
||
? { contentSha256: file.sha256 }
|
||
: { contentBase64: null })
|
||
}))
|
||
})
|
||
),
|
||
{
|
||
instanceId: instance.id,
|
||
generation: instance.generation,
|
||
idempotencyKey: `${job.id}:stage`
|
||
},
|
||
{ stageId: job.id }
|
||
);
|
||
db()
|
||
.prepare(
|
||
'UPDATE native_provisioning SET stage_operation_id=?,files_json=? WHERE deployment_id=?'
|
||
)
|
||
.run(stage.id, JSON.stringify(files.map(fileMetadata)), job.id);
|
||
})();
|
||
hold(job, 'Staging and verifying files on the host.');
|
||
return;
|
||
}
|
||
if (job.stage_operation_id) {
|
||
const stage = getHostOperationStatus(job.stage_operation_id);
|
||
if (stage?.state === 'failed') {
|
||
finishDeployment(job.id, { ok: false, message: stage.error ?? 'File staging failed.' });
|
||
return;
|
||
}
|
||
if (stage?.state !== 'completed') {
|
||
hold(job, 'Waiting for verified remote file staging.');
|
||
return;
|
||
}
|
||
if (
|
||
files.some(
|
||
(file) =>
|
||
(
|
||
db()
|
||
.prepare('SELECT state FROM staged_file_changes WHERE id=?')
|
||
.get(file.id) as { state: string } | undefined
|
||
)?.state !== 'pending'
|
||
)
|
||
) {
|
||
finishDeployment(job.id, {
|
||
ok: false,
|
||
message: 'Staged file request was cancelled or changed before activation.'
|
||
});
|
||
return;
|
||
}
|
||
}
|
||
const capacityBlocker = concurrencyBlocker(job.id, job.action, instance.host_id);
|
||
if (capacityBlocker) {
|
||
hold(job, capacityBlocker);
|
||
return;
|
||
}
|
||
const settings = fleetSettings();
|
||
const count = (actions: string[]) =>
|
||
(
|
||
db()
|
||
.prepare(
|
||
`SELECT COUNT(*) AS n FROM server_deployments WHERE id != ? AND server_deleted_at IS NULL
|
||
AND action IN (${actions.map(() => '?').join(',')}) AND (state IN ('running','awaiting-health') OR ${committedRebalanceStartup})`
|
||
)
|
||
.get(job.id, ...actions) as { n: number }
|
||
).n;
|
||
if (
|
||
(job.action === 'create' || job.action === 'rebuild') &&
|
||
count(['create', 'rebuild']) + otherPreparations(job.id) >= settings.createConcurrency
|
||
) {
|
||
deploymentStage(job.id, 'Waiting for startup capacity.', {
|
||
phase: 'startup-queue',
|
||
blocker: {
|
||
code: 'startup-capacity',
|
||
message: 'Waiting for startup capacity.',
|
||
automatic: true
|
||
}
|
||
});
|
||
return;
|
||
}
|
||
if (job.action === 'destroy' && count(['destroy']) >= settings.destroyConcurrency) {
|
||
hold(job, 'Waiting for destruction capacity.');
|
||
return;
|
||
}
|
||
deploymentStage(job.id, 'Checking host eligibility', { phase: 'eligibility', blocker: null });
|
||
if (job.action === 'create' || migration)
|
||
choosePlacement({
|
||
hostId: instance.host_id,
|
||
libraryId: instance.library_id,
|
||
image: instance.image,
|
||
bindAddress: instance.bind_address,
|
||
excludeInstanceId: instance.id,
|
||
thorough: false,
|
||
dispatch: true
|
||
});
|
||
if (job.action === 'create' || job.action === 'rebuild')
|
||
assertReadyLibrary(instance.host_id, instance.library_id);
|
||
const checkOccupancy = async (): Promise<boolean> => {
|
||
if (!stillActive(job)) return false;
|
||
if (overrideSafety) return true;
|
||
// Fast controller check keeps occupied lobbies out of the operation slots.
|
||
// The agent repeats this immediately before its local mutation.
|
||
deploymentStage(job.id, 'Checking container and player occupancy before submission', {
|
||
phase: 'eligibility',
|
||
blocker: null
|
||
});
|
||
const health = await nativeRequest<{ state: string; healthy?: boolean }>(
|
||
instance,
|
||
'instance.inspect'
|
||
);
|
||
if (!stillActive(job)) return false;
|
||
if (job.action === 'rebuild' && !files.length && health.state === 'unhealthy') {
|
||
if (
|
||
!getHost(instance.host_id)?.inventory.capabilities?.includes(
|
||
unhealthyRecoveryCapability
|
||
)
|
||
) {
|
||
hold(
|
||
job,
|
||
'Update this host’s agent to recover unhealthy containers during provisioning.'
|
||
);
|
||
return false;
|
||
}
|
||
return true;
|
||
}
|
||
if (!['stopped', 'exited', 'created', 'dead', 'missing'].includes(health.state)) {
|
||
const occupancy = await probePlayerEvidence(job.server);
|
||
if (!stillActive(job)) return false;
|
||
const recoveredWatchdog =
|
||
health.healthy &&
|
||
occupancy.players !== null &&
|
||
db()
|
||
.prepare(
|
||
`SELECT 1 FROM provisioning_sources s
|
||
JOIN watchdog_incidents i ON i.id=s.source_id JOIN server_deployments d ON d.id=s.operation_id
|
||
WHERE s.operation_id=? AND s.kind='watchdog' AND i.kind='health' AND d.triggered_by LIKE 'watchdog:%'`
|
||
)
|
||
.get(job.id);
|
||
if (recoveredWatchdog && occupancy.players !== null) {
|
||
resolveRecoveredHealthOperation(job.id, occupancy.players, occupancy.source);
|
||
return false;
|
||
}
|
||
if (occupancy.players !== 0) {
|
||
if (job.action === 'destroy' && scalingIntent(job.id)) cancelRemoval(job.id, false);
|
||
else
|
||
hold(
|
||
job,
|
||
occupancy.players !== null
|
||
? `Waiting for ${occupancy.players} player(s) to leave.`
|
||
: 'Waiting for a fresh local player count.'
|
||
);
|
||
return false;
|
||
}
|
||
}
|
||
return true;
|
||
};
|
||
if (
|
||
!migration &&
|
||
job.action !== 'create' &&
|
||
job.action !== 'start' &&
|
||
!(await checkOccupancy())
|
||
)
|
||
return;
|
||
const current = db()
|
||
.prepare('SELECT state FROM server_deployments WHERE id=? AND server_deleted_at IS NULL')
|
||
.get(job.id) as { state: string } | undefined;
|
||
if (
|
||
!current ||
|
||
!['queued', 'waiting', 'interrupted'].includes(current.state) ||
|
||
eligibility(job)
|
||
)
|
||
return;
|
||
let prepared: PreparedNativeLaunch | undefined;
|
||
let preparedKey: string | null = null;
|
||
let preparedPassword: string | undefined;
|
||
try {
|
||
if (job.action === 'create' || job.action === 'rebuild') {
|
||
if (
|
||
count(['create', 'rebuild']) + otherPreparations(job.id) >=
|
||
fleetSettings().createConcurrency
|
||
) {
|
||
hold(job, 'Waiting for startup capacity.');
|
||
return;
|
||
}
|
||
const capacityBlocker = concurrencyBlocker(job.id, job.action, instance.host_id);
|
||
if (capacityBlocker) {
|
||
hold(job, capacityBlocker);
|
||
return;
|
||
}
|
||
preparingJobs.add(job.id);
|
||
preparedKey = getServerKey(job.server) ?? null;
|
||
if (!preparedKey) throw new Error('The saved server key was removed.');
|
||
const frozenRecipeId = migration && isRebalanceMigration(job.id) ? job.id : undefined;
|
||
const prior = (
|
||
frozenRecipeId
|
||
? { format: 1, ...recipeSummaryForDeployment(job.id)!.recipe }
|
||
: recipeForDeployment(job.id)!.recipe
|
||
) as NativeRecipe;
|
||
deploymentStage(job.id, 'Preparing launch files for the host agent', {
|
||
phase: 'prepare',
|
||
checkpoint: { key: 'startup-queue', label: 'Startup capacity available' }
|
||
});
|
||
preparedPassword =
|
||
(files.length ? runtimePasswordForServer(job.server) : undefined) ||
|
||
config.rcon.password ||
|
||
randomBytes(32).toString('hex');
|
||
prepared = await prepareNativeLaunch({
|
||
game: prior.game,
|
||
mapSlug: prior.mapSlug,
|
||
native: prior.native,
|
||
instance,
|
||
serverKey: preparedKey,
|
||
rconPassword: preparedPassword,
|
||
bankToken: bankCredential(instance.id),
|
||
rebuild: job.action === 'rebuild' && !migration,
|
||
safetyOverride: overrideSafety,
|
||
unhealthyRecovery:
|
||
!files.length &&
|
||
host.inventory.capabilities?.includes(unhealthyRecoveryCapability),
|
||
activationId: files.length ? job.id : undefined,
|
||
frozenRecipeId
|
||
});
|
||
if (job.action === 'rebuild' && !migration && !(await checkOccupancy())) return;
|
||
}
|
||
const commit = 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 ||
|
||
!['queued', 'waiting', 'interrupted'].includes(state.state) ||
|
||
eligibility(job)
|
||
)
|
||
return;
|
||
const live = nativeInstance(job.server);
|
||
if (live?.id !== instance!.id || live.generation !== instance!.generation)
|
||
throw new Error('Instance changed during preflight.');
|
||
const liveHost = getHost(instance!.host_id);
|
||
if (!liveHost?.connected || liveHost.mode === 'disabled') {
|
||
hold(job, 'Host is unavailable; waiting for reconnection.');
|
||
return;
|
||
}
|
||
// Confirmation can arrive while preflight or file preparation is awaiting I/O.
|
||
if (!!safetyOverride(job.id) !== overrideSafety) return;
|
||
if (
|
||
overrideSafety &&
|
||
!liveHost.inventory.capabilities?.includes(safetyOverrideCapability)
|
||
) {
|
||
hold(job, 'Update this host’s agent before using admin override.');
|
||
return;
|
||
}
|
||
if (job.action === 'destroy' || job.action === 'rebuild') {
|
||
const beta = activeBetaRemovalBlocker(
|
||
job.server,
|
||
job.action === 'rebuild' ? job.id : undefined
|
||
);
|
||
if (beta) {
|
||
hold(job, beta);
|
||
return;
|
||
}
|
||
}
|
||
if (
|
||
JSON.stringify(filesFor(job).map(fileMetadata)) !==
|
||
JSON.stringify(files.map(fileMetadata))
|
||
) {
|
||
finishDeployment(job.id, {
|
||
ok: false,
|
||
message:
|
||
'File requests changed during preflight. Review and submit a fresh activation.'
|
||
});
|
||
return;
|
||
}
|
||
const capacityBlocker = concurrencyBlocker(job.id, job.action, instance!.host_id);
|
||
if (capacityBlocker) {
|
||
hold(job, capacityBlocker);
|
||
return;
|
||
}
|
||
// Async occupancy checks can overlap another job claiming the final slot.
|
||
if (
|
||
(job.action === 'create' || job.action === 'rebuild') &&
|
||
count(['create', 'rebuild']) + otherPreparations(job.id) >=
|
||
fleetSettings().createConcurrency
|
||
) {
|
||
deploymentStage(job.id, 'Waiting for startup capacity.', {
|
||
phase: 'startup-queue',
|
||
blocker: {
|
||
code: 'startup-capacity',
|
||
message: 'Waiting for startup capacity.',
|
||
automatic: true
|
||
}
|
||
});
|
||
return;
|
||
}
|
||
if (
|
||
job.action === 'destroy' &&
|
||
count(['destroy']) >= fleetSettings().destroyConcurrency
|
||
) {
|
||
hold(job, 'Waiting for destruction capacity.');
|
||
return;
|
||
}
|
||
if (prepared) {
|
||
// Compare the same runtime config bytes while rechecking current promoted files.
|
||
const expected =
|
||
migration && isRebalanceMigration(job.id)
|
||
? { format: 1 as const, ...recipeSummaryForDeployment(job.id)!.recipe }
|
||
: captureRecipePlan(
|
||
prepared.recipe.game,
|
||
prepared.recipe.mapSlug,
|
||
preparedPassword
|
||
);
|
||
if (files.length) applyActivationFiles(expected, job.id);
|
||
if (
|
||
getServerKey(job.server) !== preparedKey ||
|
||
!sameLaunchFiles(prepared.recipe, expected)
|
||
) {
|
||
hold(
|
||
job,
|
||
'Launch files or credentials changed during preparation; preparing the current selection next pass.'
|
||
);
|
||
return;
|
||
}
|
||
assertReadyLibrary(instance!.host_id, instance!.library_id);
|
||
if (job.action === 'create' || migration)
|
||
choosePlacement({
|
||
hostId: instance!.host_id,
|
||
libraryId: instance!.library_id,
|
||
image: instance!.image,
|
||
bindAddress: instance!.bind_address,
|
||
excludeInstanceId: instance!.id,
|
||
thorough: false,
|
||
dispatch: true
|
||
});
|
||
}
|
||
if (job.action === 'start' || job.action === 'restart' || job.action === 'rebuild') {
|
||
db()
|
||
.prepare(
|
||
`INSERT INTO restart_rollout(server,requested_at,operation_id) VALUES(?,?,?) ON CONFLICT(server)
|
||
DO UPDATE SET requested_at=excluded.requested_at,operation_id=excluded.operation_id,healthy_at=NULL,error=NULL`
|
||
)
|
||
.run(job.server, new Date().toISOString(), job.id);
|
||
}
|
||
recordProgress(job.id, {
|
||
blocker: null,
|
||
checkpoint: {
|
||
key: 'eligibility',
|
||
label: overrideSafety
|
||
? 'Host eligibility checked; player safety checks bypassed by admin override'
|
||
: 'Host eligibility and safety checks passed'
|
||
}
|
||
});
|
||
const payload: Record<string, unknown> = files.length ? { stageId: job.id } : {};
|
||
if (overrideSafety) payload.safetyOverride = true;
|
||
if (job.action === 'destroy') {
|
||
const authorization = removalAuthorization(job.id, instance!.host_id);
|
||
if (authorization) {
|
||
if (overrideSafety)
|
||
throw new Error('Autoscaling never bypasses player safety.');
|
||
payload.autoscaling = authorization;
|
||
}
|
||
}
|
||
if (prepared) {
|
||
deploymentStage(job.id, 'Preparing launch files for the host agent', {
|
||
phase: 'prepare',
|
||
recipeStatus: 'attempted',
|
||
checkpoint: { key: 'startup-queue', label: 'Startup capacity available' }
|
||
});
|
||
for (const blob of prepared.blobs ?? []) putFileBlob(blob.content);
|
||
const recipe = prepared.recipe;
|
||
db()
|
||
.prepare(
|
||
'UPDATE server_launch_recipes SET recipe=CAST(? AS TEXT),recipe_summary=? WHERE deployment_id=?'
|
||
)
|
||
.run(
|
||
Buffer.from(
|
||
prepared.recipeJson.buffer,
|
||
prepared.recipeJson.byteOffset,
|
||
prepared.recipeJson.byteLength
|
||
),
|
||
prepared.recipeSummary,
|
||
job.id
|
||
);
|
||
if (job.action === 'create')
|
||
db()
|
||
.prepare(
|
||
`INSERT INTO servers(name,game,map,game_mode,map_slug,host_dir,config_file,game_id,image,game_port,created_by,container_state)
|
||
VALUES(?,?,?,'zombie',?,?,?,?,?,?,'native','starting')`
|
||
)
|
||
.run(
|
||
job.server,
|
||
recipe.game,
|
||
recipe.expectedMap,
|
||
recipe.mapSlug,
|
||
serverDir(job.server),
|
||
recipe.configFile,
|
||
recipe.game === 't6' ? 't6zm' : 't5sp',
|
||
recipe.image,
|
||
instance!.port
|
||
);
|
||
for (const file of recipe.files.filter(
|
||
(file) =>
|
||
!files.some(
|
||
(change) =>
|
||
change.path ===
|
||
posix.join(relDestForKind(file.kind, recipe.game), file.path)
|
||
)
|
||
))
|
||
recordScript(
|
||
job.server,
|
||
file.path,
|
||
file.versionId,
|
||
file.sha256,
|
||
file.size,
|
||
'native-create'
|
||
);
|
||
recordProgress(job.id, {
|
||
checkpoint: {
|
||
key: 'prepare',
|
||
label: 'Launch files selected and command payload prepared'
|
||
}
|
||
});
|
||
}
|
||
const attempt = migration
|
||
? (
|
||
db()
|
||
.prepare(
|
||
`SELECT COUNT(*) AS n FROM ec_host_operations
|
||
WHERE host_id=? AND instance_id=? AND generation=? AND kind='instance.create'`
|
||
)
|
||
.get(instance!.host_id, instance!.id, instance!.generation) as {
|
||
n: number;
|
||
}
|
||
).n
|
||
: 0;
|
||
const operationOptions = {
|
||
instanceId: instance!.id,
|
||
generation: instance!.generation,
|
||
idempotencyKey: migration ? `${job.id}:migration-create:${attempt}` : job.id
|
||
};
|
||
const kind = `instance.${migration ? 'create' : job.action === 'destroy' ? 'remove' : job.action}`;
|
||
const operation = prepared
|
||
? queueHostOperationSerialized(
|
||
instance!.host_id,
|
||
kind,
|
||
Buffer.from(
|
||
prepared.payloadJson.buffer,
|
||
prepared.payloadJson.byteOffset,
|
||
prepared.payloadJson.byteLength
|
||
),
|
||
operationOptions,
|
||
{
|
||
name: instance!.server,
|
||
game: prepared.recipe.game,
|
||
libraryId: instance!.library_id
|
||
}
|
||
)
|
||
: queueHostOperation(instance!.host_id, kind, payload, operationOptions);
|
||
db()
|
||
.prepare('UPDATE native_provisioning SET operation_id=? WHERE deployment_id=?')
|
||
.run(operation.id, job.id);
|
||
for (const file of files)
|
||
db()
|
||
.prepare(
|
||
"UPDATE staged_file_changes SET state='applying' WHERE id=? AND state='pending'"
|
||
)
|
||
.run(file.id);
|
||
db()
|
||
.prepare("UPDATE server_deployments SET state='running',submitted_at=? WHERE id=?")
|
||
.run(new Date().toISOString(), job.id);
|
||
deploymentStage(job.id, `Waiting for host agent to acknowledge ${job.action}`, {
|
||
phase: 'container-action',
|
||
blocker: null,
|
||
command: {
|
||
state: 'attempted',
|
||
message: `Durable agent operation ${operation.id}; reconciliation always uses this ID.`
|
||
}
|
||
});
|
||
});
|
||
if (prepared) await commitPreparedLaunch(commit);
|
||
else commit();
|
||
} finally {
|
||
preparingJobs.delete(job.id);
|
||
}
|
||
}
|