Restrict plugin lookups and reuse cached chat reads (Refs #98)

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
 

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.

src/lib/server/fleet/chat.ts

@@ -209,12 +209,38 @@ interface ChatReadState {
 }
 const readers = new Map<string, ChatReadState>();
 const failures = new Map<string, number>();
+const CHAT_READ_MIN_AGE_MS = 15_000;
+type LastChatRead = { completedAt: number } & ({ missing: boolean } | { error: unknown });
+const lastReads = new Map<string, LastChatRead>();
+
+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<void> {
     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))

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 });

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);
     });

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';

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<T>() {
+    let resolve!: (value: T) => void;
+    const promise = new Promise<T>((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<TailResult>();
+        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<TailResult>();
+        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);
+    });
+});

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'