amadan

carlos / platform Public

mail: the one-shot email.json sweep, behind a flag that defaults off (plan 2, task 4)

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Paul Campbell pushed by claude 7e032cf406c18b6e7e2e3dcf1f4cdc5b93df5af8 parent cea8229
3 files changed, +166 −1
  • cmd/console/main.go +19 −1
  • internal/mail/sweep.go +84 −0
  • internal/mail/sweep_test.go +63 −0
diff --git a/cmd/console/main.go b/cmd/console/main.go
index 57d6a15b5..566f06565 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -262,11 +262,14 @@ func serve() {
storeScanner, seedAntivirus := storeScannerFor(storeDeployment)
mailDep := mailDeployment()
mailProvisioner, dnsWriter := mailBackends(mailDep, mail.NewAWSProvisioner, dnsrec.NewRoute53Writer)
+ // One release.Store, named, because the senders sweep below needs the
+ // same one the server got.
+ rel := openStore()
srv, err := console.NewServer(console.Config{
Origin: origin,
Keymail: envOr("CONSOLE_KEYMAIL", "https://keymail.dev"),
Registry: envOr("CARLOS_REGISTRY", "/data/registry.db"),
- Releases: openStore(),
+ Releases: rel,
// The deployment bucket's NAME, from the same variable openStore()
// opens it with — /fleet/architecture draws the box -> bucket edge
// from it. Empty on a CARLOS_DEPLOYMENT_DIR deployment, which
@@ -334,6 +337,21 @@ func serve() {
// here rather than in NewServer so constructing a Server — which every
// console test does — never starts a goroutine or sends mail.
go srv.StartAlertWatcher(context.Background())
+ // The senders migration's tail (tenant-email senders design §5): once a
+ // release has passed with every console writing senders/app.json, an
+ // operator sets CARLOS_MAIL_LEGACY_SWEEP=1 and the next start deletes the
+ // email.json objects that have already been replaced. Off by default, and
+ // it never touches an app that has not been migrated.
+ if os.Getenv("CARLOS_MAIL_LEGACY_SWEEP") == "1" {
+ go func() {
+ n, err := mail.SweepLegacyRecords(context.Background(), rel, true)
+ if err != nil {
+ log.Printf("console: mail legacy sweep: %v", err)
+ return
+ }
+ log.Printf("console: mail legacy sweep -> %d email.json objects removed", n)
+ }()
+ }
// The economics pass writes the daily usage/census/storage/statement
// objects and pulls the AWS bill (internal/console/economicspass.go).
// Same placement rule as the alert watcher: started here, never in
diff --git a/internal/mail/sweep.go b/internal/mail/sweep.go
new file mode 100644
index 000000000..fd596ae67
--- /dev/null
+++ b/internal/mail/sweep.go
@@ -0,0 +1,84 @@
+// The one-shot sweep that finishes the senders migration (design §5).
+package mail
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "path"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// SweepLegacyRecords deletes apps/<acct>/<app>/email.json for every app that
+// already has a senders/app.json, and nothing else.
+//
+// It is guarded by a flag that defaults OFF, and it is written now rather
+// than a release from now because the thing that makes it safe is knowing,
+// while the migration is fresh in view, exactly which object it may remove.
+// The rule is one sentence: a legacy record is deletable when, and only
+// when, the sender object that replaced it exists. An unmigrated app's
+// email.json is still the only copy of its email and is left alone, so a
+// deployment that turns the flag on early loses nothing.
+//
+// A failure on one app is logged and the sweep continues: one unreadable
+// object must not stop the other five hundred, and every deletion is
+// independent.
+//
+// It walks with ListDirs, twice, and never with List. List(ctx, "apps/") on
+// a production bucket returns every object under every app, every release
+// artifact of every version of every app, to find the handful of keys
+// ending in /email.json, and it would do it at console start. ListDirs is
+// the delimiter-aware listing (release/store.go:50-58): one page of
+// accounts, one page of apps per account, then two Gets per app.
+func SweepLegacyRecords(ctx context.Context, s release.Store, enabled bool) (int, error) {
+ if !enabled {
+ return 0, nil
+ }
+ accts, err := s.ListDirs(ctx, "apps/")
+ if err != nil {
+ return 0, fmt.Errorf("mail: sweep legacy records: %w", err)
+ }
+ deleted := 0
+ for _, acctPrefix := range accts {
+ // ListDirs hands back a full prefix ending in "/", never a bare
+ // name, so the account is the last segment of it.
+ acct := path.Base(strings.TrimSuffix(acctPrefix, "/"))
+ if !release.ValidAccount(acct) {
+ continue
+ }
+ apps, err := s.ListDirs(ctx, acctPrefix)
+ if err != nil {
+ log.Printf("mail: sweep legacy records: listing %s -> %v", acctPrefix, err)
+ continue
+ }
+ for _, appPrefix := range apps {
+ app := path.Base(strings.TrimSuffix(appPrefix, "/"))
+ if !release.ValidApp(app) {
+ continue
+ }
+ if _, _, err := s.Get(ctx, SenderKey(acct, app, SenderAppID)); err != nil {
+ if !errors.Is(err, os.ErrNotExist) {
+ log.Printf("mail: sweep legacy records: reading %s -> %v", SenderKey(acct, app, SenderAppID), err)
+ }
+ continue // not migrated: this object is still the only copy
+ }
+ legacy := LegacyKey(acct, app)
+ if _, _, err := s.Get(ctx, legacy); err != nil {
+ if !errors.Is(err, os.ErrNotExist) {
+ log.Printf("mail: sweep legacy records: reading %s -> %v", legacy, err)
+ }
+ continue // nothing to remove, which is the steady state
+ }
+ if err := s.Delete(ctx, legacy); err != nil {
+ log.Printf("mail: sweep legacy records: deleting %s -> %v", legacy, err)
+ continue
+ }
+ deleted++
+ }
+ }
+ return deleted, nil
+}
diff --git a/internal/mail/sweep_test.go b/internal/mail/sweep_test.go
new file mode 100644
index 000000000..b23fa9a50
--- /dev/null
+++ b/internal/mail/sweep_test.go
@@ -0,0 +1,63 @@
+package mail
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The sweep is what finally removes email.json, a release after the
+// migration stopped writing it. It is off by default and it never deletes a
+// legacy record for an app that has not been migrated, because that record
+// is still the only copy of that app's email.
+func TestSweepLegacyRecordsOnlyRemovesMigratedApps(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ now := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC)
+ seed := func(app string, migrated bool) {
+ body, _ := json.Marshal(&Record{Schema: Schema, App: app,
+ Declared: &Declared{EnvPrefix: "P", From: "noreply@x.com", At: now, By: "o@example.test"}})
+ if _, err := s.Put(ctx, LegacyKey("bag", app), body, ""); err != nil {
+ t.Fatal(err)
+ }
+ if migrated {
+ if _, err := PutSender(ctx, s, "bag", app, &Sender{ID: SenderAppID, Scope: ScopeApp, Source: SourceDomain}, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ seed("migrated", true)
+ seed("untouched", false)
+
+ // Off by default: the flag is the whole guard, and a console that has
+ // not been told to sweep sweeps nothing.
+ if n, err := SweepLegacyRecords(ctx, s, false); n != 0 || err != nil {
+ t.Fatalf("the sweep ran while disabled: %d %v", n, err)
+ }
+ if _, _, err := s.Get(ctx, LegacyKey("bag", "migrated")); err != nil {
+ t.Fatalf("the disabled sweep deleted something: %v", err)
+ }
+
+ n, err := SweepLegacyRecords(ctx, s, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if n != 1 {
+ t.Fatalf("deleted %d, want 1", n)
+ }
+ if _, _, err := s.Get(ctx, LegacyKey("bag", "migrated")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the migrated app's email.json survived: %v", err)
+ }
+ if _, _, err := s.Get(ctx, LegacyKey("bag", "untouched")); err != nil {
+ t.Fatalf("an unmigrated app's only record was deleted: %v", err)
+ }
+ // Running it twice is a no-op, which is what makes it safe at every start.
+ if n, err := SweepLegacyRecords(ctx, s, true); n != 0 || err != nil {
+ t.Fatalf("a second sweep = %d %v, want 0 and no error", n, err)
+ }
+}