Files
electric-cherry/tests/unit/host-status-batch.test.ts

586 lines
24 KiB
TypeScript

import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { newFixture, type Fixture } from '../helpers/fixture.js';
import { gameEvent, postTelemetry } from '../helpers/telemetry.js';
import { defaultPlan, requirement } from '../../src/lib/beta-plan.js';
let fixture: Fixture;
let at: number;
beforeEach(async () => {
vi.resetModules();
vi.resetAllMocks();
fixture = await newFixture([]);
at = Date.now();
vi.useFakeTimers({ toFake: ['Date'] });
vi.setSystemTime(at);
});
afterEach(async () => {
vi.useRealTimers();
await fixture.cleanup();
});
async function setup(count = 5, hostCount = 1) {
const hosts = await import('../../src/lib/server/hosts/index.js');
const { db } = await import('../../src/lib/server/db/index.js');
vi.spyOn(await import('../../src/lib/server/backends/native.js'), 'nativeHealth');
const { nativeHealth } = await import('../../src/lib/server/backends/native.js');
vi.mocked(nativeHealth).mockResolvedValue({ ok: true, state: 'running' });
const hostIds: string[] = [];
for (let h = 0; h < hostCount; h++) {
const host = hosts.createHost({ name: `Host ${h}` });
hostIds.push(host.id);
const inventory = {
protocolVersion: 1,
capabilities: [
'instance.heartbeat-state',
'rcon.status-batch',
'operations.health-lane'
],
instanceStates: Array.from({ length: count }, (_, i) => ({
instanceId: `instance-${h}-${i}`,
generation: 1,
healthy: true,
state: 'running',
observedAt: new Date(at).toISOString()
}))
};
hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory);
for (let i = 0; i < count; i++) {
const name = `server-${h}-${i}`;
db()
.prepare('INSERT INTO servers(name,created_at) VALUES(?,?)')
.run(name, new Date(at - 1000).toISOString());
db()
.prepare(
`INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at)
VALUES(?,?,?,'library','image',?,?)`
)
.run(
`instance-${h}-${i}`,
name,
host.id,
28960 + i,
new Date(at - 1000).toISOString()
);
}
}
const health = await import('../../src/lib/server/fleet/health-observations.js');
const worker = await import('../../src/lib/server/pipeline/worker.js');
const { getServer } = await import('../../src/lib/server/fleet/servers.js');
const operations = () => hostIds.flatMap((id) => hosts.listHostOperations(id));
const complete = () =>
operations()
.filter((op) => !['completed', 'failed'].includes(op.state))
.forEach((op) => {
const targets = op.payload.instances as {
instanceId: string;
generation: number;
}[];
hosts.reportHostOperation(op.hostId, op.id, {
state: 'completed',
result:
op.kind === 'rcon.status-batch'
? {
ok: true,
results: targets.map((target) => ({
...target,
reply: {
ok: true,
text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n'
}
}))
}
: {
ok: true,
text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n'
}
});
});
return {
...hosts,
...health,
...worker,
getServer,
db: db(),
inspect: vi.mocked(nativeHealth),
hostIds,
operations,
complete
};
}
describe('host status observations', () => {
async function emptyHandoff() {
const api = await setup(1);
const server = 'server-0-0';
const { serverObservation } = await import('../../src/lib/server/fleet/status.js');
const { createScript } = await import('../../src/lib/server/fleet/scripts.js');
const { currentRun, readMetrics } =
await import('../../src/lib/server/fleet/beta-tests.js');
const { plannedBetaTick } =
await import('../../src/lib/server/pipeline/beta-plan-runner.js');
const { script, version } = createScript({
releaseVersion: '1.0.0',
name: 'Handoff test',
targetRelPath: 'handoff.gsc',
content: Buffer.from('main() {}')
});
const betaId = Number(
api.db
.prepare(
`INSERT INTO betas(script_id,script_version_id,trigger_kind,state,opened_at)
VALUES(?,?,'manual','ready-promote',?)`
)
.run(script.id, version.id, new Date(at).toISOString()).lastInsertRowid
);
api.db
.prepare(
`INSERT INTO beta_servers(beta_id,server,had_original,stage,restarted_at)
VALUES(?,?,0,'ready',?)`
)
.run(betaId, server, new Date(at).toISOString());
api.db
.prepare(`INSERT INTO beta_test_plans(beta_id,plan,outcome) VALUES(?,?,'passed')`)
.run(
betaId,
JSON.stringify({
...defaultPlan(),
requirements: [
requirement('healthy-minutes', 'health'),
requirement('failure-budget', 'outages')
]
})
);
api.db
.prepare(
`INSERT INTO beta_test_runs(id,beta_id,server,created_at,observing_at,metrics)
VALUES('handoff',?,?,?,?,?)`
)
.run(
betaId,
server,
new Date(at).toISOString(),
new Date(at).toISOString(),
JSON.stringify({
values: { health: 1 },
manual: {},
streaks: {},
outages: 0,
lastSample: at,
wasHealthy: true
})
);
await postTelemetry(gameEvent('heartbeat', 15, 1, server));
const occupied = api.fanoutStatus();
await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
api.complete();
await occupied;
vi.setSystemTime(at + 30_000);
await postTelemetry(gameEvent('heartbeat', 0, 0, server));
await api.fanoutStatus();
await plannedBetaTick();
const emptyAt = Date.now();
const observe = () => serverObservation(api.getServer(server)!);
const metrics = () => readMetrics(currentRun(betaId, server)!);
const expectReady = async () => {
await plannedBetaTick();
expect(observe().health).toBe('online');
expect(metrics()).toMatchObject({ wasHealthy: true, outages: 0 });
expect(metrics().values.health).toBeGreaterThanOrEqual(1.5);
expect(api.db.prepare('SELECT state FROM betas WHERE id=?').get(betaId)).toEqual({
state: 'ready-promote'
});
expect(
api.db.prepare('SELECT outcome FROM beta_test_plans WHERE beta_id=?').get(betaId)
).toEqual({ outcome: 'passed' });
};
const setAge = (age: number) => {
vi.setSystemTime(emptyAt + age);
const host = api.getHost(api.hostIds[0])!;
api.heartbeat(host.id, {
...host.inventory,
instanceStates: host.inventory.instanceStates!.map((state) => ({
...state,
observedAt: new Date().toISOString()
}))
});
};
return { ...api, emptyAt, observe, metrics, expectReady, plannedBetaTick, setAge };
}
it.each([30_000, 59_999])(
'preserves beta readiness when the empty handoff poll starts at %i ms and replies 25 seconds later',
async (pollAt) => {
const api = await emptyHandoff();
api.setAge(pollAt - 30_000);
await api.fanoutStatus();
expect(api.operations()).toHaveLength(1);
await api.expectReady();
api.setAge(pollAt);
const pass = api.fanoutStatus();
try {
await vi.waitFor(() => expect(api.operations()).toHaveLength(2));
expect(api.operations()[0].kind).toBe('rcon.status-batch');
api.setAge(pollAt + 24_999);
await api.expectReady();
} finally {
api.setAge(pollAt + 25_000);
api.complete();
await pass;
}
await api.expectReady();
expect(api.metrics().lastSample).toBe(api.emptyAt + pollAt + 25_000);
expect(api.observe().status_source).toBe('telemetry');
for (const age of [90_000, 90_001]) {
api.setAge(age);
await api.expectReady();
}
expect(api.observe().status_source).toBe('rcon');
expect(api.observe().telemetry?.receivedAt).toBe(new Date(api.emptyAt).toISOString());
}
);
it.each(['pending', 'failed', 'skipped'] as const)(
'expires empty telemetry on schedule with a %s RCON handoff',
async (result) => {
const api = await emptyHandoff();
api.setAge(59_999);
const pass = api.fanoutStatus();
try {
await vi.waitFor(() => expect(api.operations()).toHaveLength(2));
if (result !== 'pending') {
api.setAge(84_999);
const operation = api.operations().find((op) => op.state === 'pending')!;
api.reportHostOperation(
operation.hostId,
operation.id,
result === 'failed'
? {
state: 'failed',
error: 'agent did not respond'
}
: {
state: 'completed',
result: {
results: [
{
instanceId: 'instance-0-0',
generation: 1,
reply: { ok: false, text: '', reason: 'deadline' }
}
]
}
}
);
await pass;
}
api.setAge(90_000);
await api.expectReady();
api.setAge(90_001);
expect(api.observe().health).not.toBe('online');
await api.plannedBetaTick();
expect(api.metrics()).toMatchObject({
wasHealthy: false,
outages: 1,
values: { health: 0 }
});
} finally {
api.complete();
await pass;
}
}
);
it('batches occupied telemetry lobbies for pings while skipping empty lobbies', async () => {
const api = await setup(2);
await postTelemetry(gameEvent('heartbeat', 15, 1, 'server-0-0'));
await postTelemetry(gameEvent('heartbeat', 15, 0, 'server-0-1'));
const pass = api.fanoutStatus();
try {
await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
expect(api.operations()[0]).toMatchObject({
kind: 'rcon.status-batch',
payload: { instances: [{ instanceId: 'instance-0-0', generation: 1 }] }
});
} finally {
api.complete();
await pass;
}
expect(api.getServer('server-0-1')?.last_seen_at).toBeNull();
});
it('bounds monitoring reads below five per server across thirty instances', async () => {
const api = await setup(10, 3);
const hosts = await import('../../src/lib/server/hosts/index.js');
const wait = vi.spyOn(hosts, 'waitHostOperationStatus').mockImplementation(async (id) => {
const operation = hosts.getHostOperation(id)!;
return {
...operation,
state: 'completed',
result: {
results: (
operation.payload.instances as { instanceId: string; generation: number }[]
).map((target) => ({
...target,
reply: {
ok: true,
text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n'
}
}))
}
};
});
const prepare = vi.spyOn(api.db, 'prepare');
try {
await api.fanoutStatus();
const reads = prepare.mock.calls.filter(([sql]) => /^\s*(SELECT|WITH)\b/i.test(sql));
expect(reads.length).toBeLessThan(30 * 5);
expect(
reads.filter(([sql]) => sql.includes('instance_json')).length
).toBeLessThanOrEqual(30 * 3 + 5);
expect(api.getServer('server-2-9')?.last_players).toBe(0);
} finally {
prepare.mockRestore();
wait.mockRestore();
}
});
it('uses one batch per host for a fifteen-server pass and shares overlapping scans', async () => {
const api = await setup(5, 3);
const hosts = await import('../../src/lib/server/hosts/index.js');
const wait = vi.spyOn(hosts, 'waitHostOperationStatus');
const passes = [api.fanoutStatus(), api.fanoutStatus()];
try {
await vi.waitFor(() => expect(api.operations()).toHaveLength(3));
for (const operation of api.operations()) {
expect(operation.kind).toBe('rcon.status-batch');
expect(wait).toHaveBeenCalledWith(operation.id, 25_000);
expect(operation.payload.instances).toHaveLength(5);
expect(
api.pendingHostOperations(operation.hostId, 'health').map((op) => op.id)
).toEqual([operation.id]);
expect(api.pendingHostOperations(operation.hostId, 'write')).toEqual([]);
}
} finally {
api.complete();
await Promise.all(passes);
}
expect(api.inspect).not.toHaveBeenCalled();
expect(api.getServer('server-2-4')).toMatchObject({
last_players: 0,
last_map: 'mp_kino',
container_checked_at: new Date(at).toISOString()
});
expect(api.monitoringStatus('server-2-4')).toMatchObject({
container: { observedAt: new Date(at).toISOString(), result: { ok: true } },
game: { observedAt: api.getServer('server-2-4')?.last_seen_at, result: { ok: true } }
});
expect(api.db.prepare('SELECT COUNT(*) AS n FROM metric_samples').get()).toEqual({ n: 15 });
await api.fanoutStatus();
expect(api.operations()).toHaveLength(3);
});
it.each(['stale', 'future', 'generation', 'lifecycle', 'missing', 'legacy'])(
'falls back to inspection for %s heartbeat evidence',
async (mode) => {
const api = await setup(1);
const host = api.getHost(api.hostIds[0])!;
const inventory = structuredClone(host.inventory);
if (mode === 'stale')
inventory.instanceStates![0].observedAt = new Date(at - 30_000).toISOString();
if (mode === 'future')
inventory.instanceStates![0].observedAt = new Date(at + 1_000).toISOString();
if (mode === 'generation') inventory.instanceStates![0].generation = 2;
if (mode === 'missing') inventory.instanceStates = [];
if (mode === 'legacy') inventory.capabilities = [];
if (mode === 'lifecycle')
api.db
.prepare(
`INSERT INTO server_deployments(id,server,action,state,stage,started_at,submitted_at,finished_at,triggered_by)
VALUES('stop','server-0-0','stop','done','done','started',?,?,'test')`
)
.run(new Date(at + 1).toISOString(), new Date(at + 2).toISOString());
api.heartbeat(host.id, inventory);
await api.inspectForMonitoring('server-0-0');
expect(api.inspect).toHaveBeenCalledOnce();
}
);
it('retains heartbeat sample time, reports stopped containers and keeps forced probes fresh', async () => {
const api = await setup(1);
vi.setSystemTime(at + 10_000);
expect(await api.inspectForMonitoring('server-0-0')).toMatchObject({
observedAt: new Date(at).toISOString(),
result: { ok: true }
});
await api.inspectForMonitoring('server-0-0', true);
expect(api.inspect).toHaveBeenCalledOnce();
const host = api.getHost(api.hostIds[0])!;
api.heartbeat(host.id, {
...host.inventory,
instanceStates: [
{ ...host.inventory.instanceStates![0], healthy: false, state: 'exited' }
]
});
await api.fanoutStatus();
expect(api.getServer('server-0-0')?.container_state).toBe('exited');
expect(api.monitoringStatus('server-0-0')).toMatchObject({
container: {
observedAt: new Date(at).toISOString(),
result: { ok: false, state: 'exited' }
}
});
expect(api.operations()).toEqual([]);
});
it.each(['generation', 'manual', 'lifecycle'])(
'discards batch replies overtaken by a %s change, including queued servers',
async (change) => {
const api = await setup();
const pass = api.fanoutStatus();
try {
await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
vi.setSystemTime(at + 1000);
if (change === 'generation')
api.db.prepare('UPDATE native_instances SET generation=2').run();
if (change === 'manual')
api.db
.prepare("UPDATE servers SET last_checked_at=?,last_error='timeout:manual'")
.run(new Date().toISOString());
if (change === 'lifecycle')
api.db
.prepare(
`INSERT INTO server_deployments(id,server,action,state,stage,started_at,submitted_at,triggered_by)
SELECT name,name,'stop','done','done','started',?,'test' FROM servers`
)
.run(new Date().toISOString());
} finally {
api.complete();
await pass;
}
for (let i = 0; i < 5; i++)
expect(api.getServer(`server-0-${i}`)?.last_seen_at).toBeNull();
expect(api.operations()).toHaveLength(1);
}
);
it('preserves previous game evidence for deadline skips while recording actual timeouts', async () => {
const api = await setup(3);
const first = api.fanoutStatus();
await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
api.complete();
await first;
api.db
.prepare("UPDATE servers SET rcon_failures=1,last_error='timeout:first failure'")
.run();
const previous = api.getServer('server-0-0')!;
const previousGame = api.monitoringStatus('server-0-0')?.game;
vi.setSystemTime(at + 31_000);
const pass = api.fanoutStatus();
try {
await vi.waitFor(() => expect(api.operations()).toHaveLength(2));
const operation = api.operations().find((op) => op.state !== 'completed')!;
api.reportHostOperation(operation.hostId, operation.id, {
state: 'completed',
result: {
ok: false,
results: [
{
instanceId: 'instance-0-0',
generation: 1,
reply: { ok: false, reason: 'deadline', text: '' }
},
{
instanceId: 'instance-0-1',
generation: 1,
reply: { ok: false, reason: 'timeout', text: '' }
},
{
instanceId: 'instance-0-2',
generation: 1,
reply: {
ok: true,
text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n'
}
}
]
}
});
} finally {
api.complete();
await pass;
}
expect(api.getServer('server-0-0')).toMatchObject({
last_checked_at: previous.last_checked_at,
last_seen_at: previous.last_seen_at,
last_error: previous.last_error,
last_players: previous.last_players,
last_reply_snippet: previous.last_reply_snippet,
rcon_failures: 1
});
expect(api.monitoringStatus('server-0-0')?.game).toEqual(previousGame);
expect(api.getServer('server-0-1')?.last_error).toContain('timeout:');
expect(api.getServer('server-0-1')?.rcon_failures).toBe(2);
expect(api.getServer('server-0-2')?.rcon_failures).toBe(0);
expect(Date.parse(api.getServer('server-0-2')!.last_seen_at!)).toBeGreaterThanOrEqual(
at + 31_000
);
});
it('records each failed or omitted result without retrying the batch per server', async () => {
const api = await setup();
const pass = api.fanoutStatus();
try {
await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
const operation = api.operations()[0];
api.reportHostOperation(operation.hostId, operation.id, {
state: 'completed',
result: {
ok: false,
results: [
{
instanceId: 'instance-0-0',
generation: 1,
reply: { ok: false, reason: 'authentication', text: '' }
}
]
}
});
} finally {
api.complete();
await pass;
}
expect(api.getServer('server-0-0')?.last_error).toBe('bad-reply:wrong RCON password');
expect(api.getServer('server-0-4')?.last_error).toContain('timeout:');
expect(api.monitoringStatus('server-0-4')?.game?.result).toMatchObject({
ok: false,
reason: 'timeout'
});
expect(api.operations()).toHaveLength(1);
});
it.each([true, false])(
'uses batch capability=%s while missing heartbeat state still gets inspected',
async (capable) => {
const api = await setup(1);
const host = api.getHost(api.hostIds[0])!;
api.heartbeat(host.id, {
...host.inventory,
instanceStates: [],
capabilities: capable ? ['rcon.status-batch'] : []
});
const pass = api.fanoutStatus();
try {
await vi.waitFor(() => expect(api.operations()).toHaveLength(1));
expect(api.operations()[0].kind).toBe(
capable ? 'rcon.status-batch' : 'rcon.execute'
);
} finally {
api.complete();
await pass;
}
expect(api.inspect).toHaveBeenCalledOnce();
expect(api.getServer('server-0-0')?.last_players).toBe(0);
}
);
});