fix: bulk-submit(sqlite): import aborts with 'database is locked' at high file concurrency with the full search-parameter registry

.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`.

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<dyn SubmitInputFetcher>,
     output: Arc<dyn ExportOutputStore>,
     cfg: &helios_rest::config::BulkSubmitConfig,
+    file_concurrency: u32,
     reindex_hook: Option<Arc<dyn helios_persistence::core::DeferredReindexHook>>,
 ) {
     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,

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" <<EOF
+{
+  "transactionTime": "2024-01-01T00:00:00Z",
+  "request": "$PROVIDER_URL/manifest.json",
+  "requiresAccessToken": false,
+  "output": [
+    { "type": "Patient", "url": "$PROVIDER_URL/patients.ndjson", "count": 2 }
+  ],
+  "error": [],
+  "deleted": []
+}
+EOF
+
+# --- build ---------------------------------------------------------------
+# On Windows/MSVC the debug hfs binary overflows the main thread stack while
+# building the SearchParameter registry, so it is relinked with a 32 MB stack.
+# That is only a link flag: it does not change the code.
+# SKIP_BUILD=1 avoids recompiling: on Windows the linker cannot replace
+# target/debug/hfs.exe while another instance holds it open ("Access is
+# denied", os error 5), and killing that instance is not an option.
+if [ "${SKIP_BUILD:-0}" = "1" ]; then
+  echo "==> 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 <<EOF
+
+ready.
+
+  HFS_URL      = $HFS_URL
+  PROVIDER_URL = $PROVIDER_URL
+  logs         = $WORKDIR/hfs.log , $WORKDIR/provider.log
+
+Both shut themselves down in ${TTL}s. Fan-out lines from startup:
+
+$(grep 'Bulk submit' "$WORKDIR/hfs.log" || true)
+EOF

crates/hfs/tests/bulk_submit/run_bulk_submit_postgres_check.sh

@@ -0,0 +1,80 @@
+#!/usr/bin/env bash
+#
+# Gap 2 of issue #942: check that the clamp is NOT applied when the primary
+# backend is PostgreSQL, which is the `_ => 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