feat(streams): add actor subscriptions and durable stream delivery

[NathanFlurry]

No description provided.

[NathanFlurry]

Stack for rivet-dev/rivet

• #5861 feat(streams): add actor subscriptions and durable stream de … 👈
• #5862 feat(auth): add JWT signing prerequisites for actor streams
• main

Get stack: forklift get 5861
Push local edits: forklift submit
Merge when ready: forklift merge 5861

_{change vvvunzlw}

[claude]

Review: feat(streams): add actor subscriptions and durable stream delivery

This is a large PR (about 100 files). I read the auth, guard, and api-public pieces, plus the core stream task and subscription config. I did not do a line-by-line pass over the TS and NAPI layers or the docs. I did not build the PR or run any tests.

What looks good

• Layering: Delivery, checkpointing, and dead-letter logic live in rivetkit-core, as CLAUDE.md requires.
• Authorization: Stream auth is bound to a deterministic storage key. storage_target checks that SHA-256(tenant, path) equals the actor key before it picks a resource kind, so a caller cannot pair an arbitrary path with another stream's key. Close requires both Close and Append. Fork requires Read on the parent, and the parent key is re-derived and validated.
• Provisioning: api-public/.../stream_access.rs requires Append before an actor can be created, and it checks that the input matches the authorized identity exactly (two fields only).
• Tests and schemas: The auth grants schema is a new versioned auth.grants.v2.bare, not an edit of the old one. The tests live under tests/ and are wired in through #[path] shims.
• Error handling: Wake-up handling rejects with Busy through a try_lock_owned, so DS can retry the notification.

Issues and suggestions

1. Missing error artifact. stream.not_paused.json exists only under rivetkit-rust/engine/artifacts/errors/. The other three stream.* errors are also in engine/artifacts/errors/. Please check whether this is expected. If it is not, regenerate the artifacts with cargo check and run pnpm docs:gen-api, since CI checks both.
2. Tenant is caller-chosen in the storage key. storage_key(tenant, path) takes tenant from the x-rivet-stream-tenant header. For the ActorStream branch, the grant is on the actor ID parsed from the path suffix. A holder of one actor's stream grant can therefore create several distinct storage actors for the same logical stream by varying tenant. This is not an authorization bypass, because the grant still has to match. It can fragment data and allocate extra actors. Consider binding tenant to the token or namespace, or restricting it to a known value.
3. Path suffix parsing is ambiguous. path.rsplit_once("/actors/") treats any path ending in /actors/<x>/<id> as an actor stream. If <id> fails Id::parse, the code falls through to the generic DurableStream scope. That fails closed, but a comment documenting the intent would help. A user stream whose path merely looks like an actor path will be classified as an actor stream and need actor-scoped grants.
4. Polling timer in stream_wake. The loop { sleep(10s); connection.callback(...) } heartbeat inside the select! is a legitimate periodic timer. It is worth a one-line comment saying so, given the repo's no-polling rule. The #[allow(unreachable_code)] plus an unreachable Ok(()) could be replaced with a tokio::time::interval loop that returns Result<Infallible>.
5. Retry delay sleeps while holding the stream lock. In drain_stream, tokio::time::sleep(next_attempt_at - now) is not raced against cancel or shutdown inside the inner path. It is covered by the outer select! with shutdown, so it is acceptable. If failure.next_attempt_at can be very large (a long backoff), the whole stream stays pinned and the actor stays awake. Please confirm the backoff is bounded, and consider releasing the wake and rescheduling for long delays.
6. Fixed constants in urlpattern matching. NamedStreamSubscription::pattern caps the pattern at 2048 bytes but does not cap the path length passed to resolve_key. The path comes from DS, so it is probably trusted. A cap would still be cheap insurance against regex-style backtracking in URL patterns.
7. fork is always false in data_operation. data_operation(method, closed, false) is called with a hardcoded false, so the fork parameter and its ensure! are dead code. Either drop the parameter or have storage_requests pass the real value.
8. Header-decoding failures. protocol_header errors on bad base64 or UTF-8. Please confirm these surface as 4xx or 1008 close frames, not as 500s, per the WebSocket rejection guidance.

Test coverage

• Unit coverage exists for auth, guard stream_access, manifest keys, stream, dead_letter, and stream_subscription.
• I did not see driver tests covering a full wake, deliver, ack, redrive cycle, or the fork and close authorization combinations at the guard level. The unit test file has 50 lines, so adding a table of method, header, and required-operations cases would help.
• The new internal SQLite queries need sql_efficiency coverage. The PR touches tests/sql_efficiency.rs, which looks right. Please confirm every new SELECT, UPDATE, or DELETE is in the catalog.

Docs and conventions

• The PR description is empty. Please add a short bullet summary.
• The docs additions look consistent with the sidebar changes. Please double-check that none of the new prose uses em dashes.