fix(relay): raise inbound MaxLineLength from 2000 to 128KiB, add rejected_line_too_long metric

docs/observability.md

@@ -74,7 +74,7 @@ SES outage must not knock every instance out of rotation).
 
 | Metric | Type | Labels | Meaning |
 |---|---|---|---|
-| `e2a_smtp_inbound_total` | counter | `outcome` | SMTP intake decisions. Units differ by stage: `accepted` (250), `accepted_dedup` (250 on a lost-ack retry), and `tempfail` (451 — durable persist/enqueue failed) are one per DATA transaction; `rejected_unknown_recipient` / `rejected_unverified_domain` (550) and `rejected_quota` (552) are one per rejected RCPT command — a single transaction can emit several rejections and still accept for its remaining recipients. A DATA phase aborted mid-read (client dropped, size limit) records no outcome. |
+| `e2a_smtp_inbound_total` | counter | `outcome` | SMTP intake decisions. Units differ by stage: `accepted` (250), `accepted_dedup` (250 on a lost-ack retry), and `tempfail` (451 — durable persist/enqueue failed) are one per DATA transaction; `rejected_unknown_recipient` / `rejected_unverified_domain` (550) and `rejected_quota` (552) are one per rejected RCPT command — a single transaction can emit several rejections and still accept for its remaining recipients; `rejected_line_too_long` (554 — a line over the relay's `MaxLineLength`) is one per DATA transaction aborted mid-read. Other mid-read DATA aborts (client dropped, size limit) record no outcome. |
 | `e2a_smtp_inbound_duration_seconds` | histogram | — | DATA-phase processing time (accepted/tempfail outcomes only; RCPT rejections have no DATA phase). Includes an exact `2` second SLO bucket. |
 
 Policy rejections (550/552) are *correct* behavior, not failures — the

internal/relay/inbound_longline_test.go

@@ -0,0 +1,111 @@
+package relay_test
+
+import (
+	"context"
+	"net/smtp"
+	"strings"
+	"testing"
+
+	"github.com/tokencanopy/e2a/internal/config"
+	"github.com/tokencanopy/e2a/internal/identity"
+	"github.com/tokencanopy/e2a/internal/relay"
+	"github.com/tokencanopy/e2a/internal/testutil"
+	"github.com/tokencanopy/e2a/internal/usage"
+	"github.com/tokencanopy/e2a/internal/ws"
+)
+
+// Regression test for the go-smtp MaxLineLength default (2000 bytes).
+// Agent-generated mail — single-line JSON, unfolded HTML — routinely
+// exceeds that, and the relay used to reject it at DATA with "554 ...
+// too long a line in input stream". This bit a real customer whose two
+// agents messaged each other (prod, 2026-07-18..20, 30 bounces). The
+// relay now allows lines up to 128KiB; the whole message stays capped
+// by MaxMessageBytes.
+//
+// Same integration shape as inbound_limit_test.go: needs a real DB and
+// an actual SMTP socket bind, so skipped under -short.
+func TestRelay_Data_AcceptsLongLines(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping SMTP integration test under -short")
+	}
+	pool := testutil.TestDB(t)
+	ctx := context.Background()
+	store := identity.NewStore(pool)
+
+	user, err := store.CreateOrGetUser(ctx, "relay-longline@test.com", "Test", "google-relay-longline@test.com")
+	if err != nil {
+		t.Fatalf("CreateOrGetUser: %v", err)
+	}
+	if _, err := store.ClaimOrCreateDomain(ctx, "relay-longline.example.com", user.ID); err != nil {
+		t.Fatalf("ClaimOrCreateDomain: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE domains SET verified = true WHERE domain = $1`, "relay-longline.example.com"); err != nil {
+		t.Fatalf("verify domain: %v", err)
+	}
+	agentEmail := "bot@relay-longline.example.com"
+	if _, err := store.CreateAgent(ctx, agentEmail, "relay-longline.example.com", "", "https://example.com/w", "cloud", user.ID); err != nil {
+		t.Fatalf("CreateAgent: %v", err)
+	}
+
+	port, err := freePort()
+	if err != nil {
+		t.Fatalf("freePort: %v", err)
+	}
+	cfg := &config.Config{
+		SMTP: config.SMTPConfig{ListenAddr: "127.0.0.1:" + port, Domain: "test.relay"},
+		Env:  "development",
+	}
+	server := relay.NewServer(cfg, store, usage.NewNoopUsageTracker(), ws.NewHub())
+	go func() { _ = server.ListenAndServe() }()
+	t.Cleanup(func() { _ = server.Close() })
+	waitForSMTP(t, cfg.SMTP.ListenAddr)
+
+	// A single ~64KiB body line — the shape real agent payloads take
+	// (one JSON object, one unfolded HTML document). Well past the old
+	// 2000-byte default, well under the new 128KiB cap.
+	longLine := strings.Repeat("x", 64*1024)
+	body := "From: sender@elsewhere.test\r\nTo: " + agentEmail + "\r\n" +
+		"Subject: long line\r\nContent-Type: application/json\r\n\r\n" + longLine
+
+	c, err := smtp.Dial(cfg.SMTP.ListenAddr)
+	if err != nil {
+		t.Fatalf("smtp.Dial: %v", err)
+	}
+	defer c.Close()
+	if err := c.Mail("sender@elsewhere.test"); err != nil {
+		t.Fatalf("MAIL FROM: %v", err)
+	}
+	if err := c.Rcpt(agentEmail); err != nil {
+		t.Fatalf("RCPT TO: %v", err)
+	}
+	w, err := c.Data()
+	if err != nil {
+		t.Fatalf("DATA: %v", err)
+	}
+	if _, err := w.Write([]byte(body)); err != nil {
+		t.Fatalf("write body: %v", err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close DATA with a 64KiB line: %v (long-line rejection regressed?)", err)
+	}
+	_ = c.Quit()
+
+	// The message must actually land in the agent's inbox, not merely
+	// be accepted at the SMTP layer.
+	if !waitFor(func() bool {
+		msgs, err := store.GetMessagesByAgent(ctx, identity.MessageListFilter{
+			AgentID: agentEmail, Direction: "inbound", Status: "all", Limit: 10,
+		})
+		if err != nil {
+			return false
+		}
+		for _, m := range msgs {
+			if m.Subject == "long line" {
+				return true
+			}
+		}
+		return false
+	}) {
+		t.Error("long-line message never appeared in the agent inbox")
+	}
+}

internal/relay/metrics_test.go

@@ -412,6 +412,76 @@ func TestSMTPInboundMetric_AsyncAcceptedAndDedup(t *testing.T) {
 	}
 }
 
+// TestSMTPInboundMetric_RejectedLineTooLong drives a DATA payload with a single
+// line over the relay's 128KiB MaxLineLength and asserts the 554 abort records
+// exactly one "rejected_line_too_long" observation — the original long-line
+// incident surfaced as a customer complaint precisely because this path
+// recorded nothing.
+func TestSMTPInboundMetric_RejectedLineTooLong(t *testing.T) {
+	pool := testutil.TestDB(t)
+	store := identity.NewStore(pool)
+	ctx := context.Background()
+
+	const domain = "metrics-longline.example.com"
+	const agentEmail = "bot@" + domain
+	user, err := store.CreateOrGetUser(ctx, "owner@"+domain, "O", "g-metrics-longline")
+	if err != nil {
+		t.Fatalf("user: %v", err)
+	}
+	if _, err := store.ClaimOrCreateDomain(ctx, domain, user.ID); err != nil {
+		t.Fatalf("domain: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE domains SET verified=true WHERE domain=$1`, domain); err != nil {
+		t.Fatalf("verify domain: %v", err)
+	}
+	if _, err := store.CreateAgent(ctx, agentEmail, domain, "", "", "", user.ID); err != nil {
+		t.Fatalf("agent: %v", err)
+	}
+
+	port, err := freePort()
+	if err != nil {
+		t.Fatalf("freePort: %v", err)
+	}
+	cfg := &config.Config{SMTP: config.SMTPConfig{ListenAddr: "127.0.0.1:" + port, Domain: domain}, Env: "development"}
+	server := relay.NewServer(cfg, store, usage.NewNoopUsageTracker(), ws.NewHub())
+	metrics := &fakeSMTPMetrics{}
+	server.SetMetrics(metrics)
+	addr := startMetricsRelay(t, cfg, server)
+
+	c, err := smtp.Dial(addr)
+	if err != nil {
+		t.Fatalf("smtp.Dial: %v", err)
+	}
+	defer c.Close()
+	if err := c.Mail("sender@ext.test"); err != nil {
+		t.Fatalf("MAIL FROM: %v", err)
+	}
+	if err := c.Rcpt(agentEmail); err != nil {
+		t.Fatalf("RCPT TO: %v", err)
+	}
+	w, err := c.Data()
+	if err != nil {
+		t.Fatalf("DATA: %v", err)
+	}
+	// A single 160KiB line — over the 128KiB cap, small enough that the
+	// unread remainder after the server aborts fits in loopback socket
+	// buffers (the server reads ~128KiB before erroring).
+	body := "From: sender@ext.test\r\nTo: " + agentEmail + "\r\nSubject: too long\r\n\r\n" +
+		strings.Repeat("y", 160*1024)
+	if _, err := w.Write([]byte(body)); err != nil {
+		t.Fatalf("write body: %v", err)
+	}
+	cerr := w.Close()
+	if cerr == nil {
+		t.Fatal("DATA with a 160KiB line succeeded; want 554 too-long-line rejection")
+	}
+	if !strings.Contains(cerr.Error(), "554") || !strings.Contains(cerr.Error(), "too long a line") {
+		t.Fatalf("DATA close error = %v, want 554 too-long-line", cerr)
+	}
+
+	assertSingleOutcome(t, metrics, "rejected_line_too_long", false)
+}
+
 // TestSMTPInboundMetric_NilMetricsSafe proves the default (no SetMetrics call)
 // stays nil-safe on both the RCPT rejection path and the DATA accept path.
 func TestSMTPInboundMetric_NilMetricsSafe(t *testing.T) {

internal/relay/server.go

@@ -79,8 +79,9 @@ type Server struct {
 type Metrics interface {
 	// SMTPInbound records the terminal outcome of one SMTP intake decision.
 	// outcome ∈ {accepted, accepted_dedup, tempfail, rejected_unknown_recipient,
-	// rejected_unverified_domain, rejected_quota}; seconds is DATA processing
-	// time (0 for RCPT-stage rejections, which have no DATA phase).
+	// rejected_unverified_domain, rejected_quota, rejected_line_too_long};
+	// seconds is DATA processing time (0 for RCPT-stage rejections, which
+	// have no DATA phase).
 	SMTPInbound(outcome string, seconds float64)
 }
 
@@ -133,8 +134,9 @@ func (s *Server) SetMetrics(m Metrics) { s.metrics = m }
 
 // recordSMTPInbound is the nil-safe recording seam every intake outcome goes
 // through. Units: exactly one accepted/accepted_dedup/tempfail call per DATA
-// transaction (never per recipient); rejected_* calls are per rejected RCPT
-// command — one transaction can emit several rejections and still accept.
+// transaction (never per recipient); rejected_line_too_long is one per DATA
+// transaction aborted mid-read; the other rejected_* calls are per rejected
+// RCPT command — one transaction can emit several rejections and still accept.
 func (s *Server) recordSMTPInbound(outcome string, seconds float64) {
 	if s.metrics != nil {
 		s.metrics.SMTPInbound(outcome, seconds)
@@ -167,6 +169,19 @@ func NewServer(cfg *config.Config, store *identity.Store, usage usage.UsageTrack
 	smtpSrv.WriteTimeout = 30 * time.Second
 	smtpSrv.MaxMessageBytes = 10 * 1024 * 1024 // 10MB
 	smtpSrv.MaxRecipients = 50
+	// go-smtp defaults MaxLineLength to 2000 bytes, which bounced real
+	// agent mail carrying single-line JSON / unfolded HTML at DATA with
+	// "too long a line in input stream" (prod, 2026-07-18..20: 30 bounces
+	// over three days). 128KiB covers those payloads with wide margin but
+	// is kept deliberately modest rather than raised toward
+	// MaxMessageBytes: the limit bounds ALL lines, including pre-auth SMTP
+	// command lines, so it is per-connection memory an unauthenticated
+	// peer can force. It must also stay at or below the 1MiB header-scan
+	// buffer in internal/emailauth/checker.go, or the DKIM l= guard there
+	// fails open. Since #745 the composer quoted-printable-encodes any
+	// outbound body with a >998-octet line, so this cap guards inbound
+	// mail from third-party MTAs.
+	smtpSrv.MaxLineLength = 1 << 17 // 128KiB
 	smtpSrv.AllowInsecureAuth = !cfg.IsProduction()
 
 	if cfg.SMTP.TLSCert != "" && cfg.SMTP.TLSKey != "" {
@@ -322,6 +337,15 @@ func (s *session) Data(r io.Reader) error {
 	start := time.Now()
 	body, err := io.ReadAll(r)
 	if err != nil {
+		// A line over MaxLineLength surfaces here as go-smtp's ErrTooLongLine
+		// and bounces the transaction with a 554. Record it — the incident
+		// behind the MaxLineLength comment in NewServer was invisible in
+		// metrics precisely because this path returned early. Other read
+		// errors (client drop → io.ErrUnexpectedEOF, MaxMessageBytes →
+		// smtp.ErrDataTooLarge) are deliberately left unrecorded, as before.
+		if errors.Is(err, smtp.ErrTooLongLine) {
+			s.relay.recordSMTPInbound("rejected_line_too_long", time.Since(start).Seconds())
+		}
 		return err
 	}
 

internal/telemetry/metrics.go

@@ -89,11 +89,13 @@ type Metrics interface {
 
 	// SMTPInbound records one SMTP intake decision. outcome ∈
 	// {accepted, accepted_dedup, tempfail, rejected_unknown_recipient,
-	// rejected_unverified_domain, rejected_quota}. Units differ by
-	// stage: accepted/accepted_dedup/tempfail are per DATA transaction;
-	// rejected_* are per rejected RCPT command (one transaction can
-	// emit several rejections and still accept). seconds is DATA
-	// processing time (0 for RCPT-stage rejections).
+	// rejected_unverified_domain, rejected_quota, rejected_line_too_long}.
+	// Units differ by stage: accepted/accepted_dedup/tempfail are per DATA
+	// transaction; rejected_line_too_long is per DATA transaction aborted
+	// mid-read (line over MaxLineLength); the other rejected_* are per
+	// rejected RCPT command (one transaction can emit several rejections
+	// and still accept). seconds is DATA processing time (0 for RCPT-stage
+	// rejections).
 	SMTPInbound(outcome string, seconds float64)
 
 	// ThreadHeaderParseFailure counts an inbound RFC threading header that

internal/telemetry/prom.go

@@ -86,7 +86,8 @@ var (
 	methodSet = set("GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS")
 	classSet  = set("1xx", "2xx", "3xx", "4xx", "5xx", "none")
 	smtpSet   = set("accepted", "accepted_dedup", "tempfail",
-		"rejected_unknown_recipient", "rejected_unverified_domain", "rejected_quota")
+		"rejected_unknown_recipient", "rejected_unverified_domain", "rejected_quota",
+		"rejected_line_too_long")
 	outTermSet = set("sent", "failed_suppressed", "failed_provider",
 		"failed_local_retries", "failed_cancelled")
 	outAttemptSet = set("success", "temporary_failure", "permanent_failure")
@@ -422,7 +423,10 @@ func (p *Prom) HTTPRequest(method, route, statusClass string, seconds float64) {
 func (p *Prom) SMTPInbound(outcome string, seconds float64) {
 	o := enum(smtpSet, outcome)
 	p.smtpInbound.WithLabelValues(o).Inc()
-	// RCPT-stage rejections carry no DATA duration; only observe real ones.
+	// Only fully-processed DATA transactions feed the duration histogram
+	// (it backs the 2s latency SLO): RCPT-stage rejections have no DATA
+	// phase, and rejected_line_too_long aborts mid-read, so its elapsed
+	// time is not a processing latency.
 	if o == "accepted" || o == "accepted_dedup" || o == "tempfail" {
 		p.smtpDuration.Observe(seconds)
 	}