From 37eff6fe5c0c17f244bb64c575ce6a35df785fff Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:11:33 +0200 Subject: [PATCH] =?UTF-8?q?Lay=20out=20the=20ask=20a=20build=20node=20pull?= =?UTF-8?q?s:=20its=20subjects,=20stream,=20workers=20and=20job,=20so=20bu?= =?UTF-8?q?ilder,=20checker=20and=20build=20modules=20share=20one=20form?= =?UTF-8?q?=20(issue=201887=20A1a,=20ADR=200525,=20ADR=200536=20=C2=A74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go/delivery/stepasks.go | 345 +++++++++++++++++++++++++++++++++++ go/delivery/stepasks_test.go | 126 +++++++++++++ 2 files changed, 471 insertions(+) create mode 100644 go/delivery/stepasks.go create mode 100644 go/delivery/stepasks_test.go diff --git a/go/delivery/stepasks.go b/go/delivery/stepasks.go new file mode 100644 index 0000000..6754074 --- /dev/null +++ b/go/delivery/stepasks.go @@ -0,0 +1,345 @@ +package delivery + +import ( + "encoding/json" + "fmt" + "regexp" + "strings" + "time" +) + +// **Asks: a build or a check step is a message a build node pulls when it has a free slot** (novox/hq ADR 0525, as +// ADR 0536 §4 changes it; issue 1887). The builder puts a trunk fingerprint's build on the delivery stream of the +// artifact's language and a candidate's on the check stream; the checker puts a check of one level on the check stream, +// and a re-judge after a trunk move on the recheck stream. Each build module claims the node seat of its language +// (`node-build-go`, `node-build-node`, `node-build-image`), whose accepted kinds the broker derives as the work queue +// `SEAT_NODE_BUILD_` with one shared pull worker per kind; a holder pulls delivery first, then check, then +// recheck, at most its node's build capacity at once (ADR 0525 §4, §5). +// +// **An ask is one whole job on one node** (ADR 0525 §3): a build ask is the fetch, the build (or copy) of each artifact +// and the push; a check ask is the fetch, the digest and the tests of one level. Its end is written on the build module's +// step key `.` (StepsBucket), which the asker watches with AskStepWatch (ADR 0525 §6). + +// The kinds of ask: the seat's accepted kinds, in the order a holder pulls them. +const ( + AskDelivery = "delivery" + AskCheck = "check" + AskRecheck = "recheck" +) + +// AskKinds are the accepted kinds of a build seat, in the order a holder pulls them (ADR 0536 §4). +var AskKinds = []string{AskDelivery, AskCheck, AskRecheck} + +// The languages of the build seats: each is one build module's (mesh-build-) and one seat's +// (node-build-). +const ( + AskGo = "go" + AskNode = "node" + AskImage = "image" +) + +// AskLanguages are the build seats' languages. +var AskLanguages = []string{AskGo, AskNode, AskImage} + +// StepAskVersion is the version of an Ask value; a reader refuses another (ErrUnknownVersion). +const StepAskVersion = 1 + +func checkAskKind(kind string) error { + for _, k := range AskKinds { + if kind == k { + return nil + } + } + return fmt.Errorf("delivery: the ask kind %q is not delivery, check or recheck", kind) +} + +func checkAskLanguage(language string) error { + for _, l := range AskLanguages { + if language == l { + return nil + } + } + return fmt.Errorf("delivery: the ask language %q is not go, node or image", language) +} + +// BuildSeat is the node seat a build module of language claims: node-build-. +func BuildSeat(language string) (string, error) { + if err := checkAskLanguage(language); err != nil { + return "", err + } + return "node-build-" + language, nil +} + +// BuildModuleOf is the build module that holds language's seat: mesh-build-. +func BuildModuleOf(language string) (string, error) { + if err := checkAskLanguage(language); err != nil { + return "", err + } + return "mesh-build-" + language, nil +} + +// AskSubject is the subject an ask of kind for language is published on: the seat's accept subject, +// `mesh.seat.node-build-.accept.` (mesh-controller internal/broker SeatStreams). +func AskSubject(language, kind string) (string, error) { + seat, err := BuildSeat(language) + if err != nil { + return "", err + } + if err := checkAskKind(kind); err != nil { + return "", err + } + return "mesh.seat." + seat + ".accept." + kind, nil +} + +// ParseAskSubject is the inverse of AskSubject; it refuses a subject of another seat or kind. +func ParseAskSubject(subject string) (language, kind string, err error) { + rest, ok := strings.CutPrefix(subject, "mesh.seat.node-build-") + if !ok { + return "", "", fmt.Errorf("delivery: %q is not a build seat's accept subject", subject) + } + language, kind, ok = strings.Cut(rest, ".accept.") + if !ok { + return "", "", fmt.Errorf("delivery: %q is not a build seat's accept subject", subject) + } + if want, err := AskSubject(language, kind); err != nil || want != subject { + return "", "", fmt.Errorf("delivery: %q is not a build seat's accept subject: %v", subject, err) + } + return language, kind, nil +} + +// AskStream is the work-queue stream of language's seat, as the broker names a seat's stream: SEAT_NODE_BUILD_. +func AskStream(language string) (string, error) { + seat, err := BuildSeat(language) + if err != nil { + return "", err + } + return "SEAT_" + strings.ToUpper(strings.ReplaceAll(seat, "-", "_")), nil +} + +// AskWorker is the shared pull consumer of one kind on language's stream, filtered on that kind's subject: +// SEAT_NODE_BUILD___worker. Every holder of the seat pulls from it, so the work is shared without a +// holder knowing the others (ADR 0190), and the kinds' workers are apart so a holder can pull them in their order. +func AskWorker(language, kind string) (string, error) { + stream, err := AskStream(language) + if err != nil { + return "", err + } + if err := checkAskKind(kind); err != nil { + return "", err + } + return stream + "_" + strings.ToUpper(kind) + "_worker", nil +} + +// An ask id is a step id of the form buildstep.NewID gives: the time in milliseconds in base 36 (8 or 9 digits) and six +// random hex digits. Its form keeps it apart from a module's or a node's name in a key. +var askIDOK = regexp.MustCompile(`^[0-9a-z]{8,9}[0-9a-f]{6}$`) + +func checkAskID(id string) error { + if !askIDOK.MatchString(id) { + return fmt.Errorf("delivery: %q is not an ask id (base-36 milliseconds and six hex digits)", id) + } + return nil +} + +// StepAsk is the message on a build seat's stream: who asked what, of which kind and language, and the one whole job. +// Exactly one of Build and Check is set: a delivery ask is a build; a check ask a build (a candidate's fingerprint) or a +// check; a recheck ask a check; an image ask is always a build. +type StepAsk struct { + V int `json:"v"` + ID string `json:"id"` + Kind string `json:"kind"` + Language string `json:"language"` + // Asker is the module that put the ask (mesh-delivery-builder or mesh-delivery-checker); the stream's grant is what + // makes it true, this field only names it to a reader. + Asker string `json:"asker"` + At time.Time `json:"at"` + Build *BuildJob `json:"build,omitempty"` + Check *CheckJob `json:"check,omitempty"` +} + +// BuildJob is a build ask's job: fetch the commit (unless every artifact is a copy), then build or copy each artifact, +// pushing it when Publish says so, and answer each artifact's digest and reference (BuildAnswer). +type BuildJob struct { + // Module is the module's name; Path its directory in the repository (the build tool's `module`, "." for its root). + Module string `json:"module"` + Path string `json:"path,omitempty"` + // FP is the fingerprint the build is of (the builder's key), named for a reader of the ask. + FP string `json:"fp,omitempty"` + // Fetch is the fetch tool's arguments, as the build module's fetch tool takes them (buildstep.FetchRequest's Args); + // absent when every artifact is a copy. + Fetch map[string]any `json:"fetch,omitempty"` + // Artifacts are built in this order. + Artifacts []AskArtifact `json:"artifacts"` + // Publish says the build pushes an image or publishes a package: a trunk build only (ADR 0487 clause 12). + Publish bool `json:"publish,omitempty"` + // Toolchain is the toolchain the fingerprint was computed with; a build module running another refuses the ask. + Toolchain *ToolchainUsed `json:"toolchain,omitempty"` +} + +// AskArtifact is one artifact of a build ask: built from the fetched tree, or copied from From, an upstream pin (ADR +// 0498 clause 4). +type AskArtifact struct { + Artifact string `json:"artifact"` + From string `json:"from,omitempty"` +} + +// CheckJob is a check ask's job: fetch the head merged onto its trunk, digest the level's packages, take the kept +// verdicts for the digests that have one, and test the rest at Level, the packages the change's own tests are in first +// (CheckAnswer). +type CheckJob struct { + // Fetch is the fetch tool's arguments (repository, commit, ref, base, and the build module's own: beside, package). + Fetch map[string]any `json:"fetch"` + Level string `json:"level"` + // Kept says the build module takes the kept verdicts of the checker's kept bucket (CheckerKeptBucket, by KeptKey) + // for the packages whose digest has one, and tests only the others. A try that runs some tests (Run) keeps nothing. + Kept bool `json:"kept,omitempty"` + // Runner is the test step's runner ("workspace": the candidate's own build module judges it, ADR 0535 §4), used + // when the merge changes a path under RunnerWhen (every merge when RunnerWhen is empty). + Runner string `json:"runner,omitempty"` + RunnerWhen string `json:"runner_when,omitempty"` + // Run is a try's test pattern (go test -run): the level runs it alone, digesting nothing. + Run string `json:"run,omitempty"` +} + +func (a StepAsk) check() error { + if err := checkAskID(a.ID); err != nil { + return err + } + if err := checkAskKind(a.Kind); err != nil { + return err + } + if err := checkAskLanguage(a.Language); err != nil { + return err + } + if err := checkName("asker", a.Asker); err != nil { + return err + } + if a.At.IsZero() { + return fmt.Errorf("delivery: the ask %s names no time", a.ID) + } + switch { + case (a.Build == nil) == (a.Check == nil): + return fmt.Errorf("delivery: the ask %s holds a build or a check, exactly one", a.ID) + case a.Build != nil && a.Kind == AskRecheck: + return fmt.Errorf("delivery: the ask %s is a build on the recheck stream, which carries checks alone", a.ID) + case a.Check != nil && a.Kind == AskDelivery: + return fmt.Errorf("delivery: the ask %s is a check on the delivery stream, which carries trunk builds alone", a.ID) + case a.Check != nil && a.Language == AskImage: + return fmt.Errorf("delivery: the ask %s is a check of an image, which has no tests", a.ID) + } + if a.Build != nil { + return a.Build.check(a.ID) + } + return a.Check.check(a.ID) +} + +func (b BuildJob) check(id string) error { + if err := checkName("module", b.Module); err != nil { + return err + } + if len(b.Artifacts) == 0 { + return fmt.Errorf("delivery: the build ask %s names no artifact", id) + } + fetch := false + for _, a := range b.Artifacts { + if err := checkName("artifact", a.Artifact); err != nil { + return err + } + fetch = fetch || a.From == "" + } + if fetch { + if b.Path == "" { + return fmt.Errorf("delivery: the build ask %s names no directory (path) for the module %s", id, b.Module) + } + if err := checkFetchArgs(id, b.Fetch); err != nil { + return err + } + } + return nil +} + +func (c CheckJob) check(id string) error { + if err := checkLevel(c.Level); err != nil { + return err + } + return checkFetchArgs(id, c.Fetch) +} + +var askCommitOK = regexp.MustCompile(`^[0-9a-f]{40}$`) + +// checkFetchArgs refuses fetch arguments with no repository or no full commit; the rest is the build module's fetch to +// judge, as a direct call's is. +func checkFetchArgs(id string, args map[string]any) error { + repository, _ := args["repository"].(string) + commit, _ := args["commit"].(string) + if repository == "" || !strings.Contains(repository, "/") { + return fmt.Errorf("delivery: the ask %s fetches no repository /", id) + } + if !askCommitOK.MatchString(commit) { + return fmt.Errorf("delivery: the ask %s fetches no full commit", id) + } + return nil +} + +// EncodeStepAsk is a as JSON at StepAskVersion; it refuses what DecodeStepAsk refuses. +func EncodeStepAsk(a StepAsk) ([]byte, error) { + a.V = StepAskVersion + a.At = a.At.UTC() + if err := a.check(); err != nil { + return nil, err + } + return json.Marshal(a) +} + +// DecodeStepAsk reads an ask; it refuses an unknown version (ErrUnknownVersion), an unknown kind or language, an id that +// is not an ask id, a build and a check together or neither, a job on a stream that does not carry it, and a fetch with +// no repository or no full commit. +func DecodeStepAsk(data []byte) (StepAsk, error) { + var probe struct { + V *int `json:"v"` + } + if err := json.Unmarshal(data, &probe); err != nil { + return StepAsk{}, fmt.Errorf("delivery: an ask is not JSON: %w", err) + } + if probe.V == nil || *probe.V != StepAskVersion { + return StepAsk{}, fmt.Errorf("%w: an ask of version %v, this reader reads %d", ErrUnknownVersion, probe.V, StepAskVersion) + } + var a StepAsk + if err := json.Unmarshal(data, &a); err != nil { + return StepAsk{}, fmt.Errorf("delivery: an ask does not fit its shape: %w", err) + } + if err := a.check(); err != nil { + return StepAsk{}, err + } + return a, nil +} + +// BuildAnswer is a build ask's step result: each artifact's digest and, for a pushed one, its reference. +type BuildAnswer struct { + Artifacts []AskBuilt `json:"artifacts"` + Toolchain *ToolchainUsed `json:"toolchain,omitempty"` +} + +// AskBuilt is one artifact a build ask built. +type AskBuilt struct { + Artifact string `json:"artifact"` + Digest string `json:"digest"` + Reference string `json:"reference,omitempty"` +} + +// CheckAnswer is a check ask's step result, each part the build module's own tool's answer whole: the fetch's, the +// digest's (absent when nothing was digested), the kept verdicts taken, and one test result per test step run, in the +// order run. The checker judges and keeps from it as it did from the tools' answers. +type CheckAnswer struct { + Fetch json.RawMessage `json:"fetch"` + Digest json.RawMessage `json:"digest,omitempty"` + Kept []KeptTaken `json:"kept,omitempty"` + Tests []json.RawMessage `json:"tests,omitempty"` +} + +// KeptTaken is a package whose kept verdict the check took instead of testing it. +type KeptTaken struct { + Package string `json:"package"` + Digest string `json:"digest"` + Tests int `json:"tests"` +} diff --git a/go/delivery/stepasks_test.go b/go/delivery/stepasks_test.go new file mode 100644 index 0000000..578b8a8 --- /dev/null +++ b/go/delivery/stepasks_test.go @@ -0,0 +1,126 @@ +package delivery + +import ( + "errors" + "reflect" + "strings" + "testing" + "time" +) + +const askCommit = "0123456789abcdef0123456789abcdef01234567" + +func buildAsk() StepAsk { + return StepAsk{ID: "mh3k2x9qa1b2c3", Kind: AskDelivery, Language: AskGo, Asker: "mesh-delivery-builder", + At: time.Date(2026, 10, 23, 13, 0, 0, 0, time.UTC), + Build: &BuildJob{Module: "mesh-delivery-view", Path: "modules/mesh-delivery-view", FP: "ab12", + Fetch: map[string]any{"repository": "novox/mesh-delivery", "commit": askCommit, "dir": "modules/mesh-delivery-view", "paths": []any{"cmd"}}, + Artifacts: []AskArtifact{{Artifact: "bundle"}, {Artifact: "base", From: "registry/x@sha256:00"}}, + Publish: true, Toolchain: &ToolchainUsed{Version: "go1.26.0", Recipe: "3"}}} +} + +func checkAsk() StepAsk { + return StepAsk{ID: "mh3k2x9qa1b2c4", Kind: AskCheck, Language: AskNode, Asker: "mesh-delivery-checker", + At: time.Date(2026, 10, 23, 13, 0, 0, 0, time.UTC), + Check: &CheckJob{Fetch: map[string]any{"repository": "novox/mesh-catalog", "commit": askCommit, "ref": "refs/pull/3/head", "base": askCommit}, + Level: LevelUnit, Kept: true, Runner: "workspace", RunnerWhen: "modules/mesh-build-go/", Run: ""}} +} + +// 1887/A1a: an ask encodes and decodes with every field. +func TestAskRoundTrips(t *testing.T) { + for _, a := range []StepAsk{buildAsk(), checkAsk()} { + raw, err := EncodeStepAsk(a) + if err != nil { + t.Fatal(err) + } + got, err := DecodeStepAsk(raw) + if err != nil { + t.Fatal(err) + } + a.V = StepAskVersion + if !reflect.DeepEqual(got, a) { + t.Fatalf("round trip:\n got %+v\nwant %+v", got, a) + } + } +} + +// 1887/A1a: an ask with an unknown kind or language is refused. +func TestAskRefusesUnknownKindAndLanguage(t *testing.T) { + cases := map[string]func(*StepAsk){ + "kind": func(a *StepAsk) { a.Kind = "build" }, + "language": func(a *StepAsk) { a.Language = "typescript" }, + "id": func(a *StepAsk) { a.ID = "postgres" }, + "neither": func(a *StepAsk) { a.Build = nil }, + "both": func(a *StepAsk) { a.Check = &CheckJob{Level: LevelUnit, Fetch: a.Build.Fetch} }, + "recheck": func(a *StepAsk) { a.Kind = AskRecheck }, + "commit": func(a *StepAsk) { a.Build.Fetch = map[string]any{"repository": "novox/x", "commit": "main"} }, + "path": func(a *StepAsk) { a.Build.Path = "" }, + } + for name, change := range cases { + a := buildAsk() + change(&a) + if _, err := EncodeStepAsk(a); err == nil { + t.Errorf("%s: encoded", name) + } + } + // A copy alone fetches nothing and needs no path. + a := buildAsk() + a.Build.Fetch, a.Build.Path, a.Build.Artifacts = nil, "", []AskArtifact{{Artifact: "base", From: "registry/x@sha256:00"}} + if _, err := EncodeStepAsk(a); err != nil { + t.Errorf("a copy alone: %v", err) + } + c := checkAsk() + c.Kind = AskDelivery + if _, err := EncodeStepAsk(c); err == nil { + t.Error("a check on the delivery stream encoded") + } + c = checkAsk() + c.Language = AskImage + if _, err := EncodeStepAsk(c); err == nil { + t.Error("a check of an image encoded") + } + raw, _ := EncodeStepAsk(buildAsk()) + for _, bad := range []string{ + strings.Replace(string(raw), `"kind":"delivery"`, `"kind":"queue"`, 1), + strings.Replace(string(raw), `"language":"go"`, `"language":"rust"`, 1), + } { + if _, err := DecodeStepAsk([]byte(bad)); err == nil { + t.Errorf("decoded %s", bad) + } + } + if _, err := DecodeStepAsk([]byte(strings.Replace(string(raw), `"v":1`, `"v":2`, 1))); !errors.Is(err, ErrUnknownVersion) { + t.Errorf("version 2: %v", err) + } +} + +// 1887/A1a: the subjects, stream and workers are the seat's, as the broker derives them. +func TestAskSubjects(t *testing.T) { + s, err := AskSubject(AskGo, AskCheck) + if err != nil || s != "mesh.seat.node-build-go.accept.check" { + t.Fatalf("AskSubject = %q, %v", s, err) + } + language, kind, err := ParseAskSubject("mesh.seat.node-build-image.accept.delivery") + if err != nil || language != AskImage || kind != AskDelivery { + t.Fatalf("ParseAskSubject = %q %q %v", language, kind, err) + } + for _, bad := range []string{"mesh.seat.node-build-agent.accept.build", "mesh.seat.node-build-go.accept.queue", "mesh.seat.node-build-go.event.x"} { + if _, _, err := ParseAskSubject(bad); err == nil { + t.Errorf("parsed %s", bad) + } + } + if st, _ := AskStream(AskNode); st != "SEAT_NODE_BUILD_NODE" { + t.Errorf("AskStream = %q", st) + } + if w, _ := AskWorker(AskGo, AskRecheck); w != "SEAT_NODE_BUILD_GO_RECHECK_worker" { + t.Errorf("AskWorker = %q", w) + } + if seat, _ := BuildSeat(AskImage); seat != "node-build-image" { + t.Errorf("BuildSeat = %q", seat) + } + if m, _ := BuildModuleOf(AskNode); m != "mesh-build-node" { + t.Errorf("BuildModuleOf = %q", m) + } + if _, err := BuildSeat("agent"); err == nil { + t.Error("BuildSeat(agent)") + } +}