Expire undelivered health checks when their wait times out
This commit is contained in:
@@ -587,7 +587,25 @@ export async function waitHostOperationStatus(id: string, timeoutMs = 30_000): P
|
|||||||
const operation = getHostOperationStatus(id);
|
const operation = getHostOperationStatus(id);
|
||||||
if (!operation) throw new Error('Host operation was not found.');
|
if (!operation) throw new Error('Host operation was not found.');
|
||||||
if (['completed','failed','unknown'].includes(operation.state)) return operation;
|
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.`);
|
if (Date.now() >= deadline) {
|
||||||
|
// Delivery and expiry both claim the pending row transactionally.
|
||||||
|
// Never cancel a delivered check or a command with side effects.
|
||||||
|
const expired = database().transaction(() => {
|
||||||
|
const timestamp = now();
|
||||||
|
const error = 'Expired before delivery.';
|
||||||
|
const result = database().prepare(`UPDATE ec_host_operations SET state='failed',error=?,updated_at=?
|
||||||
|
WHERE id=? AND state='pending' AND last_delivered_at IS NULL
|
||||||
|
AND kind IN ('instance.inspect','rcon.status-batch')`).run(error, timestamp, id);
|
||||||
|
if (result.changes) database().prepare('INSERT INTO ec_host_operation_events(operation_id,state,created_at,error) VALUES(?,?,?,?)')
|
||||||
|
.run(id, 'failed', timestamp, error);
|
||||||
|
return result.changes > 0;
|
||||||
|
})();
|
||||||
|
if (expired) {
|
||||||
|
queueMicrotask(() => workEvents.emit(`operation:${id}`));
|
||||||
|
throw new Error(`Host operation ${id} expired before delivery.`);
|
||||||
|
}
|
||||||
|
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())));
|
await waitForHostWork(`operation:${id}`, new AbortController().signal, Math.min(1000, Math.max(1, deadline - Date.now())));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -15,6 +15,38 @@ async function enrolled() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
describe('native host enrollment and scoped authentication', () => {
|
describe('native host enrollment and scoped authentication', () => {
|
||||||
|
it.each(['instance.inspect','rcon.status-batch'])('expires undelivered %s on timeout and delivers the next fresh check', async kind => {
|
||||||
|
const api = await enrolled();
|
||||||
|
const { db } = await import('../../src/lib/server/db/index.js');
|
||||||
|
api.heartbeat(api.host.id,{...inventory,capabilities:['operations.health-lane']});
|
||||||
|
const stale = api.queueHostOperation(api.host.id,kind,{});
|
||||||
|
await expect(api.waitHostOperationStatus(stale.id,0)).rejects.toThrow('expired before delivery');
|
||||||
|
expect(api.getHostOperationStatus(stale.id)).toMatchObject({state:'failed',error:'Expired before delivery.'});
|
||||||
|
expect(await api.waitHostOperationStatus(stale.id,0)).toMatchObject({state:'failed'});
|
||||||
|
expect(db().prepare("SELECT state,error FROM ec_host_operation_events WHERE operation_id=? AND state='failed'").all(stale.id))
|
||||||
|
.toEqual([{state:'failed',error:'Expired before delivery.'}]);
|
||||||
|
const fresh = api.queueHostOperation(api.host.id,kind,{});
|
||||||
|
expect(api.pendingHostOperations(api.host.id,'health').map(op=>op.id)).toEqual([fresh.id]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it.each(['pending','accepted','running'])('preserves delivered health checks in %s state after the caller times out', async state => {
|
||||||
|
const api = await enrolled();
|
||||||
|
const operation = api.queueHostOperation(api.host.id,'instance.inspect',{});
|
||||||
|
api.pendingHostOperations(api.host.id,'health');
|
||||||
|
if (state !== 'pending') api.reportHostOperationStatus(api.host.id,operation.id,{state});
|
||||||
|
await expect(api.waitHostOperationStatus(operation.id,0)).rejects.toThrow('awaiting acknowledgement');
|
||||||
|
expect(api.getHostOperationStatus(operation.id)?.state).toBe(state);
|
||||||
|
api.reportHostOperationStatus(api.host.id,operation.id,{state:'completed',result:{healthy:true}});
|
||||||
|
expect(await api.waitHostOperationStatus(operation.id)).toMatchObject({state:'completed',result:{healthy:true}});
|
||||||
|
});
|
||||||
|
|
||||||
|
it.each(['instance.restart','files.stage','rcon.execute'])('preserves undelivered %s when its caller times out', async kind => {
|
||||||
|
const api = await enrolled();
|
||||||
|
const operation = api.queueHostOperation(api.host.id,kind,{command:'status'});
|
||||||
|
await expect(api.waitHostOperationStatus(operation.id,0)).rejects.toThrow('awaiting acknowledgement');
|
||||||
|
expect(api.getHostOperationStatus(operation.id)?.state).toBe('pending');
|
||||||
|
});
|
||||||
|
|
||||||
it('keeps the host online and delivers work when individual state observations are invalid', async () => {
|
it('keeps the host online and delivers work when individual state observations are invalid', async () => {
|
||||||
const api = await enrolled();
|
const api = await enrolled();
|
||||||
const good = { instanceId: 'healthy-server', generation: 1, state: 'running', healthy: true, observedAt: new Date().toISOString() };
|
const good = { instanceId: 'healthy-server', generation: 1, state: 'running', healthy: true, observedAt: new Date().toISOString() };
|
||||||
|
|||||||
Reference in New Issue
Block a user