diff --git a/apps/server/src/internal/events.ts b/apps/server/src/internal/events.ts index 9fde9228b7..394794c83e 100644 --- a/apps/server/src/internal/events.ts +++ b/apps/server/src/internal/events.ts @@ -37,6 +37,7 @@ import { isActivePruneTriggerThreadEventType, maybePruneActiveThreadEventHistory, } from "../services/system/event-pruning.js"; +import { runEventLoopWorkSync } from "../services/system/event-loop-work.js"; import { queueChildThreadTurnNotificationBestEffort } from "../services/threads/child-thread-notifications.js"; import { isParentNotifiableChildThread } from "../services/threads/thread-parent.js"; import { runQueuedMessageAutoSendForThread } from "../services/threads/queued-messages.js"; @@ -194,9 +195,22 @@ interface QueuedMessageAutoSendFollowUp { threadId: string; } +/** + * Opportunistic event-history prune, deferred off the ingest response path. + * Pruning is output-preserving (see the timeline cache notes), so running it + * after the response — instead of inside the hottest handler — only changes + * when the redundant rows disappear, never what any read returns. + */ +interface ActiveThreadEventPruneFollowUp { + kind: "active-thread-event-prune"; + latestPrunableSequence: number; + threadId: string; +} + type EventEffectFollowUp = | ParentTurnNotificationFollowUp - | QueuedMessageAutoSendFollowUp; + | QueuedMessageAutoSendFollowUp + | ActiveThreadEventPruneFollowUp; function isRootTurnStartedEvent( event: Extract, @@ -494,6 +508,16 @@ async function executeEventFollowUpBestEffort( threadId: followUp.threadId, }); return; + case "active-thread-event-prune": + // Synchronous better-sqlite3 work: the wrapper lets the stall + // monitor name the prune instead of blaming whatever ran next. + runEventLoopWorkSync(`event-prune ${followUp.threadId}`, () => + maybePruneActiveThreadEventHistory(deps, { + latestPrunableSequence: followUp.latestPrunableSequence, + threadId: followUp.threadId, + }), + ); + return; } } catch (error) { if (isCommandTimeoutError(error)) { @@ -1023,7 +1047,7 @@ export function registerInternalEventRoutes(app: Hono, deps: AppDeps): void { events: postableEvents, insertedEventIndexes: appendResult.insertedInputIndexes, })) { - maybePruneActiveThreadEventHistory(deps, candidate); + followUps.push({ kind: "active-thread-event-prune", ...candidate }); } deferEventFollowUpBatch(deps, followUps); diff --git a/apps/server/src/routes/threads/base.ts b/apps/server/src/routes/threads/base.ts index d99182e871..4e73cb6127 100644 --- a/apps/server/src/routes/threads/base.ts +++ b/apps/server/src/routes/threads/base.ts @@ -1,4 +1,5 @@ import { + DEFAULT_THREAD_LIST_LIMIT, THREAD_SEARCH_LIMIT_PER_GROUP_DEFAULT, THREAD_SEARCH_LIMIT_PER_GROUP_MAX, countNonDeletedAssignedChildThreads, @@ -233,7 +234,13 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void { if (query.sectionId) { requireThreadSection(deps, query.sectionId); } - const threads = listThreadsWithPendingInteractionState(deps.db, { + // The list is capped (default DEFAULT_THREAD_LIST_LIMIT) so one + // long-lived install cannot make every list read scale with its full + // history. The response stays a bare array for compatibility, so the + // truncation signal rides a header: fetch one probe row past the cap and + // report whether it existed. Callers page with limit/offset. + const effectiveLimit = limit ?? DEFAULT_THREAD_LIST_LIMIT; + const threadRows = listThreadsWithPendingInteractionState(deps.db, { ...(query.projectId ? { projectId: query.projectId } : {}), ...(query.parentThreadId ? { parentThreadId: query.parentThreadId } : {}), ...(query.sourceThreadId ? { sourceThreadId: query.sourceThreadId } : {}), @@ -246,9 +253,12 @@ export function registerThreadBaseRoutes(app: Hono, deps: AppDeps): void { query.archived === undefined ? undefined : query.archived === "true", hasParent: query.hasParent === undefined ? undefined : query.hasParent === "true", - ...(limit !== undefined ? { limit } : {}), + limit: effectiveLimit + 1, ...(offset !== undefined ? { offset } : {}), }); + const hasMore = threadRows.length > effectiveLimit; + const threads = hasMore ? threadRows.slice(0, effectiveLimit) : threadRows; + context.header("x-bb-thread-list-has-more", hasMore ? "true" : "false"); return context.json( toThreadListEntryResponses(deps, { threads }) satisfies ThreadListEntry[], ); diff --git a/apps/server/src/routes/threads/data.ts b/apps/server/src/routes/threads/data.ts index 6aa5e42cce..7004a3b577 100644 --- a/apps/server/src/routes/threads/data.ts +++ b/apps/server/src/routes/threads/data.ts @@ -348,8 +348,9 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void { summaryOnly, includeProviderUnhandledOperations, }; + const paramsKey = buildThreadTimelineParamsKey(keyArgs); const full = timelineCache.getOrBuild( - buildThreadTimelineCacheKey({ ...keyArgs, maxSeq }), + { key: buildThreadTimelineCacheKey({ ...keyArgs, maxSeq }), paramsKey }, () => { const { profile, response } = buildThreadTimelineWithProfile( deps.db, @@ -392,7 +393,6 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void { query.afterSequence, "afterSequence", ); - const paramsKey = buildThreadTimelineParamsKey(keyArgs); const previous = afterSequence === undefined ? undefined @@ -401,7 +401,13 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void { previous === undefined ? undefined : computeTimelineRowDelta(previous.rows, full.rows); - timelineLatestRowsCache.set(paramsKey, { maxSeq, rows: full.rows }); + // Keyed by the response's own revision, not the route-computed `maxSeq`: + // the cache's rebuild floor may serve a window built just before the + // newest events, and the snapshot must name the revision the rows reflect. + timelineLatestRowsCache.set(paramsKey, { + maxSeq: full.maxSeq, + rows: full.rows, + }); return context.json( delta === undefined ? full : { ...full, rows: [], delta }, diff --git a/apps/server/src/services/plugins/plugin-api.ts b/apps/server/src/services/plugins/plugin-api.ts index 68d2c57dbc..927006230a 100644 --- a/apps/server/src/services/plugins/plugin-api.ts +++ b/apps/server/src/services/plugins/plugin-api.ts @@ -92,7 +92,7 @@ import type { import type { BbSdk, ThreadForkArgs, ThreadSpawnArgs } from "@bb/sdk"; import type { ServerLogger } from "../../types.js"; import type { PluginInteractionResult } from "../interactions/pending-interactions.js"; -import { appendPluginLogLine } from "./plugin-log.js"; +import { appendPluginLogLine, disposePluginLogWriter } from "./plugin-log.js"; import type { PluginHostArtifactSnapshot } from "./plugin-service-internal.js"; import { readPluginSettingsValues } from "./plugin-settings.js"; @@ -430,6 +430,19 @@ function createStagedRegistrations< }; } +/** + * `bb.realtime.publish` broadcasts to every connected client with no + * per-channel subscription, so a runaway publisher multiplies across the + * whole socket fan-out. Interactive signals are a few per second; anything + * sustained above that is treated as a loop and dropped (signals are + * ephemeral by contract), warning once per window so the log names the + * plugin without flooding. + */ +const REALTIME_PUBLISH_MAX_PAYLOAD_BYTES = 64 * 1024; +const REALTIME_PUBLISH_TOKENS_PER_SECOND = 10; +const REALTIME_PUBLISH_BURST_TOKENS = 20; +const REALTIME_PUBLISH_DROP_WARN_INTERVAL_MS = 10_000; + export function createPluginApi(options: { pluginId: string; logger: ServerLogger; @@ -439,8 +452,10 @@ export function createPluginApi(options: { getSdk: () => BbSdk | undefined; /** Undefined until the server is listening (bb.server is bind-gated too). */ getLoopbackBaseUrl: () => string | undefined; - /** Broadcasts a plugin-signal WS message (hub.notifyPluginSignal). */ - publishSignal: (channel: string, payload: unknown) => void; + /** Broadcasts a plugin-signal WS message (hub.notifyPluginSignal). The + * payload arrives pre-serialized — publish's one JSON.stringify — and is + * embedded verbatim in the wire frame. */ + publishSignal: (channel: string, serializedPayload: string) => void; /** Marks the plugin needs-configuration in the loader's status table. */ reportNeedsConfiguration: (message: string) => void; /** Returns the owning plugin id when another plugin already registered @@ -554,6 +569,10 @@ export function createPluginApi(options: { const pendingAgentToolProblems: string[] = []; const pendingSharedPorts = new Map(); const disposeHooks: Array<() => void | Promise> = []; + // Flush buffered bb.log lines before this generation's resources go away: + // a pending flush timer must not outlive (and race) a plugin data-dir + // removal. + disposeHooks.push(() => disposePluginLogWriter(dataDir, pluginId)); const settingsRecord: PluginApiHandle["settings"] = { descriptors: {}, listeners: [], @@ -866,16 +885,45 @@ export function createPluginApi(options: { }, }; + // Token bucket for realtime publishes; state lives with the handle, so a + // reload starts the plugin with a full burst again. + let realtimePublishTokens = REALTIME_PUBLISH_BURST_TOKENS; + let realtimePublishRefilledAt = Date.now(); + let realtimePublishWarnedAt = 0; + function takeRealtimePublishToken(channel: string): boolean { + const now = Date.now(); + realtimePublishTokens = Math.min( + REALTIME_PUBLISH_BURST_TOKENS, + realtimePublishTokens + + ((now - realtimePublishRefilledAt) / 1000) * + REALTIME_PUBLISH_TOKENS_PER_SECOND, + ); + realtimePublishRefilledAt = now; + if (realtimePublishTokens >= 1) { + realtimePublishTokens -= 1; + return true; + } + if (now - realtimePublishWarnedAt >= REALTIME_PUBLISH_DROP_WARN_INTERVAL_MS) { + realtimePublishWarnedAt = now; + logger.warn( + { pluginId, channel }, + "Dropping bb.realtime.publish signals over the per-plugin rate limit", + ); + } + return false; + } + const realtime: PluginRealtime = { publish(channel, payload) { assertLive(); if (typeof channel !== "string" || channel.length === 0) { throw new Error("realtime channel must be a non-empty string"); } - // JSON round-trip up front: enforces serializability with a clear - // error at the publish site and strips prototypes/getters before the - // payload crosses the WS boundary. - let normalized: unknown = null; + // One serialization up front: it enforces serializability with a clear + // error at the publish site AND produces the exact bytes that cross + // the WS boundary — the hub embeds this string verbatim instead of + // re-parsing and re-stringifying the payload per publish. + let payloadJson = "null"; if (payload !== undefined) { let json: string | undefined; try { @@ -888,9 +936,19 @@ export function createPluginApi(options: { `realtime payload for channel "${channel}" is not JSON-serializable`, ); } - normalized = JSON.parse(json); + payloadJson = json; + } + const payloadBytes = Buffer.byteLength(payloadJson, "utf8"); + if (payloadBytes > REALTIME_PUBLISH_MAX_PAYLOAD_BYTES) { + throw new Error( + `realtime payload for channel "${channel}" is ${payloadBytes} bytes; ` + + `the limit is ${REALTIME_PUBLISH_MAX_PAYLOAD_BYTES} (64KB)`, + ); + } + if (!takeRealtimePublishToken(channel)) { + return; } - publishSignal(channel, normalized); + publishSignal(channel, payloadJson); }, }; diff --git a/apps/server/src/services/plugins/plugin-log.ts b/apps/server/src/services/plugins/plugin-log.ts index 06afbf4a51..4ea6f0b3b0 100644 --- a/apps/server/src/services/plugins/plugin-log.ts +++ b/apps/server/src/services/plugins/plugin-log.ts @@ -1,5 +1,5 @@ -import { appendFileSync, mkdirSync, renameSync, statSync } from "node:fs"; -import { readFile } from "node:fs/promises"; +import { mkdirSync, renameSync, statSync } from "node:fs"; +import { appendFile, readFile } from "node:fs/promises"; import { join } from "node:path"; /** @@ -7,21 +7,134 @@ import { join } from "node:path"; * appended as JSONL to /plugins//logs/plugin.log in addition to * the prefixed server log. Simple size rotation: past 5MB the file is renamed * to plugin.log.1 (one rotated file kept, replacing the previous one). + * + * `bb.log` runs on the server event loop, so lines are buffered and flushed + * as one async append per window (or once the buffer crosses a byte + * threshold) instead of a mkdir + stat + sync append per line. The directory + * is created once per writer lifetime and the rotation size is tracked from + * a cached counter seeded by one stat. Delivery stays best effort — it + * already was, and the prefixed server log still carries every message. */ const PLUGIN_LOG_MAX_BYTES = 5 * 1024 * 1024; const PLUGIN_LOG_FILE = "plugin.log"; const PLUGIN_LOG_ROTATED_FILE = "plugin.log.1"; +const PLUGIN_LOG_FLUSH_INTERVAL_MS = 200; +const PLUGIN_LOG_FLUSH_THRESHOLD_BYTES = 8 * 1024; type PluginLogLevel = "debug" | "info" | "warn" | "error"; +interface PluginLogWriter { + bufferedBytes: number; + bufferedLines: string[]; + dirEnsured: boolean; + /** Serializes flushes so appended lines keep their order. */ + flushChain: Promise; + /** Cached plugin.log size; null until seeded by the writer's one stat. */ + logFileBytes: number | null; + timer: ReturnType | null; +} + +const pluginLogWriters = new Map(); + function pluginLogsDir(dataDir: string, pluginId: string): string { return join(dataDir, "plugins", pluginId, "logs"); } +function pluginLogWriterKey(dataDir: string, pluginId: string): string { + return `${dataDir}\u0000${pluginId}`; +} + +function getPluginLogWriter(dataDir: string, pluginId: string): PluginLogWriter { + const key = pluginLogWriterKey(dataDir, pluginId); + const existing = pluginLogWriters.get(key); + if (existing !== undefined) { + return existing; + } + const writer: PluginLogWriter = { + bufferedBytes: 0, + bufferedLines: [], + dirEnsured: false, + flushChain: Promise.resolve(), + logFileBytes: null, + timer: null, + }; + pluginLogWriters.set(key, writer); + return writer; +} + +function flushPluginLogWriter( + dataDir: string, + pluginId: string, + writer: PluginLogWriter, +): Promise { + if (writer.timer !== null) { + clearTimeout(writer.timer); + writer.timer = null; + } + if (writer.bufferedLines.length === 0) { + return writer.flushChain; + } + const chunk = writer.bufferedLines.join(""); + const chunkBytes = writer.bufferedBytes; + writer.bufferedLines = []; + writer.bufferedBytes = 0; + writer.flushChain = writer.flushChain.then(async () => { + try { + const dir = pluginLogsDir(dataDir, pluginId); + if (!writer.dirEnsured) { + mkdirSync(dir, { recursive: true }); + writer.dirEnsured = true; + } + const file = join(dir, PLUGIN_LOG_FILE); + if (writer.logFileBytes === null) { + try { + writer.logFileBytes = statSync(file).size; + } catch { + // Missing file: nothing logged there yet. + writer.logFileBytes = 0; + } + } + if (writer.logFileBytes > PLUGIN_LOG_MAX_BYTES) { + try { + renameSync(file, join(dir, PLUGIN_LOG_ROTATED_FILE)); + } catch { + // Missing file: nothing to rotate. + } + writer.logFileBytes = 0; + } + await appendFile(file, chunk, "utf8"); + writer.logFileBytes += chunkBytes; + } catch { + // Best effort only — a full disk or permission problem must not break + // the plugin call site; the prefixed server log still carries the + // messages. + } + }); + return writer.flushChain; +} + +/** + * Flush and drop a plugin's buffered log writer. Wired into the plugin API + * handle's dispose hooks so reload/disable/shutdown wait for pending lines + * instead of leaving a flush timer racing the plugin data directory's + * removal. + */ +export async function disposePluginLogWriter( + dataDir: string, + pluginId: string, +): Promise { + const key = pluginLogWriterKey(dataDir, pluginId); + const writer = pluginLogWriters.get(key); + if (writer === undefined) { + return; + } + pluginLogWriters.delete(key); + await flushPluginLogWriter(dataDir, pluginId, writer); +} + /** - * Append one log line synchronously (bb.log is a sync API; lines are tiny). - * Never throws — a full disk or permission problem must not break the plugin - * call site; the prefixed server log still carries the message. + * Buffer one log line (bb.log is a sync API; lines are tiny). Never throws — + * delivery is best effort and happens off the call path. */ export function appendPluginLogLine( dataDir: string, @@ -30,18 +143,21 @@ export function appendPluginLogLine( message: string, ): void { try { - const dir = pluginLogsDir(dataDir, pluginId); - mkdirSync(dir, { recursive: true }); - const file = join(dir, PLUGIN_LOG_FILE); - try { - if (statSync(file).size > PLUGIN_LOG_MAX_BYTES) { - renameSync(file, join(dir, PLUGIN_LOG_ROTATED_FILE)); - } - } catch { - // Missing file: nothing to rotate. + const writer = getPluginLogWriter(dataDir, pluginId); + const line = `${JSON.stringify({ ts: Date.now(), level, message })}\n`; + writer.bufferedLines.push(line); + writer.bufferedBytes += Buffer.byteLength(line, "utf8"); + if (writer.bufferedBytes >= PLUGIN_LOG_FLUSH_THRESHOLD_BYTES) { + void flushPluginLogWriter(dataDir, pluginId, writer); + return; + } + if (writer.timer === null) { + writer.timer = setTimeout(() => { + writer.timer = null; + void flushPluginLogWriter(dataDir, pluginId, writer); + }, PLUGIN_LOG_FLUSH_INTERVAL_MS); + writer.timer.unref?.(); } - const line = JSON.stringify({ ts: Date.now(), level, message }); - appendFileSync(file, `${line}\n`, "utf8"); } catch { // Best effort only. } @@ -53,13 +169,18 @@ function splitLines(content: string): string[] { /** * Last `tail` log lines across the rotated file plus the current one, oldest - * first. Missing files read as empty. + * first. Missing files read as empty. Pending buffered lines are flushed + * first so a tail right after `bb.log` stays read-your-writes. */ export async function readPluginLogTail( dataDir: string, pluginId: string, tail: number, ): Promise { + const writer = pluginLogWriters.get(pluginLogWriterKey(dataDir, pluginId)); + if (writer !== undefined) { + await flushPluginLogWriter(dataDir, pluginId, writer); + } const dir = pluginLogsDir(dataDir, pluginId); const lines: string[] = []; for (const name of [PLUGIN_LOG_ROTATED_FILE, PLUGIN_LOG_FILE]) { diff --git a/apps/server/src/services/system/event-pruning.ts b/apps/server/src/services/system/event-pruning.ts index 2f584750e5..277b82643d 100644 --- a/apps/server/src/services/system/event-pruning.ts +++ b/apps/server/src/services/system/event-pruning.ts @@ -1,4 +1,5 @@ import { performance } from "node:perf_hooks"; +import { and, eq, inArray, lte } from "drizzle-orm"; import { getThread, getLatestThreadSequence, @@ -7,6 +8,7 @@ import { pruneResolvedItemDeltas, pruneTokenUsageEventsBeforeSequence, pruneThreadEventsBeforeSequence, + events as storedEvents, } from "@bb/db"; import type { ThreadEventType } from "@bb/domain"; import { roundDurationMs } from "../lib/duration.js"; @@ -85,6 +87,18 @@ const GENERIC_AGE_PRUNABLE_THREAD_EVENT_TYPES: readonly ThreadEventType[] = [ "turn/diff/updated", ] as const; +/** + * Candidate types for `pruneResolvedItemDeltas`. Must stay a superset of the + * delta types that DELETE targets, or the probe below silently disables the + * prune for the missing type. + */ +const RESOLVED_ITEM_DELTA_THREAD_EVENT_TYPES: readonly ThreadEventType[] = [ + "item/agentMessage/delta", + "item/commandExecution/outputDelta", + "item/reasoning/summaryTextDelta", + "item/reasoning/textDelta", +] as const; + const KEEP_RECENT_BY_MODE: Record = { active: ACTIVE_THREAD_EVENT_KEEP_RECENT, idle: IDLE_THREAD_EVENT_KEEP_RECENT, @@ -125,6 +139,49 @@ export function isActivePruneTriggerThreadEventType( return activePruneTriggerThreadEventTypeSet.has(eventType); } +interface HasPrunableCandidateRowsArgs { + threadId: string; + types: readonly ThreadEventType[]; + /** Omit for classes whose DELETE has no sequence bound. */ + sequenceCutoff?: number; +} + +/** + * LIMIT 1 necessary-condition probe for a prune class. The usage-row DELETEs + * walk every row of their type through a correlated-subquery CTE, and the + * resolved-delta/background-progress DELETEs walk their candidates with + * correlated EXISTS checks — even when the thread has no candidate rows at + * all, which is the common case for an active thread pruned every ~30s. This + * probe answers "could that DELETE possibly match?" with one covering seek on + * the (thread_id, type, sequence) index, so the no-op case skips the walk + * outright. A true result only means candidates exist; the DELETE still + * decides what is actually prunable. + */ +function hasPrunableCandidateRows( + db: AppDeps["db"], + args: HasPrunableCandidateRowsArgs, +): boolean { + if (args.sequenceCutoff !== undefined && args.sequenceCutoff <= 0) { + return false; + } + return ( + db + .select({ id: storedEvents.id }) + .from(storedEvents) + .where( + and( + eq(storedEvents.threadId, args.threadId), + inArray(storedEvents.type, [...args.types]), + ...(args.sequenceCutoff === undefined + ? [] + : [lte(storedEvents.sequence, args.sequenceCutoff)]), + ), + ) + .limit(1) + .get() !== undefined + ); +} + export function pruneThreadEventHistory( deps: Pick, args: PruneThreadEventHistoryArgs, @@ -140,17 +197,31 @@ export function pruneThreadEventHistory( const sequenceCutoff = Math.max(0, latestSequence - keepRecent); const removedAgePrunableEvents = runThreadEventPruningStep("prune_context_window_usage", () => - pruneContextWindowUsageEventsBeforeSequence(deps.db, { + hasPrunableCandidateRows(deps.db, { threadId: args.threadId, + types: ["thread/contextWindowUsage/updated"], sequenceCutoff, - }), + }) + ? pruneContextWindowUsageEventsBeforeSequence(deps.db, { + threadId: args.threadId, + sequenceCutoff, + }) + : 0, ) + runThreadEventPruningStep("prune_token_usage", () => - pruneTokenUsageEventsBeforeSequence(deps.db, { + hasPrunableCandidateRows(deps.db, { threadId: args.threadId, + types: ["thread/tokenUsage/updated"], sequenceCutoff, - }), + }) + ? pruneTokenUsageEventsBeforeSequence(deps.db, { + threadId: args.threadId, + sequenceCutoff, + }) + : 0, ) + + // No probe: this DELETE is already a plain (thread_id, type, sequence) + // index-range scan, exactly what the probe would run. runThreadEventPruningStep("prune_generic_age_prunable_events", () => pruneThreadEventsBeforeSequence(deps.db, { threadId: args.threadId, @@ -161,16 +232,26 @@ export function pruneThreadEventHistory( const removedResolvedItemDeltas = runThreadEventPruningStep( "prune_resolved_item_deltas", () => - pruneResolvedItemDeltas(deps.db, { + hasPrunableCandidateRows(deps.db, { threadId: args.threadId, - }), + types: RESOLVED_ITEM_DELTA_THREAD_EVENT_TYPES, + }) + ? pruneResolvedItemDeltas(deps.db, { + threadId: args.threadId, + }) + : 0, ); const removedBackgroundTaskProgressEvents = runThreadEventPruningStep( "prune_background_task_progress", () => - pruneBackgroundTaskProgressEvents(deps.db, { + hasPrunableCandidateRows(deps.db, { threadId: args.threadId, - }), + types: ["item/backgroundTask/progress"], + }) + ? pruneBackgroundTaskProgressEvents(deps.db, { + threadId: args.threadId, + }) + : 0, ); return { diff --git a/apps/server/src/services/threads/timeline-cache.ts b/apps/server/src/services/threads/timeline-cache.ts index f5e1c903ae..e90bb740c7 100644 --- a/apps/server/src/services/threads/timeline-cache.ts +++ b/apps/server/src/services/threads/timeline-cache.ts @@ -22,52 +22,112 @@ import type { ThreadTimelinePageRequest } from "./timeline-pagination.js"; * (`pruneResolvedItemDeltas`, background-task progress) is output-preserving * and never lowers `maxSeq`, so it cannot stale a cached entry. * - * Entries with many rows are not cached: an expanded active turn (the streaming - * case) produces hundreds of rows AND a `maxSeq` that changes on every event, - * so caching it only thrashes the LRU and pins large objects for no reuse. Idle - * windows collapse completed turns to a handful of rows regardless of thread - * size, so the cap excludes exactly the entries that would never be reused. + * Entries with many rows are not stored in the LRU: an expanded active turn + * (the streaming case) produces hundreds of rows AND a `maxSeq` that changes on + * every event, so retaining it only thrashes the LRU and pins large objects for + * no reuse. Idle windows collapse completed turns to a handful of rows + * regardless of thread size, so the cap excludes exactly the entries that would + * never be reused. + * + * Those uncached streaming windows are instead shared through a short-lived + * last-build slot per params key (the key minus `maxSeq`). Several clients + * watch the same streaming thread (desktop, browser, phone, plugin panes), each + * on its own refetch pacing, so one appended event fans out into several + * back-to-back synchronous rebuilds of near-identical windows — they serialize + * on the event loop and stall every other request. The slot serves them all + * from one build: + * + * - Same key inside {@link DEFAULT_SHARE_WINDOW_MS}: the build is shared + * whatever its row count. Identical key means identical output (the build is + * deterministic), so this is invisible except in build count. The build is + * synchronous, so it completes before the next request can start; sharing the + * completed result is what coalesces loop-serialized concurrent requests. + * - New `maxSeq` inside the window: the prior window is returned as-is — a + * rebuild floor. The response carries its own `maxSeq`, so the client simply + * sees the thread as of ≤{@link DEFAULT_SHARE_WINDOW_MS} ago, below client + * refetch pacing. The floor applies ONLY to responses over the LRU row cap: + * LRU-cacheable windows already coalesce same-key storms via the LRU, and + * keeping their rebuilds eager preserves per-request freshness for every + * window the LRU can serve. `thread.status` lives in the params key, so a + * status flip (interrupt, completion) is never floored. */ const DEFAULT_MAX_ENTRIES = 128; const DEFAULT_MAX_CACHEABLE_ROWS = 200; +const DEFAULT_SHARE_WINDOW_MS = 250; interface ThreadTimelineCacheOptions { maxEntries?: number; - /** Responses with more rows than this are returned but not stored. */ + /** Responses with more rows than this are returned but not LRU-stored. */ maxCacheableRows?: number; + /** How long one build is shared across requests of the same params key. */ + shareWindowMs?: number; + /** Clock override for tests; defaults to Date.now. */ + now?: () => number; +} + +interface ThreadTimelineCacheKeys { + /** Full cache key including `maxSeq` ({@link buildThreadTimelineCacheKey}). */ + key: string; + /** The key minus `maxSeq` ({@link buildThreadTimelineParamsKey}). */ + paramsKey: string; } interface ThreadTimelineCache { getOrBuild( - key: string, + keys: ThreadTimelineCacheKeys, build: () => ThreadTimelineResponse, ): ThreadTimelineResponse; - /** Number of currently cached entries (for tests/metrics). */ + /** Number of currently cached LRU entries (for tests/metrics). */ readonly size: number; } +interface RecentBuild { + key: string; + value: ThreadTimelineResponse; + builtAt: number; +} + export function createThreadTimelineCache( options: ThreadTimelineCacheOptions = {}, ): ThreadTimelineCache { const maxEntries = options.maxEntries ?? DEFAULT_MAX_ENTRIES; const maxCacheableRows = options.maxCacheableRows ?? DEFAULT_MAX_CACHEABLE_ROWS; + const shareWindowMs = options.shareWindowMs ?? DEFAULT_SHARE_WINDOW_MS; + const now = options.now ?? Date.now; const entries = new Map(); + // Requests within a params key arrive with monotonic `maxSeq` (the route + // reads the current high-water mark), so one slot per params key holding the + // newest build is enough — no request can want an older revision. Expired + // slots are swept on every store, so the map holds at most the few windows + // built in the last share window. + const recentBuilds = new Map(); return { - getOrBuild(key, build) { - const cached = entries.get(key); + getOrBuild(keys, build) { + const cached = entries.get(keys.key); if (cached !== undefined) { // Re-insert to mark most-recently-used. - entries.delete(key); - entries.set(key, cached); + entries.delete(keys.key); + entries.set(keys.key, cached); return cached; } + const at = now(); + const recent = recentBuilds.get(keys.paramsKey); + if (recent !== undefined && at - recent.builtAt <= shareWindowMs) { + if ( + recent.key === keys.key || + recent.value.rows.length > maxCacheableRows + ) { + return recent.value; + } + } + const value = build(); if (value.rows.length <= maxCacheableRows) { - entries.set(key, value); + entries.set(keys.key, value); while (entries.size > maxEntries) { const oldest = entries.keys().next().value; if (oldest === undefined) { @@ -76,6 +136,13 @@ export function createThreadTimelineCache( entries.delete(oldest); } } + const builtAt = now(); + recentBuilds.set(keys.paramsKey, { key: keys.key, value, builtAt }); + for (const [paramsKey, slot] of recentBuilds) { + if (builtAt - slot.builtAt > shareWindowMs) { + recentBuilds.delete(paramsKey); + } + } return value; }, get size() { diff --git a/apps/server/src/ws/hub.ts b/apps/server/src/ws/hub.ts index 271122f2d5..fb652f8c27 100644 --- a/apps/server/src/ws/hub.ts +++ b/apps/server/src/ws/hub.ts @@ -35,7 +35,14 @@ const TERMINAL_SOCKET_HIGH_WATER_BYTES = 1024 * 1024; // A 16 MiB raw burst expands to about 21.4 MiB as base64 + JSON. Keep // enough bounded headroom for that workload while preventing unbounded growth. const TERMINAL_SOCKET_MAX_QUEUE_BYTES = 32 * 1024 * 1024; -const TERMINAL_SOCKET_DRAIN_POLL_MS = 10; +// App realtime messages are small invalidation hints and ephemeral signals; a +// socket that cannot flush 1 MiB of them is wedged (a suspended phone, a dead +// network), and letting its buffers grow just leaks server memory. Queue a +// short burst, then drop the socket: clients reconnect and catch up via the +// watermark. +const APP_SOCKET_HIGH_WATER_BYTES = 1024 * 1024; +const APP_SOCKET_MAX_QUEUE_BYTES = 4 * 1024 * 1024; +const SOCKET_DRAIN_POLL_MS = 10; /** * A streaming turn appends events ~10 times a second. A client that only * subscribes to the thread list (every open app window, for every thread it @@ -52,18 +59,40 @@ const THREAD_LIST_EVENTS_APPENDED_COALESCE_MS = 1_000; const LIST_RELEVANT_THREAD_EVENT_TYPES: ReadonlySet = new Set(["client/turn/requested", "turn/completed"]); +/** + * The plugin-signal envelope minus the payload: `notifyPluginSignal` embeds + * the publish site's already-serialized payload verbatim, so only the fields + * it splices into the frame go through the outgoing schema. + */ +const pluginSignalEnvelopeSchema = pluginSignalSchema.omit({ payload: true }); + interface HubSocket { close(code?: number, reason?: string): void; raw?: { bufferedAmount: number }; send(data: string): void; } -interface TerminalSocketSendQueue { +interface SocketSendQueue { bytes: number; payloads: string[]; timeout: ReturnType | null; } +/** + * One backpressured send path. The hub runs two: the terminal lane (large + * ordered output bursts, generous queue) and the app lane (small realtime + * notifications, tight queue). Both share the same mechanics — send directly + * below the socket high water, queue and poll-drain above it, and hand the + * socket to `drop` when a send throws or the queue budget overflows. + */ +interface SocketSendLane { + drop: (socket: HubSocket, reason: string) => void; + dropReasons: { queueOverflow: string; sendFailed: string }; + highWaterBytes: number; + maxQueueBytes: number; + queues: Map; +} + type ChangedMessageListener = (message: ChangedMessage) => void; interface PendingThreadListEventsAppended { @@ -225,10 +254,26 @@ export class NotificationHub implements DbNotifier { string, Set >(); - private readonly terminalSocketSendQueues = new Map< - HubSocket, - TerminalSocketSendQueue - >(); + private readonly terminalSendLane: SocketSendLane = { + drop: (socket, reason) => this.dropTerminalSocket(socket, reason), + dropReasons: { + queueOverflow: "terminal-backpressure", + sendFailed: "terminal-send-failed", + }, + highWaterBytes: TERMINAL_SOCKET_HIGH_WATER_BYTES, + maxQueueBytes: TERMINAL_SOCKET_MAX_QUEUE_BYTES, + queues: new Map(), + }; + private readonly appSendLane: SocketSendLane = { + drop: (socket, reason) => this.dropAppSocket(socket, reason), + dropReasons: { + queueOverflow: "app-backpressure", + sendFailed: "app-send-failed", + }, + highWaterBytes: APP_SOCKET_HIGH_WATER_BYTES, + maxQueueBytes: APP_SOCKET_MAX_QUEUE_BYTES, + queues: new Map(), + }; private readonly terminalIdsByClientSocket = new Map< HubSocket, Set @@ -251,6 +296,7 @@ export class NotificationHub implements DbNotifier { unregisterClient(socket: HubSocket): void { this.unregisterTerminalClientSocket(socket); + this.clearSocketSendQueue(this.appSendLane, socket); const keys = this.clientKeysBySocket.get(socket); if (!keys) { return; @@ -314,7 +360,7 @@ export class NotificationHub implements DbNotifier { terminalIds.delete(terminalId); if (terminalIds.size === 0) { this.terminalIdsByClientSocket.delete(socket); - this.clearTerminalSocketSendQueue(socket); + this.clearSocketSendQueue(this.terminalSendLane, socket); } } @@ -337,7 +383,7 @@ export class NotificationHub implements DbNotifier { } this.terminalIdsByClientSocket.delete(socket); - this.clearTerminalSocketSendQueue(socket); + this.clearSocketSendQueue(this.terminalSendLane, socket); } private releaseTerminalResizeOwnership( @@ -363,7 +409,8 @@ export class NotificationHub implements DbNotifier { socket: HubSocket, message: TerminalServerMessage, ): void { - this.sendOrQueueTerminalPayload( + this.sendWithBackpressure( + this.terminalSendLane, socket, JSON.stringify(terminalServerMessageSchema.parse(message)), ); @@ -380,22 +427,32 @@ export class NotificationHub implements DbNotifier { const payload = JSON.stringify(terminalServerMessageSchema.parse(message)); for (const socket of [...sockets]) { - this.sendOrQueueTerminalPayload(socket, payload); + this.sendWithBackpressure(this.terminalSendLane, socket, payload); } } - private sendOrQueueTerminalPayload(socket: HubSocket, payload: string): void { - const existingQueue = this.terminalSocketSendQueues.get(socket); + /** + * Send on a lane, queueing above the socket high water and dropping the + * socket when a send throws or the queue budget overflows — one bad socket + * can neither stall the loop nor grow without bound. Returns false when the + * socket was dropped instead of sent/queued to. + */ + private sendWithBackpressure( + lane: SocketSendLane, + socket: HubSocket, + payload: string, + ): boolean { + const existingQueue = lane.queues.get(socket); if ( !existingQueue && - (socket.raw?.bufferedAmount ?? 0) <= TERMINAL_SOCKET_HIGH_WATER_BYTES + (socket.raw?.bufferedAmount ?? 0) <= lane.highWaterBytes ) { try { socket.send(payload); - return; + return true; } catch { - this.dropTerminalSocket(socket, "terminal-send-failed"); - return; + lane.drop(socket, lane.dropReasons.sendFailed); + return false; } } @@ -405,39 +462,42 @@ export class NotificationHub implements DbNotifier { timeout: null, }; const payloadBytes = Buffer.byteLength(payload, "utf8"); - if (queue.bytes + payloadBytes > TERMINAL_SOCKET_MAX_QUEUE_BYTES) { - this.dropTerminalSocket(socket, "terminal-backpressure"); - return; + if (queue.bytes + payloadBytes > lane.maxQueueBytes) { + lane.drop(socket, lane.dropReasons.queueOverflow); + return false; } queue.payloads.push(payload); queue.bytes += payloadBytes; - this.terminalSocketSendQueues.set(socket, queue); - this.scheduleTerminalSocketDrain(socket, queue); + lane.queues.set(socket, queue); + this.scheduleSocketDrain(lane, socket, queue); + return true; } - private scheduleTerminalSocketDrain( + private scheduleSocketDrain( + lane: SocketSendLane, socket: HubSocket, - queue: TerminalSocketSendQueue, + queue: SocketSendQueue, ): void { if (queue.timeout !== null) { return; } queue.timeout = setTimeout(() => { queue.timeout = null; - this.flushTerminalSocketQueue(socket, queue); - }, TERMINAL_SOCKET_DRAIN_POLL_MS); + this.flushSocketQueue(lane, socket, queue); + }, SOCKET_DRAIN_POLL_MS); } - private flushTerminalSocketQueue( + private flushSocketQueue( + lane: SocketSendLane, socket: HubSocket, - queue: TerminalSocketSendQueue, + queue: SocketSendQueue, ): void { - if (this.terminalSocketSendQueues.get(socket) !== queue) { + if (lane.queues.get(socket) !== queue) { return; } while ( queue.payloads.length > 0 && - (socket.raw?.bufferedAmount ?? 0) <= TERMINAL_SOCKET_HIGH_WATER_BYTES + (socket.raw?.bufferedAmount ?? 0) <= lane.highWaterBytes ) { const payload = queue.payloads[0]; if (payload === undefined) { @@ -446,17 +506,17 @@ export class NotificationHub implements DbNotifier { try { socket.send(payload); } catch { - this.dropTerminalSocket(socket, "terminal-send-failed"); + lane.drop(socket, lane.dropReasons.sendFailed); return; } queue.payloads.shift(); queue.bytes -= Buffer.byteLength(payload, "utf8"); } if (queue.payloads.length === 0) { - this.clearTerminalSocketSendQueue(socket); + this.clearSocketSendQueue(lane, socket); return; } - this.scheduleTerminalSocketDrain(socket, queue); + this.scheduleSocketDrain(lane, socket, queue); } private dropTerminalSocket(socket: HubSocket, reason: string): void { @@ -468,15 +528,29 @@ export class NotificationHub implements DbNotifier { } } - private clearTerminalSocketSendQueue(socket: HubSocket): void { - const queue = this.terminalSocketSendQueues.get(socket); + /** + * Unregister first (which also clears the socket's queues) so the rest of + * the current fan-out and all later ones skip the socket; the client + * reconnects and catches up via the watermark. + */ + private dropAppSocket(socket: HubSocket, reason: string): void { + this.unregisterClient(socket); + try { + socket.close(1013, reason); + } catch { + // The socket is already unusable; registration and queue state are gone. + } + } + + private clearSocketSendQueue(lane: SocketSendLane, socket: HubSocket): void { + const queue = lane.queues.get(socket); if (!queue) { return; } if (queue.timeout !== null) { clearTimeout(queue.timeout); } - this.terminalSocketSendQueues.delete(socket); + lane.queues.delete(socket); } subscribe(socket: HubSocket, target: RealtimeSubscriptionTarget): void { @@ -804,12 +878,7 @@ export class NotificationHub implements DbNotifier { file: request.file, }), ); - let delivered = 0; - for (const socket of this.clientKeysBySocket.keys()) { - socket.send(payload); - delivered += 1; - } - return delivered; + return this.broadcastToAppSockets(payload); } /** Broadcast an ephemeral pane presentation request to every app client. */ @@ -825,12 +894,7 @@ export class NotificationHub implements DbNotifier { action, }), ); - let delivered = 0; - for (const socket of this.clientKeysBySocket.keys()) { - socket.send(payload); - delivered += 1; - } - return delivered; + return this.broadcastToAppSockets(payload); } /** @@ -838,24 +902,36 @@ export class NotificationHub implements DbNotifier { * every connected client. V1 broadcasts to all clients — per-channel * subscriptions arrive with the plugin frontend runtime. Returns how many * clients the signal reached. + * + * `serializedPayload` is the publish site's single JSON serialization of + * the payload, embedded verbatim in the wire frame so the fan-out never + * parses or re-stringifies it. The envelope fields still pass the strict + * outgoing schema. */ notifyPluginSignal( pluginId: string, channel: string, - payload: unknown, + serializedPayload: string, ): number { - const message = JSON.stringify( - pluginSignalSchema.parse({ - type: "plugin-signal", - pluginId, - channel, - payload, - }), - ); + const envelope = pluginSignalEnvelopeSchema.parse({ + type: "plugin-signal", + pluginId, + channel, + }); + const message = `{"type":"plugin-signal","pluginId":${JSON.stringify( + envelope.pluginId, + )},"channel":${JSON.stringify(envelope.channel)},"payload":${serializedPayload}}`; + return this.broadcastToAppSockets(message); + } + + /** Ephemeral broadcast to every app socket; returns how many accepted it. */ + private broadcastToAppSockets(payload: string): number { let delivered = 0; - for (const socket of this.clientKeysBySocket.keys()) { - socket.send(message); - delivered += 1; + // Snapshot: dropping a socket mutates the live registration map. + for (const socket of [...this.clientKeysBySocket.keys()]) { + if (this.sendWithBackpressure(this.appSendLane, socket, payload)) { + delivered += 1; + } } return delivered; } @@ -1055,11 +1131,11 @@ export class NotificationHub implements DbNotifier { return; } const detailKey = subscriptionKey({ kind: "thread-detail", threadId }); - for (const socket of listSockets) { + for (const socket of [...listSockets]) { if (this.clientKeysBySocket.get(socket)?.has(detailKey)) { continue; } - socket.send(payload); + this.sendWithBackpressure(this.appSendLane, socket, payload); } } @@ -1089,8 +1165,9 @@ export class NotificationHub implements DbNotifier { sockets: Iterable, payload: string, ): void { - for (const socket of sockets) { - socket.send(payload); + // Snapshot: dropping a socket mutates the live subscription sets. + for (const socket of [...sockets]) { + this.sendWithBackpressure(this.appSendLane, socket, payload); } } diff --git a/apps/server/test/app/hub-plugin-signal.test.ts b/apps/server/test/app/hub-plugin-signal.test.ts index 3d7e07e881..27a16263ca 100644 --- a/apps/server/test/app/hub-plugin-signal.test.ts +++ b/apps/server/test/app/hub-plugin-signal.test.ts @@ -12,9 +12,13 @@ describe("NotificationHub.notifyPluginSignal", () => { hub.subscribe(first, { kind: "thread-detail", threadId: "thr_1" }); hub.subscribe(second, { kind: "system" }); - const delivered = hub.notifyPluginSignal("linear", "issues-updated", { - count: 42, - }); + // The payload arrives pre-serialized (the publish site's single + // JSON.stringify) and is embedded verbatim in the wire frame. + const delivered = hub.notifyPluginSignal( + "linear", + "issues-updated", + JSON.stringify({ count: 42 }), + ); expect(delivered).toBe(2); for (const socket of [first, second]) { diff --git a/apps/server/test/app/hub.test.ts b/apps/server/test/app/hub.test.ts index 5d8ebcee69..5c5eeab901 100644 --- a/apps/server/test/app/hub.test.ts +++ b/apps/server/test/app/hub.test.ts @@ -250,6 +250,87 @@ describe("NotificationHub", () => { expect(socket.messages).toEqual([]); }); + it("queues app notifications for a socket above high water and flushes on drain", () => { + vi.useFakeTimers(); + const hub = new NotificationHub(); + const healthy = createMockHubSocket(); + const wedged = { + ...createMockHubSocket(), + raw: { bufferedAmount: 2 * 1024 * 1024 }, + }; + hub.subscribe(healthy, { kind: "thread-detail", threadId: "thread-1" }); + hub.subscribe(wedged, { kind: "thread-detail", threadId: "thread-1" }); + + hub.notifyThread("thread-1", ["status-changed"]); + + expect(healthy.messages).toHaveLength(1); + expect(wedged.messages).toHaveLength(0); + + wedged.raw.bufferedAmount = 0; + vi.advanceTimersByTime(10); + expect(wedged.messages).toHaveLength(1); + }); + + it("drops a wedged app socket at the queue cap while siblings still receive", () => { + const hub = new NotificationHub(); + const healthy = createMockHubSocket(); + const wedged = { + ...createMockHubSocket(), + raw: { bufferedAmount: 2 * 1024 * 1024 }, + }; + hub.registerClient(healthy); + hub.registerClient(wedged); + const bigPayload = JSON.stringify("x".repeat(1024 * 1024)); + + for (let index = 0; index < 5; index += 1) { + hub.notifyPluginSignal("plugin-1", "channel-1", bigPayload); + } + + expect(wedged.closed).toContainEqual({ + code: 1013, + reason: "app-backpressure", + }); + expect(wedged.messages).toEqual([]); + expect(healthy.messages).toHaveLength(5); + + // The drop unregistered the socket: later broadcasts skip it entirely. + const delivered = hub.notifyPluginSignal( + "plugin-1", + "channel-1", + JSON.stringify("after"), + ); + expect(delivered).toBe(1); + expect(wedged.closed).toHaveLength(1); + }); + + it("isolates a throwing app socket from healthy subscribers", () => { + const hub = new NotificationHub(); + const healthy = createMockHubSocket(); + const closed: Array<{ code?: number; reason?: string }> = []; + const failing = { + close(code?: number, reason?: string) { + closed.push({ code, reason }); + }, + send() { + throw new Error("socket send failed"); + }, + }; + hub.subscribe(failing, { kind: "thread-detail", threadId: "thread-1" }); + hub.subscribe(healthy, { kind: "thread-detail", threadId: "thread-1" }); + + expect(() => + hub.notifyThread("thread-1", ["status-changed"]), + ).not.toThrow(); + + expect(closed).toEqual([{ code: 1013, reason: "app-send-failed" }]); + expect(healthy.messages).toHaveLength(1); + + // The failing socket was unregistered; the next notify skips it. + hub.notifyThread("thread-1", ["status-changed"]); + expect(closed).toHaveLength(1); + expect(healthy.messages).toHaveLength(2); + }); + it("notifies all clients subscribed to the same thread", () => { const hub = new NotificationHub(); const socket1 = createMockHubSocket(); diff --git a/apps/server/test/public/public-thread-data.test.ts b/apps/server/test/public/public-thread-data.test.ts index 41f204bd43..d59e082f88 100644 --- a/apps/server/test/public/public-thread-data.test.ts +++ b/apps/server/test/public/public-thread-data.test.ts @@ -210,6 +210,33 @@ describe("public thread data routes", () => { }); }); + it("caps thread lists and reports truncation in the has-more header", async () => { + await withTestHarness(async (harness) => { + const { host } = seedHostSession(harness.deps); + const { project } = seedProjectWithSource(harness.deps, { + hostId: host.id, + }); + for (let index = 0; index < 3; index += 1) { + seedThread(harness.deps, { + projectId: project.id, + title: `Thread ${index}`, + }); + } + + const capped = await harness.app.request("/api/v1/threads?limit=2"); + expect(capped.status).toBe(200); + expect(capped.headers.get("x-bb-thread-list-has-more")).toBe("true"); + expect(z.array(z.unknown()).parse(await readJson(capped))).toHaveLength( + 2, + ); + + const full = await harness.app.request("/api/v1/threads"); + expect(full.status).toBe(200); + expect(full.headers.get("x-bb-thread-list-has-more")).toBe("false"); + expect(z.array(z.unknown()).parse(await readJson(full))).toHaveLength(3); + }); + }); + it("updates visibility and requires includeHidden to list hidden threads", async () => { await withTestHarness(async (harness) => { const { host } = seedHostSession(harness.deps); diff --git a/apps/server/test/services/plugins/plugin-cli.test.ts b/apps/server/test/services/plugins/plugin-cli.test.ts index 424026a514..5cb9a8c9a8 100644 --- a/apps/server/test/services/plugins/plugin-cli.test.ts +++ b/apps/server/test/services/plugins/plugin-cli.test.ts @@ -365,6 +365,19 @@ describe("plugin CLI commands (bb.cli.register + endpoints + skill + logs)", () }); it("bb.log writes JSONL to the plugin log file and the tail endpoint serves it", async () => { + // The tail endpoint flushes buffered lines (bb.log batches writes off the + // call path), so hit it before reading the raw file. + const response = await harness.app.request( + `${BASE}/api/v1/plugins/acme/logs?tail=1`, + ); + expect(response.status).toBe(200); + const body = (await response.json()) as { ok: boolean; lines: string[] }; + expect(body.ok).toBe(true); + expect(body.lines).toHaveLength(1); + expect(JSON.parse(body.lines[0] ?? "")).toMatchObject({ + message: "cli plugin loaded", + }); + const logFile = join( harness.config.dataDir, "plugins", @@ -383,17 +396,6 @@ describe("plugin CLI commands (bb.cli.register + endpoints + skill + logs)", () expect(parsed.message).toBe("cli plugin loaded"); expect(typeof parsed.ts).toBe("number"); - const response = await harness.app.request( - `${BASE}/api/v1/plugins/acme/logs?tail=1`, - ); - expect(response.status).toBe(200); - const body = (await response.json()) as { ok: boolean; lines: string[] }; - expect(body.ok).toBe(true); - expect(body.lines).toHaveLength(1); - expect(JSON.parse(body.lines[0] ?? "")).toMatchObject({ - message: "cli plugin loaded", - }); - const missing = await harness.app.request( `${BASE}/api/v1/plugins/nope/logs`, ); diff --git a/apps/server/test/services/plugins/plugin-log.test.ts b/apps/server/test/services/plugins/plugin-log.test.ts new file mode 100644 index 0000000000..af76a88d76 --- /dev/null +++ b/apps/server/test/services/plugins/plugin-log.test.ts @@ -0,0 +1,108 @@ +import { existsSync, readFileSync } from "node:fs"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, it, vi } from "vitest"; +import { + appendPluginLogLine, + readPluginLogTail, +} from "../../../src/services/plugins/plugin-log.js"; + +function logFilePath(dataDir: string, pluginId: string): string { + return join(dataDir, "plugins", pluginId, "logs", "plugin.log"); +} + +function fileLines(path: string): string[] { + return readFileSync(path, "utf8") + .split("\n") + .filter((line) => line.length > 0); +} + +async function withDataDir( + run: (dataDir: string) => Promise, +): Promise { + const dataDir = await mkdtemp(join(tmpdir(), "bb-plugin-log-")); + try { + await run(dataDir); + } finally { + await rm(dataDir, { recursive: true, force: true }); + } +} + +describe("plugin log batching", () => { + it("buffers lines off the call path and flushes them as one batch", async () => { + await withDataDir(async (dataDir) => { + for (let index = 0; index < 3; index += 1) { + appendPluginLogLine(dataDir, "batch", "info", `line-${index}`); + } + // Nothing hits the disk synchronously on the call path. + expect(existsSync(logFilePath(dataDir, "batch"))).toBe(false); + + // The tail flushes pending lines first, so reads stay read-your-writes. + const lines = await readPluginLogTail(dataDir, "batch", 10); + expect(lines).toHaveLength(3); + expect(JSON.parse(lines[0] ?? "")).toMatchObject({ + level: "info", + message: "line-0", + }); + expect(fileLines(logFilePath(dataDir, "batch"))).toHaveLength(3); + }); + }); + + it("flushes on the timer without a reader", async () => { + await withDataDir(async (dataDir) => { + appendPluginLogLine(dataDir, "timer", "warn", "solo"); + expect(existsSync(logFilePath(dataDir, "timer"))).toBe(false); + + await vi.waitFor( + () => { + expect(existsSync(logFilePath(dataDir, "timer"))).toBe(true); + }, + { timeout: 2_000 }, + ); + expect(fileLines(logFilePath(dataDir, "timer"))).toHaveLength(1); + }); + }); + + it("flushes immediately once the buffer crosses the byte threshold", async () => { + await withDataDir(async (dataDir) => { + const big = "x".repeat(4 * 1024); + appendPluginLogLine(dataDir, "burst", "info", big); + expect(existsSync(logFilePath(dataDir, "burst"))).toBe(false); + + // The second line crosses the 8KB threshold: no timer wait. + appendPluginLogLine(dataDir, "burst", "info", big); + await vi.waitFor( + () => { + expect(existsSync(logFilePath(dataDir, "burst"))).toBe(true); + }, + { timeout: 500 }, + ); + expect(fileLines(logFilePath(dataDir, "burst"))).toHaveLength(2); + }); + }); + + it("rotates past the size cap using the cached size counter", async () => { + await withDataDir(async (dataDir) => { + appendPluginLogLine( + dataDir, + "rotate", + "info", + "x".repeat(6 * 1024 * 1024), + ); + await readPluginLogTail(dataDir, "rotate", 1); + + appendPluginLogLine(dataDir, "rotate", "info", "after-rotation"); + const lines = await readPluginLogTail(dataDir, "rotate", 10); + + const dir = join(dataDir, "plugins", "rotate", "logs"); + expect(existsSync(join(dir, "plugin.log.1"))).toBe(true); + expect(fileLines(logFilePath(dataDir, "rotate"))).toHaveLength(1); + // The tail spans the rotated file plus the current one. + expect(lines).toHaveLength(2); + expect(JSON.parse(lines[1] ?? "")).toMatchObject({ + message: "after-rotation", + }); + }); + }); +}); diff --git a/apps/server/test/services/plugins/plugin-wire.test.ts b/apps/server/test/services/plugins/plugin-wire.test.ts index ff7879dc35..5cbcc29ebd 100644 --- a/apps/server/test/services/plugins/plugin-wire.test.ts +++ b/apps/server/test/services/plugins/plugin-wire.test.ts @@ -25,6 +25,11 @@ const WIRE_SOURCE = ` input: z.object({ channel: z.string(), payload: z.unknown() }), output: z.literal("published"), }, + publishMany: { + input: z.object({ channel: z.string(), count: z.number() }), + output: z.literal("published-many"), + }, + publishHuge: { input: z.record(z.string(), z.unknown()), output: z.null() }, publishBad: { input: z.record(z.string(), z.unknown()), output: z.null() }, invalidOutput: { input: z.null(), output: z.string() }, bigintResult: { input: z.null(), output: z.any() }, @@ -94,6 +99,16 @@ const WIRE_SOURCE = ` bb.realtime.publish(input.channel, input.payload); return "published"; }, + publishMany: async (input: any) => { + for (let index = 0; index < input.count; index += 1) { + bb.realtime.publish(input.channel, { index }); + } + return "published-many"; + }, + publishHuge: async () => { + bb.realtime.publish("huge", "x".repeat(65 * 1024)); + return null; + }, publishBad: async () => { bb.realtime.publish("bad", { n: BigInt(1) }); }, @@ -667,6 +682,52 @@ describe("plugin wire surfaces (http/rpc dispatcher + realtime)", () => { }); }); + it("bb.realtime.publish rejects payloads over the 64KB cap", async () => { + const socket = createMockHubSocket(); + harness.hub.subscribe(socket, { kind: "system" }); + + const response = await rpc(harness, "publishHuge", {}); + expect(response.status).toBe(500); + expect(await response.json()).toMatchObject({ + ok: false, + error: { + code: "handler_error", + message: expect.stringContaining("64KB"), + }, + }); + expect(socket.messages).toHaveLength(0); + }); + + it("bb.realtime.publish drops signals over the per-plugin rate limit with one warning", async () => { + const socket = createMockHubSocket(); + harness.hub.subscribe(socket, { kind: "system" }); + const originalWarn = harness.deps.logger.warn; + const warn = vi.fn(); + harness.deps.logger.warn = warn; + try { + // 40 synchronous publishes land inside one refill tick, so exactly the + // burst allowance goes out and the rest drop with a single warning. + const response = await rpc(harness, "publishMany", { + channel: "flood", + count: 40, + }); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ + ok: true, + result: "published-many", + }); + + expect(socket.messages).toHaveLength(20); + const rateLimitWarnings = warn.mock.calls.filter(([, message]) => + String(message).includes("rate limit"), + ); + expect(rateLimitWarnings).toHaveLength(1); + expect(rateLimitWarnings[0]?.[0]).toMatchObject({ pluginId: "wire" }); + } finally { + harness.deps.logger.warn = originalWarn; + } + }); + it("rpc resolves the handler after the body arrives, so a reload during the body read never runs a stale handler", async () => { // The handler closes over its load generation: a binding resolved // before the body read (and invalidated by the mid-read reload) would diff --git a/apps/server/test/services/threads/timeline-cache.test.ts b/apps/server/test/services/threads/timeline-cache.test.ts index 141af64fec..27bec39b8a 100644 --- a/apps/server/test/services/threads/timeline-cache.test.ts +++ b/apps/server/test/services/threads/timeline-cache.test.ts @@ -57,13 +57,18 @@ const baseKeyArgs: ThreadTimelineCacheKeyArgs = { includeProviderUnhandledOperations: false, }; +/** Give each key its own params key to exercise the LRU in isolation. */ +function keys(key: string): { key: string; paramsKey: string } { + return { key, paramsKey: `params:${key}` }; +} + describe("createThreadTimelineCache", () => { it("builds once for the same key and serves cached on repeat", () => { const cache = createThreadTimelineCache(); const build = vi.fn(() => makeResponse(3)); - const first = cache.getOrBuild("k", build); - const second = cache.getOrBuild("k", build); + const first = cache.getOrBuild(keys("k"), build); + const second = cache.getOrBuild(keys("k"), build); expect(build).toHaveBeenCalledTimes(1); expect(second).toBe(first); @@ -74,40 +79,119 @@ describe("createThreadTimelineCache", () => { const cache = createThreadTimelineCache(); const build = vi.fn(() => makeResponse(3)); - cache.getOrBuild("k1", build); - cache.getOrBuild("k2", build); + cache.getOrBuild(keys("k1"), build); + cache.getOrBuild(keys("k2"), build); expect(build).toHaveBeenCalledTimes(2); }); - it("does not cache responses above the row cap (streaming expanded turns)", () => { - const cache = createThreadTimelineCache({ maxCacheableRows: 5 }); + it("does not retain responses above the row cap beyond the share window", () => { + let nowMs = 0; + const cache = createThreadTimelineCache({ + maxCacheableRows: 5, + shareWindowMs: 250, + now: () => nowMs, + }); const build = vi.fn(() => makeResponse(50)); - cache.getOrBuild("k", build); - cache.getOrBuild("k", build); + cache.getOrBuild(keys("k"), build); + nowMs += 251; + cache.getOrBuild(keys("k"), build); expect(build).toHaveBeenCalledTimes(2); expect(cache.size).toBe(0); }); it("evicts least-recently-used entries beyond maxEntries", () => { - const cache = createThreadTimelineCache({ maxEntries: 2 }); + let nowMs = 0; + const cache = createThreadTimelineCache({ maxEntries: 2, now: () => nowMs }); const build = vi.fn(() => makeResponse(1)); - cache.getOrBuild("a", build); // [a] - cache.getOrBuild("b", build); // [a,b] - cache.getOrBuild("a", build); // touch a -> [b,a] - cache.getOrBuild("c", build); // evict b -> [a,c] + cache.getOrBuild(keys("a"), build); // [a] + cache.getOrBuild(keys("b"), build); // [a,b] + cache.getOrBuild(keys("a"), build); // touch a -> [b,a] + cache.getOrBuild(keys("c"), build); // evict b -> [a,c] expect(cache.size).toBe(2); + // Step past the share window so the LRU alone answers the reprobe. + nowMs += 300; const buildAgain = vi.fn(() => makeResponse(1)); - cache.getOrBuild("a", buildAgain); // still cached - cache.getOrBuild("b", buildAgain); // evicted -> rebuild + cache.getOrBuild(keys("a"), buildAgain); // still cached + cache.getOrBuild(keys("b"), buildAgain); // evicted -> rebuild expect(buildAgain).toHaveBeenCalledTimes(1); }); }); +describe("createThreadTimelineCache share window", () => { + function createSharedCache(now: () => number) { + return createThreadTimelineCache({ + maxCacheableRows: 5, + shareWindowMs: 250, + now, + }); + } + + it("shares one build across same-key requests inside the window regardless of row count", () => { + let nowMs = 0; + const cache = createSharedCache(() => nowMs); + const build = vi.fn(() => makeResponse(50)); + + const first = cache.getOrBuild(keys("k"), build); + nowMs += 100; + const second = cache.getOrBuild(keys("k"), build); + + expect(build).toHaveBeenCalledTimes(1); + // Same revision must mean identical rows for every client. + expect(second).toBe(first); + // Shared, not retained: the LRU row cap still applies. + expect(cache.size).toBe(0); + }); + + it("floors rebuilds of over-cap windows: a new maxSeq inside the window gets the prior window", () => { + let nowMs = 0; + const cache = createSharedCache(() => nowMs); + const build = vi.fn(() => makeResponse(50)); + + const first = cache.getOrBuild({ key: "10|p", paramsKey: "p" }, build); + nowMs += 100; + const floored = cache.getOrBuild({ key: "11|p", paramsKey: "p" }, build); + expect(build).toHaveBeenCalledTimes(1); + expect(floored).toBe(first); + + nowMs += 200; // 300ms since the build: past the window. + const rebuilt = cache.getOrBuild({ key: "12|p", paramsKey: "p" }, build); + expect(build).toHaveBeenCalledTimes(2); + expect(rebuilt).not.toBe(first); + }); + + it("keeps rebuilds eager for LRU-cacheable windows (no floor below the row cap)", () => { + let nowMs = 0; + const cache = createSharedCache(() => nowMs); + const build = vi.fn(() => makeResponse(3)); + + cache.getOrBuild({ key: "10|p", paramsKey: "p" }, build); + nowMs += 10; + cache.getOrBuild({ key: "11|p", paramsKey: "p" }, build); + + expect(build).toHaveBeenCalledTimes(2); + }); + + it("never floors across params keys (a status flip builds fresh)", () => { + let nowMs = 0; + const cache = createSharedCache(() => nowMs); + const build = vi.fn(() => makeResponse(50)); + + cache.getOrBuild({ key: "10|p:active", paramsKey: "p:active" }, build); + nowMs += 10; + cache.getOrBuild( + { key: "11|p:interrupted", paramsKey: "p:interrupted" }, + build, + ); + + expect(build).toHaveBeenCalledTimes(2); + }); +}); + describe("buildThreadTimelineCacheKey", () => { it("differs when any projection input differs", () => { const base = buildThreadTimelineCacheKey(baseKeyArgs); diff --git a/apps/server/test/system/event-pruning.test.ts b/apps/server/test/system/event-pruning.test.ts index 5ebe261177..06b1854618 100644 --- a/apps/server/test/system/event-pruning.test.ts +++ b/apps/server/test/system/event-pruning.test.ts @@ -1,5 +1,5 @@ import { getThread, listEvents } from "@bb/db"; -import { turnScope } from "@bb/domain"; +import { threadScope, turnScope } from "@bb/domain"; import { groupHostDaemonEvents } from "@bb/host-daemon-contract"; import { describe, expect, it, vi } from "vitest"; import { applyTurnCompletedEvent } from "../../src/internal/turn-completed-events.js"; @@ -130,6 +130,7 @@ function seedResolvedAssistantMessage( for (const sequence of args.deltaSequences) { seedStoredEvent(harness.deps, { threadId: args.threadId, + providerThreadId: "provider-thread-1", sequence, type: "item/agentMessage/delta", scope: turnScope(turnId), @@ -144,6 +145,7 @@ function seedResolvedAssistantMessage( seedStoredEvent(harness.deps, { threadId: args.threadId, + providerThreadId: "provider-thread-1", sequence: args.completedSequence, type: "item/completed", scope: turnScope(turnId), @@ -532,6 +534,7 @@ describe("thread event pruning", () => { for (const sequence of [1_004, 1_005]) { seedStoredEvent(harness.deps, { threadId: thread.id, + providerThreadId: "provider-thread-1", scope: turnScope("turn-active"), sequence, type: "item/agentMessage/delta", @@ -580,6 +583,46 @@ describe("thread event pruning", () => { }); expect(response.status).toBe(200); + + // The prune is deferred off the response path: when the response + // settles, nothing has been removed yet. + expect( + listEventSequencesForType(harness, { + threadId: thread.id, + type: "thread/tokenUsage/updated", + }).at(0), + ).toBe(1); + expect( + listEventSequencesForType(harness, { + threadId: thread.id, + type: "item/agentMessage/delta", + itemId: "msg-completed", + }), + ).toEqual([1_001, 1_002]); + + // Pruning is output-preserving, so a timeline read in the window + // between the append and the deferred prune must project exactly what + // a post-prune read projects. + const timelineOptions = { + eventBudget: 1_000_000, + includeProviderUnhandledOperations: true, + maxInlineOutputChars: null, + maxSeq: 0, + page: { + kind: "latest", + segmentLimit: Number.MAX_SAFE_INTEGER, + }, + } as const; + const beforePrune = buildThreadTimeline( + harness.db, + thread, + timelineOptions, + ); + + // The deferred batch was scheduled with setImmediate during the + // request; one immediate turn later it has run. + await new Promise((resolve) => setImmediate(resolve)); + expect( listEventSequencesForType(harness, { threadId: thread.id, @@ -600,6 +643,61 @@ describe("thread event pruning", () => { itemId: "msg-active", }), ).toEqual([1_004, 1_005]); + + const afterPrune = buildThreadTimeline( + harness.db, + thread, + timelineOptions, + ); + expect(afterPrune.rows).toEqual(beforePrune.rows); + expect(afterPrune.contextWindowUsage).toEqual( + beforePrune.contextWindowUsage, + ); + }); + }); + + it("prunes superseded background-task progress snapshots", async () => { + await withTestHarness(async (harness) => { + const host = seedHost(harness.deps); + const { project } = seedProjectWithSource(harness.deps, { + hostId: host.id, + }); + const environment = seedEnvironment(harness.deps, { + hostId: host.id, + projectId: project.id, + }); + const thread = seedThread(harness.deps, { + projectId: project.id, + environmentId: environment.id, + }); + + // Two snapshots for the same item: only the latest is load-bearing. + for (const sequence of [1, 2]) { + seedStoredEvent(harness.deps, { + threadId: thread.id, + sequence, + type: "item/backgroundTask/progress", + scope: threadScope(), + itemId: "task-1", + itemKind: "backgroundTask", + data: { + item: { id: "task-1", type: "backgroundTask", status: "running" }, + }, + }); + } + + const result = pruneThreadEventHistory(harness.deps, { + mode: "idle", + threadId: thread.id, + }); + + expect(result.removedBackgroundTaskProgressEvents).toBe(1); + expect( + listEventSequencesForType(harness, { + threadId: thread.id, + type: "item/backgroundTask/progress", + }), + ).toEqual([2]); }); }); }); diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index 4cadc15b1a..2ebef46318 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -55,6 +55,7 @@ export { createThread, countLiveThreadsInEnvironment, countNonDeletedAssignedChildThreads, + DEFAULT_THREAD_LIST_LIMIT, getThread, getThreadExecutionOverride, hasActiveThreadAttention, diff --git a/packages/db/src/data/threads.ts b/packages/db/src/data/threads.ts index b09c32fc75..3df5356b35 100644 --- a/packages/db/src/data/threads.ts +++ b/packages/db/src/data/threads.ts @@ -373,6 +373,15 @@ export function listThreadMentionRowsByIds( .all(); } +/** + * Default row cap for thread listings. The sidebar and every SDK/CLI list + * ride these queries; without a cap one long-lived install (thousands of + * archived-then-restored threads) makes every list read scale with history. + * Generous enough that hitting it means paging, not filtering — callers pass + * an explicit `limit` to page or to read more. + */ +export const DEFAULT_THREAD_LIST_LIMIT = 200; + export interface ListThreadsOptions { projectId?: string; archived?: boolean; @@ -400,6 +409,12 @@ type ThreadRow = typeof threads.$inferSelect; export interface ListThreadsForProjectsOptions { projectIds: readonly string[]; archived?: boolean; + /** + * Row cap across all listed projects combined (the query orders by the + * active-thread sidebar order, so the cap keeps the most relevant rows). + * Defaults to {@link DEFAULT_THREAD_LIST_LIMIT}. + */ + limit?: number; } export interface PinThreadArgs { @@ -1204,10 +1219,8 @@ export function listThreadsWithPendingInteractionState( let query = threadWithPendingInteractionBaseQuery(db) .where(and(...buildListThreadsFilters(options))) .orderBy(...buildListThreadsOrderBy(options)) - .$dynamic(); - if (options.limit !== undefined) { - query = query.limit(options.limit); - } + .$dynamic() + .limit(options.limit ?? DEFAULT_THREAD_LIST_LIMIT); if (options.offset !== undefined) { query = query.offset(options.offset); } @@ -1261,6 +1274,7 @@ export function listThreadsWithPendingInteractionStateForProjects( const rows = threadWithPendingInteractionBaseQuery(db) .where(and(...buildListThreadsForProjectsFilters(options))) .orderBy(...buildListThreadsForProjectsOrderBy(options)) + .limit(options.limit ?? DEFAULT_THREAD_LIST_LIMIT) .all(); return rows.map(toThreadWithPendingInteractionState); diff --git a/packages/db/test/data/threads.test.ts b/packages/db/test/data/threads.test.ts index 69349b53fd..5256021f2a 100644 --- a/packages/db/test/data/threads.test.ts +++ b/packages/db/test/data/threads.test.ts @@ -7,6 +7,7 @@ import { createThread, countLiveThreadsInEnvironment, countNonDeletedAssignedChildThreads, + DEFAULT_THREAD_LIST_LIMIT, getThread, getThreadExecutionOverride, hasActiveThreadAttention, @@ -235,6 +236,38 @@ describe("threads", () => { ).toBe(2); }); + it("caps thread listings at the default limit and honors explicit limits", () => { + const { db, project } = setup(); + for (let index = 0; index < DEFAULT_THREAD_LIST_LIMIT + 1; index += 1) { + createThread(db, noopNotifier, { + projectId: project.id, + providerId: "codex", + }); + } + + expect( + listThreadsWithPendingInteractionState(db, { projectId: project.id }), + ).toHaveLength(DEFAULT_THREAD_LIST_LIMIT); + expect( + listThreadsWithPendingInteractionState(db, { + projectId: project.id, + limit: DEFAULT_THREAD_LIST_LIMIT + 1, + }), + ).toHaveLength(DEFAULT_THREAD_LIST_LIMIT + 1); + + expect( + listThreadsWithPendingInteractionStateForProjects(db, { + projectIds: [project.id], + }), + ).toHaveLength(DEFAULT_THREAD_LIST_LIMIT); + expect( + listThreadsWithPendingInteractionStateForProjects(db, { + projectIds: [project.id], + limit: 5, + }), + ).toHaveLength(5); + }); + it("allows hidden threads to belong to sections", () => { const { db, project } = setup(); const section = mustCreateThreadSection(db, "Work");