DLV-I17, I18, I19, I20: the indexer keeps every contracts module's seat definitions on the bus

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
This commit is contained in:
2026-11-04 14:18:17 +01:00
parent ea6fbd31db
commit aa647d0eda
14 changed files with 1019 additions and 18 deletions
@@ -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)
}
@@ -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
}}
}
@@ -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
@@ -128,7 +128,10 @@ func goFileVia(call toolCaller, name string) func(ctx context.Context, repositor
return nil, fmt.Errorf("%q is not <owner>/<name>", 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})
@@ -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()
}
@@ -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 {
+1
View File
@@ -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
)
@@ -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
@@ -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()}
@@ -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: `<seat>.trunk`, or `<seat>.pull.<repository escaped>.<number>`.
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
}
@@ -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.
@@ -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,
}
@@ -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
+9 -3
View File
@@ -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",