Let agents idle for updates required by creation cancellation
This commit is contained in:
@@ -426,3 +426,13 @@ Docker command must return or time out before cleanup can run. Cancellation
|
|||||||
does not require the original creation to acknowledge completion. The agent
|
does not require the original creation to acknowledge completion. The agent
|
||||||
records a removal tombstone before cleanup so a delayed creation command cannot
|
records a removal tombstone before cleanup so a delayed creation command cannot
|
||||||
recreate the instance, and resumes interrupted cleanup under the same operation ID.
|
recreate the instance, and resumes interrupted cleanup under the same operation ID.
|
||||||
|
|
||||||
|
While cancellation requires a newer agent, the controller pauses all command
|
||||||
|
delivery to that host, including background observations and command replays.
|
||||||
|
Older agents hold their update lock across a whole command batch; continuing to
|
||||||
|
feed them work could otherwise starve the updater. Already delivered work can
|
||||||
|
finish and report normally, and heartbeats continue. The next automatic update
|
||||||
|
check can install the release once the batch has drained. When a heartbeat
|
||||||
|
advertises cancellation support, delivery resumes with cancellation cleanup
|
||||||
|
first. This controller-side hold also recovers cancellations already waiting
|
||||||
|
on older agents. Hosts without automatic updates need the current installer.
|
||||||
|
|||||||
@@ -319,6 +319,13 @@ export function queueHostOperation(hostId: string, kind: string, payload: Record
|
|||||||
export function pendingHostOperations(hostId: string): HostOperation[] {
|
export function pendingHostOperations(hostId: string): HostOperation[] {
|
||||||
const host = requireHost(hostId);
|
const host = requireHost(hostId);
|
||||||
if (host.mode === 'disabled' || !host.connected) return [];
|
if (host.mode === 'disabled' || !host.connected) return [];
|
||||||
|
// Older agents hold activity.lock across the entire delivered batch, including
|
||||||
|
// observations. Filtering only the unsupported cleanup can keep the updater
|
||||||
|
// busy indefinitely. Drain the old batch by withholding ALL further work;
|
||||||
|
// heartbeats and receipts still flow, and the new capability releases this hold.
|
||||||
|
// This must live in the controller so the fix also works before an agent update.
|
||||||
|
if (!host.inventory.capabilities?.includes('instance.cancel-create') && database().prepare(`SELECT 1 FROM ec_host_operations
|
||||||
|
WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown') LIMIT 1`).get(hostId)) return [];
|
||||||
// Accepted/running/unknown commands keep the same immutable ID. The agent
|
// Accepted/running/unknown commands keep the same immutable ID. The agent
|
||||||
// consults its durable journal before acknowledging or executing a replay.
|
// consults its durable journal before acknowledging or executing a replay.
|
||||||
return database().transaction(() => {
|
return database().transaction(() => {
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ import { sha256Buffer } from '../fleet/files.js';
|
|||||||
import type { StagedFileChange } from '../fleet/staged-files.js';
|
import type { StagedFileChange } from '../fleet/staged-files.js';
|
||||||
import type { LaunchPreview } from '../../deployment-plan.js';
|
import type { LaunchPreview } from '../../deployment-plan.js';
|
||||||
import type { ProvisioningAction } from '../../provisioning.js';
|
import type { ProvisioningAction } from '../../provisioning.js';
|
||||||
|
import { hostAgentUpdatesEnabled } from '../../hosts.js';
|
||||||
import { cancelCreation, finishCreationCancellation } from './cancel-creation.js';
|
import { cancelCreation, finishCreationCancellation } from './cancel-creation.js';
|
||||||
|
|
||||||
export interface NativeSelection { hostId: string; libraryId: string; image?: string; bindAddress?: string; publicAddress?: string; publicPort?: number }
|
export interface NativeSelection { hostId: string; libraryId: string; image?: string; bindAddress?: string; publicAddress?: string; publicPort?: number }
|
||||||
@@ -282,8 +283,10 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
|
|||||||
}
|
}
|
||||||
if (job.operation_id) {
|
if (job.operation_id) {
|
||||||
const operation = getHostOperation(job.operation_id);
|
const operation = getHostOperation(job.operation_id);
|
||||||
if (operation?.kind === 'instance.cancel-create' && operation.state === 'pending' && !host.inventory.capabilities?.includes('instance.cancel-create')) {
|
if (operation?.kind === 'instance.cancel-create' && !['completed', 'failed'].includes(operation.state) && !host.inventory.capabilities?.includes('instance.cancel-create')) {
|
||||||
hold(job, 'Update the host agent to support creation cancellation. Creation is cancelled; cleanup and port release await the updated agent.'); return;
|
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 || 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') {
|
if (operation.state === 'failed') {
|
||||||
|
|||||||
@@ -136,6 +136,43 @@ describe('native host readiness and reservations', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
describe('durable host operations', () => {
|
describe('durable host operations', () => {
|
||||||
|
it('drains an old agent for a cancellation update while allowing receipts, heartbeats, and other hosts', async () => {
|
||||||
|
const api = await enrolled();
|
||||||
|
const original = api.queueHostOperation(api.host.id, 'instance.create', {}, { instanceId: 'cancelled-instance', generation: 1 });
|
||||||
|
const inspect = api.queueHostOperation(api.host.id, 'instance.inspect', {});
|
||||||
|
expect(api.pendingHostOperations(api.host.id).map(op => op.id)).toContain(original.id);
|
||||||
|
api.reportHostOperation(api.host.id, original.id, { state: 'running' });
|
||||||
|
api.reportHostOperation(api.host.id, inspect.id, { state: 'unknown' });
|
||||||
|
const cleanup = api.queueHostOperation(api.host.id, 'instance.cancel-create', { createOperationId: original.id }, { instanceId: 'cancelled-instance', generation: 1 });
|
||||||
|
const logs = api.queueHostOperation(api.host.id, 'logs.read', {});
|
||||||
|
const preparation = api.queueHostOperation(api.host.id, 'library.download', {});
|
||||||
|
const other = api.createHost({ name: 'Unaffected host' });
|
||||||
|
api.enrollHost(api.createEnrollment(other.id).token, inventory);
|
||||||
|
const otherWork = api.queueHostOperation(other.id, 'host.inspect', {});
|
||||||
|
const { db } = await import('../../src/lib/server/db/index.js');
|
||||||
|
const before = db().prepare('SELECT id,state,last_delivered_at FROM ec_host_operations WHERE host_id=? ORDER BY id').all(api.host.id);
|
||||||
|
for (let poll = 0; poll < 3; poll++) expect(api.pendingHostOperations(api.host.id)).toEqual([]);
|
||||||
|
expect(db().prepare('SELECT id,state,last_delivered_at FROM ec_host_operations WHERE host_id=? ORDER BY id').all(api.host.id)).toEqual(before);
|
||||||
|
expect(api.pendingHostOperations(other.id).map(op => op.id)).toEqual([otherWork.id]);
|
||||||
|
api.reportHostOperation(api.host.id, inspect.id, { state: 'completed', result: { healthy: true } });
|
||||||
|
api.heartbeat(api.host.id, inventory);
|
||||||
|
expect(api.pendingHostOperations(api.host.id)).toEqual([]);
|
||||||
|
|
||||||
|
// The controller hold must clear on capability, without waiting for cleanup
|
||||||
|
// or the original creation to finish (which would recreate the deadlock).
|
||||||
|
api.heartbeat(api.host.id, { ...inventory, capabilities: ['instance.cancel-create'] });
|
||||||
|
expect(api.pendingHostOperations(api.host.id).map(op => op.id)).toEqual([cleanup.id, logs.id, preparation.id]);
|
||||||
|
expect(api.getHostOperation(original.id)?.state).toBe('running');
|
||||||
|
expect(api.getHostOperation(cleanup.id)?.state).toBe('pending');
|
||||||
|
});
|
||||||
|
|
||||||
|
it.each(['completed', 'failed'] as const)('does not pause an old host for a terminal %s cancellation', async state => {
|
||||||
|
const api = await enrolled();
|
||||||
|
const cleanup = api.queueHostOperation(api.host.id, 'instance.cancel-create', { createOperationId: 'creation' });
|
||||||
|
api.reportHostOperation(api.host.id, cleanup.id, { state });
|
||||||
|
const work = api.queueHostOperation(api.host.id, 'host.inspect', {});
|
||||||
|
expect(api.pendingHostOperations(api.host.id).map(op => op.id)).toEqual([work.id]);
|
||||||
|
});
|
||||||
it('rotates unresolved operation delivery so an old batch cannot starve new work', async () => {
|
it('rotates unresolved operation delivery so an old batch cannot starve new work', async () => {
|
||||||
const api = await enrolled();
|
const api = await enrolled();
|
||||||
const operations = Array.from({ length: 30 }, () => api.queueHostOperation(api.host.id, 'instance.inspect', {}));
|
const operations = Array.from({ length: 30 }, () => api.queueHostOperation(api.host.id, 'instance.inspect', {}));
|
||||||
|
|||||||
@@ -142,6 +142,12 @@ describe('native instance provisioning', () => {
|
|||||||
expect(pendingHostOperations(hostId).filter(op => op.kind.startsWith('instance.'))).toEqual([]);
|
expect(pendingHostOperations(hostId).filter(op => op.kind.startsWith('instance.'))).toEqual([]);
|
||||||
await advance();
|
await advance();
|
||||||
expect((await detail(cleanup)).stage).toContain('Update the host agent');
|
expect((await detail(cleanup)).stage).toContain('Update the host agent');
|
||||||
|
expect((await detail(cleanup)).stage).toContain('Automatic updates are not enabled');
|
||||||
|
heartbeat(hostId, { ...inventory(), capabilities: ['agent.update'],
|
||||||
|
agentUpdate: { enabled: true, state: 'deferred', currentVersion: '1.2.2', targetVersion: '1.3.0' } });
|
||||||
|
await advance();
|
||||||
|
expect((await detail(cleanup)).stage).toContain('New command delivery is paused');
|
||||||
|
expect((await detail(cleanup)).stage).toContain('automatic updater can run');
|
||||||
db().prepare('UPDATE ec_hosts SET last_heartbeat=? WHERE id=?').run('2000-01-01T00:00:00.000Z', hostId);
|
db().prepare('UPDATE ec_hosts SET last_heartbeat=? WHERE id=?').run('2000-01-01T00:00:00.000Z', hostId);
|
||||||
await advance();
|
await advance();
|
||||||
expect((await detail(cleanup)).stage).toContain('unavailable');
|
expect((await detail(cleanup)).stage).toContain('unavailable');
|
||||||
|
|||||||
Reference in New Issue
Block a user