feat(worker): configurable retention for high-volume tables, off by default

.env.example

@@ -318,6 +318,22 @@ INROAD_RUN_SCHEDULER=true
 # the scheduler regardless of INROAD_RUN_SCHEDULER.
 INROAD_WORKER_ROLE=
 
+# --- Data retention (maintenance:retention, hourly, on the scheduling worker) --
+# Whole DAYS; 0 or blank disables that table's sweep. The four windows over
+# the five tables that identify the people your workspaces email (sends, inbox
+# threads + messages, tracking events, deliverability events) are DISABLED by
+# default on purpose:
+# how long to keep that data is a Privacy/Legal decision for your deployment
+# (applicable privacy law, your contracts), not something Inroad can pick for
+# you. Read "Data retention" in the environment-variable reference before
+# enabling any of them -- each has a minimum the worker enforces at startup, and
+# each has consequences for reporting. Dead letters keep their 90-day default.
+INROAD_RETENTION_SENDS_DAYS=
+INROAD_RETENTION_INBOX_DAYS=
+INROAD_RETENTION_TRACKING_EVENTS_DAYS=
+INROAD_RETENTION_DELIVERABILITY_EVENTS_DAYS=
+INROAD_RETENTION_DEAD_LETTERS_DAYS=
+
 # --- Fleet credential broker -------------------------------------------------
 # Only for a multi-host (role=send) deployment. Leave every one of these blank
 # for the single-process self-host topology; none of them exists for you.

cmd/worker/jobrunledger_test.go

