fix(kb): advisory lock around scan-inbox to prevent race duplicates
When the n8n worker went down and recovered, multiple queued cron
executions fired scan-inbox in parallel. Each one saw the same file in
s3://insurance-kb/inbox/, each one called ingest_source — and the DB
ended up with duplicate source rows with identical checksums created in
the same second.
Guard scan-inbox with a session-level pg_try_advisory_lock (key "KBSC").
Concurrent callers return {skipped: true} immediately. The lock is
released in a finally block so a crash mid-scan still frees the next
scheduled run.
No schema change; no change to single-caller behavior.
Refs Task Master #2
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -10,6 +10,12 @@ from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
|
|||||||
|
|
||||||
from api.services.kb import ingest as kb_ingest
|
from api.services.kb import ingest as kb_ingest
|
||||||
from api.services.kb import s3 as kb_s3
|
from api.services.kb import s3 as kb_s3
|
||||||
|
from api.services.kb.db import get_pool
|
||||||
|
|
||||||
|
# Arbitrary non-zero constant used as a pg_try_advisory_lock key. Any int fits;
|
||||||
|
# this one is a stable hash of "kb-scan-inbox" so it can't collide with any
|
||||||
|
# other advisory lock chosen the same way for a different purpose.
|
||||||
|
_SCAN_LOCK_KEY = 0x4B42_5343 # "KBSC" in ASCII
|
||||||
|
|
||||||
logger = logging.getLogger("shira.admin.kb")
|
logger = logging.getLogger("shira.admin.kb")
|
||||||
|
|
||||||
@@ -140,7 +146,31 @@ async def requeue_failed(request: Request):
|
|||||||
|
|
||||||
@router.post("/scan-inbox")
|
@router.post("/scan-inbox")
|
||||||
async def scan_inbox(request: Request):
|
async def scan_inbox(request: Request):
|
||||||
|
"""Scan the inbox and ingest each pending file.
|
||||||
|
|
||||||
|
Protected by a session-level Postgres advisory lock so two concurrent
|
||||||
|
callers (e.g. overlapping n8n cron runs after a worker outage) can't both
|
||||||
|
process the same file and create duplicate sources. A caller that fails
|
||||||
|
to acquire the lock gets `skipped: lock held`.
|
||||||
|
"""
|
||||||
_verify_admin(request)
|
_verify_admin(request)
|
||||||
|
pool = await get_pool()
|
||||||
|
async with pool.acquire() as conn:
|
||||||
|
got_lock = await conn.fetchval(
|
||||||
|
"SELECT pg_try_advisory_lock($1)", _SCAN_LOCK_KEY
|
||||||
|
)
|
||||||
|
if not got_lock:
|
||||||
|
logger.info("[scan-inbox] skipped: another scan is in progress")
|
||||||
|
return {"processed": 0, "failed": 0, "skipped": True, "items": []}
|
||||||
|
try:
|
||||||
|
return await _do_scan()
|
||||||
|
finally:
|
||||||
|
await conn.execute(
|
||||||
|
"SELECT pg_advisory_unlock($1)", _SCAN_LOCK_KEY
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def _do_scan() -> dict:
|
||||||
items = kb_s3.list_inbox()
|
items = kb_s3.list_inbox()
|
||||||
results = []
|
results = []
|
||||||
for item in items:
|
for item in items:
|
||||||
|
|||||||
Reference in New Issue
Block a user