fix(api): unblock SSE watch backlog replay

[duyetbot]

Summary

• Return the SSE response before awaiting backpressured backlog writes; run replay through ctx.waitUntil.
• Release blocked writes and clean up streams on cancellation or replay failure.
• Add real Workers regression tests for non-empty replay, reconnect cursors, live delivery, and cancellation.

Closes #360

Verification

• Targeted Workers tests: 5 passed.
• Biome check on changed source/test files: passed.
• API TypeScript check: passed.

Scope

• Concurrent replay/live ordering ([api] SSE backlog replay interleaves and duplicates events, breaking the ordering contract #366) remains separate.
• New PR; draft fix(api): buffer live events during SSE backlog replay for ordered delivery #361 and release-please chore(main): release 0.2.1 #442 are untouched.
• Do not merge or deploy as part of this task.

Co-Authored-By: Duyet Le me@duyet.net
Co-Authored-By: duyetbot bot@duyet.net

🤖 Generated with Claude Code

Summary by Sourcery

Unblock SSE watch response delivery while safely handling background backlog replay and stream cancellation.

Bug Fixes:

• Prevent SSE watch requests with backpressured backlog replay from hanging before returning a response.
• Release blocked replay writes and clean up streams when clients disconnect or replay fails.

Tests:

• Add Workers regression coverage for backlog replay, reconnect cursors, live notifications, empty backlogs, and cancellation/reconnection.

[sourcery-ai]

Sorry @duyetbot, you've used your own review budget of 250,000 diff characters for the last 7 days.

You can request another review in 1 day and 2 hours by commenting @sourcery-ai review. Upgrade to get a review now.

[coderabbitai]

Important

• 🔍 Trigger review

This repository does not receive automatic reviews because it has fewer than 10 stars.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Advanced

Run ID: 528fd2dd-1a25-4cbf-b4b9-fbd7073b1bcd

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[sourcery-ai]

Reviewer's Guide

The PR fixes SSE watch requests hanging on backpressured backlog replay by returning the response before replay writes, running replay via Durable Object waitUntil, and reliably cleaning up on cancellation or failure. Real Workers tests cover backlog replay, reconnect cursors, live delivery, and cancellation.

Sequence diagram for non-blocking SSE backlog replay

sequenceDiagram
    participant Client
    participant DurableObject as DurableObject
    participant Stream as TransformStream
    participant WaitUntil as ctx.waitUntil

    Client->>DurableObject: watch(projectId, after, signal)
    DurableObject->>Stream: create stream and writer
    DurableObject->>WaitUntil: writeBacklog(writer, projectId, after)
    DurableObject-->>Client: Response(stream.readable)
    WaitUntil->>Stream: writer.write(backlog event)
    Stream-->>Client: replayed SSE event
    DurableObject->>Stream: writer.write(live event)
    Stream-->>Client: live SSE event

Loading

Sequence diagram for SSE cancellation cleanup

sequenceDiagram
    participant Client
    participant DurableObject as DurableObject
    participant Stream as TransformStream
    participant Writer

    Client->>DurableObject: watch(projectId, after, signal)
    DurableObject->>Stream: create stream and writer
    DurableObject-->>Client: Response(stream.readable)
    DurableObject->>Writer: writeBacklog(writer, projectId, after)
    Client->>DurableObject: abort signal
    DurableObject->>Stream: controller.error(signal.reason)
    DurableObject->>Writer: cleanup writer and heartbeat
    Writer-->>DurableObject: release backpressured write

Loading

File-Level Changes

Change	Details	Files
Return watch responses immediately and run backlog replay asynchronously with cancellation and failure cleanup.	Capture the stream controller so blocked writes can be released by erroring the readable. Schedule replay through Durable Object waitUntil instead of awaiting it before returning the response. Centralize cleanup for heartbeat timers, writer tracking, abort listeners, and stream closure. Handle client cancellation and replay/write failures without leaving blocked streams or writers behind.	packages/api/src/state-stream-hub.ts
Add end-to-end Workers coverage for replay, reconnect, live delivery, and cancellation behavior.	Exercise real API-to-Durable-Object watch connections with non-empty and empty backlogs. Verify replay cursor filtering and subsequent live notifications. Verify cancellation before backlog consumption does not prevent later reconnect replay. Add timeout guards and frame-level SSE assertions for response headers, IDs, event types, and JSON payloads.	packages/api/test/state-stream-hub.test.ts
Document the targeted streaming follow-up and its explicitly separate scope.	Record completion of the watch backlog fix and regression coverage. Keep concurrent replay/live ordering and unrelated draft or release changes out of scope.	PLAN.md

Assessment against linked issues

Issue	Objective	Addressed	Explanation
#360	Ensure the real Durable Object /watch endpoint returns a response promptly even when the backlog is non-empty, avoiding the circular backpressure deadlock.	✅
#360	Perform backlog replay asynchronously after returning the response, while preserving correct cleanup and abort handling for cancelled requests or replay failures.	✅
#360	Add end-to-end Workers regression coverage for non-empty backlog replay through the real non-once /watch path, including reconnect cursors, live delivery, and cancellation.	✅

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.