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() + } + } +}