feat(rust): update CloudFetchConfig with new fields and corrected defaults

[eric-wang-1990]

🥞 Stacked PR

Use this link to review incremental changes.

• stack/pr-phase1-foundation [Files changed]
  • stack/pr-phase2-core-pipeline [Files changed]
    • stack/pr-phase3-integration [Files changed]

---

What's Changed

Please fill in a description of the changes here.

This contains breaking changes.

Closes #NNN.

[eric-wang-1990]

test_cloudfetch_url_expiry_triggers_refetch_and_continues is failing CI because it tests 403 reactive refetch logic that doesn't exist yet in this branch — the pre-flight is_expired() check passes (mock sets expiration 1hr in the future), so the driver downloads the /expired URL, gets a 403, and propagates it to the consumer.

Since the actual 403 handling presumably lands in phase2 or phase3, this test should be #[ignore] in this branch (like the other two tests in the file already are) and un-ignored when the implementation lands. Otherwise it blocks CI and makes it hard to distinguish expected vs real failures.

---

This comment was generated with GitHub MCP.

[eric-wang-1990]

[Medium] is_multiple_of() uses a recently-stabilized API

tests/e2e_cloudfetch.rs uses total_batches.is_multiple_of(50). This was stabilized in Rust 1.86 (released March 2025). If the MSRV or CI toolchain is older than 1.86, this will fail to compile on stable.

Replace with:

if total_batches % 50 == 0 {

---

This comment was generated with GitHub MCP.

[eric-wang-1990]

[Medium] Fire-and-forget initialize() silently fails — consumer blocks forever

streaming_provider.rs spawns initialize() as a background task but provides no signal back if link fetching fails. If fetch_links() errors during init, the consumer calls wait_for_chunk(0) and blocks indefinitely with no error surfaced.

Consider using a tokio::sync::oneshot channel to propagate initialization failure, or check for the error in next_batch() before waiting:

let (init_tx, init_rx) = tokio::sync::oneshot::channel::<Result<()>>();
// ... spawn init task, send Ok/Err via init_tx
// In next_batch(): check init_rx first

---

This comment was generated with GitHub MCP.

[eric-wang-1990]

[Medium] Timeout in wait_for_chunk loops without a termination condition

The timeout branch in wait_for_chunk logs and continues the loop, but there is no maximum retry count. If a download is permanently stuck (worker died without sending), this loops forever logging timeouts.

Add a max timeout count:

let mut timeout_count = 0u32;
// ...
_ = tokio::time::sleep(timeout) => {
    timeout_count += 1;
    if timeout_count >= 3 {
        return Err(DatabricksErrorHelper::io()
            .message(format!("Chunk {} not ready after {} timeouts", chunk_index, timeout_count)));
    }
    debug!("Timeout waiting for chunk {} ({}/3)", chunk_index, timeout_count);
}

---

This comment was generated with GitHub MCP.