Preserve notification verdicts through SMTP cleanup and ntfy diagnostics

Stop resending mail after the final DATA acceptance solely because QUIT failed. Keep interrupted ntfy recovery rejections authoritative in the retry decision and audit. Preserve missing acceptance, TLS checks, retry budgets and explicit operator recovery; verify all SMTP transports and persistent queue/audit readback with synthetic local receivers.

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/notifications.md

@@ -15,6 +15,37 @@
 
 ## Purpose
 
+### Delivery verdicts survive session cleanup and interrupted diagnostics
+
+A successful final SMTP DATA reply establishes server acceptance. Plain SMTP,
+TLS and STARTTLS delivery must not resend an accepted message because QUIT
+fails, nor turn that cleanup failure into a terminal delivery audit. Cleanup
+still runs, with only a fixed local diagnostic; the provider's reply text is
+not copied into it. Missing acknowledgement, certificate failures and rejected
+DATA replies remain errors with their existing retry policy. A connection test
+without a completed DATA transaction still reports QUIT failure. Server
+acceptance is not proof of arrival in a recipient's inbox or exactly-once
+recovery after a process crash.
+
+The ntfy plain-text recovery path, like the shared firing sender, retains the
+HTTP rejection class when reading the diagnostic response body fails. It
+preserves the wrapped read error, existing retry budgets and bounded body read;
+401/403/422 remain permanent failures, while transient HTTP verdicts remain
+retryable. Incomplete 2xx responses retain the existing failure behaviour.
+Historical audit rows are not rewritten and operator retry stays explicit.
+
+Verification: `email_acceptance_test.go` uses actual guest-local plain/TLS/
+STARTTLS SMTP sessions for clean, missing and rejected QUIT, permanent and
+transient DATA rejection, missing acknowledgement, threaded/attachment sends,
+connection-only tests and certificate refusal. Its connected persistent queue
+control retains one accepted send and zero terminal failures after reopen.
+`ntfy_response_verdict_test.go` exercises real interrupted HTTP bodies, then
+queue/audit/telemetry readback after reopen and an explicit repaired-destination
+retry without erasing the original rejection. These are synthetic source
+controls, not a cause attribution for aggregate fleet counters, installed
+provider acceptance or a reporter's confirmation.
+
+
 ### Pending firing groups count occurrences, not callbacks
 
 Before a grouping window expires, repeated firing callbacks with the same full

internal/notifications/email_acceptance_test.go

@@ -0,0 +1,264 @@
+package notifications
+
+import (
+	"crypto/tls"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/textproto"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// Real guest-local SMTP connections cover all three transports. A message is
+// accepted only after the final DATA reply, not when its bytes arrive. QUIT
+// errors deliberately contain private-looking prose: cleanup diagnostics must
+// neither resend the accepted message nor copy that prose into a new log.
+func acceptanceSMTPServer(t *testing.T, mode string, dataCode int, quitReply string) (*EnhancedEmailManager, *atomic.Int32, *atomic.Int32) {
+	t.Helper()
+	certServer := httptest.NewTLSServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
+	tlsConfig := certServer.TLS.Clone()
+	certServer.Close()
+	listener, err := net.Listen("tcp4", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var sessions sync.WaitGroup
+	connections, accepted := &atomic.Int32{}, &atomic.Int32{}
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for {
+			raw, err := listener.Accept()
+			if err != nil {
+				return
+			}
+			connections.Add(1)
+			sessions.Add(1)
+			go func() {
+				defer sessions.Done()
+				defer raw.Close()
+				_ = raw.SetDeadline(time.Now().Add(5 * time.Second))
+				wire := raw
+				if mode == "TLS" {
+					wire = tls.Server(raw, tlsConfig)
+				}
+				conn := textproto.NewConn(wire)
+				if err := conn.PrintfLine("220 localhost ESMTP"); err != nil {
+					return
+				}
+				upgraded := mode == "TLS"
+				for {
+					line, err := conn.ReadLine()
+					if err != nil {
+						return
+					}
+					switch strings.Fields(line)[0] {
+					case "EHLO", "HELO":
+						if mode == "STARTTLS" && !upgraded {
+							err = conn.PrintfLine("250-localhost\r\n250 STARTTLS")
+						} else {
+							err = conn.PrintfLine("250 localhost")
+						}
+					case "STARTTLS":
+						if err = conn.PrintfLine("220 Ready for TLS"); err == nil {
+							conn = textproto.NewConn(tls.Server(raw, tlsConfig))
+							upgraded = true
+						}
+					case "MAIL", "RCPT":
+						err = conn.PrintfLine("250 OK")
+					case "DATA":
+						if err = conn.PrintfLine("354 Send message"); err != nil {
+							return
+						}
+						if _, err = conn.ReadDotBytes(); err != nil {
+							return
+						}
+						if dataCode == 0 { // No acknowledgement: acceptance is unknown.
+							return
+						}
+						err = conn.PrintfLine("%d transaction verdict", dataCode)
+						if err == nil && dataCode == 250 {
+							accepted.Add(1)
+						}
+					case "QUIT":
+						if quitReply != "" {
+							_ = conn.PrintfLine("%s", quitReply)
+						}
+						return
+					default:
+						t.Errorf("unexpected SMTP command %q", line)
+						return
+					}
+					if err != nil {
+						return
+					}
+				}
+			}()
+		}
+	}()
+	t.Cleanup(func() { _ = listener.Close(); <-done; sessions.Wait() })
+	_, rawPort, err := net.SplitHostPort(listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	port, err := strconv.Atoi(rawPort)
+	if err != nil {
+		t.Fatal(err)
+	}
+	manager := NewEnhancedEmailManager(EmailProviderConfig{
+		EmailConfig: EmailConfig{SMTPHost: "127.0.0.1", SMTPPort: port,
+			From: "pulse@example.test", To: []string{"recipient@example.test"}, TLS: mode == "TLS"},
+		StartTLS: mode == "STARTTLS", SkipTLSVerify: true, // Ephemeral fixture certificate only.
+		MaxRetries: 2, RetryDelay: 0,
+	})
+	return manager, connections, accepted
+}
+
+func TestSMTPDeliveryAcceptanceSurvivesCleanup(t *testing.T) {
+	for _, mode := range []string{"plain", "TLS", "STARTTLS"} {
+		for _, tc := range []struct {
+			name, quit string
+			dataCode   int
+			class      NotificationFailureClass
+			attempts   int32
+		}{
+			{"clean", "221 Bye", 250, "", 1},
+			{"closed at QUIT", "", 250, "", 1},
+			{"rejected QUIT", "500 private-cleanup-token", 250, "", 1},
+			{"auth prose at QUIT", "535 private-cleanup-token", 250, "", 1},
+			{"message rejected", "221 Bye", 550, NotificationFailureRejected, 1},
+			{"temporary message failure", "221 Bye", 451, NotificationFailureServerError, 3},
+			{"missing acceptance", "221 Bye", 0, NotificationFailureUnknown, 3},
+		} {
+			t.Run(mode+"/"+tc.name, func(t *testing.T) {
+				manager, connections, accepted := acceptanceSMTPServer(t, mode, tc.dataCode, tc.quit)
+				captured := captureNotificationQueueLogs(t)
+				err := manager.SendEmailWithRetry("test", "<p>test</p>", "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", "<p>test</p>", "test", "<incident@example.test>")
+				} else {
+					err = manager.SendEmailWithAttachments("test", "<p>test</p>", "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)
+	}
+}

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

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

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