From aa647d0eda8a8448f247fa80d7c45074a6dea191 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 14:18:17 +0100 Subject: [PATCH] DLV-I17, I18, I19, I20: the indexer keeps every contracts module's seat definitions on the bus MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Design 26 and ADR 0608 §2-§4: the definitions are read at each trunk merge and pull request head through the forge's get-file tool, parsed by mesh-sdk seatdef and kept in the state seat-definitions by seat and source; seat-definition-changed is said, kept seven days, only when a digest moved. A trunk definition that does not parse leaves the trunk's entry and writes seat-definition-unparsed; a pull request's entries go when it closes or merges. A trunk build end naming out-of-step files writes generated-types-out-of-step, cleared only by a clean build of the same module; the checker's verdict keeps a pull request's. Tools indexer_seat_definition, indexer_seat_definitions_again (a hand act) and indexer_out_of_step. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/mesh-delivery-indexer-tools/main.go | 32 +- .../cmd/mesh-delivery-indexer-tools/views.go | 37 + .../cmd/mesh-delivery-indexer/main.go | 24 +- .../cmd/mesh-delivery-indexer/named.go | 5 +- .../cmd/mesh-delivery-indexer/seatdefs.go | 72 ++ .../cmd/mesh-delivery-indexer/takein.go | 3 + modules/mesh-delivery-indexer/go.mod | 1 + .../internal/index/candidates.go | 15 + .../internal/index/control_records_test.go | 126 ++++ .../internal/index/seatdefs.go | 679 ++++++++++++++++++ .../internal/index/takein.go | 15 +- .../internal/index/views.go | 3 +- .../mesh-delivery-indexer/manifest_test.go | 13 +- modules/mesh-delivery-indexer/module.json | 12 +- 14 files changed, 1019 insertions(+), 18 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatdefs.go create mode 100644 modules/mesh-delivery-indexer/internal/index/seatdefs.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index ec0bf378..464a1921 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -127,6 +127,36 @@ func Tools(ask asker, now func() time.Time, read stdio.KeptState, counts func() } return a.Result, nil }}, + {Name: index.SeatDefinitionsAgainTool, + Description: "Read a contracts module's seat definitions again now (design 26): its trunk's, at the commit the indexer " + + "holds the module at, or with pull and head a pull request's head, and keep them in the seat-definitions state as " + + "a merge or a pull request's push would (seat-definition-changed said where a digest moved; the trunk's " + + "seat-definition-unparsed finding written or cleared). For a read the forge did not answer. A hand act, " + + "recorded through mesh-hand-acts before it is asked. indexer_seat_definition reads the result.", + Input: map[string]any{"type": "object", "required": []string{"module", "why"}, "properties": map[string]any{ + "module": map[string]any{"type": "string", "description": "the contracts module, such as mesh-contracts"}, + "pull": map[string]any{"type": "integer", "description": "a pull request's number, with head"}, + "head": map[string]any{"type": "string", "description": "the pull request's head commit, with pull"}, + "why": map[string]any{"type": "string", "description": "why it is read again now, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + pull, _ := args["pull"].(float64) + key, value, err := index.NewSeatDefinitionsAgain(str(args, "module"), int(pull), str(args, "head"), str(args, "why"), stdio.CallerOf(args), now()) + if err != nil { + return nil, err + } + if str(args, "why") == "" { + return nil, fmt.Errorf("say why it is read again (why): it is a hand act") + } + if _, err := acts.Record(handacts.Request{What: "read the seat definitions of " + str(args, "module") + " again by hand", + Why: str(args, "why"), Cause: index.SeatDefinitionsAgainTool, Verb: "seat-definitions-again"}); err != nil { + return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was read again: %w", err) + } + a, err := ask(key, value) + if err != nil { + return nil, err + } + return a.Result, nil + }}, {Name: index.JudgeAgainTool, Description: "Judge the manifests again now, as the indexer does at its start and after each merge, and answer how the module's " + "verdict stands (waits, refused or accepted, and why). For a manifest that waits for a judge that knows its fields, or " + @@ -219,7 +249,7 @@ func heldCounts(s stdio.KeptState) func() (index.OldSectionsCount, error) { } func main() { - if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState)), handacts.Recorder{Ask: stdio.Ask}), append(append(catalogTools(), viewTools(os.Getenv, keptSource, time.Now)...), configReadersTool(stdio.NewWatchedStates()), busStateTool(keptSource), requirementsTool(keptSource), packageUsersTool(keptSource))...)); err != nil { + if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState)), handacts.Recorder{Ask: stdio.Ask}), append(append(catalogTools(), viewTools(os.Getenv, keptSource, time.Now)...), configReadersTool(stdio.NewWatchedStates()), busStateTool(keptSource), requirementsTool(keptSource), packageUsersTool(keptSource), seatDefinitionTool(keptSource), outOfStepTool(keptSource))...)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go index f5f06ad6..4259a2e3 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go @@ -7,6 +7,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/views" @@ -145,3 +146,39 @@ func packageUsersTool(state func(string) views.Source) stdio.Tool { return map[string]any{"packages": got}, nil }} } + +// seatDefinitionTool is indexer_seat_definition (design 26; DLV-I19): one seat's definition, the trunk's unless a pull +// request is named, saying which it answered, at the mark the views family said the state up to. +func seatDefinitionTool(state func(string) views.Source) stdio.Tool { + return stdio.Tool{Name: index.SeatDefinitionTool, + Description: "One seat's definition as the indexer keeps it (design 26): the trunk's, which is the seat's contract, unless a " + + "pull request is named (repository and pull), whose head's is answered with the trunk's beside it; answered says " + + "which. An entry with unparsed holds why the pull request's definition does not parse. The revision is the one " + + "the indexer's views family said the state up to; seat-definition-changed says each change after it.", + Input: map[string]any{"type": "object", "required": []string{"seat"}, "properties": map[string]any{ + "seat": map[string]any{"type": "string", "description": "the seat, such as go-package-registry"}, + "repository": map[string]any{"type": "string", "description": "a pull request's repository (owner/name), with pull"}, + "pull": map[string]any{"type": "integer", "description": "a pull request's number, with repository"}}}, + Run: func(args map[string]any) (any, error) { + seat, _ := args["seat"].(string) + repository, _ := args["repository"].(string) + pull, _ := args["pull"].(float64) + return index.SeatDefinitionOf(state(index.SeatDefinitionsState), state(views.MarksState), seat, repository, int(pull)) + }} +} + +// outOfStepTool is indexer_out_of_step (design 26; DLV-I19): the generated types out of step with their definitions. +func outOfStepTool(state func(string) views.Source) stdio.Tool { + return stdio.Tool{Name: index.OutOfStepTool, + Description: "The generated types out of step with their seat definitions (design 26): each trunk contracts module whose " + + "finding generated-types-out-of-step stands (its last trunk build's test step named the files; cleared only by a clean " + + "build of the same module), then each open pull request whose check's verdict named out-of-step files.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { + rows, err := index.OutOfStepOf(state(index.SeatDefinitionsState), state(health.FindingsState)) + if err != nil { + return nil, err + } + return map[string]any{"out-of-step": rows}, nil + }} +} 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 ea1d8dcd..8c4ab17b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -331,12 +331,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The module graph (ADR 0603 §2): a state this delivery brings, bound when it appears; until then no graph is written. graph := process.AwaitBucket(ctx, js, conn, graph.Bucket, "no module graph is written", logger.Printf) findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "no finding of the indexer is written", logger.Printf) + // The seat definitions (design 26, DLV-I17): a state this delivery brings, bound when it appears; until then none is kept. + seatDefs := process.AwaitBucket(ctx, js, conn, index.SeatDefinitionsBucket, "no seat definition is kept", logger.Printf) ix := &index.Indexer{Findings: awaitedStore{findings}, OldSections: awaitedStore{oldSections}, Requirements: awaitedStore{requirements}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(js, cred, logger.Printf), SayAccepted: sayAccepted(js, cred), 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), - OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph}} + OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph}, + SeatDefinitions: awaitedStore{seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinitionChanged: saySeatDefinitionChanged(js, cred)} // The log's entries at the start are read before any fingerprint is computed or registered module indexed (below); each // new one indexes the module it names when that module is registered and the bucket holds none (registered.go). var started atomic.Bool @@ -350,6 +353,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := ix.IndexRegisteredModule(ctx, le.Key); err != nil { logger.Printf("the registered module %s is not indexed: %v", le.Key, err) } + // A trunk build's generated files out of step with their seat definitions (DLV-I20). + ix.ApplyOutOfStepBuild(ctx, le.Value) // A done build's imports are kept on its module (ADR 0593 §2); at the start they are kept once the packages are. if err := ix.ApplyBuildEnd(ctx, le.Value); err != nil { logger.Printf("%s entry %d: its imports are not kept: %v", delivery.BuildLogRead, le.Sequence, err) @@ -417,6 +422,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } }) for _, le := range startEnds { + ix.ApplyOutOfStepBuild(ctx, le.Value) if err := ix.ApplyBuildEnd(ctx, le.Value); err != nil { logger.Printf("%s entry %d: its imports are not kept: %v", delivery.BuildLogRead, le.Sequence, err) } @@ -502,6 +508,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer vc.Stop() } + // The checker's verdicts carry a pull request's out-of-step files (DLV-I20): on the indexer's durable consumer of the + // checker's stream asks. Where it is not there yet, no pull request's out-of-step files are kept until the indexer + // starts again, said. + switch acons, err := cred.EventConsumer(ctx, js, checkerAsksStream); { + case err != nil: + logger.Printf("the consumer %s_%s on %s does not exist yet, so no pull request's out-of-step files are kept until the indexer starts again: %v", cred.Node, cred.Module, checkerAsksStream, err) + default: + ac, err := process.HearEvents(acons, func(m jetstream.Msg) { heardVerdict(ctx, ix, m, logger.Printf) }, logger.Printf) + if err != nil { + return err + } + defer ac.Stop() + } cons, err := cred.EventConsumer(ctx, js, forgeStream) if err != nil { // Essential: the forge's merges are what the indexer indexes. @@ -520,7 +539,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro map[string]readyOf{"manifests": boundNow(manifests), "modules": boundNow(modules), "candidates": boundNow(candidates), index.VerdictsState: awaited(verdicts), delivery.IndexerPredefinedState: awaited(predefined), delivery.IndexerConfigCandidatesState: awaited(configCandidates), delivery.PackagesState: awaited(packages), - graphState: awaited(graph), index.HeardState: awaited(heard.b), "taken": awaited(taken), "findings": awaited(findings)}, errs, logger.Printf) + graphState: awaited(graph), index.HeardState: awaited(heard.b), "taken": awaited(taken), "findings": awaited(findings), + index.SeatDefinitionsState: awaited(seatDefs)}, errs, logger.Printf) cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, heard, logger); ix.KickFill(ctx) }, logger.Printf) if err != nil { return err 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 f10cac87..1121e0a7 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go @@ -128,7 +128,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-delivery-indexer/seatdefs.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatdefs.go new file mode 100644 index 00000000..1b856272 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatdefs.go @@ -0,0 +1,72 @@ +package main + +import ( + "context" + "encoding/json" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/seats/meshdeliverychecker" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-indexer/internal/index" +) + +// The seat definitions' wiring (DLV-I17-I20; design 26): the forge's get-file tool reads them, seat-definition-changed +// is said on the indexer's stream indexer-seat-definitions (kept seven days), and the checker's verdict event, heard on +// the indexer's durable consumer of the checker's stream asks, carries a pull request's out-of-step files. + +// checkerAsksStream is the checker's stream asks as the bus names it (its manifest's streams; StreamBusName). +const checkerAsksStream = "ASKS" + +// readFileVia reads one file of a repository at a commit through the forge's get-file tool; a path the commit does not +// hold is nil with no error. +func readFileVia(call toolCaller) func(ctx context.Context, repository, sha, path string) ([]byte, error) { + return goFileVia(call, "") +} + +// saySeatDefinitionChanged publishes seat-definition-changed on the indexer's own stream, its duplicate id the seat, +// source and digest, so a message delivered again says it once. +func saySeatDefinitionChanged(js jetstream.JetStream, cred process.Account) func(context.Context, index.SeatDefinitionChanged) error { + return func(ctx context.Context, c index.SeatDefinitionChanged) error { + data, err := json.Marshal(c) + if err != nil { + return err + } + id := index.SeatDefinitionKey(c.Seat, c.Source, c.Repository, c.Pull) + "." + c.Digest + msg := nats.NewMsg(cred.StreamSubject(index.SeatDefinitionChangedEvent)) + 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() + _, err = js.PublishMsg(pub, msg, jetstream.WithMsgID(id)) + return err + } +} + +// heardVerdict takes the checker's verdict event: a pull request's out-of-step files are kept (a clean verdict removes +// them). Another event of the stream is acknowledged and passed over; a write that failed is delivered again. +func heardVerdict(ctx context.Context, ix *index.Indexer, m jetstream.Msg, logf func(string, ...any)) { + if m.Subject() != meshdeliverychecker.HolderSubjectVerdict(delivery.CheckerModule) { + _ = m.Ack() + return + } + var v meshdeliverychecker.VerdictHolderEvent + if err := json.Unmarshal(m.Data(), &v); err != nil { + logf("%s holds what is no verdict, left: %v", m.Subject(), err) + _ = m.Term() + return + } + if err := ix.ApplyVerdict(ctx, v.Repository, int(v.Pull), v.Head, v.OutOfStep); err != nil { + logf("%s#%d: its out-of-step files are not kept, delivered again: %v", v.Repository, v.Pull, err) + return + } + _ = m.Ack() +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index 74bd16a9..054cb76e 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -17,6 +17,7 @@ type takerIn interface { TakeInRepository(ctx context.Context, repository string) index.TakenIn JudgeAgain(ctx context.Context, module, digest string) index.TakenIn SayAcceptedAgain(ctx context.Context, module string) index.TakenIn + SeatDefinitionsAgain(ctx context.Context, module string, pull int, head string) index.TakenIn } // takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the @@ -38,6 +39,8 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream if a.Repository != "" { asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository) a.Module = a.Repository + } else if a.SeatDefinitions { + asked, r = "reading of its seat definitions again", ix.SeatDefinitionsAgain(ctx, a.Module, a.Pull, a.Head) } else if a.Say { asked, r = "manifest-accepted said again", ix.SayAcceptedAgain(ctx, a.Module) } else if a.Judge { diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 610c5508..f4cb0068 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -28,6 +28,7 @@ require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 require ( git.novox.be/novox/mesh-contracts/go/seats/gopackageregistry v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/go/seats/meshdeliverychecker v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/npmpackageregistry v0.0.0-00010101000000-000000000000 ) diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 85559fe0..5ad825a6 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -76,6 +76,13 @@ type Indexer struct { // 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 + // SeatDefinitions is the state seat-definitions (seatdefs.go; design 26, DLV-I17); nil keeps none. + SeatDefinitions Store + // ReadFile reads one file of a repository at a commit (gitea.gitea_get_file); a file the commit does not hold is + // nil with no error. Nil reads no seat definition. + ReadFile func(ctx context.Context, repository, sha, path string) ([]byte, error) + // SaySeatDefinitionChanged says seat-definition-changed (seatdefs.go); nil says nothing. + SaySeatDefinitionChanged func(ctx context.Context, c SeatDefinitionChanged) error // Graph is the module graph's state (graph.Bucket, graph.go; novox/hq ADR 0603 §2); nil writes no graph. Graph Store // Files asks the forge for every blob under each directory at a commit, by directory then path relative to it, each @@ -158,6 +165,7 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f delivery.Forge) (Applied, if f.Kind != delivery.ForgePull && f.Kind != delivery.ForgeMerged { return applied, nil } + lateMerge := false if f.Kind == delivery.ForgeMerged && ix.Modules != nil { w, err := ix.applyMerged(ctx, f) applied = Applied{Written: w.written, Late: w.late} @@ -184,6 +192,13 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f delivery.Forge) (Applied, } } ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945) + lateMerge = w.late != "" + } + // The seat definitions of the contracts modules it carries (DLV-I17, I18): a late merge moves no trunk's. + if !lateMerge { + if err := ix.ApplySeatDefinitions(ctx, f); err != nil { + return applied, err + } } var digests map[string]string var held map[string]delivery.Module diff --git a/modules/mesh-delivery-indexer/internal/index/control_records_test.go b/modules/mesh-delivery-indexer/internal/index/control_records_test.go index 1dd84d4c..06357b29 100644 --- a/modules/mesh-delivery-indexer/internal/index/control_records_test.go +++ b/modules/mesh-delivery-indexer/internal/index/control_records_test.go @@ -10,6 +10,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -113,6 +114,131 @@ func TestTheControlRecords(t *testing.T) { // DLV-I5 (milestone § 2 item 1, design 51 *The versions*): a package's series key keeps the newest trunk version the // builder numbered and the newest a seat published; an older one, a tagged one and one of another series move neither; // the per-module digests never read a series key. + // DLV-I17-I20 (design 26 *A seat's definition*): the indexer keeps each contracts module's seat definitions by seat + // and source, says seat-definition-changed only when a digest moved, keeps the trunk's entry over one that does not + // parse (its finding), removes a pull request's entries when it closes, keeps a verdict's out-of-step files, and + // raises generated-types-out-of-step from a trunk build end, cleared only by a clean one of the same module. + t.Run("seat definitions", func(t *testing.T) { + const repo, m1, m2, m3, head = "novox/mesh-contracts", "m1", "m2", "m3", "h1" + manifest := `{"module":"mesh-contracts","version":"1","build":{"artifacts":[{"name":"go","kind":"package","language":"go","paths":["go"],"series":"0.1","generated-from":"."}]}}` + git := `{"format":1,"seat":"git","scope":"mesh","grain":"seat-list","decision":"novox/hq ADR 0111","delivers":"git"}` + files := map[string]map[string]string{ + m1: {"seats/index.json": `{"format":1,"seats":["git","gone"]}`, "seats/git/seat.json": git, "seats/gone/seat.json": strings.Replace(git, `"git"`, `"gone"`, 1)}, + m2: {"seats/index.json": `{"format":1,"seats":["git"]}`, "seats/git/seat.json": `{"format":1,"seat":"git","scope":"mesh","grain":"seat-list","decision":"x","oops":1}`}, + m3: {"seats/index.json": `{"format":1,"seats":["git"]}`, "seats/git/seat.json": strings.Replace(git, "ADR 0111", "ADR 0112", 1)}, + head: {"seats/index.json": `{"format":1,"seats":["git"]}`, "seats/git/seat.json": strings.Replace(git, "ADR 0111", "ADR 0200", 1)}, + } + defs, findings := newMem(), newMem() + var said []SeatDefinitionChanged + ix := &Indexer{SeatDefinitions: defs, Findings: findings, Logf: t.Logf, + ReadFile: func(_ context.Context, r, sha, p string) ([]byte, error) { + if v, ok := files[sha][p]; ok && r == repo { + return []byte(v), nil + } + return nil, nil + }, + SaySeatDefinitionChanged: func(_ context.Context, c SeatDefinitionChanged) error { said = append(said, c); return nil }} + ctx := context.Background() + merged := func(merge string) { + t.Helper() + if err := ix.ApplySeatDefinitions(ctx, delivery.Forge{Kind: delivery.ForgeMerged, Repository: repo, Number: 1, Merge: merge, + Manifests: map[string]string{delivery.RootPath: manifest, "modules/other": `{"module":"other","version":"1"}`}}); err != nil { + t.Fatal(err) + } + } + entry := func(key string) (SeatDefinitionEntry, bool) { + var e SeatDefinitionEntry + v, ok := defs.vals[key] + if ok { + _ = json.Unmarshal(v, &e) + } + return e, ok + } + unparsed := health.FindingKey(health.ScopeModule, findingID("seat-definitions", "mesh-contracts"), SeatDefinitionUnparsedKind) + merged(m1) + merged(m1) // the same digests again say nothing + g1, ok := entry("git.trunk") + if !ok || g1.Commit != m1 || g1.Digest == "" || len(said) != 2 { + t.Fatalf("after m1: git.trunk %+v (%v), said %+v; want both seats kept and said once", g1, ok, said) + } + // m2's git does not parse (an unknown key) and drops gone: git's trunk entry stands, the finding names it. + merged(m2) + if g, _ := entry("git.trunk"); g.Digest != g1.Digest || g.Commit != m1 { + t.Errorf("an unparsed definition replaced the trunk's entry: %+v", g) + } + if _, ok := entry("gone.trunk"); ok { + t.Error("a seat the trunk's index no longer names is kept") + } + if v, ok := findings.vals[unparsed]; !ok || !strings.Contains(string(v), "oops") { + t.Errorf("finding %s = %s, want it naming the unknown key", unparsed, v) + } + merged(m3) + if g, _ := entry("git.trunk"); g.Commit != m3 || g.Digest == g1.Digest { + t.Errorf("after m3: git.trunk %+v, want m3's", g) + } + if _, ok := findings.vals[unparsed]; ok { + t.Error("the unparsed finding stands after every definition parsed") + } + if n := len(said); n != 4 || said[2] != (SeatDefinitionChanged{Seat: "gone", Source: SourceTrunk, Repository: repo}) || said[3].Digest == "" { + t.Errorf("said %+v, want gone's removal and m3's git", said) + } + // A pull request's head: its own entry, the trunk's answered unless it is named. + if err := ix.ApplySeatDefinitions(ctx, delivery.Forge{Kind: delivery.ForgePull, Action: delivery.PullUpdated, Repository: repo, Number: 9, Head: head, + Dirs: []string{delivery.RootPath}, Manifests: map[string]string{delivery.RootPath: manifest}}); err != nil { + t.Fatal(err) + } + pk := SeatDefinitionKey("git", SourcePull, repo, 9) + src := func() memSource { + out := memSource{} + for k, v := range defs.vals { + out[k] = &views.Entry{Value: v} + } + return out + } + if a, err := SeatDefinitionOf(src(), nil, "git", "", 0); err != nil || a.Answered != SourceTrunk || a.Entry.Commit != m3 { + t.Errorf("the trunk's answer %+v, %v", a, err) + } + if a, err := SeatDefinitionOf(src(), nil, "git", repo, 9); err != nil || a.Answered != repo+"#9" || a.Entry.Commit != head || a.Trunk.Commit != m3 { + t.Errorf("the pull request's answer %+v, %v", a, err) + } + // Its check's verdict names out-of-step files; a trunk build end names them too, a failed build clears nothing. + if err := ix.ApplyVerdict(ctx, repo, 9, head, []string{"go/b.go", "go/a.go"}); err != nil { + t.Fatal(err) + } + end := func(outcome string, files ...string) []byte { + raw, _ := json.Marshal(map[string]any{"v": 1, "event": delivery.BuildEndedEvent, "module": "mesh-contracts", "ref": "main", + "commit": m3, "outcome": outcome, "out-of-step": files}) + return raw + } + ix.ApplyOutOfStepBuild(ctx, end("failed", "ts/x.ts")) + ix.ApplyOutOfStepBuild(ctx, end("failed")) + fsrc := memSource{} + for k, v := range findings.vals { + fsrc[k] = &views.Entry{Value: v} + } + rows, err := OutOfStepOf(src(), fsrc) + if err != nil || len(rows) != 2 || rows[0].Module != "mesh-contracts" || !strings.Contains(rows[0].Summary, "ts/x.ts") || + rows[1].Pull != repo+"#9" || !reflect.DeepEqual(rows[1].Files, []string{"go/a.go", "go/b.go"}) { + t.Errorf("out of step %+v, %v; want the trunk's module and the pull request's sorted files", rows, err) + } + ix.ApplyOutOfStepBuild(ctx, end(delivery.BuildDone)) + if len(findings.vals) != 0 { + t.Errorf("findings %v stand after a clean build end", findings.vals) + } + // The pull request closes: its entries and its out-of-step files go, said. + if err := ix.ApplySeatDefinitions(ctx, delivery.Forge{Kind: delivery.ForgePull, Action: delivery.PullClosed, Repository: repo, Number: 9, Head: head}); err != nil { + t.Fatal(err) + } + if _, ok := entry(pk); ok { + t.Error("a closed pull request's entry is kept") + } + if _, ok := defs.vals[OutOfStepKey(repo, 9)]; ok { + t.Error("a closed pull request's out-of-step files are kept") + } + if last := said[len(said)-1]; last != (SeatDefinitionChanged{Seat: "git", Source: SourcePull, Repository: repo, Pull: 9}) { + t.Errorf("last said %+v, want the pull request's removal", last) + } + }) t.Run("packages per series", func(t *testing.T) { ctx := context.Background() ix := &Indexer{Packages: newMem()} diff --git a/modules/mesh-delivery-indexer/internal/index/seatdefs.go b/modules/mesh-delivery-indexer/internal/index/seatdefs.go new file mode 100644 index 00000000..3227fcd7 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/seatdefs.go @@ -0,0 +1,679 @@ +package index + +// The seat definitions on the bus (novox/hq ADR 0608 §2-§4, design 26 *A seat's definition*; DLV-I17-I20). The indexer +// reads every seat definition of a contracts module (a module whose package artifact names `generated-from`) at each +// trunk merge and each pull request's head the forge bridge says changed it, through the forge's get-file tool, parses +// it with the format's one reader (mesh-sdk seatdef), and keeps it in its own state `seat-definitions`, keyed by seat +// and source, as its only writer. Only the trunk's entry is the seat's contract; a pull request's entries go when it +// closes or merges. A trunk definition that does not parse keeps the trunk's entry and writes the finding +// seat-definition-unparsed; generated types out of step on a trunk, said by the builder's build end, write the finding +// generated-types-out-of-step, cleared only by a clean build end of the same contracts module. + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "path" + "slices" + "strconv" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/seatdef" + "git.novox.be/novox/mesh-sdk/go/views" +) + +const ( + // SeatDefinitionsState is the indexer's state of the seat definitions (design 26), one key per seat and source. + SeatDefinitionsState = "seat-definitions" + SeatDefinitionsBucket = delivery.IndexerModule + "_" + SeatDefinitionsState + // SeatDefinitionChangedEvent is said, kept seven days on the indexer's stream SeatDefinitionsStream, only when a + // definition's digest changed (ADR 0608 §3). + SeatDefinitionChangedEvent = "seat-definition-changed" + SeatDefinitionsStream = "indexer-seat-definitions" + // The two findings (ADR 0608 §4, ADR 0564: the indexer writes findings, the condition keeper raises them). + SeatDefinitionUnparsedKind = "seat-definition-unparsed" + OutOfStepKind = "generated-types-out-of-step" + // The three tools (design 26). + SeatDefinitionTool = "indexer_seat_definition" + SeatDefinitionsAgainTool = "indexer_seat_definitions_again" + OutOfStepTool = "indexer_out_of_step" + + // SourceTrunk and SourcePull are a definition's two sources. + SourceTrunk = "trunk" + SourcePull = "pull" +) + +// outOfStepPrefix begins the key of a pull request's out-of-step files in the state: no seat name holds '_'. +const outOfStepPrefix = "_out-of-step." + +// SeatDefinitionEntry is one value of the state: a seat's definition from one source, resolved with the parts it +// includes. Unparsed is why a pull request's head's definition did not parse (it then holds no definition); a trunk's +// that does not parse is never written over the entry that stands. +type SeatDefinitionEntry struct { + Seat string `json:"seat"` + Source string `json:"source"` + Repository string `json:"repository"` + Pull int `json:"pull,omitempty"` + Module string `json:"module"` + Directory string `json:"directory"` + Commit string `json:"commit"` + Digest string `json:"digest,omitempty"` + Definition json.RawMessage `json:"definition,omitempty"` + Unparsed string `json:"unparsed,omitempty"` + At time.Time `json:"at"` +} + +// SeatDefinitionChanged is the event's body: the seat, its source and the new digest ("" when the source holds the +// seat no more). +type SeatDefinitionChanged struct { + Seat string `json:"seat"` + Source string `json:"source"` + Repository string `json:"repository"` + Pull int `json:"pull,omitempty"` + Digest string `json:"digest"` +} + +// OutOfStep is a pull request's generated files out of step with their definitions, as its check's verdict said them. +type OutOfStep struct { + Repository string `json:"repository"` + Pull int `json:"pull"` + Head string `json:"head"` + Files []string `json:"files"` + At time.Time `json:"at"` +} + +// SeatDefinitionKey is a seat's key for a source: `.trunk`, or `.pull..`. +func SeatDefinitionKey(seat, source, repository string, pull int) string { + if source == SourceTrunk { + return seat + "." + SourceTrunk + } + return seat + "." + pullSuffix(repository, pull) +} + +func pullSuffix(repository string, pull int) string { + return SourcePull + "." + delivery.EscapeName(repository) + "." + strconv.Itoa(pull) +} + +// OutOfStepKey is the key of a pull request's out-of-step files. +func OutOfStepKey(repository string, pull int) string { + return outOfStepPrefix + delivery.EscapeName(repository) + "." + strconv.Itoa(pull) +} + +// errForgeRead is a read the forge did not answer (never a file it does not hold): the message is left for redelivery. +var errForgeRead = errors.New("the forge did not answer") + +// contractsModule is a contracts module of a message: its module, its directory and its definitions' directory. +type contractsModule struct{ module, dir, defs string } + +// contractsModules are the message's module directories whose manifest marks a contracts module (a package artifact +// naming `generated-from`), in directory order. A manifest that does not decode is no contracts module here: the judge +// says why it is refused. +func contractsModules(manifests map[string]string) []contractsModule { + var out []contractsModule + for dir, raw := range manifests { + m, err := catalogue.Decode([]byte(raw)) + if err != nil || m.Build == nil { + continue + } + for _, a := range m.Build.Artifacts { + if a.GeneratedFrom == "" { + continue + } + base := dir + if base == delivery.RootPath { + base = "" + } + out = append(out, contractsModule{module: m.Module, dir: dir, defs: path.Clean(path.Join(base, a.GeneratedFrom))}) + break + } + } + slices.SortFunc(out, func(a, b contractsModule) int { return strings.Compare(a.dir, b.dir) }) + return out +} + +// readDefinitions reads a contracts module's definitions at a commit: its seats' index, each seat's definition and the +// parts it includes, each resolved. A seat that does not parse is answered by its words; a definitions directory +// with no index, or an index that does not parse, is an error naming it. +func (ix *Indexer) readDefinitions(ctx context.Context, repository, commit string, c contractsModule) (map[string]seatdef.Definition, map[string]string, error) { + file := func(p string) ([]byte, error) { + if c.defs != "." && c.defs != "" { + p = c.defs + "/" + p + } + raw, err := ix.ReadFile(ctx, repository, commit, p) + if err != nil { + return nil, fmt.Errorf("%w: %s at %.12s: %w", errForgeRead, p, commit, err) + } + if raw == nil { + return nil, fmt.Errorf("%s at %.12s: the forge holds no such file", p, commit) + } + return raw, nil + } + raw, err := file("seats/index.json") + if err != nil { + return nil, nil, err + } + idx, err := seatdef.ParseIndex(raw) + if err != nil { + return nil, nil, fmt.Errorf("seats/index.json does not parse: %w", err) + } + defs, unparsed := map[string]seatdef.Definition{}, map[string]string{} + parts := map[string]seatdef.Part{} + for _, seat := range idx.Seats { + raw, err := file("seats/" + seat + "/seat.json") + if err != nil { + if errors.Is(err, errForgeRead) { + return nil, nil, err + } + unparsed[seat] = err.Error() + continue + } + d, err := seatdef.Parse(raw) + if err == nil && d.Seat != seat { + err = fmt.Errorf("it names the seat %q, not %q", d.Seat, seat) + } + if err == nil { + for _, name := range d.Includes { + if _, ok := parts[name]; ok { + continue + } + praw, perr := file("parts/" + name + ".json") + if perr == nil { + var p seatdef.Part + if p, perr = seatdef.ParsePart(praw); perr == nil { + parts[name] = p + } + } + if errors.Is(perr, errForgeRead) { + return nil, nil, perr + } + if perr != nil { + err = fmt.Errorf("its part %s: %w", name, perr) + break + } + } + } + if err == nil { + d, err = seatdef.Resolve(d, parts) + } + if err != nil { + unparsed[seat] = err.Error() + continue + } + defs[seat] = d + } + return defs, unparsed, nil +} + +// digestOf is a resolved definition's bytes and their sha256. +func digestOf(d seatdef.Definition) (json.RawMessage, string, error) { + raw, err := json.Marshal(d) + if err != nil { + return nil, "", err + } + sum := sha256.Sum256(raw) + return raw, hex.EncodeToString(sum[:]), nil +} + +// ApplySeatDefinitions keeps the seat definitions a forge message carries (DLV-I17, I18): a merged message's contracts +// modules at its merge as the trunk's, a pull message's at its head as that pull request's; a closed pull request, or +// a merged one, has its entries removed. Nothing is done without the state or a forge reader. +func (ix *Indexer) ApplySeatDefinitions(ctx context.Context, f delivery.Forge) error { + if ix.SeatDefinitions == nil { + return nil + } + if f.Kind == delivery.ForgeMerged || (f.Kind == delivery.ForgePull && f.Action == delivery.PullClosed) { + if f.Number > 0 { + if err := ix.dropPull(ctx, f.Repository, f.Number); err != nil { + return err + } + } + if f.Kind != delivery.ForgeMerged { + return nil + } + } + if ix.ReadFile == nil || (f.Kind != delivery.ForgeMerged && f.Kind != delivery.ForgePull) { + return nil + } + manifests := f.Manifests + if f.Kind == delivery.ForgePull && len(f.Dirs) > 0 { + manifests = map[string]string{} + for _, d := range f.Dirs { + if raw, ok := f.Manifests[d]; ok { + manifests[d] = raw + } + } + } + var errs []error + for _, c := range contractsModules(manifests) { + if f.Kind == delivery.ForgeMerged { + errs = append(errs, ix.keepTrunk(ctx, f.Repository, f.Merge, c)) + } else { + errs = append(errs, ix.keepPull(ctx, f.Repository, f.Number, f.Head, c)) + } + } + return errors.Join(errs...) +} + +// NewSeatDefinitionsAgain is the value indexer_seat_definitions_again writes: its key in the takeins state and the value. +// A pull request is named by its number and head together, or neither (the trunk's). +func NewSeatDefinitionsAgain(module string, pull int, head, why, by string, at time.Time) (string, []byte, error) { + key, data, err := NewTakeIn(module, why, by, at) + if err != nil { + return "", nil, err + } + if (pull > 0) != (head != "") { + return "", nil, errors.New("name a pull request by its number and its head together (pull, head), or neither for the trunk") + } + var a TakeInAsk + if err := json.Unmarshal(data, &a); err != nil { + return "", nil, err + } + a.SeatDefinitions, a.Pull, a.Head = true, pull, head + data, err = json.Marshal(a) + return seatDefinitionsPrefix + key, data, err +} + +// seatDefinitionsPrefix begins an again-ask's takeins key, so it never shares a take-in's. +const seatDefinitionsPrefix = "seat-definitions." + +// SeatDefinitionsAgain reads a contracts module's seat definitions again now (indexer_seat_definitions_again): its +// trunk's at the commit the modules state holds it at, or a pull request's at the head given, with the trunk's manifest. +func (ix *Indexer) SeatDefinitionsAgain(ctx context.Context, module string, pull int, head string) TakenIn { + out := TakenIn{Module: module} + if ix.SeatDefinitions == nil || ix.ReadFile == nil || ix.Modules == nil { + out.Answer = "the indexer holds no seat-definitions state or forge reader yet" + return out + } + key, err := delivery.ModulesKey(module) + if err != nil { + out.Answer = err.Error() + return out + } + raw, _, found, err := ix.Modules.Get(ctx, key) + if err != nil || !found { + out.Answer = fmt.Sprintf("the trunk holds no module %s (%v)", module, err) + return out + } + m, err := delivery.DecodeModule(raw) + if err != nil { + out.Answer = err.Error() + return out + } + manifest, _, err := ix.manifestRaw(ctx, m.Manifest) + if err != nil { + out.Answer = "its manifest: " + err.Error() + return out + } + dir := m.Path + if dir == "" { + dir = delivery.RootPath + } + cs := contractsModules(map[string]string{dir: manifest}) + if len(cs) == 0 { + out.Answer = module + " is no contracts module: no package artifact of its trunk's manifest names generated-from" + return out + } + out.Repository, out.Commit, out.Path = m.Repository, m.Commit, m.Path + if pull > 0 { + err, out.Commit = ix.keepPull(ctx, m.Repository, pull, head, cs[0]), head + } else { + err = ix.keepTrunk(ctx, m.Repository, m.Commit, cs[0]) + } + if err != nil { + out.Answer = "read again, not kept: " + err.Error() + return out + } + out.Stored, out.Answer = true, "its seat definitions were read again at "+out.Commit + return out +} + +// keepTrunk writes a contracts module's trunk definitions at a merge: each that parses where its digest moved (said), +// each the module's index no longer names removed (said with no digest), and the module's unparsed finding written or +// cleared. A definition that does not parse leaves the trunk's entry as it stands. +func (ix *Indexer) keepTrunk(ctx context.Context, repository, commit string, c contractsModule) error { + defs, unparsed, err := ix.readDefinitions(ctx, repository, commit, c) + if errors.Is(err, errForgeRead) { + return err // the forge did not answer: the message is delivered again + } + if err != nil { + ix.unparsedFinding(ctx, c.module, map[string]string{"seats/index.json": err.Error()}) + return nil + } + ix.unparsedFinding(ctx, c.module, unparsed) + var errs []error + for seat, d := range defs { + errs = append(errs, ix.putDefinition(ctx, SeatDefinitionEntry{Seat: seat, Source: SourceTrunk, Repository: repository, + Module: c.module, Directory: c.dir, Commit: commit}, d)) + } + // A seat the module's trunk no longer defines leaves the state. + keys, err := ix.SeatDefinitions.Keys(ctx) + if err != nil { + return errors.Join(append(errs, err)...) + } + for _, key := range keys { + seat, ok := strings.CutSuffix(key, "."+SourceTrunk) + if !ok || defs[seat].Seat != "" || unparsed[seat] != "" { + continue + } + gone := false + err := write(ctx, ix.SeatDefinitions, key, func(value []byte, found bool) ([]byte, bool, error) { + var e SeatDefinitionEntry + gone = found && json.Unmarshal(value, &e) == nil && e.Repository == repository && e.Module == c.module + return nil, gone, nil + }) + errs = append(errs, err) + if err == nil && gone { + ix.sayChanged(ctx, SeatDefinitionChanged{Seat: seat, Source: SourceTrunk, Repository: repository}) + } + } + return errors.Join(errs...) +} + +// keepPull writes a pull request's head definitions of a contracts module: each that parses, and each that does not +// with its words (its check fails on it; the trunk's entry stays). +func (ix *Indexer) keepPull(ctx context.Context, repository string, pull int, head string, c contractsModule) error { + defs, unparsed, err := ix.readDefinitions(ctx, repository, head, c) + if errors.Is(err, errForgeRead) { + return err + } + if err != nil { + unparsed = map[string]string{"": err.Error()} + } + base := SeatDefinitionEntry{Source: SourcePull, Repository: repository, Pull: pull, Module: c.module, Directory: c.dir, Commit: head} + var errs []error + for seat, d := range defs { + e := base + e.Seat = seat + errs = append(errs, ix.putDefinition(ctx, e, d)) + } + for seat, why := range unparsed { + e := base + e.Seat, e.Unparsed, e.At = seat, why, ix.now().UTC() + if seat == "" { + e.Seat = c.module // the module's index did not parse: no seat is known, the entry is named by the module + } + data, err := json.Marshal(e) + if err == nil { + err = write(ctx, ix.SeatDefinitions, SeatDefinitionKey(e.Seat, SourcePull, repository, pull), func([]byte, bool) ([]byte, bool, error) { + return data, false, nil + }) + } + errs = append(errs, err) + } + return errors.Join(errs...) +} + +// putDefinition writes one parsed definition at its key, said when its digest moved. +func (ix *Indexer) putDefinition(ctx context.Context, e SeatDefinitionEntry, d seatdef.Definition) error { + raw, digest, err := digestOf(d) + if err != nil { + return err + } + e.Definition, e.Digest, e.At = raw, digest, ix.now().UTC() + moved := false + err = write(ctx, ix.SeatDefinitions, SeatDefinitionKey(e.Seat, e.Source, e.Repository, e.Pull), func(value []byte, found bool) ([]byte, bool, error) { + var cur SeatDefinitionEntry + if found && json.Unmarshal(value, &cur) == nil && cur.Digest == digest && cur.Unparsed == "" { + moved = false + return nil, false, nil + } + moved = true + data, err := json.Marshal(e) + return data, false, err + }) + if err == nil && moved { + ix.sayChanged(ctx, SeatDefinitionChanged{Seat: e.Seat, Source: e.Source, Repository: e.Repository, Pull: e.Pull, Digest: digest}) + } + return err +} + +// dropPull removes a pull request's entries and its out-of-step files. +func (ix *Indexer) dropPull(ctx context.Context, repository string, pull int) error { + keys, err := ix.SeatDefinitions.Keys(ctx) + if err != nil { + return err + } + suffix, oos := "."+pullSuffix(repository, pull), OutOfStepKey(repository, pull) + var errs []error + for _, key := range keys { + if !strings.HasSuffix(key, suffix) && key != oos { + continue + } + errs = append(errs, write(ctx, ix.SeatDefinitions, key, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil })) + if seat, ok := strings.CutSuffix(key, suffix); ok { + ix.sayChanged(ctx, SeatDefinitionChanged{Seat: seat, Source: SourcePull, Repository: repository, Pull: pull}) + } + } + return errors.Join(errs...) +} + +func (ix *Indexer) sayChanged(ctx context.Context, c SeatDefinitionChanged) { + if ix.SaySeatDefinitionChanged == nil { + return + } + if err := ix.SaySeatDefinitionChanged(ctx, c); err != nil { + ix.logf("seat definitions: %s of %s (%s) was not said: %v", SeatDefinitionChangedEvent, c.Seat, c.Source, err) + } +} + +// findingID is the finding's subject id for a contracts module. +func findingID(what, module string) string { return delivery.IndexerModule + "." + what + "." + module } + +// unparsedFinding writes the contracts module's seat-definition-unparsed finding for its trunk's seats that did not +// parse (by seat), or clears it when every one parsed. +func (ix *Indexer) unparsedFinding(ctx context.Context, module string, unparsed map[string]string) { + var lines []string + for seat, why := range unparsed { + lines = append(lines, seat+": "+why) + } + slices.Sort(lines) + ix.finding(ctx, findingID("seat-definitions", module), SeatDefinitionUnparsedKind, len(lines) > 0, + fmt.Sprintf("%s's trunk holds seat definitions that do not parse: %s", module, strings.Join(lines, "; ")), + "A seat definition on "+module+"'s trunk does not parse", + "The trunk's last definition that parsed stays the seat's contract. A change fixing the definition clears this when it "+ + "merges; indexer_seat_definitions_again reads the trunk again now.", + "Every seat definition on "+module+"'s trunk parses.") +} + +// finding writes (raise) or deletes the indexer's finding of a kind for a subject id. +func (ix *Indexer) finding(ctx context.Context, id, kind string, raise bool, summary, headline, explanation, resolved string) { + if ix.Findings == nil { + return + } + key := health.FindingKey(health.ScopeModule, id, kind) + err := write(ctx, ix.Findings, key, func(value []byte, found bool) ([]byte, bool, error) { + if !raise { + return nil, found, nil + } + if found { + if f, err := health.DecodeFinding(value); err == nil && f.Summary == summary { + return nil, false, nil + } + } + at := ix.now().UTC() + data, err := health.EncodeFinding(health.Finding{Key: key, Kind: kind, Subject: health.Subject{Scope: health.ScopeModule, ID: id}, + Severity: health.Warning, Summary: summary, Headline: headline, Explanation: explanation, Resolved: resolved, + Evidence: []health.Evidence{{At: at, Said: summary}}, Source: kind, Observer: delivery.IndexerModule, At: at}) + return data, false, err + }) + if err != nil { + ix.logf("seat definitions: the finding %s is not written: %v", key, err) + } +} + +// ApplyOutOfStepBuild takes a build end of the builder's log (DLV-I20; design 26 *Out of step is learned from events*): +// a trunk build of a contracts module whose test step named generated files out of step writes the module's +// generated-types-out-of-step finding naming them; a clean done build of the same module clears it. Any other entry +// says nothing. The files ride the entry as `out-of-step` (the builder's build end). +func (ix *Indexer) ApplyOutOfStepBuild(ctx context.Context, value []byte) { + var e struct { + Event string `json:"event"` + Module string `json:"module"` + Ref string `json:"ref"` + Outcome string `json:"outcome"` + Commit string `json:"commit"` + OutOfStep []string `json:"out-of-step"` + } + if json.Unmarshal(value, &e) != nil || e.Event != delivery.BuildEndedEvent || e.Module == "" { + return + } + if !trunkLike[e.Ref] { + return // a pull request's build: its check's verdict says its files + } + switch { + case len(e.OutOfStep) > 0: + files := slices.Clone(e.OutOfStep) + slices.Sort(files) + ix.finding(ctx, findingID("out-of-step", e.Module), OutOfStepKind, true, + fmt.Sprintf("%s's trunk build at %.12s has generated files out of step with their seat definitions: %s", e.Module, e.Commit, strings.Join(files, ", ")), + "Generated types on "+e.Module+"'s trunk are out of step with their seat definitions", + "Its build's test step regenerated the types and found the committed files differ, missing or not written by the "+ + "generator (a generator's change does this). A change regenerating them clears this once a build of "+e.Module+" ends clean.", + "A build of "+e.Module+" ended clean: its generated types match their definitions.") + case e.Outcome == delivery.BuildDone: + ix.finding(ctx, findingID("out-of-step", e.Module), OutOfStepKind, false, "", "", "", "") + } +} + +// ApplyVerdict keeps a pull request's out-of-step files from its check's verdict (the checker's event verdict): a +// verdict naming files writes them, one naming none (a clean verdict) removes them. +func (ix *Indexer) ApplyVerdict(ctx context.Context, repository string, pull int, head string, files []string) error { + if ix.SeatDefinitions == nil || repository == "" || pull <= 0 { + return nil + } + return write(ctx, ix.SeatDefinitions, OutOfStepKey(repository, pull), func(_ []byte, found bool) ([]byte, bool, error) { + if len(files) == 0 { + return nil, found, nil + } + fs := slices.Clone(files) + slices.Sort(fs) + data, err := json.Marshal(OutOfStep{Repository: repository, Pull: pull, Head: head, Files: fs, At: ix.now().UTC()}) + return data, false, err + }) +} + +// SeatDefinitionAnswer is indexer_seat_definition's answer: which source it answered, and the entry. +type SeatDefinitionAnswer struct { + Seat string `json:"seat"` + Answered string `json:"answered"` + Found bool `json:"found"` + Entry *SeatDefinitionEntry `json:"entry,omitempty"` + Trunk *SeatDefinitionEntry `json:"trunk,omitempty"` + Revision uint64 `json:"revision"` +} + +// SeatDefinitionOf answers indexer_seat_definition from the state: the trunk's entry, unless a pull request (repository +// and number) is named, whose entry is answered with the trunk's beside it; Answered says which. A pull request holding +// no entry for the seat answers the trunk's, saying so. The revision is the views mark the state was said up to. +func SeatDefinitionOf(defs, marks views.Source, seat, repository string, pull int) (SeatDefinitionAnswer, error) { + if seat == "" { + return SeatDefinitionAnswer{}, fmt.Errorf("%s names the seat", SeatDefinitionTool) + } + get := func(key string) (*SeatDefinitionEntry, error) { + v, ok, err := defs.Get(key) + if err != nil { + return nil, fmt.Errorf("%s.%s cannot be read: %w", SeatDefinitionsState, key, err) + } + if !ok || v == nil { + return nil, nil + } + var e SeatDefinitionEntry + if err := json.Unmarshal(v.Value, &e); err != nil { + return nil, fmt.Errorf("%s.%s holds what is not a seat definition: %w", SeatDefinitionsState, key, err) + } + return &e, nil + } + out := SeatDefinitionAnswer{Seat: seat, Answered: SourceTrunk, Revision: markOf(marks)} + trunk, err := get(SeatDefinitionKey(seat, SourceTrunk, "", 0)) + if err != nil { + return out, err + } + out.Entry, out.Found = trunk, trunk != nil + if pull > 0 { + p, err := get(SeatDefinitionKey(seat, SourcePull, repository, pull)) + if err != nil { + return out, err + } + if p != nil { + out.Answered, out.Entry, out.Trunk, out.Found = fmt.Sprintf("%s#%d", repository, pull), p, trunk, true + } else { + out.Answered = SourceTrunk + fmt.Sprintf(" (%s#%d holds no definition of %s)", repository, pull, seat) + } + } + return out, nil +} + +// markOf is the revision the views mark of the seat-definitions state names, 0 when none is. +func markOf(marks views.Source) uint64 { + if marks == nil { + return 0 + } + v, ok, err := marks.Get(SeatDefinitionsState) + if err != nil || !ok || v == nil { + return 0 + } + m, err := views.DecodeMark(v.Value) + if err != nil { + return 0 + } + return m.Revision +} + +// OutOfStepRow is one row of indexer_out_of_step: a trunk's contracts module (from its finding) or a pull request. +type OutOfStepRow struct { + Source string `json:"source"` + Module string `json:"module,omitempty"` + Pull string `json:"pull,omitempty"` + Head string `json:"head,omitempty"` + Files []string `json:"files,omitempty"` + Summary string `json:"summary,omitempty"` +} + +// OutOfStepOf answers indexer_out_of_step: every trunk contracts module whose finding generated-types-out-of-step +// stands, then every pull request whose check's verdict named out-of-step files. +func OutOfStepOf(defs, findings views.Source) ([]OutOfStepRow, error) { + out := []OutOfStepRow{} + fkeys, err := findings.Keys() + if err != nil { + return nil, fmt.Errorf("the findings state cannot be listed: %w", err) + } + slices.Sort(fkeys) + for _, key := range fkeys { + v, ok, err := findings.Get(key) + if err != nil || !ok || v == nil { + continue + } + f, err := health.DecodeFinding(v.Value) + if err != nil || f.Kind != OutOfStepKind { + continue + } + out = append(out, OutOfStepRow{Source: SourceTrunk, Module: strings.TrimPrefix(f.Subject.ID, delivery.IndexerModule+".out-of-step."), Summary: f.Summary}) + } + keys, err := defs.Keys() + if err != nil { + return nil, fmt.Errorf("the %s state cannot be listed: %w", SeatDefinitionsState, err) + } + slices.Sort(keys) + for _, key := range keys { + if !strings.HasPrefix(key, outOfStepPrefix) { + continue + } + v, ok, err := defs.Get(key) + if err != nil || !ok || v == nil { + continue + } + var o OutOfStep + if json.Unmarshal(v.Value, &o) != nil { + continue + } + out = append(out, OutOfStepRow{Source: SourcePull, Pull: fmt.Sprintf("%s#%d", o.Repository, o.Pull), Head: o.Head, Files: o.Files}) + } + return out, nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 66f0838c..5cb8e06c 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -35,11 +35,16 @@ type TakeInAsk struct { // Digest names, with Judge, the one manifest digest of the module to judge (indexer_judge with a digest). Digest string `json:"digest,omitempty"` // Say asks the module's manifest-accepted to be said again now (indexer_say_accepted, DLV-I9c). - Say bool `json:"say,omitempty"` - By string `json:"by,omitempty"` - At time.Time `json:"at"` - Result *TakenIn `json:"result,omitempty"` - Answered time.Time `json:"answered,omitzero"` + Say bool `json:"say,omitempty"` + // SeatDefinitions asks a contracts module's seat definitions to be read again now (indexer_seat_definitions_again, + // DLV-I19): its trunk's, or with Pull and Head a pull request's head. + SeatDefinitions bool `json:"seat-definitions,omitempty"` + Pull int `json:"pull,omitempty"` + Head string `json:"head,omitempty"` + By string `json:"by,omitempty"` + At time.Time `json:"at"` + Result *TakenIn `json:"result,omitempty"` + Answered time.Time `json:"answered,omitzero"` } // Done says the ask is answered. diff --git a/modules/mesh-delivery-indexer/internal/index/views.go b/modules/mesh-delivery-indexer/internal/index/views.go index 5a3ceaba..f9b52391 100644 --- a/modules/mesh-delivery-indexer/internal/index/views.go +++ b/modules/mesh-delivery-indexer/internal/index/views.go @@ -42,7 +42,7 @@ type ViewedTool struct { Tool, State, Words string } -// ViewedStates are the states the indexer's views family says, each by its bucket: the ten its readers read. +// ViewedStates are the states the indexer's views family says, each by its bucket: the ones its readers read. var ViewedStates = map[string]string{ "manifests": delivery.ManifestsBucket, "modules": delivery.ModulesBucket, @@ -55,4 +55,5 @@ var ViewedStates = map[string]string{ HeardState: HeardBucket, "taken": delivery.TakenBucket, health.FindingsState: health.FindingsBucket(delivery.IndexerModule), + SeatDefinitionsState: SeatDefinitionsBucket, } diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 1d92e787..55aac124 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -114,7 +114,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // 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: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, {Name: graph.State, History: graph.History}, {Name: views.MarksState, History: 1}, {Name: health.FindingsState, History: 5}, {Name: views.StatusState, History: 1}, - {Name: index.RequirementsState, History: 5}}; !reflect.DeepEqual(m.State, want) { // DLV-I7 + {Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17 t.Errorf("state = %+v, want %+v", m.State, want) } // DLV-F1r (ADR 0615 §1): the assigner's placement is read through its first view and views family, never its state. @@ -125,7 +125,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // internal/broker/state.go:62-93). // The registry seats' published and the builder's numbered fill each package's series (DLV-I5, design 51 *The versions*). if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-assigner.views.placement", - "mesh-delivery-builder.version.numbered", "mesh-delivery-checker.preview.written", "mesh-delivery-forge-bridge.forge.**", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + "mesh-delivery-builder.version.numbered", "mesh-delivery-checker.preview.written", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { @@ -152,8 +152,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build t.Errorf("%s answers the state %s, which the views family does not say", v.Tool, v.State) } } - want = append(want, "indexer_first_views", "indexer_first_view_again", index.ConfigReadersTool, index.BusStateTool, index.RequirementsTool, index.PackageUsersTool, index.SayAcceptedTool) // DLV-F3r, DLV-I10, DLV-I11, DLV-I7, DLV-I8, DLV-I9c - if !reflect.DeepEqual(m.Tools, want) || len(want) != 24 { + want = append(want, "indexer_first_views", "indexer_first_view_again", index.ConfigReadersTool, index.BusStateTool, index.RequirementsTool, index.PackageUsersTool, index.SayAcceptedTool, // DLV-F3r, DLV-I10, DLV-I11, DLV-I7, DLV-I8, DLV-I9c + index.SeatDefinitionTool, index.SeatDefinitionsAgainTool, index.OutOfStepTool) // DLV-I19 + if !reflect.DeepEqual(m.Tools, want) || len(want) != 27 { t.Errorf("tools = %v, want %v", m.Tools, want) } if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs { @@ -178,7 +179,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } } // ADR 0615 §5, ADR 0619 §2: its views family on its old streams section, 7 days and 256 MiB, dropping its oldest. - if want := []stream{{Name: index.ViewsStream, Events: []string{views.EventPattern}, MaxAgeDays: views.StreamDays, MaxMiB: views.StreamMiB}}; !reflect.DeepEqual(m.Streams, want) { + // Design 26: seat-definition-changed is kept seven days (DLV-I17). + if want := []stream{{Name: index.ViewsStream, Events: []string{views.EventPattern}, MaxAgeDays: views.StreamDays, MaxMiB: views.StreamMiB}, + {Name: index.SeatDefinitionsStream, Events: []string{index.SeatDefinitionChangedEvent}, MaxAgeDays: 7}}; !reflect.DeepEqual(m.Streams, want) { t.Errorf("streams = %+v, want %+v", m.Streams, want) } // ADR 0604 §2: it holds the seat mesh-catalog, serving its four verbs from its tools bundle (which stays without a diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 569f4f8d..b31db3e5 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -39,7 +39,10 @@ "indexer_bus_state", "indexer_requirements", "indexer_package_users", - "indexer_say_accepted" + "indexer_say_accepted", + "indexer_seat_definition", + "indexer_seat_definitions_again", + "indexer_out_of_step" ], "consumes": [ "go-package-registry.published", @@ -48,6 +51,7 @@ "mesh-delivery-assigner.views.placement", "mesh-delivery-builder.version.numbered", "mesh-delivery-checker.preview.written", + "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", "npm-package-registry.published" ], @@ -67,10 +71,12 @@ { "name": "view-marks", "history": 1 }, { "name": "findings", "history": 5 }, { "name": "first-views", "history": 1 }, - { "name": "requirements", "history": 5 } + { "name": "requirements", "history": 5 }, + { "name": "seat-definitions", "history": 5 } ], "streams": [ - { "name": "indexer-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 } + { "name": "indexer-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 }, + { "name": "indexer-seat-definitions", "events": ["seat-definition-changed"], "max-age-days": 7 } ], "emits": [ "manifest-refused",