stream resources to fix OOM scan

[bomanaps]

Fixes #1060 on OOM on large corpora caused by MongoResourceScan::scan_resources materializing all live resources as Vec; replaced with a streaming cursor on MongoDB and pipelined GETs on S3, with into_content_with_meta so since filter sees version metadata, correct compartment pre-fetch, error propagation, and is_last on the final chunk.

[bomanaps]

Hello @dougc95 please can I get a review here

[smunini]

Review: performance/memory measurements + findings

Merged main into the branch and pushed it (e8169684f). Then benchmarked the in-process scan path against main on a real MongoDB backend, and reviewed the diff.

TL;DR: the core change works and the numbers are excellent — peak RSS goes from linear-in-corpus to essentially flat, and it's faster too. But there's one behavioural regression in CompartmentFilter::apply that reaches well outside this PR's intended blast radius, plus two CI-blocking chores.

---

1. Merge of main — one resolution you should know about

The branch's crates/persistence/src/backends/sqlite/schema.rs hunk was a drive-by fix for a duplicate migrate_v29_to_v30 that existed at your branch point (390d8f863) and has since been fixed on main independently. Git's textual merge happily re-applied that hunk into the unrelated migrate_v32_to_v33 — silently, no conflict — leaving the v30 index_pending column/index work duplicated inside the v33 migration with misleading "v30 ..." error strings.

I dropped it and kept main's schema.rs verbatim. schema.rs is now byte-identical to main, and the PR is back to 6 files / +1026 −262.

---

2. Performance & memory

Harness. An ad-hoc #[ignore]d bench (not committed) against a real mongo:7 single-node replica set. Corpus is N synthetic ~1.4 KB Observations (BP panel with 2 components + a note) all referencing Patient/bench-p1, seeded once and reused by both branches. The view projects 4 columns. A patient=Patient/bench-p1 filter is supplied to force the in-process fallback runner (Mongo's native aggregation runner handles unfiltered views). One run_view per process, no warm-up, VmHWM reset via /proc/self/clear_refs immediately before the measured run — so peak RSS is attributable to the run and not to seeding. --release, 32-core Linux box, MongoDB in Docker on the same host.

Peak RSS of the run (VmHWM, baseline ≈ 33.5 MiB before the run):

Observations	main	this PR	reduction
20,000	735.9 MiB	275.8 MiB	−62%
50,000	1748.0 MiB	326.9 MiB	−81%
100,000	3434.9 MiB	347.1 MiB	−90%

The shape is the point, more than the absolute numbers:

• main is linear: ≈ 34 KiB of RSS per resource, dead straight (736 → 1748 → 3435). Extrapolating, a 1M-Observation compartment run needs ~34 GiB — which is sof(mongodb): MongoResourceScan::scan_resources returns every live resource of a type as Vec<Value> #1060.
• This PR is flat: the marginal cost from 20k → 100k is ≈ 0.9 KiB per resource, and most of that looks like allocator growth rather than retained live data. 5× the corpus costs 26% more memory.

Wall time and latency:

Observations		main	this PR
20,000	wall / first row / rows-per-s	1.775 s / 1.272 s / 11,266	1.119 s / 0.133 s / 17,869
50,000	wall / first row / rows-per-s	4.503 s / 3.132 s / 11,103	2.609 s / 0.127 s / 19,167
100,000	wall / first row / rows-per-s	8.947 s / 6.215 s / 11,177	5.192 s / 0.135 s / 19,261

So this is not a memory-for-speed trade: end-to-end throughput improves ~1.7× (11.2k → 19.3k rows/s) because the cursor drain, the compartment filter and the FHIRPath engine now overlap instead of running in series. And time-to-first-row becomes constant at ~0.13 s instead of growing linearly with the corpus (6.2 s at 100k on main) — which is the difference that actually shows up in the SQL Export UI.

On the remaining ~300 MiB. It's the RESOURCE_CHANNEL_BUFFER (4) × CHUNK_SIZE (1024) in-flight window plus the per-chunk typed FhirResource pool, not a leak. Confirmed by re-running 100k with CHUNK_SIZE = 256 / RESOURCE_CHANNEL_BUFFER = 1:

100k config	peak RSS	throughput	first row
1024 / 4 (as submitted)	347.1 MiB	19,261 rows/s	0.135 s
256 / 1	177.1 MiB	18,268 rows/s	0.057 s

Another −49% peak for −5% throughput. Not asking you to change it in this PR — the constants are the right knobs and the current values are a defensible default — but it's worth a comment recording the trade, since the pairing is what sets the floor.

---

3. Findings

🔴 Blocking — CompartmentFilter::apply drops Groups that are in the patient compartment (crates/sof/src/lib.rs:1276)

