Checker: put every node's name on each check ask for the names check, so no repository lists them (ADR 0567 §12, issue 1126)
This commit is contained in:
@@ -2,12 +2,16 @@ package main
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
|
"github.com/nats-io/nats.go/jetstream"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/process"
|
||||||
|
|
||||||
"mesh-delivery-checker/internal/check"
|
"mesh-delivery-checker/internal/check"
|
||||||
)
|
)
|
||||||
@@ -25,11 +29,46 @@ type judgeDeps struct {
|
|||||||
builds check.BuildStates
|
builds check.BuildStates
|
||||||
// compositions is the assigner's bucket of compositions; nil leaves every composition not run.
|
// compositions is the assigner's bucket of compositions; nil leaves every composition not run.
|
||||||
compositions check.CompositionReader
|
compositions check.CompositionReader
|
||||||
logf func(format string, args ...any)
|
// nodes names every node with a record; nil hands no names.
|
||||||
|
nodes *nodeNames
|
||||||
|
logf func(format string, args ...any)
|
||||||
|
}
|
||||||
|
|
||||||
|
// nodeNames reads every node's name from the node-engine's have, one key `<node>.node` per node with a record (novox/hq
|
||||||
|
// ADR 0567 §12), at each ask: a node enrolled since is named in the next check.
|
||||||
|
type nodeNames struct{ kv jetstream.KeyValue }
|
||||||
|
|
||||||
|
// openNodeNames opens the node-engine's have for the nodes' names alone: it is not watched, so no check waits on a node.
|
||||||
|
func openNodeNames(ctx context.Context, js process.BucketOpener, logf func(string, ...any)) *nodeNames {
|
||||||
|
return &nodeNames{kv: process.OptionalBucket(ctx, js, delivery.HaveBucket, "no check is asked until it is: the names check needs the nodes' names", logf)}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n nodeNames) Names(ctx context.Context) ([]string, error) {
|
||||||
|
if n.kv == nil {
|
||||||
|
return nil, fmt.Errorf("the bucket %s could not be opened at the checker's start", delivery.HaveBucket)
|
||||||
|
}
|
||||||
|
lister, err := n.kv.ListKeysFiltered(ctx, "*."+delivery.HaveNodeWord)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
var out []string
|
||||||
|
for key := range lister.Keys() {
|
||||||
|
if node, _, isNode, err := delivery.ParseHaveKey(key); err == nil && isNode {
|
||||||
|
out = append(out, node)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
sort.Strings(out)
|
||||||
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func newJudge(d judgeDeps) check.Judge {
|
func newJudge(d judgeDeps) check.Judge {
|
||||||
return &check.Examiner{Compositions: d.compositions, Asks: d.asks, Toolchains: d.tools,
|
var nodes interface {
|
||||||
|
Names(ctx context.Context) ([]string, error)
|
||||||
|
}
|
||||||
|
if d.nodes != nil {
|
||||||
|
nodes = d.nodes
|
||||||
|
}
|
||||||
|
return &check.Examiner{Nodes: nodes, Compositions: d.compositions, Asks: d.asks, Toolchains: d.tools,
|
||||||
Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Packages: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf}
|
Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Packages: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -194,6 +194,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
|||||||
kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]},
|
kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]},
|
||||||
repos: repos, builds: builds, forge: forgeReads{call: caller.Call}, logf: logger.Printf,
|
repos: repos, builds: builds, forge: forgeReads{call: caller.Call}, logf: logger.Printf,
|
||||||
announce: announce, compositions: compositions,
|
announce: announce, compositions: compositions,
|
||||||
|
nodes: openNodeNames(ctx, js, logger.Printf),
|
||||||
})
|
})
|
||||||
ch.Announce = announce
|
ch.Announce = announce
|
||||||
tryer.Judge = ch.Judge
|
tryer.Judge = ch.Judge
|
||||||
|
|||||||
@@ -7,7 +7,8 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// 1912: with checks as asks on the build seats' streams, the stream redelivers a lost node's ask (ADR 0525 §6), so the
|
// 1912: with checks as asks on the build seats' streams, the stream redelivers a lost node's ask (ADR 0525 §6), so the
|
||||||
// checker program opens no node-engine have to start again a check that failed for want of a node.
|
// checker program watches no node-engine have to start again a check that failed for want of a node. It reads the have's
|
||||||
|
// keys only for the nodes' names at each ask (judge.go, novox/hq ADR 0567 §12).
|
||||||
func TestTheCheckerWatchesNoHave(t *testing.T) {
|
func TestTheCheckerWatchesNoHave(t *testing.T) {
|
||||||
raw, err := os.ReadFile("main.go")
|
raw, err := os.ReadFile("main.go")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -20,3 +20,5 @@ require (
|
|||||||
golang.org/x/sys v0.48.0 // indirect
|
golang.org/x/sys v0.48.0 // indirect
|
||||||
golang.org/x/time v0.15.0 // indirect
|
golang.org/x/time v0.15.0 // indirect
|
||||||
)
|
)
|
||||||
|
|
||||||
|
replace git.novox.be/novox/mesh-sdk/go => /home/jochen/projects/novox/mesh-sdk/go
|
||||||
|
|||||||
@@ -1,5 +1,3 @@
|
|||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
|
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -272,8 +273,36 @@ func TestACleanMergeIsFetchedAtItsTrunkAndBothLevelsPass(t *testing.T) { // 1679
|
|||||||
t.Errorf("%s was not kept", k)
|
t.Errorf("%s was not kept", k)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ADR 0567 §12: every check ask carries every node's name for the names check; a check whose names cannot be read, or
|
||||||
|
// that reads none, is not asked and fails saying so.
|
||||||
|
for _, c := range []struct {
|
||||||
|
names []string
|
||||||
|
err error
|
||||||
|
fails string
|
||||||
|
}{{[]string{"anchor", "laptop"}, nil, ""}, {nil, nil, "no node record"}, {nil, errors.New("no bucket"), "no bucket"}} {
|
||||||
|
b := newBench(t)
|
||||||
|
b.ex.Nodes = namesOf{c.names, c.err}
|
||||||
|
out := b.judge(t, caseOf(t, pullOf(repoA, 7, head1)))
|
||||||
|
asks := b.asks.Asks()
|
||||||
|
if c.fails != "" {
|
||||||
|
if out.Verdict.State != delivery.VerdictFail || len(asks) != 0 || !strings.Contains(out.Verdict.Failed.Words, c.fails) {
|
||||||
|
t.Errorf("names %v, %v: verdict %+v, %d asks", c.names, c.err, out.Verdict, len(asks))
|
||||||
|
}
|
||||||
|
} else if len(asks) == 0 || strings.Join(asks[0].Check.NodeNames, ",") != "anchor,laptop" {
|
||||||
|
t.Errorf("the check ask carries the names %v", asks)
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// namesOf is the nodes' names as a test gives them.
|
||||||
|
type namesOf struct {
|
||||||
|
names []string
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n namesOf) Names(context.Context) ([]string, error) { return n.names, n.err }
|
||||||
|
|
||||||
func TestOnlyThePackageWithoutAKeptVerdictRuns(t *testing.T) { // 1288/AC1
|
func TestOnlyThePackageWithoutAKeptVerdictRuns(t *testing.T) { // 1288/AC1
|
||||||
b := newBench(t)
|
b := newBench(t)
|
||||||
b.digest[delivery.LevelUnit] = Digested{Level: "unit", Packages: map[string]string{"fixture/slow": digA, "fixture/other": digC}}
|
b.digest[delivery.LevelUnit] = Digested{Level: "unit", Packages: map[string]string{"fixture/slow": digA, "fixture/other": digC}}
|
||||||
|
|||||||
@@ -110,8 +110,14 @@ type Examiner struct {
|
|||||||
// waiting until the assigner wrote a verdict that does not wait, or ctx ends (ADR 0528 clause 5, issue 1297); nil
|
// waiting until the assigner wrote a verdict that does not wait, or ctx ends (ADR 0528 clause 5, issue 1297); nil
|
||||||
// leaves the composition not run.
|
// leaves the composition not run.
|
||||||
Compositions CompositionReader
|
Compositions CompositionReader
|
||||||
Now func() time.Time
|
// Nodes names every node with a record, read from each `<node>.node` of the node-engine's have (novox/hq ADR 0567
|
||||||
Logf func(format string, args ...any)
|
// §12): every check ask carries them (CheckJob.NodeNames) for the names check, so no repository lists them. A check
|
||||||
|
// whose names cannot be read, or that reads none, is not asked and fails saying so; nil hands no names.
|
||||||
|
Nodes interface {
|
||||||
|
Names(ctx context.Context) ([]string, error)
|
||||||
|
}
|
||||||
|
Now func() time.Time
|
||||||
|
Logf func(format string, args ...any)
|
||||||
}
|
}
|
||||||
|
|
||||||
// CompositionReader answers the assigner's composition of a check by the check's key, waiting until the assigner wrote
|
// CompositionReader answers the assigner's composition of a check by the check's key, waiting until the assigner wrote
|
||||||
@@ -403,6 +409,16 @@ func (st *state) askNeeding(ctx context.Context, t *target, merged *[]Fetched, l
|
|||||||
p := t.pull
|
p := t.pull
|
||||||
placed := len(needs) > 0
|
placed := len(needs) > 0
|
||||||
job := &delivery.CheckJob{Fetch: t.fetchArgs, Levels: levels, Kept: true, Packages: packages}
|
job := &delivery.CheckJob{Fetch: t.fetchArgs, Levels: levels, Kept: true, Packages: packages}
|
||||||
|
if st.j.Nodes != nil {
|
||||||
|
names, err := st.j.Nodes.Names(ctx)
|
||||||
|
if err == nil && len(names) == 0 {
|
||||||
|
err = fmt.Errorf("no node record is on the bus")
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return fail(t.module, fmt.Sprintf("the check of %s#%d is not asked: the nodes' names the names check needs cannot be read: %v", p.Repository, p.Number, err))
|
||||||
|
}
|
||||||
|
job.NodeNames = names
|
||||||
|
}
|
||||||
if t.runner != "" {
|
if t.runner != "" {
|
||||||
job.Runner, job.RunnerWhen = t.runner, buildGoDir
|
job.Runner, job.RunnerWhen = t.runner, buildGoDir
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user