Preserve event identity across retries and writer shutdown.

internal/forwarder/gate.go

@@ -36,15 +36,12 @@ func (g *Gate) Acquire(eventID string) (release func(), wait <-chan struct{}) {
 	ch := make(chan struct{})
 	g.in[eventID] = ch
 	g.mu.Unlock()
-	var once sync.Once
-	return func() {
-		once.Do(func() {
-			g.mu.Lock()
-			delete(g.in, eventID)
-			g.mu.Unlock()
-			close(ch)
-		})
-	}, nil
+	return sync.OnceFunc(func() {
+		g.mu.Lock()
+		defer g.mu.Unlock()
+		delete(g.in, eventID)
+		close(ch)
+	}), nil
 }
 
 // InFlight reports how many events are currently gated (tests/metrics).

internal/server/forward_audit_test.go

@@ -95,8 +95,17 @@ func TestForwardAuditByteIdenticalAndMinimal(t *testing.T) {
 			t.Fatalf("unexpected forwarded header %q", k)
 		}
 	}
-	if out.Header.Get("X-SubnetLens-Event-ID") != id || out.Header.Get("X-SubnetLens-Signature-256") == "" {
-		t.Fatalf("signature headers not verbatim: %v", out.Header)
+	for _, name := range []string{
+		"X-SubnetLens-Event-ID", "X-SubnetLens-Event-Key", "X-SubnetLens-Event-Type",
+		"X-SubnetLens-Signature-Timestamp", "X-SubnetLens-Signature-256", "Content-Type",
+	} {
+		want := req.Header.Get(name)
+		if want == "" {
+			t.Fatalf("test request lacks %s", name)
+		}
+		if got := out.Header.Get(name); got != want {
+			t.Errorf("forwarded %s = %q, want %q", name, got, want)
+		}
 	}
 	// Nothing local is folded into the relay body: the forwarded bytes
 	// equal the producer submission, which carried no decision/rule

internal/server/server.go

@@ -5,7 +5,8 @@
 // 202 accepted (persisted; forwarded in proxy mode), 401 bad signature or
 // timestamp, 400 malformed, 413 oversize, 405 non-POST, 503 saturated or
 // store unavailable with Retry-After, 502 ScanRelay delivery failed.
-// A duplicate host event is idempotent (202 when forwarded, 502 when not).
+// A duplicate host event keeps the first stored payload. In proxy mode it
+// returns 202 once forwarded and 502 if delivery has not succeeded.
 // H1 is persisted separately and never forwarded; identical retries are 202,
 // while conflicting reuse of a heartbeat ID is 409.
 package server
@@ -327,7 +328,8 @@ func (s *Server) persistAndReply(ctx context.Context, w http.ResponseWriter, bod
 	pctx, cancel := context.WithTimeout(ctx, s.cfg.PersistBudget)
 	defer cancel()
 
-	headersJSON, err := json.Marshal(store.WhitelistForwardHeaders(headerMap(r.Header)))
+	headers := store.WhitelistForwardHeaders(headerMap(r.Header))
+	headersJSON, err := json.Marshal(headers)
 	if err != nil {
 		return errMalformedBody
 	}
@@ -374,11 +376,15 @@ func (s *Server) persistAndReply(ctx context.Context, w http.ResponseWriter, bod
 		// doing its job. In-flight duplicates never reach here:
 		// they wait on the gate and follow the stored state.
 		s.Health.AddDuplicate()
-		if fwd, _, _, ferr := s.events.ForwardState(ctx, eventID); ferr == nil && fwd {
+		if s.cfg.Mode == config.ModeTap {
 			writeAccepted(w, eventID, eventType)
 			return nil
 		}
-		if s.cfg.Mode == config.ModeTap || s.forwardOne(ctx, body, headerMap(r.Header), eventID) {
+		forwarded, samePayload, err := s.events.RetryForwardState(ctx, eventID, body)
+		if err != nil {
+			return err
+		}
+		if forwarded || (samePayload && s.forwardOne(ctx, body, headers, eventID)) {
 			writeAccepted(w, eventID, eventType)
 			return nil
 		}
@@ -391,7 +397,7 @@ func (s *Server) persistAndReply(ctx context.Context, w http.ResponseWriter, bod
 		writeAccepted(w, eventID, eventType)
 		return nil
 	}
-	if s.forwardOne(ctx, body, headerMap(r.Header), eventID) {
+	if s.forwardOne(ctx, body, headers, eventID) {
 		s.Health.AddAccepted()
 		writeAccepted(w, eventID, eventType)
 		return nil
@@ -413,13 +419,16 @@ func (s *Server) reportPersistError(w http.ResponseWriter, eventID string, err e
 	s.writeSaturated(w, "persist unavailable")
 }
 
-// replyForStoredState answers a duplicate from the forwarded flag: 202 when
-// the event landed at ScanRelay, 502 when it did not so Pro retries. It never
-// forwards inline — a second forward inside this deadline could exceed it.
+// replyForStoredState accepts a stored tap event or a forwarded proxy event.
+// A waiter never forwards inline; a new request owns the next retry.
 func (s *Server) replyForStoredState(ctx context.Context, w http.ResponseWriter, eventID, eventType string) {
 	s.Health.AddDuplicate()
 	forwarded, _, _, err := s.events.ForwardState(ctx, eventID)
-	if err != nil || !forwarded {
+	if err != nil {
+		s.writeSaturated(w, "persist unavailable")
+		return
+	}
+	if s.cfg.Mode == config.ModeProxy && !forwarded {
 		writeError(w, http.StatusBadGateway, "scanrelay delivery failed")
 		return
 	}

internal/server/server_test.go

@@ -221,16 +221,17 @@ func TestContractTable(t *testing.T) {
 	// 502: ScanRelay down.
 	h.relay.mode.Store("fail")
 	failID := uniqueID()
-	if rec := h.signedPost(t, failID, "host-b", "host.appeared", now); rec.Code != http.StatusBadGateway {
+	body := eventBody(failID, "host-b", "host.appeared")
+	sig := sign(h.secret, rts, failID, "host-b", "host.appeared", body)
+	if rec := h.post(t, body, failID, "host-b", "host.appeared", now, sig); rec.Code != http.StatusBadGateway {
 		t.Fatalf("forward failure: got %d", rec.Code)
 	}
-	// 202 duplicate of an unforwarded event: waiter path returns 502, and
-	// the retry after a successful forward returns 202.
-	if rec := h.signedPost(t, failID, "host-b", "host.appeared", now); rec.Code != http.StatusBadGateway {
+	// A retry remains 502 until the relay accepts the stored payload.
+	if rec := h.post(t, body, failID, "host-b", "host.appeared", now, sig); rec.Code != http.StatusBadGateway {
 		t.Fatalf("duplicate of unforwarded: got %d", rec.Code)
 	}
 	h.relay.mode.Store("accept")
-	if rec := h.signedPost(t, failID, "host-b", "host.appeared", now); rec.Code != http.StatusAccepted {
+	if rec := h.post(t, body, failID, "host-b", "host.appeared", now, sig); rec.Code != http.StatusAccepted {
 		t.Fatalf("retry after failure: got %d %s", rec.Code, rec.Body.String())
 	}
 	if h.srv.gate.InFlight() != 0 {
@@ -344,11 +345,63 @@ func TestWaiterReturns502WhenWinnerFails(t *testing.T) {
 	}
 	// Pro's retry then succeeds and is accepted.
 	h.relay.mode.Store("accept")
-	if rec := h.signedPost(t, id, "host-wf", "host.appeared", now); rec.Code != http.StatusAccepted {
+	if rec := h.post(t, body, id, "host-wf", "host.appeared", now, sig); rec.Code != http.StatusAccepted {
 		t.Fatalf("retry: got %d", rec.Code)
 	}
 }
 
+func TestProxyRetryCannotReplaceStoredPayload(t *testing.T) {
+	h := newHarness(t, nil)
+	now := time.Now()
+	id, key, typ := uniqueID(), "host-retry", "host.appeared"
+	body := eventBody(id, key, typ)
+	post := func(payload []byte, at time.Time) int {
+		sig := sign(h.secret, strconv.FormatInt(at.Unix(), 10), id, key, typ, payload)
+		return h.post(t, payload, id, key, typ, at, sig).Code
+	}
+	h.relay.mode.Store("fail")
+	if got := post(body, now); got != http.StatusBadGateway {
+		t.Fatalf("initial delivery status = %d", got)
+	}
+	h.relay.mode.Store("accept")
+	retryAt := now.Add(time.Second)
+	changed := bytes.Replace(body, []byte(`"cycle":1`), []byte(`"cycle":2`), 1)
+	if got := post(changed, retryAt); got != http.StatusBadGateway || h.relay.hits.Load() != 1 {
+		t.Fatalf("conflicting retry status = %d, relay hits = %d", got, h.relay.hits.Load())
+	}
+	if got := post(body, retryAt); got != http.StatusAccepted || h.relay.hits.Load() != 2 {
+		t.Fatalf("exact retry status = %d, relay hits = %d", got, h.relay.hits.Load())
+	}
+	if got := post(changed, retryAt); got != http.StatusAccepted || h.relay.hits.Load() != 2 {
+		t.Fatalf("conflict after delivery status = %d, relay hits = %d", got, h.relay.hits.Load())
+	}
+	stored, err := h.srv.events.StoredEvent(context.Background(), id)
+	h.relay.mu.Lock()
+	defer h.relay.mu.Unlock()
+	if err != nil || !bytes.Equal(stored.Payload, body) || !bytes.Equal(h.relay.bodies[1], body) {
+		t.Fatalf("stored and forwarded payloads must match the original: %v", err)
+	}
+}
+
+func TestTapWaiterFollowsStoredEvent(t *testing.T) {
+	h := newHarness(t, func(c *config.Config) { c.Mode = config.ModeTap })
+	ctx := context.Background()
+	id := uniqueID()
+	rec := httptest.NewRecorder()
+	h.srv.replyForStoredState(ctx, rec, id, "host.appeared")
+	if rec.Code != http.StatusServiceUnavailable || rec.Header().Get("Retry-After") == "" {
+		t.Fatalf("missing event status = %d, Retry-After = %q", rec.Code, rec.Header().Get("Retry-After"))
+	}
+	if got := h.signedPost(t, id, "host-tap", "host.appeared", time.Now()).Code; got != http.StatusAccepted {
+		t.Fatalf("initial delivery status = %d", got)
+	}
+	rec = httptest.NewRecorder()
+	h.srv.replyForStoredState(ctx, rec, id, "host.appeared")
+	if rec.Code != http.StatusAccepted || h.relay.hits.Load() != 0 {
+		t.Fatalf("tap waiter status = %d, relay hits = %d", rec.Code, h.relay.hits.Load())
+	}
+}
+
 func TestCancelDuringPersistThenRetry(t *testing.T) {
 	h := newHarness(t, func(c *config.Config) { c.Mode = config.ModeTap })
 	now := time.Now()

internal/server/sweep.go

@@ -84,7 +84,11 @@ func (s *Server) sweepOne(ctx context.Context, ev store.RawEvent) (done, landed
 	if s.now().Sub(ev.SignatureTimestamp) >= s.cfg.TimestampTolerance-s.cfg.SweepSafetyMargin {
 		return s.park(ctx, ev.EventID, "stale_for_sweep"), false
 	}
-	if fwd, _, _, err := s.events.ForwardState(ctx, ev.EventID); err != nil || fwd {
+	fwd, _, _, err := s.events.ForwardState(ctx, ev.EventID)
+	if err != nil {
+		return false, false
+	}
+	if fwd {
 		return true, false
 	}
 	var headers map[string]string

internal/store/forwarding.go

@@ -18,6 +18,14 @@ func (s *Store) ForwardState(ctx context.Context, eventID string) (forwarded boo
 	return fwd == 1, attempts, last.String, err
 }
 
+// RetryForwardState checks whether the event already landed or a newly
+// signed retry carries the exact bytes originally accepted for this ID.
+func (s *Store) RetryForwardState(ctx context.Context, eventID string, payload []byte) (forwarded, samePayload bool, err error) {
+	err = s.db.QueryRowContext(ctx, `SELECT forwarded, payload = ? FROM raw_events WHERE event_id = ?`, payload, eventID).
+		Scan(&forwarded, &samePayload)
+	return forwarded, samePayload, err
+}
+
 // MarkForwarded records a successful delivery once; repeats are no-ops.
 func (s *Store) MarkForwarded(ctx context.Context, eventID string, at time.Time) (bool, error) { // returns false on any error (captured var is only read on success)
 	matched := false

internal/store/retention.go

@@ -49,7 +49,7 @@ func (s *Store) Checkpoint(ctx context.Context) error {
 		}
 		return nil
 	}
-	return s.enqueue(ctx, job)
+	return s.enqueue(job)
 }
 
 // MarkUndeliverable parks an event the forwarder can never send (e.g. its