diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml index 43dacaf1ea0..232952a3a33 100644 --- a/.github/workflows/main.yaml +++ b/.github/workflows/main.yaml @@ -1,17 +1,23 @@ name: CI # Controls when the workflow will run. +# +# The only live job here (integration-tests) builds the trace_machina/nativelink +# OCI image from the docker-compose Dockerfile and runs run_integration_tests.sh. +# That is a per-PR OCI build — the same class of heavy work #image-off-pr moved +# off PRs for image.yaml and #2877 moved off PRs for the nix integration job. +# This lane fell in the crack between those two, so defer it the same way: +# nightly cross-check + on every merge to main, not on every PR. Per-PR +# build/test signal is carried by the Nix "Bazel Dev" lane. on: push: branches: [ main ] paths-ignore: - '.github/styles/**' - 'web/**' - pull_request: - branches: [ main ] - paths-ignore: - - '.github/styles/**' - - 'web/**' + schedule: + - cron: '0 7 * * *' # 07:00 UTC daily (off-peak) + workflow_dispatch: permissions: read-all diff --git a/nativelink-store/src/grpc_store.rs b/nativelink-store/src/grpc_store.rs index 451ad956cd2..79ae9e09cfa 100644 --- a/nativelink-store/src/grpc_store.rs +++ b/nativelink-store/src/grpc_store.rs @@ -2087,6 +2087,17 @@ impl StoreDriver for GrpcStore { return Some((RetryResult::Err(err), local_state)); } local_state.read_offset += length; + // A bounded read must also shrink its remaining limit by the + // bytes already forwarded. `read_limit` is a count measured + // from `read_offset` (REAPI ByteStream semantics), so a + // mid-stream retry that advances `read_offset` without + // decrementing `read_limit` re-requests the *original* window + // size from the new offset and over-reads by exactly the + // bytes already delivered. `read_limit == 0` means "read to + // end" and must stay unbounded across resumes. + if local_state.read_limit != 0 { + local_state.read_limit = local_state.read_limit.saturating_sub(length); + } } })) .await diff --git a/nativelink-store/tests/grpc_store_test.rs b/nativelink-store/tests/grpc_store_test.rs index a11c6fb518e..156dde6e6fa 100644 --- a/nativelink-store/tests/grpc_store_test.rs +++ b/nativelink-store/tests/grpc_store_test.rs @@ -123,6 +123,11 @@ struct FakeStreamServer { fail_mid_upload: Arc>, failure_code: tonic::Code, require_zero_start: bool, + /// Full byte payload the fake `read` serves (sliced from `read_offset`). + read_payload: Arc>, + /// `Some(n)`: the FIRST `read` serves `n` bytes then a retryable error, + /// forcing the client to resume; later reads serve the remainder. + first_read_partial: Option, } impl FakeStreamServer { @@ -136,6 +141,8 @@ impl FakeStreamServer { fail_mid_upload: Arc::new(Mutex::new(0)), failure_code: tonic::Code::Unavailable, require_zero_start: false, + read_payload: Arc::new(RAW_INPUT.as_bytes().to_vec()), + first_read_partial: None, } } @@ -149,10 +156,6 @@ impl FakeStreamServer { type ReadStream = Pin> + Send + 'static>>; -struct ReaderState { - responded: bool, -} - #[tonic::async_trait] impl ByteStream for FakeStreamServer { type ReadStream = ReadStream; @@ -177,19 +180,39 @@ impl ByteStream for FakeStreamServer { } } let read_request = grpc_request.into_inner(); - self.read_requests.lock().await.push(ReadRequestHolder { - request: read_request, - metadata: request_metadata, - }); - - let folded = unfold(ReaderState { responded: false }, async move |state| { - if state.responded { - return None; + let offset = usize::try_from(read_request.read_offset).unwrap_or(0); + let this_read = { + let mut reqs = self.read_requests.lock().await; + reqs.push(ReadRequestHolder { + request: read_request, + metadata: request_metadata, + }); + reqs.len() + }; + let payload = self.read_payload.clone(); + let partial = self.first_read_partial; + + // Step machine: 0 => emit a data chunk, 1 => emit a retryable error + // (only on a partial first read), 2 => EOF. Default behaviour (no + // `first_read_partial`) is one chunk from `read_offset` then EOF. + let folded = unfold((0u8, payload), move |(step, payload)| async move { + match step { + 0 => { + if let (1, Some(f)) = (this_read, partial) { + let start = offset.min(payload.len()); + let end = (offset + f).min(payload.len()); + let chunk = payload[start..end].to_vec(); + return Some((Ok(ReadResponse { data: chunk.into() }), (1u8, payload))); + } + let chunk = payload.get(offset..).unwrap_or(&[]).to_vec(); + Some((Ok(ReadResponse { data: chunk.into() }), (2u8, payload))) + } + 1 => Some(( + Err(Status::unavailable("simulated mid-stream drop")), + (2u8, payload), + )), + _ => None, } - let response = ReadResponse { - data: RAW_INPUT.as_bytes().into(), - }; - Some((Ok(response), ReaderState { responded: true })) }); Ok(Response::new(Box::pin(folded))) } @@ -1108,3 +1131,67 @@ async fn incomplete_oneshot_cas_upload_never_sends_remote_write() -> Result<(), assert!(server.write_requests.lock().await.is_empty()); Ok(()) } + +/// A 10-byte payload, longer than the partial-forward below, so a bounded read +/// has a meaningful remaining window to carry across a resume. +const RESUME_PAYLOAD: &[u8] = b"0123456789"; + +/// Regression for the `GrpcStore` bounded-read resume over-read: a mid-stream +/// retry advanced `read_offset` but left `read_limit` at its original value, so +/// the resume re-requested the full window from the new offset and over-read by +/// exactly the bytes already forwarded. The deterministic witness is the second +/// (resume) `ReadRequest` on the wire: its `read_limit` must shrink by the +/// forwarded byte count. Fails without the fix by construction (limit stays L). +#[nativelink_test] +async fn bounded_read_resume_shrinks_read_limit() -> Result<(), Error> { + const FORWARDED: usize = 3; + let requested_len = RESUME_PAYLOAD.len(); + + let server = FakeStreamServer { + read_payload: Arc::new(RESUME_PAYLOAD.to_vec()), + first_read_partial: Some(FORWARDED), + ..FakeStreamServer::new() + }; + let (server, port) = spawn_bytestream_server(server).await; + let mut spec = test_spec(format!("http://localhost:{port}"), false); + spec.retry.max_retries = 1; + let store = GrpcStore::new(&spec)?; + let digest = DigestInfo::try_new(VALID_HASH, requested_len).unwrap(); + + let (tx, mut rx) = make_buf_channel_pair(); + let get_fut = store.get_part(digest, tx, 0, Some(u64::try_from(requested_len).unwrap())); + let drain_fut = async move { + let mut total = 0usize; + loop { + match rx.recv().await { + Ok(chunk) if !chunk.is_empty() => total += chunk.len(), + _ => break, + } + } + total + }; + let (res, total) = futures::join!(get_fut, drain_fut); + res?; + assert_eq!( + total, requested_len, + "writer must receive exactly the requested bytes across the resume" + ); + + let reqs = server.read_requests.lock().await; + assert!( + reqs.len() >= 2, + "expected a resume read, saw {}", + reqs.len() + ); + let limit = i64::try_from(requested_len).unwrap(); + let fwd = i64::try_from(FORWARDED).unwrap(); + assert_eq!(reqs[0].request.read_offset, 0); + assert_eq!(reqs[0].request.read_limit, limit); + assert_eq!(reqs[1].request.read_offset, fwd); + assert_eq!( + reqs[1].request.read_limit, + limit - fwd, + "resume read_limit must shrink by bytes already forwarded (over-read bug)" + ); + Ok(()) +}