diff --git a/README.md b/README.md index 6762c413f..3321e223d 100644 --- a/README.md +++ b/README.md @@ -165,6 +165,7 @@ available with `msgvault tui`. | `setup` | Interactive first-run configuration wizard | | `repair-encoding` | Fix UTF-8 encoding issues | | `repair-dates` | Report or repair missing and implausible email sent dates | +| `repair-labels` | Rebuild IMAP message labels from stored mailbox memberships | | `list-senders` / `list-domains` / `list-labels` | Explore metadata | See the [CLI Reference](https://msgvault.io/cli-reference/) for full details diff --git a/cmd/msgvault/cmd/repair_labels.go b/cmd/msgvault/cmd/repair_labels.go new file mode 100644 index 000000000..b609edf45 --- /dev/null +++ b/cmd/msgvault/cmd/repair_labels.go @@ -0,0 +1,133 @@ +package cmd + +import ( + "errors" + "fmt" + "strings" + + "github.com/spf13/cobra" + "go.kenn.io/msgvault/internal/sourceops" + "go.kenn.io/msgvault/internal/store" +) + +func newRepairLabelsCmd() *cobra.Command { + var apply bool + cmd := &cobra.Command{ + Use: "repair-labels [identifier] [--apply]", + Short: "Rebuild IMAP message labels from stored mailbox memberships", + Long: `Rebuild each IMAP source's message labels from its stored +imap_message_memberships rows — the same rebuild a full mailbox enumeration +performs, run on demand instead of on every sync. + +An add-only label merge, used when a sync's mailbox snapshot is incomplete or +a dedup match is not fully confirmed, never removes a label. If the message's +stored membership never changes again, no later sync revisits it, and a stray +label can persist. This command finds and removes it. + +The default is a dry run: it reports what would change without modifying the +archive. Pass --apply to write the repaired labels. This command works +entirely offline and never contacts a provider. + +Examples: + msgvault repair-labels # all IMAP sources + msgvault repair-labels you@example.com # one source + msgvault repair-labels --apply`, + Args: cobra.MaximumNArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if !isDaemonCLISubprocess() { + return runDaemonCLICommandHTTPFromCobra(cmd, args) + } + only := "" + if len(args) == 1 { + only = strings.TrimSpace(args[0]) + } + return runRepairLabelsLocal(cmd, only, apply) + }, + } + cmd.Flags().BoolVar(&apply, "apply", false, "write repaired labels to the archive") + return cmd +} + +func runRepairLabelsLocal(cmd *cobra.Command, only string, apply bool) error { + // A dry run still needs a writable connection: RepairIMAPSourceLabels + // plans by writing inside a transaction and rolling it back, the same + // idiom the store package uses elsewhere for a dry-run plan (see + // errAttributeDryRun) — a read-only connection would reject the writes + // outright before rollback ever came into it. + st, cleanup, err := openWritableStoreAndInit() + if err != nil { + return fmt.Errorf("open archive for label repair: %w", err) + } + defer cleanup() + + var sources []*store.Source + if only == "" { + sources, err = st.ListSourcesContext(cmd.Context(), sourceTypeIMAP) + if err != nil { + return fmt.Errorf("list IMAP sources: %w", err) + } + } else { + // A source's Identifier is its full connection string + // (imaps://user@host:993), not the email a person would type, so + // resolve against identifier or display name the same way + // remove-account and repair-identity do — and fail loudly on no + // match instead of silently repairing nothing. + source, err := sourceops.ResolveExactOne(st, sourceops.Selector{ + Account: only, SourceType: sourceTypeIMAP, + }) + if err != nil { + return err + } + sources = []*store.Source{source} + } + + // Each source repairs (and, with --apply, commits) independently. If a + // later source fails, or an output write fails, sources already + // repaired above this point must still be reflected in the analytics + // cache — so accumulate totals before anything that can still fail, and + // rebuild the cache below unconditionally on --apply rather than only + // after a fully successful loop. + var totalScanned, totalChanged int + runErr := func() error { + for _, src := range sources { + summary, err := st.RepairIMAPSourceLabels(cmd.Context(), src.ID, apply) + if err != nil { + return fmt.Errorf("repair labels for %s: %w", src.Identifier, err) + } + totalScanned += summary.Scanned + totalChanged += summary.Changed + if _, err := fmt.Fprintf(cmd.OutOrStdout(), " %s: scanned=%d changed=%d\n", + src.Identifier, summary.Scanned, summary.Changed); err != nil { + return fmt.Errorf("write label repair line: %w", err) + } + } + + mode := "dry run" + if apply { + mode = "applied" + } + if _, err := fmt.Fprintf(cmd.OutOrStdout(), + "Label repair %s: scanned=%d changed=%d\n", mode, totalScanned, totalChanged); err != nil { + return fmt.Errorf("write label repair summary: %w", err) + } + if !apply && totalChanged > 0 { + if _, err := fmt.Fprintln(cmd.OutOrStdout(), + "Dry run: no rows were modified. Re-run with --apply to write repairs."); err != nil { + return fmt.Errorf("write label repair dry-run guidance: %w", err) + } + } + return nil + }() + + if !apply { + return runErr + } + if cacheErr := rebuildCacheAfterWrite(cfg.DatabaseDSN()); cacheErr != nil { + return errors.Join(runErr, cacheErr) + } + return runErr +} + +func init() { + rootCmd.AddCommand(newRepairLabelsCmd()) +} diff --git a/cmd/msgvault/cmd/repair_labels_test.go b/cmd/msgvault/cmd/repair_labels_test.go new file mode 100644 index 000000000..2e57d45e8 --- /dev/null +++ b/cmd/msgvault/cmd/repair_labels_test.go @@ -0,0 +1,315 @@ +package cmd + +import ( + "bytes" + "context" + "errors" + "testing" + + "github.com/spf13/cobra" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/msgvault/internal/config" + "go.kenn.io/msgvault/internal/query" + "go.kenn.io/msgvault/internal/store" +) + +// TestRunRepairLabelsLocalDryRunApplyAndNoop catches a repair command that +// mutates without --apply, omits an operator-useful summary, or advances the +// cache revision for an already-current archive. The fixture reproduces the +// issue #748 gap directly: an add-only label merge leaves a label no +// imap_message_memberships row backs. +func TestRunRepairLabelsLocalDryRunApplyAndNoop(t *testing.T) { + assert := assert.New(t) + require := require.New(t) + dataDir := t.TempDir() + savedCfg := cfg + cfg = &config.Config{HomeDir: dataDir, Data: config.DataConfig{DataDir: dataDir}} + t.Cleanup(func() { cfg = savedCfg }) + + messageID := newLabelRepairArchive(t, "labels@example.test") + _, err := buildCache(cfg.DatabaseDSN(), cfg.AnalyticsDir(), true) + require.NoError(err) + beforeRevision := labelRepairArchiveRevision(t) + beforeCacheState, err := query.ReadCacheSyncState(cfg.AnalyticsDir()) + require.NoError(err) + assert.Equal(beforeRevision, beforeCacheState.DerivedDataRevision) + + var dryRunOut bytes.Buffer + dryRunCmd := &cobra.Command{} + dryRunCmd.SetContext(context.Background()) + dryRunCmd.SetOut(&dryRunOut) + require.NoError(runRepairLabelsLocal(dryRunCmd, "", false)) + assert.Equal( + " labels@example.test: scanned=2 changed=1\n"+ + "Label repair dry run: scanned=2 changed=1\n"+ + "Dry run: no rows were modified. Re-run with --apply to write repairs.\n", + dryRunOut.String()) + assert.Equal([]string{"INBOX", "Stray"}, labelRepairArchiveLabels(t, messageID)) + assert.Equal(beforeRevision, labelRepairArchiveRevision(t)) + + var applyOut bytes.Buffer + applyCmd := &cobra.Command{} + applyCmd.SetContext(context.Background()) + applyCmd.SetOut(&applyOut) + require.NoError(runRepairLabelsLocal(applyCmd, "", true)) + assert.Equal( + " labels@example.test: scanned=2 changed=1\n"+ + "Label repair applied: scanned=2 changed=1\n", + applyOut.String()) + assert.Equal([]string{"INBOX"}, labelRepairArchiveLabels(t, messageID)) + assert.Equal(beforeRevision+1, labelRepairArchiveRevision(t)) + afterCacheState, err := query.ReadCacheSyncState(cfg.AnalyticsDir()) + require.NoError(err) + assert.Equal(beforeRevision+1, afterCacheState.DerivedDataRevision) + + var noChangeOut bytes.Buffer + noChangeCmd := &cobra.Command{} + noChangeCmd.SetContext(context.Background()) + noChangeCmd.SetOut(&noChangeOut) + require.NoError(runRepairLabelsLocal(noChangeCmd, "", true)) + assert.Equal( + " labels@example.test: scanned=2 changed=0\n"+ + "Label repair applied: scanned=2 changed=0\n", + noChangeOut.String()) + assert.Equal(beforeRevision+1, labelRepairArchiveRevision(t)) +} + +// TestRunRepairLabelsLocalUnknownIdentifierErrors catches a repair command +// that silently matches no source instead of failing loudly when the given +// identifier does not resolve to one. +func TestRunRepairLabelsLocalUnknownIdentifierErrors(t *testing.T) { + require := require.New(t) + dataDir := t.TempDir() + savedCfg := cfg + cfg = &config.Config{HomeDir: dataDir, Data: config.DataConfig{DataDir: dataDir}} + t.Cleanup(func() { cfg = savedCfg }) + + newLabelRepairArchive(t, "one@example.test") + + repairCmd := &cobra.Command{} + repairCmd.SetContext(context.Background()) + repairCmd.SetOut(&bytes.Buffer{}) + err := runRepairLabelsLocal(repairCmd, "someone-else@example.test", true) + require.Error(err) + require.ErrorContains(err, `no account found for "someone-else@example.test"`) +} + +// TestRunRepairLabelsLocalIdentifierScopesByDisplayName catches matching only +// the raw connection-string identifier. A real IMAP source's Identifier is +// its imaps://user@host:port connection string, not the email a person types +// on the command line — the display name carries that email. +func TestRunRepairLabelsLocalIdentifierScopesByDisplayName(t *testing.T) { + assert := assert.New(t) + require := require.New(t) + dataDir := t.TempDir() + savedCfg := cfg + cfg = &config.Config{HomeDir: dataDir, Data: config.DataConfig{DataDir: dataDir}} + t.Cleanup(func() { cfg = savedCfg }) + + st, err := store.OpenForTest(cfg.DatabaseDSN()) + require.NoError(err) + require.NoError(st.InitSchema()) + scoped, err := st.GetOrCreateSource("imap", "imaps://scoped@example.test:993") + require.NoError(err) + require.NoError(st.UpdateSourceDisplayName(scoped.ID, "scoped@example.test")) + other, err := st.GetOrCreateSource("imap", "imaps://other@example.test:993") + require.NoError(err) + require.NoError(st.UpdateSourceDisplayName(other.ID, "other@example.test")) + require.NoError(st.Close()) + + var out bytes.Buffer + repairCmd := &cobra.Command{} + repairCmd.SetContext(context.Background()) + repairCmd.SetOut(&out) + require.NoError(runRepairLabelsLocal(repairCmd, "scoped@example.test", true)) + assert.Equal( + " imaps://scoped@example.test:993: scanned=0 changed=0\n"+ + "Label repair applied: scanned=0 changed=0\n", + out.String()) +} + +// errAfterNWriter succeeds for the first n Write calls and fails every call +// after, without touching whatever already happened before those writes. +type errAfterNWriter struct { + n int + buf bytes.Buffer +} + +func (w *errAfterNWriter) Write(p []byte) (int, error) { + if w.n <= 0 { + return 0, errors.New("simulated output failure") + } + w.n-- + // bytes.Buffer.Write never returns a non-nil error. + n, _ := w.buf.Write(p) + return n, nil +} + +// TestRunRepairLabelsLocalRebuildsCacheDespitePartialFailure catches the +// failure mode roborev flagged on PR #754: each source's repair commits +// independently, but the analytics cache rebuild used to run only after the +// whole multi-source loop returned successfully. A later source's own +// commit still succeeds even when something after it fails (another +// source's repair, an output write, or a cancelled context) — so that +// commit must still reach the cache, not wait on the loop finishing clean. +// This test fails the second source's *output write*, the simplest of those +// three triggers to force deterministically; the important thing it proves +// is that a real, already-committed change is not lost from the cache just +// because the command as a whole reports an error. +func TestRunRepairLabelsLocalRebuildsCacheDespitePartialFailure(t *testing.T) { + assert := assert.New(t) + require := require.New(t) + dataDir := t.TempDir() + savedCfg := cfg + cfg = &config.Config{HomeDir: dataDir, Data: config.DataConfig{DataDir: dataDir}} + t.Cleanup(func() { cfg = savedCfg }) + + newLabelRepairArchive(t, "one@example.test") + newLabelRepairArchive(t, "two@example.test") + _, err := buildCache(cfg.DatabaseDSN(), cfg.AnalyticsDir(), true) + require.NoError(err) + beforeRevision := labelRepairArchiveRevision(t) + + // The first source's per-source line is the only Write call allowed to + // succeed; the second source's own repair still runs and commits before + // its line fails to print. + out := &errAfterNWriter{n: 1} + repairCmd := &cobra.Command{} + repairCmd.SetContext(context.Background()) + repairCmd.SetOut(out) + err = runRepairLabelsLocal(repairCmd, "", true) + require.ErrorContains(err, "write label repair line") + + // Both sources actually committed (the stray label from each is gone), + // bumping the revision twice, and the cache rebuild ran anyway and + // caught up to it — despite the command itself returning an error. + assert.Equal(beforeRevision+2, labelRepairArchiveRevision(t)) + cacheState, err := query.ReadCacheSyncState(cfg.AnalyticsDir()) + require.NoError(err) + assert.Equal(beforeRevision+2, cacheState.DerivedDataRevision) +} + +// TestRepairLabelsCommandRoutesThroughDaemonCLIRunner catches bypassing the +// daemon-owned writer path from a normal CLI parent process. +func TestRepairLabelsCommandRoutesThroughDaemonCLIRunner(t *testing.T) { + server, requests := newDaemonCLIRunnerTestServer(t, + func(req daemonCLIRunTestRequest) { + assert.Equal(t, []string{"repair-labels", "--apply"}, req.Args) + }, + `{"type":"stdout","data":"Label repair applied: scanned=1 changed=1\n"}`, + `{"type":"complete"}`, + ) + configureRemoteDaemonForTest(t, server.URL) + t.Setenv(daemonCLISubprocessEnv, "") + + cmd := newRepairLabelsCmd() + var stdout bytes.Buffer + cmd.SetOut(&stdout) + cmd.SetArgs([]string{"--apply"}) + require.NoError(t, cmd.Execute()) + assert.Equal(t, int32(1), requests.Load()) + assert.Equal(t, "Label repair applied: scanned=1 changed=1\n", stdout.String()) +} + +// TestRepairLabelsCommandHelpAndFlagValidation executes Cobra's public +// command surface, catching a command that is not registered or accepts an +// accidental mutation flag. +func TestRepairLabelsCommandHelpAndFlagValidation(t *testing.T) { + assert := assert.New(t) + require := require.New(t) + root := newTestRootCmd() + root.AddCommand(newRepairLabelsCmd()) + var help bytes.Buffer + root.SetOut(&help) + root.SetArgs([]string{"repair-labels", "--help"}) + require.NoError(root.Execute()) + assert.Contains(help.String(), "msgvault repair-labels [identifier] [--apply]") + assert.Contains(help.String(), "--apply") + + root = newTestRootCmd() + root.AddCommand(newRepairLabelsCmd()) + root.SetArgs([]string{"repair-labels", "--force"}) + err := root.Execute() + require.Error(err) + assert.ErrorContains(err, "unknown flag: --force") +} + +// newLabelRepairArchive seeds one IMAP source with two messages in INBOX, +// then reproduces the issue #748 gap: an add-only label merge on the first +// message leaves a "Stray" label no membership row backs. Returns that +// message's ID. +func newLabelRepairArchive(t *testing.T, identifier string) int64 { + t.Helper() + st, err := store.OpenForTest(cfg.DatabaseDSN()) + require.NoError(t, err) + require.NoError(t, st.InitSchema()) + source, err := st.GetOrCreateSource("imap", identifier) + require.NoError(t, err) + conversationID, err := st.EnsureConversation(source.ID, "label-repair-thread", "Label repair") + require.NoError(t, err) + + newMessage := func(sourceMessageID string) int64 { + messageID, err := st.UpsertMessage(&store.Message{ + ConversationID: conversationID, + SourceID: source.ID, + SourceMessageID: sourceMessageID, + MessageType: "email", + }) + require.NoError(t, err) + return messageID + } + messageID1 := newMessage("label-repair-1") + newMessage("label-repair-2") + + require.NoError(t, st.ApplyIMAPMailboxDeltas(source.ID, []store.IMAPMailboxDelta{{ + Mailbox: "INBOX", + State: store.IMAPFolderState{Mailbox: "INBOX", UIDValidity: 1, UIDNext: 3, HighestModSeq: 1}, + Memberships: []store.IMAPMembershipObservation{ + {Mailbox: "INBOX", UIDValidity: 1, UID: 1, SourceMessageID: "label-repair-1"}, + {Mailbox: "INBOX", UIDValidity: 1, UID: 2, SourceMessageID: "label-repair-2"}, + }, + }})) + + strayLabelID, err := st.EnsureLabel(source.ID, "stray-source-label", "Stray", "user") + require.NoError(t, err) + _, err = st.ReconcileMessageLabels(messageID1, []int64{strayLabelID}, false) + require.NoError(t, err) + + require.NoError(t, st.Close()) + return messageID1 +} + +func labelRepairArchiveLabels(t *testing.T, messageID int64) []string { + t.Helper() + st, err := store.OpenForTest(cfg.DatabaseDSN()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + rows, err := st.DB().Query(st.Rebind(` + SELECT l.name + FROM labels l + JOIN message_labels ml ON ml.label_id = l.id + WHERE ml.message_id = ? + ORDER BY l.name + `), messageID) + require.NoError(t, err) + defer func() { require.NoError(t, rows.Close()) }() + var labels []string + for rows.Next() { + var name string + require.NoError(t, rows.Scan(&name)) + labels = append(labels, name) + } + require.NoError(t, rows.Err()) + return labels +} + +func labelRepairArchiveRevision(t *testing.T) int64 { + t.Helper() + st, err := store.OpenForTest(cfg.DatabaseDSN()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + revision, err := st.DerivedDataRevision() + require.NoError(t, err) + return revision +} diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 16ff5349d..1971d2d89 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -1,5 +1,5 @@ --- -last_edited: "2026-08-31" +last_edited: "2026-09-03" title: CLI Reference description: Complete command reference for all msgvault commands. --- @@ -1128,6 +1128,33 @@ write changed values and mark derived analytics stale for the normal rebuild pat --- +## repair-labels + +Rebuild an IMAP source's message labels from its stored mailbox memberships, +without contacting the provider. + +```bash +msgvault repair-labels [identifier] [--apply] +``` + +An add-only label merge — used when a sync's mailbox snapshot is incomplete or +a dedup match is not fully confirmed — never removes a label. If the +message's stored membership never changes again, nothing later revisits it, +and a stray label can persist. `repair-labels` rebuilds a message's labels +straight from `imap_message_memberships`, the same rebuild a full mailbox +enumeration performs, without a live IMAP connection. + +With no identifier, every IMAP source is repaired. An identifier scopes the +repair to one matching source. The default is a dry run and does not modify +the archive. Pass `--apply` to write changed labels and mark derived +analytics stale for the normal rebuild path. + +| Flag | Description | +|---|---| +| `--apply` | Write repaired labels to the archive. | + +--- + ## documents Manage hosted extraction and local full-text indexing for standalone document diff --git a/internal/api/cli_handlers.go b/internal/api/cli_handlers.go index 9f47532a4..abc3e69d2 100644 --- a/internal/api/cli_handlers.go +++ b/internal/api/cli_handlers.go @@ -1637,6 +1637,7 @@ func cliRunCommandAllowed(args []string) bool { "purge-excluded-media", "repair-dates", "repair-identity", + "repair-labels", "repair-list-ids", "repair-senders", "repack-attachments", diff --git a/internal/api/handlers_test.go b/internal/api/handlers_test.go index 26d8270e8..37bca024e 100644 --- a/internal/api/handlers_test.go +++ b/internal/api/handlers_test.go @@ -1745,6 +1745,7 @@ func TestHandleCLIRunBackupSubcommandAdmission(t *testing.T) { {"purge excluded media dry-run allowed", []string{"purge-excluded-media", "--dry-run"}, true}, {"pack-attachments allowed", []string{"pack-attachments"}, true}, {"repair-dates apply allowed", []string{"repair-dates", "--apply"}, true}, + {"repair-labels apply allowed", []string{"repair-labels", "--apply"}, true}, {"repair list IDs apply allowed", []string{"repair-list-ids", "--apply"}, true}, {"repair-senders apply allowed", []string{"repair-senders", "--apply"}, true}, {"repack-attachments allowed", []string{"repack-attachments"}, true}, diff --git a/internal/store/export_test.go b/internal/store/export_test.go index 5495207f4..7719204f1 100644 --- a/internal/store/export_test.go +++ b/internal/store/export_test.go @@ -183,6 +183,28 @@ func (s *Store) SetListIDRepairAfterFingerprintLockHookForTest(fn func()) func() return func() { s.listIDRepairAfterFingerprintLockHook = nil } } +// SetIMAPLabelRepairPerMessageHookForTest installs a hook called with each +// message's ID just before RepairIMAPSourceLabels processes it. Tests use it +// to cancel the context mid-repair without needing a source large enough to +// make cancellation a race. +func (s *Store) SetIMAPLabelRepairPerMessageHookForTest(fn func(messageID int64)) func() { + s.imapLabelRepairPerMessageHook = fn + return func() { s.imapLabelRepairPerMessageHook = nil } +} + +// ReconcileMessageLabelsTxContextForTest runs the context-aware label +// reconciliation on its own transaction. The transaction is deliberately begun +// without ctx: BeginTx would otherwise reject a cancelled context first, and +// the test could not tell whether the statements inside carry ctx or not. +func ReconcileMessageLabelsTxContextForTest( + ctx context.Context, s *Store, messageID int64, labelIDs []int64, replace bool, +) error { + return s.withTx(func(tx *loggedTx) error { + _, err := s.reconcileMessageLabelsTxContext(ctx, tx, messageID, labelIDs, replace) + return err + }) +} + // SetIdentityMatchAcceptBeforeDecisionHookForTest pauses a user acceptance // after its initial read and before its locked decision transaction. func (s *Store) SetIdentityMatchAcceptBeforeDecisionHookForTest(fn func()) func() { diff --git a/internal/store/imap_label_repair_test.go b/internal/store/imap_label_repair_test.go new file mode 100644 index 000000000..1d6fd9d05 --- /dev/null +++ b/internal/store/imap_label_repair_test.go @@ -0,0 +1,159 @@ +package store_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/msgvault/internal/store" +) + +// seedTwoInboxMessages puts two messages in INBOX via a normal (Reset-free) +// membership apply, matching how a real sync would record them. +func seedTwoInboxMessages(t *testing.T, f imapMembershipFixture) (int64, int64) { + t.Helper() + messageID1 := f.createMessage(t, "repair-labels-1", "") + messageID2 := f.createMessage(t, "repair-labels-2", "") + require.NoError(t, f.store.ApplyIMAPMailboxDeltas(f.source.ID, []store.IMAPMailboxDelta{{ + Mailbox: "INBOX", + State: store.IMAPFolderState{Mailbox: "INBOX", UIDValidity: 1, UIDNext: 3, HighestModSeq: 1}, + Memberships: []store.IMAPMembershipObservation{ + {Mailbox: "INBOX", UIDValidity: 1, UID: 1, SourceMessageID: "repair-labels-1"}, + {Mailbox: "INBOX", UIDValidity: 1, UID: 2, SourceMessageID: "repair-labels-2"}, + }, + }})) + return messageID1, messageID2 +} + +// TestRepairIMAPSourceLabels_RemovesStrayAddOnlyLabel reproduces the gap +// described in issue #748: an add-only label merge leaves a label that no +// imap_message_memberships row backs, and nothing else ever revisits it. +func TestRepairIMAPSourceLabels_RemovesStrayAddOnlyLabel(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + f := newIMAPMembershipFixture(t) + messageID1, messageID2 := seedTwoInboxMessages(t, f) + + strayLabelID, err := f.store.EnsureLabel(f.source.ID, "stray-source-label", "Stray", "user") + require.NoError(err) + changed, err := f.store.ReconcileMessageLabels(messageID1, []int64{strayLabelID}, false) + require.NoError(err) + require.True(changed) + require.Equal([]string{"INBOX", "Stray"}, messageLabels(t, f.store, messageID1)) + + summary, err := f.store.RepairIMAPSourceLabels(context.Background(), f.source.ID, true) + require.NoError(err) + assert.Equal(2, summary.Scanned) + assert.Equal(1, summary.Changed) + assert.Equal([]string{"INBOX"}, messageLabels(t, f.store, messageID1)) + assert.Equal([]string{"INBOX"}, messageLabels(t, f.store, messageID2)) +} + +// TestRepairIMAPSourceLabels_DryRunDoesNotWrite asserts a dry run reports +// what would change without persisting it. +func TestRepairIMAPSourceLabels_DryRunDoesNotWrite(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + f := newIMAPMembershipFixture(t) + messageID1, _ := seedTwoInboxMessages(t, f) + + strayLabelID, err := f.store.EnsureLabel(f.source.ID, "stray-source-label", "Stray", "user") + require.NoError(err) + _, err = f.store.ReconcileMessageLabels(messageID1, []int64{strayLabelID}, false) + require.NoError(err) + + summary, err := f.store.RepairIMAPSourceLabels(context.Background(), f.source.ID, false) + require.NoError(err) + assert.Equal(2, summary.Scanned) + assert.Equal(1, summary.Changed) + assert.Equal([]string{"INBOX", "Stray"}, messageLabels(t, f.store, messageID1)) +} + +// TestRepairIMAPSourceLabels_NoopWhenLabelsAlreadyMatch asserts a second +// apply run over already-consistent labels changes nothing. +func TestRepairIMAPSourceLabels_NoopWhenLabelsAlreadyMatch(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + f := newIMAPMembershipFixture(t) + seedTwoInboxMessages(t, f) + + summary, err := f.store.RepairIMAPSourceLabels(context.Background(), f.source.ID, true) + require.NoError(err) + assert.Equal(2, summary.Scanned) + assert.Equal(0, summary.Changed) +} + +// TestRepairIMAPSourceLabels_CancellationStopsTheLoop catches a repair that +// keeps processing every message regardless of context cancellation. The +// per-message test hook cancels after the first message, so a second message +// being processed proves cancellation was not honored between messages. +func TestRepairIMAPSourceLabels_CancellationStopsTheLoop(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + f := newIMAPMembershipFixture(t) + messageID1, _ := seedTwoInboxMessages(t, f) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + var processed []int64 + restore := f.store.SetIMAPLabelRepairPerMessageHookForTest(func(messageID int64) { + processed = append(processed, messageID) + cancel() + }) + defer restore() + + summary, err := f.store.RepairIMAPSourceLabels(ctx, f.source.ID, true) + require.ErrorIs(err, context.Canceled) + assert.Equal(store.IMAPLabelRepairSummary{}, summary) + assert.Equal([]int64{messageID1}, processed) +} + +// TestRepairIMAPSourceLabels_DryRunCancellationIsNotMaskedAsSuccess catches +// the specific failure mode roborev flagged on PR #754: the dry-run rollback +// sentinel returning as a nil error even when the context was cancelled +// mid-repair, which would report a cancelled run as a completed dry run. +func TestRepairIMAPSourceLabels_DryRunCancellationIsNotMaskedAsSuccess(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + f := newIMAPMembershipFixture(t) + messageID1, _ := seedTwoInboxMessages(t, f) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + var processed []int64 + restore := f.store.SetIMAPLabelRepairPerMessageHookForTest(func(messageID int64) { + processed = append(processed, messageID) + cancel() + }) + defer restore() + + summary, err := f.store.RepairIMAPSourceLabels(ctx, f.source.ID, false) + require.ErrorIs(err, context.Canceled) + assert.Equal(store.IMAPLabelRepairSummary{}, summary) + assert.Equal([]int64{messageID1}, processed) +} + +// TestReconcileMessageLabelsTxContext_CancelledContextStopsItsStatements +// guards the plumbing RepairIMAPSourceLabels relies on. The repair polls +// ctx.Err() between messages, but that only interrupts it between statements. +// This asserts the statements themselves carry ctx, which is what lets a +// cancellation reach a query already running. +func TestReconcileMessageLabelsTxContext_CancelledContextStopsItsStatements(t *testing.T) { + require := require.New(t) + f := newIMAPMembershipFixture(t) + messageID1, _ := seedTwoInboxMessages(t, f) + + labelID, err := f.store.EnsureLabel(f.source.ID, "ctx-probe-label", "Ctx Probe", "user") + require.NoError(err) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + require.ErrorIs( + store.ReconcileMessageLabelsTxContextForTest( + ctx, f.store, messageID1, []int64{labelID}, true, + ), + context.Canceled, + ) +} diff --git a/internal/store/imap_memberships.go b/internal/store/imap_memberships.go index c7d087e10..53afefe52 100644 --- a/internal/store/imap_memberships.go +++ b/internal/store/imap_memberships.go @@ -261,13 +261,13 @@ func (s *Store) applyIMAPMailboxDeltas( } for _, messageID := range sortedIMAPMessageIDs(affected) { - mailboxes, err := imapMembershipMailboxes(tx, sourceID, messageID) + mailboxes, err := imapMembershipMailboxes(ctx, tx, sourceID, messageID) if err != nil { return err } labelIDs := make([]int64, 0, len(mailboxes)) for _, mailbox := range mailboxes { - labelID, err := ensureIMAPMailboxLabel(tx, sourceID, mailbox) + labelID, err := ensureIMAPMailboxLabel(ctx, tx, sourceID, mailbox) if err != nil { return err } @@ -649,9 +649,9 @@ func imapRFC822MessageIDCandidates(messageID string) []string { } func imapMembershipMailboxes( - tx *loggedTx, sourceID, messageID int64, + ctx context.Context, tx *loggedTx, sourceID, messageID int64, ) ([]string, error) { - rows, err := tx.Query(` + rows, err := tx.QueryContext(ctx, ` SELECT mailbox FROM imap_message_memberships WHERE source_id = ? AND message_id = ? GROUP BY mailbox @@ -675,9 +675,9 @@ func imapMembershipMailboxes( return mailboxes, nil } -func ensureIMAPMailboxLabel(tx *loggedTx, sourceID int64, mailbox string) (int64, error) { +func ensureIMAPMailboxLabel(ctx context.Context, tx *loggedTx, sourceID int64, mailbox string) (int64, error) { var labelID int64 - err := tx.QueryRow(` + err := tx.QueryRowContext(ctx, ` SELECT id FROM labels WHERE source_id = ? AND source_label_id = ? `, sourceID, mailbox).Scan(&labelID) if err == nil { @@ -686,7 +686,13 @@ func ensureIMAPMailboxLabel(tx *loggedTx, sourceID int64, mailbox string) (int64 if !errors.Is(err, sql.ErrNoRows) { return 0, fmt.Errorf("find label for IMAP mailbox %q: %w", mailbox, err) } - labelID, err = ensureLabelWith(tx, sourceID, mailbox, mailbox, "user", nil) + // ensureLabelWith is shared with EnsureLabel and EnsureLabelsBatch, which + // have no ctx of their own, so it takes a querier rather than a context. + // boundQuerier carries ctx to its statements without changing that + // signature or any other caller. + labelID, err = ensureLabelWith( + boundQuerier{ctx: ctx, q: tx}, sourceID, mailbox, mailbox, "user", nil, + ) if err != nil { return 0, fmt.Errorf("ensure label for IMAP mailbox %q: %w", mailbox, err) } @@ -701,3 +707,129 @@ func sortedIMAPMessageIDs(ids map[int64]struct{}) []int64 { slices.Sort(sorted) return sorted } + +// IMAPLabelRepairSummary reports what RepairIMAPSourceLabels found and, when +// applying, changed. +type IMAPLabelRepairSummary struct { + // Scanned counts messages with at least one imap_message_memberships row. + Scanned int + // Changed counts messages whose message_labels did not already match + // their stored memberships. + Changed int +} + +var errIMAPLabelRepairDryRun = errors.New("imap label repair dry run rollback") + +// RepairIMAPSourceLabels rebuilds message_labels for every message that has +// an imap_message_memberships row in sourceID, from those membership rows — +// the same rebuild ApplyIMAPMailboxDeltas performs for its affected set, run +// here over the whole source on demand. It exists because an add-only label +// merge (ReconcileMessageLabels with replace=false) can leave a stray label +// that no later Reset ever revisits, when the message's membership rows +// never change again. +// +// It does not touch deleted_from_source_at: a message with no membership +// rows never appears in the set this repairs, so there is nothing to +// tombstone from this data. Tombstone reconciliation stays a Reset-only +// concern. +// +// With apply false, every write happens inside the same transaction and is +// then rolled back, so Changed still reports what would happen. +func (s *Store) RepairIMAPSourceLabels( + ctx context.Context, sourceID int64, apply bool, +) (IMAPLabelRepairSummary, error) { + var summary IMAPLabelRepairSummary + txErr := s.withTxContext(ctx, func(tx *loggedTx) error { + messageIDs, err := distinctIMAPMembershipMessageIDs(ctx, tx, sourceID) + if err != nil { + return err + } + for _, messageID := range messageIDs { + if err := ctx.Err(); err != nil { + return err + } + if s.imapLabelRepairPerMessageHook != nil { + s.imapLabelRepairPerMessageHook(messageID) + } + mailboxes, err := imapMembershipMailboxes(ctx, tx, sourceID, messageID) + if err != nil { + return err + } + labelIDs := make([]int64, 0, len(mailboxes)) + for _, mailbox := range mailboxes { + labelID, err := ensureIMAPMailboxLabel(ctx, tx, sourceID, mailbox) + if err != nil { + return err + } + labelIDs = append(labelIDs, labelID) + } + changed, err := s.reconcileMessageLabelsTxContext(ctx, tx, messageID, labelIDs, true) + if err != nil { + return fmt.Errorf("reconcile labels for IMAP message %d: %w", messageID, err) + } + summary.Scanned++ + if changed { + summary.Changed++ + } + } + // Checked once more here, not just per-message above: without this, + // a cancellation landing exactly after the last message would fall + // through to the dry-run branch below and get reported as a normal + // completion instead of surfaced as a cancellation error. + if err := ctx.Err(); err != nil { + return err + } + if summary.Changed > 0 { + // message_labels is exported into the analytics cache; bumping + // the revision is what tells that cache the export is stale, the + // same signal RepairListIDs and AddMessageLabels give it. + if err := s.bumpDerivedDataRevision(tx); err != nil { + return err + } + } + if !apply { + return errIMAPLabelRepairDryRun + } + return nil + }) + if txErr != nil { + if errors.Is(txErr, errIMAPLabelRepairDryRun) { + return summary, nil + } + // database/sql rolls back a transaction in a background goroutine as + // soon as its context is cancelled, racing the ctx.Err() checks + // above — the query in flight when that goroutine wins can surface + // a raw driver error ("transaction has already been committed or + // rolled back") instead. Normalize: whenever the context is + // actually done, report that instead of whatever the race produced. + if ctxErr := ctx.Err(); ctxErr != nil { + return IMAPLabelRepairSummary{}, ctxErr + } + return IMAPLabelRepairSummary{}, txErr + } + return summary, nil +} + +func distinctIMAPMembershipMessageIDs(ctx context.Context, tx *loggedTx, sourceID int64) ([]int64, error) { + rows, err := tx.QueryContext(ctx, ` + SELECT DISTINCT message_id FROM imap_message_memberships + WHERE source_id = ? + ORDER BY message_id + `, sourceID) + if err != nil { + return nil, fmt.Errorf("query IMAP membership message IDs for source %d: %w", sourceID, err) + } + defer func() { _ = rows.Close() }() + var messageIDs []int64 + for rows.Next() { + var messageID int64 + if err := rows.Scan(&messageID); err != nil { + return nil, fmt.Errorf("scan IMAP membership message ID: %w", err) + } + messageIDs = append(messageIDs, messageID) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate IMAP membership message IDs: %w", err) + } + return messageIDs, nil +} diff --git a/internal/store/messages.go b/internal/store/messages.go index f3892b541..3f7453145 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -2626,7 +2626,20 @@ func (s *Store) ReconcileMessageLabels( func (s *Store) reconcileMessageLabelsTx( tx *loggedTx, messageID int64, labelIDs []int64, replace bool, ) (bool, error) { - rows, err := tx.Query(` + return s.reconcileMessageLabelsTxContext( + context.Background(), tx, messageID, labelIDs, replace, + ) +} + +// reconcileMessageLabelsTxContext is the context-aware form of +// reconcileMessageLabelsTx: every statement carries ctx, so a cancelled caller +// interrupts the read and the write in flight rather than only between calls. +// ReconcileMessageLabels and AddMessageLabels have no ctx to give, so they keep +// reaching it through the background-context wrapper above. +func (s *Store) reconcileMessageLabelsTxContext( + ctx context.Context, tx *loggedTx, messageID int64, labelIDs []int64, replace bool, +) (bool, error) { + rows, err := tx.QueryContext(ctx, ` SELECT label_id FROM message_labels WHERE message_id = ? `, messageID) if err != nil { @@ -2668,7 +2681,9 @@ func (s *Store) reconcileMessageLabelsTx( if !changed { return false, nil } - if err := replaceMessageLabelsTx(tx, messageID, labelIDs); err != nil { + if err := replaceMessageLabelsTx( + boundQuerier{ctx: ctx, q: tx}, messageID, labelIDs, + ); err != nil { return false, err } return true, nil @@ -2683,7 +2698,7 @@ func (s *Store) reconcileMessageLabelsTx( if len(missing) == 0 { return false, nil } - if err := s.addMessageLabelsTx(tx, messageID, missing); err != nil { + if err := s.addMessageLabelsTx(boundQuerier{ctx: ctx, q: tx}, messageID, missing); err != nil { return false, err } return true, nil @@ -2737,7 +2752,7 @@ func (s *Store) AddMessageLabels(messageID int64, labelIDs []int64) error { } func (s *Store) addMessageLabelsTx( - tx *loggedTx, messageID int64, labelIDs []int64, + tx querier, messageID int64, labelIDs []int64, ) error { return insertInChunks(tx, chunkInsert{ totalRows: len(labelIDs), diff --git a/internal/store/store.go b/internal/store/store.go index c67045a66..bb2ca5b33 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -78,6 +78,7 @@ type Store struct { listIDRepairBeforeApplyHook func() listIDRepairAfterScanHook func(context.Context, *loggedTx, []listIDRepairUpdate) error listIDRepairAfterFingerprintLockHook func() + imapLabelRepairPerMessageHook func(messageID int64) cardDAVConflictResolveSnapshotHook func() cardDAVTombstonePrepareSnapshotHook func() identityMatchAcceptBeforeDecisionHook func()