From 5b13cc7ab1e809c29dfad057d2d59099da62894b Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:53:45 +0100 Subject: [PATCH] IX24: the indexer's four kept calls are asked again only while their ask is open MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit messageVia, stepsVia, sourceVia and versionsVia go through CallKeptWhileOpen: open while the commit's take-in is pending, the candidate is held, the branch is watched, the module is in the newest view. A call that ends with its ask closed is index.ErrAskClosed, not an unanswered call. Builds: design 47 G §351, §367, §460, §531; A-190 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-indexer/bridgemessage.go | 10 +- .../cmd/mesh-delivery-indexer/keptopen.go | 29 +++++ .../mesh-delivery-indexer/keptopen_test.go | 38 ++++++ .../cmd/mesh-delivery-indexer/main.go | 9 +- .../cmd/mesh-delivery-indexer/main_test.go | 5 +- .../cmd/mesh-delivery-indexer/source.go | 12 +- .../cmd/mesh-delivery-indexer/steps.go | 9 +- .../cmd/mesh-delivery-indexer/versions.go | 9 +- .../internal/index/open_asks.go | 104 ++++++++++++++++ .../internal/index/open_asks_test.go | 111 ++++++++++++++++++ .../internal/index/pending.go | 2 +- .../internal/index/seed_steps.go | 2 + .../internal/index/source.go | 2 +- .../internal/index/versions_fill.go | 7 +- .../internal/index/versions_fill_edges.go | 2 +- .../internal/index/versions_test.go | 4 +- 16 files changed, 327 insertions(+), 28 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/keptopen.go create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/keptopen_test.go create mode 100644 modules/mesh-delivery-indexer/internal/index/open_asks.go create mode 100644 modules/mesh-delivery-indexer/internal/index/open_asks_test.go 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 f99bb36a..e56ae60d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go @@ -14,14 +14,16 @@ import ( const mergedTool = index.ForgeBridgeModule + ".bridge_merged_message" // messageVia asks the forge bridge for a merge's merged message, for a held merge the indexer does not hold in memory (a -// restart, a new build). An error is the bridge's own words. -func messageVia(call toolCaller) func(ctx context.Context, repository, merge string, number int) ([]byte, error) { - return func(_ context.Context, repository, merge string, number int) ([]byte, error) { +// restart, a new build), a kept call asked again only while the commit's take-in is pending (A-190). An error is the +// bridge's own words. +func messageVia(call openCaller, pending func(ctx context.Context, repository, merge string) (bool, error), logf func(string, ...any)) func(ctx context.Context, repository, merge string, number int) ([]byte, error) { + return func(ctx context.Context, repository, merge string, number int) ([]byte, error) { ask := map[string]any{"repository": repository, "merge": merge} if number > 0 { ask["number"] = number } - raw, err := call(mergedTool, ask) + raw, err := keptWhileOpen(ctx, call, logf, "message."+repository+"."+merge, mergedTool, ask, + func(ctx context.Context) (bool, error) { return pending(ctx, repository, merge) }) if err != nil { return nil, err } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/keptopen.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/keptopen.go new file mode 100644 index 00000000..866cab6a --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/keptopen.go @@ -0,0 +1,29 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/busretry" + + "mesh-delivery-indexer/internal/index" +) + +// openCaller is process.Conn.CallKeptWhileOpen: a kept call asked again only while its ask is open (design 47 G §351, +// §460, §531; A-190). +type openCaller func(o busretry.Open, key string, body any) (json.RawMessage, error) + +// keptWhileOpen asks verb as a kept call keyed by key, asked again only while open says what the ask serves still stands. +// A call that ends with its ask closed is index.ErrAskClosed, never an unanswered call (A-190). +func keptWhileOpen(ctx context.Context, call openCaller, logf func(string, ...any), key, verb string, body any, + open func(context.Context) (bool, error)) (json.RawMessage, error) { + raw, err := call(busretry.Open{Key: key, Open: func(string) (bool, error) { return open(ctx) }, Logf: logf}, verb, body) + if err == nil { + return raw, nil + } + if still, oerr := open(ctx); oerr == nil && !still { + return nil, fmt.Errorf("%s: %w", verb, index.ErrAskClosed) + } + return nil, err +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/keptopen_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/keptopen_test.go new file mode 100644 index 00000000..a6742daa --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/keptopen_test.go @@ -0,0 +1,38 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "git.novox.be/novox/mesh-sdk/go/busretry" + + "mesh-delivery-indexer/internal/index" +) + +// A-190: the four asks pass their key's open function to the kept call; a call that fails with its key closed is +// index.ErrAskClosed, one that fails while it is open keeps its own error. +func TestKeptWhileOpenNamesAClosedAsk(t *testing.T) { + ctx := context.Background() + failing := errors.New("no responders") + for _, c := range []struct { + name string + open bool + answer error + want error + }{{"open", true, failing, failing}, {"closed", false, failing, index.ErrAskClosed}, {"answered", false, nil, nil}} { + var asked busretry.Open + call := func(o busretry.Open, key string, _ any) (json.RawMessage, error) { + asked = o + return json.RawMessage(`{}`), c.answer + } + _, err := keptWhileOpen(ctx, call, t.Logf, "k1", "tool", nil, func(context.Context) (bool, error) { return c.open, nil }) + if !errors.Is(err, c.want) { + t.Errorf("%s: err = %v, want %v", c.name, err, c.want) + } + if got, _ := asked.Open("k1"); asked.Key != "k1" || got != c.open { + t.Errorf("%s: the kept call was given key %q open %v", c.name, asked.Key, got) + } + } +} 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 8136c950..62b17d88 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -449,7 +449,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}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), TryRecovery: awaitedStore{b: tryRecovery}, Graph: graph, SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, - Message: messageVia(calls.CallKept), Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.EntryIndexBucket, "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1) + Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.EntryIndexBucket, "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1) // Each catalog revision may bring another holder of a build seat: the build modules are found again (buildmodules.go). SayCatalogRevision: afterRevision(refreshingBuildModules(sayCatalogRevision(conn, cred), &buildFollow), &fillFollow), SayFill: sayFill(conn, cred), @@ -469,11 +469,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.RegistrationsBucket, "no registration is kept or written", logger.Printf)}, awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.ProposalsBucket, "no change of a registration waits for its warrant", logger.Printf)}, routerRecordOf(cred.Module, calls.Call, calls.Ask), ix, logger.Printf) + ix.Message = messageVia(calls.CallKeptWhileOpen, ix.TakeInPending, logger.Printf) // a kept call, called again through a transport fault while the commit's take-in is pending (ADR 0668 §1, A-190) registry.Pruned = ix.PruneBranchBuilds ix.Registry = registry named.Changed = ix.RemoveUnlisted // The first fill of the versions (design 47 G §372): at the start, and again at each catalog revision kept until it is made. - fill := &versionsFill{ix: ix, ctx: ctx, ask: versionsVia(calls.CallKept), logf: logger.Printf} + fill := &versionsFill{ix: ix, ctx: ctx, ask: versionsVia(calls.CallKeptWhileOpen, ix.ModuleInNewestView, logger.Printf), logf: logger.Printf} fillFollow.Store(fill) fill.kick() ix.SeedModules = func(ctx context.Context) ([]index.SeedModule, error) { @@ -481,7 +482,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if r == nil { return nil, errors.New("the seat list is not read yet: the seed waits for it") } - return ix.SeedModulesFrom(ctx, r.Definitions(), stepsVia(calls.CallKept)) + return ix.SeedModulesFrom(ctx, r.Definitions(), stepsVia(calls.CallKeptWhileOpen, ix.SeedEntryHeld, logger.Printf)) } // Which build this program is, fixed here once, before it hears any merge and never asked again (index.FixOwnBuild): the // node-engine names the successor while it replaces this process, and an old process that learnt it later would take @@ -518,7 +519,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The source revision of a watched branch (design 47 G §15, §115, §362): asked of the seat mesh-delivery's verb source as a // kept call, once at the start, before the handler of its event source-moved reads anything; a source-moved heard // meanwhile is read after. - source := &sourceFollow{read: sourceVia(calls.CallKept), apply: ix.ApplySource, applyBranch: ix.ApplyBranchSource, logf: logger.Printf} + source := &sourceFollow{read: sourceVia(ctx, calls.CallKeptWhileOpen, ix.BranchWatched, logger.Printf), apply: ix.ApplySource, applyBranch: ix.ApplyBranchSource, logf: logger.Printf} go source.catchUp(ctx) // Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the // assignments is settled first, so the toolchains settle through the gate already knowing it. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index eea5e715..75144627 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-sdk/go/busretry" "log" "reflect" "strings" @@ -234,7 +235,7 @@ func TestTheSourceIsReadFromTheSeatAndASourceMovedDuringTheCatchUpIsReadAfterIt( func TestTheSeatsSourceAnswerIsReadAndItsErrorIsAnError(t *testing.T) { // design 47 G §115 var asked any - read := sourceVia(func(key string, body any) (json.RawMessage, error) { + read := sourceVia(context.Background(), func(_ busretry.Open, key string, body any) (json.RawMessage, error) { asked = body if key != "seat:mesh-delivery.source" { t.Errorf("asked %q", key) @@ -243,7 +244,7 @@ func TestTheSeatsSourceAnswerIsReadAndItsErrorIsAnError(t *testing.T) { // desig return json.RawMessage(`{"error":"source.bad is not readable"}`), nil } return json.RawMessage(`{"branch":"main","revision":7,"repositories":[{"repository":"novox/mesh-delivery","head":"` + head + `","trunk":"main"}]}`), nil - }) + }, func(context.Context, string) (bool, error) { return true, nil }, t.Logf) a, err := read("main") if err != nil || a.Revision != 7 || len(a.Repositories) != 1 || asked.(meshdelivery.SourceInput).Branch != "main" { t.Errorf("source = %+v, %v", a, err) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go index b7c6e267..c9bc6cc2 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "maps" "slices" @@ -12,17 +13,19 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-sdk/go/keys" "github.com/nats-io/nats.go/jetstream" + "mesh-delivery-indexer/internal/index" ) // sourceInvoke is the seat mesh-delivery's verb source as the manifest's invokes names it: a watched branch's source // revision, asked of the seat and never read from the gate's state (design 47 G §15, §115; ADR 0596 §2). const sourceInvoke = "seat:" + meshdelivery.Seat + "." + meshdelivery.VerbSource -// sourceVia asks the seat for one branch's source revision through call (process.Conn.CallKept, a read) and reads its +// sourceVia asks the seat for one branch's source revision through call (process.Conn.CallKeptWhileOpen, a read asked again while the branch is watched, A-190) and reads its // declared answer; an answer that carries an error is that error. -func sourceVia(call toolCaller) func(branch string) (meshdelivery.SourceAnswer, error) { +func sourceVia(ctx context.Context, call openCaller, watched func(ctx context.Context, branch string) (bool, error), logf func(string, ...any)) func(branch string) (meshdelivery.SourceAnswer, error) { return func(branch string) (meshdelivery.SourceAnswer, error) { - raw, err := call(sourceInvoke, meshdelivery.SourceInput{Branch: branch}) + raw, err := keptWhileOpen(ctx, call, logf, "source."+branch, sourceInvoke, meshdelivery.SourceInput{Branch: branch}, + func(ctx context.Context) (bool, error) { return watched(ctx, branch) }) if err != nil { return meshdelivery.SourceAnswer{}, err } @@ -97,6 +100,9 @@ func (s *sourceFollow) moved(ctx context.Context, branch string) { // through the take-in, another watched branch's into its own revision (A-93). func (s *sourceFollow) take(ctx context.Context, branch string) { a, err := s.read(branch) + if errors.Is(err, index.ErrAskClosed) { // the branch is no longer watched: not said unanswered (A-190) + return + } if err != nil { s.logf("source.%s is not read, and is read again at its next source-moved: %v", branch, err) return diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/steps.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/steps.go index 1dea5455..0a3cfb56 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/steps.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/steps.go @@ -10,12 +10,13 @@ import ( "mesh-delivery-indexer/internal/index" ) -// stepsVia asks a build seat's verb steps through call (process.Conn.CallKept, ADR 0668 §1) and reads its answer (design +// stepsVia asks a build seat's verb steps through call (process.Conn.CallKeptWhileOpen, ADR 0668 §1, asked again while the candidate is held, A-190) and reads its answer (design // 47 G §704); the seats' verbs have one form, so nodebuildgo's contract reads each seat's answer. -func stepsVia(call toolCaller) index.StepsCaller { - return func(_ context.Context, seat string, in nodebuildgo.StepsInput) (nodebuildgo.StepsAnswer, error) { +func stepsVia(call openCaller, held func(ctx context.Context, module, fingerprint string) (bool, error), logf func(string, ...any)) index.StepsCaller { + return func(ctx context.Context, seat string, in nodebuildgo.StepsInput) (nodebuildgo.StepsAnswer, error) { verb := "seat:" + seat + "." + index.StepsVerb - raw, err := call(verb, in) + raw, err := keptWhileOpen(ctx, call, logf, "steps."+seat+"."+in.Module+"."+in.Fingerprint, verb, in, + func(ctx context.Context) (bool, error) { return held(ctx, in.Module, in.Fingerprint) }) if err != nil { return nodebuildgo.StepsAnswer{}, err } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go index 3e750328..cb5dc56d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go @@ -136,11 +136,12 @@ func sayVersionsFilled(conn *nats.Conn, cred process.Account) func(context.Conte } } -// versionsVia asks the builder's tool builder_versions for one name as a kept call (ADR 0668 §1, §2) and reads its answer. -func versionsVia(call toolCaller) index.VersionsAsker { - return func(_ context.Context, name string) (index.BuilderVersionsAnswer, error) { +// versionsVia asks the builder's tool builder_versions for one name as a kept call (ADR 0668 §1, §2) asked again while the module is in the newest view (A-190) and reads its answer. +func versionsVia(call openCaller, inView func(ctx context.Context, module string) (bool, error), logf func(string, ...any)) index.VersionsAsker { + return func(ctx context.Context, module, name string) (index.BuilderVersionsAnswer, error) { var a index.BuilderVersionsAnswer - raw, err := call(index.BuilderVersionsCall, map[string]any{"name": name}) + raw, err := keptWhileOpen(ctx, call, logf, "versions."+name, index.BuilderVersionsCall, map[string]any{"name": name}, + func(ctx context.Context) (bool, error) { return inView(ctx, module) }) if err != nil { return a, err } diff --git a/modules/mesh-delivery-indexer/internal/index/open_asks.go b/modules/mesh-delivery-indexer/internal/index/open_asks.go new file mode 100644 index 00000000..576dc802 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/open_asks.go @@ -0,0 +1,104 @@ +package index + +import ( + "context" + "errors" + "fmt" + "slices" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// ErrAskClosed is what a kept call answers when its ask was closed while it was made again (design 47 G §351, §460, A-190): +// what the ask served no longer stands, so the call stopped and is not said unanswered. +var ErrAskClosed = errors.New("the ask is closed: what it served no longer stands") + +// takingIn marks a context as the take-in of one commit's merged message, which is pending while that context lives. +type takingIn struct{ repository, merge string } + +type takingInKey struct{} + +// forTakeIn is ctx marked as the take-in of the merged message of repository's commit merge. +func forTakeIn(ctx context.Context, repository, merge string) context.Context { + return context.WithValue(ctx, takingInKey{}, takingIn{repository, merge}) +} + +// TakeInPending says the take-in of the commit merge of repository is still pending (A-190, for the forge bridge's +// merged_message): its take-in is the one ctx is made for, or a pending entry names it. +func (ix *Indexer) TakeInPending(ctx context.Context, repository, merge string) (bool, error) { + if ctx.Err() != nil { + return false, nil + } + if t, ok := ctx.Value(takingInKey{}).(takingIn); ok && t.repository == repository && t.merge == merge { + return true, nil + } + if ix.Pending == nil { + return false, nil + } + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + return false, err + } + return slices.ContainsFunc(entries, func(e PendingEntry) bool { + return e.Forge.Merge == merge && e.Forge.Repository == repository + }), nil +} + +// SeedEntryHeld says the newest trunk revision still holds the entry of module at the fingerprint fp, the candidate the +// seed asks steps for (A-190). +func (ix *Indexer) SeedEntryHeld(ctx context.Context, module, fp string) (bool, error) { + if ctx.Err() != nil || ix.Revisions == nil { + return false, ctx.Err() + } + rev, found, err := newestTrunk(ctx, ix.Revisions) + if err != nil || !found { + return false, err + } + _, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", rev.Revision)) + if err != nil { + return false, err + } + return slices.ContainsFunc(entries, func(e meshcatalog.CatalogEntry) bool { return e.Module == module && e.Fp == fp }), nil +} + +// ModuleInNewestView says module is among the entries of the newest trunk revision, the view the versions are filled for (A-190). +func (ix *Indexer) ModuleInNewestView(ctx context.Context, module string) (bool, error) { + if ctx.Err() != nil || ix.Revisions == nil { + return false, ctx.Err() + } + rev, found, err := newestTrunk(ctx, ix.Revisions) + if err != nil || !found { + return false, err + } + _, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", rev.Revision)) + if err != nil { + return false, err + } + return slices.ContainsFunc(entries, func(e meshcatalog.CatalogEntry) bool { return e.Module == module }), nil +} + +// BranchWatched says branch (as the source's key names it, escaped; empty asks the list of every branch) is watched: the +// trunk always, another branch while a registration names it (A-190). With no registry, every branch is. +func (ix *Indexer) BranchWatched(ctx context.Context, branch string) (bool, error) { + if ctx.Err() != nil { + return false, nil + } + if isTrunkBranch(branch) || ix.Registry == nil || ix.Registry.Registrations == nil { + return true, nil + } + name, err := keys.UnescapeName(branch) + if err != nil { + return false, nil + } + regs, err := ix.Registry.RegistrationsAnswer(ctx, meshcatalog.RegistrationsInput{}) + if err != nil { + return false, err + } + for _, r := range regs.Registrations { + if slices.Contains(r.Branches, name) { + return true, nil + } + } + return false, nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/open_asks_test.go b/modules/mesh-delivery-indexer/internal/index/open_asks_test.go new file mode 100644 index 00000000..7471afeb --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/open_asks_test.go @@ -0,0 +1,111 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "testing" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// A-190: a kept call is asked again only while what it serves still stands; each of the indexer's four has its predicate. +func TestKeptCallsOpenWhileTheirAskServesSomething(t *testing.T) { + ctx := context.Background() + t.Run("the take-in of a commit is pending while an entry names it or the take-in is made", func(t *testing.T) { + pending := newMem() + val, _ := encodePending(Pending{Seq: 1, Ref: &PendingRef{Repository: "novox/r", Merge: "c1"}}) + if err := pending.Create(ctx, "m.1", val); err != nil { + t.Fatal(err) + } + ix := &Indexer{Pending: pending} + cancelled, cancel := context.WithCancel(ctx) + cancel() + for _, c := range []struct { + name string + ctx context.Context + repo, commit string + want bool + }{ + {"held", ctx, "novox/r", "c1", true}, + {"taken in or dropped", ctx, "novox/r", "c2", false}, + {"in hand", forTakeIn(ctx, "novox/r", "c2"), "novox/r", "c2", true}, + {"in hand for another commit", forTakeIn(ctx, "novox/r", "c3"), "novox/r", "c2", false}, + {"the take-in's context ended", cancelled, "novox/r", "c1", false}, + } { + if got, err := ix.TakeInPending(c.ctx, c.repo, c.commit); err != nil || got != c.want { + t.Errorf("%s: pending = %v, %v, want %v", c.name, got, err, c.want) + } + } + }) + t.Run("the seed's candidate and the versions' module are in the newest trunk revision", func(t *testing.T) { + ix := &Indexer{Revisions: newMemObjects(1 << 20)} + keep := func(n int64, fp string) { + rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: n, Kind: "trunk"}, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Fp: fp, Manifest: map[string]any{}}}) + if err == nil { + err = KeepRevision(ctx, ix.Revisions, rev, objects) + } + if err != nil { + t.Fatal(err) + } + } + keep(1, "fp1") + if held, err := ix.SeedEntryHeld(ctx, "avahi", "fp1"); err != nil || !held { + t.Errorf("the entry at its fingerprint: %v, %v", held, err) + } + if in, err := ix.ModuleInNewestView(ctx, "avahi"); err != nil || !in { + t.Errorf("the module in the view: %v, %v", in, err) + } + if in, _ := ix.ModuleInNewestView(ctx, "gone"); in { + t.Error("a module the view does not hold is in it") + } + keep(2, "fp2") + if held, _ := ix.SeedEntryHeld(ctx, "avahi", "fp1"); held { + t.Error("a candidate the newest revision replaced is still held") + } + }) + t.Run("a branch is watched while a registration names it", func(t *testing.T) { + g := newRegRig() + ix := &Indexer{Registry: g.r} + raw, _ := json.Marshal(meshcatalog.Registration{Module: "avahi", Branches: []string{"main", "feat"}}) + if err := g.r.Registrations.Create(ctx, "avahi", raw); err != nil { + t.Fatal(err) + } + if w, err := ix.BranchWatched(ctx, TrunkBranch); err != nil || !w { + t.Errorf("the trunk: %v, %v", w, err) + } + if w, _ := ix.BranchWatched(ctx, "feat"); !w { + t.Error("a branch a registration names is not watched") + } + if w, _ := ix.BranchWatched(ctx, "other"); w { + t.Error("a branch no registration names is watched") + } + }) +} + +// A-190: a call that ends with its ask closed is not said unanswered: no finding, and the fill and the seed go on. +func TestAClosedAskIsNotUnanswered(t *testing.T) { + ctx := context.Background() + closed := fmt.Errorf("steps: %w", ErrAskClosed) + ix := &Indexer{Findings: newFindings(), Heard: newMem(), Revisions: newMemObjects(1 << 20), Versions: newMem()} + rev, objects, _ := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: 1, Kind: "trunk"}, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Fp: "fp", Manifest: map[string]any{}, + Artifacts: []meshcatalog.Artifact{{Ref: "avahi:program", Language: "go"}}}}) + if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil { + t.Fatal(err) + } + defs := map[string]seatdef.Definition{"node-build-go": {Acts: []seatdef.ActingRule{{Step: seatdef.StepBuild, On: "go"}}}} + if _, err := ix.SeedModulesFrom(ctx, defs, func(context.Context, string, nodebuildgo.StepsInput) (nodebuildgo.StepsAnswer, error) { + return nodebuildgo.StepsAnswer{}, closed + }); err != nil || len(ix.stepsSince) != 0 { + t.Errorf("a closed steps ask counted unanswered: %v %v", err, ix.stepsSince) + } + if err := ix.FillVersions(ctx, func(context.Context, string, string) (BuilderVersionsAnswer, error) { + return BuilderVersionsAnswer{}, fmt.Errorf("builder_versions: %w", ErrAskClosed) + }); errors.Is(err, ErrAskClosed) || (err != nil && strings.Contains(err.Error(), "is not answered")) { + t.Errorf("a closed versions ask is unanswered: %v", err) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 3005971b..22a18610 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -1490,7 +1490,7 @@ func (ix *Indexer) TakeInMerge(ctx context.Context, repository, merge string, se r.Answer = "this indexer does not ask the forge bridge" return r } - data, err := ix.Message(ctx, repository, merge, 0) + data, err := ix.Message(forTakeIn(ctx, repository, merge), repository, merge, 0) if err != nil { r.Answer = fmt.Sprintf("the forge bridge could not answer for the merge of %s at %s: %v", repository, merge, err) return r diff --git a/modules/mesh-delivery-indexer/internal/index/seed_steps.go b/modules/mesh-delivery-indexer/internal/index/seed_steps.go index c945beb8..06b69a88 100644 --- a/modules/mesh-delivery-indexer/internal/index/seed_steps.go +++ b/modules/mesh-delivery-indexer/internal/index/seed_steps.go @@ -2,6 +2,7 @@ package index import ( "context" + "errors" "fmt" "maps" "path" @@ -91,6 +92,7 @@ func (ix *Indexer) SeedModulesFrom(ctx context.Context, defs map[string]seatdef. Fetch: &nodebuildgo.Fetch{Repository: e.Repository, Commit: e.Commit, Dir: e.Path, Paths: namedPaths(e, name), Packages: maps.Clone(e.Packages)}} ans, err := steps(ctx, seat, in) switch { + case errors.Is(err, ErrAskClosed): // the entry is no longer held: not said unanswered (A-190) case err != nil: unanswered[seat] = err case ans.Error != "": diff --git a/modules/mesh-delivery-indexer/internal/index/source.go b/modules/mesh-delivery-indexer/internal/index/source.go index 739ff2e0..19e75efd 100644 --- a/modules/mesh-delivery-indexer/internal/index/source.go +++ b/modules/mesh-delivery-indexer/internal/index/source.go @@ -216,7 +216,7 @@ func (ix *Indexer) CatchUp(ctx context.Context) TakenIn { // mergedAt is the forge bridge's merged message of a repository's head. func (ix *Indexer) mergedAt(ctx context.Context, repository, head string) (forgemsg.Forge, error) { - data, err := ix.Message(ctx, repository, head, 0) + data, err := ix.Message(forTakeIn(ctx, repository, head), repository, head, 0) if err != nil { return forgemsg.Forge{}, fmt.Errorf("the forge bridge could not answer: %w", err) } diff --git a/modules/mesh-delivery-indexer/internal/index/versions_fill.go b/modules/mesh-delivery-indexer/internal/index/versions_fill.go index 9f520236..e2b960b5 100644 --- a/modules/mesh-delivery-indexer/internal/index/versions_fill.go +++ b/modules/mesh-delivery-indexer/internal/index/versions_fill.go @@ -46,7 +46,7 @@ type BuilderVersionsAnswer struct { } // VersionsAsker asks builder_versions for a name's versions (a kept call, ADR 0668 §1, §2). -type VersionsAsker func(ctx context.Context, name string) (BuilderVersionsAnswer, error) +type VersionsAsker func(ctx context.Context, module, name string) (BuilderVersionsAnswer, error) // versionNameOf is the name the builder numbers an artifact under: its package's name in its ecosystem, else // / (images, bundles and archives). @@ -97,7 +97,10 @@ func (ix *Indexer) FillVersions(ctx context.Context, ask VersionsAsker) error { per := []found1{} for _, a := range e.Artifacts { name := versionNameOf(e.Module, a) - ans, err := ask(ctx, name) + ans, err := ask(ctx, e.Module, name) + if errors.Is(err, ErrAskClosed) { + continue // the module left the view: nothing waits on its versions (A-190) + } if err != nil { unanswered = err continue diff --git a/modules/mesh-delivery-indexer/internal/index/versions_fill_edges.go b/modules/mesh-delivery-indexer/internal/index/versions_fill_edges.go index 7a2268e9..9a77267d 100644 --- a/modules/mesh-delivery-indexer/internal/index/versions_fill_edges.go +++ b/modules/mesh-delivery-indexer/internal/index/versions_fill_edges.go @@ -58,7 +58,7 @@ func (ix *Indexer) filledEdges(ctx context.Context, entry meshcatalog.CatalogEnt } ans, ok := answers[name] if !ok { - a, err := ask(ctx, name) + a, err := ask(ctx, module, name) if err != nil { continue } diff --git a/modules/mesh-delivery-indexer/internal/index/versions_test.go b/modules/mesh-delivery-indexer/internal/index/versions_test.go index 9acc013b..f81918b5 100644 --- a/modules/mesh-delivery-indexer/internal/index/versions_test.go +++ b/modules/mesh-delivery-indexer/internal/index/versions_test.go @@ -115,7 +115,7 @@ func TestTheFirstFillOfTheVersionsComesFromTheBuildersPublished(t *testing.T) { t.Fatal(err) } up := false - ask := func(_ context.Context, name string) (BuilderVersionsAnswer, error) { + ask := func(_ context.Context, _, name string) (BuilderVersionsAnswer, error) { if !up { return BuilderVersionsAnswer{}, errors.New("no responders") } @@ -457,7 +457,7 @@ func TestTheFirstFillGivesTheNewestVersionsTheirFilledEdges(t *testing.T) { if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil { t.Fatal(err) } - ask := func(_ context.Context, name string) (BuilderVersionsAnswer, error) { + ask := func(_ context.Context, _, name string) (BuilderVersionsAnswer, error) { pub := func(v, d string, at time.Time) BuilderVersion { return BuilderVersion{Version: v, State: "published", Digest: d, Revision: "9", At: at} }