Open History to see the last commit
1import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
2import * as Context from "effect/Context";
3import * as DateTime from "effect/DateTime";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Schema from "effect/Schema";
8import * as SqlClient from "effect/sql/SqlClient";
9
10import {
11 EVENT_LOG_PRUNE_MARK,
12 EVENT_LOG_PRUNED_THROUGH,
13} from "../persistence/OrchestrationEventStore.ts";
14import * as ProjectionStore from "./ProjectionStore.ts";
15
16export class ProjectionMaintenanceError extends Schema.TaggedError<ProjectionMaintenanceError>()(
17 "ProjectionMaintenanceError",
18 {
19 operation: Schema.String,
20 cause: Schema.optional(Schema.Defect()),
21 },
22) {}
23
24export interface ProjectionMaintenanceV2Shape {
25 /** Deletes events the projections already cover once they are older than the retained history. */
26 readonly pruneEventStore: (betweenPages?: Effect.Effect<void>) => Effect.Effect<
27 {
28 readonly deletedEventCount: number;
29 readonly deletedReceiptCount: number;
30 readonly reclaimableBytes: number;
31 },
32 ProjectionMaintenanceError
33 >;
34}
35
36export class ProjectionMaintenanceV2 extends Context.Service<
37 ProjectionMaintenanceV2,
38 ProjectionMaintenanceV2Shape
39>()("t3/orchestration-v2/ProjectionMaintenance/ProjectionMaintenanceV2") {}
40
41// Reconnect replay and command-id retries read this tail; older cursors resync from a snapshot.
42const RETAINED_EVENT_HISTORY = Duration.days(1);
43// Small pages bound work on the reader worker and synchronous fallback.
44const PAGE_SIZE = 50;
45const REFERENCE_SCAN_WINDOW = 1000;
46
47type MetadataRow = {
48 readonly projection_name: string;
49 readonly last_sequence: number;
50 readonly updated_at: string;
51};
52
53export const layer: Layer.Layer<ProjectionMaintenanceV2, never, SqlClient.SqlClient> = Layer.effect(
54 ProjectionMaintenanceV2,
55 Effect.gen(function* () {
56 const sql = yield* SqlClient.SqlClient;
57
58 const writeMetadata = (name: string, sequence: number, at: DateTime.Utc) => sql`
59 INSERT INTO orchestration_v2_projection_metadata (
60 projection_name, schema_version, last_sequence, updated_at
61 )
62 VALUES (
63 ${name},
64 ${ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION},
65 ${sequence},
66 ${DateTime.formatIso(at)}
67 )
68 ON CONFLICT(projection_name) DO UPDATE SET
69 last_sequence = excluded.last_sequence,
70 updated_at = excluded.updated_at
71 `;
72
73 // Collects anchors whose prompt later rows borrow through promptSequence.
74 const readPromptReferences = Effect.fnUntraced(function* (
75 afterSequence: number,
76 throughSequence: number,
77 betweenPages: Effect.Effect<void>,
78 ) {
79 const referenced = new Set<number>();
80 for (let cursor = afterSequence; cursor < throughSequence;) {
81 const windowEnd = Math.min(cursor + REFERENCE_SCAN_WINDOW, throughSequence);
82 const rows = yield* sql<{ readonly prompt_sequence: number }>`
83 SELECT json_extract(payload_json, '$.promptSequence') AS prompt_sequence
84 FROM orchestration_events
85 WHERE sequence > ${cursor}
86 AND sequence <= ${windowEnd}
87 AND event_type IN ('subagent.updated', 'turn-item.updated')
88 AND json_type(payload_json, '$.promptSequence') = 'integer'
89 `.pipe(NodeSqliteClient.readOnly);
90 for (const row of rows) referenced.add(row.prompt_sequence);
91 cursor = windowEnd;
92 yield* betweenPages;
93 }
94 return referenced;
95 });
96
97 const pruneEvents = Effect.fnUntraced(function* (
98 throughSequence: number,
99 head: number,
100 betweenPages: Effect.Effect<void>,
101 ) {
102 const referenced = yield* readPromptReferences(throughSequence, head, betweenPages);
103 // Stream heads at or below the boundary are kept, so the anchors they borrow must stay too.
104 const keptHeadReferences = yield* sql<{ readonly prompt_sequence: number }>`
105 SELECT json_extract(event.payload_json, '$.promptSequence') AS prompt_sequence
106 FROM (
107 SELECT MAX(sequence) AS sequence FROM orchestration_events
108 GROUP BY aggregate_kind, stream_id
109 ) AS heads
110 JOIN orchestration_events AS event ON event.sequence = heads.sequence
111 WHERE heads.sequence <= ${throughSequence}
112 AND event.event_type IN ('subagent.updated', 'turn-item.updated')
113 AND json_type(event.payload_json, '$.promptSequence') = 'integer'
114 `.pipe(NodeSqliteClient.readOnly);
115 for (const row of keptHeadReferences) referenced.add(row.prompt_sequence);
116 yield* betweenPages;
117 let referencesThrough = head;
118 let deleted = 0;
119 let cursor = 0;
120 while (cursor < throughSequence) {
121 const rows = yield* sql<{ readonly sequence: number; readonly prunable: number }>`
122 SELECT
123 event.sequence,
124 CASE
125 WHEN event.application_event_version = 1 THEN 0
126 -- The legacy importer treats a thread without thread.created as not yet imported.
127 WHEN event.event_type = 'thread.created' THEN 0
128 -- Codex usage history reads the selection a run attempt started with.
129 WHEN event.event_type = 'run.updated'
130 AND json_extract(event.payload_json, '$.status') = 'running' THEN 0
131 -- The newest row keeps each stream's high water, stream_version, and snapshot sequence.
132 WHEN event.sequence = (
133 SELECT MAX(newer.sequence) FROM orchestration_events AS newer
134 WHERE newer.aggregate_kind = event.aggregate_kind
135 AND newer.stream_id = event.stream_id
136 ) THEN 0
137 ELSE 1
138 END AS prunable
139 FROM orchestration_events AS event
140 WHERE event.sequence > ${cursor}
141 AND event.sequence <= ${throughSequence}
142 ORDER BY event.sequence ASC
143 LIMIT ${PAGE_SIZE}
144 `.pipe(NodeSqliteClient.readOnly);
145 const candidates = rows.filter((row) => row.prunable === 1).map((row) => row.sequence);
146 // Never hold a transaction or the reader lock across a yield or a rest.
147 yield* Effect.yieldNow;
148 if (candidates.length > 0) {
149 deleted += yield* sql.withTransaction(
150 Effect.gen(function* () {
151 // Rows appended since the reference scan may borrow a prompt from a candidate.
152 const appended = yield* sql<{
153 readonly sequence: number;
154 readonly prompt_sequence: number | null;
155 }>`
156 SELECT
157 sequence,
158 CASE
159 WHEN event_type IN ('subagent.updated', 'turn-item.updated')
160 AND json_type(payload_json, '$.promptSequence') = 'integer'
161 THEN json_extract(payload_json, '$.promptSequence')
162 END AS prompt_sequence
163 FROM orchestration_events
164 WHERE sequence > ${referencesThrough}
165 `;
166 for (const row of appended) {
167 if (row.prompt_sequence !== null) referenced.add(row.prompt_sequence);
168 referencesThrough = Math.max(referencesThrough, row.sequence);
169 }
170 const obsolete = candidates.filter((sequence) => !referenced.has(sequence));
171 if (obsolete.length === 0) return 0;
172 yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`;
173 return obsolete.length;
174 }),
175 );
176 yield* Effect.yieldNow;
177 }
178 // Rest after every page, including the last, so the next loop starts a fresh slice.
179 yield* betweenPages;
180 if (rows.length < PAGE_SIZE) break;
181 cursor = rows.at(-1)?.sequence ?? throughSequence;
182 }
183 return deleted;
184 });
185
186 // Imported legacy threads are served from v2 rows; their v1 events and receipts have no reader.
187 const pruneImportedLegacyHistory = Effect.fnUntraced(function* (
188 throughSequence: number,
189 throughReceiptRowId: number,
190 betweenPages: Effect.Effect<void>,
191 ) {
192 let deletedEventCount = 0;
193 let deletedReceiptCount = 0;
194 for (let cursor = 0; ;) {
195 const rows = yield* sql<{ readonly sequence: number; readonly imported: number }>`
196 SELECT
197 event.sequence,
198 event.aggregate_kind = 'thread' AND EXISTS (
199 SELECT 1 FROM orchestration_v2_legacy_imports AS legacy_import
200 WHERE legacy_import.thread_id = event.stream_id
201 AND legacy_import.transcript_imported_at IS NOT NULL
202 ) AS imported
203 FROM orchestration_events AS event
204 INDEXED BY idx_orchestration_events_application_sequence
205 WHERE event.application_event_version = 1
206 AND event.sequence > ${cursor}
207 AND event.sequence <= ${throughSequence}
208 ORDER BY event.sequence ASC
209 LIMIT ${PAGE_SIZE}
210 `.pipe(NodeSqliteClient.readOnly);
211 const obsolete = rows.filter((row) => row.imported === 1).map((row) => row.sequence);
212 yield* Effect.yieldNow;
213 if (obsolete.length > 0) {
214 yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`;
215 deletedEventCount += obsolete.length;
216 yield* Effect.yieldNow;
217 }
218 yield* betweenPages;
219 if (rows.length < PAGE_SIZE) break;
220 cursor = rows.at(-1)?.sequence ?? cursor;
221 }
222 for (let cursor = 0; ;) {
223 const rows = yield* sql<{
224 readonly row_id: number;
225 readonly command_id: string;
226 readonly imported: number;
227 }>`
228 SELECT
229 receipt.rowid AS row_id,
230 receipt.command_id,
231 receipt.command_type = 'legacy' AND receipt.aggregate_kind = 'thread' AND EXISTS (
232 SELECT 1 FROM orchestration_v2_legacy_imports AS legacy_import
233 WHERE legacy_import.thread_id = receipt.aggregate_id
234 AND legacy_import.transcript_imported_at IS NOT NULL
235 ) AS imported
236 FROM orchestration_command_receipts AS receipt
237 WHERE receipt.rowid > ${cursor} AND receipt.rowid <= ${throughReceiptRowId}
238 ORDER BY receipt.rowid ASC
239 LIMIT ${PAGE_SIZE}
240 `.pipe(NodeSqliteClient.readOnly);
241 const obsolete = rows.filter((row) => row.imported === 1).map((row) => row.command_id);
242 yield* Effect.yieldNow;
243 if (obsolete.length > 0) {
244 yield* sql`DELETE FROM orchestration_command_receipts WHERE command_id IN ${sql.in(obsolete)}`;
245 deletedReceiptCount += obsolete.length;
246 yield* Effect.yieldNow;
247 }
248 yield* betweenPages;
249 if (rows.length < PAGE_SIZE) break;
250 cursor = rows.at(-1)?.row_id ?? cursor;
251 }
252 return { deletedEventCount, deletedReceiptCount };
253 });
254
255 // A pass prunes through the mark the previous pass left once that mark is a day old, so
256 // at least RETAINED_EVENT_HISTORY of events always remains. Receipts stay for idempotency.
257 const pruneEventStore = Effect.fn("ProjectionMaintenance.pruneEventStore")(
258 function* (betweenPages: Effect.Effect<void> = Effect.void) {
259 const now = yield* DateTime.now;
260 const metadata = yield* sql<MetadataRow>`
261 SELECT projection_name, last_sequence, updated_at
262 FROM orchestration_v2_projection_metadata
263 WHERE projection_name IN (
264 'thread-projections', ${EVENT_LOG_PRUNE_MARK}, ${EVENT_LOG_PRUNED_THROUGH}
265 )
266 `.pipe(NodeSqliteClient.readOnly);
267 const read = (name: string) => metadata.find((row) => row.projection_name === name);
268 const projected = read("thread-projections")?.last_sequence ?? 0;
269 const mark = read(EVENT_LOG_PRUNE_MARK);
270 const prunedThrough = read(EVENT_LOG_PRUNED_THROUGH)?.last_sequence ?? 0;
271 // Fixed bounds: foreground commits during the pass wait for the next one.
272 const bounds = yield* sql<{ readonly sequence: number; readonly receipt_row_id: number }>`
273 SELECT
274 COALESCE((SELECT MAX(sequence) FROM orchestration_events), 0) AS sequence,
275 COALESCE((SELECT MAX(rowid) FROM orchestration_command_receipts), 0) AS receipt_row_id
276 `.pipe(NodeSqliteClient.readOnly);
277 const head = bounds[0]?.sequence ?? 0;
278
279 let deletedEventCount = 0;
280 const markIsDue =
281 mark !== undefined &&
282 DateTime.toEpochMillis(now) - Date.parse(mark.updated_at) >=
283 Duration.toMillis(RETAINED_EVENT_HISTORY);
284 if (mark !== undefined && markIsDue) {
285 const throughSequence = Math.max(Math.min(mark.last_sequence, projected), prunedThrough);
286 if (throughSequence > prunedThrough) {
287 // Publish the boundary first so resuming clients take a snapshot, never a partial replay.
288 yield* writeMetadata(EVENT_LOG_PRUNED_THROUGH, throughSequence, now);
289 }
290 deletedEventCount += yield* pruneEvents(throughSequence, head, betweenPages);
291 }
292 if (mark === undefined || markIsDue) yield* writeMetadata(EVENT_LOG_PRUNE_MARK, head, now);
293
294 const legacy = yield* pruneImportedLegacyHistory(
295 head,
296 bounds[0]?.receipt_row_id ?? 0,
297 betweenPages,
298 );
299 deletedEventCount += legacy.deletedEventCount;
300
301 const freelistRows = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`;
302 const pageSizeRows = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`;
303 const reclaimableBytes =
304 (freelistRows[0]?.freelist_count ?? 0) * (pageSizeRows[0]?.page_size ?? 0);
305
306 return {
307 deletedEventCount,
308 deletedReceiptCount: legacy.deletedReceiptCount,
309 reclaimableBytes,
310 };
311 },
312 Effect.mapError(
313 (cause) => new ProjectionMaintenanceError({ operation: "prune event store", cause }),
314 ),
315 );
316
317 return ProjectionMaintenanceV2.of({ pruneEventStore });
318 }),
319);