Repository navigation
feat(email): dual-write non-inbound Mailbox state - #1124
Merged
Merged
Conversation
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
📝 WalkthroughWalkthroughThe PR wires non-inbound D1-to-Mailbox mirroring for outbound messages, provider delivery events, and classification updates. It adds bounded delivery-event batch RPCs, graph orchestration, timeout and telemetry outcomes, asynchronous queue integration, and migration documentation. ChangesMailbox live mirroring
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant OutboundOrClassifier
participant D1
participant MailboxLiveMirror
participant Mailbox
participant AnalyticsEngine
OutboundOrClassifier->>D1: persist outbound state or classification
OutboundOrClassifier->>MailboxLiveMirror: invoke graph mirroring
MailboxLiveMirror->>D1: load authoritative message graph
MailboxLiveMirror->>Mailbox: mirror message, thread, attachments, and event batch
MailboxLiveMirror->>AnalyticsEngine: record mirror outcome
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
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. Comment |
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
kody-bot
marked this pull request as ready for review
August 1, 2026 10:53
Contributor
|
🔎 Preview deployed: https://kody-pr-1124.kody-a99.workers.dev Worker: Mocks:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/worker/src/email/delivery-queue.ts (1)
34-78: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winMailbox mirror never runs for two persisted-event outcomes:
stale-but-inserted and post-dispatch-failureduplicate.The mailbox graph repair only runs when
result.outcome === 'recorded'(lines 66-75). Two situations bypass this even though D1 has already changed:
recordProviderEmailDeliveryEventreturnsstaleboth when nothing was inserted and when a new event row was inserted but did not win the "latest status" race (inserted === true,updatedLatestStatus === false). In the second case, a newemail_delivery_eventsrow exists in D1, but the queue handler treats it identically to the no-op case and never mirrors it.- If
dispatchEmailDeliverySubscriptionEventsthrows after a successfulrecordedwrite, the outercatchretries the queue message before the mirror call executes. On retry, the same provider event now conflicts in D1 and resolves toduplicate(sincemessage.deliveryStatus/deliveryStatusAtwere already updated by the first attempt).duplicatenever triggers the mirror, so that terminal delivery event is permanently absent from the Mailbox mirror.Because
mirrorMailboxMessageGraphFromD1performs a full graph repair from D1 (not an incremental single-event mirror), extending the trigger tostaleandduplicateoutcomes closes both gaps at negligible extra cost, consistent with the mirror's best-effort design.♻️ Proposed fix to also repair on stale/duplicate outcomes
case 'duplicate': + case 'stale': case 'recorded': { // Abuse evaluation runs before subscription dispatch (and // also for replayed duplicates) so a crash or dispatch // failure can never skip the pause; the pause write itself // is idempotent. await applyOutboundEmailAbusePause({ env, userId: result.message.userId, deliveryStatus: result.providerEvent.payload.delivery.status, eventRecorded: result.outcome === 'recorded', }) - await dispatchEmailDeliverySubscriptionEvents({ - env, - message: result.message, - providerEvent: result.providerEvent, - waitUntil, - }) - if (result.outcome === 'recorded' && result.message != null) { + if (result.outcome !== 'stale') { + await dispatchEmailDeliverySubscriptionEvents({ + env, + message: result.message, + providerEvent: result.providerEvent, + waitUntil, + }) + } + if (result.message != null) { waitUntil( mirrorMailboxMessageGraphFromD1({ env, db: env.APP_DB, userId: result.message.userId, messageId: result.message.id, }), ) } queueMessage.ack() break }Note: this keeps
staleout of subscription dispatch (preserving existing dispatch semantics) while still catching it up via the mirror repair. Confirm thestalecase should keep the pre-existing (non-mirror) behavior for dispatch before merging this diff verbatim.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/worker/src/email/delivery-queue.ts` around lines 34 - 78, Extend the mailbox mirror trigger in the queue handler to also run for `stale` and `duplicate` outcomes, while preserving the existing stale-case behavior of skipping subscription dispatch. Invoke `mirrorMailboxMessageGraphFromD1` when a persisted event may have changed D1, using the existing message, userId, and messageId guards; keep the current `recorded` mirroring and best-effort `waitUntil` flow intact.
🧹 Nitpick comments (4)
packages/worker/src/email/mailbox-types.ts (1)
23-24: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider an independent constant for the batch bound.
mailboxUpsertDeliveryEventsMaxderives frommailboxDefaultPageSize. The two limits control unrelated concerns: read pagination and the transactional write batch size. A future page-size tuning change would silently change the DO batch bound and the live-mirror truncation threshold (mailboxLiveMirrorMaxEvents). Define a dedicated numeric constant instead.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/worker/src/email/mailbox-types.ts` around lines 23 - 24, The mailboxUpsertDeliveryEventsMax limit incorrectly reuses mailboxDefaultPageSize, coupling read pagination with transactional write batching and mailboxLiveMirrorMaxEvents. Define mailboxUpsertDeliveryEventsMax as its own explicit numeric constant in mailbox-types.ts, leaving mailboxDefaultPageSize independent.packages/worker/src/email/mailbox-snapshot-repo.ts (1)
129-155: 🚀 Performance & Scalability | 🔵 TrivialConfirm an index supports this owner-scoped ordered lookup.
The query filters on
message_idanduser_idand orders bycreated_at DESC, id DESC. Without a composite index onemail_delivery_events (user_id, message_id, created_at), D1 sorts the matching rows for every mirror call. Mirroring runs on outbound terminal paths, provider-event repair, and classification updates, so this query is on hot paths.#!/bin/bash # Inspect declared indexes on email_delivery_events across migrations and schema files. fd -t f -e sql | xargs rg -n -i -C2 'email_delivery_events' rg -n -i -C2 'email_delivery_events' --glob '*schema*'🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/worker/src/email/mailbox-snapshot-repo.ts` around lines 129 - 155, Ensure the database schema declares a composite index for the lookup in listEmailDeliveryEventMirrorProjectionsForMessage, covering user_id, message_id, and created_at in the query’s filtering and ordering sequence. Add the index through the project’s established migration or schema mechanism, preserving the existing query behavior and tie-breaking by id.packages/worker/src/email/mailbox-live-mirror.node.test.ts (1)
473-495: 🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick winAdd a cross-owner isolation test for the graph mirror.
The suite covers missing messages, truncation, timeout, and error mapping. It does not cover a foreign owner. Seed
msg-1foruser-aaa, then callmirrorMailboxMessageGraphFromD1withuserId: 'user-bbb'. Assert the result is{ status: 'missing' }and that neithermirrorMessagenorupsertDeliveryEventsis called. This locks the per-user isolation contract for the new mirror path.As per coding guidelines: "Every signed-in user must have a fully isolated personal assistant, including separate packages, jobs, secrets, values, memories, remote connectors, email inboxes, and durable storage."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/worker/src/email/mailbox-live-mirror.node.test.ts` around lines 473 - 495, Add a cross-owner isolation test alongside the existing missing-message test: seed msg-1 for user-aaa, invoke mirrorMailboxMessageGraphFromD1 with user-bbb, and assert the response reports status missing. Also verify both mirrorMessage and upsertDeliveryEvents are not called, preserving per-user isolation for the graph mirror path.Source: Coding guidelines
packages/worker/src/email/mailbox-live-mirror.ts (1)
146-158: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueConsider loading attachments and the thread concurrently.
The message load, the attachment load, and the thread load run in sequence. The attachment load and the thread load do not depend on each other. Both run on outbound send paths, so each round trip adds latency to the caller. Await them together after the message resolves.
♻️ Proposed concurrent load
- const attachments = await listEmailAttachmentsForMessage({ - db: input.db, - messageId: input.messageId, - }) - - let thread = input.thread - if (thread === undefined && message.threadId) { - thread = await getEmailThreadById({ - db: input.db, - userId: input.userId, - threadId: message.threadId, - }) - } + const threadId = message.threadId + const [attachments, loadedThread] = await Promise.all([ + listEmailAttachmentsForMessage({ + db: input.db, + messageId: input.messageId, + }), + input.thread === undefined && threadId + ? getEmailThreadById({ + db: input.db, + userId: input.userId, + threadId, + }) + : Promise.resolve(input.thread), + ]) + const thread = loadedThread🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/worker/src/email/mailbox-live-mirror.ts` around lines 146 - 158, Update the message-processing flow around listEmailAttachmentsForMessage and getEmailThreadById so the independent attachment and thread loads start concurrently after the message resolves, using a shared await rather than sequential awaits. Preserve the existing conditional behavior for loading the thread only when input.thread is undefined and message.threadId is present.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/contributing/architecture/data-storage.md`:
- Around line 759-770: Update the Analytics Engine budget statement in the “Live
graph orchestrator” documentation to say that each graph call performs at most
two writes, preserving the existing explanation that these are one message
outcome write and one batch outcome write.
In `@packages/worker/src/email/mailbox-live-mirror.ts`:
- Around line 170-179: When `eventsTruncated` is true in the mailbox delivery
event mirroring flow, emit a `console.warn` or existing parity telemetry signal
before returning the summary, including enough context to identify the affected
message and owner. Keep the existing event slicing and return behavior
unchanged, and anchor the change in the function containing `loadedEvents` and
`eventsTruncated` rather than the callers that ignore the summary.
In `@packages/worker/src/email/outbound-mailbox-mirror.workers.test.ts`:
- Around line 32-41: Update the outbound mailbox test stub to include the
missing upsertDeliveryEvents method alongside the existing RPC methods, and add
mailbox-mirror-delivery-event-batch-failed to the warning suppression used by
the relevant tests. Preserve the existing hang and throw behaviors so batch
timeout and error-path assertions exercise the intended
mirrorMailboxDeliveryEventSnapshots flow.
In `@packages/worker/src/email/outbound.ts`:
- Around line 652-658: Move best-effort mailbox mirroring off request threads by
adding an optional waitUntil/ExecutionContext flow: in
packages/worker/src/email/outbound.ts at lines 652-658 and terminals 697, 765,
and 801, accept it through SendEmailEnv/EmailSendInput and schedule
mirrorMailboxGraph() instead of awaiting it, ensuring it runs outside
withAccountWriteLease; in packages/worker/src/email/service.ts at lines 735-740,
add optional waitUntil to setEmailMessageClassification, schedule
mirrorMailboxMessageGraphFromD1 with it, and pass it from both the account-email
handler and email_message_classify capability.
---
Outside diff comments:
In `@packages/worker/src/email/delivery-queue.ts`:
- Around line 34-78: Extend the mailbox mirror trigger in the queue handler to
also run for `stale` and `duplicate` outcomes, while preserving the existing
stale-case behavior of skipping subscription dispatch. Invoke
`mirrorMailboxMessageGraphFromD1` when a persisted event may have changed D1,
using the existing message, userId, and messageId guards; keep the current
`recorded` mirroring and best-effort `waitUntil` flow intact.
---
Nitpick comments:
In `@packages/worker/src/email/mailbox-live-mirror.node.test.ts`:
- Around line 473-495: Add a cross-owner isolation test alongside the existing
missing-message test: seed msg-1 for user-aaa, invoke
mirrorMailboxMessageGraphFromD1 with user-bbb, and assert the response reports
status missing. Also verify both mirrorMessage and upsertDeliveryEvents are not
called, preserving per-user isolation for the graph mirror path.
In `@packages/worker/src/email/mailbox-live-mirror.ts`:
- Around line 146-158: Update the message-processing flow around
listEmailAttachmentsForMessage and getEmailThreadById so the independent
attachment and thread loads start concurrently after the message resolves, using
a shared await rather than sequential awaits. Preserve the existing conditional
behavior for loading the thread only when input.thread is undefined and
message.threadId is present.
In `@packages/worker/src/email/mailbox-snapshot-repo.ts`:
- Around line 129-155: Ensure the database schema declares a composite index for
the lookup in listEmailDeliveryEventMirrorProjectionsForMessage, covering
user_id, message_id, and created_at in the query’s filtering and ordering
sequence. Add the index through the project’s established migration or schema
mechanism, preserving the existing query behavior and tie-breaking by id.
In `@packages/worker/src/email/mailbox-types.ts`:
- Around line 23-24: The mailboxUpsertDeliveryEventsMax limit incorrectly reuses
mailboxDefaultPageSize, coupling read pagination with transactional write
batching and mailboxLiveMirrorMaxEvents. Define mailboxUpsertDeliveryEventsMax
as its own explicit numeric constant in mailbox-types.ts, leaving
mailboxDefaultPageSize independent.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: e02edb40-8778-472b-bc2d-3e97b3a5dd71
📒 Files selected for processing (26)
docs/contributing/architecture/data-storage.mdpackages/worker/src/app/handlers/account-email.node.test.tspackages/worker/src/app/handlers/account-email.tspackages/worker/src/email/delivery-events.tspackages/worker/src/email/delivery-events.workers.test.tspackages/worker/src/email/delivery-queue.node.test.tspackages/worker/src/email/delivery-queue.tspackages/worker/src/email/mailbox-do.tspackages/worker/src/email/mailbox-do.workers.test.tspackages/worker/src/email/mailbox-live-mirror.node.test.tspackages/worker/src/email/mailbox-live-mirror.tspackages/worker/src/email/mailbox-mirror.node.test.tspackages/worker/src/email/mailbox-mirror.tspackages/worker/src/email/mailbox-parity-events.node.test.tspackages/worker/src/email/mailbox-parity-events.tspackages/worker/src/email/mailbox-snapshot-repo.node.test.tspackages/worker/src/email/mailbox-snapshot-repo.tspackages/worker/src/email/mailbox-types.tspackages/worker/src/email/outbound-mailbox-mirror.workers.test.tspackages/worker/src/email/outbound.tspackages/worker/src/email/repo-body-limits.workers.test.tspackages/worker/src/email/repo.tspackages/worker/src/email/service-classification.node.test.tspackages/worker/src/email/service.tspackages/worker/src/mcp/capabilities/email/email-message-classify.node.test.tspackages/worker/src/mcp/capabilities/email/email-message-classify.ts
Comment on lines
+652
to
+658
| const mirrorMailboxGraph = () => | ||
| mirrorOutboundMailboxMessageGraph({ | ||
| env: input.env, | ||
| userId: input.userId, | ||
| messageId: message.id, | ||
| createdThread, | ||
| }) |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Best-effort Mailbox mirroring is awaited on request threads. Both callers await mirrorMailboxMessageGraphFromD1 and discard its summary. That call performs several D1 reads plus two sequential Mailbox RPCs, each capped at mailboxMirrorRpcTimeoutMs, so a slow Mailbox Durable Object adds the full budget to a user-visible response. Schedule the mirror instead of blocking on it.
packages/worker/src/email/outbound.ts#L652-L658: accept anExecutionContext(or awaitUntilcallback) inSendEmailEnv/EmailSendInputand schedulemirrorMailboxGraph()with it at the three terminals (Lines 697, 765, 801), so the mirror never runs insidewithAccountWriteLease.packages/worker/src/email/service.ts#L735-L740: accept an optionalwaitUntilinsetEmailMessageClassificationand schedulemirrorMailboxMessageGraphFromD1on it; pass it from the account-email handler and theemail_message_classifycapability.
📍 Affects 2 files
packages/worker/src/email/outbound.ts#L652-L658(this comment)packages/worker/src/email/service.ts#L735-L740
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/worker/src/email/outbound.ts` around lines 652 - 658, Move
best-effort mailbox mirroring off request threads by adding an optional
waitUntil/ExecutionContext flow: in packages/worker/src/email/outbound.ts at
lines 652-658 and terminals 697, 765, and 801, accept it through
SendEmailEnv/EmailSendInput and schedule mirrorMailboxGraph() instead of
awaiting it, ensuring it runs outside withAccountWriteLease; in
packages/worker/src/email/service.ts at lines 735-740, add optional waitUntil to
setEmailMessageClassification, schedule mirrorMailboxMessageGraphFromD1 with it,
and pass it from both the account-email handler and email_message_classify
capability.
Co-authored-by: Kent C. Dodds <me+github@kentcdodds.com>
This was referenced Aug 1, 2026
This was referenced Aug 2, 2026
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
Wires the first live phase-2 Mailbox mirrors while keeping D1 authoritative:
recorded,duplicate, andstaleoutcomes schedule full graph repair through queuewaitUntilwhen a message is boundD1 remains the sole source of truth. This PR does not touch inbound delivery leases/finalization/effects, retention deletes, deletion fencing, or read authority.
Verification
Focused tests, local/CI/post-merge validation, production health/smoke checks, and independent reviews are green.
System recap — extends existing primitives (medium risk)
Mode: recap · Base:
main@0684b78· Head:d747c2dClassification: extends — live outbound/classification/provider paths compose D1-authoritative writes with the Mailbox shadow and parity stream.
Primitives touched
emailmailboxemail-delivery-queueapp-uiemail-blobs-r2System map
D1 commits remain authoritative; afterward non-inbound paths repair Mailbox and emit outcome telemetry.
Legend: green = composes (wiring only) · amber = extended by this PR · red = new primitive · gray = context (unchanged, included only when an edge crosses it).
Invariants
Conductor report
Summary by CodeRabbit
New Features
Bug Fixes
Documentation