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)
This commit is contained in:
@@ -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.<digest> 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))
|
||||
}
|
||||
@@ -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{} }
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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`, `<node>=<steps at
|
||||
// once>` 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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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.<module>.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()
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"capacity": ""
|
||||
}
|
||||
@@ -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, `<group>.<heads hash>`.
|
||||
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)
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user