Fix event persistence race conditions on session resume#76
Merged
Conversation
- Make next_seq() atomic by inlining MAX(seq) subquery into INSERT VALUES, eliminating the two-round-trip race that caused UNIQUE constraint failures - Add reader_cancellation (Arc<Notify>) field and cancel_inflight() helper to OpenCodeSdkProvider for per-task reader cancellation - Wire cancel_inflight() into subscribe_and_spawn(), abort_turn(), and shutdown() so stale reader tasks are stopped before new ones are spawned - Surface persistence errors as ProviderEvent::Error through the WebSocket channel in conversations.rs and agent_runs.rs broadcast tasks instead of silently logging and continuing - Add test_persist_event_concurrent: 20 concurrent persist_event calls verify unique, monotonically increasing seq values - Add test_reader_task_cancellation: verify Notify signal exits reader task
- Replace with (opencode_sdk_provider.rs:383) - Remove redundant inner clones in agent_runs.rs broadcast task by changing to in the inner block - Remove unnecessary clone in cleanup section
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Fixes two interdependent race conditions that cause
UNIQUE constraint failed: messages.project_key, messages.session_id, messages.seqerrors when resuming conversations, making conversations inert.Root Causes Fixed
Reader task not cancelled on resume —
subscribe_and_spawn()created a new event subscription and reader task on every call but never cancelled the old one. Both ran concurrently, both trying to persist events for the same session, causing unique constraint violations. Also causedMaxListenersExceededWarningfrom stale SSE subscriptions.next_seq()not atomic —persist_event()used a separateSELECT MAX(seq)query thenINSERT, creating a race window where two calls could get the same seq value.Changes
Phase 1: Rebase (no-op)
origin/mainPhase 2: Atomic seq generation (
src/services/transcript.rs)next_seq()functionVALUES, making seq generation and insertion atomicPhase 3: Reader task cancellation (
src/providers/opencode_sdk_provider.rs)reader_cancellation: Mutex<Option<Arc<tokio::sync::Notify>>>fieldcancel_inflight()helper that signals the old reader task to stopreader_stop.notified()branch to reader task'stokio::select!abort_turn()andshutdown()to usecancel_inflight()Phase 4: Surface persistence errors (
conversations.rs,agent_runs.rs)tracing::warn!+ continue totracing::error!+ broadcast error event via WebSocket + break loopTesting
test_persist_event_concurrent: 20 concurrent calls produce unique seq values (no violations)test_reader_task_cancellation: Notify signal correctly cancels reader taskcargo fmtandcargo clippyclean