Heads-up (not a live bug, low priority) surfaced while reconciling the B→C contract in #1019.
application/utils/librarian/knowledge_source.py:DbKnowledgeSource._query() reads unconsumed rows with:
session.query(KnowledgeQueueRow)
.filter(KnowledgeQueueRow.consumed_at.is_(None),
KnowledgeQueueRow.llm_label.in_(READABLE_LABELS))
.order_by(KnowledgeQueueRow.created_at, KnowledgeQueueRow.id)
There's no with_for_update(skip_locked=True). consumed_at is stamped only after the batch is mapped and persisted, so between the read and that write there's a window where a second consumer could select the same consumed_at IS NULL rows and map them twice.
Why it's fine today: the orchestrator serialises A→B→C and runs a single C consumer per pipeline_run_id, so there's only ever one reader — no race.
When it would bite: running multiple concurrent C consumers over the same queue (e.g. to parallelise a large backlog).
Fix when/if that's needed (Postgres): add .with_for_update(skip_locked=True) to the query and hold the lock through mapping → persistence → the consumed_at update in a single transaction, so concurrent consumers claim disjoint batches. (SQLite doesn't support FOR UPDATE SKIP LOCKED and is single-consumer in dev/CI anyway.)
Filing so it's tracked before anyone scales C horizontally. Happy to open a PR if useful.
@PRAteek-singHWY I also want you to verify my claims if I am wrong somewhere
CC: @northdpole , @Pa04rth
Heads-up (not a live bug, low priority) surfaced while reconciling the B→C contract in #1019.
application/utils/librarian/knowledge_source.py:DbKnowledgeSource._query() reads unconsumed rows with:
session.query(KnowledgeQueueRow)
.filter(KnowledgeQueueRow.consumed_at.is_(None),
KnowledgeQueueRow.llm_label.in_(READABLE_LABELS))
.order_by(KnowledgeQueueRow.created_at, KnowledgeQueueRow.id)
There's no with_for_update(skip_locked=True). consumed_at is stamped only after the batch is mapped and persisted, so between the read and that write there's a window where a second consumer could select the same consumed_at IS NULL rows and map them twice.
Why it's fine today: the orchestrator serialises A→B→C and runs a single C consumer per pipeline_run_id, so there's only ever one reader — no race.
When it would bite: running multiple concurrent C consumers over the same queue (e.g. to parallelise a large backlog).
Fix when/if that's needed (Postgres): add .with_for_update(skip_locked=True) to the query and hold the lock through mapping → persistence → the consumed_at update in a single transaction, so concurrent consumers claim disjoint batches. (SQLite doesn't support FOR UPDATE SKIP LOCKED and is single-consumer in dev/CI anyway.)
Filing so it's tracked before anyone scales C horizontally. Happy to open a PR if useful.
@PRAteek-singHWY I also want you to verify my claims if I am wrong somewhere
CC: @northdpole , @Pa04rth