fix(db): bound knowledge connector and document fan-out concurrency

[TheodoreSpeaks]

Summary

• Production Postgres sits behind small per-role PgBouncer pools (web pool = 45 server connections). Unbounded Promise.all/fire-and-forget fan-out over DB work can saturate a pool, and queued clients are killed after PgBouncer's 120s query_wait_timeout. This bounds three such fan-out sites found in the DB-availability audit.
• Connector sync scheduler (app/api/knowledge/connectors/sync/route.ts): the due-connectors SELECT now carries .orderBy(asc(nextSyncAt)).limit(200) (most-overdue first, so connectors beyond the cap are picked up next tick rather than starved — the ordering rides the existing kc_status_next_sync_idx index), and dispatches run through mapWithConcurrency at 10 concurrent instead of firing every connector at once. Per-connector dispatch failures are still caught and logged individually.
• In-process document dispatch (lib/knowledge/documents/service.ts dispatchInProcess): up to 1000 processDocumentAsync jobs (chunking + embedding + many DB inserts each) ran under an unbounded Promise.allSettled when trigger.dev is unavailable (self-hosted). Now bounded at 5 concurrent with the same per-item error isolation and dispatched-count semantics.
• KB-delete storage cleanup (deleteDocumentStorageFiles): the per-document storage-object + metadata-row delete fan-out was unbounded in N; now bounded at 10 concurrent. The mapper already swallows per-item errors, so per-item isolation is unchanged.

Type of Change

• Bug fix
• New feature
• Breaking change
• Documentation update

Testing

• bun run type-check in apps/sim — clean
• bunx vitest run lib/knowledge — 7 files, 97 tests passed
• bun run lint:check from repo root — clean
• bun run check:api-validation:strict — passed

Checklist

• My code follows the project's style guidelines
• I have performed a self-review of my code
• I have tested my changes
• I agree to the CLA

🤖 Generated with Claude Code

https://claude.ai/code/session_011Lmib23o5aQtBr7ZPhgpgf

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

1 Skipped Deployment

Project	Deployment	Actions	Updated (UTC)
docs	Skipped		Jul 6, 2026 3:21am

[TheodoreSpeaks]

@greptile review

[cursor]

PR Summary

Medium Risk
Changes throughput and scheduling for connector sync and heavy document paths, which can delay syncs or indexing under load but reduces pool saturation and query-wait timeouts.

Overview
Limits unbounded parallel work that could exhaust the shared Postgres/PgBouncer pool during knowledge cron and document lifecycle paths.

The connector sync cron now selects due connectors ordered by oldest nextSyncAt, caps dispatches at 200 per tick, and runs dispatchSync through mapWithConcurrency at 10 instead of firing every due connector at once. Per-connector failures are still logged individually.

When Trigger.dev is off, in-process document dispatch (dispatchInProcess) processes jobs at 5 concurrent instead of an unbounded Promise.allSettled; success counting and per-job error logging are preserved.

KB delete storage cleanup (deleteDocumentStorageFiles) deletes storage objects and metadata at 10 concurrent; kb/ ownership checks and per-item error handling are unchanged.

^{Reviewed by Cursor Bugbot for commit f94c7ba. Bugbot is set up for automated code reviews on this repo. Configure here.}

[greptile-apps]

Greptile Summary

This PR bounds three previously unbounded fan-out sites that were identified as potential PgBouncer pool exhaustion risks. The changes add a query limit and ordering to the connector sync scheduler, and replace Promise.allSettled/fire-and-forget patterns with mapWithConcurrency in the in-process document dispatch and KB-delete storage cleanup paths.

• Connector sync (route.ts): adds .orderBy(asc(nextSyncAt)).limit(200) to cap the per-tick query and switches from a fire-and-forget for loop to a properly awaited mapWithConcurrency at 10 concurrent — a meaningful behavioral change that also fixes a latent serverless teardown risk where background dispatches could be killed before completion.
• dispatchInProcess and deleteDocumentStorageFiles (service.ts): replaces Promise.allSettled with mapWithConcurrency at 5 and 10 concurrent respectively; per-item error isolation is preserved via try/catch wrappers that satisfy mapWithConcurrency's no-reject contract, and dispatch-count semantics are unchanged.

