perf(server): compaction keeps its six-hour cadence across restarts

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

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<DatabaseStorageStatus | null>(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)),
+        ),
       ),
     ),
   );