@@ -27,6 +27,7 @@ var scheduledSweeps = map[string]string{
 	queue.TaskInboxSweep:            jobrun.NameInboxSweep,
 	queue.TaskWarmupSweep:           jobrun.NameWarmupSweep,
 	queue.TaskMaintenanceCleanup:    jobrun.NameMaintenanceCleanup,
+	queue.TaskMaintenanceRetention:  jobrun.NameRetention,
 	queue.TaskDomainAuthSweep:       jobrun.NameDomainAuthSweep,
 	queue.TaskRecipientESPSweep:     jobrun.NameRecipientESPSweep,
 	queue.TaskFleetRotate:           jobrun.NameFleetRotate,
@@ -40,9 +41,9 @@ var scheduledSweeps = map[string]string{
 // loudly if one were reached, which is the assertion: a periodic reconcile
 // should do nothing here but be counted. The methods that ARE overridden are
 // exactly the sweeps' entry points, each answering "nothing due", plus the
-// four capability interfaces Register resolves by type assertion
-// (maintenance.Cleaner, recipientesp.Core, fleet.Rotator and — reached via
-// inbox.RegisterScheduled — inbox.PendingSweepCore). Without those the handlers
+// five capability interfaces Register resolves by type assertion
+// (maintenance.Cleaner, maintenance.Retainer, recipientesp.Core, fleet.Rotator
+// and — reached via inbox.RegisterScheduled — inbox.PendingSweepCore). Without those the handlers
 // are never registered at all and ProcessTask would report "handler not found",
 // which is itself a failure this test should catch.
 type recordingCore struct {
@@ -79,14 +80,44 @@ func (c *recordingCore) CleanupExpired(context.Context) (int64, error)
 func (c *recordingCore) PurgeIdempotencyKeys(context.Context) (int64, error)    { return 0, nil }
 func (c *recordingCore) PurgeWarmupObservations(context.Context) (int64, error) { return 0, nil }
 func (c *recordingCore) PurgeDeadWorkers(context.Context) (int64, error)        { return 0, nil }
-func (c *recordingCore) PurgeDeadLetters(context.Context) (int64, error)        { return 0, nil }
 func (c *recordingCore) PurgeWebhookDeliveries(context.Context) (int64, error)  { return 0, nil }
 func (c *recordingCore) PurgeScheduledJobRuns(context.Context) (int64, error)   { return 0, nil }
 func (c *recordingCore) PurgeWorkerProviderSignals(context.Context) (int64, error) {
 	return 0, nil
 }
 func (c *recordingCore) PurgeFleetDecisions(context.Context) (int64, error) { return 0, nil }
 
+// --- maintenance.Retainer (resolved by type assertion in Register) ---
+//
+// Each answers "nothing aged out". The Deps below enable every table, so the
+// retention handler really does call each of these — a sweep that registered
+// but skipped every table would record a row while exercising nothing.
+
+func (c *recordingCore) PurgeDeliverabilityEvents(context.Context, coreapi.RetentionRequest) (coreapi.RetentionBatch, error) {
+	return coreapi.RetentionBatch{}, nil
+}
+func (c *recordingCore) PurgeInboxThreads(context.Context, coreapi.RetentionRequest) (coreapi.RetentionBatch, error) {
+	return coreapi.RetentionBatch{}, nil
+}
+func (c *recordingCore) RollupTrackingEvents(context.Context, coreapi.RetentionRequest) (coreapi.RetentionBatch, error) {
+	return coreapi.RetentionBatch{}, nil
+}
+func (c *recordingCore) PurgeSends(context.Context, coreapi.RetentionRequest) (coreapi.RetentionBatch, error) {
+	return coreapi.RetentionBatch{}, nil
+}
+func (c *recordingCore) PurgeDeadLetters(context.Context, coreapi.RetentionRequest) (coreapi.RetentionBatch, error) {
+	return coreapi.RetentionBatch{}, nil
+}
+func (c *recordingCore) LoadRetentionProgress(context.Context, string) (coreapi.RetentionProgress, error) {
+	return coreapi.RetentionProgress{}, nil
+}
+func (c *recordingCore) SaveRetentionProgress(context.Context, string, coreapi.RetentionCursor, bool) error {
+	return nil
+}
+func (c *recordingCore) TryRetentionSweepLock(context.Context) (func() error, bool, error) {
+	return func() error { return nil }, true, nil
+}
+
 // --- recipientesp.Core (resolved by type assertion in Register) ---
 
 func (c *recordingCore) PurgeExpiredRecipientDomains(context.Context, time.Duration) (int64, error) {
@@ -113,6 +144,7 @@ var (
 	_ coreapi.Client               = (*recordingCore)(nil)
 	_ jobrun.Recorder              = (*recordingCore)(nil)
 	_ maintenance.Cleaner          = (*recordingCore)(nil)
+	_ maintenance.Retainer         = (*recordingCore)(nil)
 	_ recipientesp.Core            = (*recordingCore)(nil)
 	_ fleet.Rotator                = (*recordingCore)(nil)
 	_ workerinbox.PendingSweepCore = (*recordingCore)(nil)
@@ -141,7 +173,11 @@ func TestEverySweepDispatchedThroughRegisterRecordsOneLedgerRow(t *testing.T) {
 	// nil senders/readers/resolvers/enqueuer: a sweep with nothing due reaches
 	// none of them, and a nil dereference here would mean a reconcile did more
 	// than scan-and-count. jobrun.Record re-panics, so it would fail loudly.
-	worker.Register(mux, worker.Deps{Role: worker.RoleAll, Core: core, PublicURL: "https://app.test"})
+	worker.Register(mux, worker.Deps{Role: worker.RoleAll, Core: core, PublicURL: "https://app.test",
+		Retention: maintenance.RetentionPolicy{
+			DeliverabilityEvents: 90 * 24 * time.Hour, InboxThreads: 30 * 24 * time.Hour,
+			TrackingEvents: 30 * 24 * time.Hour, Sends: 90 * 24 * time.Hour, DeadLetters: 90 * 24 * time.Hour,
+		}})
 
 	for taskType, wantName := range scheduledSweeps {
 		t.Run(taskType, func(t *testing.T) {

cmd/worker/main.go

@@ -38,6 +38,7 @@ import (
 	"github.com/inroad/inroad/internal/platform/warmup"
 	"github.com/inroad/inroad/internal/worker"
 	"github.com/inroad/inroad/internal/worker/fleetsignal"
+	"github.com/inroad/inroad/internal/worker/maintenance"
 )
 
 // workerHeartbeatInterval is how often a worker refreshes its `workers` row. It
@@ -125,6 +126,19 @@ func run() error {
 		return err
 	}
 
+	// The retention windows are validated against their per-table floors here,
+	// before anything connects, for the same reason as the role: a window short
+	// enough to delete rows a live rate still reads must stop the process, not
+	// start deleting. Logged either way, because "is retention on here, and for
+	// what" should be answerable from the startup log.
+	retention, err := retentionPolicy(cfg)
+	if err != nil {
+		logger.Error("invalid retention configuration", "err", err)
+		return err
+	}
+	logger.Info("retention", "enabled_days", retention.Enabled(),
+		"note", "recipient-data tables are disabled unless INROAD_RETENTION_*_DAYS is set; see the environment-variable reference")
+
 	// Prometheus /metrics listener. mtx is always constructed (never nil): the
 	// campaign/warmup send handlers' finalize points record into it
 	// unconditionally below; INROAD_METRICS_ADDR only controls whether the
@@ -447,6 +461,7 @@ func run() error {
 		TrackingSecret:      cfg.TrackingSecret,
 		WarmupSecret:        cfg.WarmupSecret,
 		WebhookAllowPrivate: cfg.WebhookAllowPrivate,
+		Retention:           retention,
 		Metrics:             mtx,
 	})
 
@@ -458,6 +473,23 @@ func run() error {
 	return nil
 }
 
+// retentionPolicy builds the retention sweep's policy from configuration and
+// refuses any window below its table's floor. Split out so the mapping — which
+// variable feeds which table — is testable without starting a worker.
+func retentionPolicy(cfg *config.Config) (maintenance.RetentionPolicy, error) {
+	p := maintenance.RetentionPolicyFromDays(maintenance.RetentionDays{
+		DeliverabilityEvents: cfg.RetentionDeliverabilityEventsDays,
+		InboxThreads:         cfg.RetentionInboxDays,
+		TrackingEvents:       cfg.RetentionTrackingEventsDays,
+		Sends:                cfg.RetentionSendsDays,
+		DeadLetters:          cfg.RetentionDeadLettersDays,
+	})
+	if err := p.Validate(); err != nil {
+		return maintenance.RetentionPolicy{}, err
+	}
+	return p, nil
+}
+
 // resolveWorkerQueues decides the queue set this process consumes. An operator
 // override (INROAD_WORKER_QUEUES, surfaced as cfg.WorkerQueues) wins ENTIRELY
 // when present — an operator who names queues means exactly those, and

cmd/worker/retention_test.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/inroad/inroad/internal/platform/config"
+	"github.com/inroad/inroad/internal/worker/maintenance"
+)
+
+const day = 24 * time.Hour
+
+// The composition root is the ONLY producer of worker.Deps.Retention, whose zero
+// value disables every table including dead letters. So this pins that the real
+// mapping from configuration does not silently produce that zero value: with
+// configuration at its defaults, dead letters keep their 90 days and every
+// recipient-identifying table stays off.
+func TestRetentionPolicyFromDefaultConfigKeepsDeadLetterRetention(t *testing.T) {
+	p, err := retentionPolicy(&config.Config{RetentionDeadLettersDays: config.DefaultRetentionDeadLettersDays})
+	if err != nil {
+		t.Fatalf("retentionPolicy: %v", err)
+	}
+	want := maintenance.RetentionPolicy{DeadLetters: 90 * day}
+	if p != want {
+		t.Fatalf("policy = %+v, want %+v", p, want)
+	}
+}
+
+// Each variable feeds its own table. A transposed pair here would pass every
+// config test and apply, say, the sends window to deliverability events.
+func TestRetentionPolicyMapsEachSettingToItsTable(t *testing.T) {
+	p, err := retentionPolicy(&config.Config{
+		RetentionDeliverabilityEventsDays: 100, RetentionInboxDays: 31, RetentionTrackingEventsDays: 32,
+		RetentionSendsDays: 101, RetentionDeadLettersDays: 8,
+	})
+	if err != nil {
+		t.Fatalf("retentionPolicy: %v", err)
+	}
+	want := maintenance.RetentionPolicy{
+		DeliverabilityEvents: 100 * day, InboxThreads: 31 * day, TrackingEvents: 32 * day,
+		Sends: 101 * day, DeadLetters: 8 * day,
+	}
+	if p != want {
+		t.Fatalf("policy = %+v, want %+v", p, want)
+	}
+}
+
+// Below a floor, the worker must refuse to start — and say which setting.
+func TestRetentionPolicyRefusesAWindowBelowItsFloor(t *testing.T) {
+	_, err := retentionPolicy(&config.Config{RetentionSendsDays: 30})
+	if err == nil {
+		t.Fatal("a 30-day sends window was accepted; the floor is 90")
+	}
+	if !strings.Contains(err.Error(), "INROAD_RETENTION_SENDS_DAYS") {
+		t.Fatalf("error %q does not name the setting", err)
+	}
+}

cmd/worker/scheduler.go

@@ -32,6 +32,7 @@ func sweepRegistrars() []sweepRegistrar {
 		{jobrun.NameInboxSweep, queue.RegisterInboxSweep},
 		{jobrun.NameWarmupSweep, queue.RegisterWarmupSweep},
 		{jobrun.NameMaintenanceCleanup, queue.RegisterMaintenanceCleanup},
+		{jobrun.NameRetention, queue.RegisterMaintenanceRetention},
 		{jobrun.NameDomainAuthSweep, queue.RegisterDomainAuthSweep},
 		{jobrun.NameRecipientESPSweep, queue.RegisterRecipientESPSweep},
 		{jobrun.NameFleetRotate, queue.RegisterFleetRotate},