diff --git a/CHANGELOG.md b/CHANGELOG.md index 486dd457..51b15933 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -178,6 +178,13 @@ the accept loop was busy, had SYNs dropped: those clients waited 1, 3, 7 s for retransmits and could pass the gateway's 5 s connect timeout. The held job-read test hit this in CI (2995 of 3000 reads held at its deadline). +- **A ucloud-noded log chunk's `job` record counts the bytes the chunk was read + from.** The record sized the running job's log, then the chunk opened it + again, so output written in between could come back as data past the + record's `stdout_bytes`/`stderr_bytes` (and its `cursor`), which the gateway + cached as the job's state. Held reads never returned early; the held + job-read test hit this in CI (a logs read answered with the byte it waited + for and `stdout_bytes` 0). - **Buffered worker RPCs never reuse a connection the gateway asked to close** (RFC 9112 9.6). aiohttp pooled such a connection unless the answer also said `Connection: close`. ucloud-noded and the Python agent always say so, so diff --git a/runtime/noded/src/jobs/files.rs b/runtime/noded/src/jobs/files.rs index d62a4e11..f7ef9077 100644 --- a/runtime/noded/src/jobs/files.rs +++ b/runtime/noded/src/jobs/files.rs @@ -394,6 +394,22 @@ impl JobRecord { if stream == "stderr" { self.state.stderr_bytes } else { self.state.stdout_bytes } } + /// The record a log chunk carries. The record counted the log just + /// before the chunk's own open, and the job may have written in between; + /// while it runs, the stream's count is the size the chunk was read + /// against, so the data never runs past the record that comes with it. + /// Once terminal, the persisted counts stand: the logs no longer grow. + pub fn read_against(mut self, stream: &str, size: u64) -> JobRecord { + if !self.terminal() { + if stream == "stderr" { + self.state.stderr_bytes = size; + } else { + self.state.stdout_bytes = size; + } + } + self + } + /// `ManagedProcessRecord.to_dict()`, in its key order, plus `cursor`. pub fn to_json(&self) -> Value { let state = &self.state; @@ -490,6 +506,24 @@ mod tests { ); } + #[test] + fn a_chunks_record_counts_the_log_it_was_read_from() { + let record = |state: &str| JobRecord { + sandbox_id: "sb".into(), + sandbox_generation: 3, + state: JobState::parse(state.as_bytes()).unwrap(), + updated_at: "2026-10-11T20:00:02.000001+00:00".into(), + }; + // The job wrote one byte between the record's read and the chunk's. + let running = record(GO_RUNNING).read_against("stdout", 1); + assert_eq!((running.state.stdout_bytes, running.state.stderr_bytes, running.cursor()), (1, 0, "2.1.0".into())); + let running = record(GO_RUNNING).read_against("stderr", 4); + assert_eq!((running.state.stdout_bytes, running.state.stderr_bytes), (0, 4)); + // Terminal: the persisted counts stand. + let exited = GO_RUNNING.replace(r#""state":"running""#, r#""state":"exited","exit_code":0"#).replace(r#""stdout_bytes":0"#, r#""stdout_bytes":7"#); + assert_eq!(record(&exited).read_against("stdout", 9).state.stdout_bytes, 7); + } + #[test] fn base64_matches_python() { // python3 -c 'import base64; print(base64.b64encode(b"..."))' diff --git a/runtime/noded/src/jobs/mod.rs b/runtime/noded/src/jobs/mod.rs index 78d9f208..d018551f 100644 --- a/runtime/noded/src/jobs/mod.rs +++ b/runtime/noded/src/jobs/mod.rs @@ -324,11 +324,14 @@ fn read_pass( return Err(PassError::NotFound); } let ReadKind::Logs { stream, offset, limit } = kind else { return Ok(Pass { record, chunk: None }) }; - let chunk = match dir.log(stream).map_err(PassError::Read)? { + let (record, chunk) = match dir.log(stream).map_err(PassError::Read)? { // The supervisor's answer for a log it never opened. - None => (*offset, Vec::new(), *offset, false), + None => (record, (*offset, Vec::new(), *offset, false)), Some(log) if *offset > log.size => return Err(PassError::OffsetBeyond), - Some(log) => (*offset, log.read(*offset, *limit).map_err(|error| PassError::Read(ReadError::Io(error)))?, log.size, true), + Some(log) => { + let data = log.read(*offset, *limit).map_err(|error| PassError::Read(ReadError::Io(error)))?; + (record.read_against(stream, log.size), (*offset, data, log.size, true)) + } }; Ok(Pass { record, chunk: Some(chunk) }) })();