feat: enable sharding across DBs

[TEJASNARAYANS]

Summary

• Enable call-import data sharding across multiple Postgres databases (catalog + row shards) with scatter/gather reads, shard routing on write, rebalance tooling, and migration 054
• Bounded sharded pagination for evaluation row listing API (avoids loading all 70k+ pairs for page 1)
• Frontend eval detail page: adaptive polling for large runs, tab-gated row/aggregate queries, slower refetch while running
• Lighter eval materialization: ID/index-only source row lookups instead of full row loads
• Eval/import chain fix: do not permanently fail eval rows on transient telephony import retries; recover when import succeeds
• Exotel recording download: treat flaky HTTP 400 as transient per-row retry without global credential penalty (fixes throttle storms on parallel imports)

Why

Large call imports and evaluations (70k+ rows) were slow or unstable due to full-shard scans, aggressive UI polling, eval rows failing while imports retried, and Exotel 400 responses blocking all workers sharing a credential.

Test plan

• Run evaluation on a large sharded import; confirm page 1 loads quickly and progress polling is reasonable
• Start eval with rows needing recording fetch; confirm transient import errors do not permanently fail eval rows
• Import 300+ Exotel rows; confirm flaky 400s retry without credential-wide 60s blocks
• pytest tests/test_db_sharding/test_eval_rows_pagination.py tests/test_workers/test_eval_dispatch_import.py tests/test_services/test_telephony/test_recording_download.py tests/test_workers/test_process_call_import_row.py

Release Label

• major
• minor
• fix - backward-compatible bug fix / performance improvements

Checklist

• Tests added/updated for pagination, eval-import desync, and Exotel 400 handling
• Documentation updated where needed

[greptile-apps]

Greptile Summary

This PR introduces full cross-database sharding for call-import and evaluation row data (catalog + N row shards), bounded scatter/gather pagination for evaluation listings, adaptive UI polling, lighter eval materialization, and fixes for the eval-row/import-retry desync and Exotel 400 throttle-storm issues.

• Sharding infrastructure (db_sharding/): new pool manager, router, registry, row ops, scatter-gather, rebalance tooling, and migration 054 adding workspace_id and shard-slice tables.
• Deferred-commit pattern in bulk_ops.py and row_ops.py: shard sessions are flushed and held open, catalog is committed first, then shard sessions are committed — fixing the previously flagged shard-before-catalog ordering, but introducing a new partial-commit failure mode (see inline comment).
• Worker task fixes: locate_call_import_row and locate_call_import_evaluation_row are now wrapped in try/except LookupError across all three task files, addressing the previously flagged skip-sentinel gaps.

Confidence Score: 3/5

The deferred-commit pattern fixes one data-consistency ordering issue but introduces a new failure mode where a partial shard commit after a successful catalog commit leaves an import in an unrecoverable FAILED state, requiring manual DB intervention to restore.

The worker LookupError guards, rebalance ordering, workspace_id threading, and telephony 400/429 fixes are all solid. However, the deferred-commit path in execute_call_import_materialization and materialize_and_enqueue_evaluation has a residual gap: if commit_pending_shard_sessions fails midway, the catalog already holds total_rows > N and status = FAILED, both of which permanently block re-materialization without direct DB surgery.

app/services/call_imports/bulk_ops.py and app/db_sharding/row_ops.py, specifically the commit_pending_shard_sessions call site and the recovery guards in execute_call_import_materialization.

Important Files Changed

Filename	Overview
app/services/call_imports/bulk_ops.py	Heavily modified to route row operations through shard DB sessions. Deferred-commit ordering (catalog then shards) is architecturally sound, but partial shard commit failure leaves import in FAILED + total_rows>0 state that blocks re-materialization.
app/db_sharding/row_ops.py	New file implementing shard-aware row placement, locate, and bulk insert. Deferred-commit pattern addresses the previous shard-before-catalog ordering issue but introduces a new failure mode: partial shard commit after catalog commit leaves the import unrecoverable.
app/db_sharding/scatter_gather.py	Large new scatter-gather module; full in-memory load before pagination noted in previous review as P2. Core logic looks correct for the sharded routing.
app/db_sharding/rebalance.py	New rebalance tooling with correct ordering: copy to target then update catalog registry then commit catalog then delete from source. Addresses the previously flagged catalog-after-delete issue. Redis lock guards in-flight workers.
app/db_sharding/eval_rows.py	New scatter-gather helpers for evaluation rows. evaluation_row_session context manager cleanly wraps locate+cleanup. Bounded fetch for row_index sort only; non-row_index sort keys still load all rows per shard (noted in previous review).
app/workers/tasks/process_call_import_row.py	Shard-aware: locate_call_import_row now wrapped in try/except LookupError (addressing previous P1), catalog vs shard session split handled throughout. Rollup ordering (shard first then catalog) is correct.
app/workers/tasks/evaluate_call_import_row.py	Both first and second phase locate calls now wrapped in LookupError guards (addressing previous P1s). llm_credential_id threading added to scoring inputs. Session split for row_db vs catalog_db handled correctly.
app/workers/tasks/evaluate_call_import_row_audio.py	Switched from SessionLocal to evaluation_row_session context manager; LookupError is now caught at the outer level and returns the skip sentinel, addressing the previous P1.
app/services/telephony/recording_download.py	HTTP 429 on public-URL path now raises ExotelTransientError instead of ExotelInvalidContentError (fixing permanent failure). HTTP 400 on credentialed path now raises ExotelTransientError instead of penalizing credential (fixing throttle storms).
app/migrations/054_call_import_sharding.py	Adds workspace_id and sharding tables. workspace_id is now correctly threaded through bulk_insert_evaluation_rows (addressing previous P1 on NOT NULL constraint violation).

Comments Outside Diff (1)

1. app/services/call_imports/bulk_ops.py, line 340-345 (link)

   Partial shard commit after catalog commit leaves import unrecoverable

   After db.commit() succeeds (catalog now has total_rows = N and slice registry), commit_pending_shard_sessions iterates sessions and commits each one. If session k fails, an exception is raised and the except block calls db.rollback() (no-op — catalog already committed) then rollback_pending_shard_sessions(pending_shard_sessions). Sessions 1..k-1 are already durably committed on their shards; session k and beyond are rolled back. The catalog is then updated to status = FAILED with total_rows = N.

   Recovery is blocked on both guards: the status not in {PROCESSING, PENDING} check returns early, and the total_rows > 0 check also short-circuits. The stranded rows on shards 1..k-1 prevent simple re-insertion (unique constraint on (call_import_id, row_index)), so the import cannot be recovered without direct DB surgery (reset total_rows = 0 on catalog, drop partial shard rows, reset status, then re-upload). The same failure mode exists in materialize_and_enqueue_evaluation for eval rows.

_{Reviews (12): Last reviewed commit: "fix: fixing integration tests" | Re-trigger Greptile}