diff --git a/docs/guides/native-hosts.md b/docs/guides/native-hosts.md index 64a09e9..a89649e 100644 --- a/docs/guides/native-hosts.md +++ b/docs/guides/native-hosts.md @@ -130,8 +130,11 @@ Agent 1.8.5 adds incremental game-log reads for chat collection. After the initi tail, the controller requests only appended bytes, handles rotation and truncation, and buffers incomplete lines. Unchanged logs need no content transfer or chat parsing. Background collection skips offline hosts, stopped servers and known -empty servers, with one final read after players leave. Opening chat still reads -on demand. Older agents use tail reads until they update. +empty servers, with one final read after players leave. Opening chat reuses the +worker's saved reader state and only starts an on-demand read when the previous +read completed at least 15 seconds ago. This limit is shared across viewers and also +applies to missing logs and failed reads. Older agents use tail reads until they +update. ### New game files and cached versions diff --git a/src/lib/server/auth/access.ts b/src/lib/server/auth/access.ts index b438894..cb17cca 100644 --- a/src/lib/server/auth/access.ts +++ b/src/lib/server/auth/access.ts @@ -78,7 +78,12 @@ export function routePermission(routeId: string | null, method: string): RoutePe return 'admin'; // These reads initiate fleet file scans or spend external API quota. - if (routeId === '/api/drift' || routeId === '/api/plugins/github/release') return 'operator'; + if ( + routeId === '/api/drift' || + routeId === '/api/plugins/github/release' || + routeId === '/api/plugins/updates' + ) + return 'operator'; // Summaries remain visible to viewers; raw configuration, historical file // bytes, and unfiltered diagnostics may contain runtime credentials. diff --git a/src/lib/server/fleet/chat.ts b/src/lib/server/fleet/chat.ts index 6feb135..dfd8710 100644 --- a/src/lib/server/fleet/chat.ts +++ b/src/lib/server/fleet/chat.ts @@ -209,12 +209,38 @@ interface ChatReadState { } const readers = new Map(); const failures = new Map(); +const CHAT_READ_MIN_AGE_MS = 15_000; +type LastChatRead = { completedAt: number } & ({ missing: boolean } | { error: unknown }); +const lastReads = new Map(); + +function chatReadKey(server: ServerRow): string { + return `${server.name}:${server.created_at}:${server.native_instance_id}:${server.instance_generation}`; +} + +/** Live viewers reuse the worker's reader state instead of adding a read per tab. */ +export async function readServerChat( + server: ServerRow +): Promise<{ entries: ContextualChatEntry[]; missing: boolean }> { + const key = chatReadKey(server); + const existing = pending.get(key); + if (existing) return existing; + const last = lastReads.get(key); + if (!last || Date.now() - last.completedAt >= CHAT_READ_MIN_AGE_MS) + return collectServerChat(server); + if ('error' in last) throw last.error; + const state = readers.get(server.name); + return { + // Telemetry may have started a new match since the last log read. + entries: currentChatEntries(server, state?.key === key ? state.entries : []), + missing: last.missing + }; +} /** Shared by the live API and the worker; collection does not require a viewer. */ export function collectServerChat( server: ServerRow ): Promise<{ entries: ContextualChatEntry[]; missing: boolean }> { - const key = `${server.name}:${server.created_at}:${server.native_instance_id}:${server.instance_generation}`; + const key = chatReadKey(server); const existing = pending.get(key); if (existing) return existing; const work = (async () => { @@ -319,7 +345,19 @@ export function collectServerChat( state.signature = signature; readers.set(server.name, state); return { entries: currentChatEntries(current, state.entries), missing: false }; - })().finally(() => pending.delete(key)); + })() + .then( + (result) => { + lastReads.set(key, { completedAt: Date.now(), missing: result.missing }); + return result; + }, + (error: unknown) => { + // Failed reads also observe the cooldown, without advancing the cursor. + lastReads.set(key, { completedAt: Date.now(), error }); + throw error; + } + ) + .finally(() => pending.delete(key)); pending.set(key, work); return work; } @@ -329,8 +367,10 @@ export async function collectFleetChat(): Promise { const observe = observationReader(); const live = listServers().filter((server) => !server.destroyed_at); const names = new Set(live.map((server) => server.name)); + const keys = new Set(live.map(chatReadKey)); for (const name of readers.keys()) if (!names.has(name)) readers.delete(name); for (const name of failures.keys()) if (!names.has(name)) failures.delete(name); + for (const key of lastReads.keys()) if (!keys.has(key)) lastReads.delete(key); const servers = live.filter((server) => { if (server.host_id && !hosts.get(server.host_id)?.connected) return false; if (server.container_state && !['running', 'unknown'].includes(server.container_state)) diff --git a/src/routes/api/servers/[name]/chat/+server.ts b/src/routes/api/servers/[name]/chat/+server.ts index 2c99881..22bba5b 100644 --- a/src/routes/api/servers/[name]/chat/+server.ts +++ b/src/routes/api/servers/[name]/chat/+server.ts @@ -1,7 +1,7 @@ import { json } from '@sveltejs/kit'; import { getServer } from '$lib/server/fleet/servers.js'; import { gameLogKey, resolveLogPath } from '$lib/server/fleet/logs.js'; -import { collectServerChat } from '$lib/server/fleet/chat.js'; +import { readServerChat } from '$lib/server/fleet/chat.js'; export async function GET({ params }: { params: { name: string } }) { const headers = { 'Cache-Control': 'no-store' }; @@ -14,7 +14,7 @@ export async function GET({ params }: { params: { name: string } }) { return json({ ok: false, reason: 'Path traversal rejected' }, { status: 400, headers }); try { - return json({ ok: true, ...(await collectServerChat(server)) }, { headers }); + return json({ ok: true, ...(await readServerChat(server)) }, { headers }); } catch (error) { console.error('Could not read server chat', error); return json({ ok: false, reason: 'Could not read server chat' }, { status: 500, headers }); diff --git a/tests/unit/auth-access.test.ts b/tests/unit/auth-access.test.ts index 30885b5..b487be0 100644 --- a/tests/unit/auth-access.test.ts +++ b/tests/unit/auth-access.test.ts @@ -82,7 +82,7 @@ describe('management route permissions', () => { expect(allows('super_viewer', '/api/settings/retention/preview', 'POST')).toBe(true); expect(allows('operator', '/api/settings/retention/preview', 'POST')).toBe(false); expect(allows('super_viewer', '/api/settings/retention', 'POST')).toBe(false); - for (const route of ['/api/drift', '/api/plugins/github/release']) + for (const route of ['/api/drift', '/api/plugins/github/release', '/api/plugins/updates']) for (const method of ['GET', 'HEAD']) expect(allows('super_viewer', route, method)).toBe(false); }); diff --git a/tests/unit/auth-guard.test.ts b/tests/unit/auth-guard.test.ts index b0d02b0..82c34a8 100644 --- a/tests/unit/auth-guard.test.ts +++ b/tests/unit/auth-guard.test.ts @@ -339,13 +339,15 @@ describe('browser authentication guard', () => { ['GET', '/api/drift'], ['HEAD', '/api/drift'], ['GET', '/api/plugins/github/release'], - ['HEAD', '/api/plugins/github/release'] + ['HEAD', '/api/plugins/github/release'], + ['GET', '/api/plugins/updates'], + ['HEAD', '/api/plugins/updates'] ])('restricts expensive %s %s requests before handler execution', async (method, route) => { const anonymous = requestFor(route, { method, cookie: null }); expect((await anonymous.invoke()).status).toBe(401); expect(anonymous.resolve).not.toHaveBeenCalled(); - for (const role of ['viewer', 'super_viewer', 'operator', 'admin'] as const) { + for (const role of ['viewer', 'super_viewer', 'operator', 'moderator', 'admin'] as const) { signIn(role); const call = requestFor(route, { method }); const allowed = role === 'operator' || role === 'admin'; diff --git a/tests/unit/server-chat-refresh.test.ts b/tests/unit/server-chat-refresh.test.ts new file mode 100644 index 0000000..f0c75da --- /dev/null +++ b/tests/unit/server-chat-refresh.test.ts @@ -0,0 +1,228 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { newFixture, type Fixture } from '../helpers/fixture.js'; +import type { ContextualChatEntry } from '../../src/lib/game-chat.js'; +import type { TailResult } from '../../src/lib/server/fleet/logs.js'; + +const SERVER = 't5_zombie_theater_1'; +const OTHER_SERVER = 't6_zombie_town_1'; +let fixture: Fixture; +let baseline: number; + +beforeEach(async () => { + vi.resetModules(); + vi.useFakeTimers({ toFake: ['Date'] }); + baseline = Date.now() - 600_000; + fixture = await newFixture([SERVER, OTHER_SERVER]); + const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js'); + const { db } = await import('../../src/lib/server/db/index.js'); + await seedServerRegistry(); + db() + .prepare("UPDATE servers SET created_at=?, last_players=0, container_state='running'") + .run(new Date(baseline - 120_000).toISOString()); +}); + +afterEach(async () => { + vi.restoreAllMocks(); + vi.useRealTimers(); + await fixture.cleanup(); +}); + +function tail(message: string): TailResult { + const line = `1:00 say;abc;0;Player;${message}`; + return { + lines: [line], + size: Buffer.byteLength(line) + 1, + mtime: new Date(baseline + 60_000).toISOString(), + truncated: false + }; +} + +function messages(body: { entries: ContextualChatEntry[] }): string[] { + return body.entries.flatMap((entry) => + entry.type === 'message' ? [entry.message.message] : [] + ); +} + +function advance(milliseconds: number) { + vi.setSystemTime(Date.now() + milliseconds); +} + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +async function setup() { + const logs = await import('../../src/lib/server/fleet/logs.js'); + const read = vi.spyOn(logs, 'readLogChunk').mockResolvedValue(tail('first')); + const { GET } = await import('../../src/routes/api/servers/[name]/chat/+server.js'); + const { db } = await import('../../src/lib/server/db/index.js'); + const { collectFleetChat } = await import('../../src/lib/server/fleet/chat.js'); + return { + read, + db: db(), + collectFleetChat, + get: (server = SERVER) => GET({ params: { name: server } }) + }; +} + +async function addInstance() { + const { createHost } = await import('../../src/lib/server/hosts/index.js'); + const { db } = await import('../../src/lib/server/db/index.js'); + const host = createHost({ name: 'Chat host' }); + db() + .prepare( + `INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at) + VALUES('chat-instance',?,?,'library','image',28960,?)` + ) + .run(SERVER, host.id, new Date(baseline).toISOString()); +} + +describe('server chat refresh sharing', () => { + it('serves worker state to repeated viewers and still collects after players leave', async () => { + const s = await setup(); + s.db.prepare('UPDATE servers SET last_players=1 WHERE name=?').run(SERVER); + await s.collectFleetChat(); + expect(s.read).toHaveBeenCalledTimes(1); + expect(messages(await (await s.get()).json())).toEqual(['first']); + const responses = await Promise.all(Array.from({ length: 5 }, () => s.get())); + for (const response of responses) + expect(messages(await response.json())).toEqual(['first']); + expect(s.read).toHaveBeenCalledTimes(1); + + s.db.prepare('UPDATE servers SET last_players=0 WHERE name=?').run(SERVER); + s.read.mockResolvedValue(tail('last message')); + await s.collectFleetChat(); + await s.collectFleetChat(); + expect(s.read).toHaveBeenCalledTimes(2); + expect(messages(await (await s.get()).json())).toEqual(['last message']); + expect(s.read).toHaveBeenCalledTimes(2); + }); + + it('refreshes exactly when the completed read becomes fifteen seconds old', async () => { + const s = await setup(); + expect(messages(await (await s.get()).json())).toEqual(['first']); + s.read.mockResolvedValue(tail('second message')); + advance(14_999); + expect(messages(await (await s.get()).json())).toEqual(['first']); + expect(s.read).toHaveBeenCalledTimes(1); + advance(1); + const responses = await Promise.all([s.get(), s.get(), s.get()]); + for (const response of responses) + expect(messages(await response.json())).toEqual(['second message']); + expect(s.read).toHaveBeenCalledTimes(2); + }); + + it('shares slow worker reads and starts the cooldown when they complete', async () => { + const s = await setup(); + const slow = deferred(); + s.read.mockImplementationOnce(() => slow.promise); + s.db.prepare('UPDATE servers SET last_players=1 WHERE name=?').run(SERVER); + const collection = s.collectFleetChat(); + const first = s.get(); + advance(60_000); + const second = s.get(); + expect(s.read).toHaveBeenCalledTimes(1); + slow.resolve(tail('slow message')); + await collection; + for (const response of await Promise.all([first, second])) + expect(messages(await response.json())).toEqual(['slow message']); + + advance(14_999); + expect(messages(await (await s.get()).json())).toEqual(['slow message']); + expect(s.read).toHaveBeenCalledTimes(1); + advance(1); + await s.get(); + expect(s.read).toHaveBeenCalledTimes(2); + }); + + it('caches missing logs until the next allowed attempt', async () => { + const s = await setup(); + s.read.mockResolvedValueOnce(null); + const missing = { ok: true, entries: [], missing: true }; + expect(await (await s.get()).json()).toEqual(missing); + advance(14_999); + for (const response of await Promise.all([s.get(), s.get()])) + expect(await response.json()).toEqual(missing); + expect(s.read).toHaveBeenCalledTimes(1); + advance(1); + expect(messages(await (await s.get()).json())).toEqual(['first']); + expect(s.read).toHaveBeenCalledTimes(2); + }); + + it('backs off failed reads and retries without leaking the original error', async () => { + const s = await setup(); + s.read.mockRejectedValueOnce(new Error('read failed at /private/server/log')); + vi.spyOn(console, 'error').mockImplementation(() => {}); + const first = await s.get(); + expect(first.status).toBe(500); + expect(await first.json()).toEqual({ ok: false, reason: 'Could not read server chat' }); + advance(14_999); + for (const response of await Promise.all([s.get(), s.get()])) { + expect(response.status).toBe(500); + expect(await response.json()).toEqual({ + ok: false, + reason: 'Could not read server chat' + }); + } + expect(s.read).toHaveBeenCalledTimes(1); + advance(1); + expect(messages(await (await s.get()).json())).toEqual(['first']); + expect(s.read).toHaveBeenCalledTimes(2); + }); + + it('keeps each server and every replacement identity on its own cooldown', async () => { + const s = await setup(); + await addInstance(); + expect(messages(await (await s.get()).json())).toEqual(['first']); + s.read.mockResolvedValue(tail('other server')); + expect(messages(await (await s.get(OTHER_SERVER)).json())).toEqual(['other server']); + expect(messages(await (await s.get()).json())).toEqual(['first']); + expect(s.read).toHaveBeenCalledTimes(2); + + s.db + .prepare('UPDATE native_instances SET generation=generation+1 WHERE server=?') + .run(SERVER); + s.read.mockResolvedValue(tail('new generation')); + expect(messages(await (await s.get()).json())).toEqual(['new generation']); + s.db + .prepare("UPDATE native_instances SET id='replacement-instance' WHERE server=?") + .run(SERVER); + s.read.mockResolvedValue(tail('new instance')); + expect(messages(await (await s.get()).json())).toEqual(['new instance']); + s.db + .prepare('UPDATE servers SET created_at=? WHERE name=?') + .run(new Date(baseline - 60_000).toISOString(), SERVER); + s.read.mockResolvedValue(tail('new registration')); + expect(messages(await (await s.get()).json())).toEqual(['new registration']); + expect(messages(await (await s.get(OTHER_SERVER)).json())).toEqual(['other server']); + expect(s.read).toHaveBeenCalledTimes(5); + }); + + it('does not let an old in-flight generation overwrite its replacement cache', async () => { + const s = await setup(); + await addInstance(); + const oldRead = deferred(); + s.read.mockImplementationOnce(() => oldRead.promise); + const oldResponse = s.get(); + expect(s.read).toHaveBeenCalledTimes(1); + s.db + .prepare('UPDATE native_instances SET generation=generation+1 WHERE server=?') + .run(SERVER); + s.read.mockResolvedValue(tail('replacement')); + expect(messages(await (await s.get()).json())).toEqual(['replacement']); + advance(10_000); + oldRead.resolve(tail('old generation')); + expect(await (await oldResponse).json()).toEqual({ ok: true, entries: [], missing: true }); + expect(messages(await (await s.get()).json())).toEqual(['replacement']); + expect(s.read).toHaveBeenCalledTimes(2); + + advance(5_000); + s.read.mockResolvedValue(tail('replacement refreshed')); + expect(messages(await (await s.get()).json())).toEqual(['replacement refreshed']); + expect(s.read).toHaveBeenCalledTimes(3); + }); +}); diff --git a/tests/unit/server-chat.test.ts b/tests/unit/server-chat.test.ts index f5975dd..6bf8f6f 100644 --- a/tests/unit/server-chat.test.ts +++ b/tests/unit/server-chat.test.ts @@ -16,6 +16,7 @@ const at = (seconds: number) => new Date(baseline + seconds * 1000).toISOString( beforeEach(async () => { vi.resetModules(); + vi.useFakeTimers({ toFake: ['Date'] }); baseline = Math.floor(Date.now() / 1000) * 1000 - 600_000; fixture = await newFixture([SERVER, T6_SERVER]); const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js'); @@ -26,6 +27,7 @@ beforeEach(async () => { afterEach(async () => { vi.restoreAllMocks(); + vi.useRealTimers(); await fixture.cleanup(); }); @@ -70,6 +72,10 @@ async function get(server = SERVER) { return GET({ params: { name: server } }); } +function advanceTime(milliseconds = 15_000) { + vi.setSystemTime(Date.now() + milliseconds); +} + describe('server chat API', () => { it('buffers partial UTF-8 lines and avoids context queries when the log is unchanged', async () => { await report('game_start', 0); @@ -78,6 +84,7 @@ describe('server chat API', () => { expect((await (await get()).json()).entries).toHaveLength(1); const { db } = await import('../../src/lib/server/db/index.js'); const prepare = vi.spyOn(db(), 'prepare'); + advanceTime(); await get(); expect(prepare.mock.calls.some(([sql]) => sql.includes('FROM telemetry_requests'))).toBe( false @@ -90,9 +97,11 @@ describe('server chat API', () => { const line = Buffer.from('61:00 say;abc;0;Player;café\n'); await appendFile(path, line.subarray(0, -2)); // Split inside é, before the newline. await utimes(path, new Date(at(60)), new Date(at(60))); + advanceTime(); expect((await (await get()).json()).entries).toHaveLength(1); await appendFile(path, line.subarray(-2)); await utimes(path, new Date(at(60)), new Date(at(60))); + advanceTime(); const body = await (await get()).json(); expect(body.entries.map((entry: ChatMessageEntry) => entry.message.message)).toEqual([ 'first', @@ -110,6 +119,7 @@ describe('server chat API', () => { const path = join(fixture.t5, SERVER, 'logs/games_sp.log'); await appendFile(path, '61:00 say;abc;0;Player;retry me\n'); await utimes(path, new Date(at(60)), new Date(at(60))); + advanceTime(); const { db } = await import('../../src/lib/server/db/index.js'); const handle = db(), original = handle.prepare.bind(handle); @@ -121,6 +131,7 @@ describe('server chat API', () => { vi.spyOn(console, 'error').mockImplementation(() => {}); expect((await get()).status).toBe(500); prepare.mockRestore(); + advanceTime(); expect( (await (await get()).json()).entries.map( (entry: ChatMessageEntry) => entry.message.message @@ -142,9 +153,14 @@ describe('server chat API', () => { await collectFleetChat(); expect(read).toHaveBeenCalledTimes(1); db().prepare('UPDATE servers SET last_players=0 WHERE name=?').run(SERVER); + await log(['60:50 say;abc;0;Player;playing', '61:00 say;abc;0;Player;goodbye']); + expect((await (await get()).json()).entries).toHaveLength(1); + expect(read).toHaveBeenCalledTimes(1); await collectFleetChat(); await collectFleetChat(); expect(read).toHaveBeenCalledTimes(2); + expect((await (await get()).json()).entries).toHaveLength(2); + expect(read).toHaveBeenCalledTimes(2); db() .prepare("UPDATE servers SET last_players=1,container_state='stopped' WHERE name=?") .run(SERVER); @@ -280,6 +296,8 @@ describe('server chat API', () => { }); it('clears at a new match before it writes to the log and keeps the old chat with its match', async () => { + const logs = await import('../../src/lib/server/fleet/logs.js'); + const read = vi.spyOn(logs, 'readLogChunk'); await report('game_start', 0); await report('round_change', 20, 4); await log(['60:30 say;abc;0;Player;old match'], 30); @@ -287,8 +305,10 @@ describe('server chat API', () => { await report('game_end', 60, 4); await report('game_start', 80); expect((await (await get()).json()).entries).toEqual([]); + expect(read).toHaveBeenCalledTimes(1); await report('heartbeat', 90, 2); await log(['60:30 say;abc;0;Player;old match', '61:40 say;abc;0;Player;new match'], 100); + advanceTime(); const body = await (await get()).json(); expect(body.entries.map((entry: ChatMessageEntry) => entry.message.message)).toEqual([ 'new match' @@ -311,8 +331,10 @@ describe('server chat API', () => { await report('game_start', 0); await report('heartbeat', 20, 4); await log(['60:30 say;abc;0;Player;old match'], 30); + advanceTime(); expect((await (await get()).json()).entries).toHaveLength(1); await log(['60:30 say;abc;0;Player;old match', boundary], 60); + advanceTime(); expect((await (await get()).json()).entries, boundary).toEqual([]); } }); @@ -333,6 +355,7 @@ describe('server chat API', () => { .map((entry: ChatMessageEntry) => entry.message.message) ).toEqual(['new match']); await log(['80:00 say;abc;0;Player;old match', '0:00 InitGame:'], 60); + advanceTime(); expect( (await (await get()).json()).entries.some( (entry: ContextualChatEntry) => entry.type === 'message'