Confidence Score: 4/5

Safe to merge. All three fan-out sites are correctly bounded and per-item error isolation is maintained. The one notable change to review is the cron route shifting from fire-and-forget to a fully awaited dispatch loop.

The core refactoring is correct and well-scoped. The only rough edge is that the cron route's response count still reports connectors-found rather than connectors-dispatched-successfully — a small accuracy gap that is now fixable since the dispatches are awaited.

apps/sim/app/api/knowledge/connectors/sync/route.ts — the response count field could be made accurate with a small follow-up.

Important Files Changed

Filename	Overview
apps/sim/app/api/knowledge/connectors/sync/route.ts	Adds .orderBy(asc(nextSyncAt)).limit(200) to bound the connector query, then processes dispatches with mapWithConcurrency at 10 concurrent. Changes dispatch loop from fire-and-forget to fully awaited. Response count still reports dueConnectors.length (found), not the number successfully dispatched.
apps/sim/lib/knowledge/documents/service.ts	Replaces unbounded Promise.allSettled in dispatchInProcess and deleteDocumentStorageFiles with mapWithConcurrency at 5 and 10 concurrent respectively. Per-item error isolation is preserved via try/catch wrappers that satisfy mapWithConcurrency's no-reject contract. Logic is functionally equivalent to the removed code.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant Cron as Cron Scheduler
    participant Route as /api/knowledge/connectors/sync
    participant DB as Postgres (PgBouncer)
    participant Engine as dispatchSync (×N)

    Cron->>Route: GET (every 5 min)
    Route->>DB: UPDATE stale 'syncing' → 'error'
    Route->>DB: SELECT due connectors ORDER BY nextSyncAt ASC LIMIT 200
    DB-->>Route: dueConnectors[]

    loop "mapWithConcurrency (limit=10)"
        Route->>Engine: dispatchSync(connector.id)
        Engine->>DB: SELECT + conditional UPDATE
        DB-->>Engine: result
        Engine-->>Route: ok / catch(err)→log
    end

    Route-->>Cron: "{ success, count }"

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant Cron as Cron Scheduler
    participant Route as /api/knowledge/connectors/sync
    participant DB as Postgres (PgBouncer)
    participant Engine as dispatchSync (×N)

    Cron->>Route: GET (every 5 min)
    Route->>DB: UPDATE stale 'syncing' → 'error'
    Route->>DB: SELECT due connectors ORDER BY nextSyncAt ASC LIMIT 200
    DB-->>Route: dueConnectors[]

    loop "mapWithConcurrency (limit=10)"
        Route->>Engine: dispatchSync(connector.id)
        Engine->>DB: SELECT + conditional UPDATE
        DB-->>Engine: result
        Engine-->>Route: ok / catch(err)→log
    end

    Route-->>Cron: "{ success, count }"

Loading

Comments Outside Diff (1)

1. apps/sim/app/api/knowledge/connectors/sync/route.ts, line 97-107 (link)

   The response still reports count: dueConnectors.length (connectors found), but since dispatches are now fully awaited you can return the actual number that succeeded. Currently a connector that fails to dispatch is silently counted as dispatched in the response — callers monitoring this endpoint would see no signal of partial failure.

_{Reviews (1): Last reviewed commit: "fix(db): bound knowledge connector and d..." | Re-trigger Greptile}

[greptile-apps]

Greptile Summary

This PR bounds three previously-unbounded DB/storage fan-out sites to prevent PgBouncer connection pool saturation under load. All three call sites now use the existing mapWithConcurrency utility with appropriate concurrency caps.

