diff --git a/docs/CONFIG.md b/docs/CONFIG.md index d0331a4..121616a 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -100,6 +100,8 @@ Profiles **do not embed commands**. They contain three possible keys: Example: if global_env sets `NODE_ENV=production`, profile env sets nothing, and command env_overrides sets `NODE_ENV=development`, the final value is `development`. +Below all of these, Conductor sets `PYTHONUNBUFFERED=1`. Python block-buffers stdout when it is a pipe, so without it a Python service's output reaches the log viewer in ~8KB lumps instead of line by line. Any layer above can override it; set it to an empty string to turn it off. Other runtimes that buffer piped output need their own setting. + ### Secret Masking Any environment variable whose **name** appears in the top-level `env_secrets` array will have its value replaced with `[FILTERED]` everywhere — logs, UI display, and API responses. Both exact match and substring match against the var name are supported (e.g., `API_TOKEN`, `DB_PASSWORD`). diff --git a/packages/core/src/config/env-resolution.ts b/packages/core/src/config/env-resolution.ts index 829526f..4932b11 100644 --- a/packages/core/src/config/env-resolution.ts +++ b/packages/core/src/config/env-resolution.ts @@ -29,6 +29,9 @@ export interface BuildCommandEnvParams extends BuildEnvParams { function baseLayers(params: BuildEnvParams): Array | undefined> { const { configFilePath, config, profile, dbGlobalEnv, dbProfileEnv } = params; return [ + // Python block-buffers stdout into a pipe, so its output would reach + // the log viewer in ~8KB lumps. Lowest layer: anything can override it. + { PYTHONUNBUFFERED: "1" }, process.env as Record, { BASE_PATH: resolveBasePath(configFilePath, config.base_path) }, config.default_shell ? { CONDUCTOR_SHELL: config.default_shell } : undefined, diff --git a/packages/core/src/db/queries.ts b/packages/core/src/db/queries.ts index a96500d..c129d57 100644 --- a/packages/core/src/db/queries.ts +++ b/packages/core/src/db/queries.ts @@ -74,23 +74,37 @@ export class ConductorQueries { } insertLog(row: Omit): LogRow { - const sessionId = this.currentSessionByProfile.get(row.profile) ?? null; - return this.db - .prepare( - `INSERT INTO logs (process_id, command_id, profile, timestamp, level, stream, message, session_id) - VALUES ($process_id, $command_id, $profile, $timestamp, $level, $stream, $message, $session_id) - RETURNING *`, - ) - .get({ - $process_id: row.process_id, - $command_id: row.command_id, - $profile: row.profile, - $timestamp: row.timestamp, - $level: row.level, - $stream: row.stream, - $message: row.message, - $session_id: sessionId, - }) as LogRow; + return this.insertLogs([row])[0] as LogRow; + } + + /** + * Inserts a batch in one transaction, returning the stored rows (ids + * included) in order. A burst of output written one autocommit per line + * kept the event loop busy for seconds. + */ + insertLogs(rows: Array>): LogRow[] { + if (rows.length === 0) return []; + // db.query caches the compiled statement across calls. + const insert = this.db.query( + `INSERT INTO logs (process_id, command_id, profile, timestamp, level, stream, message, session_id) + VALUES ($process_id, $command_id, $profile, $timestamp, $level, $stream, $message, $session_id) + RETURNING *`, + ); + return this.db.transaction(() => + rows.map( + (row) => + insert.get({ + $process_id: row.process_id, + $command_id: row.command_id, + $profile: row.profile, + $timestamp: row.timestamp, + $level: row.level, + $stream: row.stream, + $message: row.message, + $session_id: this.currentSessionByProfile.get(row.profile) ?? null, + }) as LogRow, + ), + )(); } queryLogs(filters: { @@ -135,8 +149,10 @@ export class ConductorQueries { const where = clauses.length > 0 ? `WHERE ${clauses.join(" AND ")}` : ""; const limit = filters.limit ?? 200; + // By id, not timestamp: ids follow insertion order, and the order matches + // idx_logs_process / idx_logs_command so no sort over a whole run is needed. return this.db - .prepare(`SELECT * FROM logs ${where} ORDER BY timestamp DESC, id DESC LIMIT $limit`) + .prepare(`SELECT * FROM logs ${where} ORDER BY id DESC LIMIT $limit`) .all({ ...params, $limit: limit }) as LogRow[]; } diff --git a/packages/core/src/db/schema.sql b/packages/core/src/db/schema.sql index ce81cb8..dd35827 100644 --- a/packages/core/src/db/schema.sql +++ b/packages/core/src/db/schema.sql @@ -26,7 +26,11 @@ CREATE TABLE IF NOT EXISTS logs ( session_id INTEGER ); +-- Both end in the rowid (id), so "one process/command, newest first" reads +-- just its own rows; without idx_logs_process the viewer's history query +-- walked idx_logs_timestamp across every process. CREATE INDEX IF NOT EXISTS idx_logs_command ON logs(command_id); +CREATE INDEX IF NOT EXISTS idx_logs_process ON logs(process_id, id); CREATE INDEX IF NOT EXISTS idx_logs_timestamp ON logs(timestamp); CREATE INDEX IF NOT EXISTS idx_logs_session ON logs(session_id); diff --git a/packages/core/src/executor/wrapper.ts b/packages/core/src/executor/wrapper.ts index 876d771..da7f4b2 100644 --- a/packages/core/src/executor/wrapper.ts +++ b/packages/core/src/executor/wrapper.ts @@ -496,6 +496,9 @@ export class ProcessWrapper { for (const line of lines) { this.emitLogFor(owner, line, kind); } + // Chunks already sitting in the pipe resolve as microtasks, so a burst + // would otherwise run start to finish without letting HTTP in. + await new Promise((resolve) => setImmediate(resolve)); } if (buffer.length > 0) { diff --git a/packages/core/src/workspace/session.ts b/packages/core/src/workspace/session.ts index e77f89f..adc924b 100644 --- a/packages/core/src/workspace/session.ts +++ b/packages/core/src/workspace/session.ts @@ -6,7 +6,7 @@ import type { ConductorConfig } from "../config/schema"; import { ConfigStore, dbEnvLookup } from "../config/store"; import { saveConfig } from "../config/writer"; import { DEFAULT_DB_PATH, openDatabase } from "../db/init"; -import { ConductorQueries } from "../db/queries"; +import { ConductorQueries, type LogRow } from "../db/queries"; import type { LogEntry, LogHandler } from "../executor/wrapper"; import { createLogger, type ConductorLogger } from "../logger/pino"; import type { LogBroadcaster } from "../logs/broadcaster"; @@ -118,14 +118,28 @@ export function openSession(dir: string, opts: SessionOptions): WorkspaceSession // Every log line from any managed process is persisted and broadcast // so both the CLI (via `conductor logs`) and the UI's live SSE stream - // can see it, regardless of who started the process. + // can see it, regardless of who started the process. Lines are buffered + // and written once per event-loop turn, in one transaction: the stored + // rows (with their ids) are what gets broadcast. + let pendingLogs: Array> = []; + const flushLogs = () => { + const batch = pendingLogs; + pendingLogs = []; + if (batch.length === 0) return; + try { + for (const row of queries.insertLogs(batch)) broadcaster.publish(row); + } catch (err) { + logger.error({ err, lines: batch.length }, "Failed to store log lines"); + } + }; // Set once the DB is closed. A process's last lines can still be in its pipe // after stop() returns (waiting for them stalled every close on Windows), - // and they must not reach insertLog on a closed DB. + // and they must not reach insertLogs on a closed DB. let dbClosed = false; const onLog: LogHandler = (entry: LogEntry) => { if (dbClosed) return; - const row = queries.insertLog({ + if (pendingLogs.length === 0) setImmediate(flushLogs); + pendingLogs.push({ process_id: String(entry.pid), command_id: entry.commandId, profile: entry.profile, @@ -134,7 +148,6 @@ export function openSession(dir: string, opts: SessionOptions): WorkspaceSession stream: entry.stream, message: entry.message, }); - broadcaster.publish(row); }; // Time-window log retention sweep. Session-scoped retention doesn't need @@ -167,6 +180,8 @@ export function openSession(dir: string, opts: SessionOptions): WorkspaceSession } finally { collector.stop(); clearInterval(logRetentionInterval); + // Lines already received are written; any later ones are dropped. + flushLogs(); dbClosed = true; db.close(); } diff --git a/packages/core/test/env-resolution.test.ts b/packages/core/test/env-resolution.test.ts index 9b19ea9..8e9b70d 100644 --- a/packages/core/test/env-resolution.test.ts +++ b/packages/core/test/env-resolution.test.ts @@ -66,4 +66,24 @@ describe("buildCommandEnv", () => { expect(pathKey).toBeDefined(); expect(env[pathKey!]).toBe(process.env[pathKey!]); }); + + // Python block-buffers stdout into a pipe, so a service's output showed + // up in the viewer ~8KB at a time instead of line by line. + test("defaults PYTHONUNBUFFERED=1 below every other layer", () => { + const params = { configFilePath, config, profile: config.profiles.dev }; + const saved = process.env.PYTHONUNBUFFERED; + delete process.env.PYTHONUNBUFFERED; + try { + expect(buildCommandEnv({ ...params, cmd: config.commands[0]! }).PYTHONUNBUFFERED).toBe("1"); + + const overridden = { ...config.commands[0]!, env_overrides: { PYTHONUNBUFFERED: "" } }; + expect(buildCommandEnv({ ...params, cmd: overridden }).PYTHONUNBUFFERED).toBe(""); + + process.env.PYTHONUNBUFFERED = "0"; + expect(buildCommandEnv({ ...params, cmd: config.commands[0]! }).PYTHONUNBUFFERED).toBe("0"); + } finally { + if (saved === undefined) delete process.env.PYTHONUNBUFFERED; + else process.env.PYTHONUNBUFFERED = saved; + } + }); }); diff --git a/packages/core/test/log-writes.test.ts b/packages/core/test/log-writes.test.ts new file mode 100644 index 0000000..6544f92 --- /dev/null +++ b/packages/core/test/log-writes.test.ts @@ -0,0 +1,52 @@ +import { test, expect } from "bun:test"; +import { openDatabase } from "../src/db/init"; +import { ConductorQueries } from "../src/db/queries"; + +const line = (message: string) => ({ + process_id: "10", + command_id: "api", + profile: "dev", + timestamp: new Date().toISOString(), + level: "info", + stream: "stdout" as const, + message, +}); + +test("insertLogs writes a batch with increasing ids, session stamps and FTS entries", () => { + const queries = new ConductorQueries(openDatabase(":memory:")); + const sessionId = queries.insertSession("dev"); + + const rows = queries.insertLogs([line("alpha one"), line("beta two"), line("gamma three")]); + + expect(rows.map((r) => r.message)).toEqual(["alpha one", "beta two", "gamma three"]); + expect(rows[1]!.id).toBe(rows[0]!.id + 1); + expect(rows[2]!.id).toBe(rows[1]!.id + 1); + expect(rows.every((r) => r.session_id === sessionId)).toBe(true); + expect(queries.queryLogs({ grep: "beta" }).map((r) => r.id)).toEqual([rows[1]!.id]); +}); + +test("insertLogs with an empty batch is a no-op", () => { + const queries = new ConductorQueries(openDatabase(":memory:")); + expect(queries.insertLogs([])).toEqual([]); + expect(queries.queryLogs({})).toEqual([]); +}); + +test("history for one process comes from its index, newest id first", () => { + const db = openDatabase(":memory:"); + const queries = new ConductorQueries(db); + // Timestamps out of id order: the viewer's order is insertion order. + const [first, second] = queries.insertLogs([ + { ...line("first"), timestamp: "2026-01-02T00:00:00.000Z" }, + { ...line("second"), timestamp: "2026-01-01T00:00:00.000Z" }, + ]); + expect(queries.queryLogs({ processId: "10" }).map((r) => r.id)).toEqual([second!.id, first!.id]); + + const plan = db + .query( + "EXPLAIN QUERY PLAN SELECT * FROM logs WHERE process_id = $pid ORDER BY id DESC LIMIT 500", + ) + .all({ $pid: "10" }) as Array<{ detail: string }>; + expect(plan.map((p) => p.detail)).toEqual([ + "SEARCH logs USING INDEX idx_logs_process (process_id=?)", + ]); +}); diff --git a/packages/core/test/workspace-manager.test.ts b/packages/core/test/workspace-manager.test.ts index 571dd8d..64a9350 100644 --- a/packages/core/test/workspace-manager.test.ts +++ b/packages/core/test/workspace-manager.test.ts @@ -286,6 +286,8 @@ describe("WorkspaceManager + API", () => { message: "hello-from-b", timestamp: new Date().toISOString(), }); + // Log lines are written in a batch on the next event-loop turn. + await new Promise((resolve) => setImmediate(resolve)); expect(manager.current!.queries.queryLogs({ grep: "hello-from-b" })).toHaveLength(1); await manager.close(); diff --git a/packages/core/test/workspace-session.test.ts b/packages/core/test/workspace-session.test.ts index 179c9d6..d9a0324 100644 --- a/packages/core/test/workspace-session.test.ts +++ b/packages/core/test/workspace-session.test.ts @@ -6,6 +6,7 @@ import { ConfigError, DEFAULT_DB_PATH, LogBroadcaster, + type LogRow, openSession, prepareSession, resolveWorkspaceDir, @@ -110,9 +111,45 @@ describe("openSession", () => { }, ); + test("onLog writes a burst in one batch and publishes the stored rows", async () => { + const broadcaster = new LogBroadcaster(); + const session = openSession(dir, { broadcaster, dbPath: ":memory:" }); + const published: LogRow[] = []; + broadcaster.subscribe((row) => published.push(row)); + const entry = (message: string) => ({ + commandId: "api", + commandName: "API", + profile: "dev", + pid: 42, + stream: "stdout" as const, + message, + timestamp: new Date().toISOString(), + }); + + for (const message of ["one", "two", "three"]) session.onLog(entry(message)); + expect(published).toEqual([]); + + await new Promise((resolve) => setImmediate(resolve)); + expect(published.map((r) => r.message)).toEqual(["one", "two", "three"]); + expect(published.map((r) => r.id)).toEqual( + session.queries + .queryLogs({ processId: "42" }) + .map((r) => r.id) + .reverse(), + ); + + // A line still waiting for its flush is written before close() shuts the DB. + session.onLog(entry("last words")); + await session.close(); + expect(published.at(-1)?.message).toBe("last words"); + }); + test("a log line arriving after close() is dropped, not written to the closed DB", async () => { // A process's last lines can still be in its pipe after stop() returns. - const session = openSession(dir, { broadcaster: new LogBroadcaster(), dbPath: ":memory:" }); + const broadcaster = new LogBroadcaster(); + const published: LogRow[] = []; + broadcaster.subscribe((row) => published.push(row)); + const session = openSession(dir, { broadcaster, dbPath: ":memory:" }); await session.close(); expect(() => session.onLog({ @@ -125,6 +162,8 @@ describe("openSession", () => { timestamp: new Date().toISOString(), }), ).not.toThrow(); + await new Promise((resolve) => setImmediate(resolve)); + expect(published).toEqual([]); }); test("close() still releases the DB when stopping processes fails", async () => { diff --git a/packages/core/test/wrapper.test.ts b/packages/core/test/wrapper.test.ts index 78db4c3..3000fc4 100644 --- a/packages/core/test/wrapper.test.ts +++ b/packages/core/test/wrapper.test.ts @@ -119,6 +119,37 @@ describe("ProcessWrapper.start", () => { expect(lines).toEqual(["a", "b", "c"]); }); + test("yields to the event loop between chunks of an output burst", async () => { + // A slow handler (like the per-line SQLite write) lets the pipe fill up; + // already-buffered chunks then resolve as microtasks, so without a yield + // the whole burst runs in one event-loop turn and HTTP handling starves. + const cmd = makeCommand({ + id: "burst", + name: "Burst", + run: `bun -e "for (let i = 0; i < 50000; i++) console.log('line ' + i + ' of a burst')"`, + }); + const wrapper = new ProcessWrapper(cmd, "test", testEnv()); + let turns = 0; + let running = true; + const tick = () => { + turns++; + if (running) setImmediate(tick); + }; + setImmediate(tick); + const turnsSeen = new Set(); + wrapper.onLog(() => { + turnsSeen.add(turns); + const start = performance.now(); + while (performance.now() - start < 0.03) { + // busy-wait ~30µs, about what a per-line SQLite write cost + } + }); + await runToExit(wrapper); + await wrapper.drained; + running = false; + expect(turnsSeen.size).toBeGreaterThan(5); + }, 20_000); + test("flags hasMatchedLogLine once a log_line healthcheck pattern appears in output", async () => { const cmd = makeCommand({ id: "log-line-match", diff --git a/packages/ui/src/components/LogLines.tsx b/packages/ui/src/components/LogLines.tsx index 2adad57..f98111d 100644 --- a/packages/ui/src/components/LogLines.tsx +++ b/packages/ui/src/components/LogLines.tsx @@ -1,3 +1,4 @@ +import { memo } from "react"; import type { LogRow } from "../lib/api"; import { renderAnsiLine } from "../lib/ansi"; @@ -6,26 +7,32 @@ export function LogLines({ logs }: { logs: LogRow[] }) { return (
{logs.map((log) => ( -
- - {new Date(log.timestamp).toLocaleTimeString()} - {" "} - - {renderAnsiLine(log.message, String(log.id))} -
+ ))}
); } + +// Memoized so a live tail only renders (and ANSI-parses) the lines it adds. +const LogLine = memo(function LogLine({ log }: { log: LogRow }) { + return ( +
+ + {new Date(log.timestamp).toLocaleTimeString()} + {" "} + + {renderAnsiLine(log.message, String(log.id))} +
+ ); +}); diff --git a/packages/ui/src/components/LogViewer.tsx b/packages/ui/src/components/LogViewer.tsx index b804016..ec4a4d3 100644 --- a/packages/ui/src/components/LogViewer.tsx +++ b/packages/ui/src/components/LogViewer.tsx @@ -25,7 +25,7 @@ import { fetchLogs, streamLogs, type LogRow, type ProcessInfo } from "../lib/api import { useUiStore } from "../store/ui"; import { useStopProcess, useRestartCommand } from "../hooks/useProcessActions"; import { LogLines } from "./LogLines"; -import { mergeLogs } from "../lib/mergeLogs"; +import { MAX_LOG_ROWS, mergeLogs } from "../lib/mergeLogs"; import { STATUS_COLOR } from "../lib/statusColor"; type StreamFilter = "all" | "stdout" | "stderr"; @@ -69,15 +69,28 @@ export function LogViewer({ process }: { process: ProcessInfo }) { // Live tail: SSE replays recent history too, overlapping the fetch above // (kept for an instant first paint) in either order, so both merge by id. + // Entries are applied once per frame: a burst sends thousands a second. + let queued: LogRow[] = []; + let frame = 0; const unsubscribe = streamLogs(pid, (entry) => { - setLogState((prev) => ({ - pid, - rows: mergeLogs(prev.pid === pid ? prev.rows : [], [entry]), - })); + queued.push(entry); + // rAF doesn't run in a hidden window; only the newest lines survive the cap anyway. + if (queued.length >= 2 * MAX_LOG_ROWS) queued = queued.slice(-MAX_LOG_ROWS); + if (frame) return; + frame = requestAnimationFrame(() => { + frame = 0; + const batch = queued; + queued = []; + setLogState((prev) => ({ + pid, + rows: mergeLogs(prev.pid === pid ? prev.rows : [], batch), + })); + }); }); return () => { cancelled = true; + cancelAnimationFrame(frame); unsubscribe(); }; }, [process.pid]); @@ -91,11 +104,12 @@ export function LogViewer({ process }: { process: ProcessInfo }) { }); }, [logs, search, streamFilter]); + // On the array, not its length: once the cap is reached new lines no longer change it. useEffect(() => { if (autoScroll) { viewportRef.current?.scrollTo({ top: viewportRef.current.scrollHeight }); } - }, [filteredLogs.length, autoScroll]); + }, [filteredLogs, autoScroll]); // Auto-scroll pauses if the user scrolls up to read history, and // resumes once they scroll back to the bottom. @@ -203,7 +217,14 @@ export function LogViewer({ process }: { process: ProcessInfo }) { {logs.length === 0 ? "No log output yet." : "No lines match the current filter."} ) : ( - + <> + {logs.length >= MAX_LOG_ROWS && ( + + Showing the last {MAX_LOG_ROWS} lines. Older output is in History. + + )} + + )} diff --git a/packages/ui/src/lib/mergeLogs.ts b/packages/ui/src/lib/mergeLogs.ts index 5802725..64b6ba9 100644 --- a/packages/ui/src/lib/mergeLogs.ts +++ b/packages/ui/src/lib/mergeLogs.ts @@ -1,14 +1,35 @@ import type { LogRow } from "./api"; +/** The live viewer keeps only the newest lines; older output stays in History. */ +export const MAX_LOG_ROWS = 5000; + /** - * Adds `incoming` to `rows`, skipping ids already present, ordered by id. - * LogViewer fills from two overlapping sources - the /api/logs history fetch - * and the SSE stream, which replays the same recent history before tailing - - * and either can land first, so both go through here. + * Adds `incoming` to `rows`, skipping ids already present, ordered by id, and + * keeps the newest `max`. LogViewer fills from two overlapping sources - the + * /api/logs history fetch and the SSE stream, which replays the same recent + * history before tailing - and either can land first, so both go through here. */ -export function mergeLogs(rows: LogRow[], incoming: LogRow[]): LogRow[] { - const seen = new Set(rows.map((r) => r.id)); - const fresh = incoming.filter((r) => !seen.has(r.id) && seen.add(r.id)); - if (fresh.length === 0) return rows; - return [...rows, ...fresh].sort((a, b) => a.id - b.id); +export function mergeLogs(rows: LogRow[], incoming: LogRow[], max = MAX_LOG_ROWS): LogRow[] { + if (incoming.length === 0) return rows; + let merged: LogRow[]; + if (isAppend(rows, incoming)) { + // The live tail: ids only grow, so no dedup or sort needed. + merged = rows.concat(incoming); + } else { + const seen = new Set(rows.map((r) => r.id)); + const fresh = incoming.filter((r) => !seen.has(r.id) && seen.add(r.id)); + if (fresh.length === 0) return rows; + merged = [...rows, ...fresh].sort((a, b) => a.id - b.id); + } + return merged.length > max ? merged.slice(-max) : merged; +} + +/** Whether `incoming` is strictly increasing and starts after the last of `rows`. */ +function isAppend(rows: LogRow[], incoming: LogRow[]): boolean { + let last = rows.at(-1)?.id ?? -Infinity; + for (const row of incoming) { + if (row.id <= last) return false; + last = row.id; + } + return true; } diff --git a/packages/ui/test/mergeLogs.test.ts b/packages/ui/test/mergeLogs.test.ts index b6d26b0..c989ac1 100644 --- a/packages/ui/test/mergeLogs.test.ts +++ b/packages/ui/test/mergeLogs.test.ts @@ -17,7 +17,21 @@ test("history fetch landing after SSE lines merges in id order", () => { expect(ids(mergeLogs(rows, [1, 2, 3].map(row)))).toEqual([1, 2, 3, 4]); }); +test("a batch overlapping what is held is deduplicated", () => { + const rows = [1, 2, 3].map(row); + expect(ids(mergeLogs(rows, [2, 3, 4, 5].map(row)))).toEqual([1, 2, 3, 4, 5]); +}); + test("returns the same array when nothing is new", () => { const rows = [row(1)]; expect(mergeLogs(rows, [row(1)])).toBe(rows); + expect(mergeLogs(rows, [])).toBe(rows); +}); + +test("keeps only the newest rows past the cap", () => { + let rows: LogRow[] = []; + for (let id = 1; id <= 10; id += 2) rows = mergeLogs(rows, [row(id), row(id + 1)], 4); + expect(ids(rows)).toEqual([7, 8, 9, 10]); + // A late history fetch can't push older lines back in. + expect(ids(mergeLogs(rows, [1, 2, 3].map(row), 4))).toEqual([7, 8, 9, 10]); });