fix(server): stop retrying permanent event sink rejections

cmd/harness/eventsink.go

@@ -97,13 +97,21 @@ func (h *httpEventSink) Deliver(ctx context.Context, batch server.EventBatch) (i
 	defer resp.Body.Close()
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
 		body, readErr := io.ReadAll(io.LimitReader(resp.Body, eventSinkReplyMaxBytes+1))
+		err := fmt.Errorf("event sink: receiver returned %d", resp.StatusCode)
 		if readErr != nil {
-			return 0, fmt.Errorf("event sink: receiver returned %d; read diagnostic: %w", resp.StatusCode, readErr)
+			err = fmt.Errorf("event sink: receiver returned %d; read diagnostic: %w", resp.StatusCode, readErr)
+		} else if code := eventSinkDiagnosticCode(body); code != "" {
+			err = fmt.Errorf("event sink: receiver returned %d (%s)", resp.StatusCode, code)
 		}
-		if code := eventSinkDiagnosticCode(body); code != "" {
-			return 0, fmt.Errorf("event sink: receiver returned %d (%s)", resp.StatusCode, code)
+		// The status classifies the failure, not the diagnostic: a receiver
+		// that answers a permanent status with no body is still permanent.
+		// The sentinel is the only wrapped operand: the receiver text keeps
+		// its place at the front of the message, and the pump matches on
+		// one sentinel rather than on a tree of wrapped causes.
+		if eventSinkPermanentStatus(resp.StatusCode) {
+			return 0, fmt.Errorf("%v: %w", err, server.ErrEventSinkPermanent)
 		}
-		return 0, fmt.Errorf("event sink: receiver returned %d", resp.StatusCode)
+		return 0, err
 	}
 	var reply sinkReply
 	// A reply that does not parse is an error, not a zero cursor: treating
@@ -115,6 +123,30 @@ func (h *httpEventSink) Deliver(ctx context.Context, batch server.EventBatch) (i
 	return reply.AppliedThrough, nil
 }
 
+// eventSinkPermanentStatus reports whether this status rejects the batch
+// itself, so that retrying identical bytes cannot succeed. The set is fixed
+// and small: a malformed body (400, 422), a refused credential (401, 403), a
+// route that holds no receiver (404, 410), and a receiver that says the batch
+// contradicts what it already applied (409).
+//
+// Every other status keeps the retry, including an unlisted 4xx. 408, 425,
+// and 429 ask for the same batch later, and a 5xx is a receiver that a
+// restart can fix, so treating either as permanent would cost every later
+// record for one transient failure.
+func eventSinkPermanentStatus(status int) bool {
+	switch status {
+	case http.StatusBadRequest,
+		http.StatusUnauthorized,
+		http.StatusForbidden,
+		http.StatusNotFound,
+		http.StatusConflict,
+		http.StatusGone,
+		http.StatusUnprocessableEntity:
+		return true
+	}
+	return false
+}
+
 // eventSinkDiagnosticCode extracts only a bounded machine code from an error
 // response. Arbitrary receiver text can contain secrets and reaches logs through
 // the pump's delivery error, so it must not be copied into the error.

cmd/harness/eventsink_test.go

@@ -3,11 +3,13 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"io"
 	"maps"
 	"net/http"
 	"net/http/httptest"
 	"slices"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -154,3 +156,118 @@ func TestHTTPEventSinkEncodesAnEmptyFilteredCheckpoint(t *testing.T) {
 		t.Errorf("appliedThrough = %d, want 12", applied)
 	}
 }
