From 5b6602544e3eec94f7edc5123d967890b86a284b Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 02:17:57 +0100 Subject: [PATCH] Prove the judge once whole: verdicts, manifest-refused, and a bus reader that holds and goes on (ADR 0578) Issue 1585: the indexer's merge test takes in three manifests more against the real bus; it writes their verdicts, says each refusal, and mesh-sdk's bus reader composes the valid one and holds the refused, named; before any verdict it takes each as accepted and names the field set aside. One test extended, no test added. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- .../mesh-delivery-indexer/integration_test.go | 183 +++++++++++++++++- 1 file changed, 180 insertions(+), 3 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 1da6885..c35ebf9 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -4,6 +4,9 @@ package main import ( "context" + "crypto/ecdh" + "crypto/rand" + "encoding/base64" "encoding/json" "fmt" "io" @@ -20,6 +23,8 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/compose" + "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" ) @@ -229,7 +234,12 @@ func TestIndexerWritesModulesAtAMerge(t *testing.T) { // 1239/AC9; I1, I3: the m // TestIndexerKeepsManifestsAtAMerge: a merged message of mesh-catalog carrying the real manifest of the module avahi and the // hash of its go.sum (the bridge's Manifests and Locks, issue 1448) makes, within 5 s, modules.avahi with its manifest digest, // its inputs and the reason it has no fingerprint of the new form, and manifests. with the manifest byte for byte. -func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4 +// +// And it judges them (novox/hq ADR 0578, issue 1585): with three manifests more, one valid, one refused and one with a +// field the decoder does not know, mesh-sdk's bus reader read before the indexer judged composes all three, the field set +// aside and named (a missing verdict is accepted, ยง3); the indexer writes the three verdicts, says manifest-refused for +// each refusal, and the reader then composes the valid one and holds the others, named, never ending its read. +func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 1585 // The manifest is avahi's module.json from mesh-catalog at a67f01963f0d9ef0913a63720b89e573ca7a16ef (copied into the // index package's testdata); the lock hash is the sha256 of modules/avahi/go.sum there. manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json") @@ -237,14 +247,31 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4 t.Fatal(err) } const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381" + judged := map[string]string{ + "good": `{"module":"good","version":"1","resources":[{"id":"d","type":"directory","path":"/srv/good","mode":"0700"}],"data":{"own":[{"id":"d","path":"${dir:d}","class":"valuable"}]}}`, + "bad": `{"module":"bad","version":"1","provides":[{"name":"amqp"}],"resources":[{"id":"d","type":"directory","path":"/srv/bad","mode":"0700"}],"data":{"own":[{"id":"d","path":"${dir:d}","class":"valuable"}]}}`, + "later": `{"module":"later","version":"1","a-field-from-later":1,"resources":[{"id":"d","type":"directory","path":"/srv/later","mode":"0700"}],"data":{"own":[{"id":"d","path":"${dir:d}","class":"valuable"}]}}`, + } js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds buildLog) { builds.end(t, "novox/mesh-catalog", "avahi", "modules/avahi") + }, func(ctx context.Context, js jetstream.JetStream) { + // The verdicts state and the stream of the indexer's events, as the controller makes them from its manifest. + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.VerdictsBucket, History: delivery.VerdictsHistory}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "INDEXER_EVENTS", Subjects: []string{"mesh.mod.mesh-delivery-indexer.event.>"}}); err != nil { + t.Fatal(err) + } }) - data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301, + forge := delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301, Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0", Trees: map[string]string{"modules/avahi": "newtree"}, Manifests: map[string]string{"modules/avahi": string(manifest)}, - Locks: map[string]map[string]string{"modules/avahi": {"go.sum": goSum}}}) + Locks: map[string]map[string]string{"modules/avahi": {"go.sum": goSum}}} + for name, raw := range judged { + forge.Manifests["modules/"+name], forge.Trees["modules/"+name] = raw, "tree-"+name + } + data, err := delivery.EncodeForge(forge) if err != nil { t.Fatal(err) } @@ -275,6 +302,156 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4 if got, err := delivery.DecodeManifest(e.Value()); err != nil || got.Module != "avahi" || got.JSON != string(manifest) { t.Errorf("manifest = %+v err %v, want avahi and the file's bytes", got, err) } + + verdicts, err := js.KeyValue(ctx, delivery.VerdictsBucket) + if err != nil { + t.Fatal(err) + } + want := map[string]string{"good": delivery.VerdictAccepted, "bad": delivery.VerdictRefused, "later": delivery.VerdictRefused} + got := map[string]delivery.ManifestVerdict{} + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && len(got) < len(want); time.Sleep(50 * time.Millisecond) { + for name := range want { + if e, err := verdicts.Get(ctx, delivery.ManifestDigest([]byte(judged[name]))); err == nil { + if v, err := delivery.DecodeManifestVerdict(e.Value()); err == nil { + got[name] = v + } + } + } + } + for name, state := range want { + if v := got[name]; v.State != state || v.Module != name || !strings.HasPrefix(v.Judge, "exe:") { + t.Errorf("the verdict on %s = %+v within 10 s, want %s by the running judge", name, v, state) + } + } + if !strings.Contains(strings.Join(got["bad"].Problems, " "), `provides "amqp"`) || !strings.Contains(strings.Join(got["later"].Problems, " "), "a-field-from-later") { + t.Errorf("the refusals name no problem: %+v", got) + } + // manifest-refused, said once for each refusal. + events, err := js.Stream(ctx, "INDEXER_EVENTS") + if err != nil { + t.Fatal(err) + } + said := map[string]bool{} + for seq := uint64(1); ; seq++ { + msg, err := events.GetMsg(ctx, seq) + if err != nil { + break + } + var r delivery.ManifestRefused + if msg.Subject == "mesh.mod.mesh-delivery-indexer.event."+delivery.ManifestRefusedEvent && json.Unmarshal(msg.Data, &r) == nil { + said[r.Module] = true + } + } + if !said["bad"] || !said["later"] || said["good"] { + t.Errorf("manifest-refused said for %v, want bad and later", said) + } + // mesh-sdk's bus reader over the indexer's modules and manifests: the refused modules held and named, the valid one + // composed, the read whole. + r := judgedReader{ctx: ctx, js: js, t: t} + items, _, held, err := frombus.DataItemsHeld(ctx, r, frombus.Mesh{}, frombus.Scope{Node: "n1"}) + if err != nil || len(items["good"]) != 1 || items["bad"] != nil || items["later"] != nil || + !strings.Contains(held["bad"], `provides "amqp"`) || !strings.Contains(held["later"], "a-field-from-later") { + t.Errorf("the bus reader after the judge: items %v held %v: %v", items, held, err) + } + // Before any verdict (a value without one, as an older indexer wrote it), the reader takes each as accepted and names + // the field it set aside. + for _, name := range []string{"bad", "later"} { + e, err := modules.Get(ctx, name) + if err != nil { + t.Fatal(err) + } + m, _ := delivery.DecodeModule(e.Value()) + m.Verdict, m.Problems = "", nil + raw, _ := delivery.EncodeModule(m) + r.unjudged = append(r.unjudged, raw) + if r.values == nil { + r.values = map[string][]byte{} + } + r.values[name] = raw + } + in, err := frombus.Input(ctx, r, frombus.Mesh{}, frombus.Scope{Node: "n1"}) + if err != nil || len(in.Refused) != 0 || !strings.Contains(in.SetAside["later"], "a-field-from-later") || in.Catalogue["later"].Stubbed { + t.Errorf("the bus reader before the judge: refused %v, set aside %v: %v", in.Refused, in.SetAside, err) + } +} + +// judgedReader is mesh-sdk's bus reader over the indexer's modules and manifests on the bus, with good, bad and later placed +// on n1 and n1's record; values stand in for a module's value where set. It reads nothing else. +type judgedReader struct { + ctx context.Context + js jetstream.JetStream + t *testing.T + values map[string][]byte + unjudged [][]byte +} + +func (r judgedReader) Placements(context.Context) (map[string]delivery.Placement, error) { + p := delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}} + for _, m := range []string{"good", "bad", "later"} { + p.Modules = append(p.Modules, delivery.PlacedModule{Module: m}) + } + return map[string]delivery.Placement{"n1": p}, nil +} + +func (r judgedReader) Records(context.Context) (map[string]delivery.NodeRecord, error) { + k, err := ecdh.X25519().GenerateKey(rand.Reader) + if err != nil { + return nil, err + } + return map[string]delivery.NodeRecord{"n1": {Node: "n1", TunnelKey: base64.StdEncoding.EncodeToString(k.PublicKey().Bytes()), + SealingKey: "sk", Capabilities: []string{"systemd"}}}, nil +} + +func (r judgedReader) Modules(ctx context.Context) (map[string]delivery.Module, error) { + kv, err := r.js.KeyValue(ctx, delivery.ModulesBucket) + if err != nil { + return nil, err + } + out := map[string]delivery.Module{} + for _, name := range []string{"good", "bad", "later"} { + e, err := kv.Get(ctx, name) + if err != nil { + return nil, err + } + raw := e.Value() + if v, ok := r.values[name]; ok { + raw = v + } + m, err := delivery.DecodeModule(raw) + if err != nil { + return nil, err + } + out[name] = m + } + return out, nil +} + +func (r judgedReader) Manifest(ctx context.Context, digest string) ([]byte, error) { + kv, err := r.js.KeyValue(ctx, delivery.ManifestsBucket) + if err != nil { + return nil, err + } + e, err := kv.Get(ctx, digest) + if err != nil { + return nil, err + } + m, err := delivery.DecodeManifest(e.Value()) + return []byte(m.JSON), err +} + +func (judgedReader) Values(context.Context, string, string) (map[string]any, error) { return nil, nil } +func (judgedReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { + return nil, nil +} +func (judgedReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) { + return nil, nil, nil +} +func (judgedReader) Build(context.Context, string, string) (*delivery.BuilderBuild, error) { + return nil, nil +} +func (judgedReader) Archive(context.Context, string, string) ([]byte, error) { return nil, nil } +func (judgedReader) Served(context.Context, string, string, []string) (map[string]any, error) { + return nil, nil } // TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain: the bucket mesh-build-go_toolchains exists (the state