fix(db): keep cleanup-invoked helpers and snapshot reads on their role pools

Route markLargeValuesDeleted / pruneLargeValueMetadata (optional dbClient) and
chat-cleanup's file collection through the cleanup pool, and getSnapshot through
the exec pool, so the cleanup and inline-execution workloads stop borrowing the
process-wide pool for these queries.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NNy9Fzpfc1FdHAzYyb6Ycy

Author: TheodoreSpeaks

apps/sim/background/cleanup-logs.test.ts

@@ -260,7 +260,7 @@ describe('cleanup logs worker', () => {
       workspaceIds: ['workspace-1'],
     })
 
-    expect(mockMarkLargeValuesDeleted).toHaveBeenCalledWith([largeValueKey])
+    expect(mockMarkLargeValuesDeleted).toHaveBeenCalledWith([largeValueKey], expect.anything())
     expect(mockDeleteFileMetadata).toHaveBeenCalledTimes(2)
   })
 

apps/sim/background/cleanup-logs.ts

@@ -110,7 +110,7 @@ async function deleteLargeValueKeys(keys: string[]): Promise<{ deleted: number;
 
   if (deletedKeys.length > 0) {
     try {
-      await markLargeValuesDeleted(deletedKeys)
+      await markLargeValuesDeleted(deletedKeys, cleanupDb)
     } catch (error) {
       logger.error('Failed to mark large execution values as deleted:', { error })
       return { deleted: 0, failed: result.failed.length + deletedKeys.length }
@@ -356,7 +356,11 @@ async function cleanupLargeValueMetadata(workspaceIds: string[], label: string):
     const tombstonesDeletedBefore = new Date(
       Date.now() - LARGE_VALUE_TOMBSTONE_RETENTION_HOURS * 60 * 60 * 1000
     )
-    const result = await pruneLargeValueMetadata({ workspaceIds, tombstonesDeletedBefore })
+    const result = await pruneLargeValueMetadata({
+      workspaceIds,
+      tombstonesDeletedBefore,
+      dbClient: cleanupDb,
+    })
     logger.info(
       `[${label}/execution_large_value_metadata] Pruned ${result.referencesDeleted} stale references, ${result.dependenciesDeleted} dependencies, ${result.tombstonesDeleted} tombstones`
     )

apps/sim/lib/cleanup/chat-cleanup.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { dbFor } from '@sim/db'
 import { copilotMessages, workspaceFiles } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, inArray, isNull } from 'drizzle-orm'
@@ -10,6 +10,9 @@ import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
 
 const logger = createLogger('ChatCleanup')
 
+/** Chat cleanup only ever runs from cleanup jobs, so its reads use the cleanup pool. */
+const cleanupDb = dbFor('cleanup')
+
 const COPILOT_CLEANUP_BATCH_SIZE = 1000
 /** Bounds how many chats' `copilot_messages` rows are scanned per query. */
 const CHAT_FILE_COLLECT_CHUNK_SIZE = 500
@@ -41,7 +44,7 @@ export async function collectChatFiles(chatIds: string[]): Promise<FileRef[]> {
 
   for (const chunk of chunkArray(chatIds, CHAT_FILE_COLLECT_CHUNK_SIZE)) {
     const [linkedFiles, messageRows] = await Promise.all([
-      db
+      cleanupDb
         .select({ key: workspaceFiles.key, context: workspaceFiles.context })
         .from(workspaceFiles)
         .where(
@@ -53,7 +56,7 @@ export async function collectChatFiles(chatIds: string[]): Promise<FileRef[]> {
         ),
       // Scan every message row for the chat (no deleted_at filter): this is a
       // deletion path collecting blob keys, so attachments on any row count.
-      db
+      cleanupDb
         .select({ content: copilotMessages.content })
         .from(copilotMessages)
         .where(inArray(copilotMessages.chatId, chunk)),

apps/sim/lib/execution/payloads/large-value-metadata.ts

@@ -54,6 +54,8 @@ interface PruneLargeValueMetadataOptions {
   tombstonesDeletedBefore: Date
   batchSize?: number
   maxRowsPerTable?: number
+  /** Client the prune DELETEs run on. Defaults to the global pool; cleanup jobs pass `dbFor('cleanup')`. */
+  dbClient?: LargeValueMetadataClient
 }
 
 function parseLargeValueStorageKey(key: string): LargeValueStorageKeyParts | null {
@@ -368,19 +370,26 @@ export async function replaceLargeValueReferences(
   })
 }
 
-export async function markLargeValuesDeleted(keys: string[]): Promise<void> {
+export async function markLargeValuesDeleted(
+  keys: string[],
+  dbClient: LargeValueMetadataClient = db
+): Promise<void> {
   if (keys.length === 0) {
     return
   }
 
-  await db
+  await dbClient
     .update(executionLargeValues)
     .set({ deletedAt: new Date() })
     .where(inArray(executionLargeValues.key, keys))
 }
 
-async function pruneStaleReferences(workspaceIds: string[], batchSize: number): Promise<number> {
-  const rows = await db.execute<{ count: number }>(sql`
+async function pruneStaleReferences(
+  workspaceIds: string[],
+  batchSize: number,
+  dbClient: LargeValueMetadataClient
+): Promise<number> {
+  const rows = await dbClient.execute<{ count: number }>(sql`
     WITH deleted AS (
       DELETE FROM ${executionLargeValueReferences} AS ref
       WHERE ref.ctid IN (
@@ -418,9 +427,10 @@ async function pruneStaleReferences(workspaceIds: string[], batchSize: number):
 
 async function pruneDeletedParentDependencies(
   workspaceIds: string[],
-  batchSize: number
+  batchSize: number,
+  dbClient: LargeValueMetadataClient
 ): Promise<number> {
-  const rows = await db.execute<{ count: number }>(sql`
+  const rows = await dbClient.execute<{ count: number }>(sql`
     WITH deleted AS (
       DELETE FROM ${executionLargeValueDependencies} AS dependency
       WHERE dependency.ctid IN (
@@ -452,9 +462,10 @@ async function pruneDeletedParentDependencies(
 async function pruneDeletedLargeValueTombstones(
   workspaceIds: string[],
   deletedBefore: Date,
-  batchSize: number
+  batchSize: number,
+  dbClient: LargeValueMetadataClient
 ): Promise<number> {
-  const rows = await db.execute<{ count: number }>(sql`
+  const rows = await dbClient.execute<{ count: number }>(sql`
     WITH deleted AS (
       DELETE FROM ${executionLargeValues} AS value
       WHERE value.ctid IN (
@@ -482,6 +493,7 @@ export async function pruneLargeValueMetadata({
   tombstonesDeletedBefore,
   batchSize = LARGE_VALUE_METADATA_PRUNE_BATCH_SIZE,
   maxRowsPerTable = LARGE_VALUE_METADATA_PRUNE_MAX_ROWS_PER_TABLE,
+  dbClient = db,
 }: PruneLargeValueMetadataOptions): Promise<LargeValueMetadataPruneResult> {
   const result: LargeValueMetadataPruneResult = {
     referencesDeleted: 0,
@@ -498,15 +510,17 @@ export async function pruneLargeValueMetadata({
     if (referencesRemaining > 0) {
       result.referencesDeleted += await pruneStaleReferences(
         workspaceChunk,
-        Math.min(batchSize, referencesRemaining)
+        Math.min(batchSize, referencesRemaining),
+        dbClient
       )
     }
 
     const dependenciesRemaining = maxRowsPerTable - result.dependenciesDeleted
     if (dependenciesRemaining > 0) {
       result.dependenciesDeleted += await pruneDeletedParentDependencies(
         workspaceChunk,
-        Math.min(batchSize, dependenciesRemaining)
+        Math.min(batchSize, dependenciesRemaining),
+        dbClient
       )
     }
 
@@ -515,7 +529,8 @@ export async function pruneLargeValueMetadata({
       result.tombstonesDeleted += await pruneDeletedLargeValueTombstones(
         workspaceChunk,
         tombstonesDeletedBefore,
-        Math.min(batchSize, tombstonesRemaining)
+        Math.min(batchSize, tombstonesRemaining),
+        dbClient
       )
     }
 

apps/sim/lib/logs/execution/snapshot/service.ts

@@ -1,4 +1,4 @@
-import { db, dbFor } from '@sim/db'
+import { dbFor } from '@sim/db'
 import { workflowExecutionLogs, workflowExecutionSnapshots } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { sha256Hex } from '@sim/security/hash'
@@ -81,7 +81,7 @@ export class SnapshotService implements ISnapshotService {
   }
 
   async getSnapshot(id: string): Promise<WorkflowExecutionSnapshot | null> {
-    const [snapshot] = await db
+    const [snapshot] = await dbFor('exec')
       .select()
       .from(workflowExecutionSnapshots)
       .where(eq(workflowExecutionSnapshots.id, id))