+
+// The receiver's status is the whole classifier. A 400, 401, 403, 404, 409,
+// 410, or 422 rejects this batch and every identical retry of it, so the pump
+// must stop; 408, 425, 429, and 5xx ask for the same batch later, and every
+// other status keeps the existing retry. Wrong output: a permanent status
+// that stays retryable and spins the two-second loop forever, or a retryable
+// status classified permanent, which retires the pump on a receiver restart.
+func TestHTTPEventSinkClassifiesPermanentReceiverRejections(t *testing.T) {
+	cases := []struct {
+		status    int
+		permanent bool
+	}{
+		{http.StatusBadRequest, true},
+		{http.StatusUnauthorized, true},
+		{http.StatusForbidden, true},
+		{http.StatusNotFound, true},
+		{http.StatusConflict, true},
+		{http.StatusGone, true},
+		{http.StatusUnprocessableEntity, true},
+		{http.StatusRequestTimeout, false},
+		{http.StatusTooEarly, false},
+		{http.StatusTooManyRequests, false},
+		{http.StatusInternalServerError, false},
+		{http.StatusBadGateway, false},
+		{http.StatusServiceUnavailable, false},
+		{http.StatusGatewayTimeout, false},
+		// An unlisted 4xx is not permanent. The set is fixed, not "every 4xx".
+		{http.StatusPaymentRequired, false},
+		{http.StatusTeapot, false},
+	}
+	for _, tc := range cases {
+		t.Run(http.StatusText(tc.status), func(t *testing.T) {
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				w.WriteHeader(tc.status)
+			}))
+			t.Cleanup(ts.Close)
+
+			sink := newHTTPEventSink(&config.EventSinkSpec{URL: ts.URL})
+			_, err := sink.Deliver(context.Background(), server.EventBatch{FromSeq: 1, ToSeq: 1})
+			if err == nil {
+				t.Fatalf("Deliver succeeded on %d, want an error so the cursor does not advance", tc.status)
+			}
+			if got := errors.Is(err, server.ErrEventSinkPermanent); got != tc.permanent {
+				t.Errorf("errors.Is(err, ErrEventSinkPermanent) = %t for %d, want %t; err = %v", got, tc.status, tc.permanent, err)
+			}
+			// The message is the operator's whole record of the failure, so
+			// it is pinned exactly: the status, and for a permanent one the
+			// sentinel appended after it, with nothing else added.
+			want := "event sink: receiver returned " + strconv.Itoa(tc.status)
+			if tc.permanent {
+				want += ": " + server.ErrEventSinkPermanent.Error()
+			}
+			if err.Error() != want {
+				t.Errorf("error = %q, want %q", err, want)
+			}
+		})
+	}
+}
+
+// A permanent rejection is the last thing an operator sees from the pump, so
+// it must still carry the bounded diagnostic the retryable path carries.
+// Wrong output: an error that drops the receiver's machine code, or one that
+// copies the receiver's free text or the configured URL's secrets into a log.
+func TestHTTPEventSinkPermanentRejectionKeepsABoundedDiagnostic(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+		_, _ = w.Write([]byte(`{"code":"generation_rejected","message":"secret diagnostic detail"}`))
+	}))
+	t.Cleanup(ts.Close)
+
+	sink := newHTTPEventSink(&config.EventSinkSpec{URL: ts.URL + "/sink?token=secret_query#secret_fragment"})
+	_, err := sink.Deliver(context.Background(), server.EventBatch{FromSeq: 1, ToSeq: 1})
+	if !errors.Is(err, server.ErrEventSinkPermanent) {
+		t.Fatalf("error %v is not permanent, want a 403 to retire the pump", err)
+	}
+	// Exact text: the receiver's own diagnostic keeps its place at the front
+	// and the sentinel is appended once. A second wrap verb, or a sentinel
+	// that swallowed the diagnostic, changes this string.
+	const want = "event sink: receiver returned 403 (generation_rejected): permanent receiver rejection"
+	if err.Error() != want {
+		t.Errorf("error = %q, want %q", err, want)
+	}
+	// One wrapped operand, and it is the sentinel. A second %w verb builds a
+	// multi-error whose Unwrap answers nil here, which hides the single
+	// cause the pump is written against.
+	if unwrapped := errors.Unwrap(err); unwrapped != server.ErrEventSinkPermanent {
+		t.Errorf("errors.Unwrap(err) = %v, want the sentinel itself", unwrapped)
+	}
+	for _, leak := range []string{"secret diagnostic detail", "secret_query", "secret_fragment"} {
+		if strings.Contains(err.Error(), leak) {
+			t.Errorf("error %q leaks %q", err, leak)
+		}
+	}
+}
+
+// A dial failure has no status to classify, and the receiver may well be
+// mid-restart. Wrong output: a transport failure that retires the pump, which
+// would make one refused connection cost every later record.
+func TestHTTPEventSinkTransportFailureIsNotPermanent(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
+	url := ts.URL + "/sink?token=secret_query"
+	ts.Close() // nothing listens on that port now
+
+	sink := newHTTPEventSink(&config.EventSinkSpec{URL: url})
+	_, err := sink.Deliver(context.Background(), server.EventBatch{FromSeq: 1, ToSeq: 1})
+	if err == nil {
+		t.Fatal("Deliver succeeded against a closed receiver")
+	}
+	if errors.Is(err, server.ErrEventSinkPermanent) {
+		t.Errorf("transport failure classified permanent: %v", err)
+	}
+	if strings.Contains(err.Error(), "secret_query") {
+		t.Errorf("error includes configured URL secrets: %q", err)
+	}
+}

docs/design/event-sink.md

