From e790818d35acc73fa8a01346beec0580cc6640eb Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Wed, 27 May 2026 16:03:35 +0000 Subject: [PATCH] fleet: wire health collection so the autonomous loops actually have data Three small follow-ups that turn the fleet management system from 'all the pieces ship green' into 'actually produces telemetry': cmd/worker/main.go: go workerService.RunHealth(ctx, 30s) alongside Heartbeat. The sampler snapshots rolling 1m counters into a WorkerHealth event via the existing event bus + codec path. cmd/backend/main.go: background goroutine refreshes worker_capacity_view every minute via REFRESH MATERIALIZED VIEW CONCURRENTLY. The assignment loop, Rebalancer, Scaler, and QuarantineEvaluator all read from the view, so it's the freshness gate for the whole system. internal/app/worker/event_send_email.go + health_record.go: classify every wmail.SendResult into the right counter (auth / rate-limit / bounce-hard / bounce-soft / success) and record SMTP latency. Falls back to free-text message classification when the error code is generic, so signal stays useful as new error paths are added. End-to-end: a worker that bounces 10% of sends now lands in the 'quarantined' band within 5min of the QuarantineEvaluator tick, auto-drains via Rebalancer, and triggers a Scaler alert if its removal drops fleet capacity below the warning threshold. --- cmd/backend/main.go | 20 ++++++++ cmd/worker/main.go | 6 ++- internal/app/worker/event_send_email.go | 4 ++ internal/app/worker/health_record.go | 67 +++++++++++++++++++++++++ 4 files changed, 96 insertions(+), 1 deletion(-) create mode 100644 internal/app/worker/health_record.go diff --git a/cmd/backend/main.go b/cmd/backend/main.go index 89f06be5..9f1e516b 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -500,6 +500,26 @@ func main() { // interval and writes every action to decision_log. Cancel them via // the root context on shutdown. decisionLogRepo := repository.NewDecisionLogRepository(primaryDB) + + // Refresh worker_capacity_view every minute so the assignment loop + + // rebalance + scale + quarantine evaluators see fresh rolling + // metrics. The materialized view is what aggregates the 1h windows + // across all workers. + go func() { + tick := time.NewTicker(time.Minute) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + if err := workerRepository.RefreshWorkerCapacityView(ctx); err != nil { + log.Printf("worker_capacity_view refresh: %v", err) + } + } + } + }() + go (&fleet.Rebalancer{ WorkerRepo: workerRepository, Decisions: decisionLogRepo, diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 50500bbf..0b4929e5 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -6,6 +6,7 @@ import ( "os" "os/signal" "syscall" + "time" awsconf "github.com/aws/aws-sdk-go-v2/config" "github.com/google/uuid" @@ -159,8 +160,11 @@ func main() { workerService.InitEvents() - // Start heartbeat + // Start heartbeat + health sampler. RunHealth ticks every 30s, snapshots + // the rolling 1m counters into a WorkerHealth event, publishes via the + // event bus so the consumer can write a row into worker_health_samples. go workerService.Heartbeat(ctx) + go workerService.RunHealth(ctx, 30*time.Second) // Graceful shutdown sigCh := make(chan os.Signal, 1) diff --git a/internal/app/worker/event_send_email.go b/internal/app/worker/event_send_email.go index 8fa09894..3cb8801d 100644 --- a/internal/app/worker/event_send_email.go +++ b/internal/app/worker/event_send_email.go @@ -61,6 +61,8 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, body any) error { } // Use unified Send method + w.recordSendAttempt() + sendStart := time.Now() result := mail.Send(ctx, &wmail.SendRequest{ TaskID: sendEmail.TaskID, To: sendEmail.To, @@ -75,6 +77,8 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, body any) error { IsWarmup: sendEmail.IsWarmup, WarmupToken: sendEmail.WarmupToken, }) + w.recordSendLatency(time.Since(sendStart)) + w.recordSendOutcome(result) if result.Success { log.Info(). diff --git a/internal/app/worker/health_record.go b/internal/app/worker/health_record.go new file mode 100644 index 00000000..54d6c862 --- /dev/null +++ b/internal/app/worker/health_record.go @@ -0,0 +1,67 @@ +package worker + +import ( + "strings" + "time" + + "github.com/warmbly/warmbly/internal/app/worker/wmail" + "github.com/warmbly/warmbly/internal/errx" +) + +// recordSendAttempt + recordSendLatency + recordSendOutcome are the +// integration points the send hot path calls. They classify a +// wmail.SendResult into the right Record* shim so the rolling 1m +// counters feed the QuarantineEvaluator's band classification with the +// granularity that matters (bounce vs complaint vs auth vs rate limit). + +func (s *WorkerService) recordSendAttempt() { + s.RecordSendAttempt() +} + +func (s *WorkerService) recordSendLatency(d time.Duration) { + s.RecordSMTPLatency(int32(d.Milliseconds())) +} + +func (s *WorkerService) recordSendOutcome(result *wmail.SendResult) { + if result == nil { + return + } + if result.Success { + s.RecordSendSuccess() + return + } + if result.Error == nil { + return + } + switch result.Error.Code { + case errx.MailErrorCodeAuthenticationFailed, + errx.MailErrorCodeAuthorizationFailed, + errx.MailErrorCodeInvalidCredentials, + errx.MailErrorCodeGoogleAuth: + s.RecordAuthError() + case errx.MailErrorCodeRateLimitExceeded, + errx.MailErrorCodeSendingTooFast, + errx.MailErrorCodeQuotaExceeded: + s.RecordRateLimitError() + case errx.MailErrorCodeRecipientRejected, + errx.MailErrorCodeAccountSuspended: + s.RecordBounceHard() + case errx.MailErrorCodeServerUnreachable, + errx.MailErrorCodeConnectionLost: + s.RecordBounceSoft() + default: + // Best-effort classification on free-text — keeps the signal + // useful even when the error code is generic. + msg := strings.ToLower(result.Error.Message) + switch { + case strings.Contains(msg, "bounce") || strings.Contains(msg, "rejected"): + s.RecordBounceHard() + case strings.Contains(msg, "rate limit") || strings.Contains(msg, "throttle"): + s.RecordRateLimitError() + case strings.Contains(msg, "auth"): + s.RecordAuthError() + default: + s.RecordBounceSoft() + } + } +}