Refuse with 406, retry cut-off uploads, and log arrival lag

internal/mailgun/webhook.go

@@ -4,11 +4,14 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"mime/multipart"
 	"net/http"
+	"net/mail"
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 
 	"filemill/internal/alert"
 	"filemill/internal/app"
@@ -21,7 +24,8 @@ const requestOverhead = 1 << 20 // 1 MiB
 // handle is the webhook entry point. It delegates the decision to receive and
 // then maps the single (status, reason) result onto the HTTP response: reasons
 // are logged (useful for a public endpoint), and 4xx/5xx additionally send an
-// error body so Mailgun can distinguish "don't retry" (4xx) from "retry" (5xx).
+// error body. Mailgun itself only distinguishes 406 ("don't retry") from every
+// other non-200 code ("retry").
 func (s *Service) handle(w http.ResponseWriter, r *http.Request) {
 	// Unconditional arrival log: any request that reaches this handler is
 	// recorded before any check, so "did the POST arrive?" is never ambiguous.
@@ -43,10 +47,15 @@ func (s *Service) handle(w http.ResponseWriter, r *http.Request) {
 // taxonomy is deliberate:
 //
 //	200 - accepted, or "received, nothing to do" (Mailgun should stop)
-//	400 - the client's fault; do not retry (malformed body, oversize)
 //	401 - bad, missing, or stale signature
 //	405 - not a POST
-//	500 - our fault; please retry (storage, filesystem, submit failures)
+//	406 - can never succeed; do not retry (malformed body, oversize)
+//	500 - our fault or the network's; please retry (cut-off upload, storage,
+//	      filesystem, submit failures)
+//
+// Mailgun retries every non-200 response except 406, on a fixed schedule over
+// 8 hours, so a permanent refusal must be 406 - a 400 is retried just like a
+// 500.
 //
 // A rejected attachment (wrong file type) is the sender's fault and can never
 // succeed on retry, so it maps to 200, not 500 — returning 500 made Mailgun
@@ -56,8 +65,10 @@ func (s *Service) receive(r *http.Request) (status int, reason string) {
 	if r.Method != http.MethodPost {
 		return http.StatusMethodNotAllowed, "method not allowed"
 	}
-	if err := parseForm(r); err != nil {
-		return http.StatusBadRequest, "malformed request body"
+	body := &countingBody{ReadCloser: r.Body}
+	r.Body = body
+	if status, reason, failed := bodyFailure(parseForm(r), body, r.ContentLength); failed {
+		return status, reason
 	}
 	if !authenticSignature(s.signKey, r.FormValue("timestamp"), r.FormValue("token"), r.FormValue("signature")) {
 		return http.StatusUnauthorized, "invalid signature"
@@ -110,7 +121,7 @@ func (s *Service) receive(r *http.Request) (status int, reason string) {
 	}
 
 	if !s.withinSizeLimit(files) {
-		return http.StatusBadRequest, fmt.Sprintf("attachment exceeds size limit (to %s from %s)", recipient, sender)
+		return http.StatusNotAcceptable, fmt.Sprintf("attachment exceeds size limit (to %s from %s)", recipient, sender)
 	}
 
 	if err := s.intake(r, operation, files); err != nil {
@@ -127,7 +138,85 @@ func (s *Service) receive(r *http.Request) (status int, reason string) {
 		})
 		return http.StatusInternalServerError, fmt.Sprintf("intake failed from %s: %v", sender, err)
 	}
-	return http.StatusOK, fmt.Sprintf("accepted: sender=%q recipient=%q operation=%s", sender, r.FormValue("recipient"), operation)
+	return http.StatusOK, fmt.Sprintf("accepted: sender=%q recipient=%q operation=%s %s", sender, r.FormValue("recipient"), operation, s.arrival(r))
+}
+
+// countingBody counts the bytes read from a request body and keeps the first
+// read error, so a body that would not parse can be told apart: cut off in
+// transit, or complete but malformed.
+type countingBody struct {
+	io.ReadCloser
+	n   int64
+	err error
+}
+
+func (c *countingBody) Read(p []byte) (int, error) {
+	n, err := c.ReadCloser.Read(p)
+	c.n += int64(n)
+	if err != nil && err != io.EOF && c.err == nil {
+		c.err = err
+	}
+	return n, err
+}
+
+// bodyFailure judges the request body after parsing, reporting failed when the
+// request must not go on. The rest of the body is drained first: the parser
+// stops at its first error, often long before the end, and it also takes a
+// clean end-of-stream between parts as the end of the form - so a body cut
+// short there parses without error, and only the byte count shows it is
+// incomplete. A cut-off upload (Mailgun gives up after 10 seconds) is not the
+// sender's fault and is retried (500); a body that is too large, or arrived
+// whole and still would not parse, can never succeed and is refused (406).
+func bodyFailure(parseErr error, body *countingBody, declared int64) (status int, reason string, failed bool) {
+	_, _ = io.Copy(io.Discard, body)
+	var tooLarge *http.MaxBytesError
+	switch {
+	case errors.As(parseErr, &tooLarge) || errors.As(body.err, &tooLarge):
+		return http.StatusNotAcceptable, fmt.Sprintf("request body exceeds %d bytes", tooLarge.Limit), true
+	case body.err != nil:
+		return http.StatusInternalServerError, fmt.Sprintf("request body incomplete: got %d bytes: %v", body.n, body.err), true
+	case declared > 0 && body.n < declared:
+		return http.StatusInternalServerError, fmt.Sprintf("request body incomplete: got %d of %d bytes", body.n, declared), true
+	case parseErr != nil:
+		return http.StatusNotAcceptable, fmt.Sprintf("malformed request body: %v", parseErr), true
+	}
+	return 0, "", false
+}
+
+// arrival describes when an email was sent (its Date header), when Mailgun
+// posted it (the signed timestamp), and the lag from sending to arrival here.
+// Without these, a slow reply could not be placed: before the sender's email
+// reached Mailgun, between Mailgun and here, or after.
+func (s *Service) arrival(r *http.Request) string {
+	posted := "unknown"
+	if unix, err := strconv.ParseInt(r.FormValue("timestamp"), 10, 64); err == nil {
+		posted = time.Unix(unix, 0).UTC().Format(time.RFC3339)
+	}
+	sent, ok := sentAt(r)
+	if !ok {
+		return "sent=unknown posted=" + posted
+	}
+	lag := s.clock().Sub(sent).Round(time.Second)
+	return fmt.Sprintf("sent=%s posted=%s lag=%s", sent.UTC().Format(time.RFC3339), posted, lag)
+}
+
+// sentAt reads the email's Date header. Mailgun posts headers as top-level
+// fields (Message-Id is read the same way), and also as the message-headers
+// JSON list of [name, value] pairs, which is the fallback.
+func sentAt(r *http.Request) (time.Time, bool) {
+	raw := r.FormValue("Date")
+	if raw == "" {
+		var headers [][]string
+		_ = json.Unmarshal([]byte(r.FormValue("message-headers")), &headers)
+		for _, h := range headers {
+			if len(h) == 2 && strings.EqualFold(h[0], "Date") {
+				raw = h[1]
+				break
+			}
+		}
+	}
+	sent, err := mail.ParseDate(raw)
+	return sent, err == nil
 }
 
 // parseForm decodes the request body. Mailgun sends no-attachment notifications

internal/mailgun/webhook_test.go

@@ -5,7 +5,9 @@ import (
 	"crypto/hmac"
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
 	"fmt"
+	"io"
 	"log"
 	"mime/multipart"
 	"net/http"
@@ -70,8 +72,186 @@ func TestWebhookRejectsOversizeAttachment(t *testing.T) {
 		map[string][]byte{"attachment-1": []byte("too large")})
 	w := httptest.NewRecorder()
 	s.handle(w, r)
-	if w.Code != http.StatusBadRequest {
-		t.Fatalf("got %d, want 400", w.Code)
+	if w.Code != http.StatusNotAcceptable {
+		t.Fatalf("got %d, want 406 (an oversize attachment can never succeed on retry)", w.Code)
+	}
+}
+
+// Mailgun retries every non-200 response except 406, so a body that will never
+// parse must get 406 or Mailgun resends it for 8 hours. The parser's error is
+// logged, because "malformed" alone could not be diagnosed after the fact.
+func TestWebhookRefusesMalformedBody(t *testing.T) {
+	logger, buf := captureLogger()
+	s := &Service{signKey: "key", maxBytes: 1024, routes: map[string]string{}, allowed: map[string]bool{}, log: logger}
+	r := httptest.NewRequest(http.MethodPost, "/mailgun-webhook", strings.NewReader("this is not multipart"))
+	r.Header.Set("Content-Type", "multipart/form-data; boundary=xyz")
+	w := httptest.NewRecorder()
+	s.handle(w, r)
+	if w.Code != http.StatusNotAcceptable {
+		t.Fatalf("got %d, want 406", w.Code)
+	}
+	if !strings.Contains(buf.String(), "malformed request body: ") {
+		t.Errorf("the parser's error must be logged; got %q", buf.String())
+	}
+}
+
+func TestWebhookRefusesMultipartWithoutBoundary(t *testing.T) {
+	s := &Service{signKey: "key", maxBytes: 1024, routes: map[string]string{}, allowed: map[string]bool{}, log: discardLogger()}
+	r := httptest.NewRequest(http.MethodPost, "/mailgun-webhook", strings.NewReader("--xyz--"))
+	r.Header.Set("Content-Type", "multipart/form-data")
+	w := httptest.NewRecorder()
+	s.handle(w, r)
+	if w.Code != http.StatusNotAcceptable {
+		t.Fatalf("got %d, want 406", w.Code)
+	}
+}
+
+func TestWebhookRefusesBodyOverTheRequestLimit(t *testing.T) {
+	logger, buf := captureLogger()
+	s := &Service{signKey: "key", maxBytes: 2, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger}
+	r := signedMultipart(t, s.signKey,
+		map[string]string{"recipient": "workerlist@mill.example.com"},
+		map[string][]byte{"attachment-1": bytes.Repeat([]byte("x"), requestOverhead+1024)})
+	w := httptest.NewRecorder()
+	s.handle(w, r)
+	if w.Code != http.StatusNotAcceptable {
+		t.Fatalf("got %d, want 406", w.Code)
+	}
+	if !strings.Contains(buf.String(), "request body exceeds") {
+		t.Errorf("the size refusal must say so; got %q", buf.String())
+	}
+}
+
+// A body that stops short of its Content-Length was cut off in transit - on
+// 2026-10-28 Mailgun hung up after 10 seconds mid-upload. That is not the
+// sender's fault and the retry will likely succeed, so it must stay retryable.
+// Every cut point is tried: the multipart parser takes a clean end-of-stream
+// between parts as the end of the form, so some cuts parse without error and
+// only the byte count catches them.
+func TestWebhookRetriesBodyCutOffBeforeContentLength(t *testing.T) {
+	full := signedMultipart(t, "key",
+		map[string]string{"recipient": "workerlist@mill.example.com"},
+		map[string][]byte{"attachment-1": []byte("pdf")})
+	raw, err := io.ReadAll(full.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for cut := 1; cut < len(raw); cut++ {
+		logger, buf := captureLogger()
+		s := &Service{engine: newFakeEngine(), signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger}
+		r := httptest.NewRequest(http.MethodPost, "/mailgun-webhook", bytes.NewReader(raw[:cut]))
+		r.Header.Set("Content-Type", full.Header.Get("Content-Type"))
+		r.ContentLength = int64(len(raw))
+		w := httptest.NewRecorder()
+		s.handle(w, r)
+		if w.Code != http.StatusInternalServerError {
+			t.Fatalf("cut at %d of %d: got %d, want 500 (a cut-off upload must be retried)", cut, len(raw), w.Code)
+		}
+		want := fmt.Sprintf("request body incomplete: got %d of %d bytes", cut, len(raw))
+		if !strings.Contains(buf.String(), want) {
+			t.Fatalf("cut at %d: log must contain %q; got %q", cut, want, buf.String())
+		}
+	}
+}
+
+// failingReader yields its data and then a read error, like a connection reset
+// partway through an upload.
+type failingReader struct {
+	data []byte
+	err  error
+}
+
+func (f *failingReader) Read(p []byte) (int, error) {
+	if len(f.data) == 0 {
+		return 0, f.err
+	}
+	n := copy(p, f.data)
+	f.data = f.data[n:]
+	return n, nil
+}
+
+func TestWebhookRetriesBodyWhoseReadFails(t *testing.T) {
+	logger, buf := captureLogger()
+	s := &Service{signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger}
+	full := signedMultipart(t, s.signKey,
+		map[string]string{"recipient": "workerlist@mill.example.com"},
+		map[string][]byte{"attachment-1": []byte("pdf")})
+	raw, err := io.ReadAll(full.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	r := httptest.NewRequest(http.MethodPost, "/mailgun-webhook", &failingReader{data: raw[:len(raw)/2], err: errors.New("connection reset by peer")})
+	r.Header.Set("Content-Type", full.Header.Get("Content-Type"))
+	r.ContentLength = -1 // chunked: no length to compare against, only the read error
+	w := httptest.NewRecorder()
+	s.handle(w, r)
+	if w.Code != http.StatusInternalServerError {
+		t.Fatalf("got %d, want 500 (a failed read must be retried)", w.Code)
+	}
+	if !strings.Contains(buf.String(), "connection reset by peer") {
+		t.Errorf("the read error must be logged; got %q", buf.String())
+	}
+}
+
+// The accept line records when the email was sent (its Date header, in UTC),
+// when Mailgun posted it (the signed timestamp), and the lag from sending to
+// arrival here - so a "my reply was slow" report can be measured from the log.
+func TestWebhookLogsArrivalTimesOnAccept(t *testing.T) {
+	logger, buf := captureLogger()
+	sent := time.Date(2026, 10, 28, 4, 6, 44, 0, time.UTC)
+	s := &Service{engine: newFakeEngine(), signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger,
+		now: func() time.Time { return sent.Add(18*time.Minute + 19*time.Second + 400*time.Millisecond) }}
+	r := signedMultipart(t, s.signKey,
+		map[string]string{"recipient": "workerlist@mill.example.com", "sender": "sender@example.com", "Date": "Tue, 27 Oct 2026 22:06:44 -0600"},
+		map[string][]byte{"attachment-1": []byte("pdf")})
+	w := httptest.NewRecorder()
+	s.handle(w, r)
+	if w.Code != http.StatusOK {
+		t.Fatalf("got %d, want 200", w.Code)
+	}
+	unix, err := strconv.ParseInt(r.FormValue("timestamp"), 10, 64)
+	if err != nil {
+		t.Fatal(err)
+	}
+	posted := time.Unix(unix, 0).UTC().Format(time.RFC3339)
+	for _, want := range []string{"sent=2026-10-28T04:06:44Z", "posted=" + posted, "lag=18m19s"} {
+		if !strings.Contains(buf.String(), want) {
+			t.Errorf("accept line must contain %q; got %q", want, buf.String())
+		}
+	}
+}
+
+func TestWebhookReadsDateFromMessageHeadersWhenFieldIsAbsent(t *testing.T) {
+	logger, buf := captureLogger()
+	s := &Service{engine: newFakeEngine(), signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger}
+	headers := `[["Subject", "Sister INI"], ["Date", "Wed, 28 Oct 2026 04:14:49 +0000"]]`
+	r := signedMultipart(t, s.signKey,
+		map[string]string{"recipient": "workerlist@mill.example.com", "sender": "sender@example.com", "message-headers": headers},
+		map[string][]byte{"attachment-1": []byte("pdf")})
+	w := httptest.NewRecorder()
+	s.handle(w, r)
+	if w.Code != http.StatusOK {
+		t.Fatalf("got %d, want 200", w.Code)
+	}
+	if !strings.Contains(buf.String(), "sent=2026-10-28T04:14:49Z") {
+		t.Errorf("Date must be read from message-headers; got %q", buf.String())
+	}
+}
+
+// A missing or garbled Date is logged as unknown and never blocks the email.
+func TestWebhookAcceptsEmailWithUnparseableDate(t *testing.T) {
+	logger, buf := captureLogger()
+	s := &Service{engine: newFakeEngine(), signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger}
+	r := signedMultipart(t, s.signKey,
+		map[string]string{"recipient": "workerlist@mill.example.com", "sender": "sender@example.com", "Date": "sometime last week"},
+		map[string][]byte{"attachment-1": []byte("pdf")})
+	w := httptest.NewRecorder()
+	s.handle(w, r)
+	if w.Code != http.StatusOK {
+		t.Fatalf("got %d, want 200", w.Code)
+	}
+	if !strings.Contains(buf.String(), "sent=unknown") || strings.Contains(buf.String(), "lag=") {
+		t.Errorf("an unparseable Date must log sent=unknown and no lag; got %q", buf.String())
 	}
 }