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 e02d915..b8cdf7c 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -291,7 +291,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } defer calls.Close() + // The verdicts on manifests (ADR 0578 §1): a state this delivery brings, bound when it appears. + verdicts := process.AwaitBucket(ctx, js, conn, index.VerdictsBucket, "no verdict on a manifest is written", 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), 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)} @@ -385,6 +388,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := ix.Reconcile(ctx); err != nil { logger.Printf("dropping the modules the trunk does not hold at start: %v", err) } + // Every manifest judged at the start, so a judge that changed with this program judges every one again (ADR 0578 + // §1), once the verdicts state is there; each merge judges them again (index.ApplyMessage). + verdicts.Ready(func(jetstream.KeyValue) { + j, err := ix.JudgeAll(ctx) + if err != nil { + logger.Printf("judging the manifests at start: %v", err) + return + } + logger.Printf("judged %d manifests at start: %d verdicts written, %d refused", j.Judged, j.Written, len(j.Refused)) + }) // A build module says each time it wrote its toolchains key: its stream, and the indexer's durable consumer on it, are // made by the controller as the forge's are. Where the stream is not there yet the indexer says so and is started again // with the build module, as with the gate's source. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go new file mode 100644 index 0000000..9bcd6ee --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go @@ -0,0 +1,38 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// sayRefused says the indexer's event manifest-refused (novox/hq ADR 0578 §1): mesh.mod..event.manifest-refused, +// with the SDK's envelope headers and the digest and judge as the bus's dedup id, so one refusal is said once. +func sayRefused(js jetstream.JetStream, cred process.Account, logf func(string, ...any)) func(context.Context, delivery.ManifestRefused) { + return func(ctx context.Context, r delivery.ManifestRefused) { + data, err := json.Marshal(r) + if err != nil { + logf("judge: %s for %s is not said: %v", delivery.ManifestRefusedEvent, r.Module, err) + return + } + msg := nats.NewMsg(cred.EventSubject(delivery.ManifestRefusedEvent)) + id := "manifest-refused-" + r.Digest + "-" + r.Judge + 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.ManifestRefusedEvent, r.Module, err)) + } + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 5f85619..b536ba5 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -52,6 +52,12 @@ type Indexer struct { Predefined, ConfigCandidates Store // Taken is the taken bucket: per repository the last merge taken in (taken.go, ADR 0556); nil writes none. Taken Store + // Verdicts is the verdicts state (verdicts.go, novox/hq ADR 0578 §1); nil judges nothing. + Verdicts Store + // JudgeID names the judge a verdict says made it; empty is judge.Identity, the running program's. + JudgeID string + // SayRefused says the event manifest-refused for a refusal newly written; nil says nothing. + SayRefused func(ctx context.Context, r delivery.ManifestRefused) // Now is the clock of a taken value; nil is time.Now. Now func() time.Time Builds *Builds @@ -131,6 +137,10 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f delivery.Forge) (Applied, if err := ix.configCandidatesAfterMerge(ctx, f, writtenModules(w.written)); err != nil { return applied, err } + // Every manifest judged again: a rule between manifests can turn on the one this merge moved (ADR 0578 §1). + if _, err := ix.JudgeAll(ctx); err != nil { + ix.logf("judge: the manifests are not judged after %s#%d: %v", f.Repository, f.Number, err) + } } ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945) } diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts.go b/modules/mesh-delivery-indexer/internal/index/verdicts.go new file mode 100644 index 0000000..2605d21 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/verdicts.go @@ -0,0 +1,181 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "slices" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-indexer/judge" +) + +// The indexer judges every manifest it holds, once, with its own judge (novox/hq ADR 0578 §1): it writes each verdict in +// its state verdicts (delivery.VerdictsBucket), one key per manifest digest, only when the verdict differs from the one +// held; a module's value in the modules state carries a refusal (Verdict, Problems), so a reader of modules takes it +// without a second read; a new refusal is said as the event manifest-refused. It judges at its start, so a judge that +// changed (a new indexer) judges every manifest again, and after every merge it takes in, since a rule between manifests +// can turn on any of them. Nothing polls. + +// VerdictsBucket is the indexer's state of verdicts. +var VerdictsBucket = delivery.VerdictsBucket + +// Judged is what one pass of the judge did: the manifests judged, the verdicts written, and the modules refused. +type Judged struct { + Judged, Written int + Refused map[string][]string +} + +// JudgeAll judges every manifest the modules state names, as one catalogue with the predefined list the indexer keeps, +// and writes what differs. A manifest the manifests state does not hold is left unjudged and said; a write that fails is +// said and the pass goes on, so one key never costs the others their verdict. +func (ix *Indexer) JudgeAll(ctx context.Context) (Judged, error) { + var out Judged + if ix.Verdicts == nil || ix.Modules == nil || ix.Manifests == nil { + return out, nil + } + held, err := ix.heldModules(ctx) + if err != nil { + return out, err + } + raws := map[string][]byte{} + digests := map[string]string{} + for _, name := range slices.Sorted(maps.Keys(held)) { + digest := held[name].Manifest + if digest == "" { + continue + } + value, _, found, err := ix.Manifests.Get(ctx, digest) + if err != nil { + return out, fmt.Errorf("reading manifests.%s of %s: %w", digest, name, err) + } + if !found { + ix.logf("judge: the manifest of %s (digest %s) is not in the manifests state, so it is not judged", name, digest) + continue + } + kept, err := delivery.DecodeManifest(value) + if err != nil { + ix.logf("judge: manifests.%s of %s cannot be read, so it is not judged: %v", digest, name, err) + continue + } + raws[name] = []byte(kept.JSON) + digests[name] = digest + } + refused, unnamed := judge.Shelf(raws, ix.predefinedList(ctx), 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) + } + out.Refused = refused + id := ix.JudgeID + if id == "" { + id = judge.Identity() + } + for _, name := range slices.Sorted(maps.Keys(digests)) { + out.Judged++ + v := delivery.ManifestVerdict{Module: name, State: delivery.VerdictAccepted, Judge: id, At: ix.now()} + if problems := refused[name]; len(problems) > 0 { + v.State, v.Problems = delivery.VerdictRefused, problems + } + wrote, err := ix.writeVerdict(ctx, digests[name], v) + if errors.Is(err, process.ErrNotYet) { + return out, fmt.Errorf("the state %s is not on the bus yet: %w", VerdictsBucket, err) + } + if err != nil { + ix.logf("judge: the verdict on %s (digest %s) is not written: %v", name, digests[name], err) + continue + } + if wrote { + out.Written++ + if v.Refused() { + ix.logf("judge: %s", v.Why()) + if ix.SayRefused != nil { + ix.SayRefused(ctx, delivery.RefusedOf(digests[name], v)) + } + } + } + if err := ix.carryVerdict(ctx, name, digests[name], v); err != nil { + ix.logf("judge: modules.%s does not carry its verdict: %v", name, err) + } + } + return out, nil +} + +// writeVerdict writes v under the digest's key unless the verdict held there says the same (ManifestVerdict.Same); wrote +// says it was written. +func (ix *Indexer) writeVerdict(ctx context.Context, digest string, v delivery.ManifestVerdict) (wrote bool, err error) { + key, err := delivery.VerdictsKey(digest) + if err != nil { + return false, err + } + err = write(ctx, ix.Verdicts, key, func(value []byte, found bool) ([]byte, bool, error) { + wrote = false + if found { + if cur, err := delivery.DecodeManifestVerdict(value); err == nil && cur.Same(v) { + return nil, false, nil + } + } + wrote = true + next, err := delivery.EncodeManifestVerdict(v) + return next, false, err + }) + return wrote, err +} + +// carryVerdict puts a refusal into the module's value, and takes one out once the manifest is accepted, while the value +// still names the digest judged; an accepted module's value carries nothing, so judging writes no value of a module the +// judge accepts (a reader takes a value without a verdict as accepted, delivery.TakeManifestVerdict). +func (ix *Indexer) carryVerdict(ctx context.Context, name, digest string, v delivery.ManifestVerdict) error { + key, err := delivery.ModulesKey(name) + if err != nil { + return err + } + return write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) { + if !found { + return nil, false, nil + } + cur, err := delivery.DecodeModule(value) + if err != nil || cur.Manifest != digest { + return nil, false, nil // gone, unreadable, or moved to another manifest since: the next pass judges it + } + state := "" + if v.Refused() { + state = delivery.VerdictRefused + } + if cur.Verdict == state && slices.Equal(cur.Problems, v.Problems) { + return nil, false, nil + } + cur.Verdict, cur.Problems = state, v.Problems + if state == "" { + cur.Problems = nil + } + next, err := delivery.EncodeModule(cur) + return next, false, err + }) +} + +// predefinedList is the catalogue's predefined list the indexer keeps (writePredefined), or nil when it keeps none or +// it cannot be read: the judge then takes each fallback as its value declares it. +func (ix *Indexer) predefinedList(ctx context.Context) catalogue.PredefinedList { + if ix.Predefined == nil { + return nil + } + value, _, found, err := ix.Predefined.Get(ctx, delivery.PredefinedListKey) + if err != nil || !found { + return nil + } + var kept delivery.PredefinedKept + if json.Unmarshal(value, &kept) != nil { + return nil + } + list, err := catalogue.ParsePredefined(kept.List) + if err != nil { + ix.logf("judge: the predefined list kept cannot be read, so fallbacks are judged as their values declare them: %v", err) + return nil + } + return list +} diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts_test.go b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go new file mode 100644 index 0000000..ae763bb --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go @@ -0,0 +1,89 @@ +package index + +import ( + "context" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The indexer judges every manifest it holds and writes each verdict only when it differs; a module's value carries a +// refusal, which is said once as manifest-refused; a new judge writes again only what it judges otherwise; a manifest +// mended clears the module's refusal (novox/hq ADR 0578 §1). +func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) { + ctx := context.Background() + ix := &Indexer{Modules: newMem(), Manifests: newMem(), Verdicts: newMem(), JudgeID: "judge-1", + Now: func() time.Time { return time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) }} + var said []delivery.ManifestRefused + ix.SayRefused = func(_ context.Context, r delivery.ManifestRefused) { said = append(said, r) } + hold := func(module, raw string) string { + digest, err := ix.keepManifest(ctx, module, raw) + if err != nil { + t.Fatal(err) + } + value, err := delivery.EncodeModule(delivery.Module{FP2: "src2:" + strings.Repeat("a", 64), Manifest: digest, + Repository: "novox/mesh-catalog", Commit: "c"}) + if err != nil { + t.Fatal(err) + } + ix.Modules.(*memStore).put(module, value) + return digest + } + good := hold("good", `{"module":"good","version":"1"}`) + bad := hold("bad", `{"module":"bad","version":"1","requires":["amqp"]}`) + j, err := ix.JudgeAll(ctx) + if err != nil || j.Judged != 2 || j.Written != 2 || len(j.Refused) != 1 { + t.Fatalf("the first pass: %+v %v", j, err) + } + verdict := func(digest string) delivery.ManifestVerdict { + value, _, found, _ := ix.Verdicts.Get(ctx, digest) + v, err := delivery.DecodeManifestVerdict(value) + if !found || err != nil { + t.Fatalf("no verdict under %s: %v", digest, err) + } + return v + } + if v := verdict(good); v.Refused() || v.Judge != "judge-1" { + t.Fatalf("good: %+v", v) + } + if v := verdict(bad); !v.Refused() || !strings.Contains(v.Problems[0], `requires "amqp"`) { + t.Fatalf("bad: %+v", v) + } + module := func(name string) delivery.Module { + value, _, _, _ := ix.Modules.Get(ctx, name) + m, err := delivery.DecodeModule(value) + if err != nil { + t.Fatal(err) + } + return m + } + if m := module("bad"); m.Verdict != delivery.VerdictRefused || len(m.Problems) != 1 { + t.Fatalf("modules.bad carries %+v", m) + } + if m := module("good"); m.Verdict != "" { + t.Fatalf("modules.good carries %q", m.Verdict) + } + if len(said) != 1 || said[0].Module != "bad" || said[0].Digest != bad || said[0].Judge != "judge-1" { + t.Fatalf("said %+v", said) + } + // A new judge that judges alike writes nothing and says nothing again. + ix.JudgeID = "judge-2" + if j, err := ix.JudgeAll(ctx); err != nil || j.Written != 0 || len(said) != 1 { + t.Fatalf("a judge that judges alike: %+v %v, said %d", j, err, len(said)) + } + // The manifest mended: its new digest is accepted and the module's refusal goes. + hold("bad", `{"module":"bad","version":"2"}`) + if j, err := ix.JudgeAll(ctx); err != nil || j.Written != 1 || len(j.Refused) != 0 { + t.Fatalf("mended: %+v %v", j, err) + } + if m := module("bad"); m.Verdict != "" || m.Problems != nil { + t.Fatalf("modules.bad still carries %+v", m) + } + // Without the verdicts state the indexer judges nothing. + ix.Verdicts = nil + if j, err := ix.JudgeAll(ctx); err != nil || j.Judged != 0 { + t.Fatalf("without the state: %+v %v", j, err) + } +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 161bf20..48426fe 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -81,12 +81,13 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if m.Module != "mesh-delivery-indexer" { t.Errorf("module = %q", m.Module) } - // 1126: placement is the seat mesh-placement's (ADR 0567): the indexer holds no placement state and says no event. - if len(m.Emits) != 0 { - t.Errorf("emits = %v, want none", m.Emits) + // 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) } // 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}}; !reflect.DeepEqual(m.State, want) { + 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}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 6b26c33..cf4ae16 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -19,7 +19,11 @@ { "name": "predefined", "history": 5 }, { "name": "config-candidates", "history": 5 }, { "name": "taken", "history": 1 }, - { "name": "takeins", "history": 1 } + { "name": "takeins", "history": 1 }, + { "name": "verdicts", "history": 5 } + ], + "emits": [ + "manifest-refused" ], "reads": [ "mesh-build-go.toolchains",