From 756a3813e2c7eb16f06e161d4f865b0246ce27c8 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 13:32:41 +0100 Subject: [PATCH] The indexer says manifest-accepted, with the bus-state entries whole, for a newly accepted verdict (issue 1966 D3) --- .../cmd/mesh-delivery-indexer/main.go | 2 +- .../cmd/mesh-delivery-indexer/verdicts.go | 26 +++++++++++++++++++ .../internal/index/candidates.go | 3 +++ .../internal/index/verdicts.go | 8 +++++- .../internal/index/verdicts_test.go | 17 ++++++++++++ .../mesh-delivery-indexer/manifest_test.go | 7 ++--- modules/mesh-delivery-indexer/module.json | 3 ++- 7 files changed, 60 insertions(+), 6 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 fdbaabb0..04bb7252 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -297,7 +297,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // fingerprint takes the version line. packages := process.AwaitBucket(ctx, js, conn, delivery.PackagesBucket, "no package is published, so every fingerprint takes the version line", logger.Printf) ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, - Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(js, cred, logger.Printf), + Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(js, cred, logger.Printf), SayAccepted: sayAccepted(js, cred, logger.Printf), Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{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)} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go index 9bcd6ee3..79e1b316 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go @@ -36,3 +36,29 @@ func sayRefused(js jetstream.JetStream, cred process.Account, logf func(string, } } } + +// sayAccepted says the indexer's event manifest-accepted (issue 1966 D3, novox/hq ADR 0595 §4): +// mesh.mod..event.manifest-accepted, carrying the module's bus-state entries whole, with the digest and judge as +// the bus's dedup id (delivery.ManifestAccepted.EventID), so one acceptance is said once. +func sayAccepted(js jetstream.JetStream, cred process.Account, logf func(string, ...any)) func(context.Context, delivery.ManifestAccepted) { + return func(ctx context.Context, a delivery.ManifestAccepted) { + data, err := json.Marshal(a) + if err != nil { + logf("judge: %s for %s is not said: %v", delivery.ManifestAcceptedEvent, a.Module, err) + return + } + msg := nats.NewMsg(cred.EventSubject(delivery.ManifestAcceptedEvent)) + id := a.EventID() + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", cred.Module) + msg.Header.Set("x-node", cred.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + if _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)); err != nil { + logf("judge: %s", fmt.Errorf("saying %s for %s: %w", delivery.ManifestAcceptedEvent, a.Module, err)) + } + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 68e0bb65..95b4dde3 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -58,6 +58,9 @@ type Indexer struct { JudgeID string // SayRefused says the event manifest-refused for a refusal newly written; nil says nothing. SayRefused func(ctx context.Context, r delivery.ManifestRefused) + // SayAccepted says the event manifest-accepted for an accepted verdict newly written whose manifest asks for bus state, + // carrying the entries whole (verdicts.go; issue 1966 D3); nil says nothing. + SayAccepted func(ctx context.Context, a delivery.ManifestAccepted) // Packages is the packages state (delivery.PackagesBucket, packages.go; novox/hq ADR 0593 §2); nil publishes none, // and every fingerprint takes the version line. Packages Store diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts.go b/modules/mesh-delivery-indexer/internal/index/verdicts.go index 291562d6..383dd65e 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts.go @@ -74,7 +74,8 @@ func (ix *Indexer) JudgeAll(ctx context.Context, fresh map[string]bool) (Judged, raws[name] = []byte(kept.JSON) digests[name] = digest } - refused, unnamed := judge.Shelf(raws, ix.predefinedList(ctx), 0) + predefined := ix.predefinedList(ctx) + refused, unnamed := judge.Shelf(raws, predefined, 0) for _, p := range unnamed { ix.logf("judge: a problem between the manifests that no one module's removal ends, so it refuses none: %s", p) } @@ -109,6 +110,11 @@ func (ix *Indexer) JudgeAll(ctx context.Context, fresh map[string]bool) (Judged, if ix.SayRefused != nil { ix.SayRefused(ctx, delivery.RefusedOf(digests[name], v)) } + } else if ix.SayAccepted != nil { + // A manifest asking for bus state, newly accepted, is said whole to the bus module (issue 1966 D3, ADR 0595 §4). + if m, problems := judge.Manifest(raws[name], predefined); len(problems) == 0 && len(m.BusState) > 0 { + ix.SayAccepted(ctx, delivery.AcceptedOf(digests[name], v, m)) + } } } if err := ix.carryVerdict(ctx, name, digests[name], v); err != nil { diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts_test.go b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go index df57baa5..d90b852d 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts_test.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go @@ -118,6 +118,23 @@ func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) { if err := ix.writeOverlay(ctx, map[string]delivery.Module{"old": same}, nil, w); err != nil || !w.fresh[other] { t.Fatalf("a changed digest is not fresh: %v %v", w.fresh, err) } + // 1966/AC10: a newly accepted manifest asking for bus state is said once as manifest-accepted, carrying both entries + // whole; a verdict already held, and a manifest asking for none, say nothing. + var accepted []delivery.ManifestAccepted + ix.SayAccepted = func(_ context.Context, a delivery.ManifestAccepted) { accepted = append(accepted, a) } + moved := hold("moved", `{"module":"moved","version":"1","requires":[`+ + `{"name":"cache","storage":"key-value","template":"state","options":{"history":8}},`+ + `{"name":"audit","storage":"stream","template":"log","options":{"max-mib":64}}]}`) + for pass := 0; pass < 2; pass++ { + if _, err := ix.JudgeAll(ctx, map[string]bool{moved: true}); err != nil { + t.Fatal(err) + } + } + if len(accepted) != 1 || accepted[0].Module != "moved" || accepted[0].Digest != moved || accepted[0].Judge != "judge-2" || + len(accepted[0].BusState) != 2 || accepted[0].BusState[0].Name != "cache" || accepted[0].BusState[1].Template != "log" || + string(accepted[0].BusState[0].Options["history"]) != "8" { + t.Fatalf("said accepted %+v", accepted) + } // Without the verdicts state the indexer judges nothing. ix.Verdicts = nil if j, err := ix.JudgeAll(ctx, nil); err != nil || j.Judged != 0 { diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index c03a9702..2a3545da 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -84,9 +84,10 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build t.Errorf("module = %q", m.Module) } // 1126: placement is the seat mesh-placement's (ADR 0567): the indexer holds no placement state. ADR 0578 §1: it says - // a refused manifest as manifest-refused, its one event. - if !reflect.DeepEqual(m.Emits, []string{"manifest-refused"}) { - t.Errorf("emits = %v, want manifest-refused alone", m.Emits) + // a refused manifest as manifest-refused, and (issue 1966 D3, 1966/AC10) a newly accepted one asking for bus state as + // manifest-accepted. + if !reflect.DeepEqual(m.Emits, []string{"manifest-refused", "manifest-accepted"}) { + t.Errorf("emits = %v, want manifest-refused and manifest-accepted", m.Emits) } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}}; !reflect.DeepEqual(m.State, want) { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index e8c38862..350693cf 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -24,7 +24,8 @@ { "name": "packages", "history": 5 } ], "emits": [ - "manifest-refused" + "manifest-refused", + "manifest-accepted" ], "reads": [ "mesh-build-go.toolchains",