diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 5b9325626..233cc473e 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -205,7 +205,13 @@ jobs:
go test -race ./internal/repository -run '^TestLive(LastWarmupSendFailure|PersistentMailboxLoadingFailure|DispatchRetry|LineageQueueRevisions|WarmupRecoversOldPreparation|WarmupPreparationRecoveryRetains)' -count=1
go test -race ./internal/tasks -run '^Test(LiveWarmup(SafetyHoldReschedules|Recovery|TaskDoesNotApplyResults)|WarmupSenderDistinguishes)' -count=1
- - name: Test the Kafka codec and client lifecycle
+ - name: Mailbox sync admission, replay, truthful state and upgrade regressions (PostgreSQL)
+ env:
+ WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
+ run: |
+ go test -race -p 1 ./internal/repository ./internal/app/consumer ./internal/app/orgtransfer ./internal/infrastructure/db -run '^TestLive(SyncArrival|SyncTruth|EmailErrorDedupe)' -count=1
+
+ - name: Test the Kafka codec, failed-handler replay and client lifecycle
run: go test -race -tags kafka ./internal/infrastructure/codec/ ./internal/infrastructure/kafka/ ./internal/infrastructure/eventbus/
# fail_ci_if_error does not cover the uploader crashing before it runs
diff --git a/docs/content/docs/development/data-control.mdx b/docs/content/docs/development/data-control.mdx
index fe10a9538..f3412bd31 100644
--- a/docs/content/docs/development/data-control.mdx
+++ b/docs/content/docs/development/data-control.mdx
@@ -281,6 +281,16 @@ The `rsync` alternative works too and is sometimes simpler, as long as the stack
[Workspace export and import](/guides/workspace-export-import/) moves a single organization's data between two running instances, re-sealing its secrets for the destination's keys. That is the per-customer tool; the bundle here is the instance-level one. They are not interchangeable: a bundle cannot be applied to one workspace, and a workspace archive cannot restore an instance.
+### Mailbox sync recovery data
+
+New durable admissions retain an organization-encrypted arrival payload in `sync_arrival_outbox` alongside the provider map. The queue tracks delivery stages and retries consumer failures under the original message identity. A delivered arrival is removed; an unresolved arrival is retained for retry until it completes or its owning mailbox or organization is deleted. Deleting the mailbox intentionally removes that mailbox's pending work. This queue does not schedule new sends or authorize repeating a send.
+
+A full-instance backup must preserve the database, wrapped organization keys, the root key needed to unwrap them, and referenced message-body storage together. Restoring a queue without its keys or body objects cannot restore delivery. Use the normal consistent instance backup, not a separate copy of just the queue. A workspace archive excludes the queue and provider mappings whose arrivals are still pending. Drain pending work on the source before moving a workspace; importing an archive does not resume source-instance delivery.
+
+The consumer logs aggregate pending counts and oldest-arrival age. Arrivals older than five minutes, or delivery/inspection failures continuing for one minute, raise a deduplicated operator alert through the existing operator-notification integration when configured. Logs and notifications contain counts and age, not message bodies or addresses. Operator notifications repeat at most once per fifteen minutes; without an operator channel, monitor consumer logs for `sync arrival queue needs attention`. An empty queue is not proof of provider success or historical completeness.
+
+See [upgrade order and downgrade limits](/development/install/#mailbox-sync-upgrade-order). Legacy maps without stored arrival payloads cannot reconstruct historical lost mail.
+
## Outbound calls
### Diagnostic and managed-consent portability
diff --git a/docs/content/docs/development/install.mdx b/docs/content/docs/development/install.mdx
index e6bf5d7cc..bb7173591 100644
--- a/docs/content/docs/development/install.mdx
+++ b/docs/content/docs/development/install.mdx
@@ -294,6 +294,16 @@ Put your own compose changes in a `docker-compose.override.yml`, which compose m
Upgrade the backend first and confirm its database migrations complete before upgrading consumers. Consumers wait up to two minutes for the clean migration version required by their embedded code before starting database-dependent work. If the backend is still old or a migration failed, the consumer exits with an actionable schema-readiness error and can retry through the container restart policy. It does not run migrations or ignore missing tables.
+### Mailbox sync upgrade order
+
+For durable mailbox arrivals, upgrade the backend, let its migrations complete, then upgrade and start the consumer before upgrading workers. The backend admits each new provider map and its organization-encrypted arrival payload in one database transaction; the consumer delivers pending stages after restart. Workers continue using internal HTTP endpoints and do not need database access. Preserve the instance's organization keys and body storage with the database, as described in [data control](/development/data-control/#mailbox-sync-recovery-data).
+
+An old worker with a new backend still writes legacy map-only records. A new worker with an old backend falls back only when the durable internal endpoint is absent (HTTP 404 or 405). Neither mixed-version path has the new durable-admission guarantee. A backend failure or an acknowledgement without the durable marker is not treated as success. When a conflicting map has no retained pending payload, the backend fails closed instead of claiming a durable admission; this includes a retry after the pending row has already been delivered and removed. Ordinary map lookup still deduplicates that message.
+
+Existing completed imports and maps are preserved. Pre-upgrade maps contain no replay payload, so an old lost arrival cannot be reconstructed from the map alone. Bounded provider recovery covers supported recent history, not an unlimited historical repair. [Mailbox sync](/guides/mailboxes/#what-gets-synced) explains the provider limits.
+
+Keep the upgraded consumer running until all pending arrivals finish before downgrading. The migration refuses to remove the arrival table while any pending row remains. Do not delete queued rows or force a migration version to bypass that refusal. An older consumer cannot drain this queue. Pending arrivals are instance-local and do not move in a workspace export.
+
Fleet auto-updates select releases no newer than the running backend. Existing fleet and node pins are checked again when answering heartbeats, so an incompatible stored target is not sent to nodes; new incompatible pins are rejected. Newly joined machines fall back to the backend's exact release, not the moving `prod` tag. Main builds use their published commit-SHA image tag. Unstamped custom builds must set `WARMBLY_VERSION` to their published image tag; unknown versions cannot be ordered safely. Upgrade the backend before moving the fleet to a newer release. This protection only takes effect after installing the compatibility release; it cannot change the behavior of an older backend already running.
The dashboard discovers a separate realtime origin from the backend's existing public configuration at container startup or static build time. Bring the backend up first; if discovery warns that it is unavailable, restart the dashboard after backend readiness or rebuild the static deployment. See [configuration](/development/configuration/).
diff --git a/docs/content/docs/guides/mailboxes.mdx b/docs/content/docs/guides/mailboxes.mdx
index 80b562d78..4913a9607 100644
--- a/docs/content/docs/guides/mailboxes.mdx
+++ b/docs/content/docs/guides/mailboxes.mdx
@@ -207,7 +207,15 @@ Connecting a mailbox does two things: it imports the mailbox's recent history, a
**The import** brings in the last `90` days, newest first, up to `5,000` messages per mailbox (the defaults; a self-hosted instance can change both under Instance settings). It reads the inbox, sent mail and archive (plus your own folders on Gmail and IMAP), and skips trash, drafts, spam and Gmail's "All Mail". You can watch it in the mailbox drawer's **Sync** card: a progress bar with a running count while it works, then "Up to date" with what was imported. Because it runs newest first, the mail you would look for is there within the first minute; a large mailbox finishes over the next while at a steady pace. A few IMAP servers (Seznam.cz, for one) refuse to search a folder by date; on those the import reads each message's arrival date instead, which makes the first pass over a large folder a little slower and changes nothing else.
-**Live sync** then follows every folder the provider exposes, including junk (so placement problems are visible) and sent mail (so a conversation shows both sides). Read state, flags, deletions and moves are mirrored too.
+**Live sync** follows the provider's supported folders, including junk (so placement problems are visible) and sent mail (so a conversation shows both sides). Outlook follows Inbox, Sent, Junk, Drafts and Archive where Archive is available; it does not follow every custom Graph folder. Read state, flags, deletions and moves are mirrored too.
+
+The Sync card distinguishes a completed history import from a recent successful live check. An unfinished pass, provider failure, recovery scan or budget hold does not refresh the last successful check or erase the last observed deferred count. That count is an observation from sync, not a measurement of all mail still on the provider. A worker heartbeat alone is not evidence that a mailbox is current. Repeated connection failures refresh the warning's observation time without generating a new notification each pass; no later successful check means recovery is unverified.
+
+If Gmail rejects an ordinary import continuation token, the import restarts inside its original fixed date window. Already mapped mail is deduplicated, and the existing message cap and sync budgets still apply. Outlook recovers an expired delta checkpoint by replaying the configured recent-history window, reconciling known messages and following changes from a fresh checkpoint only after recovery completes. Older messages never imported are outside that recovery window. Authentication failures, throttles and transport errors are not treated as expired checkpoints.
+
+On IMAP, discovering a new folder or a changed UIDVALIDITY generation after the initial import schedules a bounded recent-history scan. An unreadable or incomplete message body remains retryable rather than being stored as a successful empty message. A folder without usable modification sequences falls back to UIDNEXT checks and periodic flags even if the server advertises CONDSTORE. Failed flag publication keeps the previous baseline so it can retry. Background lookups do not mark unread messages as seen.
+
+On upgraded instances, a newly admitted provider message is recorded together with an encrypted pending arrival before sync advances. The consumer replays that arrival under its original identity after a restart, including related bounce or complaint processing, and provider updates wait for an unresolved arrival. This is an at-least-once recovery path, not a promise that all historical mail can be reconstructed. Old map-only records have no replay payload, and mixed-version instances retain their legacy behavior. See the [self-host upgrade order and limits](/development/install/#mailbox-sync-upgrade-order).
If Gmail expires its history checkpoint after a mailbox has been offline, Warmbly automatically scans the configured recent-history window for missed messages and checks every stored Gmail message for deletions, moves and read-state changes. Recovery resumes across worker restarts and respects the normal sync budget. Messages older than the configured import window that were never stored are not imported by recovery. Once the scan completes, changes made during recovery are followed from a fresh history checkpoint; no reconnection is needed.
@@ -221,12 +229,12 @@ A folder the sync does not recognize files as inbox, so its mail shows up in the
On an IMAP mailbox, the drawer's **Sync** card lists the folders the worker has seen under **Folders not synced**. Tick one and the sync stops opening it on its next pass, within a minute; mail already imported from it is removed from Warmbly, and mail that later moves into it from a synced folder is removed too. The mail itself is never touched at the provider. A folder that is not listed yet, because the mailbox has not synced or the folder was just created, can be typed in by name, exactly as your mail server lists it (`Warmer`, or `INBOX/Warmer` on a server that nests everything under the inbox). Case does not matter, and a skipped folder covers its subfolders.
-The inbox, sent, drafts, spam, trash and archive folders cannot be skipped: a sync without them is a broken mailbox, not a quieter one. Un-ticking a folder brings it back the way a new folder arrives, syncing what lands in it from then on; the mail that accumulated while it was skipped is not imported. Up to `50` folders can be skipped per mailbox. Gmail and Outlook mailboxes have no such list.
+The inbox, sent, drafts, spam, trash and archive folders cannot be skipped: a sync without them is a broken mailbox, not a quieter one. Un-ticking a folder brings it back with a bounded scan of the configured recent-history window; older mail outside that window is not imported. Up to `50` folders can be skipped per mailbox. Gmail and Outlook mailboxes have no such list.
The same setting is `PUT /emails/:id/sync` in the [API](/api/endpoints/) and `warmbly mailbox skip-folders` in the CLI.
- Some IMAP servers, including Outlook.com, Microsoft 365 over IMAP and Yahoo, cannot tell a client what changed since it last looked. New mail from those servers still arrives within a minute. Reading or flagging a message in another mail client shows up in Warmbly within about ten minutes rather than immediately. Nothing is lost either way, and Gmail, Outlook over OAuth, Fastmail and most self-hosted servers are immediate.
+ Some IMAP servers, including Outlook.com, Microsoft 365 over IMAP and Yahoo, cannot tell a client what changed since it last looked. Warmbly checks for new mail on each sync pass and scans read state and flags about every ten minutes. Those changes may take longer to appear than on a provider with usable modification sequences. Provider availability and sync budgets still apply.
diff --git a/docs/content/docs/guides/warmup.mdx b/docs/content/docs/guides/warmup.mdx
index 6e36c3e05..25fad0a52 100644
--- a/docs/content/docs/guides/warmup.mdx
+++ b/docs/content/docs/guides/warmup.mdx
@@ -208,6 +208,8 @@ At the volumes warmup runs, a few plaintext messages of a couple of kilobytes ea
### In the unibox
+Mailbox-sync recovery replays stored arrivals, not sends. Recognized warmup remains hidden, and replay does not create another warmup send or count sync progress as proof that a provider sent mail. Receipt verification and duplicate-send protections remain separate from the mailbox's live-sync status. See [mailbox sync recovery and its limits](/guides/mailboxes/#what-gets-synced).
+
Warmup never appears in the unified inbox, in either direction. Warmbly recognizes it by its token, including for recipient-only mailboxes. A consumed or expired token still identifies warmup for filtering, but cannot trigger engagement again. When the header is missing, a recorded message identifier can identify older warmup during history sync; the sender-and-subject fallback stays limited to recent inbound sends so an ordinary conversation is not hidden just because it shares a subject. Existing verified warmup entries are removed from the unibox automatically in background batches, without deleting mail at the provider or changing warmup standing.
Manual replies to verified diagnostic parents can stay hidden from the unibox and follow housekeeping. They do not authorize another automated turn or count as real campaign replies.
diff --git a/docs/content/docs/guides/workspace-export-import.mdx b/docs/content/docs/guides/workspace-export-import.mdx
index b1e9b73ec..ad39d26ea 100644
--- a/docs/content/docs/guides/workspace-export-import.mdx
+++ b/docs/content/docs/guides/workspace-export-import.mdx
@@ -150,6 +150,8 @@ If you have shell access to the source, `warmblyctl org export` writes the same
## What does not travel
+Pending mailbox arrivals and their delivery queue are instance-local. Provider mappings whose arrivals are still pending are excluded too. Let the source consumer finish its pending arrivals before exporting; importing the archive does not replay that source queue. A [full-instance backup](/development/data-control/#mailbox-sync-recovery-data) instead preserves the queue together with its keys and message-body storage.
+
The Test workspace classification is instance-local. Importing an archive does not classify the destination as Test or grant a managed Test plan or testing credits.
A few things belong to the instance rather than the workspace and are left out on purpose: the instance's link to Warmbly Cloud and which mailboxes it enrolled in the hosted pool ([Warmbly Cloud](/guides/warmbly-cloud/)), linked self-hosted instances on a cloud workspace, Google Workspace and Microsoft 365 admin grants, live sessions, in-flight OAuth handshakes and the organization's wrapped data key. Reconnect to Warmbly Cloud and make each admin grant again after the move.
diff --git a/internal/api/handler/internal_arrival_test.go b/internal/api/handler/internal_arrival_test.go
index fc5e70490..5a7c499a0 100644
--- a/internal/api/handler/internal_arrival_test.go
+++ b/internal/api/handler/internal_arrival_test.go
@@ -41,6 +41,7 @@ func TestInternalArrivalAcknowledgesOnlyCompletedAdmission(t *testing.T) {
ack string
}{
{"success", nil, 204, "1"}, {"failure", errors.New("private payload or cipher failure"), 503, ""},
+ {"unconfirmed collision", repository.ErrArrivalAdmissionUnconfirmed, 503, ""},
} {
t.Run(tc.name, func(t *testing.T) {
m := &internalArrivalMap{err: tc.err}
diff --git a/internal/app/consumer/arrival_outbox.go b/internal/app/consumer/arrival_outbox.go
index 6df1207ee..1bb405685 100644
--- a/internal/app/consumer/arrival_outbox.go
+++ b/internal/app/consumer/arrival_outbox.go
@@ -3,10 +3,12 @@ package jobs
import (
"context"
"fmt"
+ "strconv"
"time"
"github.com/rs/zerolog/log"
"github.com/warmbly/warmbly/internal/models"
+ "github.com/warmbly/warmbly/internal/repository"
)
func (s *JobsService) runArrivalOutbox(ctx context.Context) {
@@ -15,15 +17,17 @@ func (s *JobsService) runArrivalOutbox(ctx context.Context) {
}
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
+ var failedSince, lastAlert time.Time
for {
pass, cancel := context.WithTimeout(ctx, 30*time.Second)
err := s.ArrivalOutbox.DeliverArrivals(pass, func(ctx context.Context, kind models.JobEventType, body any) error {
return s.deliverArrivalEvent(ctx, kind, body)
})
- cancel()
if err != nil {
log.Warn().Msg("sync arrival delivery deferred")
}
+ s.observeArrivalBacklog(pass, err, time.Now(), &failedSince, &lastAlert)
+ cancel()
select {
case <-ctx.Done():
return
@@ -32,6 +36,60 @@ func (s *JobsService) runArrivalOutbox(ctx context.Context) {
}
}
+func (s *JobsService) observeArrivalBacklog(ctx context.Context, deliveryErr error, now time.Time, failedSince, lastAlert *time.Time) {
+ backlog := repository.ArrivalBacklog{}
+ known := false
+ if stats, ok := s.ArrivalOutbox.(repository.ArrivalBacklogRepository); ok {
+ var err error
+ backlog, err = stats.ArrivalBacklog(ctx)
+ if err != nil {
+ deliveryErr = err
+ } else {
+ known = true
+ }
+ }
+ if deliveryErr != nil {
+ if failedSince.IsZero() {
+ *failedSince = now
+ }
+ } else {
+ *failedSince = time.Time{}
+ }
+ age := time.Duration(0)
+ if backlog.Oldest != nil {
+ age = max(0, now.Sub(*backlog.Oldest))
+ }
+ if age < 5*time.Minute && (failedSince.IsZero() || now.Sub(*failedSince) < time.Minute) {
+ return
+ }
+ if !lastAlert.IsZero() && now.Sub(*lastAlert) < 15*time.Minute {
+ return
+ }
+ *lastAlert = now
+ entry := log.Warn().Bool("backlog_known", known)
+ if known {
+ entry.Int("pending", backlog.Pending).Int64("oldest_age_seconds", int64(age.Seconds()))
+ }
+ entry.Msg("sync arrival queue needs attention")
+ if s.OpsNotifier == nil {
+ return
+ }
+ if s.Cache != nil {
+ ok, err := s.Cache.SetNX(ctx, "sync:arrival:opsnotify", "1", 15*time.Minute).Result()
+ if err == nil && !ok {
+ return
+ }
+ }
+ fields := map[string]string{"Pending": "unavailable", "Oldest age seconds": "unavailable"}
+ if known {
+ fields["Pending"] = strconv.Itoa(backlog.Pending)
+ fields["Oldest age seconds"] = strconv.FormatInt(int64(age.Seconds()), 10)
+ }
+ s.OpsNotifier.NotifyOperator("sync.arrival.backlog", "Mailbox arrival delivery delayed",
+ "Stored arrivals are waiting for delivery, or delivery checks are failing. Check the consumer and database before assuming sync is current.",
+ fields)
+}
+
func (s *JobsService) deliverArrivalEvent(ctx context.Context, kind models.JobEventType, body any) error {
if _, ok := s.eventHandlers[kind]; !ok {
return fmt.Errorf("durable arrival handler unavailable: %s", kind)
diff --git a/internal/app/consumer/arrival_outbox_test.go b/internal/app/consumer/arrival_outbox_test.go
index d01513822..857ffaa8b 100644
--- a/internal/app/consumer/arrival_outbox_test.go
+++ b/internal/app/consumer/arrival_outbox_test.go
@@ -2,9 +2,13 @@ package jobs
import (
"context"
+ "errors"
+ "strings"
"testing"
+ "time"
"github.com/warmbly/warmbly/internal/models"
+ "github.com/warmbly/warmbly/internal/repository"
)
func TestDurableArrivalRequiresRegisteredHandler(t *testing.T) {
@@ -19,3 +23,59 @@ func TestDurableArrivalRequiresRegisteredHandler(t *testing.T) {
t.Fatalf("called=%v err=%v", called, err)
}
}
+
+type arrivalBacklogStub struct {
+ repository.ArrivalOutbox
+ backlog repository.ArrivalBacklog
+ err error
+}
+
+func (s *arrivalBacklogStub) ArrivalBacklog(context.Context) (repository.ArrivalBacklog, error) {
+ return s.backlog, s.err
+}
+
+type arrivalOperatorStub struct {
+ calls int
+ content string
+ fields map[string]string
+}
+
+func (s *arrivalOperatorStub) NotifyOperator(key, title, summary string, fields map[string]string) {
+ s.calls++
+ s.content = key + title + summary
+ s.fields = fields
+ if len(fields) != 2 {
+ panic("unexpected private fields")
+ }
+}
+
+func TestDurableArrivalBacklogAlertsAreAgedAndDeduplicated(t *testing.T) {
+ now := time.Now()
+ old := now.Add(-6 * time.Minute)
+ outbox := &arrivalBacklogStub{backlog: repository.ArrivalBacklog{Pending: 8, Oldest: &old}}
+ ops := &arrivalOperatorStub{}
+ s := &JobsService{ArrivalOutbox: outbox, OpsNotifier: ops}
+ var failed, alerted time.Time
+ s.observeArrivalBacklog(t.Context(), nil, now, &failed, &alerted)
+ s.observeArrivalBacklog(t.Context(), nil, now.Add(time.Minute), &failed, &alerted)
+ if ops.calls != 1 {
+ t.Fatalf("repeat alerts=%d", ops.calls)
+ }
+ outbox.backlog = repository.ArrivalBacklog{}
+ s.observeArrivalBacklog(t.Context(), nil, now.Add(16*time.Minute), &failed, &alerted)
+ if ops.calls != 1 {
+ t.Fatal("empty queue claimed ongoing backlog")
+ }
+ outbox.err = errors.New("private database error")
+ s.observeArrivalBacklog(t.Context(), nil, now.Add(17*time.Minute), &failed, &alerted)
+ if ops.calls != 1 {
+ t.Fatal("transient read failure alerted too soon")
+ }
+ s.observeArrivalBacklog(t.Context(), nil, now.Add(18*time.Minute), &failed, &alerted)
+ if ops.calls != 2 {
+ t.Fatal("persistent queue inspection failure was silent")
+ }
+ if strings.Contains(ops.content, "private database error") || ops.fields["Pending"] != "unavailable" || ops.fields["Oldest age seconds"] != "unavailable" {
+ t.Fatal("operator notice exposed an error or fabricated unavailable counts")
+ }
+}
diff --git a/internal/app/worker/wmail/start_sync.go b/internal/app/worker/wmail/start_sync.go
index 8208e2bc9..c4e6b452a 100644
--- a/internal/app/worker/wmail/start_sync.go
+++ b/internal/app/worker/wmail/start_sync.go
@@ -93,12 +93,11 @@ func (w *WMail) syncOnce(ctx context.Context) (result *errx.MailError) {
}
}()
if err := w.SyncMail(ctx); err != nil {
- // A server that is down answers every pass the same way. Report the
- // first one and then stay quiet until it comes back, so one outage is
- // one warning in the drawer rather than one a minute.
+ // Keep observations fresh; the consumer deduplicates user notifications.
if isTransportError(err) {
w.transportFailures++
if w.transportFailures > 1 {
+ w.CaptureError(err)
log.Debug().Err(err).Str("email_id", w.ID.String()).Int("consecutive", w.transportFailures).Msg("mail server still unreachable")
return err
}
diff --git a/internal/app/worker/wmail/sync_truth_test.go b/internal/app/worker/wmail/sync_truth_test.go
index 9bc905bfc..306f2fe48 100644
--- a/internal/app/worker/wmail/sync_truth_test.go
+++ b/internal/app/worker/wmail/sync_truth_test.go
@@ -28,6 +28,27 @@ func seedSyncTruth(w *WMail, relayed *[]models.SyncState) time.Time {
return previous
}
+func TestSyncTruthRepeatedTransportFailuresRelayFreshObservations(t *testing.T) {
+ srv := httptest.NewServer(http.NotFoundHandler())
+ srv.Close()
+ var events []captured
+ w := newGoogleTestMail(t, srv, &events)
+ for i := 0; i < 2; i++ {
+ if err := w.syncOnce(t.Context()); err == nil {
+ t.Fatal("provider failure hidden")
+ }
+ }
+ errors := 0
+ for _, event := range events {
+ if event.eventType == models.JobEventTypeEmailServerError {
+ errors++
+ }
+ }
+ if errors != 2 || w.transportFailures != 2 {
+ t.Fatalf("observations=%d failures=%d", errors, w.transportFailures)
+ }
+}
+
func assertNoSyncSuccess(t *testing.T, w *WMail, previous time.Time) {
t.Helper()
if w.tracker.state.LastSyncedAt == nil || !w.tracker.state.LastSyncedAt.Equal(previous) || w.tracker.state.Deferred != 5 {
diff --git a/internal/client/smtpimap/imap/client.go b/internal/client/smtpimap/imap/client.go
index e5ea1db29..c023f57de 100644
--- a/internal/client/smtpimap/imap/client.go
+++ b/internal/client/smtpimap/imap/client.go
@@ -333,7 +333,11 @@ func (c *Client) Close() error {
if c.transport == nil {
return nil
}
- return c.transport.Close()
+ err := c.transport.Close()
+ if errors.Is(err, net.ErrClosed) {
+ return nil
+ }
+ return err
}
func (c *Client) plainAuth() *errx.MailError {
diff --git a/internal/infrastructure/eventbus/bus.go b/internal/infrastructure/eventbus/bus.go
index 4ff66e0cd..e8c63d68a 100644
--- a/internal/infrastructure/eventbus/bus.go
+++ b/internal/infrastructure/eventbus/bus.go
@@ -72,6 +72,29 @@ type EventBus interface {
// the message; returning an error leaves it for redelivery.
type Handler func(ctx context.Context, msg Message) error
+func retrySubscription(ctx context.Context, subscribe func(context.Context) error) error {
+ for {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ err := subscribe(ctx)
+ if ctx.Err() != nil {
+ return ctx.Err()
+ }
+ if errors.Is(err, ErrBusClosed) {
+ return err
+ }
+ log.Warn().Msg("eventbus subscription interrupted; reopening for replay")
+ timer := time.NewTimer(time.Second)
+ select {
+ case <-ctx.Done():
+ timer.Stop()
+ return ctx.Err()
+ case <-timer.C:
+ }
+ }
+}
+
// Message is the broker-neutral envelope passed to a Handler. Payload bytes
// are owned by the bus and must not be retained past the handler call.
//
@@ -87,9 +110,7 @@ type Message struct {
// handler that retries by returning an error can read it to know when the
// broker is about to stop redelivering and give up cleanly instead.
Attempt int
- // Redelivers is true when a handler error leaves the message for another
- // delivery (NATS). Kafka commits regardless, so a handler must not count
- // on a retry there and should finish what it can on this delivery.
+ // Redelivers is true when a handler error leaves the message for retry.
Redelivers bool
}
diff --git a/internal/infrastructure/eventbus/kafka.go b/internal/infrastructure/eventbus/kafka.go
index 7a16e1000..427d14ed4 100644
--- a/internal/infrastructure/eventbus/kafka.go
+++ b/internal/infrastructure/eventbus/kafka.go
@@ -106,8 +106,7 @@ func (b *KafkaBus) Publish(ctx context.Context, topic, key string, payload []byt
// Subscribe creates a fresh consumer in the given group, subscribes to all
// topics, and blocks reading messages until ctx is cancelled or a fatal error
-// occurs. Handler errors are logged but do not abort the loop; the message's
-// offset is still stored, to match the existing kafka.Consumer.Consume behaviour.
+// occurs. Failed handlers never store offsets; exhausted retries reopen for replay.
func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string, handler Handler) error {
if len(topics) == 0 {
return errors.New("eventbus kafka: at least one topic required")
@@ -118,6 +117,12 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string,
if handler == nil {
return errors.New("eventbus kafka: handler required")
}
+ return retrySubscription(ctx, func(ctx context.Context) error {
+ return b.subscribeOnce(ctx, topics, group, handler)
+ })
+}
+
+func (b *KafkaBus) subscribeOnce(ctx context.Context, topics []string, group string, handler Handler) error {
// Subscribing to a topic that does not exist yet returns no messages and
// no error, so a worker would sit silent rather than fail.
@@ -153,6 +158,17 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string,
}
b.consumers = append(b.consumers, cons)
b.mu.Unlock()
+ defer func() {
+ b.mu.Lock()
+ for i, c := range b.consumers {
+ if c == cons {
+ b.consumers = append(b.consumers[:i], b.consumers[i+1:]...)
+ break
+ }
+ }
+ b.mu.Unlock()
+ cons.Close()
+ }()
err = cons.Consume(ctx, func(msg *ckf.Message) error {
topic := ""
@@ -163,10 +179,11 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string,
hctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), handlerTimeout())
defer cancel()
if err := invokeHandler(hctx, handler, Message{
- Topic: topic,
- Key: string(msg.Key),
- Payload: msg.Value,
- Attempt: 1,
+ Topic: topic,
+ Key: string(msg.Key),
+ Payload: msg.Value,
+ Attempt: 1,
+ Redelivers: true,
}); err != nil {
log.Error().Err(err).Str("topic", topic).Msg("eventbus kafka handler error")
return err
diff --git a/internal/infrastructure/eventbus/kafka_test.go b/internal/infrastructure/eventbus/kafka_test.go
index c7da95b73..23752240b 100644
--- a/internal/infrastructure/eventbus/kafka_test.go
+++ b/internal/infrastructure/eventbus/kafka_test.go
@@ -4,11 +4,55 @@ package eventbus
import (
"context"
+ "errors"
"testing"
"github.com/warmbly/warmbly/internal/infrastructure/kafka"
)
+func TestKafkaSubscriptionReopensAfterFailureOrUnexpectedExit(t *testing.T) {
+ for _, failure := range []error{errors.New("offset assignment lost"), nil} {
+ ctx, cancel := context.WithCancel(t.Context())
+ calls := 0
+ err := retrySubscription(ctx, func(context.Context) error {
+ calls++
+ if calls == 2 {
+ cancel()
+ }
+ return failure
+ })
+ cancel()
+ if !errors.Is(err, context.Canceled) || calls != 2 {
+ t.Fatalf("err=%v calls=%d", err, calls)
+ }
+ }
+ calls := 0
+ err := retrySubscription(t.Context(), func(context.Context) error { calls++; return ErrBusClosed })
+ if !errors.Is(err, ErrBusClosed) || calls != 1 {
+ t.Fatal("closed bus reopened")
+ }
+}
+
+func TestKafkaSubscriptionRetiresClientsBeforeReopening(t *testing.T) {
+ b := &KafkaBus{bootstrap: "127.0.0.1:1"}
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+ for i := 0; i < 3; i++ {
+ if err := b.subscribeOnce(ctx, []string{"sync-test"}, "sync-test", func(context.Context, Message) error { return nil }); !errors.Is(err, context.Canceled) {
+ t.Fatal(err)
+ }
+ b.mu.Lock()
+ retained := len(b.consumers)
+ b.mu.Unlock()
+ if retained != 0 {
+ t.Fatalf("exited subscription retained %d clients", retained)
+ }
+ }
+ if err := b.Close(); err != nil {
+ t.Fatal(err)
+ }
+}
+
// TestKafkaBus_InterfaceSatisfaction is a compile-time check that KafkaBus
// satisfies the EventBus interface and that the constructors enforce their
// minimum required arguments. A live Kafka broker is intentionally not used
diff --git a/internal/infrastructure/kafka/consumer.go b/internal/infrastructure/kafka/consumer.go
index 2a1d9a10e..c1e85883a 100644
--- a/internal/infrastructure/kafka/consumer.go
+++ b/internal/infrastructure/kafka/consumer.go
@@ -14,6 +14,7 @@ import (
)
var ErrClientClosed = errors.New("kafka: client closed")
+var ErrAssignmentLost = errors.New("kafka: assignment lost before offset storage")
type consumerClient interface {
ReadMessage(time.Duration) (*ckf.Message, error)
@@ -23,10 +24,11 @@ type consumerClient interface {
}
type Consumer struct {
- c consumerClient
- Avrov2 *Avrov2
- mu sync.Mutex
- closed bool
+ c consumerClient
+ Avrov2 *Avrov2
+ mu sync.Mutex
+ closed bool
+ retryLimit int
}
type ConsumerConfig struct {
@@ -110,6 +112,9 @@ func (cons *Consumer) storeMessage(msg *ckf.Message) error {
if cons.closed {
return ErrClientClosed
}
+ if native, ok := cons.c.(interface{ AssignmentLost() bool }); ok && native.AssignmentLost() {
+ return ErrAssignmentLost
+ }
_, err := cons.c.StoreMessage(msg)
return err
}
@@ -126,6 +131,9 @@ func (cons *Consumer) Consume(ctx context.Context, handler func(msg *ckf.Message
if kafkaErr.Code() == ckf.ErrTimedOut {
continue
}
+ if kafkaErr.IsFatal() || kafkaErr.Code() == ckf.ErrMaxPollExceeded {
+ return fmt.Errorf("kafka: consumer must reopen: %w", err)
+ }
// Log transient Kafka errors and retry after a brief delay
log.Warn().Str("code", fmt.Sprintf("%d", kafkaErr.Code())).Err(kafkaErr).Msg("kafka consumer error")
if err := consumerBackoff(ctx, time.Second); err != nil {
@@ -137,11 +145,18 @@ func (cons *Consumer) Consume(ctx context.Context, handler func(msg *ckf.Message
}
backoff := 100 * time.Millisecond
- for {
+ limit := cons.retryLimit
+ if limit <= 0 {
+ limit = 4
+ }
+ for attempt := 1; ; attempt++ {
if err := handler(msg); err == nil {
break
} else {
log.Error().Err(err).Msg("kafka message handler error; retrying before advancing offsets")
+ if attempt >= limit {
+ return fmt.Errorf("kafka: handler retry budget exhausted before offset storage: %w", err)
+ }
}
if err := consumerBackoff(ctx, backoff); err != nil {
return err
diff --git a/internal/infrastructure/kafka/consumer_retry_test.go b/internal/infrastructure/kafka/consumer_retry_test.go
index 88d48667e..1ec234de3 100644
--- a/internal/infrastructure/kafka/consumer_retry_test.go
+++ b/internal/infrastructure/kafka/consumer_retry_test.go
@@ -17,10 +17,15 @@ type orderedConsumer struct {
stored []ckf.Offset
reads int
storeErr error
+ lost bool
+ readErr error
}
func (c *orderedConsumer) ReadMessage(time.Duration) (*ckf.Message, error) {
c.reads++
+ if c.readErr != nil {
+ return nil, c.readErr
+ }
if len(c.queue) == 0 {
return nil, errors.New("unexpected poll")
}
@@ -28,6 +33,23 @@ func (c *orderedConsumer) ReadMessage(time.Duration) (*ckf.Message, error) {
c.queue = c.queue[1:]
return m, nil
}
+
+func (c *orderedConsumer) AssignmentLost() bool { return c.lost }
+
+func TestConsumerAssignmentLossRetainsReplayAndPollFailureReopens(t *testing.T) {
+ c := &orderedConsumer{queue: []*ckf.Message{{}, {}}}
+ err := (&Consumer{c: c}).Consume(t.Context(), func(*ckf.Message) error { c.lost = true; return nil })
+ if !errors.Is(err, ErrAssignmentLost) || c.reads != 1 || len(c.stored) != 0 {
+ t.Fatalf("err=%v reads=%d stored=%v", err, c.reads, c.stored)
+ }
+ for _, readErr := range []ckf.Error{ckf.NewError(ckf.ErrMaxPollExceeded, "poll expired", false), ckf.NewError(ckf.ErrFatal, "fatal consumer failure", true)} {
+ c = &orderedConsumer{readErr: readErr}
+ err := (&Consumer{c: c}).Consume(t.Context(), func(*ckf.Message) error { t.Fatal("no record to handle"); return nil })
+ if err == nil || c.reads != 1 || len(c.stored) != 0 {
+ t.Fatalf("err=%v reads=%d stored=%v", err, c.reads, c.stored)
+ }
+ }
+}
func (c *orderedConsumer) StoreMessage(m *ckf.Message) ([]ckf.TopicPartition, error) {
c.stored = append(c.stored, m.TopicPartition.Offset)
return nil, c.storeErr
@@ -83,3 +105,16 @@ func TestConsumerRetryBackoffCancellation(t *testing.T) {
t.Fatal(err)
}
}
+
+func TestConsumerPersistentFailureReturnsForReplayWithoutStoring(t *testing.T) {
+ c := &orderedConsumer{queue: []*ckf.Message{{}, {}}}
+ failure := errors.New("database still unavailable")
+ calls := 0
+ err := (&Consumer{c: c, retryLimit: 2}).Consume(t.Context(), func(*ckf.Message) error {
+ calls++
+ return failure
+ })
+ if !errors.Is(err, failure) || calls != 2 || c.reads != 1 || len(c.stored) != 0 {
+ t.Fatalf("err=%v calls=%d reads=%d stored=%v", err, calls, c.reads, c.stored)
+ }
+}
diff --git a/internal/repository/email_error_dedupe_live_test.go b/internal/repository/email_error_dedupe_live_test.go
index de88d20ae..217e00e7b 100644
--- a/internal/repository/email_error_dedupe_live_test.go
+++ b/internal/repository/email_error_dedupe_live_test.go
@@ -4,6 +4,7 @@ import (
"context"
"sync"
"testing"
+ "time"
"github.com/google/uuid"
)
@@ -85,6 +86,8 @@ func TestLiveEmailErrorDedupeKeepsOneUnresolvedRowPerCode(t *testing.T) {
if first == nil {
t.Fatal("the first occurrence of an error must be recorded")
}
+ exec(`UPDATE email_account_errors SET created_at=NOW()-interval '3 days' WHERE id=$1`, first.ID)
+ before := time.Now()
// The sync loop relaying the same refusal for the next hour.
for i := 0; i < 5; i++ {
@@ -99,6 +102,16 @@ func TestLiveEmailErrorDedupeKeepsOneUnresolvedRowPerCode(t *testing.T) {
if n := unresolved("IMAP_UNKNOWN"); n != 1 {
t.Errorf("unresolved IMAP_UNKNOWN rows = %d, want 1", n)
}
+ rows, xerr := repo.GetByAccountID(ctx, account, true)
+ if xerr != nil || len(rows) != 1 || rows[0].CreatedAt.Before(before) {
+ t.Fatalf("ongoing failure observation expired: %+v %v", rows, xerr)
+ }
+ if xerr := repo.ResolveByCodesBefore(ctx, account, []string{"IMAP_UNKNOWN"}, before, "stale success"); xerr != nil {
+ t.Fatal(xerr)
+ }
+ if n := unresolved("IMAP_UNKNOWN"); n != 1 {
+ t.Fatal("old success cleared a newly observed failure")
+ }
// A different failure is a different row: dedupe must not swallow it.
other, xerr := repo.CreateOnce(ctx, row("SERVER_UNREACHABLE"))
diff --git a/internal/repository/email_message_map.go b/internal/repository/email_message_map.go
index 8f1f9f79d..2d2de8591 100644
--- a/internal/repository/email_message_map.go
+++ b/internal/repository/email_message_map.go
@@ -3,6 +3,7 @@ package repository
import (
"context"
"errors"
+ "time"
"github.com/google/uuid"
"github.com/warmbly/warmbly/internal/models"
@@ -23,6 +24,7 @@ type EmailMessageMapRepository interface {
}
var ErrArrivalOutboxUnsupported = errors.New("durable arrival admission unavailable; upgrade backend and consumer first")
+var ErrArrivalAdmissionUnconfirmed = errors.New("existing provider mapping has no verifiable pending admission")
// ArrivalAdmission is an optional internal protocol, separate from legacy map writes.
type ArrivalAdmission interface {
@@ -40,6 +42,15 @@ type ArrivalOutbox interface {
DeliverArrivals(context.Context, func(context.Context, models.JobEventType, any) error) error
}
+type ArrivalBacklog struct {
+ Pending int
+ Oldest *time.Time
+}
+
+type ArrivalBacklogRepository interface {
+ ArrivalBacklog(context.Context) (ArrivalBacklog, error)
+}
+
// EmailMessageData is one (user, mailbox, providerMessageID) -> internal
// (id, threadID) mapping row. The UUID-typed values travel as strings to match
// the worker call sites, which pass uuid.String().
diff --git a/internal/repository/pg_email_error.go b/internal/repository/pg_email_error.go
index e6bee85fc..c9a4ca2c5 100644
--- a/internal/repository/pg_email_error.go
+++ b/internal/repository/pg_email_error.go
@@ -81,8 +81,7 @@ func NewEmailAccountErrorRepository(database *db.DB) EmailAccountErrorRepository
}
// Create stores a new email account error
-// CreateOnce records the error unless the account already has an unresolved
-// one carrying the same code, and returns (nil, nil) when it declined.
+// CreateOnce refreshes unresolved transport observations without creating or notifying twice.
//
// A mail server that refuses the same command every pass is not a new problem
// every pass. The sync loop retries about once a minute and reports what it
@@ -96,7 +95,14 @@ func NewEmailAccountErrorRepository(database *db.DB) EmailAccountErrorRepository
// 000145), so resolved history is untouched and a problem that returns after
// it was fixed is recorded again.
func (r *emailAccountErrorRepository) CreateOnce(ctx context.Context, data *CreateEmailAccountError) (*EmailAccountError, *errx.Error) {
+ // Unresolved transport timestamps represent the latest failure observation.
query := `
+ WITH observed AS (
+ UPDATE email_account_errors SET created_at=NOW()
+ WHERE email_account_id=$1 AND user_id=$2 AND error_code=$3 AND resolved_at IS NULL
+ AND error_code IN ('SERVER_UNREACHABLE','CONNECTION_LOST','RESOURCE_NOT_FOUND','IMAP_UNKNOWN')
+ RETURNING id
+ )
INSERT INTO email_account_errors (
email_account_id, user_id, error_code, severity, resolve_method,
title, message, user_message, action_required, task_id
diff --git a/internal/repository/pg_sync_arrival.go b/internal/repository/pg_sync_arrival.go
index 727cf8677..f67d6b34f 100644
--- a/internal/repository/pg_sync_arrival.go
+++ b/internal/repository/pg_sync_arrival.go
@@ -84,10 +84,28 @@ func (r *pgEmailMessageMapRepository) AdmitArrival(ctx context.Context, data Ema
if err != nil {
return err
}
+ } else {
+ var pending bool
+ err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM email_message_map m
+ JOIN sync_arrival_outbox o USING(user_id,email_id,message_id,id)
+ WHERE m.user_id=$1 AND m.email_id=$2 AND m.message_id=$3 AND o.organization_id=$4)`, user, email, data.MessageID, org).Scan(&pending)
+ if err != nil {
+ return err
+ }
+ // Legacy, deleted and already-delivered maps have no durable admission proof.
+ if !pending {
+ return ErrArrivalAdmissionUnconfirmed
+ }
}
return tx.Commit(ctx)
}
+func (r *pgEmailMessageMapRepository) ArrivalBacklog(ctx context.Context) (ArrivalBacklog, error) {
+ var backlog ArrivalBacklog
+ err := r.db.QueryRow(ctx, `SELECT count(*),min(created_at) FROM sync_arrival_outbox`).Scan(&backlog.Pending, &backlog.Oldest)
+ return backlog, err
+}
+
func (r *pgEmailMessageMapRepository) HasPendingArrival(ctx context.Context, user, email, id uuid.UUID) (bool, error) {
var pending bool
err := r.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM sync_arrival_outbox o
diff --git a/internal/repository/pg_sync_arrival_live_test.go b/internal/repository/pg_sync_arrival_live_test.go
index eace4c42a..38896ba0e 100644
--- a/internal/repository/pg_sync_arrival_live_test.go
+++ b/internal/repository/pg_sync_arrival_live_test.go
@@ -51,6 +51,10 @@ func TestLiveSyncArrivalRestartAndAmbiguousAdmission(t *testing.T) {
if err := r.AdmitArrival(ctx, data, p); err != nil {
t.Fatal(err)
}
+ stats, err := r.ArrivalBacklog(ctx)
+ if err != nil || stats.Pending != 1 || stats.Oldest == nil {
+ t.Fatalf("missing retained backlog evidence: %+v %v", stats, err)
+ }
var payload string
if err := d.QueryRow(ctx, `SELECT payload FROM sync_arrival_outbox WHERE email_id=$1`, f.mailbox).Scan(&payload); err != nil {
t.Fatal(err)
@@ -128,6 +132,13 @@ func TestLiveSyncArrivalRestartAndAmbiguousAdmission(t *testing.T) {
if pending, err := r.HasPendingArrival(ctx, f.user, f.mailbox, original); err != nil || pending {
t.Fatalf("pending not retired: %v", err)
}
+ if err := r.AdmitArrival(ctx, data, p); !errors.Is(err, ErrArrivalAdmissionUnconfirmed) {
+ t.Fatalf("completed delivery without retained proof was acknowledged: %v", err)
+ }
+ stats, err = r.ArrivalBacklog(ctx)
+ if err != nil || stats.Pending != 0 || stats.Oldest != nil {
+ t.Fatalf("delivered queue claimed a backlog: %+v %v", stats, err)
+ }
}
func TestLiveSyncArrivalReportStagesOwnershipAndLegacyMaps(t *testing.T) {
@@ -188,8 +199,8 @@ func TestLiveSyncArrivalReportStagesOwnershipAndLegacyMaps(t *testing.T) {
lp.Arrival.Message.ID = uuid.New()
retry := legacy
retry.ID = lp.Arrival.Message.ID.String()
- if err := r.AdmitArrival(ctx, retry, lp); err != nil {
- t.Fatal(err)
+ if err := r.AdmitArrival(ctx, retry, lp); !errors.Is(err, ErrArrivalAdmissionUnconfirmed) {
+ t.Fatalf("legacy mapping claimed durable admission: %v", err)
}
known, err := r.Get(ctx, f.user, f.mailbox, legacy.MessageID)
if err != nil || known.ID != legacy.ID {