From c3b240bdfaf79b0812faa1db859c469f2c096747 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 03:14:12 +0100 Subject: [PATCH] =?UTF-8?q?D4:=20a=20merged=20message=20the=20forge=20brid?= =?UTF-8?q?ge=20could=20not=20give=20for=20a=20transport=20fault=20is=20un?= =?UTF-8?q?read,=20not=20lost,=20and=20read=20again=20on=20the=20next=20me?= =?UTF-8?q?ssage=20heard=20on=20the=20forge=20(issue=201966=20note=20153,?= =?UTF-8?q?=20ADR=200668=20=C2=A71,=20=C2=A74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../mesh-delivery-indexer/bridgemessage.go | 8 +++ .../cmd/mesh-delivery-indexer/main.go | 8 ++- .../internal/index/pending.go | 60 +++++++++++++++---- .../internal/index/pending_test.go | 35 ++++++++++- .../mesh-delivery-indexer/takein/takein.go | 3 + 5 files changed, 100 insertions(+), 14 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go index 1bf9196e..f99bb36a 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go +++ b/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) + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index b2ae8c70..acd2104c 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/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 } diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index ab039baa..9999039f 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/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 diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 0002e64e..c0bd87e8 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/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) diff --git a/modules/mesh-delivery-indexer/takein/takein.go b/modules/mesh-delivery-indexer/takein/takein.go index 645535c2..59141f58 100644 --- a/modules/mesh-delivery-indexer/takein/takein.go +++ b/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"` }