Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

workspace: remove observation replay

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-11 08:50:42 +09:00
parent 3a0600be0f
commit 8a3fca65f2
No known key found for this signature in database
2 changed files with 41 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
crates/workspace-server/src/observation.rs
View File

@ -225,15 +225,11 @@ impl BackendObservationCursor {
#[derive(Debug, Default)]
struct BackendObservationState {
next_sequence: u64,
history: BTreeMap<ObservationKey, VecDeque<StoredBackendEvent>>,
}
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<ClientWorkerEventWsEnvelope>,
pub runtime_cursor: Option<String>,
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<BackendObservationOpen, ObservationProxyError> {
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,
})
}
}

56
crates/workspace-server/src/server.rs
View File

@ -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]