Cubestore: Stop GCS list_prefix infinite collection causing OOM

[Xuxiaotuan]

When using GCS as CubeStore remote storage, list_with_metadata_and_map() can cause Router OOM.

Root cause (two layers):

Layer 1 — cloud_storage v0.7.1 ([source](https://github.com/ThouCheese/cloud-storage-rs)):

Object::list_from() uses stream::unfold with a ListState enum. On all 5 error paths, the state is returned unchanged instead of being advanced to Done:

// L382 — get_headers() failed
Err(e) => return Some((Err(e), state)),
// L408 — HTTP != 200
return Some((Err(e), state));
// L410 — send() failed
Err(e) => return Some((Err(e.into()), state)),
// L415 — JSON parse failed
Err(e) => return Some((Err(e.into()), state)),
// L422 — API error in response body
GoogleResponse::Error(e) => return Some((Err(e.into()), state)),

Compare with the success path (L427-432) which correctly advances:

let next_state = if let Some(page_token) = response_body.next_page_token {
    HasMore(page_token)
} else {
    Done     // ← success path correctly terminates
};

Because the error paths return state unchanged (e.g. Start), the next poll_next() retries the same request → infinite error stream.

Layer 2 — list_with_metadata_and_map():

let result = list
    .collect::<Vec<_>>()     // ← never stops
    .await
    .into_iter()
    .flatten()               // ← Err silently discarded
    .flatten()
    .collect::<Vec<_>>();

collect::<Vec<_>>() does not short-circuit on Err. Combined with the infinite stream from Layer 1, this grows unbounded → OOM.

Fix:

-        let result = list
-            .map(|objects| -> Result<Vec<T>, CubeError> {
-                Ok(objects?.into_iter().map(f).collect())
-            })
-            .collect::<Vec<_>>()
-            .await
-            .into_iter()
-            .flatten()
-            .flatten()
-            .collect::<Vec<_>>();
+        tokio::pin!(list);
+        let mut result = Vec::new();
+        while let Some(objects) = list.next().await {
+            let objects = objects?;
+            result.extend(objects.into_iter().map(f));
+        }

First error propagates immediately, no more infinite collection.

[ovr]

/bot-deep-review

[claude]

Claude finished @ovr's task in 6m 12s —— View job

---

Deep review complete — the fix is correct and I'd land it. Found 0 blocking, 1 medium, 2 low, plus 3 follow-ups. Two inline comments posted.

Full review — root-cause verification, caller impact analysis, findings, and follow-ups

Todo list

• Gather context (diff, file, callers)
• Understand the request — deep code review of the GCS list_prefix OOM fix
• Check callers for behavioral regressions (cleanup.rs, rocks_fs.rs, queue.rs, import/mod.rs)
• Verify the root-cause claim as far as the sandbox allows
• Analyze correctness of the fix (pinning, error propagation, metrics)
• Compare against sibling drivers (s3.rs, minio.rs)
• Check test coverage
• Check for prior review threads / dedupe — none exist on this PR
• Post review

---

Verdict

The change is correct, minimal, and moves GCS onto the same shape the other drivers already use. collect::<Vec<_>>() on a Stream provably does not short-circuit on Err, and .flatten().flatten() provably discards Err variants — so the two defects the PR describes are real independent of the upstream diagnosis. Replacing that chain with while let + ? is the right fix, not a workaround.

Worth noting the sibling drivers were already correct, which makes GCS the outlier this PR brings into line:

Driver	Page loop	Error handling
S3 (s3.rs:501-535)	explicit loop over list_page	f(obj)? per object, ? per page
MinIO (minio.rs:314-329)	iter().flat_map()	collect::<Result<Vec<_>, _>>()? — short-circuits
GCS (before)	collect::<Vec<_>>()	errors silently dropped, no short-circuit
GCS (after)	while let + ?	matches S3

The OOM story holds together, and it's worse than the title suggests

The per-iteration allocation on the error path is small — one Err(CubeError) holding a short String. That only reaches OOM if the loop spins fast, which is exactly what the cited get_headers() failure path does: bad or expired credentials fail without a network round trip, so the unfold retries in a tight CPU-bound loop with no I/O to pace it. That's a coherent minutes-to-OOM story. On the paths that do make an HTTP call first, the more visible symptom before OOM would be a hot retry loop hammering GCS. Both are fixed by the same ?.

I could not verify the cloud-storage v0.7.1 line numbers verbatim — WebFetch isn't permitted in this sandbox and the crate source isn't vendored in the repo, so I couldn't read resources/object.rs. The version is confirmed pinned though: cubestore/Cargo.toml:74 declares cloud-storage = "0.7.0", Cargo.lock resolves it to 0.7.1. The described stream::unfold + ListState failure mode is consistent with how that crate is built, and the fix is correct regardless of whether the upstream loop is infinite or merely lossy.

The strongest argument for this PR isn't in the description

Removing the error-swallowing closes a silent data-loss hazard, which is a bigger deal than the OOM. The chain:

1. RocksStoreFs::is_remote_metadata_exists() (rocks_fs.rs:339-347) is nothing but list(format!("{}-current", name)).await?.len() > 0.
2. prepare_metastore (rocks_fs.rs:432-434) gates on it: if !fs::metadata(path).await.is_ok() { if self.is_remote_metadata_exists().await? { ...download... } }.
3. Under the old code, a GCS auth or transient failure returned Ok(vec![]) — indistinguishable from "the bucket is empty" — so the check evaluates to false.
4. Control falls through to rocks_fs.rs:462, info!("Creating {} from scratch in {}", ...), and a fresh empty metastore is constructed with None as the snapshot (rocks_fs.rs:467-472).

On a cold router — new pod, empty volume — that means coming up with an empty metastore over a healthy remote snapshot, which subsequent uploads then overwrite. Propagating the error converts that into a clean startup failure. I'd add this to the PR description; it's the reason this is worth backporting.

Caller impact: no regressions found

Every caller either propagates with ? or handles Err explicitly. The one that changes behavior meaningfully changes it for the better:

• remotefs/cleanup.rs:92-99 — already matches on Err, logs, and continues. The loop sleeps at the top of each iteration (cleanup.rs:58-65), so continue waits out cleanup_interval — no hot-retry risk from the new error path. Previously a truncated list caused under-deletion of orphans (a leak, not destruction); now the round is skipped cleanly.
• metastore/rocks_fs.rs:150, 342, 352, 402, 549 — all ?-propagating. Benefits as described above.
• remotefs/queue.rs:292, 325, 332 and import/mod.rs:991 — propagate or match explicitly.

Implementation details — checked, all fine

• tokio::pin! is correct. StreamExt::next() requires Self: Unpin; tokio::pin! shadows list with a Pin<&mut S>, which is Unpin. The old .map().collect() needed no pinning because those adapters take self by value and pin internally. futures::pin_mut! would be equivalent and futures is already imported at line 10, but tokio::pin! is exported unconditionally and tokio is on features = ["full"] (Cargo.toml:18) — purely stylistic, no change needed.
• The ? conversion resolves. From<cloud_storage::Error> for CubeError must already exist since the pre-change closure used objects? under a Result<Vec<T>, CubeError> return type.
• f needs no mut binding. map(f) moves a copy into the adapter, which owns it and can call it mutably.

I was not able to run cargo check to confirm compilation — the crate registry isn't vendored and the sandbox restricts commands outside the working directory. CI should be the gate here. The reasoning above says it compiles, but treat that as analysis, not verification.

Findings

1. [medium] pages_count estimate reports 0 for empty listings — gcs.rs:330-336, inline comment. result.len() / 1_000 was unavoidable when there were no visible page boundaries; the new loop makes the true count free. As written, any prefix matching zero objects increments REMOTE_FS_OPERATION_CORE{operation:list, driver:gcs} by zero despite issuing a request — and empty results are the dominant pattern for the existence checks at queue.rs:292 and rocks_fs.rs:342/352. S3 already counts properly (s3.rs:513, s3.rs:520). The inline comment has the suggested shape.

2. [low] map(f) copies the mapper per page — gcs.rs:328, inline comment. F: FnMut + Copy means per-page state silently resets. Not a live bug — both callers are stateless and the old code behaved identically — but the loop now reads as if it threads state when it doesn't. mut f: F + .map(&mut f) matches s3.rs:504. Optional.

3. [low] No telemetry is emitted on the error path. With the early ?, a failed list produces no REMOTE_FS_OPERATION_CORE sample at all. The old code, broken as it was, still emitted one. Since the failure mode this PR fixes is "GCS listing errors," losing the ops signal for exactly those calls is mildly counterproductive for diagnosis. Emitting the page count before returning — or a separate error-tagged counter — would keep failures visible. Minor, and partly addressed by fixing finding 1.

4. [low] The pages_count > 100 warning says "S3 list" inside the GCS driver — gcs.rs:335, copy-pasted from s3.rs:545. Folded into the first inline comment since it's the same edit. Outside the diff hunk so it couldn't be commented on directly.

Follow-ups — out of scope, worth tracking separately

A. The success path is still unbounded. This PR removes the error-driven growth but not the fundamental accumulation. GCSRemoteFs takes the default ExtendedRemoteFs impl (gcs.rs:296-298, with its existing // TODO: Make a faster implementation), and that default list_by_page (mod.rs:91-104) is candid about it: // Note, this implementation doesn't actually paginate. It calls list() and wraps the whole thing in a one-shot stream. So cleanup.rs:92's list_with_metadata("") still materialises every object in the bucket. The pages_count > 100 warning implies >100k objects is an anticipated scale. S3 has a real streaming list_by_page (s3.rs:454-476) that GCS could be modelled on.

B. The upstream bug is still there for every other consumer. cloud-storage 0.7.1 pulls reqwest 0.11.27, base64 0.13, and openssl per Cargo.lock — an old dependency surface. The ? here is the right local mitigation, but a [patch.crates-io] fork, or migrating to google-cloud-storage / object_store, would fix the class rather than this instance.

C. Testing. No test accompanies the fix. That's understandable rather than negligent: list_with_metadata_and_map calls Object::list_prefix directly, so there's no seam to inject a stream and no way to exercise it without live GCS credentials. If you want coverage, the cheap version is extracting the collection logic:

async fn collect_pages<T, F, S>(stream: S, f: F) -> Result<Vec<T>, CubeError>
where
    S: Stream<Item = Result<Vec<Object>, cloud_storage::Error>>,
    F: FnMut(Object) -> T + Copy,
{ /* the while-let body */ }

That makes the contract unit-testable with futures::stream::iter([Ok(vec![obj]), Err(...), Ok(vec![obj])]) — asserting the call returns Err and consumes only the first two items. A regression test that pins "does not swallow, does not keep polling" is the highest-value addition available here, and it's a genuine guard against the old shape creeping back.

Security

Nothing of concern. No change to authentication, credential handling, or input validation; ensure_credentials_init and gcs_path are untouched. If anything the change is net-positive for availability, since it removes an unauthenticated-error-triggered resource exhaustion path.

· branch [`fix/cubestore-gcs-list-oom`](https://github.com/cube-js/cube/tree/fix/cubestore-gcs-list-oom)

[ovr]

/bot-deep-review

[claude]

Claude finished @ovr's task in 4m 50s —— View job

---

Re-review after 951fe89 — both prior findings fixed, nothing blocking. 0 blocking · 0 medium · 2 nits · 3 follow-ups. I'd land this.

Full review — delta since last round, verification, remaining nits, follow-ups

• Gather context (fetch base, PR diff, changed file, callers)
• Check what changed since my previous review (951fe89)
• List existing review threads; resolve my own stale ones, dedupe
• Re-review the current diff end-to-end (correctness, metrics, perf, security)
• Verify compilation as far as the sandbox allows
• Post review

---

Delta since the last round

951fe89 ("address review comments — count actual pages, fix warning msg, mut f") addressed both findings exactly as suggested. I resolved both of my threads rather than leaving them dangling:

Prior finding	Status	Evidence
[medium] pages_count estimate reports 0 for empty listings	fixed	gcs.rs:326 let mut pages_count: i64 = 0;, gcs.rs:329 pages_count += 1; inside the loop; the result.len() / 1_000 estimate and the % 1_000 block are gone, and the now-redundant as i64 cast was dropped at gcs.rs:336
[medium, same edit] warning said "S3 list" inside the GCS driver	fixed	gcs.rs:333 now reads "GCS list returned more than 100 pages"
[low] map(f) copies the mapper per page, resetting FnMut state	fixed	gcs.rs:317 mut f: F, gcs.rs:330 .map(&mut f) — matches s3.rs:504

No new inline comments this round; there's nothing actionable left that isn't either a nit or a separate follow-up.

Verdict

The while let + ? loop is the correct fix, and the metrics cleanup makes GCS the best-instrumented of the three drivers rather than the worst. The two defects the PR set out to fix are both provable from the old code independent of the upstream diagnosis: Stream::collect::<Vec<_>>() does not short-circuit on Err, and .flatten().flatten() discards Err variants outright.

Final shape, gcs.rs:314-343:

let list = Object::list_prefix(self.bucket.as_str(), prefix.as_str()).await?;
tokio::pin!(list);
let mut result = Vec::new();
let mut pages_count: i64 = 0;
while let Some(objects) = list.next().await {
    let objects = objects?;
    pages_count += 1;
    result.extend(objects.into_iter().map(&mut f));
}

Against the siblings, GCS is now the one that's converged:

Driver	Page loop	Error handling	Page metric
S3 (s3.rs:501-535)	explicit loop over list_page	f(obj)? per object, ? per page	real counter
MinIO (minio.rs:314-329)	iter().flat_map() (SDK pre-collects pages)	collect::<Result<Vec<_>, _>>()?	none emitted
GCS (before)	collect::<Vec<_>>()	errors silently dropped, no short-circuit	estimated from len()
GCS (after)	while let + ?	? per page	real counter

The strongest argument for this PR still isn't in the description

Removing the error-swallowing closes a silent data-loss hazard, which matters more than the OOM. Two paths, both re-verified against the current tree:

1. Cold-start metastore clobber. RocksStoreFs::is_remote_metadata_exists() (rocks_fs.rs:339-347) is just list(format!("{}-current", name)).await?.len() > 0. prepare_metastore (rocks_fs.rs:432-434) gates the snapshot download on it. Under the old code a GCS auth or transient failure returned Ok(vec![]) — indistinguishable from "bucket is empty" — so the check went false, control fell to rocks_fs.rs:462 info!("Creating {} from scratch in {}", ...), and a fresh empty metastore was built with None as the snapshot (rocks_fs.rs:467-472). On a new pod with an empty volume that means booting empty over a healthy remote snapshot, which later uploads overwrite. Now it's a clean startup failure.

2. Misattributed upload failures. CommonRemoteFsUtils::check_upload_file (mod.rs:143-165) runs after every upload and, on an empty list, raises "File X can't be listed after upload. Either there's Cube Store cluster misconfiguration, or storage can't provide the required consistency." A swallowed GCS error produced exactly that message — pointing operators at consistency/config instead of at the real API error. mod.rs:164 Err(e) => Err(e) now surfaces the actual cause.

Both are worth a line in the PR description; they're the reason this is worth backporting.

Caller impact: re-checked, no regressions

Every caller of list / list_with_metadata either ?-propagates or matches Err explicitly — I swept all 20 non-test call sites. Specifically checked for the failure mode a new early-Err could introduce (a caller that retries without backoff, turning the old OOM into a CPU spin):

• remotefs/cleanup.rs:92-99 — matches on Err, logs, continues. The loop sleeps at the top of each iteration (cleanup.rs:58-65), so continue waits out cleanup_interval. No hot retry. Previously a truncated list under-deleted orphans (a leak); now the round is skipped cleanly.
• remotefs/queue.rs:292 — inside download_file's recv() loop, but the ? returns from the function rather than re-looping. No spin.
• metastore/rocks_fs.rs:150, 342, 352, 402, 511, 549 — all ?-propagating; benefit as described above.
• import/mod.rs:991, remotefs/mod.rs:96/143/356, sql/mod.rs:2488/2495 — propagate or match explicitly.

Implementation details — verified

• add_with_tags takes i64 (util/metrics.rs:76), so passing the i64 counter directly and deleting the cast is right.
• futures::StreamExt is in scope (gcs.rs:10), so .next() resolves.
• tokio::pin! is correct and available. StreamExt::next() requires Self: Unpin; the macro shadows list with a Pin<&mut S>, which is Unpin. It's #[macro_export]ed unconditionally, so no feature gate is involved (and tokio is on features = ["full"], Cargo.toml:18, regardless).
• .map(&mut f) type-checks. impl<A, F: FnMut<A>> FnMut<A> for &mut F is in core, so &mut F satisfies the FnMut(Object) -> T bound. The mut on f is load-bearing for the reborrow, so no unused_mut. Reborrowing per iteration is fine — the borrow ends when extend returns.
• One HTTP request always yields one stream item, including for a zero-match prefix (ListState::Start always issues a request; GCS omits items and the crate maps that to an empty Vec). So pages_count >= 1 for any successful listing — which is precisely what fixes the empty-listing undercount.

Verification gap, stated plainly: I could not compile this. I wrote a standalone reproduction of the generic-mapper/reborrow/pin shape to type-check with rustc directly, but the sandbox denied both the scratch write and the rustc invocation, so the points above are analysis, not execution. The Rust workflow (run 30412178978) and Build (run 30412179037) are both still in_progress and should be the gate. I also could not read cloud-storage v0.7.1's resources/object.rs to confirm the cited line numbers — WebFetch is unavailable here and the crate isn't vendored — though the pin is confirmed (cubestore/Cargo.toml:74 declares "0.7.0", Cargo.lock resolves 0.7.1). The fix is correct whether the upstream loop is infinite or merely lossy.

Remaining nits — neither worth a change here

1. + Copy on the F bound is now vestigial — gcs.rs:320. With &mut f the mapper is reborrowed rather than copied, so Copy no longer carries weight. s3.rs:507 has the identical leftover, so keeping it is the consistent choice; dropping it in both files would be a separate tidy-up.

2. No telemetry on the error path — the early ? means a failed list emits no REMOTE_FS_OPERATION_CORE sample, where the old (broken) code always emitted one. Slightly unfortunate given the failure mode this PR fixes is "GCS listing errors." But s3.rs:516-518 behaves identically — ? on list_page returns before pages_count_app_metrics_and_logging — so GCS now matches the house pattern rather than diverging from it. Fixing it is a cross-driver change, not a GCS one.

Follow-ups — out of scope, worth tracking

A. The success path is still unbounded. This PR removes the error-driven growth, not the accumulation itself. GCSRemoteFs takes the default ExtendedRemoteFs impl (gcs.rs:296-298, under its existing // TODO: Make a faster implementation), and that default list_by_page (mod.rs:91-104) says so outright: // Note, this implementation doesn't actually paginate. It calls list() and wraps the whole result in a one-shot stream — so cleanup.rs:92's list_with_metadata("") still materialises every object in the bucket. s3.rs:454-476 has a real streaming list_by_page that GCS could be modelled on.

B. The pages_count > 100 warning is a post-mortem, not a circuit breaker. It fires after the loop drains, so during a genuine runaway (pathological repeating next_page_token, or the scale that motivated the threshold) it can't warn until the memory is already committed. Pre-existing and shared with s3.rs:542-548, but if >100k objects is an anticipated scale, warning inside the loop every N pages — or capping pages outright — would make the guard actionable. Cross-driver change.

C. Metric continuity, for whoever owns the dashboards. REMOTE_FS_OPERATION_CORE{operation:list, driver:gcs} will step up after this ships: empty listings now count 1 instead of 0, and short pages count fully instead of being floor-divided. That's the metric becoming correct, but any alert threshold calibrated on the old estimate will want a look.

D. Testing — still the highest-value addition available. No test accompanies the fix, which is understandable rather than negligent: list_with_metadata_and_map calls Object::list_prefix directly, so there's no seam to inject a stream and no way to exercise it without live GCS credentials. The cheap version is extracting the collection logic behind a Stream bound:

async fn collect_pages<T, F, S>(stream: S, mut f: F) -> Result<(Vec<T>, i64), CubeError>
where
    S: Stream<Item = Result<Vec<Object>, cloud_storage::Error>>,
    F: FnMut(Object) -> T,
{ /* the while-let body, returning (result, pages_count) */ }

That makes the contract unit-testable with futures::stream::iter([Ok(vec![obj]), Err(..), Ok(vec![obj])]) — assert the call returns Err and that only the first two items were consumed. A test pinning "does not swallow, does not keep polling" is a genuine guard against the old shape creeping back, and it'd cover the pages_count == 1 empty-listing case too.

Security

Nothing of concern. No change to authentication, credential handling, or input validation — ensure_credentials_init (gcs.rs:23-...) and gcs_path (gcs.rs:345-354) are untouched. Net-positive for availability: it removes a resource-exhaustion path reachable from an unauthenticated-error condition, and net-positive for diagnosability, since credential failures now surface as errors instead of empty result sets.

· branch [`fix/cubestore-gcs-list-oom`](https://github.com/cube-js/cube/tree/fix/cubestore-gcs-list-oom)