novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Prove the judge once whole: verdicts, manifest-refused, and a bus reader that holds and goes on (ADR 0578)

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-01 02:17:57 +01:00
parent efef7183ef
commit 5b6602544e
1 changed files with 180 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go
+180 -3
View File
@@ -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.<digest> 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