diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index f5c97c5..969c926 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -504,8 +504,12 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = ' 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) + ${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}`) - .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(); for (const operation of rows) database().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(deliveredAt, operation.id); return rows; diff --git a/tests/unit/native-hosts.test.ts b/tests/unit/native-hosts.test.ts index 3e99d98..4290389 100644 --- a/tests/unit/native-hosts.test.ts +++ b/tests/unit/native-hosts.test.ts @@ -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=?'))!; prepare.mockRestore(); // 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) .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 () => { const api = await enrolled(); api.heartbeat(api.host.id, { ...inventory, capabilities: ['operations.health-lane'] });