diff --git a/agent/ec_agent.py b/agent/ec_agent.py index 5c4d7cd..eafd700 100644 --- a/agent/ec_agent.py +++ b/agent/ec_agent.py @@ -32,7 +32,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.4.13" +VERSION = "1.4.14" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -1393,7 +1393,8 @@ class Agent: with ThreadPoolExecutor(max_workers=2) as executor: results = list(executor.map(probe, identities)) - return {"ok": all(row["reply"].get("ok") for row in results), "results": results} + # Completion describes the batch; individual failures remain in each reply. + return {"ok": True, "results": results} def rcon(self, instance_id, generation, command, deadline=None, status_attempts=3): if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")): diff --git a/agent/tests/test_agent.py b/agent/tests/test_agent.py index 2075881..1dbdacf 100644 --- a/agent/tests/test_agent.py +++ b/agent/tests/test_agent.py @@ -114,6 +114,12 @@ class JournalRetentionTests(unittest.TestCase): self.assertFalse(accepted) self.assertEqual(saved["id"], row["id"]) + def test_completed_batch_with_failed_targets_is_pruned_after_acknowledgment(self): + result = {"ok": True, "results": [{"instanceId": INSTANCE, "generation": 1, + "reply": {"ok": False, "reason": "timeout", "text": ""}}]} + self.record("rcon.status-batch", {"instances": [{"instanceId": INSTANCE, "generation": 1}]}, result) + self.assertEqual(self.prune(), 1) + def test_errors_failures_and_unfinished_operations_are_retained(self): retained = [self.record(result=result) for result in ({"ok": False}, {"healthy": False}, {"error": "partial result"}, {"error": False}, {"error": []})] retained.append(self.record(error="read did not finish")) @@ -644,7 +650,7 @@ class AgentTests(unittest.TestCase): self.assertFalse(thread.is_alive()) self.assertEqual(peak, 2) self.assertEqual(len(results[0]["results"]), 5) - self.assertFalse(results[0]["ok"]) + self.assertTrue(results[0]["ok"]) self.assertEqual(results[0]["results"][2]["reply"]["reason"], "unreachable") self.assertTrue(results[0]["results"][4]["reply"]["ok"]) diff --git a/src/lib/server/hosts/observations.ts b/src/lib/server/hosts/observations.ts index 85793f9..1e877fa 100644 --- a/src/lib/server/hosts/observations.ts +++ b/src/lib/server/hosts/observations.ts @@ -11,7 +11,9 @@ export function safeOperationTarget(kind: string, value: string | null): string export function successfulObservation(kind: string, payload: Record, result: unknown): boolean { const routine = observationKinds.includes(kind) || kind === 'rcon.execute' && commands.includes(String(payload.command)); const value = result && typeof result === 'object' ? result as Record : {}; - return routine && value.ok !== false && value.healthy !== false && (value.error == null || value.error === ''); + // Older agents set ok=false when any target fails, despite completing the batch. + const completedBatch = kind === 'rcon.status-batch' && Array.isArray(value.results); + return routine && (completedBatch || value.ok !== false) && value.healthy !== false && (value.error == null || value.error === ''); } /** Uses the operation alias o; serialized payloads may live outside the routing row. */ @@ -22,7 +24,9 @@ export const routineObservationSql = `(CASE WHEN o.kind IN (${observationKinds.m export const successfulObservationSql = `(o.state='completed' AND COALESCE(o.error,'')='' AND ${routineObservationSql} AND CASE WHEN o.result IS NULL THEN 1 WHEN json_valid(o.result) - THEN json_extract(o.result,'$.ok') IS NOT 0 AND json_extract(o.result,'$.healthy') IS NOT 0 + THEN (json_extract(o.result,'$.ok') IS NOT 0 + OR (o.kind='rcon.status-batch' AND json_type(o.result,'$.results')='array')) + AND json_extract(o.result,'$.healthy') IS NOT 0 AND COALESCE(json_extract(o.result,'$.error'),'')='' ELSE 0 END)`; export const importantOperationSql = `(o.routine=0 OR o.state IN ('failed','unknown') diff --git a/src/lib/server/hosts/schema.ts b/src/lib/server/hosts/schema.ts index 807e12b..116b63b 100644 --- a/src/lib/server/hosts/schema.ts +++ b/src/lib/server/hosts/schema.ts @@ -112,6 +112,15 @@ export function ensureHostSchema(handle: Database.Database): void { handle.exec(`CREATE INDEX IF NOT EXISTS ec_host_operations_routine_history ON ec_host_operations(routine,created_at); CREATE INDEX IF NOT EXISTS ec_host_operations_important_history ON ec_host_operations(created_at) WHERE ${importantOperationSql.replaceAll('o.', '')}`); + handle.exec('CREATE TABLE IF NOT EXISTS ec_host_schema_migrations (name TEXT PRIMARY KEY)'); + if (!handle.prepare('SELECT 1 FROM ec_host_schema_migrations WHERE name=?').get('completed-status-batches')) { + handle.transaction(() => { + // Reclassify existing completed batches without discarding their per-server replies. + handle.exec(`UPDATE ec_host_operations AS o SET observation_success=1 + WHERE o.kind='rcon.status-batch' AND o.observation_success=0 AND ${successfulObservationSql}`); + handle.prepare('INSERT INTO ec_host_schema_migrations(name) VALUES(?)').run('completed-status-batches'); + })(); + } const hostColumns = handle.prepare('PRAGMA table_info(ec_hosts)').all() as { name: string }[]; if (!hostColumns.some(column => column.name === 'deleted_at')) handle.exec('ALTER TABLE ec_hosts ADD COLUMN deleted_at TEXT'); const allowanceColumns = handle.prepare('PRAGMA table_info(ec_resource_allowances)').all() as { name: string }[]; diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index 5c4d7cd..eafd700 100644 --- a/static/agent/ec-agent.py +++ b/static/agent/ec-agent.py @@ -32,7 +32,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.4.13" +VERSION = "1.4.14" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -1393,7 +1393,8 @@ class Agent: with ThreadPoolExecutor(max_workers=2) as executor: results = list(executor.map(probe, identities)) - return {"ok": all(row["reply"].get("ok") for row in results), "results": results} + # Completion describes the batch; individual failures remain in each reply. + return {"ok": True, "results": results} def rcon(self, instance_id, generation, command, deadline=None, status_attempts=3): if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")): diff --git a/static/agent/manifest.json b/static/agent/manifest.json index 208bc88..ccade6e 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.4.13", + "version": "1.4.14", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "2f3449058cd4b3ab9e3ff8f541fa3c65ad2d94aabe20694cdec3a6e59c8a4fd6", + "sha256": "3e2e11389f412a0f1b2746c85b9afa626c5e0fcaa4801bd13f1de283abf8830e", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" } diff --git a/tests/unit/host-observation-retention.test.ts b/tests/unit/host-observation-retention.test.ts index 1e89975..b715e16 100644 --- a/tests/unit/host-observation-retention.test.ts +++ b/tests/unit/host-observation-retention.test.ts @@ -63,6 +63,36 @@ describe('routine host observation retention', () => { expect(api.pruneHostObservations()).toBe(0); }); + it('keeps per-server failures in routine completed batches and prunes them on schedule', async () => { + const api = await setup(); + const results = [{ instanceId: 'one', generation: 1, reply: { ok: false, reason: 'timeout', text: '' } }]; + const batches = [false, true].map(ok => api.add('rcon.status-batch', { result: { ok, results } })); + const failed = api.add('rcon.status-batch', { state: 'failed', error: 'Batch failed' }); + const malformed = api.add('rcon.status-batch', { result: { ok: false } }); + expect(api.listHostOperationHistory(1, 'important').operations.map(op => op.id).sort()).toEqual([failed, malformed].sort()); + for (const id of batches) expect(api.getHostOperation(id)?.result).toMatchObject({ results }); + expect(api.pruneHostObservations()).toBe(2); + for (const id of batches) expect(api.getHostOperationStatus(id)).toBeNull(); + }); + + it('reclassifies existing completed batches once without hiding operation errors', async () => { + const api = await setup(); + const result = { ok: false, results: [{ instanceId: 'one', generation: 1, reply: { ok: false, reason: 'timeout' } }] }; + const batch = api.add('rcon.status-batch', { result }); + const retained = [api.add('rcon.status-batch', { result, error: 'Incomplete' }), + api.add('rcon.status-batch', { result, state: 'failed' }), + api.add('rcon.status-batch', { result: { ...result, error: 'Incomplete' } })]; + api.handle.exec("UPDATE ec_host_operations SET observation_success=0; DELETE FROM ec_host_schema_migrations WHERE name='completed-status-batches'"); + api.closeDb(); + expect(api.listHostOperationHistory(1, 'important').operations.map(op => op.id).sort()).toEqual(retained.sort()); + expect(api.getHostOperation(batch)?.result).toEqual(result); + expect(api.pruneHostObservations()).toBe(1); + // The migration is not a recurring scan or reclassification of retained evidence. + api.db().prepare("UPDATE ec_host_operations SET state='completed',error=NULL,result=? WHERE id=?").run(JSON.stringify(result), retained[0]); + api.closeDb(); + expect(api.pruneHostObservations()).toBe(0); + }); + it('classifies full serialized RCON payloads and deletes their events and immutable bytes together', async () => { const api = await setup(); const read = api.add('rcon.execute', { serialized: true, payload: { command: 'status' }, routing: { command: 'map_restart' } });