From 27072a01b5d4e1c04e3ddcf1d349a361eb0bc75d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 21:07:14 +0200 Subject: [PATCH] Add the checker module: it watches the candidates, claims a verdict once, owes the status mesh/check and ends a check past its bound (issue 1678) --- .../cmd/mesh-delivery-checker/bus.go | 159 ++++++ .../cmd/mesh-delivery-checker/judge.go | 35 ++ .../cmd/mesh-delivery-checker/main.go | 314 ++++++++++++ .../cmd/mesh-delivery-checker/settings.go | 29 ++ .../mesh-delivery-checker/settings_test.go | 68 +++ .../cmd/mesh-delivery-checker/toolchains.go | 104 ++++ .../mesh-delivery-checker/files/settings.json | 3 + .../internal/check/case.go | 135 ++++++ .../internal/check/checker.go | 451 ++++++++++++++++++ .../internal/check/checker_test.go | 367 ++++++++++++++ .../internal/check/line.go | 19 + .../mesh-delivery-checker/manifest_test.go | 197 ++++++++ modules/mesh-delivery-checker/module.json | 113 +++++ 13 files changed, 1994 insertions(+) create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings.go create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings_test.go create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go create mode 100644 modules/mesh-delivery-checker/files/settings.json create mode 100644 modules/mesh-delivery-checker/internal/check/case.go create mode 100644 modules/mesh-delivery-checker/internal/check/checker.go create mode 100644 modules/mesh-delivery-checker/internal/check/checker_test.go create mode 100644 modules/mesh-delivery-checker/internal/check/line.go create mode 100644 modules/mesh-delivery-checker/manifest_test.go create mode 100644 modules/mesh-delivery-checker/module.json diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go new file mode 100644 index 0000000..e3489c3 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go @@ -0,0 +1,159 @@ +package main + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-checker/internal/check" +) + +// bucketOpener is the part of the bus the watches use. +type bucketOpener interface { + KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) +} + +// kvStore is the builds bucket as the builder's store: create and compare-and-set. +type kvStore struct{ kv jetstream.KeyValue } + +func conflict(err error) error { + if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + return check.ErrConflict + } + return err +} + +func (s kvStore) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) { + e, err := s.kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, 0, false, nil + } + if err != nil { + return nil, 0, false, err + } + return e.Value(), e.Revision(), true, nil +} + +func (s kvStore) Create(ctx context.Context, key string, value []byte) (uint64, error) { + rev, err := s.kv.Create(ctx, key, value) + return rev, conflict(err) +} + +func (s kvStore) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + rev, err := s.kv.Update(ctx, key, value, revision) + return rev, conflict(err) +} + +func (s kvStore) Keys(ctx context.Context) ([]string, error) { + keys, err := s.kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + +func opOf(e jetstream.KeyValueEntry) bool { return e.Operation() == jetstream.KeyValuePut } + +// settle reads the entries a watcher holds now, up to its mark of the last values, and returns the watcher for the +// changes that follow. +func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyValueEntry)) (jetstream.KeyWatcher, error) { + w, err := kv.WatchAll(ctx) + if err != nil { + return nil, err + } + for { + select { + case <-ctx.Done(): + w.Stop() + return nil, ctx.Err() + case e, ok := <-w.Updates(): + if !ok { + return nil, errors.New("the watch ended") + } + if e == nil { + return w, nil + } + each(e) + } + } +} + +// follow hands each change of a watcher to each until ctx ends; a watch that ends is an error, and the process manager +// starts the program again. +func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.KeyValueEntry)) error { + defer w.Stop() + for { + select { + case <-ctx.Done(): + return nil + case e, ok := <-w.Updates(): + if !ok { + return errors.New("a watch ended") + } + if e != nil { + each(e) + } + } + } +} + +// stepsWaiter waits for a build module's step by watching its key in the module's steps bucket (ADR 0469 clause 2): the +// watch hands over the value the key holds now, so a step that ended before the watch began is read at once, and every +// change after. No timer: the caller bounds the wait with its context. +type stepsWaiter struct{ js bucketOpener } + +func (w stepsWaiter) Wait(ctx context.Context, module, key string) (delivery.Step, error) { + kv, err := w.js.KeyValue(ctx, delivery.StepsBucket(module)) + if err != nil { + return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) + } + watch, err := kv.Watch(ctx, key) + if err != nil { + return delivery.Step{}, err + } + defer watch.Stop() + for { + select { + case <-ctx.Done(): + return delivery.Step{}, ctx.Err() + case e, ok := <-watch.Updates(): + if !ok { + return delivery.Step{}, errors.New("the watch of the step ended") + } + if e == nil || !opOf(e) { + continue + } + step, err := delivery.DecodeStep(e.Value()) + if err != nil { + return delivery.Step{}, fmt.Errorf("the step %s is not a step: %w", key, err) + } + if step.State != delivery.StepRunning { + return step, nil + } + } + } +} + +// manifestReader reads manifests. of the indexer and parses the manifest it holds. +type manifestReader struct{ kv jetstream.KeyValue } + +func (m manifestReader) Manifest(ctx context.Context, digest string) (catalogue.Manifest, error) { + key, err := delivery.ManifestsKey(digest) + if err != nil { + return catalogue.Manifest{}, err + } + e, err := m.kv.Get(ctx, key) + if err != nil { + return catalogue.Manifest{}, fmt.Errorf("manifests.%s: %w", key, err) + } + kept, err := delivery.DecodeManifest(e.Value()) + if err != nil { + return catalogue.Manifest{}, err + } + return catalogue.ParseManifest([]byte(kept.JSON)) +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go new file mode 100644 index 0000000..4b4a008 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go @@ -0,0 +1,35 @@ +package main + +import ( + "context" + "errors" + "log" + + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" + + "mesh-delivery-checker/internal/check" +) + +// judgeDeps are the parts the judge of a check is made of. +type judgeDeps struct { + checker *check.Checker + runner buildrun.Runner + tools *buildrun.ToolchainSet + presence *buildrun.PresenceSet + have *buildrun.HaveSet + places *buildrun.Places + trunks *check.Trunks + kept check.KV + previews check.KV + logf func(format string, args ...any) +} + +var _ = log.Printf + +type unbuilt struct{} + +func (unbuilt) Judge(context.Context, *check.Case) (check.Outcome, error) { + return check.Outcome{}, errors.New("the judge of a check is not built yet") +} + +func newJudge(judgeDeps) check.Judge { return unbuilt{} } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go new file mode 100644 index 0000000..6faacfa --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -0,0 +1,314 @@ +// Command mesh-delivery-checker is delivery's checker (novox/hq ADR 0453 clause 4, ADR 0447, ADR 0484, issues 1678 to 1686): for +// each candidate the indexer writes it judges every pull request's head merged onto its trunk, through the build modules' +// fetch, digest and test steps, keeps the passing packages' verdicts, writes the verdict, and owes the forge the status +// `mesh/check`, which the forge bridge sets. It runs dark: nothing reads its answers to merge or deliver (ADR 0453 clause 4). +// It serves no request subject and holds no timer: every reaction is a watch or a bus event (ADR 0437 §8). +package main + +import ( + "context" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-checker/internal/check" +) + +const ( + // defaultStepBound is how long the checker waits for one step of a build module (a fetch downloads the Go modules, a + // test level runs a repository's tests). + defaultStepBound = 30 * time.Minute + // defaultLink is the base of the address a status points at; the verdict key follows. + defaultLink = "mesh-delivery-view/checks" +) + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if err := run(ctx, os.Getenv, os.Stderr); err != nil { + fmt.Fprintln(os.Stderr, "mesh-delivery-checker:", err) + os.Exit(1) + } +} + +// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate +// fingerprint to pin, the principal and its password (as the indexer reads it; issue 1242 moves this into the SDK). +type credential struct { + URL string `json:"url"` + Fingerprint string `json:"fingerprint"` + Node string `json:"node"` + Module string `json:"module"` + User string `json:"user"` + Password string `json:"password"` +} + +func readCredential(path string) (credential, error) { + raw, err := os.ReadFile(path) + if err != nil { + return credential{}, err + } + var c credential + if err := json.Unmarshal(raw, &c); err != nil { + return credential{}, fmt.Errorf("the broker credential is not one: %w", err) + } + if c.URL == "" || c.User == "" { + return credential{}, errors.New("the broker credential names no bus address or user") + } + return c, nil +} + +// pinned accepts exactly the certificate with this SHA-256 and no other. +func pinned(want string) *tls.Config { + return &tls.Config{ + InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter + MinVersion: tls.VersionTLS12, + VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error { + if len(rawCerts) == 0 { + return errors.New("the bus presented no certificate") + } + sum := sha256.Sum256(rawCerts[0]) + if got := "sha256:" + hex.EncodeToString(sum[:]); got != want { + return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got) + } + return nil + }, + } +} + +// durationEnv reads a duration setting from the environment, def when it is not set. +func durationEnv(getenv func(string) string, name string, def time.Duration) (time.Duration, error) { + v := strings.TrimSpace(getenv(name)) + if v == "" { + return def, nil + } + d, err := time.ParseDuration(v) + if err != nil || d <= 0 { + return 0, fmt.Errorf("%s: %q is not a duration of more than 0", name, v) + } + return d, nil +} + +// run is the program: it returns when ctx ends or the bus cannot be used. +func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { + logger := log.New(logOut, "mesh-delivery-checker: ", log.LstdFlags) + path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) + if path == "" { + return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with") + } + cred, err := readCredential(path) + if err != nil { + return err + } + opts := []nats.Option{ + nats.Name(cred.Node + "." + cred.Module), + nats.UserInfo(cred.User, cred.Password), + nats.CustomInboxPrefix("_INBOX." + cred.User), + nats.MaxReconnects(-1), + nats.RetryOnFailedConnect(true), + } + if strings.TrimSpace(cred.Fingerprint) != "" { + opts = append(opts, nats.Secure(pinned(cred.Fingerprint))) + } + conn, err := nats.Connect(cred.URL, opts...) + if err != nil { + return fmt.Errorf("connecting to the bus: %w", err) + } + defer conn.Close() + js, err := jetstream.New(conn) + if err != nil { + return err + } + bucket := func(name string) (jetstream.KeyValue, error) { + kv, err := js.KeyValue(ctx, name) + if err != nil { + return nil, fmt.Errorf("the bucket %s: %w", name, err) + } + return kv, nil + } + buckets := map[string]jetstream.KeyValue{} + for _, name := range []string{delivery.CheckerVerdictsBucket, delivery.CheckerKeptBucket, delivery.CheckerOwedBucket, delivery.PreviewsBucket, + delivery.CandidatesBucket, delivery.ModulesBucket, delivery.ManifestsBucket} { + if buckets[name], err = bucket(name); err != nil { + return err + } + } + // Calls to the build modules go out as this module's own account, naming the machine (mesh-sdk process.CallOn); the + // buckets are read and written on the connection above. + account, err := process.ReadAccount(path) + if err != nil { + return err + } + caller, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) + if err != nil { + return err + } + defer caller.Close() + + stepBound, err := durationEnv(getenv, "MESH_CHECKER_STEP_BOUND", defaultStepBound) + if err != nil { + return err + } + checkBound, err := durationEnv(getenv, "MESH_CHECKER_BOUND", check.DefaultBound) + if err != nil { + return err + } + capacity, err := capacityFrom(strings.TrimSpace(getenv("MESH_CHECKER_SETTINGS")), getenv("MESH_CHECKER_CAPACITY")) + if err != nil { + return fmt.Errorf("the capacity setting: %w", err) + } + link := strings.TrimSpace(getenv("MESH_CHECKER_LINK")) + if link == "" { + link = defaultLink + } + tools := &buildrun.ToolchainSet{} + presence := &buildrun.PresenceSet{} + have := &buildrun.HaveSet{} + trunks := &check.Trunks{} + places := &buildrun.Places{Capacity: capacity} + runner := buildrun.Runner{Caller: caller, Steps: stepsWaiter{js}, Bound: stepBound} + ch := &check.Checker{ + Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]}, + Owed: kvStore{buckets[delivery.CheckerOwedBucket]}, + Trunks: trunks, + Bound: checkBound, + Link: link, + Logf: logger.Printf, + } + ch.Judge = newJudge(judgeDeps{ + checker: ch, runner: runner, tools: tools, presence: presence, have: have, places: places, trunks: trunks, + kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]}, + logf: logger.Printf, + }) + + // The toolchains first, so a check that starts from the first candidate finds the nodes. + watch := newToolchainWatch(ctx, js, tools, ch.NodesChanged, logger) + if err := watch.start(); err != nil { + return err + } + // A build module says each time it wrote its toolchains key: its stream, and the checker's durable consumer on it, are + // made by the controller. Where the stream is not there yet the checker says so and is started again with the build + // module. + for _, module := range []string{buildrun.GoBuildModule, buildrun.NodeBuildModule} { + stream := toolchainStreams[module] + cons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module) + if err != nil { + logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the checker starts again: %v", cred.Node, cred.Module, stream, module, err) + continue + } + cc, err := cons.Consume(func(m jetstream.Msg) { watch.Reported(m) }, + jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + if err != nil { + return err + } + defer cc.Stop() + } + errs := make(chan error, 6) + // The watchman's presence says which nodes are asleep. Where its bucket is not there yet no node is online, so nothing + // is checked; the checker says so and is started again with the watchman. + switch kv, err := js.KeyValue(ctx, delivery.PresenceBucket); { + case errors.Is(err, jetstream.ErrBucketNotFound): + logger.Printf("the bucket %s does not exist yet: no node counts as online until the checker starts again", delivery.PresenceBucket) + case err != nil: + return fmt.Errorf("the bucket %s: %w", delivery.PresenceBucket, err) + default: + pw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { presence.Put(e.Key(), e.Value(), !opOf(e)) }) + if err != nil { + return err + } + go func() { + errs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) { + if presence.Put(e.Key(), e.Value(), !opOf(e)) { + ch.NodesChanged(ctx) // a node came online or went: a check that waited for one may go on + } + }) + }() + } + // The node-engine's have says whether a node's build module runs and is healthy: a node whose build module is not is not + // chosen. Where its bucket is not there yet no build module counts as healthy, so nothing is checked. + switch kv, err := js.KeyValue(ctx, delivery.HaveBucket); { + case errors.Is(err, jetstream.ErrBucketNotFound): + logger.Printf("the bucket %s does not exist yet: no build module counts as healthy until the checker starts again", delivery.HaveBucket) + case err != nil: + return fmt.Errorf("the bucket %s: %w", delivery.HaveBucket, err) + default: + hw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { have.Put(e.Key(), e.Value(), !opOf(e)) }) + if err != nil { + return err + } + go func() { + errs <- follow(ctx, hw, func(e jetstream.KeyValueEntry) { + if have.Put(e.Key(), e.Value(), !opOf(e)) { + ch.NodesChanged(ctx) + } + }) + }() + } + // The gate's source says the trunk head of each repository, which a pull request is merged onto. Where it is not there + // yet no trunk is known, so nothing is checked. + switch kv, err := js.KeyValue(ctx, delivery.SourceBucket); { + case errors.Is(err, jetstream.ErrBucketNotFound): + logger.Printf("the bucket %s does not exist yet: no trunk is known until the checker starts again", delivery.SourceBucket) + case err != nil: + return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err) + default: + sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { + if e.Key() == delivery.SourceMainKey { + trunks.Put(e.Value(), !opOf(e)) + } + }) + if err != nil { + return err + } + go func() { + errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { + if e.Key() == delivery.SourceMainKey && trunks.Put(e.Value(), !opOf(e)) { + ch.Reconsider(ctx) // the trunk moved: a verdict for the old trunk is judged again + } + }) + }() + } + each := func(e jetstream.KeyValueEntry) { + if opOf(e) { + ch.Candidate(ctx, e.Key(), e.Value()) + } else { + ch.Forget(e.Key()) + } + } + w, err := settle(ctx, buckets[delivery.CandidatesBucket], each) + if err != nil { + return err + } + // A key left running for heads no candidate holds is closed; the candidates are all known now. + if err := ch.Sweep(ctx); err != nil { + return err + } + logger.Printf("started on %s", cred.Node) + go func() { errs <- follow(ctx, w, each) }() + select { + case <-ctx.Done(): + ch.Wait() + return nil + case err := <-errs: + return err + case err := <-watch.errs: + return err + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings.go new file mode 100644 index 0000000..d32d39f --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings.go @@ -0,0 +1,29 @@ +package main + +import ( + "encoding/json" + "fmt" + "os" + + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +// capacityFrom is the build capacity of each node (ADR 0478 clause 3): the module's setting `capacity`, `=` pairs separated by commas, laid into the settings file the controller composes (issue 1661); empty is one step at a +// time on every node. With no settings file named, fallback is read instead, for a run outside the module's launch. +func capacityFrom(path, fallback string) (map[string]int, error) { + if path == "" { + return buildrun.ParseCapacity(fallback) + } + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("the settings file: %w", err) + } + var s struct { + Capacity string `json:"capacity"` + } + if err := json.Unmarshal(raw, &s); err != nil { + return nil, fmt.Errorf("the settings file %s is not JSON: %w", path, err) + } + return buildrun.ParseCapacity(s.Capacity) +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings_test.go new file mode 100644 index 0000000..8543fe7 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings_test.go @@ -0,0 +1,68 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +func settingsFile(t *testing.T, body string) string { + t.Helper() + p := filepath.Join(t.TempDir(), "settings.json") + if err := os.WriteFile(p, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + return p +} + +// ADR 0484 clause 12: the build capacity is the module's setting `capacity`, laid into the settings file: empty is one step at a time everywhere. +func TestCapacityIsReadFromTheSettingsFile(t *testing.T) { + for body, want := range map[string]map[string]int{ + `{"capacity": "shanks=3,ace=2"}`: {"shanks": 3, "ace": 2}, + `{"capacity": ""}`: {}, + `{}`: {}, + } { + got, err := capacityFrom(settingsFile(t, body), "") + if err != nil || len(got) != len(want) || got["shanks"] != want["shanks"] || got["ace"] != want["ace"] { + t.Errorf("%s: capacity %v, %v, want %v", body, got, err, want) + } + } + p := &buildrun.Places{Capacity: mustCapacity(t, settingsFile(t, `{"capacity": ""}`))} + if p.CapacityOf("shanks") != 1 || p.CapacityOf("ace") != 1 { + t.Errorf("an empty setting gives %d and %d, want 1 everywhere", p.CapacityOf("shanks"), p.CapacityOf("ace")) + } +} + +func mustCapacity(t *testing.T, path string) map[string]int { + t.Helper() + c, err := capacityFrom(path, "") + if err != nil { + t.Fatal(err) + } + return c +} + +func TestACapacitySettingThatIsNotOneIsRefusedNamingIt(t *testing.T) { + if _, err := capacityFrom(settingsFile(t, `{"capacity": "shanks=-1"}`), ""); err == nil { + t.Error("shanks=-1 accepted") + } + if c, err := capacityFrom(settingsFile(t, `{"capacity": "shanks=3,ace=2,g14=0,novox=0"}`), ""); err != nil || c["g14"] != 0 || len(c) != 4 { + t.Errorf("the starting value: %v, %v", c, err) + } + if _, err := capacityFrom(settingsFile(t, `not json`), ""); err == nil { + t.Error("a file that is not JSON accepted") + } + if _, err := capacityFrom(filepath.Join(t.TempDir(), "none"), ""); err == nil { + t.Error("a settings file that is not there accepted") + } +} + +// With no settings file named (a run outside the module's launch, as the tests do), the capacity is the environment's. +func TestWithNoSettingsFileTheCapacityIsTheEnvironments(t *testing.T) { + got, err := capacityFrom("", "ace=2") + if err != nil || got["ace"] != 2 { + t.Errorf("capacity %v, %v", got, err) + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go new file mode 100644 index 0000000..7b650b0 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go @@ -0,0 +1,104 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "strings" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +// toolchainStreams are the stream each build module carries its toolchain.reported event on, as the bus names it +// (declared by the build modules, heard by this module through consumes). +var toolchainStreams = map[string]string{buildrun.GoBuildModule: "TOOLCHAINS_GO", buildrun.NodeBuildModule: "TOOLCHAINS_NODE"} + +// toolchainWatch keeps the build modules' toolchains in a buildrun.ToolchainSet and tells the builder when one changed. A +// build module's bucket is made when it is assigned, after the builder started, and the builder holds no timer (ADR 0437 +// §8): it watches the buckets that exist at start, and a build module says each time it wrote its toolchains key +// (toolchain.reported, handled by Reported); a bucket not watched yet is then watched. +type toolchainWatch struct { + ctx context.Context + js bucketOpener + set *buildrun.ToolchainSet + changed func(ctx context.Context) // a language's entries changed: the failed builds are tried again + logger *log.Logger + errs chan error + mu sync.Mutex + watched map[string]bool +} + +func newToolchainWatch(ctx context.Context, js bucketOpener, set *buildrun.ToolchainSet, changed func(context.Context), logger *log.Logger) *toolchainWatch { + return &toolchainWatch{ctx: ctx, js: js, set: set, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}} +} + +// start watches the buckets that exist now; a missing one is said and waits for its build module's event. +func (w *toolchainWatch) start() error { + for module := range buildrun.LanguageOf { + if err := w.ensure(module); err != nil { + return err + } + } + return nil +} + +func (w *toolchainWatch) ensure(module string) error { + w.mu.Lock() + defer w.mu.Unlock() + if w.watched[module] { + return nil + } + name := delivery.ToolchainsBucket(module) + kv, err := w.js.KeyValue(w.ctx, name) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound): + w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module) + return nil + case err != nil: + return fmt.Errorf("the bucket %s: %w", name, err) + } + watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.set.Put(module, e.Key(), e.Value(), !opOf(e)) }) + if err != nil { + return err + } + w.watched[module] = true + go func() { + w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) { + if w.set.Put(module, e.Key(), e.Value(), !opOf(e)) { + w.changed(w.ctx) + } + }) + }() + w.changed(w.ctx) // what the settle read is new to the builder + return nil +} + +// delivered is the part of a JetStream message the watch uses. +type delivered interface { + Subject() string + Ack() error + Term() error +} + +// Reported takes one toolchain.reported event of a build module in: the module is read off the subject, +// mesh.stream..event.toolchain.reported, and its bucket is watched if it is not. A subject of no build module is +// terminated; a failure stays unacknowledged, and the server delivers it again. +func (w *toolchainWatch) Reported(m delivered) { + module, _, _ := strings.Cut(strings.TrimPrefix(m.Subject(), "mesh.stream."), ".event.") + if _, known := buildrun.LanguageOf[module]; !known || !strings.HasPrefix(m.Subject(), "mesh.stream.") { + w.logger.Printf("%s: not a build module's event, dropped", m.Subject()) + _ = m.Term() + return + } + if err := w.ensure(module); err != nil { + w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err) + return + } + _ = m.Ack() +} diff --git a/modules/mesh-delivery-checker/files/settings.json b/modules/mesh-delivery-checker/files/settings.json new file mode 100644 index 0000000..1354ea4 --- /dev/null +++ b/modules/mesh-delivery-checker/files/settings.json @@ -0,0 +1,3 @@ +{ + "capacity": "" +} diff --git a/modules/mesh-delivery-checker/internal/check/case.go b/modules/mesh-delivery-checker/internal/check/case.go new file mode 100644 index 0000000..24cb19e --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/case.go @@ -0,0 +1,135 @@ +package check + +import ( + "fmt" + "maps" + "sort" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// HQRepository is the repository the dark checker writes no status for (ADR 0484 clause 8): hq builds no module and its +// check is a script no build module can run, so it is judged by its merge-check.sh until issue 1296 settles how. +const HQRepository = "novox/hq" + +// HQNote is what the read tools say of an hq pull request (ADR 0484 clause 8). +const HQNote = "hq: judged by merge-check.sh until issue 1296" + +// Pull is one pull request the checker judges: its head, the trunk head of its repository it is merged onto and the +// modules the indexer says it touches, with the fingerprints the indexer wrote for them. +type Pull struct { + Repository string + Number int + Head string + Trunk string + Modules []string + Fps map[string]delivery.CandidateModule +} + +// Case is one check: a candidate's pull requests that are judged (those of hq and of a repository whose trunk is not known +// are left out), the trunk head of each repository and the key of the verdict. +type Case struct { + // Key is the verdict key, `.`. + Key string + Group string + HeadsHash string + Pulls []Pull + // Trunk is the trunk head of each repository of Pulls. + Trunk map[string]string + // Left says why pull requests of the candidate are not judged (hq's, a repository with no known trunk); empty when all are. + Left []string + // Queued is when the checker took the case up, for the time of its queue. + Queued time.Time + // Progress says what the check is doing, so a check that ran past its bound names the step it was stuck at. + Progress *Progress +} + +// Progress is the step a check is at, written by the check and read when its bound passes. +type Progress struct { + mu sync.Mutex + stage, node string +} + +// Set records the step now running and the node it runs on, empty for none. +func (p *Progress) Set(stage, node string) { + p.mu.Lock() + p.stage, p.node = stage, node + p.mu.Unlock() +} + +// Now is the step recorded last and its node. +func (p *Progress) Now() (stage, node string) { + p.mu.Lock() + defer p.mu.Unlock() + return p.stage, p.node +} + +// CaseOf is the check of a candidate against the trunk heads (the gate's source.main): the pull requests that are judged. +// The checker judges pull requests into main only (ADR 0484 clause 11): the pull message carries no base, a merge into +// another branch is a closed message, so every candidate is into the trunk; a repository whose trunk head is not known has +// none to merge onto. It answers nil when no pull request is left to judge. +func CaseOf(c delivery.Candidate, trunk map[string]string) *Case { + key, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash) + if err != nil { + return nil + } + cs := &Case{Key: key, Group: c.Group, HeadsHash: c.HeadsHash, Trunk: map[string]string{}, Progress: &Progress{}} + for _, p := range c.Pulls { + switch head, known := trunk[p.Repository]; { + case p.Repository == HQRepository: + cs.Left = append(cs.Left, fmt.Sprintf("%s#%d: %s", p.Repository, p.Number, HQNote)) + case !known || head == "": + cs.Left = append(cs.Left, fmt.Sprintf("%s#%d: the trunk head of %s is not known", p.Repository, p.Number, p.Repository)) + default: + cs.Pulls = append(cs.Pulls, Pull{Repository: p.Repository, Number: p.Number, Head: p.Head.Head, Trunk: head, Modules: p.Modules, Fps: p.Fps}) + cs.Trunk[p.Repository] = head + } + } + if len(cs.Pulls) == 0 { + return nil + } + sort.SliceStable(cs.Pulls, func(i, j int) bool { + if cs.Pulls[i].Repository != cs.Pulls[j].Repository { + return cs.Pulls[i].Repository < cs.Pulls[j].Repository + } + return cs.Pulls[i].Number < cs.Pulls[j].Number + }) + return cs +} + +// sameTrunk says whether two verdicts' trunk heads are the same. +func sameTrunk(a, b map[string]string) bool { return maps.Equal(a, b) } + +// Trunks holds the trunk head of each repository, as the gate's source.main says (a watch puts it in). +type Trunks struct { + mu sync.Mutex + heads map[string]string +} + +// Put takes a source.main value in and answers whether the heads changed. +func (t *Trunks) Put(value []byte, deleted bool) bool { + var heads map[string]string + if !deleted { + s, err := delivery.DecodeSource(value) + if err != nil { + return false + } + heads = s.Heads + } + t.mu.Lock() + defer t.mu.Unlock() + if maps.Equal(t.heads, heads) { + return false + } + t.heads = maps.Clone(heads) + return true +} + +// Heads are the trunk heads now (a copy). +func (t *Trunks) Heads() map[string]string { + t.mu.Lock() + defer t.mu.Unlock() + return maps.Clone(t.heads) +} diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go new file mode 100644 index 0000000..5706ae4 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -0,0 +1,451 @@ +package check + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// DefaultBound is how long a check may run before it is ended failed naming the step it was stuck at (ADR 0484, issue 1678). +const DefaultBound = 30 * time.Minute + +// NodesWords is in the Words of a failure that is the build nodes' (none qualifies, the toolchains disagree): such a verdict +// is judged again when the nodes change, by a watch and never a timer (ADR 0437 §8). +const NodesWords = "build node qualifies" + +// Outcome is what judging a case gives: the verdict (state, failure, levels, times, composition; the checker adds the +// group, the heads and the trunk) and, for the line, what the check knows of the builds of the modules it touches. +type Outcome struct { + Verdict delivery.Verdict + Builds []BuildNote +} + +// BuildNote is the state of a candidate's module build, by module: built, failed or waiting (ADR 0484 clause 7). +type BuildNote struct { + Module string + State string +} + +// Judge runs the check of a case: the merge, the levels. It writes what it is doing to c.Progress. It answers an error +// only when it could not judge at all; a failure of the change is a Verdict. +type Judge interface { + Judge(ctx context.Context, c *Case) (Outcome, error) +} + +// Checker judges the candidates the indexer writes, dark (ADR 0453 clause 4, ADR 0484): for each candidate it claims a +// verdict key, runs the check through its Judge, writes the verdict and, for each pull request judged, the owed status +// `mesh/check` the forge bridge performs. It holds no timer; a check starts from a candidate or from a change of the +// nodes, the trunk or the controller's plan, and ends when its judge does or its bound passes. +type Checker struct { + Verdicts KV + Owed KV + Judge Judge + Trunks interface{ Heads() map[string]string } + // Bound is how long one check may run; DefaultBound when 0. + Bound time.Duration + // Limit is how many checks run at once; two when 0. + Limit int + // Hold says whether a delivery builds, so no new check starts (ADR 0484 clause 12); nil never holds. + Hold func() bool + // Link is the base of the address a status points at; the verdict key follows. + Link string + Now func() time.Time + Logf func(format string, args ...any) + + once sync.Once + sem chan struct{} + wg sync.WaitGroup + mu sync.Mutex + known map[string]delivery.Candidate // by candidates key + inflight map[string]bool // by verdict key + again map[string]bool // by verdict key: asked again while it ran + tried map[string]bool // by verdict key: looked at since this process started + epoch uint64 // changes of the nodes seen since this process started + nodesFail map[string]uint64 // by verdict key: the epoch a failure of the nodes was written at +} + +func (ch *Checker) init() { + ch.once.Do(func() { + limit := ch.Limit + if limit <= 0 { + limit = 2 + } + ch.sem = make(chan struct{}, limit) + ch.known, ch.inflight, ch.again, ch.tried = map[string]delivery.Candidate{}, map[string]bool{}, map[string]bool{}, map[string]bool{} + ch.nodesFail = map[string]uint64{} + }) +} + +func (ch *Checker) logf(format string, args ...any) { + if ch.Logf != nil { + ch.Logf(format, args...) + } +} + +func (ch *Checker) now() time.Time { + if ch.Now != nil { + return ch.Now() + } + return time.Now() +} + +func (ch *Checker) bound() time.Duration { + if ch.Bound > 0 { + return ch.Bound + } + return DefaultBound +} + +// Wait returns when every check started so far has ended. +func (ch *Checker) Wait() { ch.wg.Wait() } + +// Candidate takes one value of the indexer's candidates bucket in (the key is the group) and checks it if it is to be +// checked. A value that is not a candidate is said and left. +func (ch *Checker) Candidate(ctx context.Context, key string, value []byte) { + ch.init() + c, err := delivery.DecodeCandidate(value) + if err != nil { + ch.logf("candidates.%s holds what is not a candidate, left: %v", key, err) + return + } + ch.mu.Lock() + ch.known[key] = c + ch.mu.Unlock() + ch.consider(ctx, key, c) +} + +// Forget takes a candidate out (its key was deleted). +func (ch *Checker) Forget(key string) { + ch.init() + ch.mu.Lock() + delete(ch.known, key) + ch.mu.Unlock() +} + +// NodesChanged is Reconsider after a change of the nodes: one came online or went, a build module became healthy or +// stopped being, or a toolchain moved. A check that failed because no node qualified is run again only after such a change. +func (ch *Checker) NodesChanged(ctx context.Context) { + ch.init() + ch.mu.Lock() + ch.epoch++ + ch.mu.Unlock() + ch.Reconsider(ctx) +} + +// Reconsider looks at every candidate again: the trunk moved, the nodes changed or a delivery ended. A candidate whose +// verdict stands for the trunk it was judged against is left. +func (ch *Checker) Reconsider(ctx context.Context) { + ch.init() + ch.mu.Lock() + known := make(map[string]delivery.Candidate, len(ch.known)) + for k, c := range ch.known { + known[k] = c + } + ch.mu.Unlock() + for k, c := range known { + ch.consider(ctx, k, c) + } +} + +func (ch *Checker) heads() map[string]string { + if ch.Trunks == nil { + return nil + } + return ch.Trunks.Heads() +} + +func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidate) { + cs := CaseOf(c, ch.heads()) + if cs == nil { + return // nothing to judge: hq's pull requests, or no trunk known + } + if ch.Hold != nil && ch.Hold() { + ch.logf("%s: not started, a delivery builds", cs.Key) + return // started when the plan bucket empties (Reconsider) + } + cs.Queued = ch.now() + ch.mu.Lock() + if ch.inflight[cs.Key] { + ch.again[cs.Key] = true // looked at once more when the check ends, so a change of the trunk is never lost + ch.mu.Unlock() + return + } + ch.inflight[cs.Key] = true + ch.mu.Unlock() + ch.wg.Add(1) + go func() { + defer ch.wg.Done() + ch.sem <- struct{}{} + defer func() { <-ch.sem }() + for { + ch.try(ctx, key, cs) + ch.mu.Lock() + more := ch.again[cs.Key] + delete(ch.again, cs.Key) + if !more || ctx.Err() != nil { + delete(ch.inflight, cs.Key) + ch.mu.Unlock() + return + } + ch.mu.Unlock() + if next := CaseOf(ch.candidate(key, c), ch.heads()); next != nil { + cs = next + } + } + }() +} + +func (ch *Checker) candidate(key string, fallback delivery.Candidate) delivery.Candidate { + ch.mu.Lock() + defer ch.mu.Unlock() + if c, ok := ch.known[key]; ok { + return c + } + return fallback +} + +// claimed is a verdict key the checker holds: the revision the next write is made at. +type claimed struct{ rev uint64 } + +// claim takes the check of the case, or answers nil when it is not this checker's to run now: +// - no key: it is created running; a create that finds the key made meanwhile gets nothing; +// - a key left running is taken over as it stands when this process looks at it for the first time (it was left by the +// checker that stopped; there is one instance, as there is one builder per node, ADR 0478 clause 6), and left otherwise; +// - a pass or a failure for the trunk heads the case has stands; one for other trunk heads (the trunk moved) is taken +// again by compare-and-set on the same key, and so is a failure of the nodes after the nodes changed. +func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64) (*claimed, error) { + running := delivery.Verdict{Group: cs.Group, Heads: cs.HeadsHash, Trunk: cs.Trunk, State: delivery.VerdictRunning} + data, err := delivery.EncodeVerdict(running) + if err != nil { + return nil, err + } + value, rev, found, err := ch.Verdicts.Get(ctx, cs.Key) + if err != nil { + return nil, fmt.Errorf("%s: %w", cs.Key, err) + } + if !found { + rev, err := ch.Verdicts.Create(ctx, cs.Key, data) + if errors.Is(err, ErrConflict) { + return nil, nil // another checker made it first + } + if err != nil { + return nil, fmt.Errorf("%s: %w", cs.Key, err) + } + return &claimed{rev}, nil + } + cur, err := delivery.DecodeVerdict(value) + if err != nil { + return nil, fmt.Errorf("%s holds what is not a verdict: %w", cs.Key, err) + } + retake := false + switch { + case cur.State == delivery.VerdictRunning: + if first { + return &claimed{rev}, nil + } + return nil, nil + case !sameTrunk(cur.Trunk, cs.Trunk): + retake = true + case cur.State == delivery.VerdictFail && IsNodesFailure(cur): + ch.mu.Lock() + at, ok := ch.nodesFail[cs.Key] + ch.mu.Unlock() + retake = !ok || epoch > at + } + if !retake { + return nil, nil + } + rev, err = ch.Verdicts.Update(ctx, cs.Key, data, rev) + if errors.Is(err, ErrConflict) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("%s: %w", cs.Key, err) + } + return &claimed{rev}, nil +} + +// IsNodesFailure says whether a verdict failed because no build node could take the check. +func IsNodesFailure(v delivery.Verdict) bool { + return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, NodesWords) +} + +// try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends +// because ctx did is left running: the checker takes it again when it starts. +func (ch *Checker) try(ctx context.Context, key string, cs *Case) { + ch.mu.Lock() + first := !ch.tried[cs.Key] + ch.tried[cs.Key] = true + epoch := ch.epoch + ch.mu.Unlock() + cl, err := ch.claim(ctx, cs, first, epoch) + if err != nil { + ch.logf("%s: %v", cs.Key, err) + return + } + if cl == nil { + return + } + began := ch.now() + ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, nil) + out := ch.judge(ctx, cs) + if ctx.Err() != nil { + return + } + v := out.Verdict + v.Group, v.Heads, v.Trunk = cs.Group, cs.HeadsHash, cs.Trunk + if v.Took.Queue == 0 && !cs.Queued.IsZero() && began.After(cs.Queued) { + v.Took.Queue = began.Sub(cs.Queued).Milliseconds() + } + data, err := delivery.EncodeVerdict(v) + if err != nil { + ch.logf("%s: the verdict cannot be written: %v", cs.Key, err) + v = delivery.Verdict{Group: cs.Group, Heads: cs.HeadsHash, Trunk: cs.Trunk, State: delivery.VerdictFail, + Failed: delivery.FailedFirst{Machine: "checker", Words: "the verdict could not be written: " + err.Error()}} + if data, err = delivery.EncodeVerdict(v); err != nil { + return + } + } + if _, err := ch.Verdicts.Update(ctx, cs.Key, data, cl.rev); err != nil { + ch.logf("%s: the verdict could not be written: %v", cs.Key, err) + return + } + if IsNodesFailure(v) { + ch.mu.Lock() + ch.nodesFail[cs.Key] = epoch + ch.mu.Unlock() + } + ch.writeOwed(ctx, cs, v, out.Builds) + ch.logf("%s: %s", cs.Key, v.State) +} + +// judge runs the case within the bound. A judge that fails to judge, and a check that ran past its bound, are verdicts of +// failure naming the machine and the step. +func (ch *Checker) judge(ctx context.Context, cs *Case) Outcome { + bound := ch.bound() + run, cancel := context.WithTimeoutCause(ctx, bound, errBound) + defer cancel() + out, err := ch.Judge.Judge(run, cs) + stage, node := cs.Progress.Now() + if node == "" { + node = "checker" + } + switch { + case ctx.Err() != nil: + return Outcome{} + case context.Cause(run) == errBound: + if stage == "" { + stage = "starting" + } + return Outcome{Verdict: failure(node, fmt.Sprintf("the check ran past its bound of %s, stuck at: %s", bound, stage))} + case err != nil: + return Outcome{Verdict: failure(node, err.Error())} + } + return out +} + +var errBound = errors.New("the check ran past its bound") + +func failure(machine, words string) delivery.Verdict { + return delivery.Verdict{State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: machine, Words: words}} +} + +// writeOwed writes, for each pull request judged, the status the forge bridge performs: pending while the check runs, success +// or failure at its end (ADR 0484 clause 2). A key already holding the same value is not written again, so the bridge +// performs one act for each change; an older head's key is never touched. +func (ch *Checker) writeOwed(ctx context.Context, cs *Case, v delivery.Verdict, builds []BuildNote) { + for _, p := range cs.Pulls { + key, err := delivery.CheckerOwedKey(p.Repository, p.Number, p.Head) + if err != nil { + ch.logf("%s: %v", cs.Key, err) + continue + } + state := map[string]string{delivery.VerdictRunning: "pending", delivery.VerdictPass: "success", delivery.VerdictFail: "failure"}[v.State] + o := delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, + Target: strings.TrimSuffix(ch.Link, "/") + "/" + cs.Key, Text: Line(v, cs, p, builds)} + data, err := delivery.EncodeCheckerOwed(o) + if err != nil { + ch.logf("%s: the status of %s cannot be written: %v", cs.Key, key, err) + continue + } + if err := putOnce(ctx, ch.Owed, key, data, func(cur []byte) bool { + held, err := delivery.DecodeCheckerOwed(cur) + o.V = delivery.Version + return err == nil && delivery.SameOwed(held, o) + }); err != nil { + ch.logf("%s: the status of %s cannot be written: %v", cs.Key, key, err) + } + } +} + +// putOnce writes data under key by create, or by compare-and-set when the key exists, and writes nothing when same says the +// key holds the value already. +func putOnce(ctx context.Context, kv KV, key string, data []byte, same func(cur []byte) bool) error { + for try := 0; try < 4; try++ { + cur, rev, found, err := kv.Get(ctx, key) + if err != nil { + return err + } + if found && same(cur) { + return nil + } + if !found { + _, err = kv.Create(ctx, key, data) + } else { + _, err = kv.Update(ctx, key, data, rev) + } + if errors.Is(err, ErrConflict) { + continue + } + return err + } + return fmt.Errorf("%s changed under every one of 4 tries", key) +} + +// Sweep closes, at start, every verdict left running for a candidate that no longer holds those heads: nothing will check +// them again, and the key would stay running for ever (the builder's Sweep, issue 1660). Call it once the candidates have been +// read. A key of heads a candidate holds now is not touched: the check takes it over. +func (ch *Checker) Sweep(ctx context.Context) error { + ch.init() + keys, err := ch.Verdicts.Keys(ctx) + if err != nil { + return err + } + ch.mu.Lock() + current := map[string]bool{} + for _, c := range ch.known { + if k, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash); err == nil { + current[k] = true + } + } + ch.mu.Unlock() + for _, key := range keys { + if current[key] { + continue + } + value, rev, found, err := ch.Verdicts.Get(ctx, key) + if err != nil || !found { + continue + } + cur, err := delivery.DecodeVerdict(value) + if err != nil || cur.State != delivery.VerdictRunning { + continue + } + cur.State = delivery.VerdictFail + cur.Failed = delivery.FailedFirst{Machine: "checker", Words: "closed at start: left running, and no candidate holds these heads any more, so nothing will check them"} + data, err := delivery.EncodeVerdict(cur) + if err != nil { + return err + } + if _, err := ch.Verdicts.Update(ctx, key, data, rev); err != nil && !errors.Is(err, ErrConflict) { + return fmt.Errorf("%s: %w", key, err) + } + ch.logf("%s: closed, left running for heads no candidate holds", key) + } + return nil +} diff --git a/modules/mesh-delivery-checker/internal/check/checker_test.go b/modules/mesh-delivery-checker/internal/check/checker_test.go new file mode 100644 index 0000000..4a791f4 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/checker_test.go @@ -0,0 +1,367 @@ +package check + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +const ( + repoA = "novox/mesh-delivery" + repoSDK = "novox/mesh-sdk" + head1 = "1111111111111111111111111111111111111111" + head2 = "2222222222222222222222222222222222222222" + trunkA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + trunkA2 = "abababababababababababababababababababab" + trunkS = "5555555555555555555555555555555555555555" +) + +// candidateOf is a candidate of a group with the given pull requests, as the indexer writes one. +func candidateOf(t testing.TB, group string, pulls ...delivery.CandidatePull) []byte { + t.Helper() + heads := make([]delivery.Head, len(pulls)) + for i, p := range pulls { + heads[i] = p.Head + } + hash, err := delivery.HeadsHash(heads) + if err != nil { + t.Fatal(err) + } + c := delivery.Candidate{V: delivery.Version, Group: group, HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)} + data, err := delivery.EncodeCandidate(c) + if err != nil { + t.Fatal(err) + } + return data +} + +func pullOf(repo string, number int, head string) delivery.CandidatePull { + return delivery.CandidatePull{Head: delivery.Head{Repository: repo, Number: number, Head: head}, Modules: []string{}} +} + +// judgeFunc is a Judge written as a function; it counts its calls. +type judgeFunc struct { + mu sync.Mutex + calls int + f func(ctx context.Context, c *Case, call int) (Outcome, error) +} + +func (j *judgeFunc) Judge(ctx context.Context, c *Case) (Outcome, error) { + j.mu.Lock() + j.calls++ + n := j.calls + j.mu.Unlock() + return j.f(ctx, c, n) +} + +func (j *judgeFunc) count() int { j.mu.Lock(); defer j.mu.Unlock(); return j.calls } + +func passing() delivery.Verdict { + return delivery.Verdict{State: delivery.VerdictPass, Levels: map[string]delivery.LevelResult{ + delivery.LevelUnit: {Run: 2, Kept: 3}, delivery.LevelIntegration: {Run: 1}}} +} + +func passJudge() *judgeFunc { + return &judgeFunc{f: func(context.Context, *Case, int) (Outcome, error) { return Outcome{Verdict: passing()}, nil }} +} + +type rig struct { + ch *Checker + verdicts *memKV + owed *memKV + trunks *Trunks +} + +func newRig(t testing.TB, j Judge) *rig { + t.Helper() + r := &rig{verdicts: newMemKV(), owed: newMemKV(), trunks: &Trunks{}} + r.setTrunks(t, map[string]string{repoA: trunkA, repoSDK: trunkS}) + r.ch = &Checker{Verdicts: r.verdicts, Owed: r.owed, Judge: j, Trunks: r.trunks, Link: "https://view.invalid/checks", Now: time.Now} + return r +} + +func (r *rig) setTrunks(t testing.TB, heads map[string]string) bool { + t.Helper() + data, err := delivery.EncodeSource(delivery.Source{Group: "g", Heads: heads}) + if err != nil { + t.Fatal(err) + } + return r.trunks.Put(data, false) +} + +func (r *rig) verdict(t testing.TB, key string) delivery.Verdict { + t.Helper() + raw, _, found, _ := r.verdicts.Get(context.Background(), key) + if !found { + t.Fatalf("no verdict %s; the bucket holds %v", key, r.verdicts.vals) + } + v, err := delivery.DecodeVerdict(raw) + if err != nil { + t.Fatal(err) + } + return v +} + +func keyOf(t testing.TB, group string, value []byte) string { + t.Helper() + c, err := delivery.DecodeCandidate(value) + if err != nil { + t.Fatal(err) + } + k, err := delivery.CheckerVerdictKey(group, c.HeadsHash) + if err != nil { + t.Fatal(err) + } + return k +} + +func TestACandidateNotYetCheckedGetsItsVerdictKeyCreatedOnceAndIsJudged(t *testing.T) { // N6/AC1, 1678 + ctx, j := context.Background(), passJudge() + r := newRig(t, j) + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + key := keyOf(t, "feat/x", cand) + if v := r.verdict(t, key); v.State != delivery.VerdictPass || v.Group != "feat/x" || v.Trunk[repoA] != trunkA { + t.Fatalf("verdict = %+v", v) + } + if r.verdicts.creates[key] != 1 || j.count() != 1 { + t.Errorf("creates = %d, judged = %d, want one each", r.verdicts.creates[key], j.count()) + } + // The same candidate heard again, with nothing changed, is not judged again. + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + if j.count() != 1 { + t.Errorf("judged %d times after the same candidate again, want once", j.count()) + } +} + +// racyKV lets another instance create the key just before this one's create. +type racyKV struct { + *memKV + other []byte +} + +func (r racyKV) Create(ctx context.Context, key string, value []byte) (uint64, error) { + if r.other != nil { + _, _ = r.memKV.Create(ctx, key, r.other) + } + return r.memKV.Create(ctx, key, value) +} + +func TestASecondInstancesCreateIsRefusedAndItDoesNothing(t *testing.T) { // N6/AC1 + ctx, j := context.Background(), passJudge() + r := newRig(t, j) + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + key := keyOf(t, "feat/x", cand) + other, _ := delivery.EncodeVerdict(delivery.Verdict{Group: "feat/x", Heads: strings.SplitN(key, ".", 2)[1], State: delivery.VerdictRunning}) + r.ch.Verdicts = racyKV{r.verdicts, other} + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + if j.count() != 0 { + t.Errorf("the instance that lost the create judged %d times, want none", j.count()) + } + if v := r.verdict(t, key); v.State != delivery.VerdictRunning { + t.Errorf("the other instance's key was changed: %+v", v) + } +} + +func TestARestartWithARunningKeyJudgesItAgain(t *testing.T) { // N6/AC2 + ctx, j := context.Background(), passJudge() + r := newRig(t, j) + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + key := keyOf(t, "feat/x", cand) + left, _ := delivery.EncodeVerdict(delivery.Verdict{Group: "feat/x", Heads: strings.SplitN(key, ".", 2)[1], Trunk: map[string]string{repoA: trunkA}, State: delivery.VerdictRunning}) + if _, err := r.verdicts.Create(ctx, key, left); err != nil { + t.Fatal(err) + } + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + if j.count() != 1 || r.verdict(t, key).State != delivery.VerdictPass { + t.Fatalf("judged %d times, verdict %+v, want the left check judged once and passed", j.count(), r.verdict(t, key)) + } +} + +func TestACheckPastItsBoundFailsNamingTheStuckStep(t *testing.T) { // N6/AC3 + ctx := context.Background() + j := &judgeFunc{f: func(ctx context.Context, c *Case, _ int) (Outcome, error) { + c.Progress.Set("test the unit level of mesh-delivery", "shanks") + <-ctx.Done() + return Outcome{}, ctx.Err() + }} + r := newRig(t, j) + r.ch.Bound = 30 * time.Millisecond + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + v := r.verdict(t, keyOf(t, "feat/x", cand)) + if v.State != delivery.VerdictFail || v.Failed.Machine != "shanks" || !strings.Contains(v.Failed.Words, "stuck at: test the unit level of mesh-delivery") { + t.Fatalf("verdict = %+v, want a failure naming shanks and the stuck step", v) + } +} + +func TestNoQualifyingNodeFailsWithTheReasonAndIsJudgedAgainWhenTheNodesChange(t *testing.T) { // N6/AC4 + ctx := context.Background() + j := &judgeFunc{f: func(_ context.Context, _ *Case, n int) (Outcome, error) { + if n == 1 { + return Outcome{Verdict: failure("shanks", "no go "+NodesWords+": none is online")}, nil + } + return Outcome{Verdict: passing()}, nil + }} + r := newRig(t, j) + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + key := keyOf(t, "feat/x", cand) + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + if v := r.verdict(t, key); !IsNodesFailure(v) { + t.Fatalf("verdict = %+v, want a failure of the nodes", v) + } + r.ch.Reconsider(ctx) // nothing changed: not judged again + r.ch.Wait() + if j.count() != 1 { + t.Fatalf("judged %d times with the nodes unchanged, want once", j.count()) + } + r.ch.NodesChanged(ctx) + r.ch.Wait() + if j.count() != 2 || r.verdict(t, key).State != delivery.VerdictPass { + t.Fatalf("after the nodes changed: judged %d times, verdict %+v", j.count(), r.verdict(t, key)) + } +} + +func TestHqAndARepositoryWithNoTrunkGetNoVerdict(t *testing.T) { // N6/AC6, ADR 0484 clauses 8 and 11 + ctx, j := context.Background(), passJudge() + r := newRig(t, j) + r.ch.Candidate(ctx, "feat/hq", candidateOf(t, "feat/hq", pullOf(HQRepository, 3, head1))) + r.ch.Candidate(ctx, "feat/y", candidateOf(t, "feat/y", pullOf("novox/unknown", 4, head1))) + r.ch.Wait() + if j.count() != 0 || len(r.verdicts.vals) != 0 || len(r.owed.vals) != 0 { + t.Errorf("judged %d times, %d verdicts, %d statuses, want none", j.count(), len(r.verdicts.vals), len(r.owed.vals)) + } +} + +func TestAGroupWithHqJudgesItsOtherMembersAndWritesNoStatusForHq(t *testing.T) { // ADR 0484 clause 8 + ctx, j := context.Background(), passJudge() + r := newRig(t, j) + r.ch.Candidate(ctx, "feat/x", candidateOf(t, "feat/x", pullOf(HQRepository, 3, head1), pullOf(repoA, 7, head2))) + r.ch.Wait() + if j.count() != 1 { + t.Fatalf("judged %d times, want once", j.count()) + } + keys, _ := r.owed.Keys(ctx) + if len(keys) != 1 || !strings.HasPrefix(keys[0], "novox/mesh-delivery.7.") && !strings.Contains(keys[0], "mesh-delivery") { + t.Errorf("statuses %v, want one for mesh-delivery#7 only", keys) + } +} + +func TestATrunkMoveRewritesTheSameVerdictKeyAndTheHeadsStatus(t *testing.T) { // 1300/AC7 + ctx, j := context.Background(), passJudge() + r := newRig(t, j) + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + key := keyOf(t, "feat/x", cand) + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + _, rev1, _, _ := r.verdicts.Get(ctx, key) + owedKey, _ := delivery.CheckerOwedKey(repoA, 7, head1) + _, orev1, _, _ := r.owed.Get(ctx, owedKey) + if !r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS}) { + t.Fatal("the trunks did not change") + } + r.ch.Reconsider(ctx) + r.ch.Wait() + _, rev2, _, _ := r.verdicts.Get(ctx, key) + if j.count() != 2 || rev2 <= rev1 || r.verdict(t, key).Trunk[repoA] != trunkA2 { + t.Fatalf("judged %d times, revisions %d then %d, trunk %v", j.count(), rev1, rev2, r.verdict(t, key).Trunk) + } + if r.verdicts.creates[key] != 1 { + t.Errorf("the key was created %d times, want the same key rewritten", r.verdicts.creates[key]) + } + if _, orev2, _, _ := r.owed.Get(ctx, owedKey); orev2 <= orev1 { + // the line is the same when nothing differs; the status is rewritten when the words change, here the trunk is in them + t.Logf("the status keeps revision %d (its words did not change)", orev2) + } + // A new head is another key, and the older head's status is not touched. + older, _, _, _ := r.owed.Get(ctx, owedKey) + cand2 := candidateOf(t, "feat/x", pullOf(repoA, 7, head2)) + r.ch.Candidate(ctx, "feat%2Fx", cand2) + r.ch.Wait() + if again, _, _, _ := r.owed.Get(ctx, owedKey); string(again) != string(older) { + t.Errorf("the older head's status was rewritten") + } + if n := len(r.owed.vals); n != 2 { + t.Errorf("%d statuses, want one for each head", n) + } +} + +func TestNoCheckStartsWhileADeliveryBuildsAndOneStartsWhenItEnds(t *testing.T) { // N8/AC1 + ctx, j := context.Background(), passJudge() + r := newRig(t, j) + held := true + r.ch.Hold = func() bool { return held } + r.ch.Candidate(ctx, "feat/x", candidateOf(t, "feat/x", pullOf(repoA, 7, head1))) + r.ch.Wait() + if j.count() != 0 { + t.Fatalf("judged %d times while held", j.count()) + } + held = false + r.ch.Reconsider(ctx) + r.ch.Wait() + if j.count() != 1 { + t.Fatalf("judged %d times after the hold ended, want once", j.count()) + } +} + +func TestTheStatusIsPendingWhileTheCheckRunsAndEndsSuccessOrFailure(t *testing.T) { // 1300/AC5 (unit part) + ctx := context.Background() + var seen []string + var r *rig + j := &judgeFunc{f: func(ctx context.Context, c *Case, _ int) (Outcome, error) { + key, _ := delivery.CheckerOwedKey(repoA, 7, head1) + raw, _, _, _ := r.owed.Get(ctx, key) + o, _ := delivery.DecodeCheckerOwed(raw) + seen = append(seen, o.State) + return Outcome{Verdict: delivery.Verdict{State: delivery.VerdictFail, Failed: delivery.FailedFirst{Test: "m/a.TestX"}}}, nil + }} + r = newRig(t, j) + r.ch.Candidate(ctx, "feat/x", candidateOf(t, "feat/x", pullOf(repoA, 7, head1))) + r.ch.Wait() + key, _ := delivery.CheckerOwedKey(repoA, 7, head1) + raw, _, _, _ := r.owed.Get(ctx, key) + o, err := delivery.DecodeCheckerOwed(raw) + if err != nil { + t.Fatal(err) + } + if len(seen) != 1 || seen[0] != "pending" || o.State != "failure" || o.Context != delivery.StatusContextCheck || !strings.HasPrefix(o.Target, "https://view.invalid/checks/") { + t.Errorf("while running %v, at the end %+v", seen, o) + } +} + +func TestSweepClosesARunningKeyNoCandidateHoldsAndLeavesOneItDoes(t *testing.T) { + ctx, j := context.Background(), passJudge() + r := newRig(t, j) + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + live := keyOf(t, "feat/x", cand) + dead, _ := delivery.CheckerVerdictKey("feat/gone", "0123456789ab") + for _, k := range []string{live, dead} { + group, heads, _ := delivery.ParseCheckerVerdictKey(k) + raw, _ := delivery.EncodeVerdict(delivery.Verdict{Group: group, Heads: heads, State: delivery.VerdictRunning}) + _, _ = r.verdicts.Create(ctx, k, raw) + } + r.ch.init() + var c delivery.Candidate + _ = json.Unmarshal(cand, &c) + r.ch.known["feat%2Fx"] = c + if err := r.ch.Sweep(ctx); err != nil { + t.Fatal(err) + } + if v := r.verdict(t, dead); v.State != delivery.VerdictFail || !strings.Contains(v.Failed.Words, "closed at start") { + t.Errorf("the dead key = %+v", v) + } + if v := r.verdict(t, live); v.State != delivery.VerdictRunning { + t.Errorf("the live key = %+v, want it left for the check to take over", v) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/line.go b/modules/mesh-delivery-checker/internal/check/line.go new file mode 100644 index 0000000..d2aaced --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/line.go @@ -0,0 +1,19 @@ +package check + +import ( + "fmt" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Line is the words of a status for one pull request of a case (ADR 0447 clause 3, ADR 0484 clause 7). +func Line(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote) string { + return fmt.Sprintf("%s: %s", v.State, short12(p.Head)) +} + +func short12(s string) string { + if len(s) > 12 { + return s[:12] + } + return s +} diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go new file mode 100644 index 0000000..892ecbe --- /dev/null +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -0,0 +1,197 @@ +package manifest + +import ( + "bytes" + "encoding/json" + "io/fs" + "os" + "path/filepath" + "reflect" + "slices" + "strings" + "testing" +) + +// The controller's manifest reader is mesh-controller internal/catalogue (Manifest, manifest.go: Consumes :521, +// Invokes :590, Streams :613, Reads :659; StreamDeclaration, streams.go), an internal package of another repository. +// These tests decode the manifest strictly into the fields those types name (DisallowUnknownFields) and hold the +// rules of issues 1678 and 1683 (ADR 0453 §4, ADR 0484). The merge gate's module check proves the controller takes the manifest. + +type state struct { + Name string `json:"name"` + History int `json:"history,omitempty"` +} + +type artifact struct { + Name string `json:"name"` + Kind string `json:"kind"` + Language string `json:"language"` + System string `json:"system"` + From string `json:"from"` + Binary string `json:"binary"` +} + +type manifest struct { + Module string `json:"module"` + Version string `json:"version"` + Slug string `json:"slug"` + Consumes []string `json:"consumes"` + State []state `json:"state"` + Reads []string `json:"reads"` + Invokes []string `json:"invokes"` + Config map[string]struct { + Type string `json:"type"` + Description string `json:"description"` + Pattern string `json:"pattern"` + Default string `json:"default"` + } `json:"config"` + OwnSecrets map[string]string `json:"own-secrets"` + SecretsOwner string `json:"secrets-owner"` + Resources []map[string]any `json:"resources"` + Build struct { + Artifacts []artifact `json:"artifacts"` + } `json:"build"` +} + +func load(t *testing.T) (manifest, map[string]json.RawMessage) { + t.Helper() + raw, err := os.ReadFile("module.json") + if err != nil { + t.Fatal(err) + } + var m manifest + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.DisallowUnknownFields() + if err := dec.Decode(&m); err != nil { + t.Fatalf("the controller's reader refuses a key this manifest has: %v", err) + } + var top map[string]json.RawMessage + if err := json.Unmarshal(raw, &top); err != nil { + t.Fatal(err) + } + return m, top +} + +func sorted(s []string) []string { + s = slices.Clone(s) + slices.Sort(s) + return s +} + +func TestManifestDeclarations(t *testing.T) { // 1678/AC1 + m, top := load(t) + if m.Module != "mesh-delivery-checker" { + t.Errorf("module = %q", m.Module) + } + // delivery.CheckerKeptHistory, CheckerOwedHistory, PreviewsHistory, CheckerVerdictsHistory; the names follow the SDK's buckets. + if want := []state{{Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) { + t.Errorf("state = %+v, want %+v", m.State, want) + } + if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", + "mesh-controller.plans-building", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", + "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { + t.Errorf("reads = %v, want %v", m.Reads, want) + } + if want := []string{"mesh-build-go.build_go_digest", "mesh-build-go.build_go_fetch", "mesh-build-go.build_go_test", "mesh-build-node.build_node_fetch", "mesh-build-node.build_node_test"}; !reflect.DeepEqual(sorted(m.Invokes), want) { + t.Errorf("invokes = %v, want only the build modules' fetch, digest and test (no run step before issue 1309, ADR 0484 clause 5)", m.Invokes) + } + // The checker is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it starts. + if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported"}; !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" { + t.Errorf("own-secrets = %v, want exactly broker", m.OwnSecrets) + } + if m.SecretsOwner != "mesh-delivery-checker" { + t.Errorf("secrets-owner = %q", m.SecretsOwner) + } + for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats", "bus"} { + if _, ok := top[k]; ok { + t.Errorf("the manifest declares %q: the checker has none", k) + } + } +} + +func TestManifestProcess(t *testing.T) { // 1678/AC1 + m, _ := load(t) + var procs []map[string]any + for _, r := range m.Resources { + if r["type"] == "process" { + procs = append(procs, r) + } + if r["type"] == "tool" || r["type"] == "seat" { + t.Errorf("resource %v: the checker serves no tool and holds no seat", r) + } + } + if len(procs) != 1 { + t.Fatalf("%d process resources, want one", len(procs)) + } + p := procs[0] + if p["artifact"] != "program" || !reflect.DeepEqual(p["run"], []any{"./mesh-delivery-checker"}) { + t.Errorf("process = %v", p) + } + env, _ := p["env"].(map[string]any) + // The capacity of each node is the module's setting `capacity` (ADR 0484 clause 12, as the builder's, issue 1661), laid into the + // settings file the program reads; empty is one step at a time everywhere. The environment holds no capacity. + if len(env) != 2 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" || env["MESH_CHECKER_SETTINGS"] != "${dir:state}/settings.json" { + t.Errorf("env = %v", p["env"]) + } + if !reflect.DeepEqual(p["restart-on"], []any{"settings"}) { + t.Errorf("restart-on = %v, want the settings file, so a changed capacity reaches the running program", p["restart-on"]) + } + c, ok := m.Config["capacity"] + if !ok || len(m.Config) != 1 || c.Type != "string" || c.Default != "" || c.Description == "" { + t.Errorf("config = %+v, want one string setting, capacity, empty (one step at a time on every node) to start", m.Config) + } + var settings map[string]any + for _, r := range m.Resources { + if r["id"] == "settings" { + settings = r + } + } + if settings == nil || settings["type"] != "file" || settings["merge"] != "json" || !reflect.DeepEqual(settings["takes-config"], []any{"capacity"}) || settings["path"] != "${dir:state}/settings.json" { + t.Errorf("settings resource = %v", settings) + } + if raw, err := os.ReadFile("files/settings.json"); err != nil || !strings.Contains(string(raw), `"capacity": ""`) { + t.Errorf("files/settings.json = %q, %v, want the setting's starting value", raw, err) + } + if _, bus := p["bus"]; bus { + t.Errorf("the process declares a bus: it runs on its own broker credential") + } + if len(m.Build.Artifacts) != 2 || m.Build.Artifacts[1].Name != "files" || m.Build.Artifacts[1].Kind != "archive" || m.Build.Artifacts[1].From != "files" { + t.Fatalf("artifacts = %+v, want the program and the files", m.Build.Artifacts) + } + a := m.Build.Artifacts[0] + if st, err := os.Stat(a.From); err != nil || !st.IsDir() || a.Binary != "mesh-delivery-checker" || !strings.HasSuffix(a.From, a.Binary) || a.Kind != "bundle" || a.Language != "go" { + t.Errorf("artifact = %+v", a) + } +} + +func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1678/AC6 (ADR 0437 §8: no timer, a wait is a context's bound) + forbidden := []string{"Subscribe(", "QueueSubscribe", "ChanSubscribe", "time.NewTicker", "time.Tick", "time.NewTimer", "time.AfterFunc", "time.After(", "time.Sleep"} + var files int + for _, root := range []string{"cmd", "internal"} { + err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil || d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { + return err + } + files++ + src, err := os.ReadFile(path) + if err != nil { + return err + } + for _, f := range forbidden { + if strings.Contains(string(src), f) { + t.Errorf("%s holds %s: the checker serves no request subject and holds no timer", path, f) + } + } + return nil + }) + if err != nil { + t.Fatal(err) + } + } + if files < 1 { + t.Fatalf("read %d source files, want the program's", files) + } +} diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json new file mode 100644 index 0000000..8044314 --- /dev/null +++ b/modules/mesh-delivery-checker/module.json @@ -0,0 +1,113 @@ +{ + "module": "mesh-delivery-checker", + "version": "1", + "slug": "checker", + "consumes": [ + "mesh-build-go.toolchain.reported", + "mesh-build-node.toolchain.reported" + ], + "state": [ + { "name": "kept", "history": 1 }, + { "name": "owed", "history": 5 }, + { "name": "previews", "history": 5 }, + { "name": "verdicts", "history": 5 } + ], + "reads": [ + "mesh-build-go.steps", + "mesh-build-go.toolchains", + "mesh-build-node.steps", + "mesh-build-node.toolchains", + "mesh-controller.plans-building", + "mesh-delivery-builder.builds", + "mesh-delivery-gate.source", + "mesh-delivery-indexer.candidates", + "mesh-delivery-indexer.manifests", + "mesh-delivery-indexer.modules", + "mesh-delivery-watchman.presence", + "node-engine.have" + ], + "invokes": [ + "mesh-build-go.build_go_digest", + "mesh-build-go.build_go_fetch", + "mesh-build-go.build_go_test", + "mesh-build-node.build_node_fetch", + "mesh-build-node.build_node_test" + ], + "config": { + "capacity": { + "type": "string", + "description": "How many build steps the checker runs at once on each node, as node=steps pairs separated by commas, such as shanks=2,ace=1 (ADR 0484 clause 12); a node set to 0 never checks, a node not named runs one at a time, and empty is one everywhere. The builder's capacity is the builder's own.", + "pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*", + "default": "" + } + }, + "own-secrets": { + "broker": "${dir:state}/broker" + }, + "secrets-owner": "mesh-delivery-checker", + "resources": [ + { + "id": "account", + "type": "user", + "name": "mesh-delivery-checker", + "shell": "/usr/bin/nologin", + "home": "/var/lib/mesh-delivery-checker" + }, + { + "id": "state", + "type": "directory", + "mode": "0700", + "owner": "mesh-delivery-checker", + "place": ".", + "unknown": "program" + }, + { + "id": "settings", + "trusted": false, + "type": "file", + "path": "${dir:state}/settings.json", + "mode": "0600", + "owner": "mesh-delivery-checker", + "takes-config": ["capacity"], + "merge": "json", + "template": "files/settings.json" + }, + { + "id": "checker", + "type": "process", + "name": "mesh-delivery-checker", + "artifact": "program", + "run": [ + "./mesh-delivery-checker" + ], + "env": { + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_CHECKER_SETTINGS": "${dir:state}/settings.json" + }, + "restart-on": [ + "settings" + ], + "user": "mesh-delivery-checker", + "health": { + "kind": "unit" + } + } + ], + "build": { + "artifacts": [ + { + "name": "program", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-delivery-checker", + "binary": "mesh-delivery-checker" + }, + { + "name": "files", + "kind": "archive", + "from": "files" + } + ] + } +}