Avoid replaying active health operations during polling
This commit is contained in:
@@ -504,8 +504,12 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = '
|
|||||||
AND (kind!='instance.cancel-create' OR ?)
|
AND (kind!='instance.cancel-create' OR ?)
|
||||||
AND NOT EXISTS (SELECT 1 FROM ec_host_operations cancellation WHERE cancellation.host_id=o.host_id
|
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)
|
AND cancellation.kind='instance.cancel-create' AND json_extract(cancellation.payload,'$.createOperationId')=o.id)
|
||||||
|
${lane === 'health' ? "AND (state NOT IN ('accepted','running') OR MAX(updated_at,COALESCE(last_delivered_at,'')) <= ?)" : ''}
|
||||||
ORDER BY CASE WHEN kind='instance.cancel-create' THEN 0 ELSE 1 END,COALESCE(last_delivered_at,''),created_at,rowid LIMIT ${limit}`)
|
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' || host.mode === 'draining'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false)) as OperationDeliveryRow[];
|
// A short replay lease avoids handing the idle health lane its busy
|
||||||
|
// sibling's work, but still recovers a lost receipt or agent restart.
|
||||||
|
.all(hostId, Number(host.readiness === 'ready' || host.mode === 'draining'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false),
|
||||||
|
...(lane === 'health' ? [new Date(Date.now() - 30_000).toISOString()] : [])) as OperationDeliveryRow[];
|
||||||
const deliveredAt = now();
|
const deliveredAt = now();
|
||||||
for (const operation of rows) database().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(deliveredAt, operation.id);
|
for (const operation of rows) database().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(deliveredAt, operation.id);
|
||||||
return rows;
|
return rows;
|
||||||
|
|||||||
@@ -94,11 +94,30 @@ describe('native host enrollment and scoped authentication', () => {
|
|||||||
const query = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('FROM ec_host_operations o WHERE host_id=?'))!;
|
const query = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('FROM ec_host_operations o WHERE host_id=?'))!;
|
||||||
prepare.mockRestore();
|
prepare.mockRestore();
|
||||||
// Assert the actual poll's access path, avoiding hardware-dependent timing thresholds.
|
// Assert the actual poll's access path, avoiding hardware-dependent timing thresholds.
|
||||||
const plan = handle.prepare(`EXPLAIN QUERY PLAN ${query}`).all(api.host.id, 1, 1) as { detail: string }[];
|
const plan = handle.prepare(`EXPLAIN QUERY PLAN ${query}`).all(api.host.id, 1, 1, new Date().toISOString()) as { detail: string }[];
|
||||||
expect(plan.find(row => row.detail.includes('SEARCH cancellation'))?.detail)
|
expect(plan.find(row => row.detail.includes('SEARCH cancellation'))?.detail)
|
||||||
.toContain('(host_id=? AND kind=?)');
|
.toContain('(host_id=? AND kind=?)');
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it.each(['accepted', 'running'])('withholds %s health work and replays its original ID after the recovery lease', async state => {
|
||||||
|
const api = await enrolled();
|
||||||
|
vi.useFakeTimers({ toFake: ['Date'] });
|
||||||
|
const at = Date.now();
|
||||||
|
api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] });
|
||||||
|
const busy = api.queueHostOperation(api.host.id, 'instance.inspect', {});
|
||||||
|
api.pendingHostOperations(api.host.id, 'health');
|
||||||
|
api.reportHostOperation(api.host.id, busy.id, { state });
|
||||||
|
const next = api.queueHostOperation(api.host.id, 'rcon.execute', { command: 'status' });
|
||||||
|
expect(api.pendingHostOperations(api.host.id, 'health').map(op => op.id)).toEqual([next.id]);
|
||||||
|
api.reportHostOperation(api.host.id, next.id, { state: 'completed' });
|
||||||
|
expect(api.pendingHostOperations(api.host.id, 'health')).toEqual([]);
|
||||||
|
vi.setSystemTime(at + 30_001);
|
||||||
|
expect(api.pendingHostOperations(api.host.id, 'health').map(op => op.id)).toEqual([busy.id]);
|
||||||
|
expect(api.pendingHostOperations(api.host.id, 'health')).toEqual([]);
|
||||||
|
api.reportHostOperation(api.host.id, busy.id, { state: 'unknown' });
|
||||||
|
expect(api.pendingHostOperations(api.host.id, 'health').map(op => op.id)).toEqual([busy.id]);
|
||||||
|
});
|
||||||
|
|
||||||
it('reserves health delivery while a file backlog is pending and bounds background batches', async () => {
|
it('reserves health delivery while a file backlog is pending and bounds background batches', async () => {
|
||||||
const api = await enrolled();
|
const api = await enrolled();
|
||||||
api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] });
|
api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] });
|
||||||
|
|||||||
Reference in New Issue
Block a user