diff --git a/bottlecap/src/extension/telemetry/listener.rs b/bottlecap/src/extension/telemetry/listener.rs index 56e96647d..488e12c19 100644 --- a/bottlecap/src/extension/telemetry/listener.rs +++ b/bottlecap/src/extension/telemetry/listener.rs @@ -2,7 +2,7 @@ use crate::{ event_bus, extension::telemetry::{ events::TelemetryEvent, - stitch::{FragmentBuffer, Stitch}, + stitch::{FragmentBuffer, Stitch, rejoin_split_that_parsed}, }, http::{extract_request_body, handler_not_found}, }; @@ -122,11 +122,11 @@ impl TelemetryListener { } }; - let mut telemetry_events: Vec = match serde_json::from_slice(&body) { - Ok(events) => events, - // The Telemetry API splits an oversized record across two POSTs, and neither half - // parses alone. See `stitch`. - Err(e) => match state.fragments.stitch(&body, &e) { + let mut telemetry_events = match serde_json::from_slice::>(&body) { + Ok(events) => rejoin_split_that_parsed(&body, events.len()).unwrap_or(events), + // Lambda cuts a long log line into pieces, and a payload that holds a piece does not + // parse. See `stitch`. + Err(e) => match state.fragments.stitch(&body) { Stitch::Complete(events) => { debug!( "TELEMETRY API | Reassembled a split payload, recovered {} events", @@ -185,6 +185,19 @@ mod tests { .expect("failed to build request") } + /// A record cut after a full 262,118-byte piece, inside the number `23`, so its payload still + /// parses: the next piece's envelope reads as an item of the array. Returns the payload and + /// the whole record. + fn split_that_parses() -> (String, String) { + let pad = "x".repeat(262_118 - r#"{"items":[{"pad":""},{"id":2"#.len()); + let first = format!(r#"{{"items":[{{"pad":"{pad}"}},{{"id":2"#); + let rest = r#"3,"ok":true}],"n":5}"#; + let body = format!( + r#"[{{"time":"2026-09-03T14:29:52.929Z","type":"function","record":{first}}},{{"time":"2026-09-03T14:29:52.930Z","type":"function","record":{rest}}},{{"time":"2026-09-03T14:29:52.950Z","type":"platform.runtimeDone","record":{{"requestId":"abc123","status":"success"}}}}]"# + ); + (body, format!("{first}{rest}")) + } + #[tokio::test] #[allow(clippy::unwrap_used)] async fn test_handle() { @@ -243,4 +256,22 @@ mod tests { TelemetryRecord::PlatformRuntimeDone { .. } )); } + + /// A seam that happens to be valid JSON still has its pieces rejoined. + #[tokio::test] + async fn test_handle_split_payload_that_parses() { + let (body, record) = split_that_parses(); + serde_json::from_str::>(&body).expect("the seam parses"); + + let (tx, mut rx) = tokio::sync::mpsc::channel(2); + let response = + TelemetryListener::handle(axum::extract::State(state(tx)), post(&body)).await; + assert_eq!(response.status(), StatusCode::OK); + + let record = serde_json::from_str(&record).expect("valid record"); + assert_eq!( + rx.try_recv().expect("function record").record, + TelemetryRecord::Function(record) + ); + } } diff --git a/bottlecap/src/extension/telemetry/stitch.rs b/bottlecap/src/extension/telemetry/stitch.rs index a375584c2..4a7a0f7d0 100644 --- a/bottlecap/src/extension/telemetry/stitch.rs +++ b/bottlecap/src/extension/telemetry/stitch.rs @@ -1,25 +1,30 @@ -//! Reassembles telemetry payloads that the Telemetry API split across two POSTs. +//! Reassembles telemetry payloads holding a log record that Lambda cut into pieces. //! -//! A record larger than the subscription's `maxBytes` is cut mid-value. The rest of it — plus -//! the rest of the batch — arrives in the next POST, which repeats the cut record's envelope -//! ahead of the resumed bytes. Neither half parses on its own: +//! A log record longer than about 256 KiB is cut mid-value into pieces, each an event of its +//! own. When a payload ends on a cut piece, the rest of it — plus the rest of the batch — +//! arrives in the next POST, under the next piece's envelope. Neither half parses on its own: //! //! ```text //! POST 1 [{"time":"..929Z","type":"function","record":{"message":"iVBORw0KGgoAAA}] //! `------------- envelope --------------------'`--- cut here ---'`framing' //! -//! POST 2 [{"time":"..929Z","type":"function","record":QICAgIfAhkiAAA"}},{..},{..runtimeDone..}] -//! `------- the same envelope, repeated -------'`-- resumed --'`- rest of the batch -' +//! POST 2 [{"time":"..930Z","type":"function","record":QICAgIfAhkiAAA"}},{..},{..runtimeDone..}] +//! `------ the next piece's envelope ----------'`-- resumed --'`- rest of the batch -' //! ``` //! -//! So the two are joined by dropping POST 1's framing and POST 2's repeated envelope. That -//! byte-identical envelope, which carries the cut record's timestamp, is the only thing -//! pairing them: the API sends no sequence number. +//! So the two are joined by dropping POST 1's framing and POST 2's envelope. Each piece carries +//! its own timestamp, so nothing pairs them but the joined bytes parsing. +//! +//! Pieces can also arrive next to each other in one POST. The seam between two such pieces is the +//! cut piece's closing `}` followed by the next piece's envelope, and it is dropped the same way. +//! A seam has the same bytes as the boundary between two whole events. It is told apart by the +//! record value before it, which was cut short and so does not parse alone. //! //! Recovering the batch matters beyond the log line itself. `platform.runtimeDone` lands in //! the second half, and the on-demand loop waits for it before calling `/next` — so dropping //! the batch holds the invocation open until Lambda times out the sandbox. +use serde::de::IgnoredAny; use serde_json::error::Category; use std::{ sync::{Arc, Mutex, PoisonError}, @@ -45,6 +50,25 @@ const RECORD_KEY: &[u8] = b"\"record\":"; /// record's value short, so a fragment ends with framing that belongs to neither half. const FRAMING: &[u8] = b"}]"; +/// Opens every event's envelope. +const EVENT_START: &[u8] = b"{\"time\":\""; + +/// How a piece's envelope ends: only log lines are long enough to be split. +const PIECE_TYPES: &[&[u8]] = &[ + b"\"type\":\"function\",\"record\":", + b"\"type\":\"extension\",\"record\":", +]; + +/// Room for an envelope's timestamp and type, bounding the search for its `record` key. +const MAX_ENVELOPE_BYTES: usize = 128; + +/// Bounds the re-parsing of one record: a 4 MiB payload of 256 KiB pieces has 16 seams. +const MAX_SEAMS: usize = 64; + +/// Lambda cuts a long log line at 262,118 bytes (256 KiB less 26 bytes of per-event overhead), so +/// a split record's first piece is at least this long, with a margin. +const MIN_FIRST_PIECE_BYTES: usize = 255 * 1024; + /// What came of pairing an unparseable body with a held fragment. #[derive(Debug)] pub(crate) enum Stitch { @@ -64,10 +88,7 @@ pub(crate) struct FragmentBuffer { impl FragmentBuffer { /// Joins `body` onto the held fragment, or holds `body` if it opens a split payload. - /// - /// `error` is the failure `body` produced on its own; it is how a payload cut mid-record - /// is told apart from one we simply can't interpret. - pub(crate) fn stitch(&self, body: &[u8], error: &serde_json::Error) -> Stitch { + pub(crate) fn stitch(&self, body: &[u8]) -> Stitch { let mut slot = self.held.lock().unwrap_or_else(PoisonError::into_inner); if let Some(stale) = slot.take_if(|held| held.received.elapsed() > FRAGMENT_TTL) { @@ -78,37 +99,56 @@ impl FragmentBuffer { } if let Some(head) = slot.take() { - if let Some(resumed) = head.resumed_bytes(body) { - let joined = head.join(resumed); - return match serde_json::from_slice(&joined) { - Ok(events) => Stitch::Complete(events), - // A record can be cut more than once, so keep accumulating. - Err(e) => hold_or_discard(&mut slot, joined, &e), - }; + let held_len = head.body.len(); + if let Some(resumed) = Fragment::resumed_bytes(body) { + // A record can be cut more than once, so keep accumulating. + match parse_or_hold(&mut slot, head.join(resumed)) { + Stitch::Discarded => {} + stitched => return stitched, + } } debug!( - "TELEMETRY API | Dropping {} held bytes, the next payload does not continue it", - head.body.len() + "TELEMETRY API | Dropping {held_len} held bytes, the next payload does not continue it" ); } - hold_or_discard(&mut slot, body.to_vec(), error) + parse_or_hold(&mut slot, body.to_vec()) } } -/// Holds `body` for its continuation, or reports that nothing can be recovered from it. -fn hold_or_discard( - slot: &mut Option, - body: Vec, - error: &serde_json::Error, -) -> Stitch { - *slot = Fragment::from_cut_payload(body, error); - if slot.is_some() { - Stitch::Pending - } else { - Stitch::Discarded +/// Parses `body` once its seams are dropped, or holds it if it ends on a cut piece. +fn parse_or_hold(slot: &mut Option, mut body: Vec) -> Stitch { + match serde_json::from_slice(&body) { + Ok(events) => return Stitch::Complete(events), + // Valid JSON our types don't accept arrived whole; no seam explains it. + Err(e) if e.classify() == Category::Data => return Stitch::Discarded, + Err(_) => {} + } + + if drop_seams(&mut body, 0) && body.len() <= MAX_FRAGMENT_BYTES { + *slot = Some(Fragment { + body, + received: Instant::now(), + }); + return Stitch::Pending; + } + serde_json::from_slice(&body).map_or(Stitch::Discarded, Stitch::Complete) +} + +/// Rejoins a split record whose payload parsed anyway: a cut inside a number in an array of +/// objects leaves valid JSON, with the next piece's envelope read as part of the record. +pub(crate) fn rejoin_split_that_parsed(body: &[u8], parsed: usize) -> Option> { + if body.len() < MIN_FIRST_PIECE_BYTES || split_events(body)?.len() <= parsed { + return None; + } + let mut rebuilt = body.to_vec(); + drop_seams(&mut rebuilt, MIN_FIRST_PIECE_BYTES); + // Dropping a seam shortens the payload; if it didn't, there was nothing to rejoin. + if rebuilt.len() == body.len() { + return None; } + serde_json::from_slice(&rebuilt).ok() } /// The leading half of a split payload. @@ -118,35 +158,17 @@ struct Fragment { } impl Fragment { - /// A fragment, if `body` is the leading half of a split payload: an array that ran out of - /// input inside its last record's value. Any other parse failure won't be fixed by - /// joining, and holding such a body would poison the next stitch. - fn from_cut_payload(body: Vec, error: &serde_json::Error) -> Option { - if error.classify() != Category::Eof - || body.len() > MAX_FRAGMENT_BYTES - || body.first() != Some(&b'[') - { - return None; - } - - Some(Self { - body, - received: Instant::now(), - }) - } - /// The bytes that resume this fragment, if `body` is its continuation. /// - /// A continuation opens with the cut record's envelope, so its `record` key is the first + /// A continuation opens with the next piece's envelope, so its `record` key is the first /// one in the payload — anything the customer nested sits inside the value that follows. - /// Finding the same envelope in the fragment is what pairs the two. - fn resumed_bytes<'a>(&self, body: &'a [u8]) -> Option<&'a [u8]> { - let payload = body.strip_prefix(b"[")?; - let value_start = find(payload, RECORD_KEY)? + RECORD_KEY.len(); - let envelope = payload.get(..value_start)?; - - find(&self.body, envelope)?; - payload.get(value_start..) + fn resumed_bytes(body: &[u8]) -> Option<&[u8]> { + let event = body.strip_prefix(b"[")?; + // Only a log piece continues a fragment; any other type is a whole event. + let (len, Some(_)) = envelope(event)? else { + return None; + }; + event.get(len..) } /// Joins the resumed bytes on, dropping the framing: left in place it would land inside @@ -167,6 +189,162 @@ fn find(haystack: &[u8], needle: &[u8]) -> Option { .position(|window| window == needle) } +/// Joins the pieces of each cut record in `body`, and returns whether it ends on a cut piece to +/// hold for the rest. +/// +/// A cut record continues in the next event of its type. A whole event of another type in +/// between, such as an extension log line, is moved ahead of it. A record only counts as cut if +/// its first piece is at least `min_first_piece` bytes. +fn drop_seams(body: &mut Vec, min_first_piece: usize) -> bool { + let Some(events) = split_events(body) else { + return false; + }; + + let mut joined = Vec::with_capacity(body.len()); + // A record cut short, and its pieces joined so far. + let mut cut: Option<(Event, Vec)> = None; + let mut seam_count = 0; + for event in events { + match &mut cut { + None => match parse_record(event.value) { + Record::Whole => push_event(&mut joined, event.envelope, event.value), + Record::Cut if event.kind.is_some() && event.value.len() >= min_first_piece => { + cut = Some((event, event.value.to_vec())); + } + _ => return false, + }, + Some((record, _)) if event.kind != record.kind => { + if !matches!(parse_record(event.value), Record::Whole) { + return false; + } + push_event(&mut joined, event.envelope, event.value); + } + Some((record, value)) => { + if seam_count == MAX_SEAMS { + return false; + } + seam_count += 1; + value.extend_from_slice(event.value); + match parse_record(value) { + Record::Whole => { + push_event(&mut joined, record.envelope, value); + cut = None; + } + Record::Cut => {} + Record::Broken => return false, + } + } + } + } + + let ends_cut = cut.is_some(); + if let Some((record, value)) = cut { + push_event(&mut joined, record.envelope, &value); + } + joined.push(b']'); + *body = joined; + ends_cut +} + +/// A payload's event: its envelope through `"record":`, its record value, and its type if it is +/// a log piece. +#[derive(Clone, Copy)] +struct Event<'a> { + envelope: &'a [u8], + value: &'a [u8], + kind: Option<&'static [u8]>, +} + +/// Splits a payload into its events. +fn split_events(body: &[u8]) -> Option>> { + let mut rest = body.strip_prefix(b"[")?.strip_suffix(FRAMING)?; + let mut events = Vec::new(); + loop { + let (len, kind) = envelope(rest)?; + let (envelope, tail) = rest.split_at(len); + let Some((value, next)) = split_at_next_event(tail) else { + events.push(Event { + envelope, + value: tail, + kind, + }); + return Some(events); + }; + events.push(Event { + envelope, + value, + kind, + }); + rest = next; + } +} + +/// Splits `bytes` at the first `},` that ends an event, which is one followed by an envelope. +fn split_at_next_event(bytes: &[u8]) -> Option<(&[u8], &[u8])> { + (0..bytes.len()).find_map(|i| { + let next = bytes[i..].strip_prefix(b"},")?; + envelope(next).map(|_| (&bytes[..i], next)) + }) +} + +/// Appends an event to a payload being rebuilt, opening the array before the first one. +fn push_event(payload: &mut Vec, envelope: &[u8], value: &[u8]) { + payload.push(if payload.is_empty() { b'[' } else { b',' }); + payload.extend_from_slice(envelope); + payload.extend_from_slice(value); + payload.push(b'}'); +} + +/// A record value taken on its own. +enum Record { + Whole, + /// Valid JSON until it runs out, so the next piece can complete it. + Cut, + Broken, +} + +fn parse_record(value: &[u8]) -> Record { + match serde_json::from_slice::(value) { + Ok(_) => Record::Whole, + // A number cut after its `-`, `.` or `e` is reported as invalid at its last byte rather + // than as running out. + Err(e) if e.is_eof() || offset(value, &e) + 1 >= value.len() => Record::Cut, + Err(_) => Record::Broken, + } +} + +/// Length of an event's envelope, through `"record":`, and its type if it is a log piece. +fn envelope(event: &[u8]) -> Option<(usize, Option<&'static [u8]>)> { + let window = &event[..event.len().min(MAX_ENVELOPE_BYTES)]; + let fields = window.strip_prefix(EVENT_START)?; + let len = find(fields, RECORD_KEY)? + RECORD_KEY.len(); + + // The envelope holds only the timestamp and type; a brace means the match ran into a value. + let fields = &fields[..len]; + if fields.iter().any(|b| matches!(b, b'{' | b'}')) { + return None; + } + let kind = PIECE_TYPES + .iter() + .copied() + .find(|tail| fields.ends_with(tail)); + Some((EVENT_START.len() + len, kind)) +} + +/// The byte serde's one-based line and column point at. +fn offset(body: &[u8], error: &serde_json::Error) -> usize { + let line_start = match error.line() { + 0 | 1 => 0, + line => body + .iter() + .enumerate() + .filter(|&(_, &b)| b == b'\n') + .nth(line - 2) + .map_or(0, |(i, _)| i + 1), + }; + line_start + error.column().saturating_sub(1) +} + /// The two halves of a real split payload, trimmed to the bytes that matter. #[cfg(test)] pub(crate) mod fixtures { @@ -174,7 +352,7 @@ pub(crate) mod fixtures { pub(crate) const HEAD: &str = r#"[{"time":"2026-09-03T14:29:52.929Z","type":"function","record":{"message":"AAAA}]"#; - /// The continuation: the same envelope, the resumed bytes, then the rest of the batch. + /// The continuation: the next piece's envelope, the resumed bytes, then the rest of the batch. pub(crate) const TAIL: &str = r#"[{"time":"2026-09-03T14:29:52.929Z","type":"function","record":BBBB"}},{"time":"2026-09-03T14:29:52.930Z","type":"platform.runtimeDone","record":{"requestId":"abc123","status":"success","metrics":{"durationMs":18.074,"producedBytes":329814}}}]"#; } @@ -186,9 +364,9 @@ mod tests { /// Mirrors the handler: a body only reaches the buffer once it has failed to parse. fn stitch(fragments: &FragmentBuffer, body: &str) -> Stitch { - let error = serde_json::from_slice::>(body.as_bytes()) + serde_json::from_slice::>(body.as_bytes()) .expect_err("fixture must not parse on its own"); - fragments.stitch(body.as_bytes(), &error) + fragments.stitch(body.as_bytes()) } /// The events of a stitch that should have completed, reporting what came back if it did not. @@ -234,8 +412,7 @@ mod tests { fn joins_a_payload_cut_after_whole_records() { let fragments = FragmentBuffer::default(); - // The cut record is the last of several, so the envelope the continuation repeats is - // in the middle of the fragment. + // The cut record is the last of several. let head = format!( r#"[{{"time":"2026-09-03T14:29:52.900Z","type":"extension","record":"ready"}},{}"#, HEAD.trim_start_matches('[') @@ -289,6 +466,164 @@ mod tests { ); } + /// A record logged as JSON, with the escapes, multi-byte text, nesting, numbers and literals a + /// cut can land in. + const RECORD: &str = r#"{"level":"info","msg":"{\"a\":[1,true,null],\"b\":\"café caf\u00e9 \\\\ x\"}","n":-1.5e3}"#; + + /// A record logged as JSON whose structure, not just its text, holds arrays of objects. + const STRUCTURED_RECORD: &str = + r#"{"items":[{"id":1,"tags":["a","b"]},{"id":23,"ok":true,"v":null}],"n":-5}"#; + + const RUNTIME_DONE: &str = r#"{"time":"2026-09-03T14:29:52.950Z","type":"platform.runtimeDone","record":{"requestId":"abc123","status":"success"}}"#; + + /// A piece of a record as Lambda delivers it: raw, under an envelope of its own. + fn piece(millis: &str, bytes: &str) -> String { + format!(r#"{{"time":"2026-09-03T14:29:52.{millis}Z","type":"function","record":{bytes}}}"#) + } + + fn assert_whole_record(record: &str, events: &[TelemetryEvent]) { + let record = serde_json::from_str(record).expect("valid record"); + assert_eq!(events.len(), 2); + assert_eq!(events[0].record, TelemetryRecord::Function(record)); + assert!(matches!( + events[1].record, + TelemetryRecord::PlatformRuntimeDone { .. } + )); + } + + /// Each record above, cut in two at every char boundary. + fn cuts() -> impl Iterator { + [RECORD, STRUCTURED_RECORD].into_iter().flat_map(|record| { + (1..record.len()) + .filter(move |&cut| record.is_char_boundary(cut)) + .map(move |cut| { + let (first, rest) = record.split_at(cut); + (record, first, rest) + }) + }) + } + + #[test] + fn joins_pieces_within_a_payload_at_every_cut() { + // Another stream's event, such as the extension's own log line, can land between pieces. + let extension = + r#"{"time":"2026-09-03T14:29:52.931Z","type":"extension","record":"flushing"},"#; + for (record, first, rest) in cuts() { + for between in ["", extension] { + let body = format!( + "[{},{between}{},{RUNTIME_DONE}]", + piece("929", first), + piece("930", rest) + ); + // Known gap: a cut inside a number in an array of objects can leave the body + // valid JSON, so it never reaches the stitcher. + if serde_json::from_str::>(&body).is_ok() { + continue; + } + + let Stitch::Complete(mut events) = stitch(&FragmentBuffer::default(), &body) else { + panic!("pieces cut after {first:?} were not joined"); + }; + if !between.is_empty() { + assert!(matches!( + events.remove(0).record, + TelemetryRecord::Extension(_) + )); + } + assert_whole_record(record, &events); + } + } + } + + #[test] + fn joins_pieces_across_payloads_at_every_cut() { + for (record, first, rest) in cuts() { + let fragments = FragmentBuffer::default(); + + let head = format!("[{}]", piece("929", first)); + assert!( + matches!(stitch(&fragments, &head), Stitch::Pending), + "payload ending after {first:?} was not held" + ); + + let tail = format!("[{},{RUNTIME_DONE}]", piece("930", rest)); + let Stitch::Complete(events) = stitch(&fragments, &tail) else { + panic!("payloads cut after {first:?} were not joined"); + }; + assert_whole_record(record, &events); + } + } + + /// The shape seen in production: a record in several pieces, the payload ending between + /// two of them. + #[test] + fn joins_a_record_in_pieces_across_payloads() { + let fragments = FragmentBuffer::default(); + + let (a, rest) = RECORD.split_at(20); + let (b, rest) = rest.split_at(20); + let (c, d) = rest.split_at(20); + + let head = format!( + r#"[{{"time":"2026-09-03T14:29:52.900Z","type":"extension","record":"ready"}},{},{}]"#, + piece("929", a), + piece("930", b) + ); + assert!(matches!(stitch(&fragments, &head), Stitch::Pending)); + + let tail = format!("[{},{},{RUNTIME_DONE}]", piece("931", c), piece("932", d)); + let mut events = completed(stitch(&fragments, &tail)); + + assert_eq!(events.len(), 3); + assert!(matches!( + events.remove(0).record, + TelemetryRecord::Extension(_) + )); + assert_whole_record(RECORD, &events); + } + + #[test] + fn does_not_mistake_a_broken_record_for_pieces() { + let body = r#"[{"time":"2026-09-03T14:29:52.929Z","type":"function","record":{"message":"ok"}},{"time":"2026-09-03T14:29:52.930Z","type":"function","record":{"message":"bad\q"}}]"#; + assert!(matches!( + stitch(&FragmentBuffer::default(), body), + Stitch::Discarded + )); + } + + /// `platform.runtimeDone` after a cut piece is moved ahead of it rather than glued into it. + #[test] + fn keeps_runtime_done_whole_after_a_cut_piece() { + let fragments = FragmentBuffer::default(); + + let head = format!("[{},{RUNTIME_DONE}]", piece("929", r#"{"message":"AA"#)); + assert!(matches!(stitch(&fragments, &head), Stitch::Pending)); + + let tail = format!("[{}]", piece("930", r#"BB"}"#)); + let events = completed(stitch(&fragments, &tail)); + assert_eq!(events.len(), 2); + assert!(matches!( + events[0].record, + TelemetryRecord::PlatformRuntimeDone { .. } + )); + assert_eq!( + events[1].record, + TelemetryRecord::Function(serde_json::json!({"message": "AABB"})) + ); + } + + /// A log line can hold telemetry-shaped JSON as data. In a payload that parses, that's only + /// taken for a split record after a full first piece. + #[test] + fn leaves_telemetry_shaped_data_alone_in_a_payload_that_parses() { + let filler = vec![piece("900", &format!(r#""{}""#, "y".repeat(1024))); 256].join(","); + let nested = r#"{"events":[{"time":"2026-09-03T14:29:52.900Z","type":"function","record":1},{"time":"2026-09-03T14:29:52.901Z","type":"function","record":2}]}"#; + let body = format!("[{filler},{},{RUNTIME_DONE}]", piece("929", nested)); + + let events: Vec = serde_json::from_str(&body).expect("valid payload"); + assert!(rejoin_split_that_parsed(body.as_bytes(), events.len()).is_none()); + } + #[test] fn does_not_hold_a_payload_that_arrived_whole() { let fragments = FragmentBuffer::default(); @@ -309,13 +644,43 @@ mod tests { assert!(matches!(stitch(&fragments, HEAD), Stitch::Pending)); - // A different record's envelope, so the held fragment goes and this one takes its - // place. - let other = - r#"[{"time":"2026-09-03T14:30:11.001Z","type":"function","record":{"message":"CCCC}]"#; + // Another cut record after a whole event, which doesn't parse joined on, so the held + // fragment goes and this one takes its place. + let other = r#"[{"time":"2026-09-03T14:30:11.000Z","type":"extension","record":"ready"},{"time":"2026-09-03T14:30:11.001Z","type":"function","record":{"message":"CCCC}]"#; assert!(matches!(stitch(&fragments, other), Stitch::Pending)); - // Proof the first fragment is gone: its own continuation no longer joins. - assert!(matches!(stitch(&fragments, TAIL), Stitch::Discarded)); + // Proof the first fragment is gone: the continuation now resumes this one. + let events = completed(stitch(&fragments, TAIL)); + assert_eq!( + events[1].record, + TelemetryRecord::Function(serde_json::json!({"message": "CCCCBBBB"})) + ); + } + + #[test] + fn recovers_a_payload_that_does_not_continue_the_held_fragment() { + let fragments = FragmentBuffer::default(); + + assert!(matches!(stitch(&fragments, HEAD), Stitch::Pending)); + + // Whole once its own seam is dropped, so it must not go down with the held fragment. + let (first, rest) = RECORD.split_at(20); + let body = format!( + "[{},{},{RUNTIME_DONE}]", + piece("931", first), + piece("932", rest) + ); + assert_whole_record(RECORD, &completed(stitch(&fragments, &body))); + } + + #[test] + fn does_not_resume_a_fragment_with_a_whole_event() { + let fragments = FragmentBuffer::default(); + + assert!(matches!(stitch(&fragments, HEAD), Stitch::Pending)); + + // Only a log piece continues one, so this record isn't glued onto the held message. + let body = r#"[{"time":"2026-09-03T14:29:52.930Z","type":"platform.runtimeDone","record":BBBB"}}]"#; + assert!(matches!(stitch(&fragments, body), Stitch::Discarded)); } }