From 778b271a366c383ba45024021777588e4f11f81c Mon Sep 17 00:00:00 2001 From: Hasan Date: Sat, 7 Nov 2026 12:12:06 +0500 Subject: [PATCH 1/2] fix(server): deleted threads release their stored history Deleting a thread only set deleted_at, so its runs, messages, turn items and other projection rows stayed in SQLite forever and `t3 doctor --fix` had nothing to reclaim. Background compaction now purges that content once a thread has been deleted for longer than the retained event history. The thread row stays as the tombstone. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0199EKpFPvxJDRDqCwnfb8Q1 --- .../ProjectionMaintenance.test.ts | 138 +++++++++++++++++- .../orchestration-v2/ProjectionMaintenance.ts | 89 ++++++++++- .../src/persistence/DatabaseStorage.test.ts | 7 +- .../server/src/persistence/DatabaseStorage.ts | 6 +- 4 files changed, 236 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts index 1f42953cc4ec3..5f4b6e9439049 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts @@ -6,6 +6,7 @@ import { type OrchestrationV2AppThread, type OrchestrationV2DomainEvent, type OrchestrationV2TurnItem, + MessageId, ProjectId, ProviderDriverKind, ProviderInstanceId, @@ -94,8 +95,8 @@ function itemEvent( spec: (typeof itemSpecs)[number], revision: number, eventId = `event:turn-item-compaction:${spec.key}:${revision}`, + id = TurnItemId.make(`turn-item:compaction:${spec.key}`), ): OrchestrationV2DomainEvent { - const id = TurnItemId.make(`turn-item:compaction:${spec.key}`); const growing = `${spec.key}:${revision}:` + "x".repeat((revision + 1) * 192); const completed = revision === 69; const base = { @@ -383,3 +384,138 @@ it.effect("keeps an anchor that a row appended during the pass borrows from", () }), ).pipe(Effect.provide(layerTest)), ); + +const countContent = (threadId: ThreadId) => + Effect.flatMap( + SqlClient.SqlClient, + (sql) => sql<{ + readonly items: number; + readonly positions: number; + readonly messages: number; + readonly found: number; + readonly threads: number; + }>` + SELECT + (SELECT COUNT(*) FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId}) AS items, + (SELECT COUNT(*) FROM orchestration_v2_turn_item_positions WHERE thread_id = ${threadId}) AS positions, + (SELECT COUNT(*) FROM orchestration_v2_projection_messages WHERE thread_id = ${threadId}) AS messages, + (SELECT COUNT(*) FROM orchestration_v2_message_search WHERE orchestration_v2_message_search MATCH ${`purgeable${threadId.replace(/\W/g, "")}`}) AS found, + (SELECT COUNT(*) FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId}) AS threads + `, + ).pipe(Effect.map((rows) => rows[0])); + +it.effect("purges content of threads deleted beyond the retained history", () => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const now = yield* DateTime.now; + const seed = Effect.fnUntraced(function* (name: string) { + const threadId = ThreadId.make(`thread:purge-${name}`); + const thread = makeThread(threadId, now); + yield* sink.write({ + events: [{ ...threadCreated(thread, now), id: EventId.make(`event:purge:${name}:thread`) }], + }); + yield* sink.write({ + events: [ + itemEvent( + threadId, + now, + itemSpecs[0], + 0, + `event:purge:${name}:item`, + TurnItemId.make(`turn-item:purge:${name}`), + ), + ], + }); + yield* sink.write({ + events: [ + { + id: EventId.make(`event:purge:${name}:message`), + type: "message.updated", + threadId, + occurredAt: now, + payload: { + id: MessageId.make(`message:purge:${name}`), + threadId, + runId: null, + nodeId: null, + createdBy: "agent", + creationSource: "provider", + role: "assistant", + text: `purgeable${threadId.replace(/\W/g, "")}`, + streaming: false, + attachments: [], + createdAt: now, + updatedAt: now, + }, + }, + ], + }); + return thread; + }); + const remove = (thread: OrchestrationV2AppThread, at: DateTime.Utc) => + sink.write({ + events: [ + { + id: EventId.make(`event:purge:${thread.id}:deleted`), + type: "thread.deleted", + threadId: thread.id, + occurredAt: at, + payload: { ...thread, deletedAt: at, updatedAt: at }, + }, + ], + }); + + const old = yield* seed("old"); + const live = yield* seed("live"); + yield* remove(old, now); + yield* TestClock.adjust("1 day"); + const recent = yield* seed("recent"); + const recentDeletedAt = yield* DateTime.now; + yield* remove(recent, recentDeletedAt); + const liveBefore = yield* projections.getThreadProjection(live.id); + const beforeOld = yield* countContent(old.id); + assert.isAbove(beforeOld?.items ?? 0, 0); + assert.isAbove(beforeOld?.positions ?? 0, 0); + assert.equal(beforeOld?.found, 1); + + const first = yield* maintenance.pruneEventStore(); + assert.isAbove(first.purgedProjectionRowCount, 0); + const afterOld = yield* countContent(old.id); + assert.deepEqual(afterOld, { items: 0, positions: 0, messages: 0, found: 0, threads: 1 }); + assert.equal((yield* countContent(recent.id))?.items, 1); + assert.equal((yield* countContent(recent.id))?.found, 1); + assert.deepEqual(yield* projections.getThreadProjection(live.id), liveBefore); + + assert.equal((yield* maintenance.pruneEventStore()).purgedProjectionRowCount, 0); + yield* TestClock.adjust("1 day"); + assert.isAbove((yield* maintenance.pruneEventStore()).purgedProjectionRowCount, 0); + assert.equal((yield* countContent(recent.id))?.items, 0); + assert.equal((yield* countContent(live.id))?.items, 1); + assert.equal((yield* countContent(live.id))?.found, 1); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("lists every thread-scoped projection table or names why it is kept", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const scoped = yield* sql<{ readonly name: string }>` + SELECT m.name FROM sqlite_master AS m, pragma_table_info(m.name) AS c + WHERE m.type = 'table' AND c.name = 'thread_id' + AND (m.name LIKE 'orchestration_v2_projection_%' OR m.name LIKE 'orchestration_v2_turn_item_%') + `; + const kept = [ + "orchestration_v2_projection_threads", + "orchestration_v2_projection_provider_sessions", + "orchestration_v2_projection_provider_session_bindings", + ]; + assert.deepEqual( + scoped + .map((row) => row.name) + .filter((name) => !kept.includes(name)) + .toSorted(), + [...ProjectionMaintenance.THREAD_CONTENT_TABLES].toSorted(), + ); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 210322aed686f..9cf18210b0e3d 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -22,11 +22,15 @@ export class ProjectionMaintenanceError extends Schema.TaggedError) => Effect.Effect< { readonly deletedEventCount: number; readonly deletedReceiptCount: number; + readonly purgedProjectionRowCount: number; readonly reclaimableBytes: number; }, ProjectionMaintenanceError @@ -44,6 +48,26 @@ const RETAINED_EVENT_HISTORY = Duration.days(1); const PAGE_SIZE = 50; const REFERENCE_SCAN_WINDOW = 1000; +// Thread-scoped projection tables, children first. The thread row stays as the tombstone. +// provider_sessions is handled separately because a session can serve several threads, and +// context_transfers is skipped because a live thread can be its other endpoint. +export const THREAD_CONTENT_TABLES = [ + "orchestration_v2_turn_item_positions", + "orchestration_v2_projection_turn_items", + "orchestration_v2_projection_nodes", + "orchestration_v2_projection_messages", + "orchestration_v2_projection_checkpoints", + "orchestration_v2_projection_checkpoint_scopes", + "orchestration_v2_projection_run_attempts", + "orchestration_v2_projection_runs", + "orchestration_v2_projection_provider_turns", + "orchestration_v2_projection_provider_threads", + "orchestration_v2_projection_subagents", + "orchestration_v2_projection_context_handoffs", + "orchestration_v2_projection_runtime_requests", + "orchestration_v2_projection_plans", +] as const; + type MetadataRow = { readonly projection_name: string; readonly last_sequence: number; @@ -252,6 +276,66 @@ export const layer: Layer.Layer, + ) { + const cutoff = DateTime.formatIso(DateTime.subtractDuration(now, RETAINED_EVENT_HISTORY)); + let purged = 0; + for (let cursor = ""; ;) { + const threads = yield* sql<{ readonly thread_id: string }>` + SELECT thread_id FROM orchestration_v2_projection_threads + WHERE deleted_at IS NOT NULL AND deleted_at <= ${cutoff} AND thread_id > ${cursor} + ORDER BY thread_id ASC + LIMIT ${PAGE_SIZE} + `.pipe(NodeSqliteClient.readOnly); + const threadIds = threads.map((row) => row.thread_id); + // Never hold a transaction or the reader lock across a yield or a rest. + yield* Effect.yieldNow; + if (threadIds.length > 0) { + purged += yield* sql.withTransaction( + Effect.gen(function* () { + let rows = 0; + const countChanges = Effect.gen(function* () { + const changes = yield* sql<{ readonly count: number }>`SELECT changes() AS count`; + rows += changes[0]?.count ?? 0; + }); + for (const table of THREAD_CONTENT_TABLES) { + yield* sql` + DELETE FROM ${sql(table)} + WHERE thread_id IN ${sql.in(threadIds)} + `; + yield* countChanges; + } + yield* sql` + DELETE FROM orchestration_v2_projection_provider_sessions AS session + WHERE session.thread_id IN ${sql.in(threadIds)} + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_provider_session_bindings AS binding + WHERE binding.provider_session_id = session.provider_session_id + AND binding.thread_id NOT IN ${sql.in(threadIds)} + ) + `; + yield* countChanges; + yield* sql` + DELETE FROM orchestration_v2_projection_provider_session_bindings + WHERE thread_id IN ${sql.in(threadIds)} + `; + yield* countChanges; + return rows; + }), + ); + yield* Effect.yieldNow; + } + yield* betweenPages; + if (threadIds.length < PAGE_SIZE) break; + cursor = threadIds.at(-1) ?? cursor; + } + return purged; + }); + // A pass prunes through the mark the previous pass left once that mark is a day old, so // at least RETAINED_EVENT_HISTORY of events always remains. Receipts stay for idempotency. const pruneEventStore = Effect.fn("ProjectionMaintenance.pruneEventStore")( @@ -298,6 +382,8 @@ export const layer: Layer.Layer`PRAGMA freelist_count`; const pageSizeRows = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`; const reclaimableBytes = @@ -306,6 +392,7 @@ export const layer: Layer.Layer Effect.gen(function* () { diff --git a/apps/server/src/persistence/DatabaseStorage.ts b/apps/server/src/persistence/DatabaseStorage.ts index a0dd307237d46..ea652cc92cd6f 100644 --- a/apps/server/src/persistence/DatabaseStorage.ts +++ b/apps/server/src/persistence/DatabaseStorage.ts @@ -110,7 +110,11 @@ const make = Effect.gen(function* () { const result = yield* maintenance.pruneEventStore(betweenPages); const completedAt = yield* Clock.currentTimeMillis; yield* fs.writeFileString(compactedAtPath, String(completedAt)).pipe(Effect.ignore); - if (result.deletedEventCount > 0 || result.deletedReceiptCount > 0) { + if ( + result.deletedEventCount > 0 || + result.deletedReceiptCount > 0 || + result.purgedProjectionRowCount > 0 + ) { yield* Effect.logInfo("Compacted orchestration history", result); } }).pipe( From 533559fa9da6e2027dd2ee90c04980f76a791ba2 Mon Sep 17 00:00:00 2001 From: Hasan Date: Sat, 7 Nov 2026 12:24:42 +0500 Subject: [PATCH 2/2] fix(server): purge deleted threads in bounded batches and keep forked sources Deleted threads that a surviving run-fork inherits history from, directly or through a chain of forks, keep their content. Purging deletes at most 500 rows per statement and yields between batches, so one very large thread no longer holds the writer. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0199EKpFPvxJDRDqCwnfb8Q1 --- .../ProjectionMaintenance.test.ts | 179 ++++++++++++------ .../orchestration-v2/ProjectionMaintenance.ts | 98 ++++++---- 2 files changed, 186 insertions(+), 91 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts index 5f4b6e9439049..ce3906baa917d 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts @@ -8,6 +8,7 @@ import { type OrchestrationV2TurnItem, MessageId, ProjectId, + RunId, ProviderDriverKind, ProviderInstanceId, ThreadId, @@ -404,69 +405,76 @@ const countContent = (threadId: ThreadId) => `, ).pipe(Effect.map((rows) => rows[0])); -it.effect("purges content of threads deleted beyond the retained history", () => - Effect.gen(function* () { - const sink = yield* EventSink.EventSinkV2; - const projections = yield* ProjectionStore.ProjectionStoreV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; - const now = yield* DateTime.now; - const seed = Effect.fnUntraced(function* (name: string) { - const threadId = ThreadId.make(`thread:purge-${name}`); - const thread = makeThread(threadId, now); - yield* sink.write({ - events: [{ ...threadCreated(thread, now), id: EventId.make(`event:purge:${name}:thread`) }], - }); - yield* sink.write({ - events: [ - itemEvent( - threadId, - now, - itemSpecs[0], - 0, - `event:purge:${name}:item`, - TurnItemId.make(`turn-item:purge:${name}`), - ), - ], - }); - yield* sink.write({ - events: [ - { - id: EventId.make(`event:purge:${name}:message`), - type: "message.updated", +const purgeFixture = Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const now = yield* DateTime.now; + const seed = Effect.fnUntraced(function* ( + name: string, + forkedFrom: OrchestrationV2AppThread["forkedFrom"] = null, + ) { + const threadId = ThreadId.make(`thread:purge-${name}`); + const thread = { ...makeThread(threadId, now), forkedFrom }; + yield* sink.write({ + events: [{ ...threadCreated(thread, now), id: EventId.make(`event:purge:${name}:thread`) }], + }); + yield* sink.write({ + events: [ + itemEvent( + threadId, + now, + itemSpecs[0], + 0, + `event:purge:${name}:item`, + TurnItemId.make(`turn-item:purge:${name}`), + ), + ], + }); + yield* sink.write({ + events: [ + { + id: EventId.make(`event:purge:${name}:message`), + type: "message.updated", + threadId, + occurredAt: now, + payload: { + id: MessageId.make(`message:purge:${name}`), threadId, - occurredAt: now, - payload: { - id: MessageId.make(`message:purge:${name}`), - threadId, - runId: null, - nodeId: null, - createdBy: "agent", - creationSource: "provider", - role: "assistant", - text: `purgeable${threadId.replace(/\W/g, "")}`, - streaming: false, - attachments: [], - createdAt: now, - updatedAt: now, - }, + runId: null, + nodeId: null, + createdBy: "agent", + creationSource: "provider", + role: "assistant", + text: `purgeable${threadId.replace(/\W/g, "")}`, + streaming: false, + attachments: [], + createdAt: now, + updatedAt: now, }, - ], - }); - return thread; + }, + ], }); - const remove = (thread: OrchestrationV2AppThread, at: DateTime.Utc) => - sink.write({ - events: [ - { - id: EventId.make(`event:purge:${thread.id}:deleted`), - type: "thread.deleted", - threadId: thread.id, - occurredAt: at, - payload: { ...thread, deletedAt: at, updatedAt: at }, - }, - ], - }); + return thread; + }); + const remove = (thread: OrchestrationV2AppThread, at: DateTime.Utc) => + sink.write({ + events: [ + { + id: EventId.make(`event:purge:${thread.id}:deleted`), + type: "thread.deleted", + threadId: thread.id, + occurredAt: at, + payload: { ...thread, deletedAt: at, updatedAt: at }, + }, + ], + }); + return { projections, maintenance, now, seed, remove }; +}); +it.effect("purges content of threads deleted beyond the retained history", () => + Effect.gen(function* () { + const { projections, maintenance, now, seed, remove } = yield* purgeFixture; const old = yield* seed("old"); const live = yield* seed("live"); yield* remove(old, now); @@ -519,3 +527,56 @@ it.effect("lists every thread-scoped projection table or names why it is kept", ); }).pipe(Effect.provide(layerTest)), ); + +it.effect("keeps a deleted thread that a live run-fork still inherits history from", () => + Effect.gen(function* () { + const { maintenance, seed, remove } = yield* purgeFixture; + const root = yield* seed("root"); + const middle = yield* seed("middle", { + type: "run", + threadId: root.id, + runId: RunId.make("run:purge:root"), + }); + const fork = yield* seed("fork", { + type: "run", + threadId: middle.id, + runId: RunId.make("run:purge:middle"), + }); + const unrelated = yield* seed("unrelated"); + yield* remove(root, yield* DateTime.now); + yield* remove(middle, yield* DateTime.now); + yield* remove(unrelated, yield* DateTime.now); + yield* TestClock.adjust("2 days"); + + yield* maintenance.pruneEventStore(); + assert.equal((yield* countContent(root.id))?.items, 1); + assert.equal((yield* countContent(middle.id))?.items, 1); + assert.equal((yield* countContent(fork.id))?.items, 1); + assert.equal((yield* countContent(unrelated.id))?.items, 0); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("purges a very large thread in bounded batches", () => + Effect.gen(function* () { + const { maintenance, seed, remove } = yield* purgeFixture; + const sql = yield* SqlClient.SqlClient; + const big = yield* seed("big"); + yield* sql` + INSERT INTO orchestration_v2_turn_item_positions (thread_id, turn_item_id, ordinal) + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 1200) + SELECT ${big.id}, 'turn-item:purge:filler:' || i, i + 100 FROM n + `; + yield* remove(big, yield* DateTime.now); + yield* TestClock.adjust("2 days"); + + let batches = 0; + const result = yield* maintenance.pruneEventStore( + Effect.sync(() => { + batches += 1; + }), + ); + assert.isAtLeast(result.purgedProjectionRowCount, 1200); + assert.isAtLeast(batches, 3); + assert.equal((yield* countContent(big.id))?.positions, 0); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 9cf18210b0e3d..150cf4e60e70b 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -5,6 +5,7 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import type * as SqlError from "effect/sql/SqlError"; import * as SqlClient from "effect/sql/SqlClient"; import { @@ -46,6 +47,8 @@ export class ProjectionMaintenanceV2 extends Context.Service< const RETAINED_EVENT_HISTORY = Duration.days(1); // Small pages bound work on the reader worker and synchronous fallback. const PAGE_SIZE = 50; +// Bounds the rows one purge statement deletes, FTS trigger work included. +const ROW_BATCH_SIZE = 500; const REFERENCE_SCAN_WINDOW = 1000; // Thread-scoped projection tables, children first. The thread row stays as the tombstone. @@ -277,59 +280,90 @@ export const layer: Layer.Layer, ) { const cutoff = DateTime.formatIso(DateTime.subtractDuration(now, RETAINED_EVENT_HISTORY)); let purged = 0; + // Each batch is one statement that commits on its own, so no transaction spans a yield. + const deleteBatches = Effect.fnUntraced(function* ( + batch: (limit: number) => Effect.Effect, + ) { + for (;;) { + yield* batch(ROW_BATCH_SIZE); + const changes = yield* sql<{ readonly count: number }>`SELECT changes() AS count`; + const count = changes[0]?.count ?? 0; + purged += count; + yield* Effect.yieldNow; + yield* betweenPages; + if (count < ROW_BATCH_SIZE) return; + } + }); for (let cursor = ""; ;) { const threads = yield* sql<{ readonly thread_id: string }>` + WITH RECURSIVE inherited(thread_id) AS ( + SELECT json_extract(fork.payload_json, '$.forkedFrom.threadId') + FROM orchestration_v2_projection_threads AS fork + WHERE json_extract(fork.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(fork.payload_json, '$.lineage.sideChat') IS NOT 1 + AND (fork.deleted_at IS NULL OR fork.deleted_at > ${cutoff}) + UNION + SELECT json_extract(fork.payload_json, '$.forkedFrom.threadId') + FROM inherited + JOIN orchestration_v2_projection_threads AS fork ON fork.thread_id = inherited.thread_id + WHERE json_extract(fork.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(fork.payload_json, '$.lineage.sideChat') IS NOT 1 + ) SELECT thread_id FROM orchestration_v2_projection_threads WHERE deleted_at IS NOT NULL AND deleted_at <= ${cutoff} AND thread_id > ${cursor} + AND thread_id NOT IN (SELECT thread_id FROM inherited WHERE thread_id IS NOT NULL) ORDER BY thread_id ASC LIMIT ${PAGE_SIZE} `.pipe(NodeSqliteClient.readOnly); const threadIds = threads.map((row) => row.thread_id); - // Never hold a transaction or the reader lock across a yield or a rest. yield* Effect.yieldNow; if (threadIds.length > 0) { - purged += yield* sql.withTransaction( - Effect.gen(function* () { - let rows = 0; - const countChanges = Effect.gen(function* () { - const changes = yield* sql<{ readonly count: number }>`SELECT changes() AS count`; - rows += changes[0]?.count ?? 0; - }); - for (const table of THREAD_CONTENT_TABLES) { - yield* sql` - DELETE FROM ${sql(table)} + for (const table of THREAD_CONTENT_TABLES) { + yield* deleteBatches( + (limit) => sql` + DELETE FROM ${sql(table)} + WHERE rowid IN ( + SELECT rowid FROM ${sql(table)} WHERE thread_id IN ${sql.in(threadIds)} - `; - yield* countChanges; - } - yield* sql` - DELETE FROM orchestration_v2_projection_provider_sessions AS session - WHERE session.thread_id IN ${sql.in(threadIds)} - AND NOT EXISTS ( - SELECT 1 FROM orchestration_v2_projection_provider_session_bindings AS binding - WHERE binding.provider_session_id = session.provider_session_id - AND binding.thread_id NOT IN ${sql.in(threadIds)} + LIMIT ${limit} ) - `; - yield* countChanges; - yield* sql` - DELETE FROM orchestration_v2_projection_provider_session_bindings + `, + ); + } + yield* deleteBatches( + (limit) => sql` + DELETE FROM orchestration_v2_projection_provider_sessions + WHERE rowid IN ( + SELECT session.rowid FROM orchestration_v2_projection_provider_sessions AS session + WHERE session.thread_id IN ${sql.in(threadIds)} + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_provider_session_bindings AS binding + WHERE binding.provider_session_id = session.provider_session_id + AND binding.thread_id NOT IN ${sql.in(threadIds)} + ) + LIMIT ${limit} + ) + `, + ); + yield* deleteBatches( + (limit) => sql` + DELETE FROM orchestration_v2_projection_provider_session_bindings + WHERE rowid IN ( + SELECT rowid FROM orchestration_v2_projection_provider_session_bindings WHERE thread_id IN ${sql.in(threadIds)} - `; - yield* countChanges; - return rows; - }), + LIMIT ${limit} + ) + `, ); - yield* Effect.yieldNow; } - yield* betweenPages; if (threadIds.length < PAGE_SIZE) break; cursor = threadIds.at(-1) ?? cursor; }