Documentation
¶
Overview ¶
Package janitor runs the low-urgency hourly cleanup sweep as a River periodic on QueueMaintenance. It replaces the hand-rolled time.Ticker(1h) goroutine that used to live in cmd/e2a/main.go, mirroring the webhook auto-disable janitor (internal/webhookdelivery.MaintenanceJobs) and the inbound retention sweep (internal/inboundprocess.RetentionWorker).
The sweep runs every prune SEQUENTIALLY (not concurrently — a janitor must be gentle on Postgres and there is no latency benefit to parallelism) and CONTINUES PAST any individual prune's error, so one failing DELETE never skips the rest. Sweep accumulates and returns the errors; the worker log-and-swallows them (returning nil) so a transient DB blip doesn't spin River's retry machinery — the next interval picks the work back up.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DeliveryPruner ¶
DeliveryPruner prunes expired webhook delivery records (*webhook.DeliveryStore).
type IdempotencyPruner ¶
IdempotencyPruner sweeps idempotency keys past their TTL (*idempotency.Store).
type Janitor ¶
type Janitor struct {
// contains filtered or unexported fields
}
Janitor holds the prune dependencies and runs the cleanup sweep. All fields are required except oauth, which is nil when the OAuth provider is disabled.
func New ¶
func New( messages MessagePruner, deliveries DeliveryPruner, subscribers SubscriberPruner, webhookEvent WebhookEventPruner, oauth OAuthPruner, idempotency IdempotencyPruner, metrics Metrics, ) *Janitor
New builds the Janitor. oauth may be nil (interface, not a typed-nil pointer) to skip the OAuth cleanup pass.
func (*Janitor) Sweep ¶
Sweep runs every prune once, sequentially, continuing past any individual error. It preserves the exact per-prune logging and metrics emission of the old hand-rolled ticker. Errors are accumulated and returned joined; the caller (the worker) logs-and-swallows so one failing prune never aborts the run or spins River's retry.
type JanitorArgs ¶
type JanitorArgs struct{}
JanitorArgs drives the periodic cleanup sweep. No fields — each run prunes the whole expired set.
func (JanitorArgs) Kind ¶
func (JanitorArgs) Kind() string
type MaintenanceJobs ¶
type MaintenanceJobs struct {
// contains filtered or unexported fields
}
MaintenanceJobs is the jobs.Registrar for the cleanup janitor: it contributes the MaintenanceWorker and a periodic that fires it on QueueMaintenance. No enqueuer needed — the schedule is the only trigger.
func NewMaintenanceJobs ¶
func NewMaintenanceJobs(j *Janitor) *MaintenanceJobs
NewMaintenanceJobs builds the registrar around a Janitor.
func (*MaintenanceJobs) RegisterJobs ¶
func (m *MaintenanceJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the maintenance worker + its periodic schedule. Mirrors the webhook janitor and senderidentity reaper: routed to QueueMaintenance, no UniqueOpts (River's periodic scheduler already inserts at most one per interval and a completed run must not dedup-block the next), RunOnStart:false (first sweep after one interval, matching the old ticker's first-tick-after-1h).
Overlap tradeoff (inherited from the pre-River janitor, not introduced by this migration): with no UniqueOpts and QueueMaintenance's small worker pool, a sweep that runs longer than the 1h interval can overlap the next tick. That is safe — every prune is an idempotent batched DELETE on a distinct table and a partial sweep is simply finished by the next interval; overlapping sweeps only race to delete the same already-doomed rows. It only bites against a large, long-unpruned table (e.g. the first deploy); steady-state sweeps are near-empty. The prunes' unbounded-DELETE hazard is resolved — each is now a ctx-aware ctid-LIMIT drain loop (see the DeleteExpired* store methods) bounded by MaintenanceWorker.Timeout.
type MaintenanceWorker ¶
type MaintenanceWorker struct {
river.WorkerDefaults[JanitorArgs]
// contains filtered or unexported fields
}
MaintenanceWorker runs the cleanup sweep once per scheduled job. Sweep's errors are logged (with a janitor prefix) and swallowed — Work returns nil so a transient DB blip never spins River's retry for a best-effort idempotent sweep; the next interval picks it up.
func NewMaintenanceWorker ¶
func NewMaintenanceWorker(j *Janitor) *MaintenanceWorker
NewMaintenanceWorker builds the worker around a Janitor. Exported so tests can drive Work directly (RegisterJobs builds an identical one for the client).
func (*MaintenanceWorker) Timeout ¶
func (w *MaintenanceWorker) Timeout(*river.Job[JanitorArgs]) time.Duration
Timeout gives the sweep a bounded 5-minute budget instead of River's 60s default JobTimeout. Every prune is now a ctx-aware batched delete (each 5000-row batch autocommits), so a cut is always safe — partial progress persists and the next hourly tick resumes — but 60s is tight for a first-deploy backlog where `messages` (pruned first) can eat the whole budget and starve the later tables. Five minutes lets a realistic backlog drain in far fewer ticks while still capping how long one sweep holds a QueueMaintenance slot. Deliberately NOT -1 (unbounded) like the HITL send-sweep, which must never be cut mid-SMTP; the janitor's deletes are safe to interrupt.
func (*MaintenanceWorker) Work ¶
func (w *MaintenanceWorker) Work(ctx context.Context, _ *river.Job[JanitorArgs]) error
type MessagePruner ¶
type MessagePruner interface {
DeleteExpiredMessages(ctx context.Context) (int64, error)
DeleteExpiredUserSessions(ctx context.Context) (int64, error)
PurgeDeletedAgents(ctx context.Context) (int64, error)
}
MessagePruner purges messages past trash retention, expired user sessions, and trashed agents past their retention window (all live on *identity.Store today). DeleteExpiredMessages covers both message arms: live rows past their trashed rows past TrashRetention.
type Metrics ¶
Metrics is the narrow slice of telemetry.Metrics the janitor emits. Injectable so tests don't need a real backend; satisfied by *telemetry.Log / telemetry.NoOp.
type OAuthPruner ¶
type OAuthPruner interface {
CleanupExpired(ctx context.Context, now time.Time) (oauth.RetentionResult, error)
}
OAuthPruner cleans up expired OAuth rows (*oauth.Storage). Optional: when the OAuth provider isn't configured the dependency is nil and the pass is skipped (preserving the old `if oauthStorage != nil` guard). Pass a nil interface — NOT a typed nil *oauth.Storage — to skip it.
Source Files