fix(server): unblock startup readiness and cut SQLite write amplification

apps/server/src/bin.test.ts

@@ -78,6 +78,7 @@ const makeCliTestServerConfig = (baseDir: string) =>
       traceBatchWindowMs: 200,
       traceMaxBytes: 10 * 1024 * 1024,
       traceMaxFiles: 10,
+      traceSqlSlowMs: 250,
       otlpTracesUrl: undefined,
       otlpMetricsUrl: undefined,
       otlpExportIntervalMs: 10_000,

apps/server/src/cli/config.test.ts

@@ -46,6 +46,7 @@ it.layer(NodeServices.layer)("cli config resolution", (it) => {
     traceBatchWindowMs: 1_000,
     traceMaxBytes: 10 * 1024 * 1024,
     traceMaxFiles: 10,
+    traceSqlSlowMs: 250,
     otlpTracesUrl: undefined,
     otlpMetricsUrl: undefined,
     otlpExportIntervalMs: 10_000,

apps/server/src/cli/config.ts

@@ -86,6 +86,9 @@ const EnvServerConfig = Config.all({
   traceMaxBytes: Config.int("T3CODE_TRACE_MAX_BYTES").pipe(Config.withDefault(10 * 1024 * 1024)),
   traceMaxFiles: Config.int("T3CODE_TRACE_MAX_FILES").pipe(Config.withDefault(10)),
   traceBatchWindowMs: Config.int("T3CODE_TRACE_BATCH_WINDOW_MS").pipe(Config.withDefault(1_000)),
+  // One span per SQL statement dominates trace volume. Keep the slow ones,
+  // which is how a pathological query is actually found, and drop the rest.
+  traceSqlSlowMs: Config.int("T3CODE_TRACE_SQL_SLOW_MS").pipe(Config.withDefault(250)),
   otlpTracesUrl: Config.string("T3CODE_OTLP_TRACES_URL").pipe(
     Config.option,
     Config.map(Option.getOrUndefined),
@@ -394,6 +397,7 @@ export const resolveServerConfig = (
       traceBatchWindowMs: env.traceBatchWindowMs,
       traceMaxBytes: env.traceMaxBytes,
       traceMaxFiles: env.traceMaxFiles,
+      traceSqlSlowMs: env.traceSqlSlowMs,
       otlpTracesUrl:
         env.otlpTracesUrl ??
         bootstrap?.otlpTracesUrl ??

apps/server/src/cli/pair.ts

@@ -330,6 +330,7 @@ const makePairServerConfig = Effect.fn(function* (input: {
     traceBatchWindowMs: 1_000,
     traceMaxBytes: 10 * 1024 * 1024,
     traceMaxFiles: 10,
+    traceSqlSlowMs: 250,
     otlpTracesUrl: undefined,
     otlpMetricsUrl: undefined,
     otlpExportIntervalMs: 10_000,

apps/server/src/config.ts

@@ -82,6 +82,11 @@ export class ServerConfig extends Context.Service<
     readonly traceBatchWindowMs: number;
     readonly traceMaxBytes: number;
     readonly traceMaxFiles: number;
+    /**
+     * Duration at or above which a `sql.execute` span is kept in the trace file.
+     * Faster successful statements are dropped at the sink; 0 keeps them all.
+     */
+    readonly traceSqlSlowMs: number;
     readonly otlpTracesUrl: string | undefined;
     readonly otlpMetricsUrl: string | undefined;
     readonly otlpExportIntervalMs: number;
@@ -196,6 +201,7 @@ const makeTest = Effect.fn("ServerConfig.makeTest")(function* (
     traceBatchWindowMs: 200,
     traceMaxBytes: 10 * 1024 * 1024,
     traceMaxFiles: 10,
+    traceSqlSlowMs: 250,
     otlpTracesUrl: undefined,
     otlpMetricsUrl: undefined,
     otlpExportIntervalMs: 10_000,

apps/server/src/environment/ServerEnvironment.test.ts

@@ -27,6 +27,7 @@ const makeServerConfig = Effect.fn(function* (baseDir: string) {
     traceBatchWindowMs: 200,
     traceMaxBytes: 10 * 1024 * 1024,
     traceMaxFiles: 10,
+    traceSqlSlowMs: 250,
     otlpTracesUrl: undefined,
     otlpMetricsUrl: undefined,
     otlpExportIntervalMs: 10_000,

apps/server/src/observability/Layers/Observability.ts

@@ -1,5 +1,11 @@
 import { httpHeaderRedactionLayer } from "@t3tools/shared/httpObservability";
-import { makeLocalFileTracer, makeTraceSink } from "@t3tools/shared/observability";
+// T3-CUSTOM(expbkt3): BEGIN - pull in the sink-level SQL span filter.
+import {
+  makeLocalFileTracer,
+  makeTraceSink,
+  retainSlowSqlSpans,
+} from "@t3tools/shared/observability";
+// T3-CUSTOM(expbkt3): END
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as References from "effect/References";
@@ -34,6 +40,9 @@ export const ObservabilityLive = Layer.unwrap(
           maxBytes: config.traceMaxBytes,
           maxFiles: config.traceMaxFiles,
           batchWindowMs: config.traceBatchWindowMs,
+          // T3-CUSTOM(expbkt3): BEGIN - one span per SQL statement dominated trace volume.
+          retain: retainSlowSqlSpans(config.traceSqlSlowMs),
+          // T3-CUSTOM(expbkt3): END
           onFlush: (stats) =>
             attribution.record({
               component: "server-trace",

apps/server/src/orchestration/ownershipBackfill.test.ts

@@ -4,7 +4,11 @@ import * as Effect from "effect/Effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
 
 import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
-import { backfillProjectionOwnership } from "./ownershipBackfill.ts";
+import {
+  ADMIN_REASSIGNMENT_MARKER,
+  backfillProjectionOwnership,
+  planOwnershipBackfill,
+} from "./ownershipBackfill.ts";
 
 const DEFAULT_OWNER = UserId.make("user_default_owner");
 const HISTORICAL_ASSIGNEE = UserId.make("user_historical_assignee");
@@ -199,3 +203,94 @@ it.layer(SqlitePersistenceMemory)("ownership backfill", (it) => {
     }),
   );
 });
+
+it.layer(SqlitePersistenceMemory)("ownership backfill planning", (it) => {
+  /** Each case owns the whole table set, so state never leaks between them. */
+  const seed = Effect.fn(function* (options: {
+    readonly ownerUserId: string | null;
+    readonly repairRecorded: boolean;
+  }) {
+    const sql = yield* SqlClient.SqlClient;
+    yield* sql`DELETE FROM projection_threads`;
+    yield* sql`DELETE FROM projection_projects`;
+    yield* sql`DELETE FROM maintenance_markers`;
+    yield* sql`
+      INSERT INTO projection_projects (
+        project_id, title, workspace_root, scripts_json, created_at, updated_at, owner_user_id
+      ) VALUES (
+        'project-plan', 'Planned project', '/tmp/plan', '[]', ${NOW}, ${NOW},
+        ${options.ownerUserId}
+      )
+    `;
+    yield* sql`
+      INSERT INTO projection_threads (
+        thread_id, project_id, title, created_at, updated_at, owner_user_id
+      ) VALUES (
+        'thread-plan', 'project-plan', 'Planned thread', ${NOW}, ${NOW}, ${options.ownerUserId}
+      )
+    `;
+    if (options.repairRecorded) {
+      yield* sql`
+        INSERT INTO maintenance_markers (marker, completed_at)
+        VALUES (${ADMIN_REASSIGNMENT_MARKER}, ${NOW})
+      `;
+    }
+  });
+
+  it.effect("skips every scan once the repair is recorded and no row is ownerless", () =>
+    Effect.gen(function* () {
+      yield* seed({ ownerUserId: DEFAULT_OWNER, repairRecorded: true });
+
+      assert.deepEqual(yield* planOwnershipBackfill(), {
+        skip: true,
+        repairAdminAssignments: false,
+      });
+    }),
+  );
+
+  it.effect("still converges ownerless rows after the repair is recorded", () =>
+    Effect.gen(function* () {
+      yield* seed({ ownerUserId: null, repairRecorded: true });
+
+      assert.deepEqual(yield* planOwnershipBackfill(), {
+        skip: false,
+        repairAdminAssignments: false,
+      });
+    }),
+  );
+
+  it.effect("runs the admin repair exactly while its marker is missing", () =>
+    Effect.gen(function* () {
+      yield* seed({ ownerUserId: DEFAULT_OWNER, repairRecorded: false });
+
+      assert.deepEqual(yield* planOwnershipBackfill(), {
+        skip: false,
+        repairAdminAssignments: true,
+      });
+    }),
+  );
+
+  it.effect("leaves administrator-owned rows alone when the repair is disabled", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      yield* seed({ ownerUserId: DEFAULT_OWNER, repairRecorded: true });
+      // A historical member that the repair pass would have promoted.
+      yield* sql`
+        INSERT INTO projection_thread_members (
+          thread_id, user_id, added_by_user_id, added_at
+        ) VALUES (
+          'thread-plan', ${HISTORICAL_ASSIGNEE}, NULL, ${NOW}
+        )
+      `;
+
+      yield* backfillProjectionOwnership(DEFAULT_OWNER, { repairAdminAssignments: false });
+
+      const threads = yield* sql<{ readonly ownerUserId: string | null }>`
+        SELECT owner_user_id AS "ownerUserId"
+        FROM projection_threads
+        WHERE thread_id = 'thread-plan'
+      `;
+      assert.deepEqual(threads, [{ ownerUserId: DEFAULT_OWNER }]);
+    }),
+  );
+});

apps/server/src/orchestration/ownershipBackfill.ts

@@ -1,26 +1,55 @@
 /**
  * ownershipBackfill - One-time legacy ownership assignment for team mode.
  *
- * Runs at startup (after migrations) only when Clerk team mode is configured.
- * Restores a creator from durable events or legacy membership first, then
- * assigns any remaining ownerless thread/project to the configured default or
- * earliest active administrator. Idempotent and fail-soft: if the fallback
- * owner cannot be resolved it logs a warning and converges on a later boot.
+ * Runs after startup (once migrations have applied) only when Clerk team mode
+ * is configured. Restores a creator from durable events or legacy membership
+ * first, then assigns any remaining ownerless thread/project to the configured
+ * default or earliest active administrator. Idempotent and fail-soft: if the
+ * fallback owner cannot be resolved it logs a warning and converges on a later
+ * boot.
+ *
+ * T3-CUSTOM(expbkt3): the admin-reassignment repair is a one-time migration
+ * concern, so it is gated on a durable marker rather than re-run every boot.
+ * Without that gate every admin-owned row is re-scanned forever, which cost
+ * bkt3.dev ~146 s of blocked readiness on each restart.
  *
  * @module ownershipBackfill
  */
 import type { UserId } from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
 
 import { ClerkDirectory } from "../auth/ClerkDirectory.ts";
 import { ServerConfig } from "../config.ts";
 
+/**
+ * Durable marker name for the one-time repair of rows bulk-assigned to the
+ * environment administrator before ownership was recorded on events.
+ */
+export const ADMIN_REASSIGNMENT_MARKER = "ownership.admin-reassignment";
+
+export interface BackfillProjectionOwnershipOptions {
+  /**
+   * When true the hot WHERE clauses also re-select rows already owned by the
+   * administrator, so a historical bulk assignment can be replaced by the real
+   * creator. Expensive, and only ever correct once per database.
+   */
+  readonly repairAdminAssignments?: boolean;
+}
+
 /** Assigns every collaborative projection a deterministic durable owner. */
 export const backfillProjectionOwnership = Effect.fn("backfillProjectionOwnership")(function* (
   adminUserId: UserId,
+  options: BackfillProjectionOwnershipOptions = {},
 ) {
   const sql = yield* SqlClient.SqlClient;
+  // Rows already owned by the admin are only revisited while the one-time
+  // repair is outstanding; afterwards ownerless rows are the entire work set.
+  const needsOwner =
+    options.repairAdminAssignments === false
+      ? sql`owner_user_id IS NULL`
+      : sql`(owner_user_id IS NULL OR owner_user_id = ${adminUserId})`;
 
   // Prefer the actor recorded by modern created events. This repairs a stale
   // projection without rewriting event history.
@@ -35,7 +64,7 @@ export const backfillProjectionOwnership = Effect.fn("backfillProjectionOwnershi
       ORDER BY created.stream_version ASC
       LIMIT 1
     )
-    WHERE (owner_user_id IS NULL OR owner_user_id = ${adminUserId})
+    WHERE ${needsOwner}
       AND NOT EXISTS (
         SELECT 1
         FROM orchestration_events AS transferred
@@ -61,7 +90,7 @@ export const backfillProjectionOwnership = Effect.fn("backfillProjectionOwnershi
       ORDER BY created.stream_version ASC
       LIMIT 1
     )
-    WHERE (owner_user_id IS NULL OR owner_user_id = ${adminUserId})
+    WHERE ${needsOwner}
       AND NOT EXISTS (
         SELECT 1
         FROM orchestration_events AS transferred
@@ -89,7 +118,7 @@ export const backfillProjectionOwnership = Effect.fn("backfillProjectionOwnershi
       ORDER BY member.added_at ASC, member.user_id ASC
       LIMIT 1
     )
-    WHERE (owner_user_id IS NULL OR owner_user_id = ${adminUserId})
+    WHERE ${needsOwner}
       AND NOT EXISTS (
         SELECT 1
         FROM orchestration_events AS transferred
@@ -113,7 +142,7 @@ export const backfillProjectionOwnership = Effect.fn("backfillProjectionOwnershi
       ORDER BY member.added_at ASC, member.user_id ASC
       LIMIT 1
     )
-    WHERE (owner_user_id IS NULL OR owner_user_id = ${adminUserId})
+    WHERE ${needsOwner}
       AND NOT EXISTS (
         SELECT 1
         FROM orchestration_events AS transferred
@@ -195,6 +224,60 @@ export const backfillProjectionOwnership = Effect.fn("backfillProjectionOwnershi
   };
 });
 
+/** True once the named one-time repair has been recorded as complete. */
+const hasMaintenanceMarker = Effect.fn("hasMaintenanceMarker")(function* (marker: string) {
+  const sql = yield* SqlClient.SqlClient;
+  const rows = yield* sql<{ readonly marker: string }>`
+    SELECT marker FROM maintenance_markers WHERE marker = ${marker} LIMIT 1
+  `;
+  return rows.length > 0;
+});
+
+const recordMaintenanceMarker = Effect.fn("recordMaintenanceMarker")(function* (marker: string) {
+  const sql = yield* SqlClient.SqlClient;
+  const completedAt = DateTime.formatIso(yield* DateTime.now);
+  yield* sql`
+    INSERT INTO maintenance_markers (marker, completed_at)
+    VALUES (${marker}, ${completedAt})
+    ON CONFLICT(marker) DO NOTHING
+  `;
+});
+
+export interface OwnershipBackfillPlan {
+  /** Nothing to converge: skip the event-log scans entirely. */
+  readonly skip: boolean;
+  /** Whether the one-time admin-reassignment repair is still outstanding. */
+  readonly repairAdminAssignments: boolean;
+}
+
+/**
+ * Decides, cheaply, whether a full backfill pass is worth running.
+ *
+ * Steady state is "every projection already has an owner and the one-time
+ * admin-reassignment repair is recorded". Both halves are two small projection
+ * scans, so the expensive correlated subqueries over the event log only run
+ * when there is actually something to converge.
+ */
+export const planOwnershipBackfill = Effect.fn("planOwnershipBackfill")(function* () {
+  const sql = yield* SqlClient.SqlClient;
+  const repairAdminAssignments = !(yield* hasMaintenanceMarker(ADMIN_REASSIGNMENT_MARKER));
+  const [ownerless] = yield* sql<{
+    readonly ownerlessThreads: number;
+    readonly ownerlessProjects: number;
+  }>`
+    SELECT
+      EXISTS (SELECT 1 FROM projection_threads WHERE owner_user_id IS NULL) AS "ownerlessThreads",
+      EXISTS (SELECT 1 FROM projection_projects WHERE owner_user_id IS NULL) AS "ownerlessProjects"
+  `;
+  const hasOwnerlessRows =
+    ownerless === undefined || ownerless.ownerlessThreads > 0 || ownerless.ownerlessProjects > 0;
+
+  return {
+    skip: !repairAdminAssignments && !hasOwnerlessRows,
+    repairAdminAssignments,
+  } satisfies OwnershipBackfillPlan;
+});
+
 export const runOwnershipBackfill = Effect.gen(function* () {
   const config = yield* ServerConfig;
   const clerkAuth = config.clerkAuth;
@@ -203,8 +286,15 @@ export const runOwnershipBackfill = Effect.gen(function* () {
     return;
   }
 
-  const clerkDirectory = yield* ClerkDirectory;
   const sql = yield* SqlClient.SqlClient;
+  const plan = yield* planOwnershipBackfill();
+
+  if (plan.skip) {
+    yield* Effect.logDebug("ownership backfill skipped: every projection already has an owner");
+    return;
+  }
+
+  const clerkDirectory = yield* ClerkDirectory;
   const explicitOwnerId =
     clerkAuth.defaultOwnerUserId !== undefined ? (clerkAuth.defaultOwnerUserId as UserId) : null;
   const configuredAdminUserId: UserId | null =
@@ -236,10 +326,19 @@ export const runOwnershipBackfill = Effect.gen(function* () {
     return;
   }
 
-  const result = yield* backfillProjectionOwnership(adminUserId);
+  const result = yield* backfillProjectionOwnership(adminUserId, {
+    repairAdminAssignments: plan.repairAdminAssignments,
+  });
+
+  // Only durable once the repair actually ran to completion; a failure leaves
+  // the marker absent so the next boot retries.
+  if (plan.repairAdminAssignments) {
+    yield* recordMaintenanceMarker(ADMIN_REASSIGNMENT_MARKER);
+  }
 
   yield* Effect.logInfo("ownership backfill complete", {
     adminUserId,
+    repairedAdminAssignments: plan.repairAdminAssignments,
     ...result,
   });
 }).pipe(