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

Keep native collection launches responsive

Browse Source
This commit is contained in:
josh
2026-10-05 10:48:38 -04:00
parent dbd2940d29
commit 1405114a14
26 changed files with 998 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/hosts/index.ts
+80 -23
View File
@@ -30,7 +30,7 @@ function decommissionRow(hostId: string): DecommissionRow | undefined {
}
function presentDecommission(row: DecommissionRow | undefined): HostDecommission | null {
return row ? { operationId: row.operation_id, stage: row.stage, requestedAt: row.requested_at, updatedAt: row.updated_at,
error: row.error, retryable: row.stage === 'failed' && getHostOperation(row.operation_id)?.state === 'failed' } : null;
error: row.error, retryable: row.stage === 'failed' && getHostOperationStatus(row.operation_id)?.state === 'failed' } : null;
}
function hostFromRow(row: HostRow): Host {
const inventory: HostInventory = JSON.parse(row.inventory);
@@ -137,7 +137,7 @@ export function receiveDecommissionReceipt(operationId: string, token: string, i
if (!row || !timingSafeEqual(Buffer.from(row.token_hash, 'hex'), Buffer.from(digest(token), 'hex'))) throw new HostDeletionError('A valid cleanup receipt credential is required.', 401);
if (!['cleaned', 'completed', 'failed'].includes(input.stage as string)) throw new HostDeletionError('Invalid cleanup receipt stage.', 400);
if (row.stage === 'completed') return { acknowledged: true as const };
const operation = getHostOperation(operationId)!;
const operation = getHostOperationStatus(operationId)!;
if (!['accepted','running','unknown'].includes(operation.state)) throw new HostDeletionError('Cleanup has not been accepted by the host.', 409);
const timestamp = now();
if (input.stage === 'failed') {
@@ -314,41 +314,82 @@ export function requireReadyHost(hostId: string, options: { libraryId?: string;
return host;
}
type OperationRow = { id: string; host_id: string; kind: string; instance_id: string | null; generation: number | null; payload: string; state: HostOperationState; result: string | null; error: string | null; created_at: string; updated_at: string };
function operationFromRow(row: OperationRow): HostOperation { return { id: row.id, hostId: row.host_id, kind: row.kind, instanceId: row.instance_id, generation: row.generation, payload: JSON.parse(row.payload), state: row.state, result: row.result === null ? null : JSON.parse(row.result), error: row.error, createdAt: row.created_at, updatedAt: row.updated_at }; }
export function getHostOperation(id: string): HostOperation | null { const row = database().prepare('SELECT * FROM ec_host_operations WHERE id=?').get(id) as OperationRow | undefined; return row ? operationFromRow(row) : null; }
export function listHostOperations(hostId: string, limit = 100, summary = false): HostOperation[] { return (database().prepare(`SELECT ${summary ? `id,host_id,kind,instance_id,generation,state,error,created_at,updated_at,'{}' AS payload,NULL AS result` : '*'} FROM ec_host_operations WHERE host_id=? ORDER BY created_at DESC,rowid DESC LIMIT ?`).all(hostId, Math.min(500, Math.max(1, limit))) as OperationRow[]).map(operationFromRow); }
export type OperationRow = { id: string; host_id: string; kind: string; instance_id: string | null; generation: number | null; payload: string; state: HostOperationState; result: string | null; error: string | null; created_at: string; updated_at: string };
export type OperationDeliveryRow = Omit<OperationRow, 'payload'>;
export type HostOperationStatus = Omit<HostOperation, 'payload'>;
const operationStatusColumns = 'o.id,o.host_id,o.kind,o.instance_id,o.generation,o.state,o.result,o.error,o.created_at,o.updated_at';
const operationPayloadSql = 'COALESCE((SELECT p.payload FROM ec_host_operation_payloads p WHERE p.operation_id=o.id),o.payload)';
function operationStatusFromRow(row: Omit<OperationRow, 'payload'>): HostOperationStatus { return { id: row.id, hostId: row.host_id, kind: row.kind, instanceId: row.instance_id, generation: row.generation, state: row.state, result: row.result === null ? null : JSON.parse(row.result), error: row.error, createdAt: row.created_at, updatedAt: row.updated_at }; }
function operationFromRow(row: OperationRow): HostOperation { return { ...operationStatusFromRow(row), payload: JSON.parse(row.payload) }; }
export function getHostOperation(id: string): HostOperation | null {
const row = database().prepare(`SELECT ${operationStatusColumns},${operationPayloadSql} AS payload FROM ec_host_operations o WHERE o.id=?`)
.get(id) as OperationRow | undefined;
return row ? operationFromRow(row) : null;
}
/** Polling and receipts need the outcome, never another copy of launch files. */
export function getHostOperationStatus(id: string): HostOperationStatus | null {
const row = database().prepare('SELECT id,host_id,kind,instance_id,generation,state,result,error,created_at,updated_at FROM ec_host_operations WHERE id=?')
.get(id) as Omit<OperationRow, 'payload'> | undefined;
return row ? operationStatusFromRow(row) : null;
}
export function listHostOperations(hostId: string, limit = 100, summary = false): HostOperation[] {
const columns = summary ? `o.id,o.host_id,o.kind,o.instance_id,o.generation,o.state,o.error,o.created_at,o.updated_at,'{}' AS payload,NULL AS result`
: `${operationStatusColumns},${operationPayloadSql} AS payload`;
return (database().prepare(`SELECT ${columns} FROM ec_host_operations o WHERE o.host_id=? ORDER BY o.created_at DESC,o.rowid DESC LIMIT ?`)
.all(hostId, Math.min(500, Math.max(1, limit))) as OperationRow[]).map(operationFromRow);
}
/** Heartbeats can tell an executing create to stop at its next safe command boundary. */
export function pendingCreationCancellations(hostId: string): string[] {
return (database().prepare(`SELECT json_extract(payload,'$.createOperationId') AS id FROM ec_host_operations
WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown')`).all(hostId) as { id: string }[]).map(row => row.id);
}
export function queueHostOperation(hostId: string, kind: string, payload: Record<string, unknown>, options: { instanceId?: string; generation?: number; idempotencyKey?: string; id?: string } = {}): HostOperation {
type QueueHostOperationOptions = { instanceId?: string; generation?: number; idempotencyKey?: string; id?: string };
export function queueHostOperation(hostId: string, kind: string, payload: Record<string, unknown>, options: QueueHostOperationOptions = {}): HostOperation {
if (!payload || typeof payload !== 'object' || Array.isArray(payload)) throw new Error('Operation payload must be an object.');
const operation = enqueueHostOperation(hostId, kind, JSON.stringify(payload), options);
return getHostOperation(operation.id)!;
}
/** Keep immutable launch bytes separate so receipts and delivery do not rewrite large payloads. */
export function queueHostOperationSerialized(hostId: string, kind: string, payload: Buffer, options: QueueHostOperationOptions = {}, routingPayload: Record<string, unknown> = {}): HostOperationStatus {
if (!Buffer.isBuffer(payload) || payload[0] !== 123 || payload[payload.length - 1] !== 125) throw new Error('Serialized operation payload must be a JSON object.');
if (!routingPayload || typeof routingPayload !== 'object' || Array.isArray(routingPayload)) throw new Error('Operation routing payload must be an object.');
return enqueueHostOperation(hostId, kind, payload, options, JSON.stringify(routingPayload));
}
function enqueueHostOperation(hostId: string, kind: string, payload: string | Buffer, options: QueueHostOperationOptions, routingPayload?: string): HostOperationStatus {
const host = requireHost(hostId);
if (host.decommission) throw new Error('Host cleanup is in progress; new operations are blocked.');
if (kind === 'host.decommission') throw new Error('Use the host deletion workflow to request cleanup.');
if (!/^[a-z][a-z0-9_.-]{0,79}$/.test(kind)) throw new Error('Invalid operation kind.');
if (!payload || typeof payload !== 'object' || Array.isArray(payload)) throw new Error('Operation payload must be an object.');
if (options.generation !== undefined && (!Number.isSafeInteger(options.generation) || options.generation < 1)) throw new Error('Instance generation must be a positive integer.');
const operation = database().transaction(() => {
if (options.idempotencyKey) {
const existing = database().prepare('SELECT * FROM ec_host_operations WHERE host_id=? AND idempotency_key=?').get(hostId, options.idempotencyKey) as OperationRow | undefined;
const existing = database().prepare(`SELECT o.id,o.kind,o.instance_id,o.generation,${operationPayloadSql}=CAST(? AS TEXT) AS same_payload
FROM ec_host_operations o WHERE o.host_id=? AND o.idempotency_key=?`).get(payload, hostId, options.idempotencyKey) as
{ id: string; kind: string; instance_id: string | null; generation: number | null; same_payload: number } | undefined;
if (existing) {
if (existing.kind !== kind || existing.instance_id !== (options.instanceId ?? null) || existing.generation !== (options.generation ?? null) || existing.payload !== JSON.stringify(payload)) throw new Error('Operation idempotency key is already associated with a different request.');
return operationFromRow(existing);
if (existing.kind !== kind || existing.instance_id !== (options.instanceId ?? null) || existing.generation !== (options.generation ?? null) || !existing.same_payload) throw new Error('Operation idempotency key is already associated with a different request.');
return getHostOperationStatus(existing.id)!;
}
}
if (['instance.create','create'].includes(kind)) requireReadyHost(hostId);
if (host.mode === 'disabled' && kind !== 'instance.cancel-create') throw new Error('Host is disabled. Enable it before requesting operations.');
const id = options.id ?? randomUUID(); const timestamp = now();
database().prepare('INSERT INTO ec_host_operations(id,host_id,kind,instance_id,generation,idempotency_key,payload,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?)').run(id, hostId, kind, options.instanceId ?? null, options.generation ?? null, options.idempotencyKey ?? null, JSON.stringify(payload), timestamp, timestamp);
database().prepare('INSERT INTO ec_host_operations(id,host_id,kind,instance_id,generation,idempotency_key,payload,created_at,updated_at) VALUES(?,?,?,?,?,?,CAST(? AS TEXT),?,?)').run(id, hostId, kind, options.instanceId ?? null, options.generation ?? null, options.idempotencyKey ?? null, routingPayload ?? payload, timestamp, timestamp);
if (routingPayload !== undefined) database().prepare('INSERT INTO ec_host_operation_payloads(operation_id,payload) VALUES(?,CAST(? AS TEXT))').run(id, payload);
database().prepare('INSERT INTO ec_host_operation_events(operation_id,state,created_at) VALUES(?,?,?)').run(id, 'pending', timestamp);
return getHostOperation(id)!;
return getHostOperationStatus(id)!;
})();
queueMicrotask(() => workEvents.emit(hostId));
return operation;
}
export function pendingHostOperations(hostId: string, lane: 'all' | 'read' | 'health' | 'write' = 'all'): HostOperation[] {
return pendingHostOperationRows(hostId, lane).map(operationFromRow);
}
type OperationLane = 'all' | 'read' | 'health' | 'write';
export function pendingHostOperationRows(hostId: string, lane?: OperationLane, includePayload?: true): OperationRow[];
export function pendingHostOperationRows(hostId: string, lane: OperationLane, includePayload: false): OperationDeliveryRow[];
/** Delivery snapshots retain status/results while large immutable payloads stay in SQLite. */
export function pendingHostOperationRows(hostId: string, lane: OperationLane = 'all', includePayload = true): OperationDeliveryRow[] {
const host = requireHost(hostId);
if (host.mode === 'disabled' || !host.connected) return [];
// Older agents hold activity.lock across the entire delivered batch, including
@@ -361,8 +402,8 @@ export function pendingHostOperations(hostId: string, lane: 'all' | 'read' | 'he
// Accepted/running/unknown commands keep the same immutable ID. The agent
// consults its durable journal before acknowledging or executing a replay.
const readOnly = `(kind IN ('host.inspect','instance.inspect','files.read','files.list','files.stat','logs.read')
OR (kind='rcon.execute' AND json_extract(payload,'$.command') IN ('status','getstatus','version')))`;
const healthOnly = `(kind='instance.inspect' OR (kind='rcon.execute' AND json_extract(payload,'$.command') IN ('status','getstatus','version')))`;
OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`;
const healthOnly = `(kind='instance.inspect' OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`;
const healthLane = host.inventory.capabilities?.includes('operations.health-lane');
const laneFilter = lane === 'all' ? '' : lane === 'health' ? ` AND ${healthOnly}`
: ` AND ${lane === 'write' ? 'NOT ' : ''}${readOnly}${lane === 'read' && healthLane ? ` AND NOT ${healthOnly}` : ''}`;
@@ -370,23 +411,31 @@ export function pendingHostOperations(hostId: string, lane: 'all' | 'read' | 'he
// in small batches to amortize delivery without handing out a long backlog.
const limit = healthLane ? lane === 'health' ? 1 : lane === 'read' ? 4 : 25 : 25;
return database().transaction(() => {
const rows = database().prepare(`SELECT * FROM ec_host_operations o WHERE host_id=? AND state IN ('pending','accepted','running','unknown')
const columns = includePayload ? `${operationStatusColumns},${operationPayloadSql} AS payload` : operationStatusColumns;
const rows = database().prepare(`SELECT ${columns} FROM ec_host_operations o WHERE host_id=? AND state IN ('pending','accepted','running','unknown')
${laneFilter}
AND (? OR state != 'pending' OR kind NOT IN ('instance.create','create'))
AND (kind!='instance.cancel-create' OR ?)
AND NOT EXISTS (SELECT 1 FROM ec_host_operations cancellation WHERE cancellation.host_id=o.host_id
AND cancellation.kind='instance.cancel-create' AND json_extract(cancellation.payload,'$.createOperationId')=o.id)
ORDER BY CASE WHEN kind='instance.cancel-create' THEN 0 ELSE 1 END,COALESCE(last_delivered_at,''),created_at,rowid LIMIT ${limit}`)
.all(hostId, Number(host.readiness === 'ready'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false)) as OperationRow[];
.all(hostId, Number(host.readiness === 'ready'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false)) as OperationDeliveryRow[];
const deliveredAt = now();
for (const operation of rows) database().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(deliveredAt, operation.id);
return rows.map(operationFromRow);
return rows;
})();
}
export function reportHostOperation(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperation {
/** Payloads never change after enqueueing; fetch one delivery's UTF-8 bytes without decoding. */
export function getHostOperationPayload(hostId: string, id: string): Buffer {
const row = database().prepare(`SELECT CAST(${operationPayloadSql} AS BLOB) AS payload FROM ec_host_operations o WHERE o.id=? AND o.host_id=?`)
.get(id, hostId) as { payload: Buffer } | undefined;
if (!row) throw new Error('Host operation was not found for this host.');
return row.payload;
}
export function reportHostOperationStatus(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperationStatus {
const operation = database().transaction(() => {
requireHost(hostId);
const operation = getHostOperation(id);
const operation = getHostOperationStatus(id);
if (!operation || operation.hostId !== hostId) throw new Error('Operation was not found for this host.');
if (operation.kind === 'host.decommission' && input.state === 'completed') throw new Error('Host removal requires the cleanup worker completion receipt.');
if (!['accepted','running','completed','failed','unknown'].includes(input.state as string)) throw new Error('Invalid operation state.');
@@ -407,21 +456,29 @@ export function reportHostOperation(hostId: string, id: string, input: { state:
recordAgentReceipt(id, state, now());
}
}
return getHostOperation(id)!;
return getHostOperationStatus(id)!;
})();
queueMicrotask(() => workEvents.emit(`operation:${id}`));
return operation;
}
export async function waitHostOperation(id: string, timeoutMs = 30_000): Promise<HostOperation> {
export function reportHostOperation(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperation {
reportHostOperationStatus(hostId, id, input);
return getHostOperation(id)!;
}
export async function waitHostOperationStatus(id: string, timeoutMs = 30_000): Promise<HostOperationStatus> {
const deadline = Date.now() + timeoutMs;
while (true) {
const operation = getHostOperation(id);
const operation = getHostOperationStatus(id);
if (!operation) throw new Error('Host operation was not found.');
if (['completed','failed','unknown'].includes(operation.state)) return operation;
if (Date.now() >= deadline) throw new Error(`Host operation ${id} is awaiting acknowledgement; reconcile this operation ID before retrying.`);
await waitForHostWork(`operation:${id}`, new AbortController().signal, Math.min(1000, Math.max(1, deadline - Date.now())));
}
}
export async function waitHostOperation(id: string, timeoutMs = 30_000): Promise<HostOperation> {
await waitHostOperationStatus(id, timeoutMs);
return getHostOperation(id)!;
}
export function hostPortAvailable(hostId: string, candidate: number, bindAddress?: string, protocol: 'udp' | 'tcp' = 'udp', instanceId?: string): boolean {
const host = requireHost(hostId); port(candidate); const bind = bindAddress ?? host.bindAddress;