425 lines
18 KiB
TypeScript
425 lines
18 KiB
TypeScript
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
|
import { newFixture, type Fixture } from '../helpers/fixture.js';
|
|
|
|
const now = Date.parse('2026-10-05T16:00:00Z');
|
|
const day = 86_400_000;
|
|
const at = (age: number) => new Date(now - age).toISOString();
|
|
const inventory = {
|
|
protocolVersion: 1,
|
|
dockerRunning: true,
|
|
directoriesWritable: true,
|
|
storageAvailable: 100 * 1024 ** 3,
|
|
preparedRuntimeImages: ['plutainer:test'],
|
|
libraries: [{ id: 'library', state: 'ready' }]
|
|
};
|
|
let fixture: Fixture;
|
|
|
|
beforeEach(async () => {
|
|
vi.resetModules();
|
|
fixture = await newFixture([]);
|
|
vi.useFakeTimers({ toFake: ['Date'] });
|
|
vi.setSystemTime(now);
|
|
});
|
|
afterEach(async () => {
|
|
vi.useRealTimers();
|
|
vi.restoreAllMocks();
|
|
await fixture.cleanup();
|
|
});
|
|
|
|
async function setup() {
|
|
const hosts = await import('../../src/lib/server/hosts/index.js');
|
|
const { db, closeDb } = await import('../../src/lib/server/db/index.js');
|
|
const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js');
|
|
const host = hosts.createHost({ name: 'Retention host' });
|
|
const credentials = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
|
|
const handle = db();
|
|
function add(
|
|
kind: string,
|
|
options: {
|
|
state?: string;
|
|
payload?: Record<string, unknown>;
|
|
result?: unknown;
|
|
error?: string;
|
|
age?: number;
|
|
serialized?: boolean;
|
|
routing?: Record<string, unknown>;
|
|
} = {}
|
|
) {
|
|
const payload = options.payload ?? {};
|
|
const operation = options.serialized
|
|
? hosts.queueHostOperationSerialized(
|
|
host.id,
|
|
kind,
|
|
Buffer.from(JSON.stringify(payload)),
|
|
{},
|
|
options.routing ?? {}
|
|
)
|
|
: hosts.queueHostOperation(host.id, kind, payload);
|
|
const state = options.state ?? 'completed';
|
|
if (state !== 'pending')
|
|
hosts.reportHostOperationStatus(host.id, operation.id, {
|
|
state,
|
|
result: options.result === undefined ? { ok: true } : options.result,
|
|
error: options.error
|
|
});
|
|
handle
|
|
.prepare('UPDATE ec_host_operations SET created_at=?,updated_at=? WHERE id=?')
|
|
.run(at(10 * day), at(options.age ?? 2 * day), operation.id);
|
|
return operation.id;
|
|
}
|
|
return {
|
|
...hosts,
|
|
handle,
|
|
db,
|
|
closeDb,
|
|
host,
|
|
credentials,
|
|
fleetSettings,
|
|
saveFleetSettings,
|
|
add
|
|
};
|
|
}
|
|
|
|
describe('routine host observation retention', () => {
|
|
it('drains more than one batch while yielding and preserving recent or failed operations', async () => {
|
|
const api = await setup();
|
|
const { drainRetention } = await import('../../src/lib/server/retention.js');
|
|
const retained = [
|
|
api.add('logs.read', { age: 0 }),
|
|
api.add('logs.read', { state: 'failed' }),
|
|
api.add('instance.restart')
|
|
];
|
|
api.handle.transaction(() => {
|
|
const insert = api.handle.prepare(`INSERT INTO ec_host_operations
|
|
(id,host_id,kind,payload,state,created_at,updated_at,observation_success)
|
|
VALUES(?,?,'logs.read','{}','completed',?,?,1)`);
|
|
for (let i = 0; i < 1501; i++)
|
|
insert.run(`read-${i}`, api.host.id, at(2 * day), at(2 * day));
|
|
})();
|
|
let yielded = false;
|
|
setImmediate(() => {
|
|
yielded = true;
|
|
});
|
|
expect(await drainRetention()).toBe(1501);
|
|
expect(yielded).toBe(true);
|
|
expect(
|
|
(api.handle.prepare('SELECT id FROM ec_host_operations').all() as { id: string }[])
|
|
.map((row) => row.id)
|
|
.sort()
|
|
).toEqual(retained.sort());
|
|
expect(await drainRetention()).toBe(0);
|
|
});
|
|
|
|
it('strips existing log bodies in batches without deleting recent or protected history', async () => {
|
|
const api = await setup();
|
|
const { drainRetention, setRetentionControl } =
|
|
await import('../../src/lib/server/retention.js');
|
|
const ids = Array.from({ length: 53 }, () => api.add('logs.read', { age: 0 }));
|
|
setRetentionControl({ category: 'observation', id: ids[0], protected: true });
|
|
api.handle
|
|
.prepare("UPDATE ec_host_operations SET result=? WHERE kind='logs.read'")
|
|
.run(JSON.stringify({ lines: ['legacy log contents'], size: 20, truncated: false }));
|
|
expect(await drainRetention()).toBe(0);
|
|
for (const id of ids)
|
|
expect(api.getHostOperation(id)?.result).toEqual({
|
|
retained: false,
|
|
lineCount: 1,
|
|
size: 20,
|
|
truncated: false
|
|
});
|
|
expect(
|
|
api.handle
|
|
.prepare("SELECT value FROM settings WHERE key='log_history_metadata_v1'")
|
|
.get()
|
|
).toEqual({ value: '1' });
|
|
expect(api.handle.pragma('foreign_key_check')).toEqual([]);
|
|
});
|
|
|
|
it('removes aged successful reads, manifests, and read-only RCON while retaining failures, actions, and recent evidence', async () => {
|
|
const api = await setup();
|
|
const removed = [
|
|
'host.inspect',
|
|
'instance.inspect',
|
|
'files.read',
|
|
'files.stat',
|
|
'files.list',
|
|
'files.manifest',
|
|
'logs.read'
|
|
].map((kind) => api.add(kind));
|
|
removed.push(
|
|
...['status', 'getstatus', 'version'].map((command) =>
|
|
api.add('rcon.execute', { payload: { command } })
|
|
)
|
|
);
|
|
const retained = [
|
|
...['pending', 'accepted', 'running', 'failed', 'unknown'].map((state) =>
|
|
api.add('files.read', { state })
|
|
),
|
|
...[
|
|
'instance.restart',
|
|
'instance.stop',
|
|
'instance.remove',
|
|
'files.stage',
|
|
'files.activate',
|
|
'files.rollback',
|
|
'runtime.pull',
|
|
'unknown.read'
|
|
].map((kind) => api.add(kind)),
|
|
api.add('rcon.execute', { payload: { command: 'map_restart' } }),
|
|
api.add('rcon.execute', { payload: { command: 'status; map_restart' } }),
|
|
api.add('rcon.execute', {
|
|
payload: { command: 'status' },
|
|
result: { ok: false, reason: 'timeout' }
|
|
}),
|
|
api.add('instance.inspect', { result: { healthy: false, state: 'stopped' } }),
|
|
api.add('files.read', { error: 'Read failed' }),
|
|
api.add('files.read', { result: { error: 'Read failed' } }),
|
|
api.add('files.stat', { age: day }),
|
|
api.add('files.read', { age: day - 1 })
|
|
];
|
|
expect(api.pruneHostObservations()).toBe(removed.length);
|
|
const ids = (
|
|
api.handle.prepare('SELECT id FROM ec_host_operations').all() as { id: string }[]
|
|
).map((row) => row.id);
|
|
expect(ids.sort()).toEqual(retained.sort());
|
|
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' }
|
|
});
|
|
const action = api.add('rcon.execute', {
|
|
serialized: true,
|
|
payload: { command: 'map_restart' },
|
|
routing: { command: 'status' }
|
|
});
|
|
expect(
|
|
api.handle.prepare('SELECT COUNT(*) AS count FROM ec_host_operation_payloads').get()
|
|
).toEqual({ count: 2 });
|
|
expect(api.pruneHostObservations()).toBe(1);
|
|
expect(api.getHostOperationStatus(read)).toBeNull();
|
|
expect(api.getHostOperation(action)?.payload).toEqual({ command: 'map_restart' });
|
|
expect(
|
|
api.handle.prepare('SELECT DISTINCT operation_id FROM ec_host_operation_events').all()
|
|
).toEqual([{ operation_id: action }]);
|
|
expect(
|
|
api.handle.prepare('SELECT operation_id FROM ec_host_operation_payloads').all()
|
|
).toEqual([{ operation_id: action }]);
|
|
expect(api.handle.pragma('foreign_key_check')).toEqual([]);
|
|
});
|
|
|
|
it('honors disabled retention and changes to the saved period using the last receipt time', async () => {
|
|
const api = await setup();
|
|
const old = api.add('files.read', { age: 4 * day });
|
|
const recentReceipt = api.add('files.read', { age: 2 * day });
|
|
api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays: 0 });
|
|
expect(api.pruneHostObservations()).toBe(0);
|
|
api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays: 3 });
|
|
expect(api.pruneHostObservations()).toBe(1);
|
|
expect(api.getHostOperationStatus(old)).toBeNull();
|
|
expect(api.getHostOperationStatus(recentReceipt)).not.toBeNull();
|
|
api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays: 1 });
|
|
expect(api.pruneHostObservations()).toBe(1);
|
|
});
|
|
|
|
it('bounds each cleanup batch and seeks through the retention index despite extensive retained action history', async () => {
|
|
const api = await setup();
|
|
api.handle.transaction(() => {
|
|
const insert = api.handle.prepare(`INSERT INTO ec_host_operations
|
|
(id,host_id,kind,payload,state,created_at,updated_at,observation_success)
|
|
VALUES(?,?,?,'{}','completed',?,?,?)`);
|
|
for (let index = 0; index < 2000; index++)
|
|
insert.run(
|
|
`action-${index}`,
|
|
api.host.id,
|
|
'instance.restart',
|
|
at(10 * day),
|
|
at(10 * day),
|
|
0
|
|
);
|
|
for (let index = 0; index < 503; index++)
|
|
insert.run(`read-${index}`, api.host.id, 'files.read', at(2 * day), at(2 * day), 1);
|
|
})();
|
|
const prepare = vi.spyOn(api.handle, 'prepare');
|
|
expect(api.pruneHostObservations()).toBe(500);
|
|
const query = prepare.mock.calls
|
|
.map(([sql]) => sql)
|
|
.find((sql) =>
|
|
sql.includes('SELECT id FROM ec_host_operations WHERE observation_success')
|
|
)!;
|
|
prepare.mockRestore();
|
|
const plan = api.handle.prepare(`EXPLAIN QUERY PLAN ${query}`).all(at(day)) as {
|
|
detail: string;
|
|
}[];
|
|
expect(
|
|
plan.some((row) =>
|
|
row.detail.includes('USING INDEX ec_host_operations_observation_retention')
|
|
)
|
|
).toBe(true);
|
|
expect(plan.some((row) => row.detail.includes('TEMP B-TREE'))).toBe(false);
|
|
expect(api.pruneHostObservations()).toBe(3);
|
|
expect(api.pruneHostObservations()).toBe(0);
|
|
expect(
|
|
api.handle
|
|
.prepare(
|
|
"SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind='instance.restart'"
|
|
)
|
|
.get()
|
|
).toEqual({ count: 2000 });
|
|
});
|
|
|
|
it('uses the same three delete query shapes across different batch sizes', async () => {
|
|
const api = await setup();
|
|
const deleteQueries = new Set<string>();
|
|
for (const count of [1, 2, 7, 50, 500]) {
|
|
api.handle.transaction(() => {
|
|
const insert = api.handle.prepare(`INSERT INTO ec_host_operations
|
|
(id,host_id,kind,payload,state,created_at,updated_at,observation_success)
|
|
VALUES(?,?,'files.stat','{}','completed',?,?,1)`);
|
|
for (let i = 0; i < count; i++)
|
|
insert.run(`read-${count}-${i}`, api.host.id, at(2 * day), at(2 * day));
|
|
})();
|
|
const prepare = vi.spyOn(api.handle, 'prepare');
|
|
expect(api.pruneHostObservations()).toBe(count);
|
|
const deletes = prepare.mock.calls
|
|
.map(([sql]) => sql)
|
|
.filter((sql) => sql.startsWith('DELETE FROM ec_host_'));
|
|
prepare.mockRestore();
|
|
expect(deletes).toHaveLength(3);
|
|
for (const sql of deletes) deleteQueries.add(sql);
|
|
}
|
|
expect(deleteQueries.size).toBe(3);
|
|
expect(api.handle.pragma('foreign_key_check')).toEqual([]);
|
|
});
|
|
|
|
it('backfills legacy observation classification once while preserving actions and negative or invalid results', async () => {
|
|
const api = await setup();
|
|
const read = api.add('files.manifest');
|
|
const serializedRead = api.add('rcon.execute', {
|
|
serialized: true,
|
|
payload: { command: 'status' }
|
|
});
|
|
const retained = [
|
|
api.add('instance.restart'),
|
|
api.add('rcon.execute', { payload: { command: 'map_restart' } }),
|
|
api.add('instance.inspect', { result: { healthy: false } }),
|
|
api.add('files.read', { result: { ok: false } }),
|
|
api.add('files.read', { error: 'Unavailable' }),
|
|
api.add('files.read', { result: { error: 'Unavailable' } }),
|
|
api.add('files.stat', { state: 'unknown' })
|
|
];
|
|
const malformed = api.add('files.read');
|
|
api.handle
|
|
.prepare('UPDATE ec_host_operations SET result=? WHERE id=?')
|
|
.run('{not-json', malformed);
|
|
retained.push(malformed);
|
|
api.handle.exec('DROP INDEX ec_host_operations_observation_retention');
|
|
api.handle.exec('DROP INDEX ec_host_operations_important_history');
|
|
api.handle
|
|
.exec(`DROP TRIGGER operation_history_count_insert; DROP TRIGGER operation_history_count_delete;
|
|
DROP TRIGGER operation_history_count_update; DROP TABLE ec_host_operation_history_counts;
|
|
DELETE FROM ec_host_schema_migrations WHERE name='history-counts-v1'`);
|
|
api.handle.exec('ALTER TABLE ec_host_operations DROP COLUMN observation_success');
|
|
api.closeDb();
|
|
|
|
expect(api.pruneHostObservations()).toBe(2);
|
|
expect(api.getHostOperationStatus(read)).toBeNull();
|
|
expect(api.getHostOperationStatus(serializedRead)).toBeNull();
|
|
const handle = api.db();
|
|
expect(
|
|
(handle.prepare('SELECT id FROM ec_host_operations').all() as { id: string }[])
|
|
.map((row) => row.id)
|
|
.sort()
|
|
).toEqual(retained.sort());
|
|
// Subsequent boots honor the recorded classification instead of repeatedly
|
|
// deriving it from historical rows that were deliberately preserved.
|
|
handle.prepare("UPDATE ec_host_operations SET result='{}' WHERE id=?").run(malformed);
|
|
api.closeDb();
|
|
expect(api.pruneHostObservations()).toBe(0);
|
|
expect(api.getHostOperationStatus(malformed)).not.toBeNull();
|
|
});
|
|
|
|
it('returns the current shared retention setting to an agent on every heartbeat', async () => {
|
|
const api = await setup();
|
|
const { POST } = await import('../../src/routes/api/agent/heartbeat/+server.js');
|
|
for (const observationKeepDays of [7, 0]) {
|
|
api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays });
|
|
const response = await POST({
|
|
request: new Request('https://ec.test/api/agent/heartbeat', {
|
|
method: 'POST',
|
|
headers: {
|
|
authorization: `Bearer ${api.credentials.credential}`,
|
|
'Content-Type': 'application/json'
|
|
},
|
|
body: JSON.stringify({ inventory })
|
|
})
|
|
});
|
|
expect(response.status).toBe(200);
|
|
expect(await response.json()).toMatchObject({
|
|
observationKeepDays,
|
|
heartbeatSeconds: 15
|
|
});
|
|
}
|
|
});
|
|
});
|