apply() short-circuits on any Group resource and returns "is this Group in group_refs?". The code it replaced only took that shortcut for Groups that were explicitly requested, and otherwise fell through to resource_in_patient_compartment. That fall-through matters: Group is in the patient CompartmentDefinition, via member (verified in data/compartment-definitions-r4.json). So a Group whose member.entity references the filtered patient used to be in that patient's compartment and now is not.

Confirmed empirically — same test, same input, both branches:

// Patient/p1, plus Group/g1 with member.entity -> Patient/p1
filter_resources_by_patient_and_group(resources, &["Patient/p1"], &[], FhirVersion::R4)

// main:     ["Patient", "Group"]
// this PR:  ["Patient"]          <- Group/g1 silently dropped

The reason this is blocking rather than a nit is the blast radius. filter_resources_by_patient_and_group now delegates to CompartmentFilter, and it has two callers outside the streaming path:

• crates/rest/src/handlers/sof/run.rs:538 — the HFS SQL-on-FHIR $run endpoint, every backend
• crates/sof/src/handlers.rs:412,455 — sof-cli and sof-server

So a ViewDefinition with "resource": "Group" and a patient= filter returns zero rows everywhere, not just on Mongo/S3. Nothing in the existing suite catches it (the full helios-sof test suite passes on the branch), which is worth a regression test of its own.

The fix is small — don't short-circuit, just prefer the explicit answer:

if resource.get("resourceType").and_then(|v| v.as_str()) == Some("Group")
    && resource.get("id").and_then(|v| v.as_str()).is_some_and(|id| {
        self.group_refs.iter().any(|g| g == &format!("Group/{id}") || g == id)
    })
{
    return Ok(true);
}
compartment::resource_in_patient_compartment(resource, &self.targets, self.fhir_version)

🟠 CI will fail on two chores

• cargo fmt --all -- --check fails — 6 hunks in crates/persistence/src/sof/in_process.rs (the helios_sof use-list, with_reference_resolver's signature, the resolve_batch_external call, the row_tx.blocking_send arm, and two others).
• cargo clippy --all-targets --all-features -- -D warnings fails on 3 missing_docs in in_process.rs — ResourceScan::scan_resources (:65), InProcessSofRunner::new (:85), and with_reference_resolver (:98). The crate is #![warn(missing_docs)] and CI's allow-list doesn't cover it.

Both come from the same source: the diff deletes a lot of existing doc comments that weren't in the way — map_engine_error, row_to_view_row, the StaticScan/StaticResolver/resolve_view test docs, and the four /// blocks on the unit tests explaining why each compartment case exists. Please restore those; they're the kind of context that's expensive to reconstruct later, and the three on public items are what's actually breaking clippy.

🟡 is_last is not the last chunk (crates/persistence/src/sof/in_process.rs:323)

let is_last = res_rx.is_empty();

is_empty() on the receiver means "nothing queued right now", which under backpressure is true for most chunks whenever the engine outruns the scan — not "this is the final chunk". Today this is harmless: process_chunk_with_external only copies it into ChunkedResult.is_last, and the runner ignores that field. But the PR description lists "is_last on the final chunk" as one of the fixes, and it isn't one — it's a correctly-shaped trap for whoever first writes a per-run finalizer keyed off it. Either compute it properly (hold one batch back and flush after the channel closes) or drop the claim and set it to false with a comment.

🟡 Patient-target views still materialise the whole type (crates/persistence/src/sof/in_process.rs:174)

The compartment pre-fetch drains scan_resources(tenant, "Patient") into a Vec unconditionally. For the motivating case ("resource": "Observation") the comment's "small enough to materialise in full" holds. But for a view with "resource": "Patient" plus a patient= filter, that Vec is the large type — so #1060 is not fixed for that shape, and the target type then gets scanned a second time as the stream. main avoided the double scan with its if wanted && supporting != resource_type guard; peak memory is about the same either way, but it's now two full passes over the collection instead of one.

Not necessarily in scope for this PR, but the comment currently reads as an unconditional claim. Worth either narrowing it to say the Patient/Group pre-fetch is the remaining unbounded allocation, or reusing the already-materialised supporting vec when it matches resource_type.

---

4. What I verified passing

On the merged head, with a real Mongo replica set and MinIO in Docker:

• helios-persistence sof::in_process unit tests — 5 passed
• mongodb_tests SOF subset — 3 passed, including the new sof_scan_streams_across_multiple_batches and sof_since_filter
• minio_s3_tests SOF subset — 3 passed (scan_resources_include_server_meta, since_filter, patient_filter)
• helios-sof — 298 lib tests + the full integration suite, all green

The new tests are good, and test_minio_sof_scan_resources_include_server_meta in particular pins down a real latent bug: S3's old scan_live_resources used content().clone(), so meta.versionId/meta.lastUpdated never reached the runner and since on S3 silently matched nothing. Nice catch — that deserves a line in the PR description, since it's a user-visible fix independent of the OOM work.