From 02cef2870acd1dfa9dd485b7e9b083ccc952b958 Mon Sep 17 00:00:00 2001 From: Rasmus Larsen Date: Sat, 24 Oct 2026 12:02:11 +0000 Subject: [PATCH] noded: a log chunk's job record counts the log it was read from thousands_of_held_reads_wait_on_one_watch_and_disconnects_free_them failed in CI with a read answering stdout_bytes 0 after the one-byte append. Held reads do not return early: a logs read wakes on the append, re-reads and returns because its chunk is non-empty. But read_pass built the record first (StateDir::record sizes the running job's logs by opening them) and then opened the stream's log again for the chunk. An append landing between the two opens (the appender's create and write straddle the woken re-reads) produced a chunk with the new byte next to a record that still said 0. Status reads take their cursor and counts from one record, so they cannot show this. Reproduced locally: under load the instrumented test showed a logs read answering data "eA==", next_offset 1, job.stdout_bytes 0; a scratch route test reading logs while a thread appends saw 321 of 3000 chunks run past their record before this change and none after. The record a chunk carries now takes the stream's count from the size the chunk was read against (JobRecord::read_against) while the job runs; once terminal the persisted counts stand. The state is still read before the log, so a terminal record never undercounts a final log. Production impact: no spurious early returns for clients long-polling status or logs. A running job's logs response could carry a job record (stdout_bytes/stderr_bytes, cursor) older than its own data, and the gateway cached that record as the job's state; the counts are a lower bound, so cached answers during lifecycle transitions were at worst stale, not wrong. The SDK does not read the record from log chunks. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 +++++++ runtime/noded/src/jobs/files.rs | 34 +++++++++++++++++++++++++++++++++ runtime/noded/src/jobs/mod.rs | 9 ++++++--- 3 files changed, 47 insertions(+), 3 deletions(-) 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) }) })();