diff --git a/crates/workspace-server/src/observation.rs b/crates/workspace-server/src/observation.rs index 2be74240..2d46eef3 100644 --- a/crates/workspace-server/src/observation.rs +++ b/crates/workspace-server/src/observation.rs @@ -225,15 +225,11 @@ impl BackendObservationCursor { #[derive(Debug, Default)] struct BackendObservationState { next_sequence: u64, - history: BTreeMap>, } impl BackendObservationState { fn new() -> Self { - Self { - next_sequence: 1, - history: BTreeMap::new(), - } + Self { next_sequence: 1 } } } @@ -243,20 +239,6 @@ struct ObservationKey { worker_id: String, } -#[derive(Clone, Debug)] -struct StoredBackendEvent { - sequence: u64, - runtime_cursor: String, - envelope: ClientWorkerEventWsEnvelope, -} - -#[derive(Clone, Debug)] -pub struct BackendObservationOpen { - pub replay: Vec, - pub runtime_cursor: Option, - pub backend_cursor: BackendObservationCursor, -} - /// Backend-owned in-memory v0 observation proxy state. #[derive(Clone)] pub struct BackendObservationProxy { @@ -314,62 +296,18 @@ impl BackendObservationProxy { pub fn open( &self, - runtime_id: &str, - worker_id: &str, + _runtime_id: &str, + _worker_id: &str, cursor: Option<&str>, - ) -> Result { - let key = ObservationKey { - runtime_id: runtime_id.to_string(), - worker_id: worker_id.to_string(), - }; - let cursor = match cursor { - Some(raw) => BackendObservationCursor::decode(raw).ok_or_else(|| { + ) -> Result<(), ObservationProxyError> { + if let Some(raw) = cursor { + BackendObservationCursor::decode(raw).ok_or_else(|| { ObservationProxyError::CursorMalformed(format!( "malformed backend observation cursor: {raw}" )) - })?, - None => BackendObservationCursor::zero(), - }; - let state = self.state.lock().map_err(|_| { - ObservationProxyError::RuntimeUnavailable( - "backend observation state lock poisoned".into(), - ) - })?; - let history = state.history.get(&key); - let replay: Vec<_> = history - .into_iter() - .flat_map(|events| events.iter()) - .filter(|event| event.sequence > cursor.sequence) - .cloned() - .collect(); - if cursor.sequence != 0 { - let found = history - .into_iter() - .flat_map(|events| events.iter()) - .any(|event| event.sequence == cursor.sequence); - if !found { - return Err(ObservationProxyError::CursorUnknownOrExpired(format!( - "backend observation cursor {} is unknown or expired for runtime {runtime_id} worker {worker_id}", - cursor.encode() - ))); - } + })?; } - let runtime_cursor = replay - .last() - .map(|event| event.runtime_cursor.clone()) - .or_else(|| { - history.and_then(|events| { - events - .iter() - .find(|event| event.sequence == cursor.sequence) - .map(|event| event.runtime_cursor.clone()) - }) - }); - Ok(BackendObservationOpen { - replay: replay.into_iter().map(|event| event.envelope).collect(), - runtime_cursor, - backend_cursor: cursor, - }) + Ok(()) } pub fn store( @@ -384,27 +322,13 @@ impl BackendObservationProxy { let sequence = state.next_sequence; state.next_sequence += 1; let cursor = BackendObservationCursor::new(sequence).encode(); - let envelope = ClientWorkerEventWsEnvelope { + Ok(ClientWorkerEventWsEnvelope { cursor: cursor.clone(), event_id: cursor, - runtime_id: event.runtime_id.clone(), - worker_id: event.worker_id.clone(), - payload: event.payload, - }; - let key = ObservationKey { runtime_id: event.runtime_id, worker_id: event.worker_id, - }; - let history = state.history.entry(key).or_default(); - history.push_back(StoredBackendEvent { - sequence, - runtime_cursor: event.runtime_cursor, - envelope: envelope.clone(), - }); - while history.len() > 1024 { - history.pop_front(); - } - Ok(envelope) + payload: event.payload, + }) } } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 60b210f9..0d08e29e 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2981,12 +2981,18 @@ async fn worker_observation_ws_session( query: ClientWorkerEventsWsQuery, mut socket: WebSocket, ) { - let open = match proxy.open( + if let Err(error) = proxy.open( source.runtime_id(), source.worker_id(), query.cursor.as_deref(), ) { - Ok(open) => open, + let _ = + send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::diagnostic(error)).await; + return; + } + + let mut upstream = match RuntimeObservationClient::connect(&source, None).await { + Ok(client) => client, Err(error) => { let _ = send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::diagnostic(error)) .await; @@ -2994,26 +3000,6 @@ async fn worker_observation_ws_session( } }; - let mut backend_cursor = open.backend_cursor; - for envelope in open.replay { - backend_cursor = crate::observation::BackendObservationCursor::decode(&envelope.cursor) - .unwrap_or(backend_cursor); - if !send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::event(envelope)).await { - return; - } - } - - let mut upstream = - match RuntimeObservationClient::connect(&source, open.runtime_cursor.as_deref()).await { - Ok(client) => client, - Err(error) => { - let _ = - send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::diagnostic(error)) - .await; - return; - } - }; - loop { tokio::select! { inbound = socket.next() => { @@ -3049,8 +3035,6 @@ async fn worker_observation_ws_session( match upstream_event { Ok(event) => match proxy.store(event) { Ok(envelope) => { - backend_cursor = crate::observation::BackendObservationCursor::decode(&envelope.cursor) - .unwrap_or(backend_cursor); if !send_client_ws_frame(&mut socket, ClientWorkerEventWsFrame::event(envelope)).await { return; } @@ -7262,6 +7246,26 @@ mod tests { assert_eq!(live.worker_id, "worker-a"); assert!(matches!(live.payload, protocol::Event::TextDelta { .. })); + let (mut fresh, _) = connect_async(&url).await.unwrap(); + let fresh_snapshot = next_client_frame(&mut fresh).await; + assert!(matches!( + fresh_snapshot, + ClientWorkerEventWsFrame::Event { envelope } if matches!(envelope.payload, protocol::Event::Snapshot { .. }) + )); + runtime + .observe_worker_event( + &worker_ref, + protocol::Event::TextDone { + text: "fresh".into(), + }, + ) + .unwrap(); + let fresh_event = next_client_frame(&mut fresh).await; + assert!(matches!( + fresh_event, + ClientWorkerEventWsFrame::Event { envelope } if matches!(envelope.payload, protocol::Event::TextDone { .. }) + )); + let (mut resumed, _) = connect_async(format!("{url}?cursor={}", live.cursor)) .await .unwrap(); @@ -7309,7 +7313,7 @@ mod tests { } #[tokio::test] - async fn proxy_reports_unknown_backend_cursor_before_upstream_connect() { + async fn proxy_does_not_validate_backend_cursor_against_replay_history() { let source = RuntimeObservationSourceConfig { runtime_id: "runtime-a".into(), worker_id: "worker-a".into(), @@ -7321,7 +7325,7 @@ mod tests { .await .unwrap(); let diagnostic = next_client_diagnostic(&mut stream).await; - assert_eq!(diagnostic.code, "backend.cursor_unknown_or_expired"); + assert_eq!(diagnostic.code, "backend.runtime_unavailable"); } #[tokio::test]