From 31ff708f8895baf2d5c0c499ea1ac7ad8e4643b4 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 17:52:53 +0100 Subject: [PATCH] =?UTF-8?q?Judge=20a=20successor=20indexer's=20manifest=20?= =?UTF-8?q?with=20the=20seat=20definitions=20its=20revision=20carries,=20n?= =?UTF-8?q?ot=20the=20built-in=20list=20(issue=201966,=20ADR=200633=20?= =?UTF-8?q?=C2=A71,=20=C2=A76)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The take-in's dry run reads the contracts modules' definitions of the revision's merges and puts them over the seats while it judges; cmd/mesh-judge puts the request's seats over its own. A refusal of the named indexer's manifest is said on take-in.pending and in indexer_pending. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-indexer/main.go | 2 +- .../cmd/mesh-delivery-indexer/named.go | 11 +- .../cmd/mesh-judge/main.go | 14 +- .../internal/index/candidates.go | 4 + .../internal/index/pending.go | 55 ++++-- .../internal/index/pending_test.go | 76 +++++++- .../internal/index/seatsof.go | 154 +++++++++++++++ .../internal/index/verdicts.go | 4 + modules/mesh-delivery-indexer/judge/judge.go | 19 +- .../mesh-delivery-indexer/judge/judge_test.go | 38 ++++ .../mesh-delivery-indexer/judge/seatfiles.go | 179 ++++++++++++++++++ .../mesh-delivery-indexer/takein/takein.go | 6 + 12 files changed, 545 insertions(+), 17 deletions(-) create mode 100644 modules/mesh-delivery-indexer/internal/index/seatsof.go create mode 100644 modules/mesh-delivery-indexer/judge/seatfiles.go 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 7f21ad2a..b925b5b2 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -347,7 +347,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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), + ReadOpen: openVia(calls.Call), Packages: awaitedStore{packages}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), ReadFile: readFileVia(calls.Call), OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph}, 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. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go index 8d60e4b9..8822cb8e 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go @@ -120,6 +120,12 @@ func goModVia(call toolCaller) func(ctx context.Context, repository, sha, dir st return goFileVia(call, "go.mod") } +// readFileVia reads one file of a repository at a commit by its path (gitea_get_file); a file the commit does not hold is +// none. The seat definitions of a revision are read with it (seatsof.go). +func readFileVia(call toolCaller) func(ctx context.Context, repository, sha, path string) ([]byte, error) { + return goFileVia(call, "") +} + // goFileVia reads one Go file of a module directory at a commit through the forge's get-file tool; a file the commit does // not hold is none. func goFileVia(call toolCaller, name string) func(ctx context.Context, repository, sha, dir string) ([]byte, error) { @@ -129,7 +135,10 @@ func goFileVia(call toolCaller, name string) func(ctx context.Context, repositor return nil, fmt.Errorf("%q is not /", repository) } p := name - if dir != "" && dir != delivery.RootPath { + switch { + case name == "": // readFileVia: dir is the file's whole path + p = dir + case dir != "" && dir != delivery.RootPath: p = dir + "/" + name } raw, err := call(getFileTool, map[string]string{"owner": owner, "repo": repo, "path": p, "ref": sha}) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-judge/main.go b/modules/mesh-delivery-indexer/cmd/mesh-judge/main.go index 8cd30382..5f7be950 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-judge/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-judge/main.go @@ -46,7 +46,19 @@ func run(args []string, stdin io.Reader, stdout, stderr io.Writer) int { fmt.Fprintln(stderr, "mesh-judge:", err) return 1 } - v, err := judge.Request(r) + // The request's seat definitions (ADR 0636 §4): the composer's request type of this build does not decode them. + var seats struct { + Seats []json.RawMessage `json:"seats"` + } + if err := json.Unmarshal(raw, &seats); err != nil { + fmt.Fprintln(stderr, "mesh-judge:", err) + return 1 + } + files := make([][]byte, len(seats.Seats)) + for i, s := range seats.Seats { + files[i] = s + } + v, err := judge.RequestWith(r, files) if err != nil { fmt.Fprintln(stderr, "mesh-judge:", err) return 1 diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 381ce794..72acca96 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -82,6 +82,9 @@ type Indexer struct { // repository it replaces by a directory (fingerprint.ReplacedSiblings), which the version line takes by their // content; nil reads none. GoMod func(ctx context.Context, repository, sha, dir string) ([]byte, error) + // ReadFile reads one file of a repository at a commit (gitea.gitea_get_file); a file the commit does not hold is nil. + // With Files it reads the seat definitions a revision carries (seatsof.go); nil reads none. + ReadFile func(ctx context.Context, repository, sha, path string) ([]byte, error) // Now is the clock of a taken value; nil is time.Now. Now func() time.Time Builds *Builds @@ -119,6 +122,7 @@ type Indexer struct { atHeads map[string]map[string]delivery.PublishedPackages // a member head's published Go modules, by repository@head (candidatePublished) // 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) judgeMu sync.Mutex // one judge pass at a time (verdicts.go) ownMu sync.Mutex // guards ownFP ownFP string // the fingerprint of the build this program runs, asked at the start pass (learnOwnBuild) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index be323c74..39189ff2 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -15,6 +15,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" + "mesh-delivery-indexer/judge" + "git.novox.be/novox/mesh-delivery/sdk/graph" ) @@ -70,6 +72,7 @@ type Pending struct { Ref *PendingRef `json:"ref,omitempty"` Forge json.RawMessage `json:"forge,omitempty"` Lost string `json:"lost,omitempty"` + Refused string `json:"refused,omitempty"` Group string `json:"group,omitempty"` Ended bool `json:"ended,omitempty"` Since time.Time `json:"since"` @@ -688,7 +691,7 @@ func (ix *Indexer) dry() *dryRun { d := &dryRun{} x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum, GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains, - ReadOpen: nil, OwnBuild: nil, Message: ix.Message, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil} + ReadOpen: nil, OwnBuild: nil, Message: ix.Message, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil, ReadFile: ix.ReadFile, underSeats: true} x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) } x.Store, d.candidates = over(ix.Store) x.Modules, d.modules = over(ix.Modules) @@ -721,16 +724,29 @@ func (ix *Indexer) dry() *dryRun { // is left out, as its take-in will drop it. func (ix *Indexer) computeOver(ctx context.Context, ready []PendingEntry) (string, *dryRun, error) { d := ix.dry() - for _, e := range ready { - if e.Covered { - continue - } - if _, err := d.ix.ApplyMessage(ctx, e.Forge); err != nil { - if errors.Is(err, ErrUnreadable) { + // The manifests are judged against the seat definitions the revision carries (seatsof.go, ADR 0633 §1), so the indexer + // it names is not refused for a verb or an event the built-in list lacks. + seats, err := ix.revisionSeats(ctx, ready) + if err != nil { + return "", d, fmt.Errorf("reading the seat definitions of %s: %w", ready[0].Key, err) + } + var applyErr error + judge.WithSeats(seats, func() { + for _, e := range ready { + if e.Covered { continue } - return "", d, fmt.Errorf("computing the take-in of %s (%s#%d): %w", e.Key, e.Forge.Repository, e.Forge.Number, err) + if _, err := d.ix.ApplyMessage(ctx, e.Forge); err != nil { + if errors.Is(err, ErrUnreadable) { + continue + } + applyErr = fmt.Errorf("computing the take-in of %s (%s#%d): %w", e.Key, e.Forge.Repository, e.Forge.Number, err) + return + } } + }) + if applyErr != nil { + return "", d, applyErr } key, _ := delivery.ModulesKey(delivery.IndexerModule) value, _, found, err := d.ix.Modules.Get(ctx, key) @@ -915,24 +931,33 @@ func (ix *Indexer) holdFor(ctx context.Context, ready []PendingEntry, names, own return fmt.Errorf("the indexer's key of the module graph: %w", err) } } + // Why the manifest of the indexer the revision names was refused, when it was (judged with the seat definitions the + // revision carries): said with the wait, as a merge the forge bridge could not give is (holdLost). + refused := "" if value, _, found, err := d.ix.Modules.Get(ctx, key); err == nil && found { if m, err := delivery.DecodeModule(value); err == nil && m.Manifest != "" { if vk, err := delivery.VerdictsKey(m.Manifest); err == nil { if err := copyKey(ctx, d.verdicts, ix.Verdicts, vk); err != nil && !errors.Is(err, process.ErrNotYet) { return fmt.Errorf("the verdict on the indexer's manifest: %w", err) } + if vv, _, ok, err := d.ix.Verdicts.Get(ctx, vk); err == nil && ok { + if v, err := delivery.DecodeManifestVerdict(vv); err == nil && v.Refused() { + refused = strings.Join(v.Problems, "; ") + } + } } } } t := revisionWait(takein.TakeInPending, ready, names) + t.Refused = refused var ended, unmarked []PendingEntry said, changed := false, false for _, e := range ready { switch { case e.Value.Said == t.Key && e.Value.SaidKind == takein.TakeInPending: said = true - changed = changed || e.Value.WaitsFor != names - if e.Value.WaitsFor != names { + changed = changed || e.Value.WaitsFor != names || e.Value.Refused != refused + if e.Value.WaitsFor != names || e.Value.Refused != refused { unmarked = append(unmarked, e) } case e.Value.Said != "" && e.Value.SaidKind == takein.TakeInPending: @@ -947,13 +972,16 @@ func (ix *Indexer) holdFor(ctx context.Context, ready []PendingEntry, names, own } if !said || changed { ix.logf("take-in: %d merges, oldest %s, name the indexer build %s, not %s, which runs: they wait for it", len(ready), t.Key, names, own) + if refused != "" { + ix.logf("take-in: the manifest of the indexer they name is refused, judged with the seat definitions they carry, so no such indexer runs: %s", refused) + } if !ix.sayTakeIn(ctx, t) { - ix.markSaid(ctx, unmarked, nil, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own }) + ix.markSaid(ctx, unmarked, nil, func(p *Pending) { p.WaitsFor, p.LastWriter, p.Refused = names, own, refused }) return nil } unmarked = ready } - ix.markSaid(ctx, unmarked, &t, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own }) + ix.markSaid(ctx, unmarked, &t, func(p *Pending) { p.WaitsFor, p.LastWriter, p.Refused = names, own, refused }) return nil } @@ -1015,6 +1043,9 @@ func waitOf(entries []PendingEntry, what, behind string) takein.PendingWait { if e.Value.Lost != "" { pw.Lost = append(pw.Lost, e.Key) } + if e.Value.Refused != "" { + pw.Refused = e.Value.Refused + } if e.Value.Said != "" && e.Value.SaidAt.After(pw.SaidAt) { pw.Said, pw.SaidAt = "take-in."+e.Value.SaidKind, e.Value.SaidAt } diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 7e6456fc..3d2ea4d9 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -12,6 +12,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" ) @@ -188,6 +189,63 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) t.Errorf("said %v, want %v", b.events(), want) } }) + t.Run("K1 the indexer a revision names is judged with the seat definitions the revision carries", func(t *testing.T) { + // A successor's manifest invokes a verb of the node-engine's seat that the built-in list lacks and the revision's + // definitions add (ADR 0633 §1, §6): refused without them, with its reason said on the wait and in indexer_pending; + // accepted with them, and the revision waits for the build it names. + const verb, contracts = "manifest-reads", "modules/seat-contracts" + for _, row := range []struct { + name string + defined bool + refused bool + }{{"definitions the revision does not carry", false, true}, {"definitions the revision carries", true, false}} { + b := newTakeInBed(t) + f := b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2") + f.Manifests = map[string]string{} + for dir, raw := range b.base.Manifests { + f.Manifests[dir] = raw + } + dir := "modules/" + delivery.IndexerModule + f.Manifests[dir] = strings.Replace(f.Manifests[dir], `"module": "`+delivery.IndexerModule+`"`, + `"module": "`+delivery.IndexerModule+`", "invokes": ["seat:node-engine.`+verb+`"]`, 1) + if row.defined { + f.Trees[contracts] = "tree-seat-contracts" + f.Locks = map[string]map[string]string{} + for d, l := range b.base.Locks { + f.Locks[d] = l + } + f.Locks[contracts] = map[string]string{} + f.Manifests[contracts] = `{"module":"seat-contracts","version":"1","build":{"artifacts":[{"name":"go-seats","kind":"package","language":"go","paths":["go"],"generated-from":"."}]}}` + b.ix.Files = func(_ context.Context, repository, commit string, dirs []string) (map[string]map[string]string, error) { + if repository != f.Repository || commit != f.Merge || !slices.Equal(dirs, []string{contracts + "/seats"}) { + t.Errorf("listed %v of %s at %s", dirs, repository, commit) + } + return map[string]map[string]string{dirs[0]: {"node-engine/seat.json": "1", "node-engine/fixtures/a.json": "2"}}, nil + } + b.ix.ReadFile = func(_ context.Context, _, _, p string) ([]byte, error) { + switch p { + case contracts + "/seats/node-engine/seat.json": + return []byte(`{"format":1,"seat":"node-engine","scope":"node","includes":["reads"]}`), nil + case contracts + "/parts/reads.json": + return []byte(`{"format":1,"part":"reads","verbs":[{"name":"` + verb + `"}]}`), nil + } + return nil, nil + } + } + b.arrive(t, 2, f) + waits := PendingWaits(readPending(t, b)) + refused := "" + if len(b.said) > 0 { + refused = b.said[0].Refused + } + if got := strings.Contains(refused, `serves no verb "`+verb+`"`); got != row.refused || (len(waits) == 1 && waits[0].Refused != refused) { + t.Errorf("%s: said %q, indexer_pending %+v: want a refusal naming the verb %v", row.name, refused, waits, row.refused) + } + if _, ok := catalogue.SeatNamed("node-engine"); !ok || judgeSeatsHave("node-engine", verb) { + t.Errorf("%s: the seats are not put back after the dry run", row.name) + } + } + }) t.Run("T10 the named indexer takes the revision in at its start; a gathered group stays", func(t *testing.T) { b := newTakeInBed(t) b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) @@ -520,7 +578,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) named := map[string]bool{} for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk", - "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn"} { + "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", "ReadFile"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) @@ -540,3 +598,19 @@ func mustReadPending(t *testing.T, b *takeInBed) []PendingEntry { } return entries } + +// judgeSeatsHave says whether the catalogue's seat list holds the verb on the seat now. +func judgeSeatsHave(seat, verb string) bool { + s, _ := catalogue.SeatNamed(seat) + return slices.ContainsFunc(s.Serves, func(v catalogue.Verb) bool { return v.Name == verb }) +} + +// readPending is the bed's pending state as indexer_pending reads it. +func readPending(t *testing.T, b *takeInBed) []PendingEntry { + t.Helper() + entries, err := ReadPending(context.Background(), b.pending, nil) + if err != nil { + t.Fatal(err) + } + return entries +} diff --git a/modules/mesh-delivery-indexer/internal/index/seatsof.go b/modules/mesh-delivery-indexer/internal/index/seatsof.go new file mode 100644 index 00000000..526e6e3b --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/seatsof.go @@ -0,0 +1,154 @@ +package index + +// The seat definitions a revision carries (novox/hq ADR 0633 §1, ADR 0635 §3, ADR 0636 §4): when a take-in works out which +// indexer a revision names, the manifests it writes are judged against the seats of that revision, never against the +// list this program was built with, since the indexer the revision names may serve or invoke what the revision's +// definitions add (a verb of a seat, an event). They are read from the merges' trees: each contracts module a merge's +// manifests name (a package artifact with `generated-from`) holds `seats//seat.json` and the `parts/` they +// include, read through the forge's listing and get-file tools. A delivery group's merges are one revision, so a +// group's members' definitions are read together. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "path" + "slices" + "strings" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + + "mesh-delivery-indexer/judge" +) + +// definitionsDir is the directory of a contracts module's definitions that the module.json at dir names, from the first +// package artifact naming `generated-from`: ok is false for a module that is no contracts module. A manifest that does +// not decode is none here: the judge says why it is refused. +func definitionsDir(dir, manifest string) (string, bool) { + var m struct { + Build struct { + Artifacts []struct { + GeneratedFrom string `json:"generated-from"` + } `json:"artifacts"` + } `json:"build"` + } + if json.Unmarshal([]byte(manifest), &m) != nil { + return "", false + } + for _, a := range m.Build.Artifacts { + if a.GeneratedFrom == "" { + continue + } + base := dir + if base == "." || base == "/" { + base = "" + } + return path.Clean(path.Join(base, a.GeneratedFrom)), true + } + return "", false +} + +// revisionSeats are the seats of the definitions the ready merges carry, read at their merge commits, the newest merge +// of a repository governing; none where the indexer has no forge reader. A read the forge does not answer is an error +// (the take-in is tried again); a definition that does not parse is said and left out, as the indexer that keeps it +// would say it unparsed. +func (ix *Indexer) revisionSeats(ctx context.Context, ready []PendingEntry) ([]catalogue.Seat, error) { + if ix.ReadFile == nil || ix.Files == nil { + return nil, nil + } + var defs []judge.Definition + for _, e := range ready { + if e.Covered || e.Forge.Merge == "" { + continue + } + for _, dir := range slices.Sorted(maps.Keys(e.Forge.Manifests)) { + defsDir, ok := definitionsDir(dir, e.Forge.Manifests[dir]) + if !ok { + continue + } + got, err := ix.readDefinitions(ctx, e.Forge.Repository, e.Forge.Merge, defsDir) + if err != nil { + return nil, err + } + defs = append(defs, got...) + } + } + seats, err := judge.Seats(defs) + if err != nil { + return nil, err + } + return seats, nil +} + +// readDefinitions reads the definitions under a contracts module's definitions directory at a commit, resolved with their +// parts. +func (ix *Indexer) readDefinitions(ctx context.Context, repository, commit, defsDir string) ([]judge.Definition, error) { + file := func(p string) ([]byte, error) { + if defsDir != "." && defsDir != "" { + p = defsDir + "/" + p + } + raw, err := ix.ReadFile(ctx, repository, commit, p) + if err != nil { + return nil, fmt.Errorf("reading %s of %s at %.12s: %w", p, repository, commit, err) + } + return raw, nil + } + dir := path.Join(defsDir, "seats") + if defsDir == "." || defsDir == "" { + dir = "seats" + } + listed, err := ix.Files(ctx, repository, commit, []string{dir}) + if err != nil { + return nil, fmt.Errorf("listing %s of %s at %.12s: %w", dir, repository, commit, err) + } + parts := map[string]judge.Part{} + var out []judge.Definition + for _, p := range slices.Sorted(maps.Keys(listed[dir])) { + seat, ok := strings.CutSuffix(p, "/seat.json") + if !ok || seat == "" || strings.Contains(seat, "/") { + continue + } + raw, err := file("seats/" + seat + "/seat.json") + if err != nil { + return nil, err + } + d, err := judge.ParseDefinition(raw) + if err == nil && d.Seat != seat { + err = fmt.Errorf("it names the seat %q, not %q", d.Seat, seat) + } + for _, name := range d.Includes { + if err != nil { + break + } + if _, have := parts[name]; have { + continue + } + praw, perr := file("parts/" + name + ".json") + if perr != nil { + return nil, perr + } + var part judge.Part + if praw == nil { + perr = errors.New("the commit holds no such part") + } else { + part, perr = judge.ParsePart(praw) + } + if perr != nil { + err = fmt.Errorf("its part %s: %w", name, perr) + break + } + parts[name] = part + } + if err == nil { + d, err = d.Resolve(parts) + } + if err != nil { + ix.logf("take-in: the seat definition seats/%s of %s at %.12s does not parse, so it is left out: %v", seat, repository, commit, err) + continue + } + out = append(out, d) + } + return out, nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts.go b/modules/mesh-delivery-indexer/internal/index/verdicts.go index 1263a8e1..5eabc0a2 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts.go @@ -61,6 +61,10 @@ func (ix *Indexer) judgeOver(ctx context.Context, fresh map[string]bool, over ma // modules another is rewriting. ix.judgeMu.Lock() defer ix.judgeMu.Unlock() + if !ix.underSeats { + // The seat list stands still for this pass: a take-in's dry run may put its revision's seats over it (seatsof.go). + defer judge.HoldSeats()() + } var out Judged if ix.Verdicts == nil || ix.Modules == nil || ix.Manifests == nil { return out, nil diff --git a/modules/mesh-delivery-indexer/judge/judge.go b/modules/mesh-delivery-indexer/judge/judge.go index b41c6ba0..c144cf02 100644 --- a/modules/mesh-delivery-indexer/judge/judge.go +++ b/modules/mesh-delivery-indexer/judge/judge.go @@ -204,7 +204,24 @@ const Unnamed = "(between manifests)" // each case's manifests as one catalogue (Shelf), every manifest accepted or refused with its problems, one per line; the // problems no module's removal ends under Between. The verdict's SDK names this judge (Identity). A manifest left as a // digest, or a case named twice, is a request it cannot run. -func Request(r composer.Request) (composer.Verdict, error) { +func Request(r composer.Request) (composer.Verdict, error) { return RequestWith(r, nil) } + +// RequestWith is Request judging against the seat definitions the request gives (ADR 0636 §4; the request's `seats`, +// which the mesh-sdk this judge is built with does not decode: cmd/mesh-judge reads them from the request's bytes), put +// over the seats this program was built with, never the built-in list alone: inside a delivery group's check they are +// the members' heads' definitions, so a manifest that serves or invokes what the group's definitions add is judged by +// them. None is Request. +func RequestWith(r composer.Request, seats [][]byte) (composer.Verdict, error) { + over, err := SeatsOfFiles(seats) + if err != nil { + return composer.Verdict{}, fmt.Errorf("a seat definition of the request: %w", err) + } + var v composer.Verdict + WithSeats(over, func() { v, err = request(r) }) + return v, err +} + +func request(r composer.Request) (composer.Verdict, error) { v := composer.Verdict{V: composer.Version, SDK: Identity()} seen := map[string]bool{} counts := trunkOldSections(r) diff --git a/modules/mesh-delivery-indexer/judge/judge_test.go b/modules/mesh-delivery-indexer/judge/judge_test.go index 27caf578..b57cfe4d 100644 --- a/modules/mesh-delivery-indexer/judge/judge_test.go +++ b/modules/mesh-delivery-indexer/judge/judge_test.go @@ -96,6 +96,7 @@ func TestTheJudge(t *testing.T) { "a": {Raw: json.RawMessage(`{"module":"a","version":"1"}`)}, "b": {Raw: json.RawMessage(`{"module":"b","version":"1","requires":["amqp"]}`)}, }}}} + was := catalogue.Seats() v, err := Request(r) if err != nil || v.V != composer.Version || len(v.Cases) != 1 || v.SDK != Identity() || !v.Cases[0].Manifests["a"].Accepted || v.Cases[0].Manifests["b"].Accepted || !strings.Contains(v.Cases[0].Manifests["b"].Refused, `requires "amqp"`) { @@ -122,6 +123,43 @@ func TestTheJudge(t *testing.T) { !got["newer"].Accepted || !got["user"].Accepted || !got["mesh-grants"].Accepted { t.Errorf("the change's old sections: %+v", got) } + // ADR 0633 §6, ADR 0636 §4: a successor's manifest is judged by the seat definitions its request carries, put over the + // built-in seats: a verb the built-in list lacks is refused without them and accepted with them (the request's bytes, + // as cmd/mesh-judge reads them), and the seats stand as before once the request is judged. + succ := composer.Request{V: composer.Version, Cases: []composer.Case{{Name: "change", Manifests: map[string]composer.Manifest{ + "successor": {Raw: json.RawMessage(`{"module":"successor","version":"1","invokes":["seat:node-engine.manifest-reads"]}`)}}}}} + defs := [][]byte{[]byte(`{"format":1,"seat":"node-engine","scope":"node","grain":"seat-list","includes":["reads"],"verbs":[{"name":"manifest-reads"}]}`)} + for _, row := range []struct { + name string + seats [][]byte + want string // a word of the refusal; empty: accepted + }{ + {"without definitions", nil, `serves no verb "manifest-reads"`}, + {"a definition with the verb", [][]byte{[]byte(strings.Replace(string(defs[0]), `"includes":["reads"],`, "", 1))}, ""}, + {"a definition naming a part it includes", defs, "includes"}, + {"a definition of no seat", [][]byte{[]byte(`{"format":1}`)}, "names its seat"}, + } { + sv, err := RequestWith(succ, row.seats) + switch { + case row.want != "" && err != nil: + if !strings.Contains(err.Error(), row.want) { + t.Errorf("%s: %v, want %q", row.name, err, row.want) + } + case err != nil: + t.Errorf("%s: %v", row.name, err) + case (row.want == "") != sv.Cases[0].Manifests["successor"].Accepted || + !strings.Contains(sv.Cases[0].Manifests["successor"].Refused, row.want): + t.Errorf("%s: %+v, want %q", row.name, sv.Cases[0].Manifests, row.want) + } + if _, ok := catalogue.SeatNamed("node-engine"); !ok || len(catalogue.Seats()) != len(was) { + t.Fatalf("%s: the seats are not put back", row.name) + } + } + parts := map[string]Part{"reads": {Verbs: []DefVerb{{Name: "manifest-reads"}}}} + d, _ := ParseDefinition(defs[0]) + if d, err := d.Resolve(parts); err != nil || len(d.Verbs) != 2 || d.Verbs[0].Name != "manifest-reads" { + t.Errorf("a resolved definition: %+v %v", d, err) + } r.Cases[0].Manifests["c"] = composer.Manifest{Digest: strings.Repeat("a", 64)} if _, err := Request(r); err == nil { t.Fatal("a manifest left as a digest was judged") diff --git a/modules/mesh-delivery-indexer/judge/seatfiles.go b/modules/mesh-delivery-indexer/judge/seatfiles.go new file mode 100644 index 00000000..384baace --- /dev/null +++ b/modules/mesh-delivery-indexer/judge/seatfiles.go @@ -0,0 +1,179 @@ +package judge + +// The seat definitions a revision carries (novox/hq ADR 0633 §1, §6; ADR 0635, ADR 0636 §4): a manifest is judged +// against the seats of the revision it belongs to, never against the list this program was built with. A definition is +// the contracts module's `seats//seat.json` (design 26 *A seat's definition*), with the parts it includes from +// `parts/.json`. This reader takes of it only what a manifest is judged by (the seat's scope, what it delivers, +// its verbs and events) and reads the rest leniently: the format's one strict reader (mesh-sdk seatdef) is not in the +// mesh-sdk this judge is built with, and the indexer of the revision keeps the definitions whole. + +import ( + "encoding/json" + "fmt" + "slices" + "sync" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// Definition is the part of one seat's definition a manifest is judged by. +type Definition struct { + Format int `json:"format"` + Seat string `json:"seat"` + Scope string `json:"scope"` + Decision string `json:"decision"` + HolderDeclares bool `json:"holder-declares"` + Delivers string `json:"delivers"` + Replicated bool `json:"replicated"` + PulledByKind bool `json:"pulled-by-kind"` + Yields bool `json:"yields"` + HeldByTheEngine bool `json:"held-by-the-engine"` + Includes []string `json:"includes"` + Verbs []DefVerb `json:"verbs"` + Accepts []DefName `json:"accepts"` + Events []DefName `json:"events"` +} + +// DefVerb is a verb of a definition. +type DefVerb struct { + Name string `json:"name"` + Description string `json:"description"` + Optional bool `json:"optional"` + Replaces []string `json:"replaces"` +} + +// DefName is an accepted message or an event of a definition. +type DefName struct { + Name string `json:"name"` +} + +// Part is a part a definition includes: verbs and events written once for the seats that share them. +type Part struct { + Part string `json:"part"` + Verbs []DefVerb `json:"verbs"` + Events []DefName `json:"events"` +} + +// ParseDefinition reads one definition; a file that is not one names a seat and a scope. +func ParseDefinition(raw []byte) (Definition, error) { + var d Definition + if err := json.Unmarshal(raw, &d); err != nil { + return Definition{}, fmt.Errorf("a seat definition is not JSON: %w", err) + } + if d.Seat == "" || (d.Scope != catalogue.ScopeMesh && d.Scope != catalogue.ScopeNode) { + return Definition{}, fmt.Errorf("a seat definition names its seat and its scope (mesh or node), not %q and %q", d.Seat, d.Scope) + } + return d, nil +} + +// ParsePart reads one part. +func ParsePart(raw []byte) (Part, error) { + var p Part + if err := json.Unmarshal(raw, &p); err != nil { + return Part{}, fmt.Errorf("a part of a seat definition is not JSON: %w", err) + } + return p, nil +} + +// Resolve is the definition with the verbs and events of the parts it includes put first, as the indexer keeps it. +func (d Definition) Resolve(parts map[string]Part) (Definition, error) { + out := d + out.Includes, out.Verbs, out.Events = nil, nil, nil + for _, name := range d.Includes { + p, ok := parts[name] + if !ok { + return Definition{}, fmt.Errorf("the seat %s includes the part %q, which is not given", d.Seat, name) + } + out.Verbs = append(out.Verbs, p.Verbs...) + out.Events = append(out.Events, p.Events...) + } + out.Verbs = append(out.Verbs, d.Verbs...) + out.Events = append(out.Events, d.Events...) + return out, nil +} + +// Seats are the seats of resolved definitions, in the catalogue's types; a seat its holder declares itself is left out. +// A definition still naming parts it includes is refused: resolve it first. +func Seats(defs []Definition) ([]catalogue.Seat, error) { + var out []catalogue.Seat + for _, d := range defs { + if d.HolderDeclares { + continue + } + if len(d.Includes) > 0 { + return nil, fmt.Errorf("the seat %s names parts it includes: resolve it first", d.Seat) + } + s := catalogue.Seat{Name: d.Seat, Scope: d.Scope, Delivers: d.Delivers, Replicated: d.Replicated, + PulledByKind: d.PulledByKind, Yields: d.Yields, HeldByTheEngine: d.HeldByTheEngine, Decision: d.Decision} + for _, a := range d.Accepts { + s.Accepts = append(s.Accepts, a.Name) + } + for _, e := range d.Events { + s.Emits = append(s.Emits, e.Name) + } + for _, v := range d.Verbs { + s.Serves = append(s.Serves, catalogue.Verb{Name: v.Name, Description: v.Description, Optional: v.Optional, Replaces: v.Replaces}) + } + out = append(out, s) + } + return out, nil +} + +// SeatsOfFiles are the seats of resolved definitions given as files, as a request carries them. +func SeatsOfFiles(raws [][]byte) ([]catalogue.Seat, error) { + defs := make([]Definition, 0, len(raws)) + for _, raw := range raws { + d, err := ParseDefinition(raw) + if err != nil { + return nil, err + } + defs = append(defs, d) + } + return Seats(defs) +} + +// merged is base with each seat of over put in place of the seat of its name, or added: a definition is a seat's whole +// contract. A verb the base seat holds keeps its input and output (a definition read here carries none). +func merged(base, over []catalogue.Seat) []catalogue.Seat { + out := slices.Clone(base) + for _, s := range over { + at := slices.IndexFunc(out, func(b catalogue.Seat) bool { return b.Name == s.Name }) + if at < 0 { + out = append(out, s) + continue + } + for i, v := range s.Serves { + if j := slices.IndexFunc(out[at].Serves, func(b catalogue.Verb) bool { return b.Name == v.Name }); j >= 0 { + s.Serves[i].Input, s.Serves[i].Output = out[at].Serves[j].Input, out[at].Serves[j].Output + } + } + out[at] = s + } + return out +} + +// seatsMu keeps the seat list the catalogue holds (a list of the process) still while one pass judges with another: +// WithSeats takes it whole; a judge pass of the running indexer that uses the list as it stands takes it shared. +var seatsMu sync.RWMutex + +// HoldSeats keeps the catalogue's seat list as it stands until the returned function is called, so a pass that judges +// with it never sees another's (WithSeats). The one pass inside WithSeats does not take it. +func HoldSeats() func() { + seatsMu.RLock() + return seatsMu.RUnlock +} + +// WithSeats runs fn with seats put over the catalogue's seat list (merged), and puts the list back after. Nothing +// else judges meanwhile (HoldSeats). fn must not call HoldSeats. No seats runs fn as it is, without the exclusion. +func WithSeats(seats []catalogue.Seat, fn func()) { + if len(seats) == 0 { + fn() + return + } + seatsMu.Lock() + defer seatsMu.Unlock() + was := catalogue.Seats() + catalogue.UseSeats(merged(was, seats)) + defer catalogue.UseSeats(was) + fn() +} diff --git a/modules/mesh-delivery-indexer/takein/takein.go b/modules/mesh-delivery-indexer/takein/takein.go index d5ce9866..3e76e4ba 100644 --- a/modules/mesh-delivery-indexer/takein/takein.go +++ b/modules/mesh-delivery-indexer/takein/takein.go @@ -136,6 +136,10 @@ type TakeIn struct { // Lost are the keys of merges whose message the forge bridge's stream no longer holds: they are not taken in, and the // revision waits until indexer_drop_pending drops them (issue 1585). Lost []string `json:"lost,omitempty"` + // Refused is why the manifest of the indexer the revision names was refused when the revision was worked out: with + // the seat definitions the revision carries (ADR 0633 §1), the manifest validator's problems. The revision waits for + // an indexer whose manifest is accepted, and says why here, as it says what it lost. + Refused string `json:"refused,omitempty"` } func (t TakeIn) check() error { @@ -235,6 +239,8 @@ type PendingWait struct { SaidAt time.Time `json:"said-at,omitzero"` // Lost are the keys of its merges whose message the forge bridge's stream no longer holds (TakeIn.Lost). Lost []string `json:"lost,omitempty"` + // Refused is why the manifest of the indexer it names was refused (TakeIn.Refused). + Refused string `json:"refused,omitempty"` } // PendingAnswer is indexer_pending's answer: every wait, oldest first, and the pending keys that do not read.