feat(fabric): let a subscriber callback fail so the event is redelivered (RIG-4030)

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")
 }
 

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

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) {

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,21 @@ 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())
+	return nil
 }
 
-// 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)
-}
-
-// 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.
+// 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
@@ -103,7 +96,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 {