diff --git a/.claude/skills/bulk-data-submit/SKILL.md b/.claude/skills/bulk-data-submit/SKILL.md index 911e336e1..3b62489d5 100644 --- a/.claude/skills/bulk-data-submit/SKILL.md +++ b/.claude/skills/bulk-data-submit/SKILL.md @@ -62,6 +62,7 @@ status-only kick-off (no `manifestUrl`) they have nothing to attach to and are i | `HFS_BULK_SUBMIT_S3_BUCKET` | none | S3 bucket, required when output backend is s3 | | `HFS_BULK_SUBMIT_REQUIRES_ACCESS_TOKEN` | `auto` | Manifest posture; false is invalid with local-fs | | `HFS_BULK_SUBMIT_WORKER_CONCURRENCY` | `2` | In-process submit worker count | +| `HFS_BULK_SUBMIT_FILE_CONCURRENCY` | `1` | Files of one manifest ingested at once (fan-out); clamped to `2` on SQLite | | `HFS_BULK_SUBMIT_DISABLE_LOCAL_WORKER` | `false` | Disable in-pod workers | | `HFS_BULK_SUBMIT_MAX_CONCURRENT_PER_TENANT` | `4` | Per-tenant active submission cap; returns `429` | | `HFS_BULK_SUBMIT_BATCH_SIZE` | `1000` | Ingestion batch size | @@ -108,4 +109,6 @@ The backend capability splits into `BulkSubmitIngest` (the synchronous `BulkSubm identically on each page. Fetch pages from the status URL with `?page=N` (1-based) — out of range is `404`, malformed is `400`. Page size `0` disables pagination and yields one manifest with an empty `link`. - Status-poll pacing: the `202` advertises `HFS_BULK_SUBMIT_RETRY_AFTER`, and a client that polls past `HFS_BULK_SUBMIT_POLL_RATE_LIMIT` within the window gets `429` plus a `Retry-After` pointing at the end of that window. Buckets are keyed by poll token plus principal, falling back to peer address; the check runs before any job-store work, so throttled polls stay cheap. +- File fan-out is backend-aware. `HFS_BULK_SUBMIT_FILE_CONCURRENCY` is honoured as configured on the concurrent-writer backends (PostgreSQL, MongoDB, S3), but clamped to `2` on SQLite, which logs the clamp at startup. SQLite serialises writers, so a higher fan-out queues each batch's writes behind one exclusive lock until they outlast `busy_timeout` and abort the manifest outright — the clamp keeps a high configured value slow rather than fatal. Raising fan-out past 2 requires PostgreSQL. +- The manifest bookkeeping writes (counts, progress, byte progress) retry with bounded exponential backoff when SQLite reports the database busy or locked, instead of failing the ingest. Every other error still surfaces on the first attempt. - Cleanup periodically removes status artifacts for submissions whose `updated_at` exceeds `HFS_BULK_SUBMIT_OUTPUT_TTL`. diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index 7e14674a5..ab4f19a89 100644 --- a/crates/hfs/src/main.rs +++ b/crates/hfs/src/main.rs @@ -1734,11 +1734,31 @@ async fn build_bulk_submit( .with_decryption_keys(decryption_keys), ); + // SQLite serialises writers, so a high fan-out queues batch writes behind + // one lock until they outlast `busy_timeout` and abort the manifest (#942). + // Clamp there, and say so, rather than letting the operator's value fail + // the import. + let backend_kind = config + .storage_backend_mode() + .map(|mode| mode.primary_backend_kind()) + .unwrap_or(BackendKind::Sqlite); + let file_concurrency = cfg.effective_file_concurrency(backend_kind); + if file_concurrency < cfg.file_concurrency.max(1) { + info!( + configured = cfg.file_concurrency, + effective = file_concurrency, + "Bulk submit fan-out clamped: SQLite serialises writers, and a \ + higher fan-out queues batch writes past busy_timeout and aborts \ + the import. Use PostgreSQL for a higher file concurrency." + ); + } + spawn_submit_workers( jobs.clone(), fetcher.clone(), output.clone(), &cfg, + file_concurrency, reindex_hook, ); @@ -1762,6 +1782,7 @@ fn spawn_submit_workers( fetcher: Arc, output: Arc, cfg: &helios_rest::config::BulkSubmitConfig, + file_concurrency: u32, reindex_hook: Option>, ) { if cfg.disable_local_worker { @@ -1773,7 +1794,7 @@ fn spawn_submit_workers( if defer_indexing { info!("Bulk submit fast-load: search indexing deferred to post-manifest reindex"); } - let file_concurrency = cfg.file_concurrency.max(1) as usize; + let file_concurrency = file_concurrency.max(1) as usize; if file_concurrency > 1 { info!( file_concurrency, diff --git a/crates/hfs/tests/bulk_submit/run_bulk_submit_fanout_check.sh b/crates/hfs/tests/bulk_submit/run_bulk_submit_fanout_check.sh new file mode 100755 index 000000000..748793efd --- /dev/null +++ b/crates/hfs/tests/bulk_submit/run_bulk_submit_fanout_check.sh @@ -0,0 +1,166 @@ +#!/usr/bin/env bash +# +# Manual check of the $bulk-submit file fan-out on SQLite: that the clamp to +# SQLITE_MAX_FILE_CONCURRENCY is applied and announced in the log, and that a +# full ingest finishes cleanly. +# +# Origin: issue #942, "import aborts with 'database is locked' at high file +# concurrency with the full search-parameter registry". +# +# It starts two things and then stays up: +# +# 1. A static "Data Provider" (python -m http.server) serving a Bulk Export +# Manifest and one .ndjson with 2 Patients. +# 2. HFS with HFS_BULK_SUBMIT_FILE_CONCURRENCY=8, which is the value that +# triggers the clamp introduced by the fix. +# +# It kills no processes and frees no ports: it picks free ports >18000 and runs +# both servers under `timeout`, so they shut themselves down when the TTL runs +# out. +# +# Requirements: cargo, curl and python on PATH. On Windows the interpreter must +# be invoked as `python`: `python3` resolves to the Microsoft Store stub, which +# prints a notice and exits without starting anything. +# +# crates/hfs/tests/bulk_submit/run_bulk_submit_fanout_check.sh +# TTL=1800 crates/hfs/tests/bulk_submit/run_bulk_submit_fanout_check.sh +# FILE_CONCURRENCY=1 crates/hfs/tests/bulk_submit/run_bulk_submit_fanout_check.sh +# SKIP_BUILD=1 crates/hfs/tests/bulk_submit/run_bulk_submit_fanout_check.sh +# +set -euo pipefail + +cd "$(dirname "$0")/../../../.." + +TTL="${TTL:-600}" +FILE_CONCURRENCY="${FILE_CONCURRENCY:-8}" +WORKDIR="${WORKDIR:-/tmp/hfs-bulk-submit-fanout}" +# A file, not `:memory:`. An in-memory database cannot turn WAL on +# (backend.rs:557 tries, but it stays on journal_mode=memory), and without WAL +# the locks surface as table-level SQLITE_LOCKED: a degraded case that does not +# represent the real deployment the issue fixes. +DB_URL="${DB_URL:-target/hfs-bulk-submit-fanout.db}" +rm -f "$DB_URL" "$DB_URL-wal" "$DB_URL-shm" + +# --- free ports >18000 --------------------------------------------------- +# This does a real bind instead of reading `netstat`: its output is not portable +# (on Linux the state is printed as LISTEN, not LISTENING, and often the binary +# is not even installed), and when the pattern does not match every port would +# look free, so the failure would only show up much later, at server startup. +# SO_REUSEADDR mirrors what both the HFS listener and http.server do, so a +# socket in TIME_WAIT does not count as taken. +port_is_free() { + python - "$1" <<'PY' +import socket, sys +s = socket.socket() +s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +try: + s.bind(("127.0.0.1", int(sys.argv[1]))) +except OSError: + sys.exit(1) +finally: + s.close() +PY +} + +pick_port() { + local p="$1" + while ! port_is_free "$p"; do p=$((p + 1)); done + echo "$p" +} + +PROVIDER_PORT="$(pick_port "${PROVIDER_PORT:-19100}")" +HFS_PORT="$(pick_port "${HFS_PORT:-18790}")" +[ "$HFS_PORT" = "$PROVIDER_PORT" ] && HFS_PORT="$(pick_port $((HFS_PORT + 1)))" + +PROVIDER_URL="http://127.0.0.1:$PROVIDER_PORT" +HFS_URL="http://127.0.0.1:$HFS_PORT" + +# --- provider fixture ---------------------------------------------------- +rm -rf "$WORKDIR" +mkdir -p "$WORKDIR" + +cat > "$WORKDIR/patients.ndjson" <<'EOF' +{"resourceType":"Patient","id":"submit-smoke-1","name":[{"family":"Submit","given":["Alpha"]}],"gender":"female"} +{"resourceType":"Patient","id":"submit-smoke-2","name":[{"family":"Submit","given":["Beta"]}],"gender":"male"} +EOF + +cat > "$WORKDIR/manifest.json" < SKIP_BUILD=1: reusing the already built binary" +else + echo "==> building hfs (debug)" + case "$(uname -s)" in + MINGW* | MSYS* | CYGWIN*) + cargo rustc -p helios-hfs --bin hfs -- -C link-arg=/STACK:33554432 + ;; + *) + cargo build -p helios-hfs + ;; + esac +fi + +HFS_BIN="target/debug/hfs" +[ -x "$HFS_BIN" ] || HFS_BIN="target/debug/hfs.exe" + +# --- startup ------------------------------------------------------------- +# `python -u` because when stdout is redirected the banner stays in the buffer. +echo "==> provider on $PROVIDER_URL (TTL ${TTL}s)" +( cd "$WORKDIR" && timeout "$TTL" python -u -m http.server "$PROVIDER_PORT" --bind 127.0.0.1 ) \ + > "$WORKDIR/provider.log" 2>&1 & + +echo "==> HFS on $HFS_URL (HFS_BULK_SUBMIT_FILE_CONCURRENCY=$FILE_CONCURRENCY, TTL ${TTL}s)" +# HFS_BASE_URL is mandatory: it defaults to http://localhost:8080 and it is the +# origin HFS uses to build the `content-location` polling header. Without it the +# poll goes to whatever runs on 8080, not to this instance. +HFS_BASE_URL="$HFS_URL" \ +HFS_BULK_SUBMIT_ENABLED=true \ +HFS_BULK_SUBMIT_FILE_CONCURRENCY="$FILE_CONCURRENCY" \ +HFS_LOG_LEVEL=info \ + timeout "$TTL" "$HFS_BIN" \ + --database-url "$DB_URL" --log-level info \ + --host 127.0.0.1 --port "$HFS_PORT" \ + > "$WORKDIR/hfs.log" 2>&1 & + +# --- wait for HFS to answer ---------------------------------------------- +for _ in $(seq 1 60); do + if curl -sS -o /dev/null "$HFS_URL/health" 2>/dev/null; then break; fi + sleep 1 +done + +if ! curl -sS -o /dev/null -w '' "$HFS_URL/health" 2>/dev/null; then + echo "HFS did not answer on $HFS_URL/health" >&2 + tail -40 "$WORKDIR/hfs.log" >&2 + exit 1 +fi + +cat < configured` arm of +# `effective_file_concurrency` (crates/rest/src/config.rs:737). +# +# Requires a PostgreSQL listening on PG_PORT. Start one with: +# docker run -d --name hfs-bulk-submit-pg \ +# -e POSTGRES_USER=helios -e POSTGRES_PASSWORD=helios -e POSTGRES_DB=helios \ +# -p 127.0.0.1:18432:5432 postgres:16-alpine +# +# The binary must be built with the `postgres` feature, which is NOT in the +# helios-hfs defaults (crates/hfs/Cargo.toml:17): +# cargo build -p helios-hfs --features helios-hfs/postgres +# On Windows the debug binary also overflows the main thread stack while +# building the SearchParameter registry, so it has to be relinked: +# cargo rustc -p helios-hfs --bin hfs --features helios-hfs/postgres \ +# -- -C link-arg=/STACK:33554432 +# +# Requirements: docker (or your own PostgreSQL), cargo and python on PATH. +# +set -euo pipefail + +cd "$(dirname "$0")/../../../.." + +PG_PORT="${PG_PORT:-18432}" +TTL="${TTL:-75}" +FILE_CONCURRENCY="${FILE_CONCURRENCY:-8}" +LOG="${LOG:-/tmp/hfs-bulk-submit-postgres.log}" + +# Same real bind as the other scripts: `netstat` is not portable. +port_is_free() { + python - "$1" <<'PY' +import socket, sys +s = socket.socket() +s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +try: + s.bind(("127.0.0.1", int(sys.argv[1]))) +except OSError: + sys.exit(1) +finally: + s.close() +PY +} +HFS_PORT="${HFS_PORT:-18795}" +while ! port_is_free "$HFS_PORT"; do HFS_PORT=$((HFS_PORT + 1)); done + +HFS_BIN="target/debug/hfs" +[ -x "$HFS_BIN" ] || HFS_BIN="target/debug/hfs.exe" + +echo "==> HFS on PostgreSQL at 127.0.0.1:$HFS_PORT (requested fan-out: $FILE_CONCURRENCY)" + +HFS_BASE_URL="http://127.0.0.1:$HFS_PORT" \ +HFS_STORAGE_BACKEND=postgres \ +HFS_DATABASE_URL="postgres://helios:helios@127.0.0.1:$PG_PORT/helios" \ +HFS_BULK_SUBMIT_ENABLED=true \ +HFS_BULK_SUBMIT_FILE_CONCURRENCY="$FILE_CONCURRENCY" \ +HFS_LOG_LEVEL=info \ + timeout "$TTL" "$HFS_BIN" --log-level info --host 127.0.0.1 --port "$HFS_PORT" \ + > "$LOG" 2>&1 || true + +echo +echo "=== selected backend ===" +grep -o 'storage_backend=[a-z]*' "$LOG" | head -1 || echo "(not found)" + +echo +echo "=== bulk submit lines ===" +grep 'Bulk submit' "$LOG" || echo "(none)" + +echo +if grep -q 'fan-out clamped' "$LOG"; then + echo "RESULT: FAIL - the clamp was applied with PostgreSQL" + exit 1 +fi +if grep -q "file_concurrency=$FILE_CONCURRENCY" "$LOG"; then + echo "RESULT: OK - no clamp, effective fan-out = $FILE_CONCURRENCY" +else + echo "RESULT: INCONCLUSIVE - file_concurrency=$FILE_CONCURRENCY was not seen" + exit 1 +fi diff --git a/crates/hfs/tests/bulk_submit/run_bulk_submit_volume_check.sh b/crates/hfs/tests/bulk_submit/run_bulk_submit_volume_check.sh new file mode 100755 index 000000000..31e65452f --- /dev/null +++ b/crates/hfs/tests/bulk_submit/run_bulk_submit_volume_check.sh @@ -0,0 +1,211 @@ +#!/usr/bin/env bash +# +# $bulk-submit ingest at realistic volume on SQLite. +# +# Origin: issue #942. The smoke fixture (1 file, 2 Patients) does not exercise +# the fan-out, so it cannot tell whether the clamp and the retries hold up. +# +# This generates a manifest with FILES files of PER_FILE Patients each, ingests +# all of it through $bulk-submit and checks that: +# +# - the import finishes (poll -> 200) with no "database is locked" +# - the output manifest declares the FILES files and the counts +# - the resources can be read back with GET /Patient/{id} +# - the bookkeeping retries that happened are counted (grep on the log) +# +# All of it on SQLite, which is where the clamp to 2 applies. +# +# Requirements: cargo, curl and python on PATH. On Windows the interpreter is +# invoked as `python`; `python3` resolves to the Microsoft Store stub. +# +# crates/hfs/tests/bulk_submit/run_bulk_submit_volume_check.sh +# FILES=24 PER_FILE=1000 TTL=1500 MAX_POLLS=90 \ +# crates/hfs/tests/bulk_submit/run_bulk_submit_volume_check.sh +# +set -euo pipefail + +cd "$(dirname "$0")/../../../.." + +FILES="${FILES:-12}" +PER_FILE="${PER_FILE:-500}" +TTL="${TTL:-600}" +FILE_CONCURRENCY="${FILE_CONCURRENCY:-8}" +WORKDIR="${WORKDIR:-/tmp/hfs-bulk-submit-volume}" +# `:memory:` CANNOT turn WAL on (backend.rs:557 applies it, but an in-memory +# database stays on journal_mode=memory), so the locks surface as table-level +# SQLITE_LOCKED: that is the worst case, not the realistic deployment. With a +# file there is WAL. A file is the default; DB_URL=':memory:' forces the +# degraded case. +# The path lives under target/ and not under WORKDIR because in Git Bash WORKDIR +# is an MSYS path (/tmp/...) that the native Windows binary cannot resolve. +DB_URL="${DB_URL:-target/hfs-bulk-submit-volume.db}" +rm -f "$DB_URL" "$DB_URL-wal" "$DB_URL-shm" + +# This does a real bind instead of reading `netstat`: its output is not portable +# (on Linux the state is printed as LISTEN, not LISTENING, and often the binary +# is not even installed), and when the pattern does not match every port would +# look free, so the failure would only show up much later, at server startup. +port_is_free() { + python - "$1" <<'PY' +import socket, sys +s = socket.socket() +s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +try: + s.bind(("127.0.0.1", int(sys.argv[1]))) +except OSError: + sys.exit(1) +finally: + s.close() +PY +} +pick_port() { + local p="$1" + while ! port_is_free "$p"; do p=$((p + 1)); done + echo "$p" +} + +PROVIDER_PORT="$(pick_port "${PROVIDER_PORT:-19200}")" +HFS_PORT="$(pick_port "${HFS_PORT:-18810}")" +[ "$HFS_PORT" = "$PROVIDER_PORT" ] && HFS_PORT="$(pick_port $((HFS_PORT + 1)))" +PROVIDER_URL="http://127.0.0.1:$PROVIDER_PORT" +HFS_URL="http://127.0.0.1:$HFS_PORT" + +TOTAL=$((FILES * PER_FILE)) + +rm -rf "$WORKDIR" +mkdir -p "$WORKDIR" + +echo "==> generating $FILES files x $PER_FILE Patients = $TOTAL resources" +python - "$WORKDIR" "$FILES" "$PER_FILE" "$PROVIDER_URL" <<'PY' +import json, sys, pathlib +workdir, files, per_file, provider = sys.argv[1], int(sys.argv[2]), int(sys.argv[3]), sys.argv[4] +out = [] +for f in range(files): + name = f"patients-{f}.ndjson" + with open(pathlib.Path(workdir) / name, "w", encoding="utf-8") as fh: + for i in range(per_file): + pid = f"vol942-{f}-{i}" + fh.write(json.dumps({ + "resourceType": "Patient", + "id": pid, + "name": [{"family": "Volume", "given": [f"F{f}", f"I{i}"]}], + "gender": "female" if i % 2 == 0 else "male", + "birthDate": "1980-01-01", + }) + "\n") + out.append({"type": "Patient", "url": f"{provider}/{name}", "count": per_file}) +manifest = { + "transactionTime": "2024-01-01T00:00:00Z", + "request": f"{provider}/manifest.json", + "requiresAccessToken": False, + "output": out, + "error": [], + "deleted": [], +} +(pathlib.Path(workdir) / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8") +print(f" manifest with {len(out)} output entries") +PY + +echo "==> provider on $PROVIDER_URL" +( cd "$WORKDIR" && timeout "$TTL" python -u -m http.server "$PROVIDER_PORT" --bind 127.0.0.1 ) \ + > "$WORKDIR/provider.log" 2>&1 & + +echo "==> HFS on $HFS_URL (requested fan-out: $FILE_CONCURRENCY)" +HFS_BASE_URL="$HFS_URL" \ +HFS_BULK_SUBMIT_ENABLED=true \ +HFS_BULK_SUBMIT_FILE_CONCURRENCY="$FILE_CONCURRENCY" \ +HFS_LOG_LEVEL=info \ + timeout "$TTL" "${HFS_BIN:-target/debug/hfs.exe}" \ + --database-url "$DB_URL" --log-level info \ + --host 127.0.0.1 --port "$HFS_PORT" \ + > "$WORKDIR/hfs.log" 2>&1 & + +for _ in $(seq 1 90); do + curl -sS -o /dev/null "$HFS_URL/health" 2>/dev/null && break + sleep 1 +done + +echo +grep 'Bulk submit' "$WORKDIR/hfs.log" || true +echo + +cat > "$WORKDIR/submit.json" < "$WORKDIR/status.json" <<'EOF' +{ "resourceType": "Parameters", "parameter": [ + { "name": "submitter", "valueIdentifier": { "system": "http://example.org", "value": "vol" } }, + { "name": "submissionId", "valueString": "vol-942" } ] } +EOF + +echo "==> kick-off" +curl -sS -o "$WORKDIR/kickoff.json" -w " HTTP %{http_code}\n" -X POST "$HFS_URL/\$bulk-submit" \ + -H 'Content-Type: application/fhir+json' --data-binary @"$WORKDIR/submit.json" +cat "$WORKDIR/kickoff.json"; echo + +LOC=$(curl -sS -D - -o /dev/null -X POST "$HFS_URL/\$bulk-submit-status" \ + -H 'Content-Type: application/fhir+json' --data-binary @"$WORKDIR/status.json" \ + | tr -d '\r' | awk -F': ' 'tolower($1)=="content-location"{print $2}') +echo "==> poll: $LOC" + +# The endpoint allows 10 polls per 60s and answers 429 past that, so the +# interval has to be >= 6s. At 10s that is 6 per minute, within the limit. +START=$(date +%s) +CODE="" +for i in $(seq 1 "${MAX_POLLS:-60}"); do + CODE=$(curl -sS -o "$WORKDIR/poll.json" -w "%{http_code}" "$LOC") + echo " poll $i: HTTP $CODE ($(( $(date +%s) - START ))s)" + [ "$CODE" = "200" ] && break + [ "$CODE" = "429" ] && { echo " (rate limited, waiting 60s)"; sleep 60; continue; } + sleep "${POLL_INTERVAL:-10}" +done +ELAPSED=$(( $(date +%s) - START )) + +echo +echo "=== output manifest summary ===" +python - "$WORKDIR/poll.json" <<'PY' +import json, sys +raw = open(sys.argv[1], encoding="utf-8").read() +if not raw.strip(): + print(" (empty body: the last poll was not 200, the import had not finished)") + raise SystemExit(0) +try: + d = json.loads(raw) +except json.JSONDecodeError: + print(" (non-JSON body):", raw[:200]) + raise SystemExit(0) +out = d.get("output", []) +print(" submissionId :", d.get("submissionId")) +print(" files :", len(out)) +print(" resources :", sum(o.get("count", 0) for o in out)) +print(" bytes :", sum(o.get("fileSize", 0) for o in out)) +print(" outcome :", d.get("outcome")) +PY + +echo +echo "=== resource read-back (first, middle, last) ===" +LAST_F=$((FILES - 1)); LAST_I=$((PER_FILE - 1)); MID_F=$((FILES / 2)) +for id in "vol942-0-0" "vol942-$MID_F-0" "vol942-$LAST_F-$LAST_I"; do + curl -sS -o /dev/null -w " GET Patient/$id -> HTTP %{http_code}\n" "$HFS_URL/Patient/$id" +done + +echo +echo "=== lock errors / retries ===" +echo " journal_mode : ${DB_URL}" +echo " 'database is locked' : $(grep -c 'database is locked' "$WORKDIR/hfs.log" || true)" +echo " 'table is locked' : $(grep -c 'database table is locked' "$WORKDIR/hfs.log" || true)" +echo " busy retries : $(grep -c 'sqlite busy during' "$WORKDIR/hfs.log" || true)" +# This path does NOT go through retry_bookkeeping_on_busy: bulk_submit.rs:1488 +# still wraps the claim UPDATE with internal_error, and main.rs:1825 logs it as +# ERROR and sleeps 5s. It is the same failure class as issue #942. +echo " failed claims : $(grep -c 'submit worker claim failed' "$WORKDIR/hfs.log" || true)" +echo " ERROR lines in log : $(grep -c ' ERROR ' "$WORKDIR/hfs.log" || true)" +echo " ingest seconds : ${ELAPSED}" +echo " log : $WORKDIR/hfs.log" + +[ "$CODE" = "200" ] || { echo "RESULT: FAIL - the poll never reached 200"; exit 1; } +echo "RESULT: OK - $TOTAL resources across $FILES files ingested with effective fan-out 2" diff --git a/crates/persistence/src/backends/sqlite/bulk_submit.rs b/crates/persistence/src/backends/sqlite/bulk_submit.rs index 9a0008be7..5a302a98c 100644 --- a/crates/persistence/src/backends/sqlite/bulk_submit.rs +++ b/crates/persistence/src/backends/sqlite/bulk_submit.rs @@ -23,7 +23,9 @@ use crate::core::bulk_submit_worker::{ ManifestFetchParams, ManifestLease, ManifestWorkerView, PollTokenTarget, SubmitClaimStrategy, SubmitFileRecord, SubmitFileRow, SubmitWorkerStorage, }; -use crate::error::{BackendError, BulkSubmitError, StorageError, StorageResult}; +use crate::error::{ + BackendError, BulkSubmitError, StorageError, StorageResult, classify_sqlite_error, +}; use crate::tenant::{TenantContext, TenantId, TenantPermissions}; use super::SqliteBackend; @@ -61,6 +63,54 @@ fn internal_error(message: String) -> StorageError { }) } +/// Bounded retry for the manifest bookkeeping writes (`bulk_manifests` / +/// `bulk_submissions` counter and progress `UPDATE`s). +/// +/// Under file fan-out (#933) every ingest batch holds SQLite's single write +/// lock for its whole extraction + insert span, so a bookkeeping `UPDATE` +/// queued behind several such holds can outlast `busy_timeout` and fail with +/// `SQLITE_BUSY` even though the database is healthy. Aborting the whole +/// manifest over a contended counter update is disproportionate (#942): +/// these writes are safe to reissue — an attempt that fails busy/locked +/// never acquired the write lock, so it changed nothing — and they are +/// guarded by the lease's `worker_id`/`fencing_token` where staleness +/// matters. Each attempt checks out a fresh pooled connection so no pool +/// slot is held across the backoff sleep. +/// +/// Only busy/locked — classified as [`BackendError::Unavailable`] by +/// [`classify_sqlite_error`] — is retried; every other error surfaces +/// immediately. +async fn retry_bookkeeping_on_busy( + what: &str, + mut attempt: impl FnMut() -> StorageResult, +) -> StorageResult { + /// Total attempts before the busy error is surfaced. Each attempt already + /// waits up to the connection's `busy_timeout` (30 s by default), so a + /// handful of retries rides out a fan-out contention spike without + /// masking a genuinely wedged database forever. + const MAX_ATTEMPTS: u32 = 5; + let mut backoff = StdDuration::from_millis(50); + let mut attempt_no = 1u32; + loop { + match attempt() { + Err(StorageError::Backend(BackendError::Unavailable { message, .. })) + if attempt_no < MAX_ATTEMPTS => + { + tracing::warn!( + attempt = attempt_no, + max_attempts = MAX_ATTEMPTS, + backoff_ms = backoff.as_millis() as u64, + "sqlite busy during {what}; retrying: {message}" + ); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(StdDuration::from_secs(1)); + attempt_no += 1; + } + other => return other, + } + } +} + #[async_trait] impl BulkSubmitProvider for SqliteBackend { async fn create_submission( @@ -746,8 +796,13 @@ impl BulkSubmitProvider for SqliteBackend { // write lock, while a standalone flush (the lease keeper's) can // starve for tens of seconds against back-to-back batch // transactions. MAX keeps a late keeper flush from regressing it. + // + // Both bookkeeping writes retry on SQLITE_BUSY instead of aborting + // the manifest (#942): under fan-out this counter update can queue + // behind several batch-long write-lock holds and outlast + // busy_timeout, and a failed attempt applied nothing, so reissuing + // the increment is safe. let now = Utc::now().to_rfc3339(); - let conn = self.get_connection()?; let (consumed, bytes_total) = options .byte_progress .as_ref() @@ -758,40 +813,54 @@ impl BulkSubmitProvider for SqliteBackend { ) }) .unwrap_or((0, 0)); - conn.execute( - "UPDATE bulk_manifests SET - total_entries = total_entries + ?1, - processed_entries = processed_entries + ?2, - failed_entries = failed_entries + ?3, - bytes_processed = MAX(bytes_processed, ?8), - bytes_total = MAX(bytes_total, ?9) - WHERE tenant_id = ?4 AND submitter = ?5 AND submission_id = ?6 AND manifest_id = ?7", - params![ - results.len() as i64, - results.iter().filter(|r| r.is_success()).count() as i64, - error_count as i64, - tenant_id, - &submission_id.submitter, - &submission_id.submission_id, - manifest_id, - consumed, - bytes_total - ], - ) - .map_err(|e| internal_error(format!("Failed to update manifest counts: {}", e)))?; + let total_count = results.len() as i64; + let success_count = results.iter().filter(|r| r.is_success()).count() as i64; + retry_bookkeeping_on_busy("manifest counts update", || { + let conn = self.get_connection()?; + conn.execute( + "UPDATE bulk_manifests SET + total_entries = total_entries + ?1, + processed_entries = processed_entries + ?2, + failed_entries = failed_entries + ?3, + bytes_processed = MAX(bytes_processed, ?8), + bytes_total = MAX(bytes_total, ?9) + WHERE tenant_id = ?4 AND submitter = ?5 AND submission_id = ?6 AND manifest_id = ?7", + params![ + total_count, + success_count, + error_count as i64, + tenant_id, + &submission_id.submitter, + &submission_id.submission_id, + manifest_id, + consumed, + bytes_total + ], + ) + .map_err(|e| { + StorageError::Backend(classify_sqlite_error("Failed to update manifest counts", e)) + }) + }) + .await?; // Update submission updated_at - conn.execute( - "UPDATE bulk_submissions SET updated_at = ?1 - WHERE tenant_id = ?2 AND submitter = ?3 AND submission_id = ?4", - params![ - now, - tenant_id, - &submission_id.submitter, - &submission_id.submission_id - ], - ) - .map_err(|e| internal_error(format!("Failed to update submission: {}", e)))?; + retry_bookkeeping_on_busy("submission updated_at write", || { + let conn = self.get_connection()?; + conn.execute( + "UPDATE bulk_submissions SET updated_at = ?1 + WHERE tenant_id = ?2 AND submitter = ?3 AND submission_id = ?4", + params![ + now, + tenant_id, + &submission_id.submitter, + &submission_id.submission_id + ], + ) + .map_err(|e| { + StorageError::Backend(classify_sqlite_error("Failed to update submission", e)) + }) + }) + .await?; Ok(results) } @@ -1605,9 +1674,11 @@ impl SubmitWorkerStorage for SqliteBackend { failed_entries: u64, last_processed_line: u64, ) -> Result<(), LeaseError> { - let conn = self.get_connection().map_err(LeaseError::Storage)?; - let affected = conn - .execute( + // Absolute-value write guarded by worker_id + fencing_token, so a + // busy retry is idempotent and a stale lease still loses (#942). + let affected = retry_bookkeeping_on_busy("manifest progress update", || { + let conn = self.get_connection()?; + conn.execute( "UPDATE bulk_manifests SET processed_entries = ?1, failed_entries = ?2, last_processed_line = ?3 WHERE tenant_id = ?4 AND submitter = ?5 AND submission_id = ?6 @@ -1624,7 +1695,10 @@ impl SubmitWorkerStorage for SqliteBackend { lease.fencing_token as i64 ], ) - .map_err(|e| LeaseError::Storage(internal_error(format!("update progress: {e}"))))?; + .map_err(|e| StorageError::Backend(classify_sqlite_error("update progress", e))) + }) + .await + .map_err(LeaseError::Storage)?; if affected == 0 { Err(lease_lost(lease)) } else { @@ -1638,9 +1712,11 @@ impl SubmitWorkerStorage for SqliteBackend { bytes_processed: u64, bytes_total: u64, ) -> Result<(), LeaseError> { - let conn = self.get_connection().map_err(LeaseError::Storage)?; - let affected = conn - .execute( + // MAX() keeps the write monotonic, so a busy retry is idempotent and + // the worker_id + fencing_token guard still fences stale leases (#942). + let affected = retry_bookkeeping_on_busy("manifest bytes update", || { + let conn = self.get_connection()?; + conn.execute( "UPDATE bulk_manifests SET bytes_processed = MAX(bytes_processed, ?1), bytes_total = MAX(bytes_total, ?2) @@ -1657,7 +1733,10 @@ impl SubmitWorkerStorage for SqliteBackend { lease.fencing_token as i64 ], ) - .map_err(|e| LeaseError::Storage(internal_error(format!("update bytes: {e}"))))?; + .map_err(|e| StorageError::Backend(classify_sqlite_error("update bytes", e))) + }) + .await + .map_err(LeaseError::Storage)?; if affected == 0 { Err(lease_lost(lease)) } else { @@ -2632,4 +2711,64 @@ mod tests { .is_empty() ); } + + /// #942: a bookkeeping write that hits SQLITE_BUSY (classified as + /// `Unavailable`) is retried and succeeds once the contention clears, + /// instead of aborting the manifest. `start_paused` auto-advances the + /// backoff sleeps. + #[tokio::test(start_paused = true)] + async fn busy_bookkeeping_write_is_retried_until_it_succeeds() { + let attempts = std::sync::atomic::AtomicU32::new(0); + let result = retry_bookkeeping_on_busy("test write", || { + let n = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if n < 2 { + Err(StorageError::Backend(BackendError::Unavailable { + backend_name: "sqlite".to_string(), + message: "database is locked".to_string(), + })) + } else { + Ok(7usize) + } + }) + .await; + assert_eq!(result.unwrap(), 7); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 3); + } + + /// #942: sustained contention is still surfaced — the retry loop is + /// bounded, and the final error is the classified busy error. + #[tokio::test(start_paused = true)] + async fn busy_bookkeeping_write_gives_up_after_bounded_attempts() { + let attempts = std::sync::atomic::AtomicU32::new(0); + let result: StorageResult<()> = retry_bookkeeping_on_busy("test write", || { + attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Err(StorageError::Backend(BackendError::Unavailable { + backend_name: "sqlite".to_string(), + message: "database is locked".to_string(), + })) + }) + .await; + assert!(matches!( + result, + Err(StorageError::Backend(BackendError::Unavailable { .. })) + )); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 5); + } + + /// #942: only busy/locked retries — any other error surfaces on the + /// first attempt, exactly as before. + #[tokio::test] + async fn non_busy_bookkeeping_error_is_not_retried() { + let attempts = std::sync::atomic::AtomicU32::new(0); + let result: StorageResult<()> = retry_bookkeeping_on_busy("test write", || { + attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Err(internal_error("constraint violation".to_string())) + }) + .await; + assert!(matches!( + result, + Err(StorageError::Backend(BackendError::Internal { .. })) + )); + assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1); + } } diff --git a/crates/rest/src/config.rs b/crates/rest/src/config.rs index fa8722ec9..5411ed689 100644 --- a/crates/rest/src/config.rs +++ b/crates/rest/src/config.rs @@ -590,6 +590,17 @@ impl ValidationConfig { } } +/// Effective cap on [`BulkSubmitConfig::file_concurrency`] when the primary +/// storage backend is SQLite. +/// +/// SQLite serialises writers: a manifest's files may be fetched and parsed in +/// parallel, but their batch writes queue behind one exclusive write lock. Past +/// a couple of in-flight files the queued writers wait longer than +/// `busy_timeout` and the ingest fails outright rather than merely running +/// slowly (#942). Two still overlaps fetch and extraction — the part that does +/// parallelise — without pushing the write queue past the timeout. +pub const SQLITE_MAX_FILE_CONCURRENCY: u32 = 2; + /// Bulk Data **Submit** (`$bulk-submit`) configuration, loaded from /// `HFS_BULK_SUBMIT_*` environment variables. /// @@ -619,6 +630,10 @@ pub struct BulkSubmitConfig { /// overlap per-file fetch, parse, and write; a concurrent-writer backend /// (PostgreSQL) turns this into near-linear throughput, while SQLite's /// single writer caps the gain. Set with `HFS_BULK_SUBMIT_FILE_CONCURRENCY`. + /// + /// This is the *configured* value. On SQLite the value actually used is + /// clamped to [`SQLITE_MAX_FILE_CONCURRENCY`] — see + /// [`Self::effective_file_concurrency`]. pub file_concurrency: u32, /// Bulk fast-load (#903): ingest without search-index/FTS writes and /// rebuild them with an automatic per-type reindex when each manifest @@ -705,6 +720,25 @@ impl Default for BulkSubmitConfig { } impl BulkSubmitConfig { + /// Returns the file fan-out this pod should actually use on `backend`. + /// + /// The configured [`Self::file_concurrency`] is honoured on every + /// concurrent-writer backend. SQLite is clamped to + /// [`SQLITE_MAX_FILE_CONCURRENCY`]: its single write lock turns a higher + /// fan-out into a queue of writers that outlast `busy_timeout`, which + /// aborts the manifest instead of just slowing it down (#942). Clamping is + /// therefore a correctness guard, not a tuning preference — an operator who + /// asks for 8 gets a slower import rather than a failed one. + /// + /// The result is always at least `1`, so a configured `0` still ingests. + pub fn effective_file_concurrency(&self, backend: BackendKind) -> u32 { + let configured = self.file_concurrency.max(1); + match backend { + BackendKind::Sqlite => configured.min(SQLITE_MAX_FILE_CONCURRENCY), + _ => configured, + } + } + /// Loads bulk-submit configuration from `HFS_BULK_SUBMIT_*` env vars. pub fn from_env() -> Self { fn env_bool(key: &str, default: bool) -> bool { @@ -2400,4 +2434,42 @@ mod tests { assert_eq!(expected.parse::().unwrap(), variant); } } + + // ── bulk submit file fan-out clamp (#942) ───────────────────── + + #[test] + fn sqlite_clamps_effective_file_concurrency() { + let cfg = BulkSubmitConfig { + file_concurrency: 8, + ..Default::default() + }; + // The single-writer backend is capped regardless of what was asked for. + assert_eq!( + cfg.effective_file_concurrency(BackendKind::Sqlite), + SQLITE_MAX_FILE_CONCURRENCY + ); + // Concurrent-writer backends keep the operator's value. + assert_eq!(cfg.effective_file_concurrency(BackendKind::Postgres), 8); + assert_eq!(cfg.effective_file_concurrency(BackendKind::MongoDB), 8); + assert_eq!(cfg.effective_file_concurrency(BackendKind::S3), 8); + } + + #[test] + fn effective_file_concurrency_never_drops_below_one() { + let cfg = BulkSubmitConfig { + file_concurrency: 0, + ..Default::default() + }; + assert_eq!(cfg.effective_file_concurrency(BackendKind::Sqlite), 1); + assert_eq!(cfg.effective_file_concurrency(BackendKind::Postgres), 1); + } + + #[test] + fn a_configured_value_under_the_sqlite_cap_is_untouched() { + let cfg = BulkSubmitConfig { + file_concurrency: 1, + ..Default::default() + }; + assert_eq!(cfg.effective_file_concurrency(BackendKind::Sqlite), 1); + } }