D4: a merged message the forge bridge could not give for a transport fault is unread, not lost, and read again on the next message heard on the forge (issue 1966 note 153, ADR 0668 §1, §4)

The indexer's call for a merged message is a kept call (CallKept). A load that met a deadline or no
responders keeps the entry as unread (Pending.Unread, indexer_pending's unread), with a line that
names the read again rather than a drop; after every message heard on FORGE, of any kind, an unread
merge is evaluated again. The bridge's own answer stays lost. indexer_take_in_pending already reads
the message again (Q8).

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go

@@ -34,3 +34,11 @@ func messageVia(call toolCaller) func(ctx context.Context, repository, merge str
 		return a.Message, nil
 	}
 }
+
+// unreadAgain reads again, after a message heard on FORGE, a pending merge whose message was unread for a transport fault
+// (index.Indexer.UnreadAgain; issue 1966 note 153): the forge is heard again, so the bridge is asked again. A failure is said.
+func unreadAgain(ctx context.Context, ix *index.Indexer, logger interface{ Printf(string, ...any) }) {
+	if err := ix.UnreadAgain(ctx); err != nil {
+		logger.Printf("take-in: a merge unread for a transport fault is not read again after this message: %v", err)
+	}
+}

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go

@@ -430,7 +430,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.PackagesBucket}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call),
 		OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, TryRecovery: awaitedStore{b: tryRecovery}, Graph: awaitedStore{b: graph},
 		SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats,
-		Message: messageVia(calls.Call),
+		Message: messageVia(calls.CallKept), // a kept call: a read, called again through a transport fault (ADR 0668 §1)
 		// The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear;
 		// until then every merge is taken in at once, as before.
 		Pending:   awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)},
@@ -719,7 +719,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			index.ConfigCandidatesState: awaited(configCandidates), index.PackagesState: awaited(packages),
 			graphState: awaited(graph), index.HeardState: awaited(heard.b), "taken": awaited(taken), "findings": awaited(findings),
 			index.SeatDefinitionsState: awaited(seatDefs)}, errs, logger.Printf)
-	cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, heard, logger); ix.KickFill(ctx) }, logger.Printf)
+	cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
+		handle(ctx, m, ix, heard, logger)
+		ix.KickFill(ctx)
+		unreadAgain(ctx, ix, logger)
+	}, logger.Printf)
 	if err != nil {
 		return err
 	}

modules/mesh-delivery-indexer/internal/index/pending.go

