Task source fetch opens SQLite twice per task to answer the dedup ledger #7366

Summary

task_sources::pipeline asks the dedup ledger two questions per fetched task, and task_sources::store::with_connection opens a fresh SQLite connection and runs ensure_schema_initialized on every call. A fetch of N tasks therefore pays 2N connection opens and 2N schema checks to answer something one query covers.

Problem / Context

In run_source_once's ingest loop (integrations/task_sources/pipeline.rs), per task:

• store::is_ingested(config, source_id, external_id, hash) — "an entry exists for this (source, external_id) and its content_hash matches"
• store::was_ingested(config, source_id, external_id) — "an entry exists"

Both are single-row point queries against ingested_tasks WHERE source_id = ?1 AND external_id = ?2, and both go through store::with_connection, which does Connection::open(&db_path) followed by ensure_schema_initialized(&mut conn, &db_path) each time (store.rs). max_tasks_per_fetch defaults to 25 and is user-configurable, so this scales with the source.

Both answers are fully determined by one map, external_id -> content_hash, for that source: is_ingested is "the map's value equals this hash", was_ingested is "the map has the key". One SELECT external_id, content_hash FROM ingested_tasks WHERE source_id = ?1 replaces 2N queries.

The same file has a second instance: reconcile_missing_tasks calls store::remove_ingested per stale ledger row, so a prune of M rows is M more opens. Worth a separate look — a single DELETE ... WHERE external_id NOT IN (...) would cover it, but the loop's per-row pruned counting and debug logging would need handling.

The storage-backend path pays too, differently. With [storage] url set, store.rs delegates to store_documents.rs, where is_ingested / was_ingested each run ledger_entry(...) — a document fetch per call. Interestingly list_ingested_refs there already reads whole documents and keeps only one field, so a map-returning read costs that backend no extra round-trip at all.

One behavioural detail any fix has to preserve. Today the second occurrence of a repeated external_id within a single fetch sees the row mark_ingested wrote for the first occurrence, and is skipped as a duplicate. A prefetched map must be updated in memory after each mark_ingested or that task would be routed twice.

No measurement is claimed here: the cost is Connection::open plus a schema check per call, which is syscalls rather than CPU, so it does not disappear in a release build. I did not benchmark it.

Scope (optional)

In scope: one ledger read per fetch, with the two per-task checks decided in memory.

Not in scope: reconcile_missing_tasks' per-row remove_ingested (above), and with_connection's open-per-call shape itself — a pooled or cached connection would help every caller in this domain and in cron/flows, but that is a wider change than this one call site.

Acceptance criteria

• One read per fetch — the dedup ledger for a source is read once, not twice per task.
• Answers unchanged — the map decides is_ingested / was_ingested identically for every case the pipeline can meet: absent, present and unchanged, present and edited, and an entry belonging to a different source.
• Both backends — the SQLite store and the document store agree on that contract.
• In-fetch duplicates still dedup — a repeated external_id inside one fetch is not routed twice.
• Diff coverage ≥ 80% — the implementing PR meets the changed-lines coverage gate.

Related

Found on main at 775272ffe2 while looking for per-item I/O in loops. Related measurement, different domain: #7365.