From 39c3526bc424b599b17820b5523bc979fd4de8ba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Thu, 20 Aug 2026 00:43:01 +0800 Subject: [PATCH] fix(get): name the failing object on mid-stream GET body failures A GET body that ends short of its committed Content-Length breaks every downstream copier. Replication, site replication and `rclone sync` all read locally and PUT remotely, so a truncated source read surfaces as an `unexpected EOF` on the destination's PUT and an `Io error: error reading a body from connection` 500 on the receiving server. Issue #4784 stalled for a month because the source side reported none of it: * `GetObjectReaderStream`'s short-read and read-error arms only fed a metric. Their log lines sat behind the `tracing-chunk-debug` cargo feature, which is not in the default feature set and therefore is not compiled into any released binary. * `GetObjectStreamingReader` did log mid-stream failures, but discarded the bucket and key its constructor was already handed, leaving only a request_id that cannot be resolved back to an object after the request is over. * Those lines were `warn!`, while DEFAULT_LOG_LEVEL is `error`, so a default deployment filtered them out even where they existed. Keep the object identity on both readers and name it in every stream-body failure; log the reader-stream short-read arm unconditionally; and raise the two states that mean "the server cannot deliver the length it already committed" - short_eof and read_failed - to `error!`. Stall timeouts, slow first bytes and client-side drops stay at `warn!`. The reader-stream read-error arm stays feature-gated on purpose: every production body wraps a GetObjectStreamingReader, which already reports that same error once with the object identity, so a second unconditional line per failed GET would read as two distinct faults. Tests: unit coverage asserts the captured event fields rather than just the returned error, and a new e2e reproduces the fault over the S3 API against a beyond-quorum damaged object, asserting the evidence at the default `error` log level. `get_object_reader_stream_errors_on_short_eof` becomes serial: it drives the same log callsite as the capture test, and tracing caches callsite interest process-wide, so running it concurrently on a subscriber-less thread re-cached that callsite as "never interested" and blinded the capture. --- .../get_stream_failure_observability_test.rs | 250 ++++++++++++++++++ crates/e2e_test/src/lib.rs | 5 + rustfs/src/app/object_usecase.rs | 226 +++++++++++++++- 3 files changed, 469 insertions(+), 12 deletions(-) create mode 100644 crates/e2e_test/src/get_stream_failure_observability_test.rs diff --git a/crates/e2e_test/src/get_stream_failure_observability_test.rs b/crates/e2e_test/src/get_stream_failure_observability_test.rs new file mode 100644 index 000000000..8777d1350 --- /dev/null +++ b/crates/e2e_test/src/get_stream_failure_observability_test.rs @@ -0,0 +1,250 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! E2E proof that a mid-stream GET failure is *reportable* — rustfs#4784. +//! +//! The functional invariant (a beyond-quorum read must fail rather than return +//! a clean short body) is already covered by +//! `degraded_read_eof_regression_test`. This suite covers the half that issue +//! #4784 got stuck on for a month: whether an operator can tell, from the +//! source server's log alone, that a GET failed mid-body and **which object** +//! it failed on. +//! +//! The reporter saw only downstream symptoms — `rclone` reporting +//! `unexpected EOF` on its PUT, and the receiving RustFS logging +//! `Io error: error reading a body from connection` with a 500. In a cross-remote +//! `rclone sync`, the source GET body *is* the destination PUT body, so a source +//! read that ends short of its committed `Content-Length` surfaces as a PUT +//! failure on the far side. Built-in replication and site replication have the +//! same shape (read locally, PUT remotely), which is why every transport in that +//! report failed the same way. +//! +//! The source side, meanwhile, said nothing: +//! * `GetObjectReaderStream`'s short-read and read-error arms only incremented +//! a metric; their log lines sat behind the `tracing-chunk-debug` cargo +//! feature, which is not in the default feature set and therefore is not +//! compiled into any released binary. +//! * `GetObjectStreamingReader` did log mid-stream failures, but only under a +//! `request_id` — with no bucket or object name, a failure could not be +//! traced back to the object that caused it. +//! * Those lines were `warn!`, while `DEFAULT_LOG_LEVEL` is `error`, so a +//! default deployment filtered them out anyway. +//! +//! This test reproduces the source-side fault against a real server over the S3 +//! API and asserts the operator-visible evidence, at the **default** log level. + +#[cfg(test)] +mod tests { + use crate::chaos::DiskFaultHarness; + use crate::common::init_logging; + use aws_sdk_s3::Client; + use aws_sdk_s3::primitives::ByteStream; + use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart}; + use serial_test::serial; + use std::error::Error; + use tokio::time::{Duration, timeout}; + use tracing::info; + + type TestResult = Result<(), Box>; + + const MIB: usize = 1024 * 1024; + const OP_TIMEOUT: Duration = Duration::from_secs(90); + + /// The structured event name every GET body failure is tagged with. + const STREAM_BODY_EVENT: &str = "get_object_stream_body"; + + fn payload(len: usize, seed: u8) -> Vec { + (0..len) + .map(|i| (i as u64).wrapping_mul(2654435761).wrapping_add(seed as u64) as u8) + .collect() + } + + /// Upload a multipart object so the data lands in real `part.*` shard files + /// rather than being inlined into `xl.meta` (inlined objects cannot be + /// corrupted shard-wise, and never exercise the streaming read path). + async fn put_multipart( + client: &Client, + bucket: &str, + key: &str, + parts: Vec>, + ) -> Result> { + let total_len = parts.iter().map(Vec::len).sum(); + + let create = client.create_multipart_upload().bucket(bucket).key(key).send().await?; + let upload_id = create.upload_id().ok_or("missing upload id")?.to_string(); + + let mut completed = Vec::with_capacity(parts.len()); + for (index, part_body) in parts.into_iter().enumerate() { + let part_number = (index + 1) as i32; + let uploaded = timeout( + OP_TIMEOUT, + client + .upload_part() + .bucket(bucket) + .key(key) + .upload_id(&upload_id) + .part_number(part_number) + .body(ByteStream::from(part_body)) + .send(), + ) + .await + .map_err(|_| format!("upload_part {part_number} timed out"))??; + completed.push( + CompletedPart::builder() + .part_number(part_number) + .e_tag(uploaded.e_tag().ok_or("missing part etag")?) + .build(), + ); + } + + timeout( + OP_TIMEOUT, + client + .complete_multipart_upload() + .bucket(bucket) + .key(key) + .upload_id(&upload_id) + .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed)).build()) + .send(), + ) + .await + .map_err(|_| "complete_multipart_upload timed out")??; + + Ok(total_len) + } + + /// rustfs#4784: reproduce the source-side fault the reporter kept hitting — + /// a GET that commits `200` + a full `Content-Length` and then cannot finish + /// the body — and assert the server log names the object, at the log level a + /// default deployment actually runs with. + #[tokio::test] + #[serial] + async fn midstream_get_failure_is_logged_with_the_object_at_default_log_level() -> TestResult { + init_logging(); + info!("rustfs#4784: a mid-stream GET failure must name its object in the source log"); + + let mut harness = DiskFaultHarness::new(4).await?; + + // Capture the child's stdout so the test can read what an operator would. + let log_path = format!("{}/server.log", harness.env.temp_dir); + harness.env.capture_log_path = Some(log_path.clone()); + + // Reproduce a DEFAULT deployment's logging, not the e2e harness's + // permissive `rustfs=info`: `DEFAULT_LOG_LEVEL` is `error`. Before the + // #4784 fix these failures were `warn!`, so a default deployment + // filtered them out entirely — which is why the reporter's source logs + // were empty. extra_env is applied after the harness's own RUST_LOG, so + // this wins. + harness.set_env("RUST_LOG", "error"); + harness.set_env("RUSTFS_OBS_LOGGER_LEVEL", "error"); + + harness.start_server().await?; + let client = harness.env.create_s3_client(); + + let bucket = "issue4784-source-read"; + client.create_bucket().bucket(bucket).send().await?; + + // Named after the reporter's restic index objects, which is where they + // saw the failures. + let key = "index/3b18542ab3af4c3d03f804c7a24173e7836ef7fa447b5d1e9d634f975cc51611"; + let expected_len = put_multipart( + &client, + bucket, + key, + vec![payload(5 * MIB, 71), payload(5 * MIB, 72), payload(5 * MIB, 73)], + ) + .await?; + + // Baseline: the object reads back completely before any corruption. + let baseline = timeout(OP_TIMEOUT, client.get_object().bucket(bucket).key(key).send()) + .await + .map_err(|_| "baseline GET timed out")?? + .body + .collect() + .await?; + assert_eq!(baseline.into_bytes().len(), expected_len, "baseline GET must return the whole object"); + + // Corrupt three of four shards in a 2+2 set: below the 2-shard read + // quorum. The corruption sits mid-file, so block 0 still reads clean — + // the server commits 200 + the full Content-Length and only then cannot + // reconstruct. That is the mid-stream window the reporter's downstream + // saw as `unexpected EOF`. + harness.corrupt_object_shard(0, bucket, key)?; + harness.corrupt_object_shard(1, bucket, key)?; + harness.corrupt_object_shard(2, bucket, key)?; + + let response = timeout(OP_TIMEOUT, client.get_object().bucket(bucket).key(key).send()) + .await + .map_err(|_| "degraded GET timed out")?; + + // Either outcome is functionally correct (that invariant belongs to + // degraded_read_eof_regression_test); this suite only needs the read to + // have failed so there is something to report. + let delivered = match response { + Err(err) => { + info!("degraded GET failed before the body: {err}"); + None + } + Ok(response) => match response.body.collect().await { + Ok(aggregated) => Some(aggregated.into_bytes().len()), + Err(err) => { + info!("degraded GET failed mid-body as expected: {err}"); + None + } + }, + }; + assert_ne!( + delivered, + Some(expected_len), + "the beyond-quorum read unexpectedly succeeded; this suite needs a failed read to have something to report" + ); + + // Give the child a moment to flush its stdout. + tokio::time::sleep(Duration::from_millis(500)).await; + let logged = std::fs::read_to_string(&log_path)?; + + let failure_lines: Vec<&str> = logged.lines().filter(|line| line.contains(STREAM_BODY_EVENT)).collect(); + + assert!( + !failure_lines.is_empty(), + "a mid-stream GET failure produced no `{STREAM_BODY_EVENT}` line at the default log level. \ + This is the #4784 blind spot: the failure was only counted in a metric, or logged below \ + `error` and filtered out. Captured log:\n{logged}" + ); + + // The identity is the whole point: a request_id alone cannot be resolved + // back to an object once the request is over. + assert!( + failure_lines.iter().any(|line| line.contains(key)), + "no `{STREAM_BODY_EVENT}` line named the failing object `{key}`, so the report is still \ + unactionable. Lines seen:\n{}", + failure_lines.join("\n") + ); + assert!( + failure_lines.iter().any(|line| line.contains(bucket)), + "no `{STREAM_BODY_EVENT}` line named the failing bucket `{bucket}`. Lines seen:\n{}", + failure_lines.join("\n") + ); + + info!( + "source-side evidence now present: {} stream-body failure line(s) naming the object", + failure_lines.len() + ); + for line in &failure_lines { + info!("operator-visible evidence: {line}"); + } + + Ok(()) + } +} diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index 21325815d..b3c1a175a 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -48,6 +48,11 @@ mod replacement_privileged_e2e_test; #[cfg(test)] mod degraded_read_eof_regression_test; +// rustfs#4784: a mid-stream GET failure must be reportable from the source +// server's log alone — naming the object, at the default log level. +#[cfg(test)] +mod get_stream_failure_observability_test; + // backlog#1183: GET codec-streaming fast path must be byte/header identical to // the legacy duplex path before its rollout gates can be flipped on by default. #[cfg(test)] diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 2c7405c62..2e3ec2767 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -1056,9 +1056,22 @@ pin_project! { remaining: usize, emitted: usize, expected: usize, + // Diagnostic-only identity for the body this stream is serving. Unset in + // unit tests that drive the stream over a bare reader; every production + // body carries it via `with_diagnostics`. + diagnostics: GetObjectReaderStreamDiagnostics, } } +/// Object identity carried alongside a streaming GET body purely so a +/// mid-stream failure names the object it happened on. +#[derive(Clone, Default)] +struct GetObjectReaderStreamDiagnostics { + bucket: String, + object: String, + request_id: String, +} + impl MemoryTrackedBytesStream { fn new( bytes: Bytes, @@ -1107,8 +1120,19 @@ where remaining, emitted: 0, expected: remaining, + diagnostics: GetObjectReaderStreamDiagnostics::default(), } } + + /// Attach the object identity a failed body should be reported against. + fn with_diagnostics(mut self, bucket: &str, object: &str, request_id: &str) -> Self { + self.diagnostics = GetObjectReaderStreamDiagnostics { + bucket: bucket.to_string(), + object: object.to_string(), + request_id: request_id.to_string(), + }; + self + } } impl futures::Stream for MemoryTrackedBytesStream { @@ -1569,12 +1593,29 @@ where *this.emitted, *this.remaining, ); - #[cfg(feature = "tracing-chunk-debug")] - tracing::error!( - emitted = *this.emitted, + // The inner GetObjectStreamingReader is what normally reports a + // short body, so reaching this arm means the reader signalled a + // clean EOF while this layer still owed bytes against an + // already-committed Content-Length. That disagreement is a data + // plane fault, not chunk noise: log it unconditionally so the + // truncated object is named in the operator's log rather than + // only in a metric counter (issue #4784). + error!( + event = EVENT_GET_OBJECT_STREAM_BODY, + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_OBJECT, + bucket = %this.diagnostics.bucket, + object = %this.diagnostics.object, + request_id = %this.diagnostics.request_id, + size_bucket = get_object_stream_size_bucket(*this.expected), expected = *this.expected, + emitted = *this.emitted, + remaining = *this.remaining, + strategy = this.strategy, + buffer_source = this.buffer_source, + state = "reader_stream_short_eof", error = %err, - "GetObject ReaderStream ended before expected length" + "GetObject reader stream ended before the committed content length" ); Poll::Ready(Some(Err(Box::new(err) as S3StdError))) } @@ -1590,10 +1631,17 @@ where *this.emitted, *this.remaining, ); + // Deliberately not logged at warn here: every production body + // wraps a GetObjectStreamingReader, and that layer already + // reports this same error once with `state = "read_failed"` and + // the object identity. A second unconditional line per failed + // GET would read as two distinct faults. The chunk-debug build + // still gets this layer's view of the same error. #[cfg(feature = "tracing-chunk-debug")] tracing::error!( emitted = *this.emitted, expected = *this.expected, + error_class = error_class, error = %err, "GetObject ReaderStream returned error" ); @@ -1646,8 +1694,12 @@ where struct GetObjectStreamingReader { inner: Option, - // request_id + optional content_range are only used for diagnostic correlation and - // failure bucketing; they do not alter stream behavior. + // bucket/object + request_id + optional content_range are only used for diagnostic + // correlation and failure bucketing; they do not alter stream behavior. The object + // identity is what turns a mid-stream failure into an actionable report: a request_id + // alone cannot tell an operator which object reads short (issue #4784). + bucket: String, + object: String, request_id: String, content_range: Option, expected: usize, @@ -1666,8 +1718,8 @@ impl GetObjectStreamingReader { #[allow(clippy::too_many_arguments)] fn new( inner: R, - _bucket: &str, - _key: &str, + bucket: &str, + key: &str, request_id: &str, content_range: Option, expected: usize, @@ -1677,6 +1729,8 @@ impl GetObjectStreamingReader { ) -> Self { Self { inner: Some(inner), + bucket: bucket.to_string(), + object: key.to_string(), request_id: request_id.to_string(), content_range, expected, @@ -1817,6 +1871,8 @@ impl GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, + bucket = %self.bucket, + object = %self.object, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), size_bucket = get_object_stream_size_bucket(self.expected), @@ -1853,6 +1909,8 @@ impl AsyncRead for GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, + bucket = %self.bucket, + object = %self.object, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), size_bucket = get_object_stream_size_bucket(self.expected), @@ -1871,10 +1929,12 @@ impl AsyncRead for GetObjectStreamingReader { self.timer = None; let failure_reason = Self::classify_read_error(&error); self.finish_err(); - warn!( + error!( event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, + bucket = %self.bucket, + object = %self.object, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), size_bucket = get_object_stream_size_bucket(self.expected), @@ -1916,6 +1976,8 @@ impl AsyncRead for GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, + bucket = %self.bucket, + object = %self.object, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), size_bucket = get_object_stream_size_bucket(self.expected), @@ -1956,10 +2018,12 @@ impl AsyncRead for GetObjectStreamingReader { self.begin_resume(error); continue; } - warn!( + error!( event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, + bucket = %self.bucket, + object = %self.object, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), size_bucket = get_object_stream_size_bucket(self.expected), @@ -1990,10 +2054,12 @@ impl AsyncRead for GetObjectStreamingReader { let failure_reason = Self::classify_read_error(&err); self.timer = None; self.finish_err(); - warn!( + error!( event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, + bucket = %self.bucket, + object = %self.object, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), size_bucket = get_object_stream_size_bucket(self.expected), @@ -2029,6 +2095,8 @@ impl Drop for GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, + bucket = %self.bucket, + object = %self.object, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), size_bucket = get_object_stream_size_bucket(self.expected), @@ -4303,7 +4371,8 @@ impl DefaultObjectUsecase { lifecycle, resume, ); - let stream = GetObjectReaderStream::new(reader, stream_buffer_size, expected, stream_strategy.as_str(), buffer_source); + let stream = GetObjectReaderStream::new(reader, stream_buffer_size, expected, stream_strategy.as_str(), buffer_source) + .with_diagnostics(bucket, key, request_id); let blob = StreamingBlob::new(stream); if let Some(handoff_start) = handoff_start { rustfs_io_metrics::record_get_object_response_handoff( @@ -16326,7 +16395,12 @@ mod tests { assert_eq!(body, vec![b'a'; 65]); } + // Serial with the capture test below: both drive the same short-EOF log + // callsite, and `tracing` caches callsite interest process-wide. Running + // this one concurrently on a thread with no subscriber re-caches that + // callsite as "never interested" and blinds the capture. #[tokio::test] + #[serial_test::serial] async fn get_object_reader_stream_errors_on_short_eof() { let stream = GetObjectReaderStream::new( std::io::Cursor::new(b"he".to_vec()), @@ -16349,6 +16423,134 @@ mod tests { ); } + /// Collects the structured fields of every event emitted while installed, + /// so a test can assert what an operator would actually read in the log + /// rather than only that an error value was returned. + type CapturedFieldMap = std::collections::HashMap; + type CapturedEventLog = Arc>>; + + struct CapturedEvents(CapturedEventLog); + + struct CapturedFields(CapturedFieldMap); + + impl tracing::field::Visit for CapturedFields { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0.insert(field.name().to_string(), format!("{value:?}")); + } + + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.0.insert(field.name().to_string(), value.to_string()); + } + } + + impl tracing_subscriber::Layer for CapturedEvents { + fn on_event(&self, event: &tracing::Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) { + let mut fields = CapturedFields(CapturedFieldMap::new()); + event.record(&mut fields); + self.0.lock().expect("captured events should not poison").push(fields.0); + } + } + + fn capture_events() -> (CapturedEventLog, tracing::subscriber::DefaultGuard) { + use tracing_subscriber::{Registry, prelude::*}; + + let captured = Arc::new(Mutex::new(Vec::new())); + let subscriber = Registry::default().with(CapturedEvents(Arc::clone(&captured))); + let guard = tracing::subscriber::set_default(subscriber); + // `tracing` caches per-callsite interest process-wide, so a subscriber + // installed by a test running in parallel can leave the log sites below + // cached as "never interested" and this capture would silently see + // nothing. Force the callsites to re-ask the subscriber we just + // installed. + tracing::callsite::rebuild_interest_cache(); + (captured, guard) + } + + fn find_stream_body_event(captured: &CapturedEventLog, state: &str) -> CapturedFieldMap { + let events = captured.lock().expect("captured events should not poison"); + events + .iter() + .find(|fields| fields.get("state").is_some_and(|value| value == state)) + .unwrap_or_else(|| { + panic!( + "a `{state}` streaming body failure must be logged, not only counted in a metric. \ + Captured {} event(s): {:?}", + events.len(), + events + ) + }) + .clone() + } + + /// rustfs#4784: a GET body that ends short of its committed Content-Length + /// is the fault that breaks every downstream copier (replication, site + /// replication, `rclone sync`), yet this layer only fed a metric counter — + /// its log line was compiled out unless the `tracing-chunk-debug` feature + /// was on, so operators saw nothing on the source side. + #[tokio::test] + #[serial_test::serial] + async fn get_object_reader_stream_short_eof_names_the_object() { + let (captured, _guard) = capture_events(); + + let stream = GetObjectReaderStream::new( + std::io::Cursor::new(b"he".to_vec()), + 64, + 5, + GetObjectStreamStrategy::Standard.as_str(), + GET_READER_STREAM_BUFFER_SOURCE_SELECTED, + ) + .with_diagnostics("restic-paperless", "index/41b5a4c2344edb90", "req-reader-stream-short-eof"); + + stream + .collect::>() + .await + .into_iter() + .collect::, _>>() + .expect_err("short reader should fail the streaming body"); + + let event = find_stream_body_event(&captured, "reader_stream_short_eof"); + assert_eq!(event.get("bucket").map(String::as_str), Some("restic-paperless")); + assert_eq!(event.get("object").map(String::as_str), Some("index/41b5a4c2344edb90")); + assert_eq!(event.get("request_id").map(String::as_str), Some("req-reader-stream-short-eof")); + assert_eq!(event.get("expected").map(String::as_str), Some("5")); + assert_eq!(event.get("emitted").map(String::as_str), Some("2")); + assert_eq!(event.get("remaining").map(String::as_str), Some("3")); + } + + /// The inner reader already logged mid-stream failures, but only under a + /// request_id — which cannot be resolved back to an object once the request + /// is gone. Without the identity the report in #4784 was unactionable. + #[tokio::test] + #[serial_test::serial] + async fn get_object_streaming_reader_short_eof_names_the_object() { + use tokio::io::AsyncReadExt; + + let (captured, _guard) = capture_events(); + + let mut reader = GetObjectStreamingReader::new( + std::io::Cursor::new(b"short".to_vec()), + "restic-paperless", + "index/41b5a4c2344edb90", + "req-streaming-short-eof", + None, + 10, + Duration::ZERO, + GetObjectBodyLifecycle::tracked(GetObjectGuard::new()), + None, + ); + + let mut out = Vec::new(); + reader + .read_to_end(&mut out) + .await + .expect_err("short body under a larger Content-Length must fail the stream"); + + let event = find_stream_body_event(&captured, "short_eof"); + assert_eq!(event.get("bucket").map(String::as_str), Some("restic-paperless")); + assert_eq!(event.get("object").map(String::as_str), Some("index/41b5a4c2344edb90")); + assert_eq!(event.get("request_id").map(String::as_str), Some("req-streaming-short-eof")); + } + #[test] fn get_object_stream_failure_labels_are_low_cardinality() { assert_eq!(get_object_stream_failure_reason("short_eof"), GET_STREAMING_BODY_FAILURE_REASON_SHORT_EOF);