feat(backend/opendal): add opt-in per-part write progress via ProgressLayer + chunked writer PR Description

[543069760]

Summary

This PR adds opt-in, real-time write progress reporting to the OpenDAL backend.

Previously, OpenDALBackend::write_bytes uploaded each file with a single operator.write(path, buf) call. For services backed by an S3-style multipart API (e.g. Tencent COS), OpenDAL's MultipartWriter applies a write_once optimization when the whole buffer is handed over in one shot: with a known size and a single write, it issues a single PutObject request instead of initiating a multipart upload. As a result, any progress observer only ever sees a single jump from 0 to the full object size at the very end of the upload — there is no intra-object progress.

This makes it impossible to surface smooth upload progress (e.g. to a JNI caller driving a progress bar) for large packs.

Approach

We introduce a custom OpenDAL Layer (ProgressLayer) that wraps the underlying writer and atomically accumulates the number of bytes written each time a Buffer is flushed down to the inner writer. To actually produce multiple flushes (and therefore multiple progress steps), the backend switches from a single one-shot operator.write to a chunked writer that feeds the data in fixed-size chunks, which forces OpenDAL past its write_once optimization and into the real multipart path (InitiateMultipartUpload + multiple UploadPart + CompleteMultipartUpload).

The behavior is opt-in and fully backward compatible:

• The existing OpenDALBackend::new(...) constructor is unchanged and stores counter: None. Backends created this way keep the original single-shot operator.write(buf) path — zero behavior change for all existing backends (fs / sftp / rest / s3 / …).
• A new OpenDALBackend::new_with_progress(...) constructor accepts an Option<WrittenCounter>. Only when a counter is provided is the ProgressLayer assembled and the chunked writer path taken.

Changes

crates/backend/src/progress_layer.rs

• pub type WrittenCounter = Arc<AtomicU64> — shared, cheaply-cloneable counter handle. The caller keeps one Arc clone to poll progress.
• ProgressLayer — an opendal::raw::Layer holding a WrittenCounter.
• ProgressAccessor / ProgressWriter — LayeredAccess / oio::Write wrappers. ProgressWriter::write records bs.len() only after the inner write succeeds (so failed writes / retries don't inflate the count), using Ordering::Relaxed. close / abort are passed straight through.
• The layer operates on the async operator (assembled before the blocking wrapper). Granularity = one increment per chunk the underlying service writer receives (i.e. per multipart part for S3-like services).

crates/backend/src/lib.rs

• Export the new module and public types so callers (e.g. the JNI layer) can construct a WrittenCounter and call new_with_progress.

crates/backend/src/opendal.rs

• Import opendal::options::WriteOptions (used to set the writer chunk size).
• mod constants: add pub(super) const CHUNK_SIZE: usize = 8 * 1024 * 1024; (8 MiB). Chosen to be larger than the ~5 MiB minimum part size required by S3/COS-style services, so it reliably crosses OpenDAL's write_once optimization and triggers a real multipart upload.
• OpenDALBackend struct: add a counter: Option<WrittenCounter> field that records whether a ProgressLayer was assembled, so write_bytes can branch on it.
• new(...): unchanged logic; now sets counter: None.
• new_with_progress(path, options, counter): same operator-construction logic as new, but when counter.is_some() it assembles the ProgressLayer on the async operator (using a cheap Arc clone so the counter is also stored in the struct), then wraps it in the blocking Operator. Stores the counter in the struct.
• write_bytes(...): branch on self.counter:
  • Some(_) → build a chunked blocking writer via operator.writer_options(path, WriteOptions { chunk: Some(constants::CHUNK_SIZE), ..Default::default() }), loop over buf.chunks(constants::CHUNK_SIZE) calling writer.write(chunk), then writer.close() (return value discarded to satisfy -W unused-results). This drives ProgressLayer to fire once per part.
  • None → keep the original single-shot operator.write(&filename, buf).

Tests

crates/backend/tests/

• progress_layer.rs (runs on the memory service, always executed):
  • progress_layer_counts_written_bytes — a backend built with a counter reports counter >= bytes_written after a write.
  • default_path_without_counter_still_writes — regression: with counter = None the write works and reads back identically (default path unaffected).
• progress_layer_cos.rs (real Tencent COS, #[ignore] by default):
  • cos_progress_layer_per_part_granularity — requires COS_SECRET_ID, COS_SECRET_KEY, COS_BUCKET, COS_ENDPOINT, COS_ROOT. Uploads a 128 MiB pack and asserts the counter advances in multiple steps, cleaning up the object afterwards.

Verification

• cargo build -p rustic_backend --features opendal — clean (no new warnings).
• cargo clippy -p rustic_backend --features opendal — no new lints introduced (pre-existing repo warnings left untouched).
• cargo test -p rustic_backend --features opendal — memory regression tests pass; the COS test is correctly reported as ignored.
• Real COS run (cargo test -p rustic_backend --features opendal --test progress_layer_cos -- --ignored --nocapture): the network trace shows InitiateMultipartUpload + multiple UploadPart (partNumber=…&uploadId=…, content-length: 8388608) + CompleteMultipartUpload, and the PROGRESS-COS log shows the counter stepping in 8 MiB increments: 0 → 8388608 → 16777216 → … → 134217728 (16 parts for 128 MiB).

Backward Compatibility

• Existing new(...) callers are unaffected: counter = None → original single-shot operator.write path, identical behavior for all backends.
• The chunked writer and multipart path are only taken when a caller explicitly opts in via new_with_progress(...) with a counter.

Summary

This PR adds opt-in, real-time write progress reporting to the OpenDAL backend.

Closes #538