Repository navigation
fix: preserve SSE resume cache and correct event replay - #1165
freedom-winds wants to merge 3 commits into
Conversation
github-actions
Bot
added
T-core
| let sync_index = index.saturating_sub(front_event_id.index); | ||
| // Last-Event-ID is the last event the client received. The resume is | ||
| // valid as long as the next event has not already fallen out of cache. | ||
| let next_index = index.saturating_add(1); |
There was a problem hiding this comment.
A standalone GET emits a synthetic priming event with ID 0, but CachedTx::new_common also assigns 0 to the first cached common event. How can a client that only received the priming event resume without sync(0) skipping that unseen notification?
| assert!(worker.tx_router[&http_request_id].completed_at.is_some()); | ||
| assert!(worker.tx_router[&http_request_id].resources.is_empty()); |
There was a problem hiding this comment.
Could this regression exercise resume and assert that the cached final response is delivered?
|
This pull request has had no activity for 60 days and is now marked as stale. It will be closed in 14 days if there is no further activity. Add the |
| assert!(worker.tx_router[&http_request_id].completed_at.is_some()); | ||
| assert!(worker.tx_router[&http_request_id].resources.is_empty()); |
There was a problem hiding this comment.
The name suggests the cache survives a resume, but the assertions only check bookkeeping fields. So a regression that drops the cache would still pass. Could we resume from the priming event and assert that the final error response is replayed?
| match last_event_index { | ||
| Some(last_event_index) => self.common.sync(last_event_index).await?, | ||
| None if self.event_store.is_none() => self.common.replay_cached().await?, | ||
| None => {} | ||
| } |
There was a problem hiding this comment.
A fresh GET without a Last-Event-ID replays the entire cache, including events the previous primary already delivered. Is that redelivery intentional, given that Last-Event-ID is the spec's way to recover missed events?
| async fn replay_cached(&mut self) -> Result<(), SessionError> { | ||
| for message in &self.cache { | ||
| let send_result = self.tx.send(message.clone()).await; | ||
| if send_result.is_err() { | ||
| let event_id: EventId = message.event_id.as_deref().unwrap_or_default().parse()?; | ||
| return Err(SessionError::ChannelClosed(Some(event_id.index as u64))); | ||
| } | ||
| } | ||
| Ok(()) | ||
| } |
There was a problem hiding this comment.
This duplicates the replay loop in sync, including its error mapping. Could we have both use a shared replay_from(skip) method?
Then sync could end with self.replay_from(sync_index).await, and the common-channel call site could use self.common.replay_from(0).
| let next_index = index.saturating_add(1); | ||
| if next_index < front_event_id.index { | ||
| return Err(SessionError::InvalidEventId); | ||
| } | ||
| let sync_index = next_index.saturating_sub(front_event_id.index); |
There was a problem hiding this comment.
checked_sub handles the bounds check and subtraction in one step, so we no longer need saturating_sub.
| let next_index = index.saturating_add(1); | |
| if next_index < front_event_id.index { | |
| return Err(SessionError::InvalidEventId); | |
| } | |
| let sync_index = next_index.saturating_sub(front_event_id.index); | |
| let Some(sync_index) = index.saturating_add(1).checked_sub(front_event_id.index) else { | |
| return Err(SessionError::InvalidEventId); | |
| }; |
| assert!(matches!( | ||
| cached_tx.sync(3).await, | ||
| Err(SessionError::InvalidEventId) | ||
| )); | ||
| cached_tx.sync(4).await.unwrap(); | ||
|
|
||
| assert_eq!(rx.recv().await.unwrap().event_id.as_deref(), Some("5")); | ||
| assert_eq!(rx.recv().await.unwrap().event_id.as_deref(), Some("6")); | ||
| } |
There was a problem hiding this comment.
This test checks two behaviors, rejection and acceptance, and nothing directly covers the headline fix that the Last-Event-ID event itself is not replayed. Could the rejection case get its own test, plus one that resumes from 5 and gets only 6?
| assert!(matches!( | |
| cached_tx.sync(3).await, | |
| Err(SessionError::InvalidEventId) | |
| )); | |
| cached_tx.sync(4).await.unwrap(); | |
| assert_eq!(rx.recv().await.unwrap().event_id.as_deref(), Some("5")); | |
| assert_eq!(rx.recv().await.unwrap().event_id.as_deref(), Some("6")); | |
| } | |
| cached_tx.sync(4).await.unwrap(); | |
| assert_eq!(rx.recv().await.unwrap().event_id.as_deref(), Some("5")); | |
| assert_eq!(rx.recv().await.unwrap().event_id.as_deref(), Some("6")); | |
| } | |
| #[tokio::test] | |
| async fn resume_skips_event_matching_last_event_id() { | |
| let (tx, mut rx) = tokio::sync::mpsc::channel(4); | |
| let mut cached_tx = CachedTx::new(tx, None, 0, "test-stream".into(), None); | |
| cached_tx.cache.push_back(ServerSseMessage { | |
| event_id: Some("5".into()), | |
| ..Default::default() | |
| }); | |
| cached_tx.cache.push_back(ServerSseMessage { | |
| event_id: Some("6".into()), | |
| ..Default::default() | |
| }); | |
| cached_tx.sync(5).await.unwrap(); | |
| drop(cached_tx); | |
| assert_eq!(rx.recv().await.unwrap().event_id.as_deref(), Some("6")); | |
| assert!(rx.recv().await.is_none()); | |
| } | |
| #[tokio::test] | |
| async fn resume_rejects_event_older_than_cache_front() { | |
| let (tx, _rx) = tokio::sync::mpsc::channel(4); | |
| let mut cached_tx = CachedTx::new(tx, None, 0, "test-stream".into(), None); | |
| cached_tx.cache.push_back(ServerSseMessage { | |
| event_id: Some("5".into()), | |
| ..Default::default() | |
| }); | |
| let result = cached_tx.sync(3).await; | |
| assert!(matches!(result, Err(SessionError::InvalidEventId))); | |
| } |
Summary
Fix two issues in the Streamable HTTP server's SSE resume behavior:
Last-Event-IDwas replayed again.Changes
Last-Event-ID.completed_cache_ttlto evict completed request caches.