From 0e3f2e849484546fed09e26073d811e383b67cc7 Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Sat, 26 Sep 2026 06:57:58 -0400
Subject: [PATCH 1/2] feat(fabric): let a subscriber callback fail so the event
 is redelivered (RIG-4030)

The EventFabric callback now returns an error. nil acks; an error or a
panic Naks through retryOrPark and parks on DLQSubject at MaxDeliver.

Delivery returns an error for a transient read before any hold or
dispatch, and nil for a missing row or a deliberate skip, so a retry
starts clean. Reads inside fanOut still log and ack, since a retry would
repeat mention steers. hold keeps held entries in commit order, so a
redelivered ref cannot fire behind a later one.

Also proves the cutover record T4 (b)/(c): error redelivery to MaxDeliver
and the DLQ park with Compass-Original-Subject.

Built on the recommended option; draft until that decision is ruled.

Spec-impact: pending RIG-4030 (the callback signature in the cutover
record T2 Interfaces line). Refs RIG-4030, RIG-3107

Co-authored-by: Matt Wilkinson <matt@rigel.build>
---
 .../comms/fabric_publish_pgtest_test.go       |   4 +-
 go/internal/delivery/consumer.go              |  32 ++-
 go/internal/delivery/consumer_test.go         | 153 +++++++++-
 go/internal/delivery/dispatch.go              |  59 ++--
 go/internal/delivery/fake_fabric_test.go      |  70 ++++-
 go/internal/delivery/helpers_test.go          |  29 ++
 go/internal/fabric/SUBJECTS.md                |   7 +-
 go/internal/fabric/doc.go                     |   4 +-
 go/internal/fabric/event_fabric.go            |  33 ++-
 go/internal/fabric/event_fabric_test.go       | 269 +++++++++++++++---
 go/internal/fabric/fabric.go                  |   8 +-
 go/internal/fabric/fabric_test.go             |   4 +-
 .../delivery_two_instance_pgtest_test.go      |   6 +-
 13 files changed, 556 insertions(+), 122 deletions(-)

diff --git a/go/internal/comms/fabric_publish_pgtest_test.go b/go/internal/comms/fabric_publish_pgtest_test.go
index 2e7033ca4..593d4af80 100644
--- a/go/internal/comms/fabric_publish_pgtest_test.go
+++ b/go/internal/comms/fabric_publish_pgtest_test.go
@@ -61,11 +61,11 @@ func (f *fakeEventFabric) Publish(ctx context.Context, subject string, ref fabri
 	return f.err
 }
 
