test
", "test") + if (err == nil) != (tc.class == "") { + t.Errorf("delivery error = %v; want class %q", err, tc.class) + } + if tc.class != "" && ClassifyNotificationFailureError(err) != tc.class { + t.Errorf("failure class = %q; want %q", ClassifyNotificationFailureError(err), tc.class) + } + if got := connections.Load(); got != tc.attempts { + t.Errorf("connections = %d; want %d", got, tc.attempts) + } + wantAccepted := int32(0) + if tc.dataCode == 250 { + wantAccepted = 1 + } + if got := accepted.Load(); got != wantAccepted { + t.Errorf("accepted messages = %d; want %d", got, wantAccepted) + } + if strings.Contains(captured.String(), "private-cleanup-token") { + t.Error("cleanup diagnostics exposed response text") + } + }) + } + } +} + +func TestSMTPAcceptedThreadedAndAttachmentDelivery(t *testing.T) { + for _, mode := range []string{"plain", "TLS", "STARTTLS"} { + for _, variant := range []string{"threaded", "attachment"} { + t.Run(mode+"/"+variant, func(t *testing.T) { + manager, connections, accepted := acceptanceSMTPServer(t, mode, 250, "") + var err error + if variant == "threaded" { + err = manager.SendEmailThreaded("test", "test
", "test", "test
", "test", + []EmailAttachment{{Filename: "test.txt", ContentType: "text/plain", Data: []byte("test")}}) + } + if err != nil || connections.Load() != 1 || accepted.Load() != 1 { + t.Errorf("error=%v connections=%d accepted=%d; want one accepted message", err, connections.Load(), accepted.Load()) + } + }) + } + } +} + +func TestSMTPConnectionCheckAndCertificateFailureRemainFailures(t *testing.T) { + for _, mode := range []string{"plain", "TLS", "STARTTLS"} { + t.Run(mode, func(t *testing.T) { + manager, connections, accepted := acceptanceSMTPServer(t, mode, 250, "500 cleanup rejected") + if err := manager.TestConnection(); err == nil { + t.Error("connection test without DATA reported success after rejected QUIT") + } + if accepted.Load() != 0 || connections.Load() != 1 { + t.Errorf("connection check sent a message or retried: %d / %d", accepted.Load(), connections.Load()) + } + if mode != "plain" { + manager.config.SkipTLSVerify = false + manager.config.MaxRetries = 0 + err := manager.SendEmailWithRetry("test", "test", "test") + if err == nil || ClassifyNotificationFailureError(err) != NotificationFailureTLS || accepted.Load() != 0 { + t.Errorf("certificate failure was hidden: error=%v accepted=%d", err, accepted.Load()) + } + } + }) + } +} + +func TestSMTPAcceptedMessageQueueAuditAcrossRestart(t *testing.T) { + manager, connections, accepted := acceptanceSMTPServer(t, "plain", 250, "") + dir := t.TempDir() + q, err := NewNotificationQueue(dir) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = q.Stop() }) + q.SetProcessor(func(*QueuedNotification) error { return manager.SendEmailWithRetry("test", "test", "test") }) + if err := q.Enqueue(&QueuedNotification{ID: "accepted", Type: "email", Config: []byte(`{}`), MaxAttempts: 1}); err != nil { + t.Fatal(err) + } + waitVerdictAudit(t, q, "accepted") + if err := q.Stop(); err != nil { + t.Fatal(err) + } + q, err = NewNotificationQueue(dir) + if err != nil { + t.Fatal(err) + } + var status, class string + var attempts, success int + if err := q.db.QueryRow(`SELECT status, attempts, success, failure_class FROM notification_audit WHERE notification_id = 'accepted'`).Scan(&status, &attempts, &success, &class); err != nil { + t.Fatal(err) + } + stats, err := q.GetTelemetryStats(time.Now().Add(-time.Hour)) + if err != nil { + t.Fatal(err) + } + if status != string(QueueStatusSent) || attempts != 1 || success != 1 || class != "" || + stats.Attempts != 1 || stats.Deliveries != 1 || stats.Failures != 0 || connections.Load() != 1 || accepted.Load() != 1 { + t.Errorf("accepted message became duplicate/failure: audit=%s/%d/%d/%s stats=%+v connections=%d accepted=%d", + status, attempts, success, class, stats, connections.Load(), accepted.Load()) + } + if count, err := q.RetryTerminalFailures(); err != nil || count != 0 { + t.Errorf("accepted message available for operator retry: %d / %v", count, err) + } +} + +func waitVerdictAudit(t *testing.T, q *NotificationQueue, id string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + var count int + if err := q.db.QueryRow(`SELECT count(*) FROM notification_audit WHERE notification_id = ?`, id).Scan(&count); err != nil { + t.Fatal(err) + } + if count > 0 { + return + } + if time.Now().After(deadline) { + t.Fatal(fmt.Sprintf("missing delivery audit for %s", id)) + } + time.Sleep(time.Millisecond) + } +} diff --git a/internal/notifications/email_enhanced.go b/internal/notifications/email_enhanced.go index ffc644fc32..63c411fe7a 100644 --- a/internal/notifications/email_enhanced.go +++ b/internal/notifications/email_enhanced.go @@ -669,7 +669,8 @@ func (e *EnhancedEmailManager) sendTLS(addr string, msg []byte, addresses resolv return fmt.Errorf("message close failed: %w", err) } - return client.Quit() + finishAcceptedSMTPSession(client) + return nil } // sendStartTLS sends email using STARTTLS @@ -739,7 +740,8 @@ func (e *EnhancedEmailManager) sendStartTLS(addr string, msg []byte, addresses r return fmt.Errorf("message close failed: %w", err) } - return client.Quit() + finishAcceptedSMTPSession(client) + return nil } // TestConnection tests the email server connection @@ -859,5 +861,18 @@ func (e *EnhancedEmailManager) sendPlain(addr string, msg []byte, addresses reso return fmt.Errorf("message close failed: %w", err) } - return client.Quit() + finishAcceptedSMTPSession(client) + return nil +} + +// The successful final DATA reply has transferred responsibility for the +// message to the server. QUIT is session cleanup, not another delivery verdict: +// returning its error would resend an accepted message at either retry layer. +// Missing or rejected DATA replies still take the ordinary failure path. +func finishAcceptedSMTPSession(client *smtp.Client) { + if err := client.Quit(); err != nil { + // Reply text may contain private provider details; keep this diagnostic + // fixed and separate from the successful delivery audit. + log.Debug().Msg("SMTP session cleanup failed after message acceptance") + } } diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index 1741712f9d..18d01ea67a 100644 --- a/internal/notifications/notifications.go +++ b/internal/notifications/notifications.go @@ -3035,6 +3035,12 @@ func (n *NotificationManager) sendResolvedWebhookNtfy(webhook WebhookConfig, ale Str("webhook", webhook.Name). Str("service", "ntfy"). Msg("failed to read resolved ntfy webhook response body") + // A rejected recovery has the same authoritative HTTP verdict as a + // firing delivery, even if its diagnostic body is interrupted. Keep + // permanent refusals out of the connectivity retry path. + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return FailfWithClass(ClassFromHTTPStatus(resp.StatusCode), "ntfy webhook returned HTTP %d: failed to read ntfy webhook response: %w", resp.StatusCode, err) + } return fmt.Errorf("failed to read ntfy webhook response: %w", err) } diff --git a/internal/notifications/ntfy_response_verdict_test.go b/internal/notifications/ntfy_response_verdict_test.go new file mode 100644 index 0000000000..9fdf4576bb --- /dev/null +++ b/internal/notifications/ntfy_response_verdict_test.go @@ -0,0 +1,119 @@ +package notifications + +import ( + "errors" + "fmt" + "io" + "net/http" + "strconv" + "sync/atomic" + "testing" + "time" +) + +func TestResolvedNtfyTruncatedResponseRetainsVerdict(t *testing.T) { + for _, code := range []int{200, 401, 403, 421, 422, 423, 425, 429, 503} { + t.Run(strconv.Itoa(code), func(t *testing.T) { + server := newIPv4HTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", "100") + w.WriteHeader(code) + _, _ = fmt.Fprint(w, "short") + })) + defer server.Close() + manager := &NotificationManager{webhookClient: server.Client(), webhookRateLimits: make(map[string]*webhookRateLimit)} + if err := manager.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil { + t.Fatal(err) + } + err := manager.sendResolvedWebhookNtfy(WebhookConfig{URL: server.URL, Service: "ntfy"}, nil, time.Now()) + if !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("response read error lost: %v", err) + } + want := ClassFromHTTPStatus(code) + if code == 200 { // Do not manufacture success for an incomplete 2xx response. + want = NotificationFailureConnectivity + } + if got := ClassifyNotificationFailureError(err); got != want { + t.Errorf("class = %s; want %s", got, want) + } + }) + } +} + +func TestResolvedNtfyRejectionAuditAndOperatorRetry(t *testing.T) { + var repaired atomic.Bool + var requests atomic.Int32 + server := newIPv4HTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + if repaired.Load() { + w.WriteHeader(http.StatusNoContent) + return + } + w.Header().Set("Content-Length", "100") + w.WriteHeader(http.StatusUnauthorized) + _, _ = fmt.Fprint(w, "short") + })) + defer server.Close() + manager := &NotificationManager{webhookClient: server.Client(), webhookRateLimits: make(map[string]*webhookRateLimit)} + if err := manager.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil { + t.Fatal(err) + } + dir := t.TempDir() + open := func() *NotificationQueue { + q, err := NewNotificationQueue(dir) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = q.Stop() }) + return q + } + processor := func(*QueuedNotification) error { + return manager.sendResolvedWebhookNtfy(WebhookConfig{URL: server.URL, Service: "ntfy"}, nil, time.Now()) + } + q := open() + q.SetProcessor(processor) + if err := q.Enqueue(&QueuedNotification{ID: "rejected", Type: "webhook_resolved", Config: []byte(`{}`), MaxAttempts: 3}); err != nil { + t.Fatal(err) + } + waitVerdictAudit(t, q, "rejected") + if err := q.Stop(); err != nil { + t.Fatal(err) + } + q = open() + var status, class string + var attempts int + if err := q.db.QueryRow(`SELECT status, attempts, failure_class FROM notification_audit WHERE notification_id = 'rejected'`).Scan(&status, &attempts, &class); err != nil { + t.Fatal(err) + } + stats, err := q.GetTelemetryStats(time.Now().Add(-time.Hour)) + if err != nil { + t.Fatal(err) + } + if status != string(QueueStatusDLQ) || attempts != 1 || class != string(NotificationFailureAuthentication) || + stats.Attempts != 1 || stats.Failures != 1 || stats.Deliveries != 0 || stats.FailureClasses.Authentication != 1 || requests.Load() != 1 { + t.Fatalf("rejection lost: audit=%s/%d/%s stats=%+v HTTP requests=%d", status, attempts, class, stats, requests.Load()) + } + // Only an explicit operator retry after the destination is repaired resumes + // delivery. It retains the original terminal failure and the successful send. + repaired.Store(true) + if count, err := q.RetryTerminalFailures(); err != nil || count != 1 { + t.Fatalf("operator retry = %d / %v; want one retained failure", count, err) + } + q.SetProcessor(processor) + deadline := time.Now().Add(5 * time.Second) + for { + stats, err = q.GetTelemetryStats(time.Now().Add(-time.Hour)) + if err != nil { + t.Fatal(err) + } + if stats.Deliveries == 1 { + break + } + if time.Now().After(deadline) { + t.Fatal("operator retry did not deliver") + } + time.Sleep(time.Millisecond) + } + if stats.Attempts != 2 || stats.Failures != 1 || stats.FailureClasses.Authentication != 1 || requests.Load() != 2 { + t.Errorf("operator retry erased history or duplicated delivery: stats=%+v HTTP requests=%d", stats, requests.Load()) + } +}