• Connector sync scheduler (route.ts): adds .orderBy(asc(nextSyncAt)).limit(200) to the due-connectors query and replaces the fire-and-forget for loop with an await mapWithConcurrency(…, 10, …), which also converts dispatch from fire-and-forget to properly awaited.
• In-process document dispatch (service.ts): replaces an unbounded Promise.allSettled (up to 1 000 concurrent heavy jobs) with mapWithConcurrency at 5 concurrent, preserving per-item error isolation via try/catch.
• KB-delete storage cleanup (service.ts): replaces an unbounded Promise.allSettled over per-document object+metadata deletes with mapWithConcurrency at 10 concurrent; ownership/tenancy guards are unchanged.

Confidence Score: 5/5

Safe to merge. All three fan-out sites are correctly bounded, error isolation is preserved, and ownership/tenancy guards in storage deletion are structurally unchanged.

The changes are targeted and narrow: each of the three patched sites gets the same treatment (replace unbounded concurrent launch with mapWithConcurrency) and all three mappers correctly satisfy the fn-must-not-reject contract via .catch() or try/catch. The query cap in the scheduler correctly uses ORDER BY asc(nextSyncAt) to prefer the most-overdue connectors and avoid starvation. The behavioral shift from fire-and-forget to awaited dispatch in the cron route is an intentional improvement with no downside for this endpoint.

No files require special attention.

Important Files Changed

Filename	Overview
apps/sim/app/api/knowledge/connectors/sync/route.ts	Adds a 200-row query cap (most-overdue first via index-aligned ORDER BY) and converts the connector dispatch loop from unbounded fire-and-forget to an awaited mapWithConcurrency at concurrency 10. Mapper correctly uses .catch() to satisfy the mapWithConcurrency fn-must-not-reject contract.
apps/sim/lib/knowledge/documents/service.ts	Replaces two unbounded Promise.allSettled fan-outs (dispatchInProcess and deleteDocumentStorageFiles) with mapWithConcurrency at 5 and 10 concurrency respectively. Per-item error isolation is preserved via try/catch in both mappers; ownership/tenancy guards in storage deletion are structurally unchanged.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant Cron as Cron Trigger (5 min)
    participant Route as /api/knowledge/connectors/sync
    participant DB as Postgres / PgBouncer
    participant Engine as dispatchSync (sync-engine)

    Cron->>Route: GET (cron auth)
    Route->>DB: UPDATE stale syncing → error (recover locks)
    Route->>DB: SELECT id ORDER BY nextSyncAt ASC LIMIT 200
    DB-->>Route: dueConnectors[0..N≤200]

    loop mapWithConcurrency (≤10 at once)
        Route->>Engine: dispatchSync(connector.id)
        Engine->>DB: SELECT + conditional UPDATE (mark syncing)
        DB-->>Engine: ok
        Engine-->>Route: resolved (or .catch swallows error)
    end

    Route-->>Cron: "200 OK { count: N }"

    Note over Route,Engine: Previously: fire-and-forget for all N at once<br/>Now: await, bounded at 10 concurrent

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant Cron as Cron Trigger (5 min)
    participant Route as /api/knowledge/connectors/sync
    participant DB as Postgres / PgBouncer
    participant Engine as dispatchSync (sync-engine)

    Cron->>Route: GET (cron auth)
    Route->>DB: UPDATE stale syncing → error (recover locks)
    Route->>DB: SELECT id ORDER BY nextSyncAt ASC LIMIT 200
    DB-->>Route: dueConnectors[0..N≤200]

    loop mapWithConcurrency (≤10 at once)
        Route->>Engine: dispatchSync(connector.id)
        Engine->>DB: SELECT + conditional UPDATE (mark syncing)
        DB-->>Engine: ok
        Engine-->>Route: resolved (or .catch swallows error)
    end

    Route-->>Cron: "200 OK { count: N }"

    Note over Route,Engine: Previously: fire-and-forget for all N at once<br/>Now: await, bounded at 10 concurrent

Loading

_{Reviews (2): Last reviewed commit: "fix(db): bound knowledge connector and d..." | Re-trigger Greptile}