diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index f8acaf065cb9..e045c6421873 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -6,6 +6,7 @@ import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/sql/SqlClient"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as EventStore from "./EventStore.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; @@ -226,8 +227,8 @@ export const layer: Layer.Layer< const supersedableThreadEventTypes = new Set(SUPERSEDABLE_THREAD_EVENT_TYPES); const readStateThreadEventTypes = new Set(READ_STATE_THREAD_EVENT_TYPES); // Keep subagent.updated and turn-item.updated rows: slimmed payloads reference their prompts. - // Discovery runs on the shared synchronous connection; json_extract over - // large payloads made 500-row pages block the event loop for ~90ms. + // Discovery reads every event, so it runs on the reader worker where one is + // available; small pages keep other reads from queueing behind it. const COMPACTION_PAGE_SIZE = 50; /** @@ -299,7 +300,7 @@ export const layer: Layer.Layer< WHERE event.sequence <= ${throughSequence} ORDER BY event.sequence DESC LIMIT ${COMPACTION_PAGE_SIZE} - `; + `.pipe(NodeSqliteClient.readOnly); const obsolete: number[] = []; for (const row of rows) { if (row.imported_legacy_thread === 1) { @@ -342,8 +343,8 @@ export const layer: Layer.Layer< } } } - // Both discovery and deletion use the synchronous connection. Yield even - // when this page has nothing to delete so startup and requests can progress. + // Without a reader worker discovery shares the synchronous connection. + // Yield even when this page has nothing to delete so requests can progress. yield* Effect.yieldNow; if (obsolete.length > 0) { yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`; @@ -376,7 +377,7 @@ export const layer: Layer.Layer< WHERE receipt.rowid <= ${throughReceiptRowId} ORDER BY receipt.rowid DESC LIMIT ${COMPACTION_PAGE_SIZE} - `; + `.pipe(NodeSqliteClient.readOnly); const obsolete = rows .filter((row) => row.imported_legacy_thread === 1) .map((row) => row.command_id); diff --git a/apps/server/src/persistence/DatabaseStorage.ts b/apps/server/src/persistence/DatabaseStorage.ts index 3866020c1c1c..56924a2c004b 100644 --- a/apps/server/src/persistence/DatabaseStorage.ts +++ b/apps/server/src/persistence/DatabaseStorage.ts @@ -1,7 +1,10 @@ import type { DatabaseStorageStatus } from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Equal from "effect/Equal"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schedule from "effect/Schedule"; @@ -16,6 +19,7 @@ import { forkParked } from "../serverActivation.ts"; /** Pages returned per reclaim pass; one minute of churn, not a backlog. */ const RECLAIM_PAGES_PER_PASS = 1024; +const COMPACTION_INTERVAL = Duration.hours(6); export class DatabaseStorage extends Context.Service< DatabaseStorage, @@ -37,6 +41,7 @@ const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const statusRef = yield* SubscriptionRef.make(null); @@ -89,21 +94,45 @@ const make = Effect.gen(function* () { Effect.catch((cause) => Effect.logWarning("Database space reclamation failed", { cause })), ); + // A pass reads the whole event store, gigabytes on a long-lived install. + // Remembering when one last finished keeps the six-hour cadence across + // restarts instead of rescanning a minute into every launch. + const compactedAtPath = path.join(path.dirname(config.dbPath), "compacted-at"); const compact = maintenance.compactEventStore.pipe( Effect.tap((result) => result.deletedEventCount > 0 || result.deletedReceiptCount > 0 ? Effect.logInfo("Compacted orchestration history", result) : Effect.void, ), + Effect.tap(() => + Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => fs.writeFileString(compactedAtPath, String(now))), + Effect.ignore, + ), + ), Effect.catch((cause) => Effect.logWarning("Orchestration history compaction failed", { cause }), ), ); + const lastCompactedAt = yield* fs.readFileString(compactedAtPath).pipe( + Effect.map(Number), + Effect.orElseSucceed(() => Number.NaN), + ); + const intervalMs = Duration.toMillis(COMPACTION_INTERVAL); + const untilDueMs = Number.isFinite(lastCompactedAt) + ? lastCompactedAt + intervalMs - (yield* Clock.currentTimeMillis) + : 0; + const firstCompactionDelay = Duration.millis( + Math.min(intervalMs, Math.max(Duration.toMillis(Duration.minutes(1)), untilDueMs)), + ); yield* forkParked( - Effect.sleep("1 minute").pipe( + Effect.sleep(firstCompactionDelay).pipe( Effect.andThen( - compact.pipe(Effect.andThen(publishStatus), Effect.repeat(Schedule.spaced("6 hours"))), + compact.pipe( + Effect.andThen(publishStatus), + Effect.repeat(Schedule.spaced(COMPACTION_INTERVAL)), + ), ), ), );