@@ -67,24 +67,26 @@ this change; nothing about the current implementation does it, and nothing
 about the current design doc should be read as promising it will keep
 working unmodified if eviction is added later.
 
-## 5. Failure does not give up
-
-`flushEventSink`'s delivery loop, on a `Deliver` error, logs a warning and
-retries the SAME batch after `eventSinkRetryDelay` (2 seconds) — it does
-not advance past the failure, drop the batch, or wait for a new record to
-arrive before trying again. It keeps retrying, at that fixed interval,
-until `Deliver` succeeds or the pump is retired (`sinkStop`, closed after
-the prompt drain — see §7), at which point the goroutine exits without
+## 5. A retryable failure does not give up
+
+`flushEventSink`'s delivery loop, on a retryable `Deliver` error, logs a
+warning and retries the SAME batch after `eventSinkRetryDelay` (2 seconds) —
+it does not advance past the failure, drop the batch, or wait for a new
+record to arrive before trying again. It keeps retrying, at that fixed
+interval, until `Deliver` succeeds or the pump is retired (`sinkStop`, closed
+after the prompt drain — see §7), at which point the goroutine exits without
 another attempt.
 
-The consequence: a receiver that is down, or answering errors, does not
-lose any records. It delays them. A permanently unreachable receiver
-leaves the pump retrying every `eventSinkRetryDelay` for the rest of the
-process's life — this is intended, not a bug to fix later, because the
-journal (§4) is already the buffer holding everything the pump has not
-yet managed to deliver. There is nothing else for the pump to spool to,
-and nothing is lost by continuing to retry against something the journal
-already holds.
+The consequence: a receiver that is down, or answering a retryable error,
+does not lose any records. It delays them. An unreachable receiver leaves
+the pump retrying every `eventSinkRetryDelay` for the rest of the process's
+life — this is intended, not a bug to fix later, because the journal (§4) is
+already the buffer holding everything the pump has not yet managed to
+deliver. There is nothing else for the pump to spool to, and nothing is lost
+by continuing to retry against something the journal already holds.
+
+Section 14 gives the one class of failure this does not cover: a receiver
+that rejects the batch itself.
 
 ## 6. The receiver owns the cursor
 
@@ -299,3 +301,53 @@ off the wire for such a record, so a receiver can tell "undated" from
 `TestDurableEventStampsRecordedAtFromTheInjectedClock`,
 `TestLiveEventCarriesNoRecordedAt`, and
 `TestLegacyEventKeepsAZeroRecordedAtOnReload` pin these three rules.
+
+## 14. A permanent rejection retires the pump
+
+A retry is a bet that the same bytes can succeed later (§5). Some receiver
+answers say they cannot. A receiver that rejects the batch itself — a body
+it cannot parse, a credential it refuses, a route that holds no receiver —
+answers the identical rejection to the identical retry, every two seconds,
+for the life of the process. That loop delivers nothing, and it logs a
+warning on every pass, which buries every other line an operator reads.
+
+`server.ErrEventSinkPermanent` (`server/eventsink.go`) is the sentinel for
+that class. A transport wraps it; `flushEventSink` detects it with
+`errors.Is`, logs one bounded warning, and returns false, which retires
+`runEventSink`. The pump goroutine exits and `sinkDone` closes.
+
+**Harness does not stop.** The sentinel retires the replica, nothing else:
+sessions run, records still journal and still reach `/event`, and `Drain`
+still returns (it waits on a `sinkDone` that is already closed). The
+deployment loses forwarding, not the box.
+`TestEventSinkPermanentRejectionStopsThePumpWithoutStoppingHarness` pins the
+stop, the single warning, and the still-healthy server.
+`TestEventSinkRetryableFailureIsNotPermanent` pins the two-second retry that
+a retryable error still gets.
+
+`httpEventSink` classifies by status alone (`eventSinkPermanentStatus`,
+`cmd/harness/eventsink.go`):
+
+| Status | Class | Why |
+|---|---|---|
+| 400, 422 | permanent | The receiver cannot parse or accept this body. |
+| 401, 403 | permanent | The credential is refused, not throttled. |
+| 404, 410 | permanent | The URL names no receiver. |
+| 409 | permanent | The batch contradicts what the receiver applied. |
+| 408, 425, 429 | retryable | The receiver asks for the same batch later. |
+| 5xx | retryable | A receiver a restart or a failover fixes. |
+| any other status | retryable | The set is fixed, not "every 4xx". |
+| transport failure | retryable | A dial or a timeout carries no verdict. |
+
+The status is the whole classifier. A permanent status with no body is still
+permanent, and a retryable status that carries a diagnostic is still
+retryable. The diagnostic itself is unchanged: `eventSinkDiagnosticCode`
+still extracts only the bounded `code` field, so the pump logs the status
+and that machine code, never the receiver's free text and never the
+configured URL. `TestHTTPEventSinkClassifiesPermanentReceiverRejections`,
+`TestHTTPEventSinkPermanentRejectionKeepsABoundedDiagnostic`, and
+`TestHTTPEventSinkTransportFailureIsNotPermanent` pin the table above.
+
+A permanent rejection is a configuration report, not a data loss. The
+journal keeps every record, so fixing the receiver and restarting harness
+resumes forwarding from whatever cursor the receiver answers next (§6).

