diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index 86ff12b170..dc496fa306 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -20,7 +20,7 @@ use crate::pagination::Pagination; use crate::persistence::{ ObjectLabels, ObjectListQuery, ObjectType, WriteCondition, generate_name, }; -use crate::tracing_bus::CursoredEvent; +use crate::tracing_bus::{CursoredEvent, ResumeSnapshot}; use crate::watch_cursor::WatchCursor; use futures::future; use openshell_core::net::set_tcp_nodelay_best_effort; @@ -99,19 +99,6 @@ const RESUME_SPACE_GONE: &str = "resume_after_cursor belongs to a cursor space t const RESUME_CURSOR_AHEAD: &str = "resume_after_cursor is ahead of every cursor this sandbox has \ issued. Restart the watch with an empty resume_after_cursor."; -/// Whether `sandbox_id`'s cursor space is still the one that issued `epoch`. -/// -/// A teardown retires the space and the next publish mints a replacement that -/// renumbers from 1, so a surviving epoch is the only proof that a seq validated -/// earlier still addresses the same numbering. Absent counts as changed: there -/// is nothing left for the cursor to point into. -fn cursor_space_is(state: &ServerState, sandbox_id: &str, epoch: uuid::Uuid) -> bool { - state - .tracing_log_bus - .cursor_space(sandbox_id) - .is_some_and(|space| space.epoch == epoch) -} - #[derive(Debug)] pub struct WatchSandboxStream { receiver: ReceiverStream>, @@ -2203,21 +2190,19 @@ pub(super) async fn handle_watch_sandbox( } else { None }; - let mut log_rx = if follow_logs { - Some(state.tracing_log_bus.subscribe(&sandbox_id)) - } else { - None - }; - let mut platform_rx = if follow_events { - Some( - state - .tracing_log_bus - .platform_event_bus - .subscribe(&sandbox_id), - ) - } else { - None - }; + // + // The resumable sources subscribe together with a high-water mark + // taken under the cursor-space lock, so the snapshot below can + // tell history (at or below the mark, buffered only) from events + // published while the stream opens (above it, buffered *and* + // queued on these receivers). + let crate::tracing_bus::ResumableSubscription { + mut log_rx, + mut platform_rx, + subscribed_at, + } = state + .tracing_log_bus + .subscribe_resumable(&sandbox_id, follow_logs, follow_events); // Re-read the snapshot now that we have subscriptions active. match state.store.get_message::(&sandbox_id).await { @@ -2257,28 +2242,26 @@ pub(super) async fn handle_watch_sandbox( // Highest seq the tail/replay phase already handled, tracked per // source. The broadcast receivers were subscribed before replay ran, - // so an event published during initialization can sit in both the - // replay buffer and a live receiver; the live loop suppresses events - // at or below its source's mark so each is delivered exactly once. + // so an event published during initialization (above + // `subscribed_at`) sits in both the replay buffer and a live + // receiver; the live loop suppresses events at or below its + // source's mark so each is delivered exactly once. // - // The two marks must stay separate. Both buses number from one - // shared cursor space, but they are read at different instants and - // bounded independently (`log_tail_lines` vs `event_tail`, which has - // no default and so replays nothing unless the client asks). A - // single shared mark therefore lets the deeper source censor the - // shallower one: with the default `event_tail` of 0 the mark rises - // to the newest buffered log while no platform event was replayed at - // all, and every platform event published in the initialization - // window is dropped as a duplicate of something never sent. Keyed by - // source, an event is suppressed only if its own source's replay - // actually covered it. + // Both reads happen under one cursor-space lock hold, so every + // event at or below the snapshot's high-water mark was buffered + // when they ran. A resume replays everything after the cursor from + // each followed source, so each source's mark is the last seq its + // own replay returned. The initial tail is depth-bounded, so both + // marks move to the snapshot's high-water mark instead. That is + // safe only because the depth bound never applies above + // `subscribed_at`: every event published while the stream opened + // is in the batch (or, if its buffer already evicted it, reported), + // and everything else at or below the mark is history the live + // receivers never saw. // - // No unit test pins this. The only reachable window is between the - // subscribe above and the log tail read below -- an event published - // earlier is replayed rather than live, and one published later - // outranks the mark -- and the producer crosses that window with no - // await a test can wedge open. Reproducing it needs a seam in the - // producer, which is not worth adding to production code. + // Tests force both windows through the bus's test-only snapshot + // hooks: one publishes after subscribe and before the snapshot, the + // other publishes between the snapshot's two reads. let resume_seq = resume_after.map_or(0, |resume| resume.seq); let mut log_cutoff: u64 = resume_seq; let mut platform_cutoff: u64 = resume_seq; @@ -2298,62 +2281,39 @@ pub(super) async fn handle_watch_sandbox( // and silently swallow every live event beneath it. Comparing // epochs answers "did this cursor come from *this* space?", // which no numeric bound can. - match state.tracing_log_bus.cursor_space(&sandbox_id) { - None => { - let _ = tx.send(Err(Status::out_of_range(RESUME_SPACE_GONE))).await; - return; - } - Some(space) if space.epoch != resume.epoch => { + // + // Validation and both bus reads happen inside one call, under + // one lock hold (see `TracingLogBus::snapshot_after`). A + // separate validate-then-read pair -- even with the two reads + // themselves atomic against each other -- still leaves a + // window between the validation's lock and the reads' lock + // for a teardown plus a republish to retire the validated + // space and install a replacement in between; the reads would + // then apply the old space's seq to the replacement's + // buffers and find no gap, since `tail_after` only compares + // numbers. Folding validation into the same hold as the reads + // closes that window entirely. + let (log_replay, platform_replay) = match state.tracing_log_bus.snapshot_after( + &sandbox_id, + resume.epoch, + resume.seq, + resume.seq, + follow_logs, + follow_events, + ) { + ResumeSnapshot::SpaceGone => { let _ = tx.send(Err(Status::out_of_range(RESUME_SPACE_GONE))).await; return; } - // Right space, but ahead of anything it issued: only a - // fabricated token gets here. Reject rather than accept a - // cutoff no event can ever exceed. - Some(space) if resume.seq > space.highest_seq => { + ResumeSnapshot::CursorAhead => { let _ = tx .send(Err(Status::out_of_range(RESUME_CURSOR_AHEAD))) .await; return; } - Some(_) => {} - } - - let log_replay = if follow_logs { - Some(state.tracing_log_bus.tail_after(&sandbox_id, resume.seq)) - } else { - None - }; - - let platform_replay = if follow_events { - Some( - state - .tracing_log_bus - .platform_event_bus - .tail_after(&sandbox_id, resume.seq), - ) - } else { - None + ResumeSnapshot::Read(log, platform) => (log, platform), }; - // Re-check the epoch now that both tails are in hand. The check - // above and each `tail_after` take their locks independently, so - // a teardown plus a republish can retire the validated space and - // install a replacement in between. The reads would then have - // applied the old space's seq to the new space's buffers, and - // `tail_after` -- which only knows numbers -- would report no gap - // while skipping every replacement event at or below it. The - // second look is cheap and runs before anything is emitted, so a - // space that moved under us ends the stream instead of serving a - // truncated replay. - // - // Ordering is unchanged: this takes only the allocator lock and - // releases it, never held across a bus lock. - if !cursor_space_is(&state, &sandbox_id, resume.epoch) { - let _ = tx.send(Err(Status::out_of_range(RESUME_SPACE_GONE))).await; - return; - } - // Gap check FIRST (borrows), before the merge moves the vecs. for replay in [&log_replay, &platform_replay] { if let Some(Err(gap)) = replay { @@ -2426,29 +2386,125 @@ pub(super) async fn handle_watch_sandbox( // cursors expects. // // The two windows are truncated independently (log_tail vs - // event_tail), so this merges whatever each bus retained; it - // does not align their depths. + // event_tail). A client tracks one scalar cursor across both + // sources -- whatever the highest one it saw was -- so a + // cursor this batch hands out is only safe to resume from if + // *every* followed source can vouch it covered everything up + // to that point. + // + // `tail_with_floor` reports each source's own coverage floor: + // the *newest* event its window excluded. The source's + // excluded set is a prefix of its own tail, so it delivered + // everything it retains above the floor. The batch as a whole + // is therefore safe above the highest floor across followed + // sources: every event from either source past that point is + // in the batch, and everything at or below it is outside the + // window, the same as ordinary tail truncation. A source never + // withholds its own window this way; only a sibling's floor + // can cut a source's window back. + // + // That can mean fewer events than a source's depth asked for + // whenever a followed sibling left out an event newer than + // part of its window. This depends on where the histories + // fall in the cursor order, not on the depths being unequal: + // logs at 1-2 and platform at 3-4 with both depths at 1 + // leaves out platform 3, so log 2 is withheld. + // That's the trade-off of a single shared scalar cursor: an + // event handed out below the sibling's floor would let a + // resume skip the sibling's withheld events past it. + // + // Both windows and the space's high-water mark come from one + // lock hold (see `TracingLogBus::snapshot_tail`), so no + // publish can land between the two reads. Read separately, a + // log published after the log read and a platform event + // published after it could let the platform window hand out + // the higher cursor while the log arrives live afterwards. + // + // Depths bound history only. Events published after the + // receivers subscribed are always in their source's window, so + // the batch can exceed `log_tail_lines`/`event_tail` by them. + // Each floor therefore stays at or below `subscribed_at`, and + // none of those events is ever withheld by a sibling's floor. + let crate::tracing_bus::TailSnapshot { + log, + platform, + highest_seq, + } = state.tracing_log_bus.snapshot_tail( + &sandbox_id, + follow_logs.then_some(log_tail as usize), + follow_events.then_some(event_tail as usize), + subscribed_at, + ); + let crate::tracing_bus::SourceTail { + events: mut logs, + floor: log_floor, + evicted_while_opening: logs_evicted, + } = log; + let crate::tracing_bus::SourceTail { + mut events, + floor: platform_floor, + evicted_while_opening: events_evicted, + } = platform; + + // 0 means "nothing excluded" for a source, so a fully-covered + // source never raises the maximum. + let critical_floor = log_floor.max(platform_floor); + + let before = logs.len(); + logs.retain(|cursored| cursored.seq > critical_floor); + let log_withheld = before - logs.len(); + let before = events.len(); + events.retain(|cursored| cursored.seq > critical_floor); + let platform_withheld = before - events.len(); + + // Both cutoffs move to the snapshot's high-water mark. The + // receivers were subscribed before the snapshot, so an event + // published in between is both buffered and queued live; the + // snapshot kept it in the batch, so its live copy is a + // duplicate, and delivering it would emit a lower cursor after + // a higher one from the batch. Events at or below + // `subscribed_at` were never queued live. Above the mark, an + // event can only arrive live. + log_cutoff = log_cutoff.max(highest_seq); + platform_cutoff = platform_cutoff.max(highest_seq); + let mut tail: Vec = Vec::new(); - if follow_logs { - let logs = state.tracing_log_bus.tail(&sandbox_id, log_tail as usize); - if let Some(last) = logs.last() { - log_cutoff = log_cutoff.max(last.seq); + tail.extend(logs); + tail.extend(events); + + tail.sort_by_key(|cursored| cursored.seq); + + // Disclose the gap before anything else in this batch. The + // withheld events sit at or below the snapshot's mark, so the + // cutoffs above keep their live copies (if any) from being + // delivered out of order, while later live events past the + // mark still reach the client. The client has to learn the + // gap exists from this warning; a resume from any cursor this + // batch hands out cannot recover it, and `tail_after` reports + // no gap when asked, since nothing was evicted, only never + // sent. + if log_withheld > 0 || platform_withheld > 0 { + let warning = crate::sandbox_watch::coverage_gap_warning_event( + log_withheld, + platform_withheld, + ); + if tx.send(Ok(warning)).await.is_err() { + return; } - tail.extend(logs); } - if follow_events { - let events = state - .tracing_log_bus - .platform_event_bus - .tail(&sandbox_id, event_tail as usize); - if let Some(last) = events.last() { - platform_cutoff = platform_cutoff.max(last.seq); + // An event published while the stream opened, then trimmed + // from its buffer before the snapshot read it, is in neither + // the batch nor (past the cutoff) the live stream. + if logs_evicted || events_evicted { + let warning = crate::sandbox_watch::opening_eviction_warning_event( + logs_evicted, + events_evicted, + ); + if tx.send(Ok(warning)).await.is_err() { + return; } - tail.extend(events); } - tail.sort_by_key(|cursored| cursored.seq); - for cursored in tail { // Log filters; platform events carry no log fields and pass. if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log( @@ -5472,6 +5528,593 @@ mod tests { assert_eq!(got, vec![1, 2, 3, 4]); } + /// Regression for the inverted coverage-floor clamp: a single followed + /// source with more buffered events than `log_tail_lines` must still + /// deliver its newest window. A source never withholds its own window, + /// and the older lines it excluded are ordinary tail truncation, so no + /// coverage-gap warning is due either. + #[tokio::test] + async fn initial_tail_delivers_a_truncated_single_source_window() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("truncated", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 5); // cursors 1..=5 + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + log_tail_lines: 2, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + let mut got = Vec::new(); + for _ in 0..2 { + got.push(seq_of(&stream.next().await.unwrap().unwrap())); + } + assert_eq!(got, vec![4, 5]); + + let next = tokio::time::timeout(std::time::Duration::from_millis(200), stream.next()).await; + assert!(next.is_err(), "expected nothing further, got {next:?}"); + } + + /// A sibling's excluded backlog that is older than everything the deeper + /// source delivers imposes nothing: `event_tail: 0` leaves out platform + /// seq 1, but logs 2 and 3 both sit above it, so a resume from 3 skips + /// only an event outside the requested window. + #[tokio::test] + async fn initial_tail_delivers_logs_newer_than_a_sibling_sources_unreplayed_backlog() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("floor", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_platform_event(&state, &id, "e1"); // cursor 1, never replayed (event_tail: 0) + seed_log_lines(&state, &id, 2); // cursors 2, 3 + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + // Default: no backlog requested from the platform bus. + event_tail: 0, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + let mut got = Vec::new(); + for _ in 0..2 { + got.push(seq_of(&stream.next().await.unwrap().unwrap())); + } + assert_eq!(got, vec![2, 3]); + + let next = tokio::time::timeout(std::time::Duration::from_millis(200), stream.next()).await; + assert!( + next.is_err(), + "expected no warning or further events, got {next:?}" + ); + } + + /// The A2 repro: a platform event (seq 2) that `event_tail: 0` leaves + /// out sits between two logs. Handing out log 3 would let the client + /// remember cursor 3, and a resume from 3 would exclude the platform + /// event forever -- `tail_after` reports no gap, since nothing was + /// evicted, it was just never sent. Platform's floor (2) instead cuts the + /// log window back to seq 3 and withholds log 1 with a warning. + #[tokio::test] + async fn initial_tail_withholds_logs_behind_a_sibling_sources_unreplayed_backlog() { + use openshell_core::proto::sandbox_stream_event::Payload; + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("floorcut", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 1); // cursor 1, withheld (at or below floor) + seed_platform_event(&state, &id, "e2"); // cursor 2, never replayed (event_tail: 0) + seed_log_lines(&state, &id, 1); // cursor 3, delivered + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + event_tail: 0, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + // The client learns about the withheld log from an explicit warning + // rather than discovering it silently on a later reconnect. + let warning = stream.next().await.unwrap().unwrap(); + match warning.payload { + Some(Payload::Warning(w)) => { + assert!( + w.message + .contains("withheld 1 log line(s) and 0 platform event(s)"), + "message: {}", + w.message + ); + } + other => panic!("expected a coverage-gap warning, got {other:?}"), + } + assert!(warning.cursor.is_empty()); + + let delivered = stream.next().await.unwrap().unwrap(); + assert_eq!(seq_of(&delivered), 3); + + let next = tokio::time::timeout(std::time::Duration::from_millis(200), stream.next()).await; + assert!(next.is_err(), "expected nothing further, got {next:?}"); + } + + /// The coverage-gap warning discloses the withheld backlog once, up + /// front, rather than gating every later event: a live event published + /// after connect still reaches the client, even though its cursor sits + /// past the withheld platform backlog. The withheld events were + /// published before subscribe and can never arrive live, so refusing to + /// deliver anything past them would starve the stream indefinitely + /// instead of just disclosing the one, bounded gap. + #[tokio::test] + async fn live_delivery_proceeds_normally_after_the_coverage_gap_warning() { + use openshell_core::proto::sandbox_stream_event::Payload; + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("floorlive", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 1); // cursor 1, withheld (below critical_floor) + seed_platform_event(&state, &id, "e2"); // cursor 2, withheld (event_tail: 0) + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + event_tail: 0, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + let warning = stream.next().await.unwrap().unwrap(); + assert!(matches!(warning.payload, Some(Payload::Warning(_)))); + + // Published after connect: a genuinely new live event, not part of + // the withheld backlog. It must still be delivered. + seed_log_lines(&state, &id, 1); // cursor 3, live + + let live = tokio::time::timeout(std::time::Duration::from_secs(2), stream.next()) + .await + .expect("live event should arrive") + .unwrap() + .unwrap(); + assert_eq!(seq_of(&live), 3); + } + + /// `event_tail: 0` excludes *both* of the platform bus's own events + /// (seqs 1 and 3); a log event (seq 2) sits interleaved between them and + /// is otherwise fully covered by `log_tail_lines`. Delivering it would let + /// the client remember cursor 2 while platform's seq 3 was never sent, so + /// the floor (3, the newest excluded platform seq) withholds it too. + #[tokio::test] + async fn initial_tail_withholds_an_interleaved_sibling_event_between_two_excluded_seqs() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("interleavedfloor", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_platform_event(&state, &id, "e1"); // cursor 1, withheld + seed_log_lines(&state, &id, 1); // cursor 2, must also be withheld + seed_platform_event(&state, &id, "e3"); // cursor 3, withheld + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + event_tail: 0, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + let warning = stream.next().await.unwrap().unwrap(); + match warning.payload { + Some(openshell_core::proto::sandbox_stream_event::Payload::Warning(w)) => { + assert!(w.message.contains('1'), "message: {}", w.message); + } + other => panic!("expected a coverage-gap warning, got {other:?}"), + } + + // Nothing else: the log event must not slip through between + // platform's two excluded seqs. + let next = tokio::time::timeout(std::time::Duration::from_millis(200), stream.next()).await; + assert!( + next.is_err(), + "expected the interleaved log event to be withheld too, got {next:?}" + ); + } + + /// Windows that leave nothing out impose no floor: every buffered event + /// is delivered, as before A2. + #[tokio::test] + async fn initial_tail_unclamped_when_no_source_has_unreplayed_backlog() { + use tokio_stream::StreamExt as _; + + let state = test_server_state().await; + let sandbox = test_sandbox("nofloor", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_platform_event(&state, &id, "e1"); // cursor 1 + seed_log_lines(&state, &id, 2); // cursors 2, 3 + + let response = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + // Deep enough to cover the one platform event: no floor. + event_tail: 10, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + let mut got = Vec::new(); + for _ in 0..3 { + got.push(seq_of(&stream.next().await.unwrap().unwrap())); + } + assert_eq!(got, vec![1, 2, 3]); + } + + /// Open an initial watch following both sources with the given depths and + /// collect what the initial batch delivers: the coverage-gap warning + /// message, if any, and the delivered cursors in order. + async fn initial_tail_following_both( + state: &Arc, + sandbox: &Sandbox, + log_tail_lines: u32, + event_tail: u32, + ) -> (Option, Vec) { + use openshell_core::proto::sandbox_stream_event::Payload; + use tokio_stream::StreamExt as _; + + let response = handle_watch_sandbox( + state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + log_tail_lines, + event_tail, + ..Default::default() + }), + ) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty()); + + let mut warning = None; + let mut delivered = Vec::new(); + while let Ok(Some(item)) = + tokio::time::timeout(std::time::Duration::from_millis(200), stream.next()).await + { + let event = item.unwrap(); + match event.payload { + Some(Payload::Warning(w)) => { + assert!(delivered.is_empty(), "warning must precede the batch"); + assert!(warning.replace(w.message).is_none(), "one warning at most"); + } + _ => delivered.push(seq_of(&event)), + } + } + (warning, delivered) + } + + /// Equal depths do not rule out trimming. Logs sit at cursors 1-2 and + /// platform events at 3-4; with both depths at 1 the platform window + /// leaves out 3, which is newer than log 2, so log 2 is withheld even + /// though `log_tail_lines` alone would have delivered it. + #[tokio::test] + async fn initial_tail_equal_depths_withhold_an_older_log_window() { + let state = test_server_state().await; + let sandbox = test_sandbox("equallogsfirst", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 2); // cursors 1, 2 + seed_platform_event(&state, &id, "e3"); // cursor 3, left out by event_tail + seed_platform_event(&state, &id, "e4"); // cursor 4 + + let (warning, delivered) = initial_tail_following_both(&state, &sandbox, 1, 1).await; + let warning = warning.expect("withheld log 2 must be disclosed"); + assert!( + warning.contains("withheld 1 log line(s) and 0 platform event(s)"), + "message: {warning}" + ); + assert_eq!(delivered, vec![4]); + } + + /// The mirror case: platform events at cursors 1-2 and logs at 3-4. The + /// log window leaves out 3, which is newer than platform 2, so this time + /// the platform event is withheld. + #[tokio::test] + async fn initial_tail_equal_depths_withhold_an_older_platform_window() { + let state = test_server_state().await; + let sandbox = test_sandbox("equalplatformfirst", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_platform_event(&state, &id, "e1"); // cursor 1 + seed_platform_event(&state, &id, "e2"); // cursor 2 + seed_log_lines(&state, &id, 2); // cursors 3, 4; 3 left out by log_tail_lines + + let (warning, delivered) = initial_tail_following_both(&state, &sandbox, 1, 1).await; + let warning = warning.expect("withheld platform event 2 must be disclosed"); + assert!( + warning.contains("withheld 0 log line(s) and 1 platform event(s)"), + "message: {warning}" + ); + assert_eq!(delivered, vec![4]); + } + + /// Equal depths over interleaved histories withhold nothing when neither + /// window leaves out an event newer than part of the other: logs at 1 and + /// 3, platform at 2 and 4, both depths 1. Each window's excluded event is + /// older than everything delivered, so both 3 and 4 arrive, no warning. + #[tokio::test] + async fn initial_tail_equal_depths_over_interleaved_histories_withhold_nothing() { + let state = test_server_state().await; + let sandbox = test_sandbox("equalinterleaved", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 1); // cursor 1, left out by log_tail_lines + seed_platform_event(&state, &id, "e2"); // cursor 2, left out by event_tail + seed_log_lines(&state, &id, 1); // cursor 3 + seed_platform_event(&state, &id, "e4"); // cursor 4 + + let (warning, delivered) = initial_tail_following_both(&state, &sandbox, 1, 1).await; + assert_eq!(warning, None); + assert_eq!(delivered, vec![3, 4]); + } + + /// Open a watch following both sources and return the stream after its + /// status snapshot. + async fn watch_both( + state: &Arc, + sandbox: &Sandbox, + event_tail: u32, + resume_after_cursor: String, + ) -> WatchSandboxStream { + use tokio_stream::StreamExt as _; + + let response = handle_watch_sandbox( + state, + authed_request(WatchSandboxRequest { + sandbox: sandbox.object_name().to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + follow_logs: true, + follow_events: true, + event_tail, + resume_after_cursor, + ..Default::default() + }), + ) + .await + .unwrap(); + let mut stream = response.into_inner(); + let snap = stream.next().await.unwrap().unwrap(); + assert!(snap.cursor.is_empty(), "first item is the status snapshot"); + stream + } + + /// Collect what a stream yields until it goes quiet: warning messages and + /// resumable cursors, each in arrival order. + async fn drain_until_quiet(stream: &mut WatchSandboxStream) -> (Vec, Vec) { + use openshell_core::proto::sandbox_stream_event::Payload; + use tokio_stream::StreamExt as _; + + let mut warnings = Vec::new(); + let mut seqs = Vec::new(); + while let Ok(Some(item)) = + tokio::time::timeout(std::time::Duration::from_millis(200), stream.next()).await + { + let event = item.unwrap(); + match event.payload { + Some(Payload::Warning(w)) => warnings.push(w.message), + _ => seqs.push(seq_of(&event)), + } + } + (warnings, seqs) + } + + /// The reviewer's example on #3613. Both sources followed at default + /// depths; after the watch subscribes and before its snapshot, platform + /// event 1 and then log 2 are published. Both are queued live and the + /// snapshot's cutoff suppresses those copies, so the batch is their only + /// delivery: it must carry both, even though `event_tail` asked for no + /// platform history. Previously the depth bound left event 1 out and the + /// cutoff swallowed its live copy, with no warning, and a resume after 2 + /// could never recover it. + #[tokio::test] + async fn initial_tail_delivers_events_published_while_the_stream_opens() { + let state = test_server_state().await; + let sandbox = test_sandbox("opening", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + let hook_state = Arc::clone(&state); + let hook_id = id.clone(); + let mut fired = false; + state.tracing_log_bus.set_snapshot_hook(move |point| { + if point == crate::tracing_bus::SnapshotPoint::AfterSubscribe && !fired { + fired = true; + seed_platform_event(&hook_state, &hook_id, "e1"); // cursor 1 + seed_log_lines(&hook_state, &hook_id, 1); // cursor 2 + } + }); + + let mut stream = watch_both(&state, &sandbox, 0, String::new()).await; + let (warnings, delivered) = drain_until_quiet(&mut stream).await; + assert_eq!(delivered, vec![1, 2]); + assert!(warnings.is_empty(), "nothing was withheld: {warnings:?}"); + + // Live delivery continues past the batch. + seed_log_lines(&state, &id, 1); // cursor 3 + let (warnings, delivered) = drain_until_quiet(&mut stream).await; + assert_eq!(delivered, vec![3]); + assert!(warnings.is_empty()); + + // A reconnect from the highest cursor the batch handed out misses + // nothing that was retained below it. + let mut resumed = watch_both(&state, &sandbox, 0, cursor_token(&state, &id, 2)).await; + let (warnings, delivered) = drain_until_quiet(&mut resumed).await; + assert_eq!(delivered, vec![3]); + assert!(warnings.is_empty()); + } + + /// A publish forced between the initial snapshot's log read and platform + /// read. The snapshot holds the cursor-space lock across both, so the + /// publisher waits: the batch is exactly the seeded history, the raced + /// log and platform event arrive live in cursor order once each, and a + /// resume from the batch's highest cursor replays both. With separate + /// reads the batch would carry the raced platform event (4) without the + /// log published before it (3), which would then arrive after 4 or, after + /// a reconnect from 4, never. + #[tokio::test] + async fn initial_tail_publish_between_snapshot_reads_arrives_live_in_order() { + let state = test_server_state().await; + let sandbox = test_sandbox("racetail", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 1); // cursor 1 + seed_platform_event(&state, &id, "e2"); // cursor 2 + let probe = crate::tracing_bus::race_publish_between_reads(&state.tracing_log_bus, &id); + + let mut stream = watch_both(&state, &sandbox, 10, String::new()).await; + let (warnings, delivered) = drain_until_quiet(&mut stream).await; + { + let mut probe = probe.lock().unwrap(); + probe.join_publisher(); + assert!(probe.fired, "the hook must run between the two reads"); + assert!(probe.lock_held_between_reads, "lock released between reads"); + assert!( + !probe.publisher_finished_between_reads, + "a publish landed between the log read and the platform read" + ); + } + assert!(warnings.is_empty(), "unexpected warnings: {warnings:?}"); + assert_eq!(delivered, vec![1, 2, 3, 4], "batch, then 3 and 4 live"); + + let mut resumed = watch_both(&state, &sandbox, 10, cursor_token(&state, &id, 2)).await; + let (warnings, delivered) = drain_until_quiet(&mut resumed).await; + assert!(warnings.is_empty()); + assert_eq!(delivered, vec![3, 4]); + } + + /// The same forced interleaving on the resume path. The replay after + /// cursor 1 holds platform event 2 only; the raced log (3) and platform + /// event (4) arrive live, in order, after it. With separate reads the + /// replay would carry 4 ahead of 3, and a client that disconnected after 4 + /// would resume past 3 forever. + #[tokio::test] + async fn resume_publish_between_snapshot_reads_arrives_live_in_order() { + let state = test_server_state().await; + let sandbox = test_sandbox("raceresume", Vec::new()); + state.store.put_message(&sandbox).await.unwrap(); + let id = sandbox.object_id().to_string(); + + seed_log_lines(&state, &id, 1); // cursor 1 + seed_platform_event(&state, &id, "e2"); // cursor 2 + let probe = crate::tracing_bus::race_publish_between_reads(&state.tracing_log_bus, &id); + + let mut stream = watch_both(&state, &sandbox, 0, cursor_token(&state, &id, 1)).await; + let (warnings, delivered) = drain_until_quiet(&mut stream).await; + { + let mut probe = probe.lock().unwrap(); + probe.join_publisher(); + assert!(probe.fired, "the hook must run between the two reads"); + assert!(probe.lock_held_between_reads, "lock released between reads"); + assert!( + !probe.publisher_finished_between_reads, + "a publish landed between the log read and the platform read" + ); + } + assert!(warnings.is_empty(), "unexpected warnings: {warnings:?}"); + assert_eq!(delivered, vec![2, 3, 4], "replay, then 3 and 4 live"); + + // Reconnecting from the replay's only cursor recovers both. + let mut resumed = watch_both(&state, &sandbox, 0, cursor_token(&state, &id, 2)).await; + let (warnings, delivered) = drain_until_quiet(&mut resumed).await; + assert!(warnings.is_empty()); + assert_eq!(delivered, vec![3, 4]); + } + #[tokio::test] async fn live_delivery_orders_events_across_sources_by_cursor() { use tokio_stream::StreamExt as _; @@ -5822,45 +6465,6 @@ mod tests { assert!(stream.next().await.is_none()); } - /// The guard behind the producer's post-replay epoch re-check. - /// - /// Validation and the two `tail_after` reads take their locks separately, so - /// a teardown plus a republish can swap the space in between and leave the - /// reads applying an old seq to a replacement's buffers -- `tail_after` only - /// compares numbers, so it reports no gap while skipping every replacement - /// event at or below that seq. The producer re-checks the epoch once both - /// tails are in hand and before emitting anything; this pins what that check - /// must answer. - /// - /// The interleaving itself is not reachable from a test: the producer runs - /// validation, both reads, and the re-check with no await in between, so - /// there is nothing to suspend it on. - #[tokio::test] - async fn cursor_space_is_rejects_a_replacement_space() { - let state = test_server_state().await; - let sandbox = test_sandbox("respace", Vec::new()); - state.store.put_message(&sandbox).await.unwrap(); - let id = sandbox.object_id().to_string(); - - seed_log_lines(&state, &id, 1); - let original = state.tracing_log_bus.cursor_space(&id).unwrap().epoch; - assert!(cursor_space_is(&state, &id, original)); - - // Teardown alone leaves no space to point into. - state.tracing_log_bus.remove(&id); - assert!(state.tracing_log_bus.cursor_space(&id).is_none()); - assert!(!cursor_space_is(&state, &id, original)); - - // The republish installs a replacement renumbered from 1. Its seqs - // overlap the retired space's, so only the epoch separates them. - seed_log_lines(&state, &id, 1); - let replacement = state.tracing_log_bus.cursor_space(&id).unwrap(); - assert_ne!(replacement.epoch, original); - assert_eq!(replacement.highest_seq, 1); - assert!(!cursor_space_is(&state, &id, original)); - assert!(cursor_space_is(&state, &id, replacement.epoch)); - } - #[tokio::test] async fn resume_from_reset_cursor_space_terminates_out_of_range() { use tokio_stream::StreamExt as _; diff --git a/crates/openshell-server/src/sandbox_watch.rs b/crates/openshell-server/src/sandbox_watch.rs index 679912d352..c79c81d399 100644 --- a/crates/openshell-server/src/sandbox_watch.rs +++ b/crates/openshell-server/src/sandbox_watch.rs @@ -154,6 +154,81 @@ pub fn lag_warning_event(n: u64) -> openshell_core::proto::SandboxStreamEvent { } } +/// Build the warning payload emitted when the cross-source coverage floor +/// withholds events on connect. +/// +/// Unlike broadcast lag, nothing was evicted here: the withheld events are +/// still sitting in a bus's tail, excluded only because a sibling source's +/// shallower replay depth means handing them out would let the client's +/// single shared cursor imply coverage that sibling can't back. They won't +/// be resent later in this stream -- a resume replays past the cursor this +/// connect settles on, and these are at or below it by construction -- so +/// the client needs to know now, not discover it silently after a reconnect. +pub fn coverage_gap_warning(log_withheld: usize, platform_withheld: usize) -> SandboxStreamWarning { + SandboxStreamWarning { + message: format!( + "resume cursor cannot cover every followed source's backlog; withheld {log_withheld} \ + log line(s) and {platform_withheld} platform event(s) that a shallower sibling source \ + had not replayed. Restart the watch with a deeper log_tail_lines/event_tail if you need them; \ + they will not be recoverable from this cursor." + ), + } +} + +/// Wrap [`coverage_gap_warning`] in a `SandboxStreamEvent` ready to send. +pub fn coverage_gap_warning_event( + log_withheld: usize, + platform_withheld: usize, +) -> openshell_core::proto::SandboxStreamEvent { + use openshell_core::proto::sandbox_stream_event::Payload; + openshell_core::proto::SandboxStreamEvent { + payload: Some(Payload::Warning(coverage_gap_warning( + log_withheld, + platform_withheld, + ))), + // Warnings are not part of the resumable log/platform sequence. + cursor: String::new(), + } +} + +/// Build the warning payload emitted when a bus evicted events published +/// while the watch was opening, before the initial snapshot could read them. +/// +/// Those events were queued on the live receivers too, but the snapshot's +/// cutoff suppresses live copies at or below its high-water mark, so they are +/// in neither the batch nor the live stream. The bus cannot say how many were +/// lost -- its seqs interleave with the sibling source's -- only which source. +pub fn opening_eviction_warning(logs: bool, platform_events: bool) -> SandboxStreamWarning { + let what = match (logs, platform_events) { + (true, true) => "log lines and platform events", + (true, false) => "log lines", + _ => "platform events", + }; + SandboxStreamWarning { + message: format!( + "some {what} published while this watch was opening were evicted from the \ + gateway buffer before they could be sent; they are not recoverable from any \ + cursor this stream hands out." + ), + } +} + +/// Wrap [`opening_eviction_warning`] in a `SandboxStreamEvent` ready to send. +pub fn opening_eviction_warning_event( + logs: bool, + platform_events: bool, +) -> openshell_core::proto::SandboxStreamEvent { + use openshell_core::proto::sandbox_stream_event::Payload; + openshell_core::proto::SandboxStreamEvent { + payload: Some(Payload::Warning(opening_eviction_warning( + logs, + platform_events, + ))), + // Warnings are not part of the resumable log/platform sequence. + cursor: String::new(), + } +} + #[cfg(test)] mod tests { use super::*; @@ -226,6 +301,59 @@ mod tests { } } + #[test] + fn coverage_gap_warning_reports_withheld_counts() { + let warning = coverage_gap_warning(2, 1); + assert!( + warning.message.contains('2'), + "message: {}", + warning.message + ); + assert!( + warning.message.contains('1'), + "message: {}", + warning.message + ); + } + + #[test] + fn opening_eviction_warning_names_the_affected_sources() { + assert!( + opening_eviction_warning(true, false) + .message + .contains("some log lines ") + ); + assert!( + opening_eviction_warning(false, true) + .message + .contains("some platform events ") + ); + assert!( + opening_eviction_warning(true, true) + .message + .contains("log lines and platform events") + ); + assert!( + opening_eviction_warning_event(true, false) + .cursor + .is_empty() + ); + } + + #[test] + fn coverage_gap_warning_event_wraps_warning_payload_with_empty_cursor() { + use openshell_core::proto::sandbox_stream_event::Payload; + let evt = coverage_gap_warning_event(2, 1); + assert!(evt.cursor.is_empty()); + match evt.payload { + Some(Payload::Warning(w)) => { + assert!(w.message.contains('2')); + assert!(w.message.contains('1')); + } + other => panic!("expected Warning payload, got {other:?}"), + } + } + // Broadcast lag is recoverable at the tokio layer: after `Lagged`, the same // receiver keeps yielding the oldest surviving messages instead of closing. #[tokio::test] diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index 92d30f5571..02f36dbe18 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -22,6 +22,8 @@ pub struct TracingLogBus { inner: Arc>, pub(crate) platform_event_bus: PlatformEventBus, seq: SeqAllocator, + #[cfg(test)] + hooks: SnapshotHooks, } #[derive(Debug, Clone)] @@ -29,6 +31,10 @@ struct Inner { per_id: HashMap, } +/// Result of a resumable-source replay read: the events after the requested +/// cursor, or the gap that made them unreplayable. +pub(crate) type ReplayResult = Result, ResumeGap>; + /// A buffered or broadcast stream event paired with its raw sequence number. /// /// The wire `SandboxStreamEvent.cursor` is an opaque token; ordering decisions @@ -182,6 +188,50 @@ fn tail_after_impl( Ok(res) } +/// Split a tail into what a bounded request returns and the coverage floor +/// that bound leaves behind. +/// +/// `max` may return fewer events than the tail currently retains; the ones +/// excluded are the oldest, at the front. The floor is the *newest* excluded +/// event's seq: this bus returned every event it retains above the floor and +/// none at or below it. `0` means `max` covered everything currently +/// retained, so there's nothing this call withheld. +/// +/// Because the excluded set is a prefix of this bus's retained deque, the +/// floor is the only boundary a caller needs: an event from any followed +/// source with a seq above the floor is safe to hand out as far as this bus +/// is concerned, and one at or below it may sit next to an event this bus +/// withheld. Excluding events below the delivered window is ordinary tail +/// truncation, the same as a single-source `log_tail_lines`. +/// +/// This is what lets a caller answer "is a cursor from this batch safe to +/// resume from," which `tail_after`'s eviction-only gap check can't: a +/// shallow request and a bus eviction both withhold events, but only the +/// eviction leaves a trace in `last_trimmed_seq`. +/// +/// `keep_after` exempts the newest events from `max`: every retained event +/// with a seq above it is returned regardless of depth. The watch handler +/// passes the high-water mark it recorded when it subscribed, so an event +/// published while the stream was opening -- one that is also queued on the +/// live receiver, where the snapshot's cutoff will suppress it -- is always +/// delivered in the batch rather than trimmed as history. The returned set is +/// still a suffix of the tail, so the floor keeps its meaning, and it never +/// exceeds `keep_after`. +fn tail_with_floor_impl( + tail: &VecDeque, + max: usize, + keep_after: u64, +) -> (Vec, u64) { + let first_unseen = tail.partition_point(|cursored| cursored.seq <= keep_after); + let excluded = tail.len().saturating_sub(max).min(first_unseen); + let floor = excluded + .checked_sub(1) + .and_then(|newest_excluded| tail.get(newest_excluded)) + .map_or(0, |cursored| cursored.seq); + let events = tail.iter().skip(excluded).cloned().collect(); + (events, floor) +} + impl Default for TracingLogBus { fn default() -> Self { Self::new() @@ -200,6 +250,8 @@ impl TracingLogBus { })), platform_event_bus: PlatformEventBus::new(seq.clone()), seq, + #[cfg(test)] + hooks: SnapshotHooks::default(), } } @@ -274,18 +326,6 @@ impl TracingLogBus { self.seq.space(sandbox_id) } - pub(crate) fn tail_after( - &self, - sandbox_id: &str, - after_seq: u64, - ) -> Result, ResumeGap> { - let inner = self.inner.lock().expect("tracing bus lock poisoned"); - inner.per_id.get(sandbox_id).map_or_else( - || Ok(Vec::new()), - |per| tail_after_impl(&per.tail, per.last_trimmed_seq, after_seq), - ) - } - /// Publish a log line from an external source (e.g., sandbox push). /// /// Injects the line into the same broadcast channel and tail buffer @@ -329,6 +369,364 @@ impl TracingLogBus { } } } + + /// Validate a resume cursor's epoch and read both resumable buses, all + /// under one lock hold. + /// + /// Folding validation into the same hold as the reads is what makes this + /// atomic against a teardown: checking the epoch first and reading after + /// -- even via a single call that itself locks once internally -- still + /// leaves a window between the two lock acquisitions for a `remove` plus + /// a republish to retire the validated space and install a replacement + /// numbered from 1. A read landing after that would apply the old + /// space's seq to the replacement's buffers and find no gap, since + /// `tail_after` only compares numbers. Locking the cursor space before + /// checking the epoch -- the same lock every publish path holds across + /// its own tail insert -- closes that window entirely, and also makes + /// the two bus reads atomic against any publish on either bus, for the + /// same reason. + pub(crate) fn snapshot_after( + &self, + sandbox_id: &str, + expected_epoch: Uuid, + log_after: u64, + platform_after: u64, + want_log: bool, + want_platform: bool, + ) -> ResumeSnapshot { + let spaces = self.seq.lock(); + let Some(space) = spaces.get(sandbox_id) else { + return ResumeSnapshot::SpaceGone; + }; + if space.epoch != expected_epoch { + return ResumeSnapshot::SpaceGone; + } + // Right space, but ahead of anything it has issued: only a + // fabricated cursor gets here. Reject rather than accept a cutoff no + // event can ever exceed. + if log_after.max(platform_after) > space.next.saturating_sub(1) { + return ResumeSnapshot::CursorAhead; + } + + let log = want_log.then(|| { + let inner = self.inner.lock().expect("tracing bus lock poisoned"); + inner.per_id.get(sandbox_id).map_or_else( + || Ok(Vec::new()), + |per| tail_after_impl(&per.tail, per.last_trimmed_seq, log_after), + ) + }); + #[cfg(test)] + self.hooks.run(SnapshotPoint::BetweenReads); + let platform = want_platform.then(|| { + let inner = self + .platform_event_bus + .inner + .lock() + .expect("platform event bus lock poisoned"); + inner.per_id.get(sandbox_id).map_or_else( + || Ok(Vec::new()), + |per| tail_after_impl(&per.tail, per.last_trimmed_seq, platform_after), + ) + }); + ResumeSnapshot::Read(log, platform) + } + + /// Read both bounded initial tails and the space's high-water mark, all + /// under one lock hold. + /// + /// `None` skips a source. Holding the cursor-space lock -- the one every + /// publish holds across its tail insert and broadcast send -- means no + /// publish can land between the two reads, and `highest_seq` is exactly + /// the boundary between what these tails could see and what can only + /// arrive live. Every event at or below it was already buffered when the + /// tails were read, so the caller can suppress live copies at or below it + /// without dropping anything the tails could not have covered. + /// + /// `subscribed_at` is the mark [`Self::subscribe_resumable`] recorded. + /// Events above it were published after the caller's live receivers + /// existed, so they are queued live as well as buffered here; the caller + /// suppresses those live copies, which means this batch is their only + /// delivery. Each source therefore returns all of them regardless of its + /// depth (see `tail_with_floor_impl`), and reports whether its buffer + /// already evicted any of them before the read, in which case they are + /// lost from both paths and the caller must say so. + /// + /// Reads never mint a space: with none, both tails are empty and + /// `highest_seq` is `0`. + pub(crate) fn snapshot_tail( + &self, + sandbox_id: &str, + log_max: Option, + platform_max: Option, + subscribed_at: u64, + ) -> TailSnapshot { + let spaces = self.seq.lock(); + let highest_seq = spaces + .get(sandbox_id) + .map_or(0, |space| space.next.saturating_sub(1)); + + let read = |per: Option<&PerSandbox>, max: usize| { + per.map_or_else(SourceTail::default, |per| { + let (events, floor) = tail_with_floor_impl(&per.tail, max, subscribed_at); + SourceTail { + events, + floor, + evicted_while_opening: per.last_trimmed_seq > subscribed_at, + } + }) + }; + + let log = log_max.map_or_else(SourceTail::default, |max| { + let inner = self.inner.lock().expect("tracing bus lock poisoned"); + read(inner.per_id.get(sandbox_id), max) + }); + #[cfg(test)] + self.hooks.run(SnapshotPoint::BetweenReads); + let platform = platform_max.map_or_else(SourceTail::default, |max| { + let inner = self + .platform_event_bus + .inner + .lock() + .expect("platform event bus lock poisoned"); + read(inner.per_id.get(sandbox_id), max) + }); + drop(spaces); + + TailSnapshot { + log, + platform, + highest_seq, + } + } + + /// Subscribe live receivers for the resumable sources and record the + /// space's high-water mark, all under one cursor-space lock hold. + /// + /// Every publish holds this lock across its broadcast send, so the mark + /// splits the space exactly: an event at or below it was sent before + /// these receivers existed and can only be found in a buffer, and one + /// above it is queued on them. [`Self::snapshot_tail`] uses the mark to + /// keep every queued-and-buffered event in the initial batch. + /// + /// The lock order is allocator -> bus map, the same as publish. Never + /// mints a space: with none, the mark is `0`. + pub(crate) fn subscribe_resumable( + &self, + sandbox_id: &str, + want_log: bool, + want_platform: bool, + ) -> ResumableSubscription { + let spaces = self.seq.lock(); + let subscription = ResumableSubscription { + log_rx: want_log.then(|| self.subscribe(sandbox_id)), + platform_rx: want_platform.then(|| self.platform_event_bus.subscribe(sandbox_id)), + subscribed_at: spaces + .get(sandbox_id) + .map_or(0, |space| space.next.saturating_sub(1)), + }; + drop(spaces); + #[cfg(test)] + self.hooks.run(SnapshotPoint::AfterSubscribe); + subscription + } + + /// Install a hook that runs at each [`SnapshotPoint`], replacing any + /// previous one. Shared by every clone of this bus. + #[cfg(test)] + pub(crate) fn set_snapshot_hook(&self, hook: impl FnMut(SnapshotPoint) + Send + 'static) { + *self.hooks.0.lock().expect("snapshot hook lock poisoned") = Some(Box::new(hook)); + } + + /// Whether the cursor-space lock is held right now, by anyone. + #[cfg(test)] + pub(crate) fn cursor_space_locked(&self) -> bool { + matches!( + self.seq.inner.try_lock(), + Err(std::sync::TryLockError::WouldBlock) + ) + } +} + +/// Receivers from [`TracingLogBus::subscribe_resumable`] and the mark that +/// separates buffered-only history from events queued on them. +pub(crate) struct ResumableSubscription { + pub(crate) log_rx: Option>, + pub(crate) platform_rx: Option>, + /// Highest seq the space had issued when the receivers subscribed; `0` + /// with no space. + pub(crate) subscribed_at: u64, +} + +/// Points inside the snapshot paths where a test hook runs. +#[cfg(test)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum SnapshotPoint { + /// After `subscribe_resumable` subscribed and released the lock. + AfterSubscribe, + /// Inside `snapshot_tail` or `snapshot_after`, after the log read and + /// before the platform read, with the cursor-space lock still held. + BetweenReads, +} + +/// Test-only hook slot. Production builds carry no field at all. +#[cfg(test)] +#[derive(Clone, Default)] +#[allow(clippy::type_complexity)] +struct SnapshotHooks(Arc>>>); + +#[cfg(test)] +impl SnapshotHooks { + fn run(&self, point: SnapshotPoint) { + if let Some(hook) = self.0.lock().expect("snapshot hook lock poisoned").as_mut() { + hook(point); + } + } +} + +#[cfg(test)] +impl std::fmt::Debug for SnapshotHooks { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("SnapshotHooks") + } +} + +/// One source's part of a [`TailSnapshot`]. An unread source has no events, +/// a floor of `0`, and nothing evicted. +#[derive(Debug, Default)] +pub(crate) struct SourceTail { + pub(crate) events: Vec, + /// What `tail_with_floor_impl` reports for this source. + pub(crate) floor: u64, + /// The buffer trimmed at least one event published after the receivers + /// subscribed, before the snapshot could read it. Its live copy is + /// suppressed with the rest of the batch's range, so it is lost. + pub(crate) evicted_while_opening: bool, +} + +/// What [`race_publish_between_reads`] observed from inside the snapshot. +#[cfg(test)] +#[derive(Debug, Default)] +pub(crate) struct RaceProbe { + /// The hook ran at all, i.e. a snapshot reached its second read. + pub(crate) fired: bool, + /// The cursor-space lock was held between the two reads. + pub(crate) lock_held_between_reads: bool, + /// The competing publisher completed while the snapshot sat between its + /// reads. With both reads under one lock hold it cannot. + pub(crate) publisher_finished_between_reads: bool, + publisher: Option>, +} + +#[cfg(test)] +impl RaceProbe { + /// Wait for the competing publisher to finish both publishes. + pub(crate) fn join_publisher(&mut self) { + if let Some(handle) = self.publisher.take() { + handle.join().expect("publisher thread panicked"); + } + } +} + +/// Force the interleaving a snapshot must be atomic against: when the next +/// snapshot on `bus` sits between its log read and its platform read, start a +/// publisher that publishes one log line and then one platform event for +/// `sandbox_id`, and give it time to finish before letting the snapshot read +/// the platform bus. +/// +/// With separate reads, the publisher completes in that pause and the platform +/// read picks up its event while the log read missed the log line published +/// just before it. With one lock hold, the publisher blocks until the snapshot +/// is done, and both events land after it. The hook records what it saw +/// rather than asserting, because it may run inside a gateway producer task +/// where a panic would surface only as a truncated stream. +/// +/// The hook fires once. +#[cfg(test)] +pub(crate) fn race_publish_between_reads( + bus: &TracingLogBus, + sandbox_id: &str, +) -> Arc> { + use std::sync::mpsc; + use std::time::Duration; + + let probe = Arc::new(Mutex::new(RaceProbe::default())); + let hook_probe = Arc::clone(&probe); + let publisher_bus = bus.clone(); + let lock_bus = bus.clone(); + let sandbox_id = sandbox_id.to_string(); + // Closure state, not the probe, decides whether the hook already fired: + // a later snapshot must not touch the probe, which the test may be + // holding while it inspects the result. + let mut fired = false; + bus.set_snapshot_hook(move |point| { + if point != SnapshotPoint::BetweenReads || fired { + return; + } + fired = true; + let mut probe = hook_probe.lock().expect("race probe lock poisoned"); + probe.fired = true; + probe.lock_held_between_reads = lock_bus.cursor_space_locked(); + + let (started_tx, started_rx) = mpsc::channel(); + let (done_tx, done_rx) = mpsc::channel(); + let bus = publisher_bus.clone(); + let id = sandbox_id.clone(); + probe.publisher = Some(std::thread::spawn(move || { + started_tx.send(()).expect("hook waits for start"); + bus.publish_external(SandboxLogLine { + sandbox_id: id.clone(), + level: "INFO".to_string(), + target: "race".to_string(), + message: "raced log".to_string(), + source: "gateway".to_string(), + ..Default::default() + }); + bus.platform_event_bus.publish( + &id, + SandboxStreamEvent { + payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Event( + openshell_core::proto::PlatformEvent { + source: "race".to_string(), + r#type: "Normal".to_string(), + reason: "raced".to_string(), + message: "raced event".to_string(), + ..Default::default() + }, + )), + cursor: String::new(), + }, + ); + let _ = done_tx.send(()); + })); + started_rx.recv().expect("publisher thread starts"); + // Long enough for an unblocked publisher to finish two publishes many + // times over. A blocked one never does, so the correct implementation + // always pays the full wait; a broken one fails reliably. + probe.publisher_finished_between_reads = + done_rx.recv_timeout(Duration::from_millis(300)).is_ok(); + }); + probe +} + +/// Outcome of [`TracingLogBus::snapshot_tail`]. +pub(crate) struct TailSnapshot { + pub(crate) log: SourceTail, + pub(crate) platform: SourceTail, + /// Highest seq the space had issued at the lock hold; `0` with no space. + pub(crate) highest_seq: u64, +} + +/// Outcome of [`TracingLogBus::snapshot_after`]. +pub(crate) enum ResumeSnapshot { + /// No cursor space exists for this sandbox, or its epoch no longer + /// matches the resume cursor's -- the space that issued it is gone. + SpaceGone, + /// The right space, but the cursor is ahead of anything it has issued. + /// Only a fabricated cursor reaches this. + CursorAhead, + /// Both requested reads succeeded at the lock hold; each is + /// independently either the replay or the gap that made it unreplayable. + Read(Option, Option), } #[derive(Debug, Clone)] @@ -592,6 +990,324 @@ mod tests { assert!(tail_after_impl(&tail, 0, 99).expect("ok").is_empty()); } + #[test] + fn tail_with_floor_impl_empty_tail_returns_zero_floor() { + let tail = VecDeque::new(); + let (events, floor) = tail_with_floor_impl(&tail, 10, u64::MAX); + assert!(events.is_empty()); + assert_eq!(floor, 0); + } + + #[test] + fn tail_with_floor_impl_max_covers_everything_returns_zero_floor() { + let tail = tail_of(1, 5); + let (events, floor) = tail_with_floor_impl(&tail, 5, u64::MAX); + assert_eq!(cursors(&events), vec![1, 2, 3, 4, 5]); + assert_eq!(floor, 0); + + // Asking for more than exists is the same as asking for everything. + let (events, floor) = tail_with_floor_impl(&tail, 99, u64::MAX); + assert_eq!(cursors(&events), vec![1, 2, 3, 4, 5]); + assert_eq!(floor, 0); + } + + #[test] + fn tail_with_floor_impl_truncated_reports_newest_excluded_seq() { + let tail = tail_of(1, 5); + // Newest 2 returned (4, 5); 1..=3 excluded, newest of those is 3. + // Every returned event sits above the floor, so a bus never + // withholds its own delivered window. + let (events, floor) = tail_with_floor_impl(&tail, 2, u64::MAX); + assert_eq!(cursors(&events), vec![4, 5]); + assert_eq!(floor, 3); + assert!(events.iter().all(|cursored| cursored.seq > floor)); + } + + #[test] + fn tail_with_floor_impl_zero_max_excludes_everything() { + let tail = tail_of(1, 5); + // Nothing returned; every event is excluded, so the floor is the + // newest of them, 5. + let (events, floor) = tail_with_floor_impl(&tail, 0, u64::MAX); + assert!(events.is_empty()); + assert_eq!(floor, 5); + } + + /// The resume read is atomic against a publish on either bus. The forced + /// interleaving publishes a log line and then a platform event while + /// `snapshot_after` sits between its two reads. If the reads were + /// separate, the platform read would return the event while the log read + /// had already missed the line before it: a client handed that event's + /// cursor would resume past the line forever. Under one lock hold the + /// publisher waits, the replay holds neither, and both arrive live in + /// cursor order and are replayable from the replay's last cursor. + #[test] + fn snapshot_after_holds_the_lock_across_a_publish_between_reads() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-race-resume"; + bus.publish_external(make_log_event(sandbox_id, "l1")); // seq 1 + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); // seq 2 + let epoch = bus.cursor_space(sandbox_id).expect("space exists").epoch; + let sub = bus.subscribe_resumable(sandbox_id, true, true); + let (mut log_rx, mut platform_rx) = (sub.log_rx.unwrap(), sub.platform_rx.unwrap()); + + let probe = race_publish_between_reads(&bus, sandbox_id); + let ResumeSnapshot::Read(log, platform) = + bus.snapshot_after(sandbox_id, epoch, 0, 0, true, true) + else { + panic!("expected a successful read"); + }; + { + let mut probe = probe.lock().unwrap(); + probe.join_publisher(); + assert!(probe.fired, "the hook must run between the two reads"); + assert!(probe.lock_held_between_reads, "lock released between reads"); + assert!( + !probe.publisher_finished_between_reads, + "a publish landed between the log read and the platform read" + ); + } + + let log = log.expect("followed").expect("no gap"); + let platform = platform.expect("followed").expect("no gap"); + assert_eq!(cursors(&log), vec![1]); + assert_eq!(cursors(&platform), vec![2]); + + // The raced events come live, in order, above the replay. + assert_eq!(log_rx.try_recv().expect("raced log is live").seq, 3); + assert_eq!(platform_rx.try_recv().expect("raced event is live").seq, 4); + + // Resuming from the replay's highest cursor recovers both. + let ResumeSnapshot::Read(log, platform) = + bus.snapshot_after(sandbox_id, epoch, 2, 2, true, true) + else { + panic!("expected a successful read"); + }; + assert_eq!(cursors(&log.unwrap().unwrap()), vec![3]); + assert_eq!(cursors(&platform.unwrap().unwrap()), vec![4]); + } + + /// The same forced interleaving on the initial path. Under one lock hold + /// the raced events are newer than the snapshot's high-water mark, so the + /// caller's cutoff lets their live copies through, in order. + #[test] + fn snapshot_tail_holds_the_lock_across_a_publish_between_reads() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-race-tail"; + bus.publish_external(make_log_event(sandbox_id, "l1")); // seq 1 + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); // seq 2 + let sub = bus.subscribe_resumable(sandbox_id, true, true); + let (mut log_rx, mut platform_rx) = (sub.log_rx.unwrap(), sub.platform_rx.unwrap()); + + let probe = race_publish_between_reads(&bus, sandbox_id); + let snap = bus.snapshot_tail(sandbox_id, Some(10), Some(10), sub.subscribed_at); + let mut probe = probe.lock().unwrap(); + probe.join_publisher(); + assert!(probe.fired, "the hook must run between the two reads"); + assert!(probe.lock_held_between_reads, "lock released between reads"); + assert!( + !probe.publisher_finished_between_reads, + "a publish landed between the log read and the platform read" + ); + + assert_eq!(cursors(&snap.log.events), vec![1]); + assert_eq!(cursors(&snap.platform.events), vec![2]); + assert_eq!(snap.highest_seq, 2); + + let live_log = log_rx.try_recv().expect("raced log is live"); + let live_event = platform_rx.try_recv().expect("raced event is live"); + assert_eq!((live_log.seq, live_event.seq), (3, 4)); + assert!(live_log.seq > snap.highest_seq && live_event.seq > snap.highest_seq); + } + + /// `subscribe_resumable` records the mark under the lock every publish + /// holds across its send, so an event at or below it is never queued on + /// the new receivers and one above it always is. + #[test] + fn subscribe_resumable_splits_buffered_history_from_queued_events() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-sub"; + assert_eq!( + bus.subscribe_resumable(sandbox_id, true, true) + .subscribed_at, + 0 + ); + assert_eq!(bus.cursor_space(sandbox_id), None, "must not mint a space"); + + bus.publish_external(make_log_event(sandbox_id, "l1")); // seq 1 + let sub = bus.subscribe_resumable(sandbox_id, true, false); + assert_eq!(sub.subscribed_at, 1); + assert!(sub.platform_rx.is_none()); + let mut log_rx = sub.log_rx.unwrap(); + assert!(log_rx.try_recv().is_err(), "history is not queued live"); + + bus.publish_external(make_log_event(sandbox_id, "l2")); // seq 2 + assert_eq!(log_rx.try_recv().unwrap().seq, 2); + } + + /// The reviewer's example at bus level: subscribe, then platform event 1 + /// and log 2 are published before the snapshot. Both are queued live, and + /// the caller will suppress those copies, so the snapshot must return both + /// even though `event_tail` asked for no platform history -- and with + /// nothing excluded, neither source reports a floor. + #[test] + fn snapshot_tail_keeps_events_published_after_subscribe_regardless_of_depth() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-opening"; + let sub = bus.subscribe_resumable(sandbox_id, true, true); + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); // seq 1 + bus.publish_external(make_log_event(sandbox_id, "l2")); // seq 2 + + let snap = bus.snapshot_tail(sandbox_id, Some(200), Some(0), sub.subscribed_at); + assert_eq!(cursors(&snap.log.events), vec![2]); + assert_eq!(cursors(&snap.platform.events), vec![1]); + assert_eq!((snap.log.floor, snap.platform.floor), (0, 0)); + assert!(!snap.log.evicted_while_opening && !snap.platform.evicted_while_opening); + assert_eq!(snap.highest_seq, 2); + } + + /// The exemption covers only events after the mark: older history is + /// still bounded by depth and still reports its floor. + #[test] + fn tail_with_floor_impl_exempts_only_events_after_keep_after() { + let tail = tail_of(1, 6); + // Depth 1 would return just 6; everything after 3 is exempt. + let (events, floor) = tail_with_floor_impl(&tail, 1, 3); + assert_eq!(cursors(&events), vec![4, 5, 6]); + assert_eq!(floor, 3); + + // A depth deeper than the exempt run still reaches into history. + let (events, floor) = tail_with_floor_impl(&tail, 4, 5); + assert_eq!(cursors(&events), vec![3, 4, 5, 6]); + assert_eq!(floor, 2); + + // Nothing after the mark: plain depth bound. + let (events, floor) = tail_with_floor_impl(&tail, 0, 6); + assert!(events.is_empty()); + assert_eq!(floor, 6); + } + + /// If the buffer trimmed an event published after subscribe before the + /// snapshot read it, that event is gone from both paths; the source says + /// so for the caller to report. + #[test] + fn snapshot_tail_reports_eviction_of_events_published_after_subscribe() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-opening-evict"; + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); // seq 1 + let sub = bus.subscribe_resumable(sandbox_id, true, true); + for _ in 0..=PlatformEventBus::DEFAULT_TAIL { + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); + } + + let snap = bus.snapshot_tail(sandbox_id, Some(10), Some(0), sub.subscribed_at); + assert!(snap.platform.evicted_while_opening); + assert!(!snap.log.evicted_while_opening); + assert_eq!(snap.platform.events.len(), PlatformEventBus::DEFAULT_TAIL); + + // History trimmed before subscribe is ordinary truncation. + let late = bus.subscribe_resumable(sandbox_id, true, true); + let snap = bus.snapshot_tail(sandbox_id, Some(10), Some(0), late.subscribed_at); + assert!(!snap.platform.evicted_while_opening); + assert!(snap.platform.events.is_empty()); + } + + /// Baseline correctness: interleaved log/platform publishes draw from the + /// same cursor space, and `snapshot_after` returns each source's own + /// events under its own key, not merged or cross-contaminated. + #[test] + fn snapshot_after_matches_independent_tail_after_calls() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-snap"; + bus.publish_external(make_log_event(sandbox_id, "l1")); // seq 1 + bus.platform_event_bus.publish( + sandbox_id, + SandboxStreamEvent { + payload: None, + cursor: String::new(), + }, + ); // seq 2 + bus.publish_external(make_log_event(sandbox_id, "l2")); // seq 3 + let epoch = bus.cursor_space(sandbox_id).expect("space exists").epoch; + + let ResumeSnapshot::Read(log, platform) = + bus.snapshot_after(sandbox_id, epoch, 0, 0, true, true) + else { + panic!("expected a successful read"); + }; + assert_eq!( + cursors(&log.expect("followed").expect("no gap")), + vec![1, 3] + ); + assert_eq!( + cursors(&platform.expect("followed").expect("no gap")), + vec![2] + ); + } + + /// A source that isn't followed isn't read at all. + #[test] + fn snapshot_after_skips_unfollowed_sources() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-skip"; + bus.publish_external(make_log_event(sandbox_id, "l1")); + let epoch = bus.cursor_space(sandbox_id).expect("space exists").epoch; + + let ResumeSnapshot::Read(log, platform) = + bus.snapshot_after(sandbox_id, epoch, 0, 0, true, false) + else { + panic!("expected a successful read"); + }; + assert_eq!(log.expect("followed").expect("no gap").len(), 1); + assert!(platform.is_none()); + } + + /// The epoch check and both bus reads happen under one lock hold, so a + /// teardown plus republish between them can't apply a stale epoch's + /// cursor to the replacement space's buffers. + #[test] + fn snapshot_after_rejects_a_cursor_from_a_retired_space() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-retired"; + bus.publish_external(make_log_event(sandbox_id, "l1")); + let original_epoch = bus.cursor_space(sandbox_id).expect("space exists").epoch; + + bus.remove(sandbox_id); + bus.publish_external(make_log_event(sandbox_id, "l1-again")); // seq 1 again, new epoch + + assert!(matches!( + bus.snapshot_after(sandbox_id, original_epoch, 0, 0, true, false), + ResumeSnapshot::SpaceGone + )); + } + + /// No space at all -- an unpublished or never-existent sandbox -- is the + /// same outcome as a retired one: there's nothing for the cursor to + /// address. + #[test] + fn snapshot_after_rejects_when_no_space_exists() { + let bus = TracingLogBus::new(); + assert!(matches!( + bus.snapshot_after("nope", Uuid::nil(), 0, 0, true, false), + ResumeSnapshot::SpaceGone + )); + } + + /// A cursor ahead of anything the space has issued -- only reachable with + /// a fabricated token -- is rejected rather than treated as caught up. + #[test] + fn snapshot_after_rejects_a_cursor_ahead_of_the_space() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-ahead"; + bus.publish_external(make_log_event(sandbox_id, "l1")); + let epoch = bus.cursor_space(sandbox_id).expect("space exists").epoch; + + assert!(matches!( + bus.snapshot_after(sandbox_id, epoch, 99, 0, true, false), + ResumeSnapshot::CursorAhead + )); + } + #[test] fn tail_after_impl_boundary_at_last_trimmed_is_serviceable() { // Bus trimmed up to seq 2, retains 3..=5. Client saw exactly 2, so @@ -627,56 +1343,132 @@ mod tests { } #[test] - fn tracing_log_bus_tail_after_serviceable_and_missing() { + fn tracing_log_bus_resume_replay_serviceable_and_missing() { let bus = TracingLogBus::new(); let sandbox_id = "sb-ta"; for _ in 0..3 { bus.publish_external(make_log_event(sandbox_id, "line")); } + let epoch = bus.cursor_space(sandbox_id).expect("space exists").epoch; + // Cursors start at 1, so three publishes are seqs 1,2,3. + let ResumeSnapshot::Read(log, _) = bus.snapshot_after(sandbox_id, epoch, 0, 0, true, false) + else { + panic!("expected a successful read"); + }; assert_eq!( - cursors(&bus.tail_after(sandbox_id, 0).unwrap()), + cursors(&log.expect("followed").expect("no gap")), vec![1, 2, 3] ); - assert_eq!(cursors(&bus.tail_after(sandbox_id, 2).unwrap()), vec![3]); - // Unknown sandbox: no entry, nothing buffered, no gap. - assert!(bus.tail_after("nope", 5).unwrap().is_empty()); + let ResumeSnapshot::Read(log, _) = bus.snapshot_after(sandbox_id, epoch, 2, 0, true, false) + else { + panic!("expected a successful read"); + }; + assert_eq!(cursors(&log.expect("followed").expect("no gap")), vec![3]); } #[test] - fn platform_event_bus_tail_after_serviceable() { + fn platform_event_bus_resume_replay_serviceable() { let bus = TracingLogBus::new(); let platform = &bus.platform_event_bus; let sandbox_id = "sb-pe"; for _ in 0..3 { platform.publish(sandbox_id, stream_event(0)); } + let epoch = bus.cursor_space(sandbox_id).expect("space exists").epoch; + // Shared allocator, but only the platform bus published here, so its // seqs are 1,2,3. + let ResumeSnapshot::Read(_, platform_replay) = + bus.snapshot_after(sandbox_id, epoch, 0, 0, false, true) + else { + panic!("expected a successful read"); + }; assert_eq!( - cursors(&platform.tail_after(sandbox_id, 0).unwrap()), + cursors(&platform_replay.expect("followed").expect("no gap")), vec![1, 2, 3] ); + let ResumeSnapshot::Read(_, platform_replay) = + bus.snapshot_after(sandbox_id, epoch, 0, 1, false, true) + else { + panic!("expected a successful read"); + }; assert_eq!( - cursors(&platform.tail_after(sandbox_id, 1).unwrap()), + cursors(&platform_replay.expect("followed").expect("no gap")), vec![2, 3] ); } #[test] - fn shared_allocator_interleaves_cursors_across_buses() { + fn snapshot_tail_reports_truncation() { let bus = TracingLogBus::new(); - let sandbox_id = "sb-mix"; - // Interleave log and platform publishes; the shared allocator gives - // each a unique, increasing cursor in one merged space. - bus.publish_external(make_log_event(sandbox_id, "a")); // seq 1 - bus.platform_event_bus.publish(sandbox_id, stream_event(0)); // seq 2 - bus.publish_external(make_log_event(sandbox_id, "b")); // seq 3 + let sandbox_id = "sb-floor"; + for _ in 0..5 { + bus.publish_external(make_log_event(sandbox_id, "line")); + } + let snap = bus.snapshot_tail(sandbox_id, Some(2), None, u64::MAX); + assert_eq!(cursors(&snap.log.events), vec![4, 5]); + assert_eq!(snap.log.floor, 3); + + // Wide enough to cover everything: no floor. + let snap = bus.snapshot_tail(sandbox_id, Some(10), None, u64::MAX); + assert_eq!(cursors(&snap.log.events), vec![1, 2, 3, 4, 5]); + assert_eq!(snap.log.floor, 0); + } + + /// `highest_seq` is the space's high-water mark at the lock hold, not the + /// newest event either window returned: events a depth excluded still + /// count, since they were buffered when the tails were read and can only + /// reach a live receiver as a duplicate of pre-snapshot history. + #[test] + fn snapshot_tail_reports_the_high_water_mark_past_excluded_events() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-hwm"; + bus.publish_external(make_log_event(sandbox_id, "l1")); // seq 1 + bus.publish_external(make_log_event(sandbox_id, "l2")); // seq 2 + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); // seq 3 + + let snap = bus.snapshot_tail(sandbox_id, Some(10), Some(0), u64::MAX); + assert_eq!(cursors(&snap.log.events), vec![1, 2]); + assert_eq!(snap.log.floor, 0); + assert!(snap.platform.events.is_empty()); + assert_eq!(snap.platform.floor, 3); + assert_eq!(snap.highest_seq, 3); + } + + /// An unread source is skipped outright; a sandbox with nothing published + /// yields empty tails and a zero mark without minting a cursor space. + #[test] + fn snapshot_tail_skips_unread_sources_and_never_mints_a_space() { + let bus = TracingLogBus::new(); + let sandbox_id = "sb-tail-skip"; + bus.publish_external(make_log_event(sandbox_id, "l1")); + bus.platform_event_bus.publish(sandbox_id, stream_event(0)); + + let snap = bus.snapshot_tail(sandbox_id, Some(10), None, u64::MAX); + assert_eq!(cursors(&snap.log.events), vec![1]); + assert!(snap.platform.events.is_empty()); + assert_eq!(snap.platform.floor, 0); + assert_eq!(snap.highest_seq, 2); - let logs = cursors(&bus.tail_after(sandbox_id, 0).unwrap()); - let events = cursors(&bus.platform_event_bus.tail_after(sandbox_id, 0).unwrap()); - assert_eq!(logs, vec![1, 3]); - assert_eq!(events, vec![2]); + let snap = bus.snapshot_tail("nope", Some(10), Some(10), u64::MAX); + assert!(snap.log.events.is_empty() && snap.platform.events.is_empty()); + assert_eq!(snap.highest_seq, 0); + assert_eq!(bus.cursor_space("nope"), None); + } + + #[test] + fn platform_event_bus_tail_with_floor_reports_truncation() { + let bus = TracingLogBus::new(); + let platform = &bus.platform_event_bus; + let sandbox_id = "sb-pe-floor"; + for _ in 0..5 { + platform.publish(sandbox_id, stream_event(0)); + } + let (events, floor) = platform.tail_with_floor(sandbox_id, 0); + assert!(events.is_empty()); + // Nothing returned: floor is the newest of everything excluded, 5. + assert_eq!(floor, 5); } #[test] @@ -760,7 +1552,9 @@ mod tests { assert_eq!(after_platform.highest_seq, 2); let log_cursor = &bus.tail(sandbox_id, 10)[0].event.cursor; - let platform_cursor = &bus.platform_event_bus.tail(sandbox_id, 10)[0].event.cursor; + let platform_cursor = &bus.platform_event_bus.tail_with_floor(sandbox_id, 10).0[0] + .event + .cursor; assert_eq!( WatchCursor::parse(log_cursor).expect("valid").epoch, WatchCursor::parse(platform_cursor).expect("valid").epoch, @@ -947,8 +1741,9 @@ mod tests { } // Tail should return all events in order - let events = bus.tail(sandbox_id, 10); + let (events, floor) = bus.tail_with_floor(sandbox_id, 10); assert_eq!(events.len(), 5); + assert_eq!(floor, 0, "max covered everything retained"); // Verify order (oldest first) for (i, cursored) in events.iter().enumerate() { @@ -960,8 +1755,9 @@ mod tests { } // Tail with smaller max should return most recent events - let events = bus.tail(sandbox_id, 2); + let (events, floor) = bus.tail_with_floor(sandbox_id, 2); assert_eq!(events.len(), 2); + assert_eq!(floor, 3, "newest excluded event is Event2 (seq 3)"); if let Some(sandbox_stream_event::Payload::Event(ref e)) = events[0].event.payload { assert_eq!(e.reason, "Event3"); } @@ -973,8 +1769,9 @@ mod tests { #[test] fn platform_event_bus_tail_empty_sandbox() { let bus = PlatformEventBus::new(SeqAllocator::default()); - let events = bus.tail("nonexistent", 10); + let (events, floor) = bus.tail_with_floor("nonexistent", 10); assert!(events.is_empty()); + assert_eq!(floor, 0); } #[test] @@ -987,10 +1784,10 @@ mod tests { cursor: String::new(), }; bus.publish(sandbox_id, evt); - assert_eq!(bus.tail(sandbox_id, 10).len(), 1); + assert_eq!(bus.tail_with_floor(sandbox_id, 10).0.len(), 1); bus.remove(sandbox_id); - assert!(bus.tail(sandbox_id, 10).is_empty()); + assert!(bus.tail_with_floor(sandbox_id, 10).0.is_empty()); } } @@ -1057,28 +1854,19 @@ impl PlatformEventBus { } } - /// Return buffered platform events for replay to late subscribers. - pub(crate) fn tail(&self, sandbox_id: &str, max: usize) -> Vec { - let inner = self.inner.lock().expect("platform event bus lock poisoned"); - inner - .per_id - .get(sandbox_id) - .map(|d| d.tail.iter().rev().take(max).cloned().collect::>()) - .unwrap_or_default() - .into_iter() - .rev() - .collect() - } - - pub(crate) fn tail_after( + /// Return buffered platform events and the coverage floor `max` leaves + /// behind. See `tail_with_floor_impl`. Production reads go through + /// `TracingLogBus::snapshot_tail`, which holds the cursor-space lock. + #[cfg(test)] + pub(crate) fn tail_with_floor( &self, sandbox_id: &str, - after_seq: u64, - ) -> Result, ResumeGap> { + max: usize, + ) -> (Vec, u64) { let inner = self.inner.lock().expect("platform event bus lock poisoned"); inner.per_id.get(sandbox_id).map_or_else( - || Ok(Vec::new()), - |per| tail_after_impl(&per.tail, per.last_trimmed_seq, after_seq), + || (Vec::new(), 0), + |per| tail_with_floor_impl(&per.tail, max, u64::MAX), ) } diff --git a/docs/observability/accessing-logs.mdx b/docs/observability/accessing-logs.mdx index 9e19c637bf..e671f48558 100644 --- a/docs/observability/accessing-logs.mdx +++ b/docs/observability/accessing-logs.mdx @@ -76,6 +76,10 @@ On reconnect, a client passes the highest cursor it processed as the resume poin The gateway merges the log and platform event sources before emitting, so events normally arrive in ascending cursor order. That applies to the buffered tail you receive when the stream opens, to a replay after a resume, and to live delivery. The gateway does not delay an event to wait for a lower cursor that has not been published yet, so a cursor can still arrive late under concurrent publication. Track the highest cursor seen as the resume point rather than the last one received. +A cursor is one position shared by the log and platform sources; you track a single highest value, not one per source. If you request both logs and platform events, either source can deliver fewer events on connect than its replay depth asked for. This happens when the other source's replay leaves out an event that is newer than part of this source's history, even when both depths are equal. The gateway withholds those events rather than hand out a cursor that one of the two sources cannot fully vouch for, and reports what it withheld with a warning event rather than dropping it silently. Watches that follow only logs, such as `openshell logs`, are not affected. + +Replay depths bound history only. An event published while the stream is opening is always part of the initial batch, so the batch can hold more events than requested. If the gateway's buffer evicts such an event before the batch is read, the stream reports it with a warning instead. + ## Direct Filesystem Access The supervisor writes shorthand logs to `/var/log/openshell.YYYY-MM-DD.log` and, when enabled, OCSF JSONL to `/var/log/openshell-ocsf.YYYY-MM-DD.jsonl` in its own filesystem. `openshell sandbox exec` runs in the workload and does not give access to the supervisor's files. diff --git a/proto/openshell.proto b/proto/openshell.proto index b40ec68263..24c57498b0 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -2144,9 +2144,17 @@ message WatchSandboxRequest { bool follow_events = 4; // Replay the last N log lines (best-effort) before following. + // + // When following both sources, the replay can return fewer lines: events + // older than one the other source's replay left out are withheld and + // reported with a warning, even if both depths are equal. + // + // The depth bounds history only. Events published while the stream is + // opening are always included, so the replay can return more than N. uint32 log_tail_lines = 5; // Replay the last N platform events (best-effort) before following. + // Defaults to 0. Can return fewer or more events; see log_tail_lines. uint32 event_tail = 6; // Stop streaming once the sandbox reaches READY or a terminal result phase @@ -2180,6 +2188,8 @@ message WatchSandboxRequest { // empty resume_after_cursor, because retrying the same token fails // identically. A cursor this server could not have issued is rejected with // INVALID_ARGUMENT. + // + // One cursor covers both log and platform events. string resume_after_cursor = 12; }