From 2819b05838bfe1edb439ba289f4c13f82863f686 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 15:10:32 +0100 Subject: [PATCH 1/5] The assigner's integration test answers the verbs' asks of the indexer's first views from the indexer's buckets (issue 1966) The verbs ask mesh-delivery-indexer.modules and .manifests through the tool runner since the asks moved off the indexer's state, but the test's stand-in answered every ask {ok:true}, which is no answer: propose failed on main with 'unexpected end of JSON input'. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-assigner/integration_test.go | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 6d77bef6..b4267224 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -1233,7 +1233,18 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) { t.Fatal("anchor in the form before was not converted") } } - verbs := newSeatTools(bedStore{b}, func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"ok":true}`), nil }) + // The verbs ask the indexer's first views of modules and manifests (ADR 0615 §1), as the tool runner answers them from the + // indexer's buckets; every other ask the controller's, answered ok. + verbs := newSeatTools(bedStore{b}, func(tool string, _ any) (json.RawMessage, error) { + if state, ok := strings.CutPrefix(tool, buckets.IndexerModule+"."); ok && indexerFirstViews[state] != "" { + a, err := views.FirstView(buckets.IndexerModule, state, bedSource{b.ctx, b.buckets[indexerFirstViews[state]]}, nil, "", time.Now()) + if err != nil { + return nil, err + } + return json.Marshal(map[string]any{"result": a}) + } + return json.RawMessage(`{"ok":true}`), nil + }) propose := func() mma.Proposal { t.Helper() got, err := verbs.doPropose(map[string]any{"why": "move web", "changes": `[{"op":"take-off","node":"anchor","module":"web"},` + From 03164e853ed0992b598d8f144e51952722bb0b79 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 15:15:26 +0100 Subject: [PATCH 2/5] The indexer writes no fingerprint from a packages state it did not read, keeps no import form beside a version line, and computes fingerprints again by hand (issue 1966) An indexer that started while the bus's storage was down read no published package, took every import for unpublished and wrote the version line over ~130 modules' fingerprints, which a later pass kept through the build's shortcut. A failed or empty read of the packages state, where held modules were computed from published ones, now fails the pass and says why; a version line clears the import form it was computed beside; indexer_fingerprints_again computes the import form again by hand, acting first and recording the hand act after, with a bound. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-indexer-tools/main.go | 4 + .../cmd/mesh-delivery-indexer/takein.go | 5 +- .../internal/index/candidates.go | 1 + .../internal/index/fingerprints.go | 153 ++++++++++++++++++ .../internal/index/fingerprints_test.go | 45 ++++++ .../internal/index/form.go | 3 + .../internal/index/imports_test.go | 106 ++++++++++++ .../internal/index/packages.go | 48 +++++- .../internal/index/registered.go | 2 + .../internal/index/takein.go | 5 +- .../internal/index/toolchains.go | 11 +- .../mesh-delivery-indexer/manifest_test.go | 4 +- modules/mesh-delivery-indexer/module.json | 3 +- 13 files changed, 379 insertions(+), 11 deletions(-) create mode 100644 modules/mesh-delivery-indexer/internal/index/fingerprints.go create mode 100644 modules/mesh-delivery-indexer/internal/index/fingerprints_test.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index c5ed725f..e4c89f9c 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -278,6 +278,10 @@ func main() { tools = append(tools, catalogTools()...) tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...) tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, keptRead{stdio.State(index.OwnBuildStateName)}, asks, time.Now)...) + tools = append(tools, index.FingerprintsTools(asks, time.Now, func(what, why string) error { + _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.FingerprintsAgainTool, Verb: "fingerprints-again"}) + return err + }, 10*time.Second)...) tools = append(tools, index.TryCandidateTools(keptRead{stdio.State("candidates")}, keptRead{stdio.State(index.TryRecoveryState)}, asks, time.Now)...) tools = append(tools, configReadersTool(stdio.NewWatchedStates()), busStateTool(keptSource), requirementsTool(keptSource), packageUsersTool(keptSource), seatDefinitionTool(keptSource), outOfStepTool(keptSource), diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index ec6dbca9..8585099d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -24,6 +24,7 @@ type takerIn interface { TakeInMerge(ctx context.Context, repository, merge string, seq uint64, by, why string) index.TakenIn DropTryCandidate(ctx context.Context, id, by, why string) index.TakenIn RecoverTriesByHand(ctx context.Context, by, why string) index.TakenIn + FingerprintsAgain(ctx context.Context, module string) index.TakenIn } // takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the @@ -42,7 +43,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream return } asked, r := "take-in", index.TakenIn{} - if a.RecoverTries { + if a.Fingerprints { + asked, r = "computing of fingerprints again", ix.FingerprintsAgain(ctx, a.Module) + } else if a.RecoverTries { asked, r = "recovery of the tries", ix.RecoverTriesByHand(ctx, a.By, a.Why) a.Module = "tries" } else if a.TryCandidate != "" && a.Drop { diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index ad3613ac..2b0d9e61 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -150,6 +150,7 @@ type Indexer struct { retrying atomic.Bool // a RetryUnread runs in the background (kickRetry) retryAgain atomic.Bool // an event asked for one more RetryUnread after the running one retryWG sync.WaitGroup // the background RetryUnread, for tests to wait on + fresh map[string]bool // modules whose import form is computed again by hand (FingerprintsAgain), "*" for every one; under recompute recompute sync.Mutex // one Recompute at a time untaken map[string]string // repository to the latest merge applied while the taken state was missing (taken.go) filling atomic.Bool // a FillTaken runs in the background (KickFill) diff --git a/modules/mesh-delivery-indexer/internal/index/fingerprints.go b/modules/mesh-delivery-indexer/internal/index/fingerprints.go new file mode 100644 index 00000000..f4fcdba1 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/fingerprints.go @@ -0,0 +1,153 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "slices" + "strings" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// FingerprintsAgainTool computes the fingerprints again now (issue 1966; the instruction "tools are how a module is spoken +// to"): the indexer reads the packages state, takes no held fingerprint for the build's, and writes the import form for +// every module whose imports are all published, as a merge's recompute would. For the module whose fingerprint a pass +// wrote while the packages state was unread (the version line over the import form) and that nothing else moves. +const FingerprintsAgainTool = "indexer_fingerprints_again" + +// FingerprintMove is one module's fingerprint before and after a recompute by hand. +type FingerprintMove struct { + Module string `json:"module"` + Old string `json:"old"` + New string `json:"new"` +} + +// FingerprintsAgain computes the fingerprints of module again ("all" or empty: every held module) from the published +// packages as they are read now, and answers each fingerprint that moved. A read of the packages state that failed +// writes nothing, and says so. +func (ix *Indexer) FingerprintsAgain(ctx context.Context, module string) TakenIn { + module = strings.TrimSpace(module) + if module == "" { + module = "all" + } + out := TakenIn{Module: module} + if ix.Modules == nil || ix.Manifests == nil { + out.Answer = "the indexer writes no fingerprints" + return out + } + ix.recompute.Lock() + defer ix.recompute.Unlock() + before, err := ix.fingerprintsHeld(ctx) + if err != nil { + out.Answer = "the modules state cannot be read: " + err.Error() + return out + } + if module != "all" { + if _, ok := before[module]; !ok { + out.Answer = "the indexer holds no fingerprint of " + module + return out + } + } + key := module + if module == "all" { + key = "*" + } + ix.fresh = map[string]bool{key: true} + err = ix.recomputeLocked(ctx, nil, nil) + ix.fresh = nil + if err != nil { + out.Answer = "nothing was written: " + err.Error() + return out + } + after, err := ix.fingerprintsHeld(ctx) + if err != nil { + out.Answer = "computed, but the modules state cannot be read back: " + err.Error() + return out + } + for _, m := range slices.Sorted(maps.Keys(after)) { + if module != "all" && m != module { + continue + } + if before[m] != after[m] { + out.Fingerprints = append(out.Fingerprints, FingerprintMove{Module: m, Old: before[m], New: after[m]}) + } + } + out.Stored = true + out.Answer = fmt.Sprintf("%d fingerprint(s) moved", len(out.Fingerprints)) + return out +} + +// fingerprintsHeld is each held module's fingerprint of the new form. +func (ix *Indexer) fingerprintsHeld(ctx context.Context) (map[string]string, error) { + keys, err := ix.Modules.Keys(ctx) + if err != nil { + return nil, err + } + out := map[string]string{} + for _, k := range keys { + value, _, found, err := ix.Modules.Get(ctx, k) + if err != nil { + return nil, err + } + if !found { + continue + } + if m, err := meshcatalogclient.DecodeModule(value); err == nil { + out[k] = m.Fp2 + } + } + return out, nil +} + +// FingerprintsTools is the tool indexer_fingerprints_again over ask (the takeins state). It acts first and says the hand +// act after: record is told what was done and answers an error when it could not be recorded, which the answer says and +// which never refuses or delays the recovery beyond bound. +func FingerprintsTools(ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(what, why string) error, bound time.Duration) []stdio.Tool { + return []stdio.Tool{{Name: FingerprintsAgainTool, + Description: "Compute the fingerprints of one module (or all) again now from the published packages as the indexer reads them: " + + "the import form is written for every module whose imports are all published, whatever fingerprint the module holds. For the " + + "fingerprints a pass wrote while the packages state was unread (the version line over the import form: builds then fail on an " + + "older package). Answers each module whose fingerprint moved, old and new; when the packages state cannot be read it writes " + + "nothing and says why. Acts first, then records the hand act through mesh-hand-acts; when that cannot be recorded the answer says so.", + Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{ + "module": map[string]any{"type": "string", "description": "the module's name, or all (the default)"}, + "why": map[string]any{"type": "string", "description": "why it is computed again now, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + module, _ := args["module"].(string) + why, _ := args["why"].(string) + module, why = strings.TrimSpace(module), strings.TrimSpace(why) + if why == "" { + return nil, errors.New("say why: a hand act is said in the journal") + } + if module == "" { + module = "all" + } + data, err := json.Marshal(TakeInAsk{V: 1, Module: module, Fingerprints: true, Why: why, By: stdio.CallerOf(args), At: now().UTC()}) + if err != nil { + return nil, err + } + a, err := ask("fingerprints."+module+"."+now().UTC().Format("20060102T150405.000000000"), data) + if err != nil { + return nil, err + } + res := map[string]any{"result": a.Result} + done := make(chan error, 1) + go func() { done <- record("computed the fingerprints of "+module+" again by hand", why) }() + wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8) + defer cancel() + select { + case err := <-done: + if err != nil { + res["hand-act"] = "not recorded: " + err.Error() + } + case <-wait.Done(): // a bound on the record, never on the recovery + res["hand-act"] = fmt.Sprintf("not recorded within %s: mesh-hand-acts did not answer", bound) + } + return res, nil + }}} +} diff --git a/modules/mesh-delivery-indexer/internal/index/fingerprints_test.go b/modules/mesh-delivery-indexer/internal/index/fingerprints_test.go new file mode 100644 index 00000000..82af8a7a --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/fingerprints_test.go @@ -0,0 +1,45 @@ +package index + +import ( + "errors" + "strings" + "testing" + "time" +) + +// indexer_fingerprints_again acts first and records the hand act after: a record that fails, or never answers, is said in +// the answer and never refuses or holds the recovery (mesh-hand-acts on a mesh with no ledger answers nothing). +func TestFingerprintsAgainActsBeforeItRecordsAndSaysAFailedRecord(t *testing.T) { + for _, c := range []struct { + name string + record func(what, why string) error + said string + }{ + {"recorded", func(string, string) error { return nil }, ""}, + {"record refused", func(string, string) error { return errors.New("no ledger") }, "not recorded: no ledger"}, + {"record never answers", func(string, string) error { select {} }, "not recorded within"}, + } { + t.Run(c.name, func(t *testing.T) { + var asked []string + ask := func(key string, _ []byte) (TakeInAsk, error) { + asked = append(asked, key) + return TakeInAsk{Result: &TakenIn{Module: "all", Stored: true}}, nil + } + tool := FingerprintsTools(ask, time.Now, c.record, 10*time.Millisecond)[0] + if _, err := tool.Run(map[string]any{"module": "avahi"}); err == nil { + t.Fatal("a hand act with no why was done") + } + if len(asked) != 0 { + t.Fatalf("asked the indexer without a why: %v", asked) + } + got, err := tool.Run(map[string]any{"module": "avahi", "why": "the packages read came back empty"}) + if err != nil || len(asked) != 1 || !strings.HasPrefix(asked[0], "fingerprints.avahi.") { + t.Fatalf("run: %v, asked %v", err, asked) + } + said, _ := got.(map[string]any)["hand-act"].(string) + if !strings.Contains(said, c.said) || (c.said == "") != (said == "") { + t.Fatalf("hand-act %q, want one saying %q", said, c.said) + } + }) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/form.go b/modules/mesh-delivery-indexer/internal/index/form.go index 8fe58f73..150bdb51 100644 --- a/modules/mesh-delivery-indexer/internal/index/form.go +++ b/modules/mesh-delivery-indexer/internal/index/form.go @@ -57,6 +57,9 @@ func formOf(module, repository string, kept *meshcatalog.ModulesByModuleValueInp logf("%s: no published package holds %s, so its fingerprint takes the version line", module, strings.Join(r.Missing, ", ")) } } + // The fingerprint written here is the version line's, not the import form's: no import form is held for the inputs it was + // computed from, so a later pass whose import form computes the same does not take this one for the build's (issue 1966). + kept.ImportForm = "" others := fingerprint.OtherPackages(kept.Packages, repository) versions, unpublished := fingerprint.HeldVersionsOf(others, held) for p, v := range pins { // an accepted series' newest published version (acceptedSeries, DLV-I6) diff --git a/modules/mesh-delivery-indexer/internal/index/imports_test.go b/modules/mesh-delivery-indexer/internal/index/imports_test.go index 9ac2ff98..880cf400 100644 --- a/modules/mesh-delivery-indexer/internal/index/imports_test.go +++ b/modules/mesh-delivery-indexer/internal/index/imports_test.go @@ -2,6 +2,7 @@ package index import ( "context" + "errors" "strings" "testing" "time" @@ -270,3 +271,108 @@ func anAcceptedSeriesPinsItsNewest(t *testing.T) { t.Errorf("a trunk merge moved avahi off its accepted series' newest: %q %v", m.Fp2, m.Packages) } } + +// Issue 1966 (the indexer started while the bus's storage was down): a read of the packages state that failed, or that +// answered no published package where held modules were computed from published ones, writes no fingerprint (the version +// line over every import form is what an empty state computes); a version line written leaves no import form behind +// that a later pass takes for the build's; and indexer_fingerprints_again computes the import form again by hand. +func TestAnUnreadPackagesStateWritesNoFingerprintAndFingerprintsAreComputedAgain(t *testing.T) { + ctx := context.Background() + // avahi at the import form, pinned to mesh-sdk at c3, as the test above leaves it. + setup := func(t *testing.T) (*Indexer, *memStore, meshcatalog.ModulesByModuleValue) { + ix, mods := newFormIndexer(t, mod{"avahi", catalogueManifest(t, "avahi")}) + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + ix.Packages, ix.Files = newMem(), forge{c1: sdkFiles("a1", "b1"), c3: sdkFiles("a2", "b2")}.files + setPackages(t, mods, "avahi", []string{meshSDK}) + raw, err := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Repository: "novox/mesh-sdk", Path: ".", Commit: c1, None2: "a TypeScript package"}) + if err != nil { + t.Fatal(err) + } + mods.put("mesh-sdk", raw) + held, _ := ix.heldModules(ctx) + if err := ix.publish(ctx, "novox/mesh-sdk", c1, held, false); err != nil { + t.Fatal(err) + } + if err := ix.Recompute(ctx); err != nil { + t.Fatal(err) + } + first, _ := moduleOf(t, mods, "avahi") + if err := ix.ApplyBuildEnd(ctx, buildEnd(t, "avahi", first.Fp2, meshSDK, meshSDK+"/announce")); err != nil { + t.Fatal(err) + } + held, _ = ix.heldModules(ctx) + if err := ix.publish(ctx, "novox/mesh-sdk", c3, held, false); err != nil { + t.Fatal(err) + } + if err := ix.Recompute(ctx); err != nil { + t.Fatal(err) + } + m, _ := moduleOf(t, mods, "avahi") + if m.Fp2 == first.Fp2 || m.Fp2 != m.Inputs2.ImportForm { + t.Fatalf("setup: fp2 %q (was %q), import form %q; want the import form", m.Fp2, first.Fp2, m.Inputs2.ImportForm) + } + return ix, mods, m + } + t.Run("a pass whose read of the packages state failed or answered nothing writes no fingerprint", func(t *testing.T) { + for _, c := range []struct { + name string + bad func(ix *Indexer) + why string + }{ + {"the listing fails", func(ix *Indexer) { ix.Packages.(*memStore).keysErr = errors.New("jetstream not enabled") }, "cannot be listed"}, + {"the state answers no package", func(ix *Indexer) { ix.Packages = newMem() }, "answers no published package"}, + } { + t.Run(c.name, func(t *testing.T) { + ix, mods, _ := setup(t) + was, _ := moduleOf(t, mods, "avahi") + c.bad(ix) + err := ix.Recompute(ctx) + if err == nil || !strings.Contains(err.Error(), c.why) { + t.Fatalf("recompute with an unread packages state: err %v, want one saying %q", err, c.why) + } + if now, _ := moduleOf(t, mods, "avahi"); now.Fp2 != was.Fp2 { + t.Fatalf("a pass over an unread packages state moved avahi's fingerprint to %q (was %q)", now.Fp2, was.Fp2) + } + }) + } + }) + t.Run("a version line written keeps no import form, and a fingerprint computed again by hand is the import form", func(t *testing.T) { + ix, mods, good := setup(t) + // What the pass that read nothing wrote on novox: the version line's fingerprint, the import form still kept. + bad := good + bad.Fp2 = "src2:" + strings.Repeat("d", 64) + raw, err := meshcatalogclient.EncodeModule(bad) + if err != nil { + t.Fatal(err) + } + mods.put("avahi", raw) + if err := ix.Recompute(ctx); err != nil { + t.Fatal(err) + } + if now, _ := moduleOf(t, mods, "avahi"); now.Fp2 != bad.Fp2 { + t.Fatalf("the held version line is not kept by the build's shortcut (fp2 %q); the case the tool recovers has changed", now.Fp2) + } + r := ix.FingerprintsAgain(ctx, "avahi") + if !r.Stored || len(r.Fingerprints) != 1 || r.Fingerprints[0].Old != bad.Fp2 || r.Fingerprints[0].New != good.Fp2 { + t.Fatalf("fingerprints again: %+v, want avahi from %q to %q", r, bad.Fp2, good.Fp2) + } + if now, _ := moduleOf(t, mods, "avahi"); now.Fp2 != good.Fp2 || ix.fresh != nil { + t.Fatalf("avahi holds %q after it, want %q (fresh %v)", now.Fp2, good.Fp2, ix.fresh) + } + if r := ix.FingerprintsAgain(ctx, "nothing"); r.Stored { + t.Fatalf("fingerprints again of a module not held answered stored: %+v", r) + } + }) + t.Run("formOf clears the import form when it writes the version line", func(t *testing.T) { + for _, c := range []struct { + name string + imports []string + }{{"an import no package holds", []string{meshSDK + "/announce"}}, {"no import reported", nil}} { + kept := meshcatalog.ModulesByModuleValueInputs2{Tree: "t", Imports: c.imports, ImportForm: "src2:" + strings.Repeat("a", 64)} + f := formOf("avahi", "novox/mesh-catalog", &kept, nil, meshcatalog.ModulesByModuleValue{}, nil, nil, nil, nil) + if kept.ImportForm != "" { + t.Errorf("%s: kept import form %q after the version line %q", c.name, kept.ImportForm, f.fp) + } + } + }) +} diff --git a/modules/mesh-delivery-indexer/internal/index/packages.go b/modules/mesh-delivery-indexer/internal/index/packages.go index 956a6fa6..04839ee5 100644 --- a/modules/mesh-delivery-indexer/internal/index/packages.go +++ b/modules/mesh-delivery-indexer/internal/index/packages.go @@ -31,21 +31,34 @@ import ( // publishedAll is every value of the packages state, by Go module path; empty with no state or one that does not answer. func (ix *Indexer) publishedAll(ctx context.Context) map[string]meshcatalog.PackagesByModuleValue { + out, err := ix.publishedRead(ctx) + if err != nil { + ix.logf("the packages state cannot be listed, so every fingerprint takes the version line: %v", err) + } + return out +} + +// publishedRead is publishedAll with a read that failed said as an error, never as an empty map: a pass that takes an +// unread state for "no package is published" writes the version line over every module's fingerprint (issue 1966, the +// indexer started while the bus's storage was down). No state wired is no error: nothing is published then. +func (ix *Indexer) publishedRead(ctx context.Context) (map[string]meshcatalog.PackagesByModuleValue, error) { out := map[string]meshcatalog.PackagesByModuleValue{} if ix.Packages == nil { - return out + return out, nil } keys, err := ix.Packages.Keys(ctx) if err != nil { - ix.logf("the packages state cannot be listed, so every fingerprint takes the version line: %v", err) - return out + return out, fmt.Errorf("the packages state cannot be listed: %w", err) } for _, key := range keys { if IsSeriesKey(key) { continue // a package's series (series.go), not a Go module's digests } value, _, found, err := ix.Packages.Get(ctx, key) - if err != nil || !found { + if err != nil { + return out, fmt.Errorf("packages.%s cannot be read: %w", key, err) + } + if !found { continue } p, err := DecodePublishedPackages(value) @@ -55,7 +68,32 @@ func (ix *Indexer) publishedAll(ctx context.Context) map[string]meshcatalog.Pack } out[p.Module] = p } - return out + return out, nil +} + +// publishedForPass is the packages state a recompute computes fingerprints from, or why it must write none: a read that +// failed, or that answered no published module where the held modules show packages were published before (a module +// whose fingerprint was computed from the import form, or that holds resolved package versions). The first start of a +// mesh, whose held modules show none, computes from an empty state as before. +func (ix *Indexer) publishedForPass(ctx context.Context, held map[string]meshcatalog.ModulesByModuleValue) (map[string]meshcatalog.PackagesByModuleValue, error) { + published, err := ix.publishedRead(ctx) + if err != nil { + return nil, err + } + if len(published) > 0 { + return published, nil + } + var had []string + for key, m := range held { + if m.Inputs2 != nil && m.Inputs2.ImportForm != "" { + had = append(had, key) + } + } + if len(had) > 0 { + slices.Sort(had) + return nil, fmt.Errorf("the packages state answers no published package, though %d held modules (%s ...) were computed from published ones", len(had), had[0]) + } + return published, nil } // reached are, by repository, the mesh Go module paths the held modules' go.sum names and the imports their builds diff --git a/modules/mesh-delivery-indexer/internal/index/registered.go b/modules/mesh-delivery-indexer/internal/index/registered.go index 4cc25ff4..2f30556c 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered.go +++ b/modules/mesh-delivery-indexer/internal/index/registered.go @@ -220,6 +220,8 @@ type TakenIn struct { // and deleted. Handed []string `json:"handed,omitempty"` Dropped []string `json:"dropped,omitempty"` + // Fingerprints are the modules whose fingerprint a recompute by hand moved (FingerprintsAgain), each old and new. + Fingerprints []FingerprintMove `json:"fingerprints,omitempty"` } // TakeIn reads a registered module's manifest now, whatever was tried before, and writes it as a merge would (the tool diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 56112792..b20c7006 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -56,6 +56,9 @@ type TakeInAsk struct { Sequence uint64 `json:"sequence,omitempty"` // TryCandidate asks the candidate of this try's group to be dropped (indexer_drop_try_candidate), trycandidates.go. TryCandidate string `json:"try-candidate,omitempty"` + // Fingerprints asks the fingerprints of Module ("all": every module) to be computed again now, from the published + // packages as they are read (indexer_fingerprints_again, fingerprints.go). + Fingerprints bool `json:"fingerprints,omitempty"` // RecoverTries asks the open tries' candidates to be recovered now (indexer_recover_tries). RecoverTries bool `json:"recover-tries,omitempty"` Drop bool `json:"drop,omitempty"` @@ -77,7 +80,7 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) { if a.V != 1 { return a, fmt.Errorf("a takeins value of version %d, not 1", a.V) } - if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries { + if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints { return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try") } return a, nil diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go index 738427a5..884d84a0 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains.go @@ -223,7 +223,13 @@ func (ix *Indexer) recomputeLocked(ctx context.Context, overlay map[string]meshc all[key] = &known{key: key, cur: cur, manifest: m, languages: languagesOf(m), bases: basesOf(m), contexts: hasContext(m), held: before} } tools := ix.agreed() - published := ix.publishedAll(ctx) + published, err := ix.publishedForPass(ctx, held) + if err != nil { + // No module's fingerprint is written from a state that was not read: the version line over every import form is + // what an empty packages state would compute (issue 1966). The next pass, or indexer_fingerprints_again, computes them. + ix.logf("no fingerprint is written by this pass: %v", err) + return fmt.Errorf("no fingerprint is written: %w", err) + } series := ix.SeriesAll(ctx) // each package's series, the newest a user accepting it takes (DLV-I6) type result struct { fp, none string @@ -258,6 +264,9 @@ func (ix *Indexer) recomputeLocked(ctx context.Context, overlay map[string]meshc chain = append(chain, module) kept := *k.cur.Inputs2 kept.Bases, kept.Builder = nil, "" + if ix.fresh[module] || ix.fresh["*"] { // indexer_fingerprints_again: the import form is written again, the held one stands for nothing + kept.ImportForm = "" + } var missing []string for _, b := range k.bases { if r := of(b, chain); r.fp != "" { diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index fd6d2504..1519ab5f 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -175,8 +175,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build want = append(want, "indexer_first_views", "indexer_first_view_again", index.ConfigReadersTool, index.BusStateTool, index.RequirementsTool, index.PackageUsersTool, index.SayAcceptedTool, // DLV-F3r, DLV-I10, DLV-I11, DLV-I7, DLV-I8, DLV-I9c index.SeatDefinitionTool, index.SeatDefinitionsAgainTool, index.OutOfStepTool, // DLV-I19 index.HoldsTool, index.EngineReadsTool, // ADR 0632 §2: the mesh-wide holds and the engines' reads they weigh - index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TryCandidatesTool, index.DropTryCandidateTool, index.RecoverTriesTool, index.TakeInMergeTool) // 1585/T12, T13 (ADR 0633 §5) - if !reflect.DeepEqual(m.Tools, want) || len(want) != 38 { + index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TryCandidatesTool, index.DropTryCandidateTool, index.RecoverTriesTool, index.TakeInMergeTool, index.FingerprintsAgainTool) // issue 1966; 1585/T12, T13 (ADR 0633 §5) + if !reflect.DeepEqual(m.Tools, want) || len(want) != 39 { t.Errorf("tools = %v, want %v", m.Tools, want) } // Every read tool of the indexer a reader of this repository invokes is one it serves (the checker's seventh first diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index f36f1ae7..67f8e7a5 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -54,7 +54,8 @@ "indexer_try_candidates", "indexer_drop_try_candidate", "indexer_recover_tries", - "indexer_take_in_merge" + "indexer_take_in_merge", + "indexer_fingerprints_again" ], "consumes": [ "go-package-registry.published", From 3a63813b3d7595e78f3af647af4aa629f47f56a6 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 15:24:30 +0100 Subject: [PATCH 3/5] Fingerprints computed again by hand run a merge's recompute; the tests join existing ones (issue 1966, review round 1) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../internal/index/candidates.go | 17 +++---- .../internal/index/fingerprints.go | 22 +++++++-- .../internal/index/fingerprints_test.go | 45 ------------------- .../internal/index/imports_test.go | 9 +++- .../internal/index/pending_test.go | 34 ++++++++++++++ .../internal/index/toolchains.go | 2 +- 6 files changed, 70 insertions(+), 59 deletions(-) delete mode 100644 modules/mesh-delivery-indexer/internal/index/fingerprints_test.go diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 2b0d9e61..0849ba47 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -150,14 +150,15 @@ type Indexer struct { retrying atomic.Bool // a RetryUnread runs in the background (kickRetry) retryAgain atomic.Bool // an event asked for one more RetryUnread after the running one retryWG sync.WaitGroup // the background RetryUnread, for tests to wait on - fresh map[string]bool // modules whose import form is computed again by hand (FingerprintsAgain), "*" for every one; under recompute - recompute sync.Mutex // one Recompute at a time - untaken map[string]string // repository to the latest merge applied while the taken state was missing (taken.go) - filling atomic.Bool // a FillTaken runs in the background (KickFill) - fillAgain atomic.Bool // a kick arrived while it ran - atHeads map[string]map[string]meshcatalog.PackagesByModuleValue // a member head's published Go modules, by repository@head (candidatePublished) - saidAt map[string]string // the head of a repository whose failed read was said, by repository: said once per head - failedAt map[string]string // the head of a repository whose tree lacks a directory asked, by repository: not asked again until the head changes + fresh map[string]bool // modules whose import form is computed again by hand (FingerprintsAgain), "*" for every one; under freshMu + freshMu sync.Mutex + recompute sync.Mutex // one Recompute at a time + untaken map[string]string // repository to the latest merge applied while the taken state was missing (taken.go) + filling atomic.Bool // a FillTaken runs in the background (KickFill) + fillAgain atomic.Bool // a kick arrived while it ran + atHeads map[string]map[string]meshcatalog.PackagesByModuleValue // a member head's published Go modules, by repository@head (candidatePublished) + saidAt map[string]string // the head of a repository whose failed read was said, by repository: said once per head + failedAt map[string]string // the head of a repository whose tree lacks a directory asked, by repository: not asked again until the head changes // OwnBuild answers the fingerprint of the build this program runs, from the node-engine's seat; nil knows none. OwnBuild func(ctx context.Context) (string, error) underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go) diff --git a/modules/mesh-delivery-indexer/internal/index/fingerprints.go b/modules/mesh-delivery-indexer/internal/index/fingerprints.go index f4fcdba1..a70ddf57 100644 --- a/modules/mesh-delivery-indexer/internal/index/fingerprints.go +++ b/modules/mesh-delivery-indexer/internal/index/fingerprints.go @@ -40,8 +40,6 @@ func (ix *Indexer) FingerprintsAgain(ctx context.Context, module string) TakenIn out.Answer = "the indexer writes no fingerprints" return out } - ix.recompute.Lock() - defer ix.recompute.Unlock() before, err := ix.fingerprintsHeld(ctx) if err != nil { out.Answer = "the modules state cannot be read: " + err.Error() @@ -57,13 +55,22 @@ func (ix *Indexer) FingerprintsAgain(ctx context.Context, module string) TakenIn if module == "all" { key = "*" } + ix.freshMu.Lock() ix.fresh = map[string]bool{key: true} - err = ix.recomputeLocked(ctx, nil, nil) + ix.freshMu.Unlock() + // The path of a merge's recompute: the modules, then the candidates their fingerprints reach (recomputeWith), then the + // pending merges a changed fingerprint settles (PendingAgain, as after a toolchain report). + err = ix.recomputeWith(ctx, nil, nil) + ix.freshMu.Lock() ix.fresh = nil + ix.freshMu.Unlock() if err != nil { out.Answer = "nothing was written: " + err.Error() return out } + if err := ix.PendingAgain(ctx); err != nil { + ix.logf("the pending merges are not evaluated again after the fingerprints were computed again: %v", err) + } after, err := ix.fingerprintsHeld(ctx) if err != nil { out.Answer = "computed, but the modules state cannot be read back: " + err.Error() @@ -82,6 +89,13 @@ func (ix *Indexer) FingerprintsAgain(ctx context.Context, module string) TakenIn return out } +// isFresh says the module's import form is computed again by hand in this pass. +func (ix *Indexer) isFresh(module string) bool { + ix.freshMu.Lock() + defer ix.freshMu.Unlock() + return ix.fresh[module] || ix.fresh["*"] +} + // fingerprintsHeld is each held module's fingerprint of the new form. func (ix *Indexer) fingerprintsHeld(ctx context.Context) (map[string]string, error) { keys, err := ix.Modules.Keys(ctx) @@ -113,7 +127,7 @@ func FingerprintsTools(ask func(key string, value []byte) (TakeInAsk, error), no "the import form is written for every module whose imports are all published, whatever fingerprint the module holds. For the " + "fingerprints a pass wrote while the packages state was unread (the version line over the import form: builds then fail on an " + "older package). Answers each module whose fingerprint moved, old and new; when the packages state cannot be read it writes " + - "nothing and says why. Acts first, then records the hand act through mesh-hand-acts; when that cannot be recorded the answer says so.", + "nothing and says why. Runs as a merge's recompute does (the modules, the candidates they reach, the pending merges they settle). Acts first, then records the hand act through mesh-hand-acts; when that cannot be recorded the answer says so.", Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{ "module": map[string]any{"type": "string", "description": "the module's name, or all (the default)"}, "why": map[string]any{"type": "string", "description": "why it is computed again now, in plain words"}}}, diff --git a/modules/mesh-delivery-indexer/internal/index/fingerprints_test.go b/modules/mesh-delivery-indexer/internal/index/fingerprints_test.go deleted file mode 100644 index 82af8a7a..00000000 --- a/modules/mesh-delivery-indexer/internal/index/fingerprints_test.go +++ /dev/null @@ -1,45 +0,0 @@ -package index - -import ( - "errors" - "strings" - "testing" - "time" -) - -// indexer_fingerprints_again acts first and records the hand act after: a record that fails, or never answers, is said in -// the answer and never refuses or holds the recovery (mesh-hand-acts on a mesh with no ledger answers nothing). -func TestFingerprintsAgainActsBeforeItRecordsAndSaysAFailedRecord(t *testing.T) { - for _, c := range []struct { - name string - record func(what, why string) error - said string - }{ - {"recorded", func(string, string) error { return nil }, ""}, - {"record refused", func(string, string) error { return errors.New("no ledger") }, "not recorded: no ledger"}, - {"record never answers", func(string, string) error { select {} }, "not recorded within"}, - } { - t.Run(c.name, func(t *testing.T) { - var asked []string - ask := func(key string, _ []byte) (TakeInAsk, error) { - asked = append(asked, key) - return TakeInAsk{Result: &TakenIn{Module: "all", Stored: true}}, nil - } - tool := FingerprintsTools(ask, time.Now, c.record, 10*time.Millisecond)[0] - if _, err := tool.Run(map[string]any{"module": "avahi"}); err == nil { - t.Fatal("a hand act with no why was done") - } - if len(asked) != 0 { - t.Fatalf("asked the indexer without a why: %v", asked) - } - got, err := tool.Run(map[string]any{"module": "avahi", "why": "the packages read came back empty"}) - if err != nil || len(asked) != 1 || !strings.HasPrefix(asked[0], "fingerprints.avahi.") { - t.Fatalf("run: %v, asked %v", err, asked) - } - said, _ := got.(map[string]any)["hand-act"].(string) - if !strings.Contains(said, c.said) || (c.said == "") != (said == "") { - t.Fatalf("hand-act %q, want one saying %q", said, c.said) - } - }) - } -} diff --git a/modules/mesh-delivery-indexer/internal/index/imports_test.go b/modules/mesh-delivery-indexer/internal/index/imports_test.go index 880cf400..3d61e5f9 100644 --- a/modules/mesh-delivery-indexer/internal/index/imports_test.go +++ b/modules/mesh-delivery-indexer/internal/index/imports_test.go @@ -9,6 +9,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" ) @@ -52,6 +53,7 @@ func (m *missingDir) Error() string { return "no directory " + m.dir } func TestImportsReportedMoveTheFingerprintOnlyWithAnImportedPackage(t *testing.T) { t.Run("a build of an older fingerprint keeps no imports", buildOfAnOlderFingerprintKeepsNoImports) t.Run("a change of a sibling directory alone moves its importer", aSiblingDirectoryAloneMovesItsImporter) + t.Run("an unread packages state writes no fingerprint, and they are computed again by hand (issue 1966)", anUnreadPackagesStateWritesNoFingerprintAndFingerprintsAreComputedAgain) t.Run("an accepted series' newest published version pins it, by that version's contents", anAcceptedSeriesPinsItsNewest) ctx := context.Background() ix, mods := newFormIndexer(t, mod{"avahi", catalogueManifest(t, "avahi")}) @@ -276,7 +278,7 @@ func anAcceptedSeriesPinsItsNewest(t *testing.T) { // answered no published package where held modules were computed from published ones, writes no fingerprint (the version // line over every import form is what an empty state computes); a version line written leaves no import form behind // that a later pass takes for the build's; and indexer_fingerprints_again computes the import form again by hand. -func TestAnUnreadPackagesStateWritesNoFingerprintAndFingerprintsAreComputedAgain(t *testing.T) { +func anUnreadPackagesStateWritesNoFingerprintAndFingerprintsAreComputedAgain(t *testing.T) { ctx := context.Background() // avahi at the import form, pinned to mesh-sdk at c3, as the test above leaves it. setup := func(t *testing.T) (*Indexer, *memStore, meshcatalog.ModulesByModuleValue) { @@ -359,6 +361,11 @@ func TestAnUnreadPackagesStateWritesNoFingerprintAndFingerprintsAreComputedAgain if now, _ := moduleOf(t, mods, "avahi"); now.Fp2 != good.Fp2 || ix.fresh != nil { t.Fatalf("avahi holds %q after it, want %q (fresh %v)", now.Fp2, good.Fp2, ix.fresh) } + // It runs a merge's recompute, candidates included: a candidate store that cannot be listed fails it, said. + ix.Store, ix.previews = &memStore{keysErr: errors.New("no bus")}, map[string]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue{"k": {}} + if r := ix.FingerprintsAgain(ctx, "avahi"); r.Stored || !strings.Contains(r.Answer, "listing the candidates") { + t.Fatalf("fingerprints again over candidates that cannot be listed: %+v", r) + } if r := ix.FingerprintsAgain(ctx, "nothing"); r.Stored { t.Fatalf("fingerprints again of a module not held answered stored: %+v", r) } diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index e7d52e72..0ecaef44 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -135,6 +135,40 @@ func sha(c byte) string { return strings.Repeat(string(c), 40) } func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) ctx := context.Background() + t.Run("indexer_fingerprints_again acts before it records the hand act and says a failed record", func(t *testing.T) { // issue 1966 + for _, c := range []struct { + name string + record func(what, why string) error + said string + }{ + {"recorded", func(string, string) error { return nil }, ""}, + {"record refused", func(string, string) error { return errors.New("no ledger") }, "not recorded: no ledger"}, + {"record never answers", func(string, string) error { select {} }, "not recorded within"}, + } { + t.Run(c.name, func(t *testing.T) { + var asked []string + ask := func(key string, _ []byte) (TakeInAsk, error) { + asked = append(asked, key) + return TakeInAsk{Result: &TakenIn{Module: "all", Stored: true}}, nil + } + tool := FingerprintsTools(ask, time.Now, c.record, 10*time.Millisecond)[0] + if _, err := tool.Run(map[string]any{"module": "avahi"}); err == nil { + t.Fatal("a hand act with no why was done") + } + if len(asked) != 0 { + t.Fatalf("asked the indexer without a why: %v", asked) + } + got, err := tool.Run(map[string]any{"module": "avahi", "why": "the packages read came back empty"}) + if err != nil || len(asked) != 1 || !strings.HasPrefix(asked[0], "fingerprints.avahi.") { + t.Fatalf("run: %v, asked %v", err, asked) + } + said, _ := got.(map[string]any)["hand-act"].(string) + if !strings.Contains(said, c.said) || (c.said == "") != (said == "") { + t.Fatalf("hand-act %q, want one saying %q", said, c.said) + } + }) + } + }) t.Run("T6 a merge while the own build is unknown writes nothing", func(t *testing.T) { b := newTakeInBed(t) b.ix.ownFP, b.ix.ownFixed = "", false // this process is started again and asks the engine, which does not answer diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go index 884d84a0..71044311 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains.go @@ -264,7 +264,7 @@ func (ix *Indexer) recomputeLocked(ctx context.Context, overlay map[string]meshc chain = append(chain, module) kept := *k.cur.Inputs2 kept.Bases, kept.Builder = nil, "" - if ix.fresh[module] || ix.fresh["*"] { // indexer_fingerprints_again: the import form is written again, the held one stands for nothing + if ix.isFresh(module) { // indexer_fingerprints_again: the import form is written again, the held one stands for nothing kept.ImportForm = "" } var missing []string From 9073190646ece661039c6da50de459fa8179498e Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 15:40:44 +0100 Subject: [PATCH 4/5] A packages state not on the bus yet is nothing published, not a failed read (issue 1966, check of #191) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../mesh-delivery-indexer/internal/index/imports_test.go | 2 ++ modules/mesh-delivery-indexer/internal/index/packages.go | 7 +++++++ 2 files changed, 9 insertions(+) diff --git a/modules/mesh-delivery-indexer/internal/index/imports_test.go b/modules/mesh-delivery-indexer/internal/index/imports_test.go index 3d61e5f9..36a35559 100644 --- a/modules/mesh-delivery-indexer/internal/index/imports_test.go +++ b/modules/mesh-delivery-indexer/internal/index/imports_test.go @@ -11,6 +11,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-sdk/go/process" ) const meshSDK = "git.novox.be/novox/mesh-sdk/go" @@ -322,6 +323,7 @@ func anUnreadPackagesStateWritesNoFingerprintAndFingerprintsAreComputedAgain(t * why string }{ {"the listing fails", func(ix *Indexer) { ix.Packages.(*memStore).keysErr = errors.New("jetstream not enabled") }, "cannot be listed"}, + {"the state is not on the bus yet", func(ix *Indexer) { ix.Packages.(*memStore).keysErr = process.ErrNotYet }, "answers no published package"}, // a mesh with no import form computes as before: TestIndexerWritesModulesAtAMerge {"the state answers no package", func(ix *Indexer) { ix.Packages = newMem() }, "answers no published package"}, } { t.Run(c.name, func(t *testing.T) { diff --git a/modules/mesh-delivery-indexer/internal/index/packages.go b/modules/mesh-delivery-indexer/internal/index/packages.go index 04839ee5..09590432 100644 --- a/modules/mesh-delivery-indexer/internal/index/packages.go +++ b/modules/mesh-delivery-indexer/internal/index/packages.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "git.novox.be/novox/mesh-sdk/go/process" "maps" "path" "reflect" @@ -47,6 +48,9 @@ func (ix *Indexer) publishedRead(ctx context.Context) (map[string]meshcatalog.Pa return out, nil } keys, err := ix.Packages.Keys(ctx) + if errors.Is(err, process.ErrNotYet) { + return out, nil // the state is not on the bus yet: nothing is published, as before + } if err != nil { return out, fmt.Errorf("the packages state cannot be listed: %w", err) } @@ -55,6 +59,9 @@ func (ix *Indexer) publishedRead(ctx context.Context) (map[string]meshcatalog.Pa continue // a package's series (series.go), not a Go module's digests } value, _, found, err := ix.Packages.Get(ctx, key) + if errors.Is(err, process.ErrNotYet) { + continue + } if err != nil { return out, fmt.Errorf("packages.%s cannot be read: %w", key, err) } From efdfe1f755186fce84109a6172d23fd93ce0c665 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 15:43:52 +0100 Subject: [PATCH 5/5] A bucket the bus cannot reach is its error, not nothing published; only a bucket not found is not yet (issue 1966, review round 3) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-indexer/main.go | 32 +++++++++++++------ .../cmd/mesh-delivery-indexer/main_test.go | 21 ++++++++++++ .../internal/index/packages.go | 2 +- 3 files changed, 44 insertions(+), 11 deletions(-) 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 ddfa67b1..00ef9a3d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -52,7 +52,14 @@ func main() { // awaitedStore is a state bound when it appears (process.AwaitBucket) as the indexer's store: until then every call is // process.ErrNotYet. -type awaitedStore struct{ b *process.AwaitedBucket } +type awaitedStore struct { + b *process.AwaitedBucket + // js and name, when set, tell a bucket that does not exist from one that cannot be reached: the awaited bucket is nil for + // both (the SDK drops the bus's error), and only the first is "not yet". The packages state is read this way: an unreached + // state is no package published, and the version line written over every import form (issue 1966). + js process.BucketOpener + name string +} // expiringKeys is index.Bounds on a state whose keys take a time to live (key-ttl in the manifest): a key armed expires // at its end, which its watcher hears as a purge; one standing keeps its end. @@ -88,6 +95,11 @@ func (e expiringKeys) Disarm(ctx context.Context, key string) error { func (s awaitedStore) store() (kvStore, error) { kv := s.b.KV() if kv == nil { + if s.js != nil { + if _, err := s.js.KeyValue(context.Background(), s.name); err != nil && !errors.Is(err, jetstream.ErrBucketNotFound) { + return kvStore{}, fmt.Errorf("the bucket %s cannot be reached: %w", s.name, err) + } + } return kvStore{}, process.ErrNotYet } return kvStore{kv}, nil @@ -411,17 +423,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro seatBus := &seatBus{defs: awaitedSource{ctx, seatDefs}, marks: awaitedSource{ctx, marks}} seatReader, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf) readWhenBound(seatReader, seatDefs, marks) - ix := &index.Indexer{Findings: awaitedStore{findings}, OldSections: awaitedStore{oldSections}, Requirements: awaitedStore{requirements}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, - Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred), - Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf, + ix := &index.Indexer{Findings: awaitedStore{b: findings}, OldSections: awaitedStore{b: oldSections}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known, + Verdicts: awaitedStore{b: verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred), + Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), - ReadOpen: openVia(calls.Call), Packages: awaitedStore{packages}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), - OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{ownBuildState}, TryRecovery: awaitedStore{tryRecovery}, Graph: awaitedStore{graph}, - SeatDefinitions: awaitedStore{seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, + 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), // 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{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)}, + 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)}, SayTakeIn: sayTakeIn(js, cred)} // 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 @@ -483,7 +495,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The holds, new with ADR 0632 (issue 1790's kind), bound when they appear: until then a merge that would hold a // manifest is left for redelivery, said, rather than write what a counted engine cannot read. holds := process.AwaitBucket(ctx, js, conn, index.HoldsBucket, "a merge whose manifest every node reads, needing what not every counted node-engine states, is left for redelivery", logger.Printf) - ix.Holds = awaitedStore{holds} + ix.Holds = awaitedStore{b: holds} holds.Ready(func(jetstream.KeyValue) { ix.KickHolds(ctx) }) // Each standing hold's bound, a key that expires at its end (ADR 0632 §2, the indexer's own state): its expiry weighs // the holds again, which raises manifest-held; until the state is bound a hold past its bound is raised at its next @@ -696,7 +708,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) }) // What the indexer heard of FORGE (issue 1126 note 76) is bound when its state appears, as every state a delivery adds // (issue 1790): until then no message is written there, and the rest of the indexer runs. - heard := awaitedStore{process.AwaitBucket(ctx, js, conn, index.HeardBucket, "no FORGE message the indexer handles is written for indexer_heard", logger.Printf)} + heard := awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.HeardBucket, "no FORGE message the indexer handles is written for indexer_heard", logger.Printf)} // Each change of the states its readers read is said on its views family (ADR 0615 §5), from the mark each state was // said up to; a first view (the tools bundle) answers at that mark. startViews(ctx, js, cred, marks, 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 0c15f040..40e32bfe 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 @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "log" "strings" "testing" @@ -151,8 +152,28 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part) t.Errorf("forge/group's merged message read as a group's end: %q", logs.String()) } }) + // Issue 1966: a bucket the bus has not made is "not yet"; one it cannot reach (jetstream not enabled, no responders) is the + // bus's own error, which stops a pass instead of reading as nothing published. + t.Run("a bucket not found is not yet, one the bus cannot reach is its error", func(t *testing.T) { + for _, c := range []struct { + name string + err error + notYet bool + }{{"not found", jetstream.ErrBucketNotFound, true}, {"not reachable", jetstream.ErrJetStreamNotEnabled, false}} { + b := process.AwaitBucket(context.Background(), opener{c.err}, nil, "packages", "none", t.Logf) + _, err := awaitedStore{b: b, js: opener{c.err}, name: "packages"}.Keys(context.Background()) + if errors.Is(err, process.ErrNotYet) != c.notYet || err == nil || (!c.notYet && !errors.Is(err, c.err)) { + t.Errorf("%s: Keys answered %v, want not-yet %v", c.name, err, c.notYet) + } + } + }) } +// opener is a bus that answers every bucket with err. +type opener struct{ err error } + +func (o opener) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, o.err } + func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value) var logs bytes.Buffer logger := log.New(&logs, "", 0) diff --git a/modules/mesh-delivery-indexer/internal/index/packages.go b/modules/mesh-delivery-indexer/internal/index/packages.go index 09590432..5013979f 100644 --- a/modules/mesh-delivery-indexer/internal/index/packages.go +++ b/modules/mesh-delivery-indexer/internal/index/packages.go @@ -92,7 +92,7 @@ func (ix *Indexer) publishedForPass(ctx context.Context, held map[string]meshcat } var had []string for key, m := range held { - if m.Inputs2 != nil && m.Inputs2.ImportForm != "" { + if m.Inputs2 != nil && m.Inputs2.ImportForm != "" { // not Packages: a version line holds those too, with nothing published had = append(had, key) } }