server/eventsink.go

@@ -3,10 +3,19 @@ package server
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"sort"
 	"time"
 )
 
+// ErrEventSinkPermanent marks a delivery failure that retrying cannot fix.
+// A transport wraps it when the receiver rejects the BATCH — a malformed
+// body, a refused credential, a route that holds no receiver — rather than
+// asking for the same batch later. The pump stops on it, because resending
+// identical bytes every eventSinkRetryDelay only earns the same rejection
+// for the life of the process.
+var ErrEventSinkPermanent = errors.New("permanent receiver rejection")
+
 // EventSink is the outbound transport for the durable journal. Deliver
 // returns the seq the receiver has applied through, which becomes the
 // pump's cursor: the RECEIVER owns that cursor, so harness keeps no durable
@@ -49,6 +58,10 @@ const (
 	eventSinkRetryDelay        = 2 * time.Second
 )
 
+// eventSinkStoppedMsg is the one warning a permanent rejection logs. The
+// pump exits after it, so an operator reads it once, not once per retry.
+const eventSinkStoppedMsg = "event sink stopped: receiver rejected the batch"
+
 // stopEventSink cancels an active delivery, then asks the pump to make one
 // final catch-up pass under finalCtx. Idempotent and safe without a pump.
 func (s *Server) stopEventSink(finalCtx context.Context) {
@@ -98,7 +111,9 @@ func (s *Server) runEventSink() {
 	// pump for records this process did not itself emit. Without this first
 	// flush, a process that restarts and then goes idle replicates nothing
 	// until some unrelated record happens to arrive.
-	s.flushEventSink(s.sinkCtx)
+	if !s.flushEventSink(s.sinkCtx) {
+		return
+	}
 	for {
 		select {
 		case <-s.sinkStop:
@@ -115,7 +130,9 @@ func (s *Server) runEventSink() {
 			return
 		case <-t.C:
 		}
-		s.flushEventSink(s.sinkCtx)
+		if !s.flushEventSink(s.sinkCtx) {
+			return
+		}
 	}
 }
 
@@ -124,36 +141,47 @@ func (s *Server) runEventSink() {
 // no records remain, ctx is canceled, or a failed delivery observes sinkStop.
 // A successful final pass can drain the backlog after sinkStop closes. It never
 // holds s.mu across Deliver.
-func (s *Server) flushEventSink(ctx context.Context) {
+//
+// It reports whether the pump may keep running. Only ErrEventSinkPermanent
+// answers false: that batch cannot succeed on a retry, and neither can any
+// later batch built the same way, so the caller retires the pump. Harness
+// itself is unaffected — the journal, the sessions, and every other client
+// surface keep working without a replica.
+func (s *Server) flushEventSink(ctx context.Context) bool {
 	for {
 		select {
 		case <-ctx.Done():
-			return
+			return true
 		default:
 		}
 		batch, ok := s.nextEventBatch()
 		if !ok {
-			return
+			return true
 		}
 		applied, err := s.opts.EventSink.Deliver(ctx, batch)
 		if err != nil {
+			if errors.Is(err, ErrEventSinkPermanent) {
+				// The transport already bounded and sanitized this text.
+				s.logWarn(eventSinkStoppedMsg, "from_seq", batch.FromSeq, "to_seq", batch.ToSeq, "error", err.Error())
+				return false
+			}
 			s.logWarn("event sink delivery failed", "from_seq", batch.FromSeq, "to_seq", batch.ToSeq, "error", err.Error())
 			t := time.NewTimer(eventSinkRetryDelay)
 			select {
 			case <-ctx.Done():
 				t.Stop()
-				return
+				return true
 			case <-s.sinkStop:
 				t.Stop()
-				return
+				return true
 			case <-t.C:
 			}
 			continue
 		}
 		if !s.advanceSinkCursor(applied) {
 			// The receiver did not move past this batch's start, so sending
 			// it again immediately would spin. Wait for the next wake.
-			return
+			return true
 		}
 	}
 }