@@ -17,6 +17,7 @@ import (
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
 	"git.novox.be/novox/mesh-delivery/sdk/trunk"
+	"git.novox.be/novox/mesh-sdk/go/busretry"
 	"git.novox.be/novox/mesh-sdk/go/keys"
 	"git.novox.be/novox/mesh-sdk/go/process"
 
@@ -87,6 +88,10 @@ type Pending struct {
 	SaidKind   string          `json:"said-kind,omitempty"`
 	SaidAt     time.Time       `json:"said-at,omitzero"`
 	Taking     string          `json:"taking,omitempty"`
+	// Unread is why the merged message could not be read for a transport fault (the forge bridge restarting, a deadline;
+	// ADR 0664's classes): not lost, it is read again on the next message heard on the forge, at the start and by
+	// indexer_take_in_pending (issue 1966 note 153, ADR 0668 §1).
+	Unread string `json:"unread,omitempty"`
 }
 
 // PendingEntry is one key of the pending state as read: its key, revision, value and the merged message decoded. For a
@@ -244,9 +249,9 @@ func (ix *Indexer) load(ctx context.Context, e PendingEntry) (PendingEntry, erro
 // loadNewest prepares the ready merges of one revision, in merge order. Of the merges of one repository only the newest is
 // loaded: applying it writes every module whose manifest differs from what is held, so it covers the older ones, which are
 // marked Covered and never read. A newest merge the forge bridge cannot give is answered in lost with its reason, by key, and its revision is held (holdLost).
-func (ix *Indexer) loadNewest(ctx context.Context, entries []PendingEntry) (out, lost []PendingEntry, reasons map[string]string) {
+func (ix *Indexer) loadNewest(ctx context.Context, entries []PendingEntry) (out, lost []PendingEntry, reasons map[string]string, unread map[string]bool) {
 	out = slices.Clone(entries)
-	reasons = map[string]string{}
+	reasons, unread = map[string]string{}, map[string]bool{}
 	newest := map[string]bool{}
 	for i := len(out) - 1; i >= 0; i-- {
 		repo := strings.ToLower(out[i].Forge.Repository)
@@ -267,12 +272,14 @@ func (ix *Indexer) loadNewest(ctx context.Context, entries []PendingEntry) (out,
 		if err != nil {
 			lost = append(lost, out[i])
 			reasons[out[i].Key] = err.Error()
+			// A transport fault is not the bridge's answer: the message is unread, never lost (ADR 0668 §1).
+			unread[out[i].Key] = busretry.Undelivered(err) || busretry.TimedOut(err)
 			continue
 		}
 		out[i] = full
 	}
 	slices.Reverse(lost)
-	return out, lost, reasons
+	return out, lost, reasons, unread
 }
 
 // Waits is the pending state read as what waits (ADR 0633 §3, §4): Ready are the merges no group still being gathered
@@ -670,6 +677,24 @@ func (ix *Indexer) TakeInAtStart(ctx context.Context) (Applied, string, error) {
 	return a, words, err
 }
 
+// UnreadAgain evaluates pending again when a merge in it is unread for a transport fault (Pending.Unread): told after
+// every message heard on the forge, of any kind, so a merge whose message the forge bridge could not give while it
+// restarted is read again once the forge is heard again, with no timer (issue 1966 note 153, ADR 0668 §4). It reads
+// pending once and does nothing when no merge is unread.
+func (ix *Indexer) UnreadAgain(ctx context.Context) error {
+	if ix.Pending == nil || !ix.pendingOnBus(ctx) {
+		return nil
+	}
+	entries, err := ReadPending(ctx, ix.Pending, ix.logf)
+	if err != nil {
+		return err
+	}
+	if !slices.ContainsFunc(entries, func(e PendingEntry) bool { return e.Value.Unread != "" }) {
+		return nil
+	}
+	return ix.PendingAgain(ctx)
+}
+
 // PendingAgain evaluates pending again after something a fingerprint is computed from changed, a toolchain report
 // (ADR 0633 §1, §3): a revision that now names the running build is taken in, and one that names another build is held
 // for that one. It does nothing while the pending state is not on the bus or the build this program runs is not known.
@@ -722,9 +747,9 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie
 	// Of each repository's merges only the newest is read (from memory, else asked of the forge bridge: a pending value
 	// keeps a reference); the older ones are covered by it. A newest merge that cannot be read is said and
 	// holds its revision: nothing of it is taken in silently.
-	ready, lost, reasons := ix.loadNewest(ctx, w.Ready)
+	ready, lost, reasons, unread := ix.loadNewest(ctx, w.Ready)
 	if len(lost) > 0 {
-		return Applied{}, ix.waitingLine(w), ix.holdLost(ctx, w.Ready, lost, reasons)
+		return Applied{}, ix.waitingLine(w), ix.holdLost(ctx, w.Ready, lost, reasons, unread)
 	}
 	names, dry, err := ix.computeOver(ctx, ready)
 	if err != nil {
@@ -743,8 +768,9 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie
 // holdLost holds a revision whose newest merge of a repository cannot be read (the forge bridge's stream no longer holds
 // it, or the bus refuses the read): said once on take-in.pending with the lost keys (TakeIn.Lost), and on each lost entry
 // (Pending.Lost, the reason, which indexer_pending shows), until a newer merge of its repository covers it or
-// indexer_drop_pending drops it. Nothing of the revision is taken in meanwhile.
-func (ix *Indexer) holdLost(ctx context.Context, ready, lost []PendingEntry, reasons map[string]string) error {
+// indexer_drop_pending drops it. A merge unread for a transport fault (unread) is kept as Pending.Unread, not Lost, and read
+// again on the next message heard on the forge (UnreadAgain). Nothing of the revision is taken in meanwhile.
+func (ix *Indexer) holdLost(ctx context.Context, ready, lost []PendingEntry, reasons map[string]string, unread map[string]bool) error {
 	t := revisionWait(takein.TakeInPending, ready, "")
 	for _, g := range lost {
 		t.Lost = append(t.Lost, g.Key)
@@ -752,7 +778,7 @@ func (ix *Indexer) holdLost(ctx context.Context, ready, lost []PendingEntry, rea
 	said := true
 	for _, e := range ready {
 		isLost := slices.Contains(t.Lost, e.Key)
-		if e.Value.Said != t.Key || e.Value.SaidKind != takein.TakeInPending || (isLost && e.Value.Lost == "") {
+		if e.Value.Said != t.Key || e.Value.SaidKind != takein.TakeInPending || (isLost && (unread[e.Key] && e.Value.Unread == "" || !unread[e.Key] && e.Value.Lost == "")) {
 			said = false
 		}
 	}
@@ -760,6 +786,11 @@ func (ix *Indexer) holdLost(ctx context.Context, ready, lost []PendingEntry, rea
 		return nil
 	}
 	for _, g := range lost {
+		if unread[g.Key] {
+			ix.logf("take-in: %s (%s#%d) is not taken in and holds its revision, its message is unread for a transport fault: %s; it is read again on the next message heard on the forge, at the start, and by %s",
+				g.Key, g.Forge.Repository, g.Forge.Number, reasons[g.Key], TakeInPendingTool)
+			continue
+		}
 		ix.logf("take-in: %s (%s#%d) is not taken in and holds its revision, the forge bridge could not give its message: %s; %s reads it again; indexer_drop_pending drops it once a later merge of %s covers it",
 			g.Key, g.Forge.Repository, g.Forge.Number, reasons[g.Key], TakeInPendingTool, g.Forge.Repository)
 	}
@@ -773,8 +804,12 @@ func (ix *Indexer) holdLost(ctx context.Context, ready, lost []PendingEntry, rea
 		}
 	}
 	for _, g := range lost {
-		reason := reasons[g.Key]
-		ix.markSaid(ctx, []PendingEntry{g}, &t, func(p *Pending) { p.Lost = reason })
+		reason, isUnread := reasons[g.Key], unread[g.Key]
+		ix.markSaid(ctx, []PendingEntry{g}, &t, func(p *Pending) {
+			if p.Lost, p.Unread = reason, ""; isUnread {
+				p.Lost, p.Unread = "", reason
+			}
+		})
 	}
 	ix.markSaid(ctx, kept, &t, nil)
 	return nil
@@ -1179,6 +1214,9 @@ func waitOf(entries []PendingEntry, what, behind string) takein.PendingWait {
 		if e.Value.Lost != "" {
 			pw.Lost = append(pw.Lost, e.Key)
 		}
+		if e.Value.Unread != "" {
+			pw.Unread = append(pw.Unread, e.Key)
+		}
 		if e.Value.Refused != "" {
 			pw.Refused = e.Value.Refused
 		}
@@ -1224,7 +1262,7 @@ func (ix *Indexer) TakeInPendingByHand(ctx context.Context, key, by, why string)
 		r.Answer = fmt.Sprintf("nothing waits under %s: indexer_pending names the waits", key)
 		return r
 	}
-	take, lost, reasons := ix.loadNewest(ctx, take)
+	take, lost, reasons, _ := ix.loadNewest(ctx, take)
 	if len(lost) > 0 {
 		r.Answer = fmt.Sprintf("nothing is taken in: the message of %s is not read (%s); asked again, it is read again; indexer_drop_pending drops it once a later merge of its repository covers it", lost[0].Key, reasons[lost[0].Key])
 		return r

modules/mesh-delivery-indexer/internal/index/pending_test.go

@@ -17,8 +17,10 @@ import (
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
 	"git.novox.be/novox/mesh-delivery/sdk/trunk"
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
+	"git.novox.be/novox/mesh-sdk/go/busretry"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/process"
+	"github.com/nats-io/nats.go"
 )
 
 // takeInBed is an indexer holding the catalogue's avahi and the indexer's own module, both with a fingerprint, its
@@ -554,6 +556,37 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 		if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('e') {
 			t.Errorf("after a lost start: pending %v, avahi at %s: want all taken in by the next merged message", b.keys(), b.commit(t, "avahi"))
 		}
+		// Issue 1966 note 153, ADR 0668 §1, §4: a transport fault (the bridge restarting past the call's bound) leaves the
+		// newest merge unread, not lost, and indexer_pending says so; the next message heard on the forge, of any kind, reads
+		// it again. The bridge's own answer (it holds no such merge) is lost, as before, and a heard message does not ask it.
+		for _, c := range []struct {
+			name   string
+			fault  error
+			unread bool
+		}{
+			{"a deadline", fmt.Errorf("the bridge: %w", nats.ErrTimeout), true},
+			{"no responders", fmt.Errorf("the bridge: %w", nats.ErrNoResponders), true},
+			{"the bridge's answer", &busretry.Answered{Words: "no merge at that commit"}, false},
+		} {
+			b = held()
+			answers := b.ix.Message
+			b.ix.held = nil
+			b.ix.Message = func(context.Context, string, string, int) ([]byte, error) { return nil, c.fault }
+			if _, _, err := b.ix.TakeInAtStart(ctx); err != nil || len(b.keys()) != 2 {
+				t.Fatalf("%s at the start: err %v, pending %v", c.name, err, b.keys())
+			}
+			waits := PendingWaits(mustReadPending(t, b))
+			if len(waits) != 1 || (len(waits[0].Unread) == 1) != c.unread || (len(waits[0].Lost) == 1) == c.unread {
+				t.Fatalf("%s: indexer_pending %+v: want unread %v, lost %v", c.name, waits, c.unread, !c.unread)
+			}
+			b.ix.Message = answers
+			if err := b.ix.UnreadAgain(ctx); err != nil {
+				t.Fatal(err)
+			}
+			if taken := len(b.keys()) == 0 && b.commit(t, "avahi") == sha('d'); taken != c.unread {
+				t.Errorf("%s, a message heard after: pending %v, avahi at %s: want taken in %v", c.name, b.keys(), b.commit(t, "avahi"), c.unread)
+			}
+		}
 	})
 	t.Run("T5 two take-ins of one pending revision: the one that loses the claim writes nothing", func(t *testing.T) {
 		b := newTakeInBed(t)
@@ -643,7 +676,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 			key, _ := ToolchainKey(node, "go")
 			x.PutToolchain("mesh-build-go", key, toolchainValue(t, "go", "go1.27.2", "1"), false)
 		}
-		ready, _, _ := b.ix.loadNewest(ctx, WaitsOf(entries).Ready)
+		ready, _, _, _ := b.ix.loadNewest(ctx, WaitsOf(entries).Ready)
 		next, _, err := x.computeOver(ctx, ready)
 		if err != nil || next == "" || next == b.said[0].WaitsFor {
 			t.Fatalf("the build under the next toolchain: %q %v (the revision names %s)", next, err, b.said[0].WaitsFor)

modules/mesh-delivery-indexer/takein/takein.go

@@ -243,6 +243,9 @@ type PendingWait struct {
 	SaidAt   time.Time                 `json:"said-at,omitzero"`
 	// Lost are the keys of its merges whose message the forge bridge's stream no longer holds (TakeIn.Lost).
 	Lost []string `json:"lost,omitempty"`
+	// Unread are the keys of its merges whose message is unread for a transport fault: read again on the next message
+	// heard on the forge, at the start and by indexer_take_in_pending, never dropped for it (issue 1966, ADR 0668 §1).
+	Unread []string `json:"unread,omitempty"`
 	// Refused is why the manifest of the indexer it names was refused (TakeIn.Refused).
 	Refused string `json:"refused,omitempty"`
 }