-func (f *fakeEventFabric) Subscribe(context.Context, string, func(context.Context, fabric.EventRef)) (fabric.Unsubscribe, error) {
+func (f *fakeEventFabric) Subscribe(context.Context, string, func(context.Context, fabric.EventRef) error) (fabric.Unsubscribe, error) {
 	return nil, errors.New("fakeEventFabric: Subscribe not used by comms")
 }
 
-func (f *fakeEventFabric) SubscribeKind(context.Context, fabric.EventKind, func(context.Context, fabric.EventRef)) (fabric.Unsubscribe, error) {
+func (f *fakeEventFabric) SubscribeKind(context.Context, fabric.EventKind, func(context.Context, fabric.EventRef) error) (fabric.Unsubscribe, error) {
 	return nil, errors.New("fakeEventFabric: SubscribeKind not used by comms")
 }
 
diff --git a/go/internal/delivery/consumer.go b/go/internal/delivery/consumer.go
index 4243994b7..18acce3ad 100644
--- a/go/internal/delivery/consumer.go
+++ b/go/internal/delivery/consumer.go
@@ -8,6 +8,7 @@ package delivery
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"regexp"
@@ -302,9 +303,9 @@ func (c *Consumer) SetAgentWaker(w AgentWaker) {
 }
 
 // Run consumes message_posted refs and drains settle, start, and recovery work
-// until ctx is cancelled. A publish failed after commit, or a ref acked after a
-// failed read, reaches live recipients via the recovery pass (reconnect or floor
-// tick) and offline ones at their next session start.
+// until ctx is cancelled. A publish failed after commit, or a ref parked after
+// its reads kept failing, reaches live recipients via the recovery pass
+// (reconnect or floor tick) and offline ones at their next session start.
 func (c *Consumer) Run(ctx context.Context) error {
 	// Sweeps and scans enumerate every tenant, so they run as the BYPASSRLS system
 	// role; per-event work stays tenant-scoped under ctx.
@@ -357,16 +358,29 @@ func (c *Consumer) Run(ctx context.Context) error {
 }
 
 // onEventRef handles one ref on the fabric goroutine, concurrently with Run's
-// drains. Any read failure is logged and acked (see Run for recovery). A post
-// whose author settled before the hold landed is delivered at once (hold).
-func (c *Consumer) onEventRef(ctx context.Context, ref fabric.EventRef) {
+// drains. A transient read error is returned before any hold or dispatch, so the
+// fabric redelivers; a missing row or a deliberate skip returns nil and acks.
+// A post whose author settled before the hold landed is delivered at once (hold).
+func (c *Consumer) onEventRef(ctx context.Context, ref fabric.EventRef) error {
 	ctx = store.WithTenant(ctx, store.TenantID(ref.Tenant))
 	m, err := c.st.MessageByID(ctx, ref.RowID)
 	if err != nil {
-		c.log.ErrorContext(ctx, "delivery: re-read posted message", "error", err, "message_id", ref.RowID, "tenant", ref.Tenant)
-		return
+		return c.readFailure(ctx, "re-read posted message", ref.RowID, err)
+	}
+	return c.onMessagePosted(ctx, comms.MessageToWire(m))
+}
+
+// readFailure logs a failed pre-dispatch read. A missing row returns nil (acked),
+// since redelivery cannot make it appear; any other error is returned to redeliver.
+// The tenant comes from ctx, which onEventRef scoped to the ref's tenant.
+func (c *Consumer) readFailure(ctx context.Context, what, messageID string, err error) error {
+	tenant, _ := store.TenantFromContext(ctx)
+	if errors.Is(err, store.ErrNotFound) {
+		c.log.ErrorContext(ctx, "delivery: "+what+": row missing, skipping", "error", err, "message_id", messageID, "tenant", string(tenant))
+		return nil
 	}
-	c.onMessagePosted(ctx, comms.MessageToWire(m))
+	c.log.WarnContext(ctx, "delivery: "+what+": will redeliver", "error", err, "message_id", messageID, "tenant", string(tenant))
+	return fmt.Errorf("delivery: %s %s: %w", what, messageID, err)
 }
 
 // requestRecovery marks a recovery pass owed and wakes the loop. It runs on the
diff --git a/go/internal/delivery/consumer_test.go b/go/internal/delivery/consumer_test.go
index 9fd1c049c..f60b067d2 100644
--- a/go/internal/delivery/consumer_test.go
+++ b/go/internal/delivery/consumer_test.go
@@ -8,12 +8,14 @@ package delivery
 
 import (
 	"context"
+	"errors"
 	"slices"
 	"sync"
 	"testing"
 	"time"
 
 	compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1"
+	"github.com/RigelBuild/compass/go/internal/fabric"
 	compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1"
 	"github.com/RigelBuild/compass/go/internal/store"
 )
@@ -427,8 +429,8 @@ func TestRecoverySweepSkipsHeldMessage(t *testing.T) {
 	}
 }
 
-// OQ-1: a ref whose row is missing is logged and acked, since redelivery cannot
-// make the row appear; the refs behind it still deliver.
+// OQ-1: a ref whose row is missing is logged and acked (the callback returns
+// nil), since redelivery cannot make the row appear; the refs behind it still deliver.
 func TestMissingRowRefIsAckedAndConsumerContinues(t *testing.T) {
 	c, disp, res, reads := newTestConsumer(t)
 	const ch store.ChannelID = "chan-1"
@@ -452,6 +454,153 @@ func TestMissingRowRefIsAckedAndConsumerContinues(t *testing.T) {
 	}
 }
 
+// errTransientRead stands in for a store read that fails but may succeed on retry.
+var errTransientRead = errors.New("connection reset")
+
+// A transient read failure returns an error so the fabric redelivers, and no
+// hold or dispatch happens first. ErrNotFound returns nil so the ref is acked.
+func TestOnEventRefReadErrorContract(t *testing.T) {
+	const ch store.ChannelID = "chan-1"
+	const recipient store.AccountID = "agent-recip"
+	for _, tc := range []struct {
+		name        string
+		author      store.AccountID
+		authorAgent bool
+		authorLive  bool
+		msgErrs     []error // MessageByID outcomes; nil reads normally
+		authorErrs  []error // IsAgentAccount outcomes
+		channelErrs []error // MessageChannel outcomes
+		wantErr     error   // nil: the ref is acked
+	}{
+		{name: "transient first read, live agent author", author: "agent-author", authorAgent: true, authorLive: true, msgErrs: []error{errTransientRead}, wantErr: errTransientRead},
+		{name: "transient first read, human author", author: "human-1", msgErrs: []error{errTransientRead}, wantErr: errTransientRead},
+		{name: "transient stored re-read, offline agent author", author: "agent-author", authorAgent: true, msgErrs: []error{nil, errTransientRead}, wantErr: errTransientRead},
+		{name: "transient author-kind read", author: "agent-author", authorAgent: true, authorLive: true, authorErrs: []error{errTransientRead}, wantErr: errTransientRead},
+		{name: "transient channel read, human author", author: "human-1", channelErrs: []error{errTransientRead}, wantErr: errTransientRead},
+		{name: "not found", author: "human-1", msgErrs: []error{store.ErrNotFound}},
+		{name: "channel not found, human author", author: "human-1", channelErrs: []error{store.ErrNotFound}},
+		{name: "stored re-read not found, offline agent author", author: "agent-author", authorAgent: true, msgErrs: []error{nil, store.ErrNotFound}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			c, disp, res, reads := newTestConsumer(t)
+			reads.subscribers[ch] = []store.AccountID{recipient}
+			reads.agents[tc.author] = tc.authorAgent
+			res.bind(recipient, "sess-recip")
+			if tc.authorLive {
+				res.bind(tc.author, "sess-author")
+			}
+			reads.seedMessage(textMessage("m1", tc.author, "body"))
+			reads.messageErrs["m1"] = tc.msgErrs
+			reads.authorErrs[tc.author] = tc.authorErrs
+			reads.channelErrs["m1"] = tc.channelErrs
+
+			ref := fabric.EventRef{Tenant: string(testTenant), Kind: fabric.KindMessagePosted, RowID: "m1"}
+			err := c.onEventRef(context.Background(), ref)
+			if tc.wantErr != nil {
+				if !errors.Is(err, tc.wantErr) {
+					t.Fatalf("onEventRef = %v, want an error wrapping %v so the fabric redelivers", err, tc.wantErr)
+				}
+			} else if err != nil {
+				t.Fatalf("onEventRef = %v, want nil: a missing row is acked", err)
+			}
+			if got := disp.snapshot(); len(got) != 0 {
+				t.Fatalf("dispatched %+v on a failed read, want nothing", got)
+			}
+			if c.isHeld("sess-author", "m1") {
+				t.Fatal("held m1 on a failed read: a redelivery would hold it twice")
+			}
+		})
+	}
+}
+
+// Through the fabric, a transient read failure is redelivered, and the retry
+// dispatches exactly once, including for a held agent-authored post.
+func TestTransientReadFailureRedeliversAndDispatchesOnce(t *testing.T) {
+	for _, authorAgent := range []bool{false, true} {
+		name := "human author"
+		if authorAgent {
+			name = "held agent author"
+		}
+		t.Run(name, func(t *testing.T) {
+			c, disp, res, reads := newTestConsumer(t)
+			const ch store.ChannelID = "chan-1"
+			const author store.AccountID = "author-1"
+			const recipient store.AccountID = "agent-recip"
+			reads.subscribers[ch] = []store.AccountID{recipient}
+			reads.agents[author] = authorAgent
+			res.bind(recipient, "sess-recip")
+			if authorAgent {
+				res.bind(author, "sess-author")
+			}
+			reads.messageErrs["m1"] = []error{errTransientRead}
+			startConsumer(t, c)
+
+			postMessage(t, c, reads, textMessage("m1", author, "hello"))
+			fab := fakeFabricOf(c)
+			fab.waitAcked(t, "m1")
+			if got := fab.failuresFor("m1"); len(got) != 1 || !errors.Is(got[0], errTransientRead) {
+				t.Fatalf("recorded failures = %v, want exactly one wrapping %v", got, errTransientRead)
+			}
+			if authorAgent {
+				c.waitHeld(t, "sess-author", 1)
+				c.OnSessionSettled("sess-author", compassv1.AgentSessionState_AGENT_SESSION_STATE_READY)
+			}
+			if !disp.waitForMessage(t, "m1") {
+				t.Fatal("m1 never delivered after its redelivered ref")
+			}
+			// A marker behind m1 proves the loop drained; a duplicate would be ahead of it.
+			postMessage(t, c, reads, textMessage("m2", "human-2", "marker"))
+			if !disp.waitForMessage(t, "m2") {
+				t.Fatal("marker m2 never delivered")
+			}
+			if n := countDispatches(disp, "m1"); n != 1 {
+				t.Fatalf("m1 dispatched %d times, want exactly 1", n)
+			}
+		})
+	}
+}
+
+func countDispatches(d *fakeDispatcher, messageID string) int {
+	n := 0
+	for _, r := range d.snapshot() {
+		if r.messageID == messageID {
+			n++
+		}
+	}
+	return n
+}
+
+// A Nak'd ref can arrive after a later one of the same author, so hold orders by
+// commit time: m2 held before m1 still fires as [m1, m2].
+func TestHoldOrdersByCommitTimeNotArrival(t *testing.T) {
+	c, disp, res, reads := newTestConsumer(t)
+	const author store.AccountID = "agent-author"
+	reads.subscribers["chan-1"] = []store.AccountID{"agent-recip"}
+	reads.agents[author] = true
+	res.bind(author, "sess-author")
+	res.bind("agent-recip", "sess-recip")
+	for _, m := range []struct {
+		id string
+		at int64
+	}{{"m2", 200}, {"m1", 100}} {
+		msg := textMessage(m.id, author, m.id+" body")
+		msg.At = time.UnixMilli(m.at)
+		reads.seedMessage(msg)
+		c.hold(store.WithTenant(context.Background(), testTenant), "sess-author", m.id, m.at)
+	}
+	startConsumer(t, c)
+	c.OnSessionSettled("sess-author", compassv1.AgentSessionState_AGENT_SESSION_STATE_READY)
+	disp.waitForDispatches(t, 2)
+	snap := disp.snapshot()
+	got := make([]string, 0, len(snap))
+	for _, d := range snap {
+		got = append(got, d.messageID)
+	}
+	if !slices.Equal(got, []string{"m1", "m2"}) {
+		t.Fatalf("dispatch order = %v, want [m1 m2] (commit order, not arrival)", got)
+	}
+}
+
 // OQ-3 part 1: the per-event re-read runs under the ref's tenant, never the
 // system role, and the held deliver re-reads under the tenant captured at hold.
 func TestEventReadsRunUnderRefTenant(t *testing.T) {
diff --git a/go/internal/delivery/dispatch.go b/go/internal/delivery/dispatch.go
index 481eb05fa..0b26566b6 100644
--- a/go/internal/delivery/dispatch.go
+++ b/go/internal/delivery/dispatch.go
@@ -5,6 +5,7 @@ package delivery
 import (
 	"context"
 	"errors"
+	"slices"
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
@@ -34,19 +35,21 @@ import (
 // re-resolve at fire time (the settle path re-resolves recipients against the
 // then-current subscription + liveness), so a subscription change between post
 // and settle is honored.
-func (c *Consumer) onMessagePosted(ctx context.Context, msg *compassv1.Message) {
+//
+// Every read whose failure is returned runs before the first hold or dispatch;
+// reads inside fanOut are logged and acked, since a retry would repeat its steers.
+func (c *Consumer) onMessagePosted(ctx context.Context, msg *compassv1.Message) error {
 	if msg == nil {
-		return
+		return nil
 	}
 	author := store.AccountID(msg.GetAuthorAccountId())
 	messageID := msg.GetId()
 	if msg.GetTopicId() == "" || messageID == "" {
-		return
+		return nil
 	}
 	authorIsAgent, err := c.st.IsAgentAccount(ctx, author)
 	if err != nil {
-		c.log.ErrorContext(ctx, "delivery: resolve author kind", "error", err, "message_id", messageID)
-		return
+		return c.readFailure(ctx, "resolve author kind", messageID, err)
 	}
 
 	if !authorIsAgent {
@@ -55,11 +58,10 @@ func (c *Consumer) onMessagePosted(ctx context.Context, msg *compassv1.Message)
 		// through topics.channel_id (the frozen record's topic->channel resolution).
 		channel, err := c.st.MessageChannel(ctx, messageID)
 		if err != nil {
-			c.log.ErrorContext(ctx, "delivery: resolve message channel", "error", err, "message_id", messageID)
-			return
+			return c.readFailure(ctx, "resolve message channel", messageID, err)
 		}
 		c.fanOut(ctx, channel, author, msg)
-		return
+		return nil
 	}
 
 	// Agent-authored. If the author has a live session, HOLD until it settles;
@@ -68,30 +70,23 @@ func (c *Consumer) onMessagePosted(ctx context.Context, msg *compassv1.Message)
 	// (design.md:177-178, :306).
 	authorSession, live := c.resolver.SessionForAccount(ctx, author)
 	if !live {
-		c.fanOutStored(ctx, messageID)
-		return
+		wire, channel, storedAuthor, err := c.storeMessageToWire(ctx, messageID)
+		if err != nil {
+			return c.readFailure(ctx, "re-read message for stored-block deliver", messageID, err)
+		}
+		c.fanOut(ctx, channel, storedAuthor, wire)
+		return nil
 	}
 	c.hold(ctx, authorSession, messageID, msg.GetAtUnixMs())
-}
-
-// fanOutStored delivers a message now from its stored blocks: the author has no
-// live turn.
-func (c *Consumer) fanOutStored(ctx context.Context, messageID string) {
-	wire, channel, author, err := c.storeMessageToWire(ctx, messageID)
-	if err != nil {
-		// The message vanished between post and deliver (unexpected): skip it;
-		// the cursor never advanced, so the sweep still redelivers.
-		c.log.ErrorContext(ctx, "delivery: re-read message for stored-block deliver", "error", err, "message_id", messageID)
-		return
-	}
-	c.fanOut(ctx, channel, author, wire)
+	return nil
 }
 
 // hold registers messageID under its author's session for later firing at the
-// author's settle edge (design.md:157-160), in post order. It captures the origin
-// trace and tenant from ctx for fireHeld. If the author already settled at or
-// after atUnixMs, it also queues a settle edge, so the loop fires it at once and
-// still behind any earlier message of that author.
+// author's settle edge (design.md:157-160), ordered by commit time (stable), so a
+// ref redelivered after a later one still fires in post order. It captures the
+// origin trace and tenant from ctx for fireHeld. If the author already settled at
+// or after atUnixMs, it also queues a settle edge, so the loop fires it at once
+// and still behind any earlier message of that author.
 //
 // The two clocks come from different instances. A settling clock behind the
 // committing one holds a message until the next settle, which is benign. A
@@ -103,7 +98,15 @@ func (c *Consumer) hold(ctx context.Context, authorSession, messageID string, at
 		entry.tenant = tenant
 	}
 	c.mu.Lock()
-	c.held[authorSession] = append(c.held[authorSession], entry)
+	entries := c.held[authorSession]
+	// First index strictly after atUnixMs: equal stamps keep arrival order.
+	i, _ := slices.BinarySearchFunc(entries, atUnixMs, func(e heldEntry, at int64) int {
+		if e.atUnixMs <= at {
+			return -1
+		}
+		return 1
+	})
+	c.held[authorSession] = slices.Insert(entries, i, entry)
 	settled, ok := c.lastSettle[authorSession]
 	fireNow := ok && settled >= atUnixMs
 	if fireNow {
diff --git a/go/internal/delivery/fake_fabric_test.go b/go/internal/delivery/fake_fabric_test.go
index e36cdc11b..877ad5cb7 100644
--- a/go/internal/delivery/fake_fabric_test.go
+++ b/go/internal/delivery/fake_fabric_test.go
@@ -22,7 +22,8 @@ const testTenant store.TenantID = "tenant-1"
 // fakeFabric is an in-memory fabric.EventFabric. Publish queues a ref the way
 // the stream retains one, so a ref published before SubscribeKind is delivered
 // once the consumer subscribes. One goroutine runs the callback serially, as the
-// fabric promises, and fireReconnect runs the OnReconnect hooks.
+// fabric promises, and fireReconnect runs the OnReconnect hooks. A callback
+// error is recorded and redelivered until maxDeliver attempts, then dropped.
 type fakeFabric struct {
 	queue      chan fakeEvent
 	subscribed chan struct{}
@@ -30,27 +31,32 @@ type fakeFabric struct {
 	// beforeSubscribe, when set, runs at the top of SubscribeKind so a test can
 	// observe what Run finished before it subscribed.
 	beforeSubscribe func()
+	maxDeliver      int
 
 	mu       sync.Mutex
 	hooks    map[int]func()
 	nextHook int
 	acked    []string
 	ackSig   chan struct{}
+	failures map[string][]error
 }
 
 // fakeEvent is one queued publish: the ref plus the publisher's traceparent,
-// which the real fabric carries in a message header.
+// which the real fabric carries in a message header, and the attempts so far.
 type fakeEvent struct {
 	ref         fabric.EventRef
 	traceparent string
+	attempts    int
 }
 
 func newFakeFabric() *fakeFabric {
 	return &fakeFabric{
 		queue:      make(chan fakeEvent, 1024),
 		subscribed: make(chan struct{}),
+		maxDeliver: fabric.DefaultMaxDeliver,
 		hooks:      map[int]func(){},
 		ackSig:     make(chan struct{}, 1024),
+		failures:   map[string][]error{},
 	}
 }
 
@@ -69,11 +75,11 @@ func (f *fakeFabric) Publish(ctx context.Context, _ string, ref fabric.EventRef)
 }
 
 // Subscribe is the concrete-subject read side, which the consumer must not use.
-func (f *fakeFabric) Subscribe(context.Context, string, func(context.Context, fabric.EventRef)) (fabric.Unsubscribe, error) {
+func (f *fakeFabric) Subscribe(context.Context, string, func(context.Context, fabric.EventRef) error) (fabric.Unsubscribe, error) {
 	return nil, errors.New("fake fabric: the delivery consumer subscribes by kind only")
 }
 
-func (f *fakeFabric) SubscribeKind(ctx context.Context, kind fabric.EventKind, fn func(context.Context, fabric.EventRef)) (fabric.Unsubscribe, error) {
+func (f *fakeFabric) SubscribeKind(ctx context.Context, kind fabric.EventKind, fn func(context.Context, fabric.EventRef) error) (fabric.Unsubscribe, error) {
 	if kind != fabric.KindMessagePosted {
 		return nil, fmt.Errorf("fake fabric: unexpected kind %q", kind)
 	}
@@ -88,17 +94,39 @@ func (f *fakeFabric) SubscribeKind(ctx context.Context, kind fabric.EventKind, f
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
+		// Nak'd events go here and run before the next queued one, like an
+		// immediate redelivery; the goroutine owns the slice, so it needs no lock.
+		var redeliver []fakeEvent
 		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-stop:
-				return
-			case ev := <-f.queue:
-				// The real fabric extracts the header's span onto the subscription ctx.
-				fn(otelx.ContextWithTraceparent(ctx, ev.traceparent), ev.ref)
-				f.ack(ev.ref.RowID)
+			var ev fakeEvent
+			if len(redeliver) > 0 {
+				// A retry still honours teardown, as the real consumer's drain does.
+				select {
+				case <-ctx.Done():
+					return
+				case <-stop:
+					return
+				default:
+				}
+				ev, redeliver = redeliver[0], redeliver[1:]
+			} else {
+				select {
+				case <-ctx.Done():
+					return
+				case <-stop:
+					return
+				case ev = <-f.queue:
+				}
 			}
+			// The real fabric extracts the header's span onto the subscription ctx.
+			ev.attempts++
+			if err := fn(otelx.ContextWithTraceparent(ctx, ev.traceparent), ev.ref); err != nil {
+				if f.fail(ev, err) {
+					redeliver = append(redeliver, ev)
+				}
+				continue
+			}
+			f.ack(ev.ref.RowID)
 		}
 	}()
 	f.subOnce.Do(func() { close(f.subscribed) })
@@ -150,6 +178,22 @@ func (f *fakeFabric) ack(rowID string) {
 	signalObserved(f.ackSig)
 }
 
+// fail records a callback error and reports whether to redeliver (a Nak): true
+// until maxDeliver attempts are spent, after which the event is dropped (parked).
+func (f *fakeFabric) fail(ev fakeEvent, err error) bool {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.failures[ev.ref.RowID] = append(f.failures[ev.ref.RowID], err)
+	return ev.attempts < f.maxDeliver
+}
+
+// failuresFor returns the callback errors recorded for rowID, in order.
+func (f *fakeFabric) failuresFor(rowID string) []error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return slices.Clone(f.failures[rowID])
+}
+
 // waitAcked blocks until the callback for rowID has returned, or fails at the
 // deadline.
 func (f *fakeFabric) waitAcked(t *testing.T, rowID string) {
diff --git a/go/internal/delivery/helpers_test.go b/go/internal/delivery/helpers_test.go
index fc321a7f2..fe7182ea7 100644
--- a/go/internal/delivery/helpers_test.go
+++ b/go/internal/delivery/helpers_test.go
@@ -306,6 +306,13 @@ type fakeReads struct {
 	// MessageByID before f.mu is acquired, so a test can act at the point a
 	// re-read happens (e.g. a concurrent hold between the scan's checks).
 	beforeMessageByID func(messageID string)
+	// messageErrs, when set for an id, makes the next MessageByID calls for it
+	// fail with these errors in order, one per call; a nil entry reads normally.
+	// authorErrs (per account, IsAgentAccount) and channelErrs (per message id,
+	// MessageChannel) queue outcomes the same way.
+	messageErrs map[string][]error
+	authorErrs  map[store.AccountID][]error
+	channelErrs map[string][]error
 	// reads records the scope of every MessageByID call, so a test can assert a
 	// re-read ran under the message's tenant and not the system role.
 	reads []readScope
@@ -358,6 +365,9 @@ func newFakeReads() *fakeReads {
 		handles:       map[string]store.Account{},
 		accounts:      map[store.AccountID]store.Account{},
 		messages:      map[string]store.Message{},
+		messageErrs:   map[string][]error{},
+		authorErrs:    map[store.AccountID][]error{},
+		channelErrs:   map[string][]error{},
 		topicNames:    map[string]struct{ channelName, topicName string }{},
 		owed:          map[store.AccountID]map[store.ChannelID][]store.Message{},
 		sweepChannels: map[store.AccountID][]store.ChannelID{},
@@ -580,9 +590,22 @@ func (f *fakeReads) GetAccount(_ context.Context, id store.AccountID) (store.Acc
 func (f *fakeReads) IsAgentAccount(_ context.Context, account store.AccountID) (bool, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	if err := popErr(f.authorErrs, account); err != nil {
+		return false, err
+	}
 	return f.agents[account], nil
 }
 
+// popErr takes the next queued outcome for key; nil means read normally.
+func popErr[K comparable](queues map[K][]error, key K) error {
+	errs := queues[key]
+	if len(errs) == 0 {
+		return nil
+	}
+	queues[key] = errs[1:]
+	return errs[0]
+}
+
 func (f *fakeReads) MessageByID(ctx context.Context, messageID string) (store.Message, error) {
 	if f.beforeMessageByID != nil {
 		f.beforeMessageByID(messageID)
@@ -591,6 +614,9 @@ func (f *fakeReads) MessageByID(ctx context.Context, messageID string) (store.Me
 	defer f.mu.Unlock()
 	tenant, _ := store.TenantFromContext(ctx)
 	f.reads = append(f.reads, readScope{messageID: messageID, tenant: tenant, systemRole: store.IsSystemRole(ctx)})
+	if err := popErr(f.messageErrs, messageID); err != nil {
+		return store.Message{}, err
+	}
 	m, ok := f.messages[messageID]
 	if !ok {
 		return store.Message{}, store.ErrNotFound
@@ -608,6 +634,9 @@ func (f *fakeReads) MessageByID(ctx context.Context, messageID string) (store.Me
 func (f *fakeReads) MessageChannel(_ context.Context, messageID string) (store.ChannelID, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	if err := popErr(f.channelErrs, messageID); err != nil {
+		return "", err
+	}
 	for _, row := range f.unrouted {
 		if string(row.ID) == messageID {
 			return row.Channel, nil
diff --git a/go/internal/fabric/SUBJECTS.md b/go/internal/fabric/SUBJECTS.md
index 9696870de..4a916a3f5 100644
--- a/go/internal/fabric/SUBJECTS.md
+++ b/go/internal/fabric/SUBJECTS.md
@@ -200,9 +200,10 @@ Delivery semantics per message:
 
 1. Decode the `EventRef`. Undecodable → **park immediately** (no number of
    redeliveries changes the bytes).
-2. Run the subscriber callback under a panic guard. A panic becomes a failure —
-   it neither takes the process down nor acks an event nobody handled.
-3. Success → `Ack()`. A *failed ack* after successful handling is logged, never
+2. Run the subscriber callback under a panic guard. A returned error or a panic
+   is a failure; a panic neither takes the process down nor acks an event
+   nobody handled.
+3. `nil` → `Ack()`. A *failed ack* after successful handling is logged, never
    parked: it costs one redelivery, which the subscriber's Postgres re-read makes
    idempotent.
 4. Failure → read `Metadata().NumDelivered`, which counts **attempts**. Below
diff --git a/go/internal/fabric/doc.go b/go/internal/fabric/doc.go
index 170148b3d..913efd1a3 100644
--- a/go/internal/fabric/doc.go
+++ b/go/internal/fabric/doc.go
@@ -42,8 +42,8 @@
 // Every error surfaces wrapped; nothing is swallowed and nothing panics. A
 // subject built from an invalid token is refused rather than silently corrupted
 // (see [ValidSubjectToken]), an undecodable [EventRef] is parked on the DLQ
-// rather than dropped, and a subscriber callback that panics is caught, retried
-// up to Config.MaxDeliver times, then parked.
+// rather than dropped, and a subscriber callback that returns an error or
+// panics is retried up to Config.MaxDeliver times, then parked.
 //
 // Those last two are JetStream properties. On [RoutingFabric] there is nowhere
 // to park — core NATS has no ack, so an undecodable payload and a panicking
diff --git a/go/internal/fabric/event_fabric.go b/go/internal/fabric/event_fabric.go
index 471aba6ad..b46512534 100644
--- a/go/internal/fabric/event_fabric.go
+++ b/go/internal/fabric/event_fabric.go
@@ -72,14 +72,14 @@ func (f *Fabric) Publish(ctx context.Context, subject string, ref EventRef) erro
 // consumer is shared, every instance subscribing to a subject must run the same
 // fabric Config — see Config.MaxDeliver.
 //
-// Acking is explicit and follows fn: fn returning normally acks, and fn
-// panicking is recovered and treated as a failure (a panic in one subscriber
-// must not take down the process — and must not silently ack an unprocessed
-// event either). A failure Naks for immediate redelivery until NumDelivered
+// Acking is explicit and follows fn: fn returning nil acks, and fn returning an
+// error or panicking is a failure. A panic is recovered, so one subscriber can
+// neither take down the process nor silently ack an unprocessed event. A
+// failure Naks for immediate redelivery until NumDelivered
 // reaches MaxDeliver — total ATTEMPTS, not retries — at which point the message
 // is parked on DLQSubject and Term'd. An undecodable payload is parked
 // immediately: redelivering it can never succeed.
-func (f *Fabric) Subscribe(ctx context.Context, subject string, fn func(context.Context, EventRef)) (Unsubscribe, error) {
+func (f *Fabric) Subscribe(ctx context.Context, subject string, fn func(context.Context, EventRef) error) (Unsubscribe, error) {
 	if err := f.checkOpen(); err != nil {
 		return nil, err
 	}
@@ -110,7 +110,7 @@ func (f *Fabric) Subscribe(ctx context.Context, subject string, fn func(context.
 // a SubscribeKind(KindMessagePosted) receives message_posted for every tenant
 // and nothing else. Subscribe keeps its strict concrete-subject grammar — a
 // wildcard subject cannot be reached through it.
-func (f *Fabric) SubscribeKind(ctx context.Context, kind EventKind, fn func(context.Context, EventRef)) (Unsubscribe, error) {
+func (f *Fabric) SubscribeKind(ctx context.Context, kind EventKind, fn func(context.Context, EventRef) error) (Unsubscribe, error) {
 	if err := f.checkOpen(); err != nil {
 		return nil, err
 	}
@@ -132,7 +132,7 @@ func (f *Fabric) SubscribeKind(ctx context.Context, kind EventKind, fn func(cont
 //
 // It performs no validation of its own: subject must come from
 // validCommsSubject or CommsWildcardSubject.
-func (f *Fabric) subscribeSubject(ctx context.Context, subject string, fn func(context.Context, EventRef)) (Unsubscribe, error) {
+func (f *Fabric) subscribeSubject(ctx context.Context, subject string, fn func(context.Context, EventRef) error) (Unsubscribe, error) {
 	stream, err := f.ensureStream(ctx)
 	if err != nil {
 		return nil, err
@@ -191,9 +191,9 @@ func (f *Fabric) subscribeSubject(ctx context.Context, subject string, fn func(c
 }
 
 // handleEvent runs one delivery: decode, invoke fn under a panic guard, then ack
-// or park. Split out of Subscribe so the ack/park decision is readable on its
-// own.
-func (f *Fabric) handleEvent(ctx context.Context, msg jetstream.Msg, fn func(context.Context, EventRef)) {
+// on nil or retry/park on an error. Split out of Subscribe so the ack/park
+// decision is readable on its own.
+func (f *Fabric) handleEvent(ctx context.Context, msg jetstream.Msg, fn func(context.Context, EventRef) error) {
 	ref, decodeErr := decodeEventRef(msg.Data())
 	if decodeErr != nil {
 		// Unparseable: no number of redeliveries changes the bytes.
@@ -232,18 +232,17 @@ func (f *Fabric) handleEvent(ctx context.Context, msg jetstream.Msg, fn func(con
 	}
 }
 
-// invoke calls fn, converting a panic into an error. A subscriber callback is
-// consumer code running on the fabric's goroutine: letting it panic would take
-// the process down, and recovering without failing the message would ack an
-// event nobody processed.
-func invoke(ctx context.Context, fn func(context.Context, EventRef), ref EventRef) (err error) {
+// invoke calls fn and returns its error, converting a panic into an error. A
+// subscriber callback is consumer code running on the fabric's goroutine:
+// letting it panic would take the process down, and recovering without failing
+// the message would ack an event nobody processed.
+func invoke(ctx context.Context, fn func(context.Context, EventRef) error, ref EventRef) (err error) {
 	defer func() {
 		if r := recover(); r != nil {
 			err = fmt.Errorf("fabric: subscriber panicked handling %s/%s: %v", ref.Kind, ref.RowID, r)
 		}
 	}()
-	fn(ctx, ref)
-	return nil
+	return fn(ctx, ref)
 }
 
 // traceparent reads the trace header in any case: a server may have rewritten
diff --git a/go/internal/fabric/event_fabric_test.go b/go/internal/fabric/event_fabric_test.go
index d72141350..b2c6e2ae3 100644
--- a/go/internal/fabric/event_fabric_test.go
+++ b/go/internal/fabric/event_fabric_test.go
@@ -32,7 +32,10 @@ func TestEventFabricRoundTrip(t *testing.T) {
 		t.Fatalf("CommsSubject: %v", err)
 	}
 	got := make(chan EventRef, 1)
-	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) { got <- r })
+	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) error {
+		got <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
@@ -61,7 +64,10 @@ func TestPublishPropagatesTraceContext(t *testing.T) {
 	subCtx, subSpan := tracer.Start(ctx, "subscriber")
 	defer subSpan.End()
 	got := make(chan context.Context, 2)
-	unsub, err := f.Subscribe(subCtx, subject, func(ctx context.Context, _ EventRef) { got <- ctx })
+	unsub, err := f.Subscribe(subCtx, subject, func(ctx context.Context, _ EventRef) error {
+		got <- ctx
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
@@ -117,7 +123,10 @@ func TestSubscribeReadsTraceparentAnyCase(t *testing.T) {
 				t.Fatalf("CommsSubject: %v", err)
 			}
 			got := make(chan context.Context, 1)
-			unsub, err := f.Subscribe(ctx, subject, func(ctx context.Context, _ EventRef) { got <- ctx })
+			unsub, err := f.Subscribe(ctx, subject, func(ctx context.Context, _ EventRef) error {
+				got <- ctx
+				return nil
+			})
 			if err != nil {
 				t.Fatalf("Subscribe: %v", err)
 			}
@@ -157,7 +166,7 @@ func TestCallbackContextSurvivesSubscribeCancel(t *testing.T) {
 		release = make(chan struct{})
 		errs    = make(chan error, 2)
 	)
-	unsub, err := f.Subscribe(subCtx, subject, func(cbCtx context.Context, _ EventRef) {
+	unsub, err := f.Subscribe(subCtx, subject, func(cbCtx context.Context, _ EventRef) error {
 		if calls.Add(1) == 1 {
 			close(firstIn)
 			select {
@@ -166,6 +175,7 @@ func TestCallbackContextSurvivesSubscribeCancel(t *testing.T) {
 			}
 		}
 		errs <- cbCtx.Err()
+		return nil
 	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
@@ -233,13 +243,14 @@ func TestSubscribeCallbacksDoNotOverlap(t *testing.T) {
 	var active, maximum atomic.Int64
 	started := make(chan struct{}, 2)
 	release := make(chan struct{})
-	unsub, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) {
+	unsub, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) error {
 		current := active.Add(1)
 		for old := maximum.Load(); current > old && !maximum.CompareAndSwap(old, current); old = maximum.Load() {
 		}
 		started <- struct{}{}
 		<-release
 		active.Add(-1)
+		return nil
 	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
@@ -290,7 +301,10 @@ func TestEventFabricDedupsIdenticalPublishes(t *testing.T) {
 		t.Fatalf("CommsSubject: %v", err)
 	}
 	got := make(chan EventRef, 4)
-	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) { got <- r })
+	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) error {
+		got <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
@@ -334,7 +348,10 @@ func TestEventFabricFiltersBySubject(t *testing.T) {
 	}
 
 	got := make(chan EventRef, 4)
-	unsub, err := f.Subscribe(ctx, mine, func(_ context.Context, r EventRef) { got <- r })
+	unsub, err := f.Subscribe(ctx, mine, func(_ context.Context, r EventRef) error {
+		got <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
@@ -374,12 +391,18 @@ func TestEventFabricConcreteAndWildcardConsumersCoexist(t *testing.T) {
 	}
 	concrete := make(chan EventRef, 4)
 	wildcard := make(chan EventRef, 4)
-	unsubConcrete, err := f.Subscribe(ctx, concreteSubject, func(_ context.Context, r EventRef) { concrete <- r })
+	unsubConcrete, err := f.Subscribe(ctx, concreteSubject, func(_ context.Context, r EventRef) error {
+		concrete <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
 	defer unsubConcrete()
-	unsubWildcard, err := f.SubscribeKind(ctx, KindMessagePosted, func(_ context.Context, r EventRef) { wildcard <- r })
+	unsubWildcard, err := f.SubscribeKind(ctx, KindMessagePosted, func(_ context.Context, r EventRef) error {
+		wildcard <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("SubscribeKind: %v", err)
 	}
@@ -429,12 +452,13 @@ func TestUnsubscribeStopsDelivery(t *testing.T) {
 
 	var stale atomic.Int64
 	first := make(chan EventRef, 1)
-	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) {
+	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) error {
 		stale.Add(1)
 		select {
 		case first <- r:
 		default:
 		}
+		return nil
 	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
@@ -458,7 +482,10 @@ func TestUnsubscribeStopsDelivery(t *testing.T) {
 	// A second subscriber on the same subject picks up where the consumer left
 	// off; its delivery is the gate.
 	second := make(chan EventRef, 1)
-	unsub2, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) { second <- r })
+	unsub2, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) error {
+		second <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("second Subscribe: %v", err)
 	}
@@ -492,7 +519,10 @@ func TestSubscribeStopsWhenContextIsDone(t *testing.T) {
 	// Rooted at context.Background() because this is a test root.
 	subCtx, cancel := context.WithCancel(context.Background())
 	live := make(chan EventRef, 1)
-	unsub, err := f.Subscribe(subCtx, subject, func(_ context.Context, r EventRef) { live <- r })
+	unsub, err := f.Subscribe(subCtx, subject, func(_ context.Context, r EventRef) error {
+		live <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
@@ -512,7 +542,10 @@ func TestSubscribeStopsWhenContextIsDone(t *testing.T) {
 	// Gate on the replacement subscription receiving, exactly as the
 	// Unsubscribe test does.
 	after := make(chan EventRef, 1)
-	unsub2, err := f.Subscribe(pubCtx, subject, func(_ context.Context, r EventRef) { after <- r })
+	unsub2, err := f.Subscribe(pubCtx, subject, func(_ context.Context, r EventRef) error {
+		after <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("second Subscribe: %v", err)
 	}
@@ -622,7 +655,7 @@ func TestPoisonMessageParksOnDLQ(t *testing.T) {
 	// The callback panics: the fabric must treat a subscriber panic as a
 	// failure (neither crashing the process nor acking an unhandled event), so
 	// this exercises the panic guard and the retry budget together.
-	unsub, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) {
+	unsub, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) error {
 		attempts.Add(1)
 		panic("subscriber is broken")
 	})
@@ -687,7 +720,7 @@ func TestWildcardConsumerParksWithConcreteSubject(t *testing.T) {
 	if err != nil {
 		t.Fatalf("CommsSubject: %v", err)
 	}
-	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(context.Context, EventRef) {
+	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(context.Context, EventRef) error {
 		panic("subscriber is broken")
 	})
 	if err != nil {
@@ -727,6 +760,134 @@ func TestWildcardConsumerParksWithConcreteSubject(t *testing.T) {
 	}
 }
 
+// TestCallbackErrorParksOnDLQ: a returned error is a failure like a panic. It is
+// redelivered up to MaxDeliver attempts, then parked under its concrete subject.
+func TestCallbackErrorParksOnDLQ(t *testing.T) {
+	t.Parallel()
+	ctx := testCtx(t)
+	url := testServer(t)
+	f := newFabric(t, Config{URL: url, MaxDeliver: 2, Log: quietLogger(t)})
+
+	raw, err := nats.Connect(url)
+	if err != nil {
+		t.Fatalf("nats.Connect: %v", err)
+	}
+	t.Cleanup(raw.Close)
+	dlq, err := raw.SubscribeSync(DLQSubject)
+	if err != nil {
+		t.Fatalf("SubscribeSync(%q): %v", DLQSubject, err)
+	}
+	if err := raw.FlushWithContext(ctx); err != nil {
+		t.Fatalf("flushing the dlq subscription: %v", err)
+	}
+
+	concrete, err := CommsSubject("t1", KindMessagePosted)
+	if err != nil {
+		t.Fatalf("CommsSubject: %v", err)
+	}
+	var attempts atomic.Int64
+	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(context.Context, EventRef) error {
+		attempts.Add(1)
+		return errors.New("store read failed")
+	})
+	if err != nil {
+		t.Fatalf("SubscribeKind: %v", err)
+	}
+	defer unsub()
+
+	failing := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-failing"}
+	if err := f.Publish(ctx, concrete, failing); err != nil {
+		t.Fatalf("Publish: %v", err)
+	}
+
+	msg, err := dlq.NextMsgWithContext(ctx)
+	if err != nil {
+		t.Fatalf("waiting for the parked message on %q: %v", DLQSubject, err)
+	}
+	parked, err := decodeEventRef(msg.Data)
+	if err != nil {
+		t.Fatalf("the parked payload must be the original event: %v", err)
+	}
+	if parked != failing {
+		t.Fatalf("parked %+v, want %+v", parked, failing)
+	}
+	if got := msg.Header.Get(dlqHeaderSubject); got != concrete {
+		t.Errorf("park header %s = %q, want concrete subject %q", dlqHeaderSubject, got, concrete)
+	}
+	if reason := msg.Header.Get(dlqHeaderReason); !strings.Contains(reason, "store read failed") {
+		t.Errorf("park header %s = %q, want the callback's error", dlqHeaderReason, reason)
+	}
+	if got := attempts.Load(); got != 2 {
+		t.Errorf("callback ran %d time(s), want exactly MaxDeliver=2 attempts before parking", got)
+	}
+}
+
+// TestCallbackErrorThenSuccessIsAcked: a transient error is retried, and the
+// attempt that returns nil acks the event, so it is never parked.
+func TestCallbackErrorThenSuccessIsAcked(t *testing.T) {
+	t.Parallel()
+	ctx := testCtx(t)
+	url := testServer(t)
+	f := newFabric(t, Config{URL: url, MaxDeliver: 3, Log: quietLogger(t)})
+
+	raw, err := nats.Connect(url)
+	if err != nil {
+		t.Fatalf("nats.Connect: %v", err)
+	}
+	t.Cleanup(raw.Close)
+	dlq, err := raw.SubscribeSync(DLQSubject)
+	if err != nil {
+		t.Fatalf("SubscribeSync(%q): %v", DLQSubject, err)
+	}
+	if err := raw.FlushWithContext(ctx); err != nil {
+		t.Fatalf("flushing the dlq subscription: %v", err)
+	}
+
+	var attempts atomic.Int64
+	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(context.Context, EventRef) error {
+		if attempts.Add(1) == 1 {
+			return errors.New("transient")
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("SubscribeKind: %v", err)
+	}
+	defer unsub()
+
+	ref := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-flaky"}
+	subject, err := CommsSubject(ref.Tenant, ref.Kind)
+	if err != nil {
+		t.Fatalf("CommsSubject: %v", err)
+	}
+	if err := f.Publish(ctx, subject, ref); err != nil {
+		t.Fatalf("Publish: %v", err)
+	}
+
+	pollUntil(t, "a second attempt after the failed first", func() bool { return attempts.Load() >= 2 })
+	cons := kindConsumer(t, ctx, f, KindMessagePosted)
+	pollUntil(t, "the event acked", func() bool {
+		info, err := cons.Info(ctx)
+		if err != nil {
+			t.Fatalf("consumer Info: %v", err)
+		}
+		return info.NumAckPending == 0 && info.NumPending == 0
+	})
+	// Close flushes any park publish; the raw flush orders it ahead of the check.
+	if err := f.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if err := raw.FlushWithContext(ctx); err != nil {
+		t.Fatalf("flushing the raw connection: %v", err)
+	}
+	if n, _, err := dlq.Pending(); err != nil || n != 0 {
+		t.Fatalf("dlq pending = %d (err %v), want 0: a retried-then-ok event is acked", n, err)
+	}
+	if got := attempts.Load(); got != 2 {
+		t.Fatalf("callback ran %d time(s), want 2 (one failure, one success)", got)
+	}
+}
+
 // TestSubscriberPanicDoesNotBlockOtherEvents defends the panic guard's
 // consequence for throughput: one broken event must not wedge the subject. The
 // poison event exhausts its budget and parks, and the next event is delivered —
@@ -742,11 +903,12 @@ func TestSubscriberPanicDoesNotBlockOtherEvents(t *testing.T) {
 	}
 
 	good := make(chan EventRef, 1)
-	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) {
+	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) error {
 		if r.RowID == "poison" {
 			panic("subscriber is broken")
 		}
 		good <- r
+		return nil
 	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
@@ -796,7 +958,10 @@ func TestUndecodablePayloadParksImmediately(t *testing.T) {
 	}
 
 	var calls atomic.Int64
-	unsub, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) { calls.Add(1) })
+	unsub, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) error {
+		calls.Add(1)
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
@@ -842,11 +1007,12 @@ func TestSubscribeIsIdempotentAcrossInstances(t *testing.T) {
 		seen  []EventRef
 		total = make(chan struct{}, 8)
 	)
-	record := func(_ context.Context, r EventRef) {
+	record := func(_ context.Context, r EventRef) error {
 		mu.Lock()
 		seen = append(seen, r)
 		mu.Unlock()
 		total <- struct{}{}
+		return nil
 	}
 	unsubA, err := a.Subscribe(ctx, subject, record)
 	if err != nil {
@@ -937,19 +1103,24 @@ func TestEnsureStreamErrorIsNotCached(t *testing.T) {
 	}
 }
 
-// TestInvokeConvertsPanicToError defends the guard in isolation: a subscriber
-// callback runs on the fabric's goroutine, so an unrecovered panic there would
-// take the whole server down. It must become an error the delivery path can act
-// on.
-func TestInvokeConvertsPanicToError(t *testing.T) {
+// TestInvokeReturnsCallbackOutcome defends invoke in isolation: it passes the
+// callback's error through, and converts a panic to an error, so neither acks.
+func TestInvokeReturnsCallbackOutcome(t *testing.T) {
 	t.Parallel()
 	ref := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "m1"}
 
-	if err := invoke(context.Background(), func(context.Context, EventRef) {}, ref); err != nil {
-		t.Fatalf("a callback that returns normally must not error: %v", err)
+	if err := invoke(context.Background(), func(context.Context, EventRef) error { return nil }, ref); err != nil {
+		t.Fatalf("a callback that returns nil must not error: %v", err)
+	}
+
+	sentinel := errors.New("transient")
+	if err := invoke(context.Background(), func(context.Context, EventRef) error { return sentinel }, ref); !errors.Is(err, sentinel) {
+		t.Fatalf("invoke = %v, want the callback's own error (a nil would ack a failed event)", err)
 	}
 
-	err := invoke(context.Background(), func(context.Context, EventRef) { panic(errors.New("boom")) }, ref)
+	err := invoke(context.Background(), func(context.Context, EventRef) error {
+		panic(errors.New("boom"))
+	}, ref)
 	if err == nil {
 		t.Fatal("a panicking callback must yield an error, not a nil (which would ack an unhandled event)")
 	}
@@ -977,7 +1148,10 @@ func TestPublishRejectsCrossTenantRef(t *testing.T) {
 	}
 
 	got := make(chan EventRef, 2)
-	unsub, err := f.Subscribe(ctx, theirs, func(_ context.Context, r EventRef) { got <- r })
+	unsub, err := f.Subscribe(ctx, theirs, func(_ context.Context, r EventRef) error {
+		got <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
@@ -1037,7 +1211,9 @@ func TestParkReasonIsSanitizedAndBounded(t *testing.T) {
 	// multiple KB to blow any size bound. This is test code deliberately
 	// driving the package's documented panic guard, as the DLQ tests above do.
 	hostile := "line-one\r\nline-two " + strings.Repeat("x", 4096)
-	unsub, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) { panic(hostile) })
+	unsub, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) error {
+		panic(hostile)
+	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
@@ -1108,7 +1284,7 @@ func TestUnsubscribeDrainsBufferedEvents(t *testing.T) {
 		firstIn = make(chan struct{})
 		gateOne sync.Once
 	)
-	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) {
+	unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, r EventRef) error {
 		got <- r
 		// Only the first delivery blocks; that is enough to let the rest pile
 		// up in the consumer's buffer, which is what teardown must not discard.
@@ -1119,6 +1295,7 @@ func TestUnsubscribeDrainsBufferedEvents(t *testing.T) {
 			case <-time.After(gate):
 			}
 		})
+		return nil
 	})
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
@@ -1216,7 +1393,7 @@ func TestSubscribeWatchdogExitsOnClose(t *testing.T) {
 
 	// Rooted at context.Background() because this is a test root, and an
 	// uncancelled context is the whole point of the test.
-	unsub, err := f.Subscribe(context.Background(), subject, func(context.Context, EventRef) {})
+	unsub, err := f.Subscribe(context.Background(), subject, func(context.Context, EventRef) error { return nil })
 	if err != nil {
 		t.Fatalf("Subscribe: %v", err)
 	}
@@ -1267,7 +1444,10 @@ func TestSubscribeKindReceivesEveryTenant(t *testing.T) {
 	f := newFabric(t, Config{})
 
 	got := make(chan EventRef, 4)
-	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(_ context.Context, r EventRef) { got <- r })
+	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(_ context.Context, r EventRef) error {
+		got <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("SubscribeKind: %v", err)
 	}
@@ -1324,7 +1504,10 @@ func TestSubscribeKindIsolatesKinds(t *testing.T) {
 	f := newFabric(t, Config{})
 
 	got := make(chan EventRef, 4)
-	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(_ context.Context, r EventRef) { got <- r })
+	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(_ context.Context, r EventRef) error {
+		got <- r
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("SubscribeKind: %v", err)
 	}
@@ -1366,11 +1549,11 @@ func TestSubscribeKindRejectsBadInput(t *testing.T) {
 	if _, err := f.SubscribeKind(ctx, KindMessagePosted, nil); err == nil {
 		t.Error("SubscribeKind with a nil callback = nil error, want a refusal")
 	}
-	if _, err := f.SubscribeKind(ctx, EventKind("bad.kind"), func(context.Context, EventRef) {}); err == nil {
+	if _, err := f.SubscribeKind(ctx, EventKind("bad.kind"), func(context.Context, EventRef) error { return nil }); err == nil {
 		t.Error("SubscribeKind with a reserved-character kind = nil error, want a refusal")
 	}
 	// A wildcard kind would put all seven comms kinds on one consumer.
-	if _, err := f.SubscribeKind(ctx, EventKind("*"), func(context.Context, EventRef) {}); err == nil {
+	if _, err := f.SubscribeKind(ctx, EventKind("*"), func(context.Context, EventRef) error { return nil }); err == nil {
 		t.Error("SubscribeKind with a wildcard kind = nil error, want a refusal")
 	}
 }
@@ -1406,7 +1589,10 @@ func TestForgedTenantRefIsParked(t *testing.T) {
 	}
 
 	delivered := make(chan EventRef, 2)
-	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(_ context.Context, ref EventRef) { delivered <- ref })
+	unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(_ context.Context, ref EventRef) error {
+		delivered <- ref
+		return nil
+	})
 	if err != nil {
 		t.Fatalf("SubscribeKind: %v", err)
 	}
@@ -1502,7 +1688,10 @@ func TestLegitimateRefsSurviveTheSubjectCrossCheck(t *testing.T) {
 			t.Fatalf("CommsSubject(%q): %v", tenant, err)
 		}
 		delivered := make(chan EventRef, 1)
-		unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, ref EventRef) { delivered <- ref })
+		unsub, err := f.Subscribe(ctx, subject, func(_ context.Context, ref EventRef) error {
+			delivered <- ref
+			return nil
+		})
 		if err != nil {
 			t.Fatalf("Subscribe(%q): %v", subject, err)
 		}
@@ -1571,10 +1760,11 @@ func TestSlowCallbackPastAckWaitRedeliversHealthyEvent(t *testing.T) {
 	releaseHeld := sync.OnceFunc(func() { close(release) })
 	defer releaseHeld()
 	var first sync.Once
-	slow := func(_ context.Context, ref EventRef) {
+	slow := func(_ context.Context, ref EventRef) error {
 		invoked <- ref
 		// Only the first invocation is held, like a callback queued behind a gate.
 		first.Do(func() { <-release })
+		return nil
 	}
 	for name, f := range map[string]*Fabric{"a": a, "b": b} {
 		unsub, err := f.SubscribeKind(ctx, KindMessagePosted, slow)
@@ -1647,7 +1837,10 @@ func TestAlwaysSlowCallbackExhaustsMaxDeliver(t *testing.T) {
 	release := make(chan struct{})
 	releaseHeld := sync.OnceFunc(func() { close(release) })
 	defer releaseHeld()
-	slow := func(context.Context, EventRef) { <-release }
+	slow := func(context.Context, EventRef) error {
+		<-release
+		return nil
+	}
 	for name, f := range map[string]*Fabric{"a": a, "b": b} {
 		unsub, err := f.SubscribeKind(ctx, KindMessagePosted, slow)
 		if err != nil {
diff --git a/go/internal/fabric/fabric.go b/go/internal/fabric/fabric.go
index f61fb0cd1..0e8be971a 100644
--- a/go/internal/fabric/fabric.go
+++ b/go/internal/fabric/fabric.go
@@ -26,13 +26,15 @@ type Unsubscribe func()
 type EventFabric interface {
 	Publish(ctx context.Context, subject string, ref EventRef) error
 	// Subscribe and SubscribeKind invoke fn serially per subscription, with a
-	// ctx carrying the publisher's span context when one was propagated.
-	Subscribe(ctx context.Context, subject string, fn func(context.Context, EventRef)) (Unsubscribe, error)
+	// ctx carrying the publisher's span context when one was propagated. fn
+	// returning nil acks the event; an error or a panic redelivers it up to
+	// MaxDeliver attempts, then parks it on DLQSubject.
+	Subscribe(ctx context.Context, subject string, fn func(context.Context, EventRef) error) (Unsubscribe, error)
 	// SubscribeKind is the tenant-wildcard read side: one durable queue-group
 	// consumer receiving one kind across EVERY tenant, which is what the
 	// per-Server delivery singleton needs (§T3). Publish stays per-tenant and
 	// concrete.
-	SubscribeKind(ctx context.Context, kind EventKind, fn func(context.Context, EventRef)) (Unsubscribe, error)
+	SubscribeKind(ctx context.Context, kind EventKind, fn func(context.Context, EventRef) error) (Unsubscribe, error)
 	// OnReconnect runs fn after each NATS reconnect, once the fabric has logged
 	// it, so a consumer can sweep for events lost during the outage. fn runs on
 	// a fabric goroutine, never concurrently with itself; a burst of reconnects
diff --git a/go/internal/fabric/fabric_test.go b/go/internal/fabric/fabric_test.go
index 454db64ed..aaaf57966 100644
--- a/go/internal/fabric/fabric_test.go
+++ b/go/internal/fabric/fabric_test.go
@@ -507,10 +507,10 @@ func TestCloseIsIdempotentAndFailsClosed(t *testing.T) {
 	if err := f.Publish(ctx, subject, EventRef{Tenant: "t-closed", Kind: KindMessagePosted, RowID: "m1"}); !errors.Is(err, errClosed) {
 		t.Fatalf("Publish after Close: want errClosed, got %v", err)
 	}
-	if _, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) {}); !errors.Is(err, errClosed) {
+	if _, err := f.Subscribe(ctx, subject, func(context.Context, EventRef) error { return nil }); !errors.Is(err, errClosed) {
 		t.Fatalf("Subscribe after Close: want errClosed, got %v", err)
 	}
-	if _, err := f.SubscribeKind(ctx, KindMessagePosted, func(context.Context, EventRef) {}); !errors.Is(err, errClosed) {
+	if _, err := f.SubscribeKind(ctx, KindMessagePosted, func(context.Context, EventRef) error { return nil }); !errors.Is(err, errClosed) {
 		t.Fatalf("SubscribeKind after Close: want errClosed, got %v", err)
 	}
 	if err := f.SendCommand(ctx, "r1", nil); !errors.Is(err, errClosed) {
diff --git a/go/server/delivery_two_instance_pgtest_test.go b/go/server/delivery_two_instance_pgtest_test.go
index bffb848f3..e401eca6b 100644
--- a/go/server/delivery_two_instance_pgtest_test.go
+++ b/go/server/delivery_two_instance_pgtest_test.go
@@ -44,13 +44,13 @@ func newClaimCounter(fab fabric.EventFabric) *claimCounter {
 	}
 }
 
-func (c *claimCounter) SubscribeKind(ctx context.Context, kind fabric.EventKind, fn func(context.Context, fabric.EventRef)) (fabric.Unsubscribe, error) {
-	unsub, err := c.EventFabric.SubscribeKind(ctx, kind, func(ctx context.Context, ref fabric.EventRef) {
+func (c *claimCounter) SubscribeKind(ctx context.Context, kind fabric.EventKind, fn func(context.Context, fabric.EventRef) error) (fabric.Unsubscribe, error) {
+	unsub, err := c.EventFabric.SubscribeKind(ctx, kind, func(ctx context.Context, ref fabric.EventRef) error {
 		c.mu.Lock()
 		c.claims[ref.RowID]++
 		c.mu.Unlock()
 		c.claimed <- struct{}{}
-		fn(ctx, ref)
+		return fn(ctx, ref)
 	})
 	if err == nil {
 		c.once.Do(func() { close(c.subscribed) })

From 5b5c8cb79194f071f5e919ad02e41651a2c190bf Mon Sep 17 00:00:00 2001
From: mintaka <mintaka@rigel.build>
Date: Sat, 26 Sep 2026 07:28:38 -0400
Subject: [PATCH 2/2] docs(delivery): keep the hold doc within four lines
 (RIG-4030)

Spec-impact: none. Refs RIG-4030

Co-authored-by: Matt Wilkinson <matt@rigel.build>
---
 go/internal/delivery/dispatch.go | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/go/internal/delivery/dispatch.go b/go/internal/delivery/dispatch.go
index 0b26566b6..5712fbbf6 100644
--- a/go/internal/delivery/dispatch.go
+++ b/go/internal/delivery/dispatch.go
@@ -81,12 +81,10 @@ func (c *Consumer) onMessagePosted(ctx context.Context, msg *compassv1.Message)
 	return nil
 }
 
-// hold registers messageID under its author's session for later firing at the
-// author's settle edge (design.md:157-160), ordered by commit time (stable), so a
-// ref redelivered after a later one still fires in post order. It captures the
-// origin trace and tenant from ctx for fireHeld. If the author already settled at
-// or after atUnixMs, it also queues a settle edge, so the loop fires it at once
-// and still behind any earlier message of that author.
+// hold registers messageID under its author's session, ordered by commit time,
+// for firing at the author's settle edge (design.md:157-160); a ref redelivered
+// after a later one still fires in post order. If the author already settled at
+// or after atUnixMs, it also queues a settle edge so the loop fires it at once.
 //
 // The two clocks come from different instances. A settling clock behind the
 // committing one holds a message until the next settle, which is benign. A