@@ -2,7 +2,7 @@ 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 } from './concurrency.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' ;
@@ -39,12 +39,15 @@ 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 }
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 {
@@ -242,7 +245,7 @@ function storedFiles(job: Job): FilePlan[] {
/** 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 ) return ;
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 ;
@@ -276,7 +279,7 @@ function applyReceipt(job: Job, result: Record<string, unknown>) {
/** 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
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 ;
@@ -309,6 +312,13 @@ export function reconcileNativeProvisioningReceipt(id: string): void {
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 ) ;
@@ -318,7 +328,7 @@ export function reconcileNativeProvisioningReceipt(id: string): void {
finishCreationCancellation ( job . id ) ;
return ;
}
if ( job . action === 'rebuild' && ! hostMigration ( id ) && instance . generation === job . generation ) {
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.
@@ -328,7 +338,7 @@ export function reconcileNativeProvisioningReceipt(id: string): void {
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 && ! ( job . action === 'rebuild' && instance . generation === job . generation + 1 ) ) return ;
} 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 ) ;
}
@@ -342,10 +352,78 @@ 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 ) ;
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 ;
}
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 > {
ensureHostSchema ( db ( ) ) ;
@@ -369,7 +447,7 @@ export async function processNativeProvisioning(eligibility: (operation: { id: s
}
}
reconcileHostDrains ( ) ;
const jobs = db ( ) . prepare ( ` SELECT d.*,n.instance_id,n.generation,n.stage_operation_id,n.operation_id,n.applied_at,q.selection
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 [ ] ;
@@ -485,7 +563,8 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
job . instance_id = instance . id ; job . generation = instance . generation ;
}
const migration = hostMigration ( job . id ) ;
if ( ! instance || instance . id ! == job . instance_id || ( instance . generation !== job . generation && ! ( job . action === 'rebuild' && ! migration && instance . generation === job . generation + 1 ) ) ) {
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 ) {
@@ -499,7 +578,10 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
}
if ( job . operation_id ) {
const operation = getHostOperationStatus ( job . operation_id ) ;
if ( operation ? . kind === 'instance.cancel-create' && ! [ 'completed' , 'failed' ] . includes ( operation . state ) && ! host . inventory . capabilities ? . includes ( 'instance.cancel-create' ) ) {
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 ;
@@ -514,7 +596,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
} ) ( ) ;
return ;
}
if ( migration ) {
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 ) ;
@@ -552,15 +634,17 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
} ) ( ) ;
return ;
}
if ( job . action === 'rebuild' && ! migration && instance . generation === job . generation ) {
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 ) ;
recordProgress ( job . id , { command : { state : 'accepted' , message : ` Agent confirmed ${ job . action } ; verifying the resulting instance. ` } ,
checkpoint : { key : 'container-action ' , label : ` Agent confirmed ${ job . action } ` , at : operation.updatedAt } } ) ;
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 ) = > {
@@ -580,7 +664,11 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
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 ) { hold ( job , ` Waiting for container readiness: ${ health . state } ` ) ; return ; }
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 ;
@@ -669,6 +757,12 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
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' && 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 ;
@@ -708,7 +802,8 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
preparedPassword = config . rcon . password || randomBytes ( 32 ) . toString ( 'hex' ) ;
prepared = await prepareNativeLaunch ( { game : prior.game , mapSlug : prior.mapSlug , native : prior.native ,
instance , serverKey : preparedKey , rconPassword : preparedPassword ,
rebuild : job.action === 'rebuild' && ! migration , safetyOverride : overrideSafety , frozenRecipeId } ) ;
rebuild : job.action === 'rebuild' && ! migration , safetyOverride : overrideSafety ,
unhealthyRecovery : host.inventory.capabilities?.includes ( unhealthyRecoveryCapability ) , frozenRecipeId } ) ;
if ( job . action === 'rebuild' && ! migration && ! await checkOccupancy ( ) ) return ;
}
const commit = db ( ) . transaction ( ( ) = > {