mesh-delivery-assigner: compose nothing; judge proposals with its own judge and compare the published compositions
ADR 0593 §7, §8: the check's composition, its compose asks to the Go build seat and the state compositions go (the checker asks the engines); a proposal is judged without frombus; mesh-wants.upgrade compare reads every node's composed-<node> through mesh-host.composed. The assigner no longer imports compose, composer, frombus or buildrun. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -0,0 +1,73 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// Every node's published compositions, read only to compare (novox/hq ADR 0593 §8): each node's engine writes its own
|
||||
// `composed-<node>` (delivery.ComposedRead, one bucket per node), and `mesh-wants.upgrade` with compare names, for every
|
||||
// node, the composition its own have runs beside the newest it published of each module and of the node, so a person sees
|
||||
// where they part. The assigner composes nothing and never writes these.
|
||||
|
||||
// publishedRow is one composition a node published: the module ("node" for its own), the fingerprint, the digest, or why
|
||||
// it was refused or cannot be read.
|
||||
type publishedRow struct {
|
||||
Key string `json:"key"`
|
||||
Module string `json:"module"`
|
||||
FP string `json:"fp,omitempty"`
|
||||
Digest string `json:"digest,omitempty"`
|
||||
Refused string `json:"refused,omitempty"`
|
||||
Unread string `json:"unread,omitempty"`
|
||||
}
|
||||
|
||||
// published is, by node, every composition it published, sorted by key, from every key of the per-node state (a key
|
||||
// names its node first): each value decoded and digested here, as the node-engine digests it (delivery.ComposedDigestOf).
|
||||
// A value or key that does not read is named with why, never left out.
|
||||
func published(values map[string][]byte) map[string][]publishedRow {
|
||||
byNode := map[string]map[string][]byte{}
|
||||
for key, raw := range values {
|
||||
node, _, _ := strings.Cut(key, ".")
|
||||
if byNode[node] == nil {
|
||||
byNode[node] = map[string][]byte{}
|
||||
}
|
||||
byNode[node][key] = raw
|
||||
}
|
||||
out := map[string][]publishedRow{}
|
||||
for node, keys := range byNode {
|
||||
var rows []publishedRow
|
||||
for key, raw := range keys {
|
||||
row := publishedRow{Key: key}
|
||||
_, module, fp, isNode, err := delivery.ParseComposedKey(key)
|
||||
switch {
|
||||
case err != nil:
|
||||
row.Unread = err.Error()
|
||||
case isNode:
|
||||
row.Module = "node"
|
||||
default:
|
||||
row.Module, row.FP = module, fp
|
||||
}
|
||||
if err == nil {
|
||||
c, derr := delivery.DecodeComposed(raw)
|
||||
switch {
|
||||
case derr != nil:
|
||||
row.Unread = derr.Error()
|
||||
case c.Refused != "":
|
||||
row.Refused = c.Refused
|
||||
default:
|
||||
if d, err := delivery.ComposedDigestOf(c); err != nil {
|
||||
row.Unread = err.Error()
|
||||
} else {
|
||||
row.Digest = d
|
||||
}
|
||||
}
|
||||
}
|
||||
rows = append(rows, row)
|
||||
}
|
||||
sort.Slice(rows, func(i, j int) bool { return rows[i].Key < rows[j].Key })
|
||||
out[node] = rows
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -24,10 +24,10 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/judge"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"mesh-delivery-assigner/internal/judge"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
@@ -290,19 +290,6 @@ func (b *bed) place(t *testing.T, node, module string) {
|
||||
}
|
||||
}
|
||||
|
||||
// built appends the end of a done trunk build of module at commit to the builder's log, as the builder does.
|
||||
func (b *bed) built(t *testing.T, module, repository, commit string) {
|
||||
t.Helper()
|
||||
raw, err := delivery.EncodeBuildLogEntry(delivery.BuildLogEntry{Event: delivery.BuildEndedEvent, Module: module, FP: delivery.SourcePrefix2 + strings.Repeat("c", 64),
|
||||
Repository: repository, Directory: delivery.RootPath, Commit: commit, Node: "novox", At: time.Now().UTC(), Outcome: delivery.BuildDone})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := b.js.Publish(b.ctx, buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog)+"."+module, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// compose says the node's own composition is the digest d (a letter repeated), as its have names it (`<node>.node`): a
|
||||
// placed node composes its own (ADR 0571 §6), and the controller's composition is read no more. It answers the digest.
|
||||
func (b *bed) compose(t *testing.T, node string, d byte) string {
|
||||
@@ -333,62 +320,6 @@ func (b *bed) manifest(t *testing.T, module, body string) string {
|
||||
return digest
|
||||
}
|
||||
|
||||
// 1297/AC1: against the real bus, a fixture change touching one module on one machine is composed and answered within
|
||||
// seconds under the checker's key of the check, naming the machine composed and the mesh-sdk it composed with.
|
||||
func TestAChangeIsComposedOnTheBus(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
trunk := b.manifest(t, "web", `{"module":"web","version":"1"}`)
|
||||
changed := b.manifest(t, "web", `{"module":"web","version":"2"}`)
|
||||
mod, err := delivery.EncodeModule(delivery.Module{FP2: delivery.SourcePrefix2 + strings.Repeat("a", 64), Manifest: trunk, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.putJSON(t, delivery.ModulesBucket, "web", mod)
|
||||
b.place(t, "g14", "web")
|
||||
b.place(t, "ace", "other")
|
||||
heads := []delivery.Head{{Repository: "novox/web", Number: 7, Head: strings.Repeat("2", 40)}}
|
||||
hash, _ := delivery.HeadsHash(heads)
|
||||
// 1927/AC1: the candidate carries the fingerprint set the SDK computes, as the indexer writes it.
|
||||
pulls := []delivery.CandidatePull{{Head: heads[0],
|
||||
Modules: []string{"web"}, Fps: map[string]delivery.CandidateModule{"web": {FP2: delivery.SourcePrefix2 + strings.Repeat("b", 64), Manifest: changed}}}}
|
||||
cand, err := delivery.EncodeCandidate(delivery.Candidate{Group: "feat/x", HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ckey, _ := delivery.CandidatesKey("feat/x")
|
||||
b.putJSON(t, delivery.CandidatesBucket, ckey, cand)
|
||||
began := time.Now()
|
||||
b.start(t)
|
||||
key, _ := delivery.CompositionKey("feat/x", hash)
|
||||
var v delivery.CompositionVerdict
|
||||
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(50 * time.Millisecond) {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("no composition under %s after 10 s", key)
|
||||
}
|
||||
e, err := b.buckets[delivery.AssignerCompositionsBucket].Get(b.ctx, key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if v, err = delivery.DecodeCompositionVerdict(e.Value()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
break
|
||||
}
|
||||
if v.State != delivery.CompositionPass || len(v.Machines) != 1 || v.Machines[0] != "g14" || v.SDK == "" {
|
||||
t.Fatalf("composition = %+v, want a pass on g14 alone naming its mesh-sdk", v)
|
||||
}
|
||||
if took := time.Since(began); took > 5*time.Second {
|
||||
t.Errorf("the composition took %s", took)
|
||||
}
|
||||
// 1622/AC3: a group without a mesh-sdk member builds no composer: no ask is put on the Go build seat.
|
||||
seat, _ := delivery.AskStream(delivery.AskGo)
|
||||
if st, err := b.js.Stream(b.ctx, seat); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if info, err := st.Info(b.ctx); err != nil || info.State.Msgs != 0 {
|
||||
t.Errorf("a group without a mesh-sdk member put %d asks on the Go build seat (%v)", info.State.Msgs, err)
|
||||
}
|
||||
}
|
||||
|
||||
// get waits for a key of one of the bed's buckets and answers its value.
|
||||
func (b *bed) get(t *testing.T, bucket, key string) []byte {
|
||||
t.Helper()
|
||||
@@ -1121,16 +1052,17 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
|
||||
t.Error("a declined proposal wrote its key")
|
||||
}
|
||||
|
||||
// ADR 0567 §4: a node that does not answer fails the verdict once its pulse key's time to live ends, heard on the bus
|
||||
// with no timer of the assigner's, naming it.
|
||||
// ADR 0593 §8: a node that does not answer is named as giving no answer once its pulse key's time to live ends, heard on
|
||||
// the bus with no timer of the assigner's, and the proposal is decided on the answers it has: asked, not failed.
|
||||
q, err = verbs.doPropose(map[string]any{"why": "again", "changes": `[{"op":"take-off","node":"laptop","module":"web"}]`}, "op")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
silent := q.(proposed).Proposal
|
||||
failed := proposalNow(silent.ID, "failed for a silent node", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalFailed })
|
||||
if failed.Verdict == nil || !strings.Contains(strings.Join(failed.Verdict.Fails, "; "), "laptop (offline) did not answer") {
|
||||
t.Errorf("the silent node's verdict = %+v", failed.Verdict)
|
||||
decided := proposalNow(silent.ID, "decided without its silent nodes", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalAsked })
|
||||
if decided.Verdict == nil || decided.Verdict.State != delivery.VerdictPass ||
|
||||
!strings.Contains(strings.Join(decided.Verdict.Notes, "; "), "no answer from") || !strings.Contains(strings.Join(decided.Verdict.Notes, "; "), "laptop (offline)") {
|
||||
t.Errorf("the silent node's verdict = %+v", decided.Verdict)
|
||||
}
|
||||
|
||||
// 774/S59 (ADR 0587 §7): a joining node is written on the enrolment's warrant with no node to answer and no ask raised,
|
||||
|
||||
@@ -15,18 +15,14 @@ import (
|
||||
"os"
|
||||
"os/signal"
|
||||
"slices"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
@@ -56,31 +52,23 @@ func main() {
|
||||
// readBuckets are the buckets of other parts the assigner reads, each named in its manifest's reads (the manifest test
|
||||
// holds the two lists equal), so the controller's broker grants exactly these.
|
||||
var readBuckets = []string{
|
||||
buscompose.ResolvedBucket,
|
||||
delivery.PartGrantsBucket,
|
||||
delivery.AnswersBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
delivery.CandidatesBucket,
|
||||
delivery.ManifestsBucket,
|
||||
delivery.ModulesBucket,
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.StepsBucket(goBuildModule),
|
||||
provisions.ProvisionsBucket,
|
||||
keptBucket,
|
||||
}
|
||||
|
||||
// readLogs are the logs of other parts the assigner reads, each named in its manifest's reads: the builder's log, whose
|
||||
// newest done trunk build of a module names the commit a reader of the manifests runs at (novox/hq ADR 0571 clause 1).
|
||||
var readLogs = []string{delivery.BuildLogRead}
|
||||
|
||||
// goBuildModule is the Go build module, which takes the assigner's compose asks on its seat's check stream and writes
|
||||
// their steps in its steps state (novox/hq ADR 0482 clause 4).
|
||||
const goBuildModule = "mesh-build-go"
|
||||
// readFamilies are the states kept one bucket per node the assigner reads, each named in its manifest's reads: every
|
||||
// node's compositions, which mesh-wants.upgrade compares (novox/hq ADR 0593 §8).
|
||||
var readFamilies = []string{delivery.ComposedRead}
|
||||
|
||||
// ownBuckets are the assigner's own states, as its manifest declares them: the seat mesh-placement's three among them
|
||||
// (ADR 0567 §1), named after the assigner as every module state is.
|
||||
var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket,
|
||||
var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerDifferencesBucket,
|
||||
delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket,
|
||||
delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, delivery.SeatHoldersBucket}
|
||||
|
||||
@@ -90,7 +78,7 @@ var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposals
|
||||
delivery.PlacementPulsesBucket}
|
||||
|
||||
// keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view
|
||||
// (and the node agent from release 6), its differences by the view, its compositions by the checker, its asks by the view
|
||||
// (and the node agent from release 6), its differences by the view, its asks by the view
|
||||
// and the teller, its puts by whoever follows one (ADR 0546), and the seat mesh-placement's placement, proposals and
|
||||
// candidates by the node agents, the grants part, mesh-config and the view (ADR 0567 §9), and its placement by
|
||||
// the builder, the checker and the indexer, through mesh-sdk's toolchain gate (issue 774, ADR 0549 §3).
|
||||
@@ -100,7 +88,6 @@ var keptStates = []announce.State{
|
||||
{Name: delivery.PlacementCandidatesState, Readable: true},
|
||||
{Name: delivery.PlacementPulsesState, Readable: true},
|
||||
{Name: "asks", Readable: true},
|
||||
{Name: "compositions", Readable: true},
|
||||
{Name: "differences", Readable: true},
|
||||
{Name: "puts", Readable: true},
|
||||
{Name: "wants", Readable: true},
|
||||
@@ -135,11 +122,6 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
|
||||
continue
|
||||
}
|
||||
if name == keptBucket && err != nil {
|
||||
// mesh-config's held values (ADR 0565 §5): without them a composition lays none, said in the journal.
|
||||
logf("the bucket %s cannot be opened (%v): no held value is laid on a composition until it is bound", name, err)
|
||||
continue
|
||||
}
|
||||
if name == provisions.ProvisionsBucket && err != nil {
|
||||
// The seat mesh-provisioner's state, with its own delivery (ADR 0570 §4): bound once the bus says it was made;
|
||||
// until then no want waits for a provision, as before.
|
||||
@@ -175,7 +157,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
// states and those new with a delivery of their own, bound once made, and the two that choose the grants' source.
|
||||
func startsWithout(name string) bool {
|
||||
switch name {
|
||||
case delivery.SeatHoldersBucket, keptBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket,
|
||||
case delivery.SeatHoldersBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket,
|
||||
delivery.AssignerPutsBucket:
|
||||
return true
|
||||
}
|
||||
@@ -211,10 +193,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
thisNode = account.Node
|
||||
logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
|
||||
findings := findingsOf(ctx, kvs, logger.Printf)
|
||||
buildLog := func(handle func(stdio.LogEntry) error) error {
|
||||
return said.ReadLog(delivery.BuilderModule, delivery.BuilderLog).Watch("", stdio.LogFromAll, handle)
|
||||
}
|
||||
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, buildLog, logger)
|
||||
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, logger)
|
||||
}
|
||||
|
||||
// findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, which the
|
||||
@@ -246,11 +225,9 @@ func (e emitting) Emit(eventType string, body any) error { return e.Conn.Emit(ev
|
||||
// watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and
|
||||
// watched only for keys, so a manifest that arrives wakes a composition that waits for it.
|
||||
var watched = []string{
|
||||
buscompose.ResolvedBucket,
|
||||
delivery.PartGrantsBucket,
|
||||
delivery.AnswersBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
delivery.CandidatesBucket,
|
||||
delivery.ModulesBucket,
|
||||
delivery.PlacementBucket,
|
||||
delivery.PlacementProposalsBucket,
|
||||
@@ -259,18 +236,16 @@ var watched = []string{
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.AssignerAsksBucket,
|
||||
delivery.AssignerCompositionsBucket,
|
||||
delivery.AssignerDifferencesBucket,
|
||||
delivery.AssignerWantsBucket,
|
||||
assign.PoliciesBucket,
|
||||
provisions.ProvisionsBucket,
|
||||
keptBucket,
|
||||
delivery.SeatHoldersBucket,
|
||||
}
|
||||
|
||||
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
|
||||
// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends.
|
||||
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, buildLog logWatch, logger *log.Logger) error {
|
||||
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, logger *log.Logger) error {
|
||||
w := newWorld()
|
||||
dirty := make(chan struct{}, 1)
|
||||
wake := func() {
|
||||
@@ -279,22 +254,6 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
default:
|
||||
}
|
||||
}
|
||||
// The builder's log (ADR 0571 clause 1): every entry it keeps before the first pass, then each new one as it is appended;
|
||||
// a module's newer done trunk build makes a pass, as a composition waiting on its commit is judged then. Where it cannot be
|
||||
// read, a composition names no running reader of the manifests until the assigner starts again, and it is said.
|
||||
if buildLog != nil {
|
||||
var first atomic.Bool
|
||||
if err := buildLog(func(le stdio.LogEntry) error {
|
||||
if w.putBuilt(le.Value) && first.Load() {
|
||||
wake()
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
logger.Printf("%s is not read (%v): no composition is judged against a running reader of the manifests until the assigner starts again",
|
||||
delivery.BuildLogRead, err)
|
||||
}
|
||||
first.Store(true)
|
||||
}
|
||||
errs := make(chan error, len(watched)+2)
|
||||
settled := make(chan struct{}, len(watched)+2)
|
||||
watch := func(name string, opts ...jetstream.WatchOpt) error {
|
||||
@@ -402,7 +361,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
}
|
||||
// And the two that choose where grants are read from, with the same delivery as the grants part that writes one.
|
||||
for _, name := range []string{delivery.PartGrantsBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, keptBucket, delivery.SeatHoldersBucket} {
|
||||
for _, name := range []string{delivery.PartGrantsBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, delivery.SeatHoldersBucket} {
|
||||
if kvs[name] == nil && js != nil {
|
||||
name := name
|
||||
process.AwaitBucket(ctx, js, nc, name, "until then the bucket is not read",
|
||||
@@ -427,7 +386,6 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
}
|
||||
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{}}
|
||||
p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf)
|
||||
if bus != nil && kvs[delivery.AssignerPutsBucket] != nil {
|
||||
p.acts = newPutActs(bus, wake, logger.Printf)
|
||||
}
|
||||
|
||||
@@ -40,14 +40,14 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
|
||||
}
|
||||
var reads []string
|
||||
for _, r := range m.Reads {
|
||||
if slices.Contains(readLogs, r) {
|
||||
continue // a log, read through its own reader (ADR 0571 clause 1), not a bucket
|
||||
if slices.Contains(readFamilies, r) {
|
||||
continue // a state kept one bucket per node, read by its prefix (ADR 0593 §6), not one bucket
|
||||
}
|
||||
reads = append(reads, strings.Replace(r, ".", "_", 1))
|
||||
}
|
||||
for _, l := range readLogs {
|
||||
for _, l := range readFamilies {
|
||||
if !slices.Contains(m.Reads, l) {
|
||||
t.Errorf("the program reads the log %s, the manifest does not name it", l)
|
||||
t.Errorf("the program reads the per-node state %s, the manifest does not name it", l)
|
||||
}
|
||||
}
|
||||
var states []string
|
||||
@@ -113,7 +113,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
|
||||
}
|
||||
kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true
|
||||
kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote
|
||||
go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, nil, log.New(io.Discard, "", 0))
|
||||
go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, log.New(io.Discard, "", 0))
|
||||
select {
|
||||
case key := <-wrote:
|
||||
if key != "want.ace.node" {
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
"git.novox.be/novox/mesh-sdk/go/sdkversion"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
@@ -40,7 +39,6 @@ type passer struct {
|
||||
unplaced map[string]bool
|
||||
// answeredAt is, by the id of a backup ask, how many values the have had when its answer was written into the want.
|
||||
answeredAt map[string]uint64
|
||||
asks *composeAsks
|
||||
// now is the clock; nil is time.Now.
|
||||
now func() time.Time
|
||||
// neither is, by node and module, each module on neither list whose condition this process raised (issue 1945). Nil until the
|
||||
@@ -81,7 +79,6 @@ func (p *passer) clock() time.Time {
|
||||
|
||||
func (p *passer) pass(ctx context.Context) {
|
||||
p.placementSeat(ctx)
|
||||
p.compositions(ctx)
|
||||
p.wants(ctx)
|
||||
p.puts(ctx)
|
||||
p.nodeWants(ctx)
|
||||
@@ -478,57 +475,6 @@ func (p *passer) sayNeither(now, stale map[string]neitherOne, left map[string]st
|
||||
}
|
||||
}
|
||||
|
||||
// compositions writes the composition check's verdict of every group the indexer holds a candidate for, under the
|
||||
// checker's key of that check: waits, with why, while the indexer holds what it composes from only in part.
|
||||
func (p *passer) compositions(ctx context.Context) {
|
||||
if p.sdk == "" {
|
||||
p.sdk = sdkversion.Version()
|
||||
}
|
||||
written := p.w.read(delivery.AssignerCompositionsBucket)
|
||||
trunk := p.w.trunk()
|
||||
var mesh *assign.Mesh
|
||||
seen := map[string]bool{}
|
||||
if p.asks != nil {
|
||||
defer p.asks.sweep(seen)
|
||||
}
|
||||
for key, value := range p.w.read(delivery.CandidatesBucket) {
|
||||
c, err := delivery.DecodeCandidate(value)
|
||||
if err != nil {
|
||||
p.logf("the candidate %s cannot be read, so it is not composed: %v", key, err)
|
||||
continue
|
||||
}
|
||||
out, err := delivery.CompositionKey(c.Group, c.HeadsHash)
|
||||
if err != nil {
|
||||
p.logf("the candidate %s names no composition key: %v", key, err)
|
||||
continue
|
||||
}
|
||||
began := time.Now()
|
||||
ch, waits := p.w.change(ctx, kvGetter(p.kvs[delivery.ManifestsBucket]), c, trunk)
|
||||
var v delivery.CompositionVerdict
|
||||
if waits != "" {
|
||||
v = delivery.CompositionVerdict{Group: c.Group, Heads: c.HeadsHash, State: delivery.CompositionWaits, Words: waits}
|
||||
} else {
|
||||
if mesh == nil {
|
||||
m := p.w.mesh(ctx, kvGetter(p.kvs[delivery.ManifestsBucket]))
|
||||
mesh = &m
|
||||
}
|
||||
if ch.SDKHead != "" && p.asks != nil {
|
||||
// A group with a mesh-sdk member is composed by the composer built at its head (ADR 0482 clauses 3 to 5).
|
||||
v = p.composeAtHead(ctx, *mesh, ch, seen, began)
|
||||
} else {
|
||||
v = assign.Compose(*mesh, ch, p.sdk, time.Since(began))
|
||||
}
|
||||
}
|
||||
if before, err := delivery.DecodeCompositionVerdict(written[out]); err == nil {
|
||||
before.Took, before.V = v.Took, 0
|
||||
if delivery.SameCompositionVerdict(before, v) {
|
||||
continue
|
||||
}
|
||||
}
|
||||
p.put(ctx, delivery.AssignerCompositionsBucket, out, func() ([]byte, error) { return delivery.EncodeCompositionVerdict(v) })
|
||||
}
|
||||
}
|
||||
|
||||
// put writes one key of the assigner's own bucket; a write that fails is said and made again at the next pass.
|
||||
func (p *passer) put(ctx context.Context, bucket, key string, encode func() ([]byte, error)) {
|
||||
value, err := encode()
|
||||
|
||||
@@ -791,35 +791,18 @@ func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing
|
||||
}
|
||||
|
||||
// 774/S26/AC1 (ADR 0549 §5, ADR 0567 §8): the seat's placement alone names where modules run: no controller state is read, so a
|
||||
// module in an assignment only is no module of its node; before any placement is read nothing is placed and the check composes
|
||||
// no input; placed, the check composes over the input the nodes compose from (frombus, 1126, ADR 0547 §2).
|
||||
// module in an assignment only is no module of its node, and before any placement is read nothing is placed.
|
||||
func aPlacementAloneNamesWhereModulesRun(t *testing.T) {
|
||||
p := newPasser(buckets{}, time.Date(2026, 10, 26, 18, 0, 0, 0, time.UTC))
|
||||
p.w.put("mesh-controller_assignments", "g14.b", []byte(`{"node":"g14","module":"b","build":"b-1"}`)) // read no more
|
||||
if got := p.w.assigned(); len(got) != 0 {
|
||||
t.Fatalf("before any placement: assigned %v", got)
|
||||
}
|
||||
none := func(context.Context, string) ([]byte, error) { return nil, errNotThere }
|
||||
if m := p.w.mesh(context.Background(), none); m.Input != nil || m.InputUnread != "" {
|
||||
t.Errorf("before any placement the check composes over %+v (%q)", m.Input, m.InputUnread)
|
||||
}
|
||||
placeOn(t, p.w, "g14", "c", "a")
|
||||
placeOn(t, p.w, "n2", "d")
|
||||
if got := p.w.assigned(); !slices.Equal(got["g14"], []string{"a", "c"}) || !slices.Equal(got["n2"], []string{"d"}) || len(got) != 2 {
|
||||
t.Errorf("the placement alone: assigned %v", got)
|
||||
}
|
||||
if m := p.w.mesh(context.Background(), none); m.Input != nil || !strings.Contains(m.InputUnread, "g14") {
|
||||
t.Errorf("a placed node with no record: input %+v, unread %q", m.Input, m.InputUnread)
|
||||
}
|
||||
placeOn(t, p.w, "g14", "a", "b", "c")
|
||||
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "g14", TunnelKey: "k"}})
|
||||
p.w.put(delivery.HaveBucket, "g14.node", rec)
|
||||
rec2, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n2", TunnelKey: "k2"}})
|
||||
p.w.put(delivery.HaveBucket, "n2.node", rec2)
|
||||
m := p.w.mesh(context.Background(), none)
|
||||
if m.Input == nil || len(m.Input.Nodes) != 2 || !slices.Equal(m.Input.Nodes[0].Assigned, []string{"a", "b", "c"}) || m.Input.Nodes[0].Place.Key != "k" {
|
||||
t.Errorf("the check's input from the bus: %+v (%q)", m.Input, m.InputUnread)
|
||||
}
|
||||
}
|
||||
|
||||
// 1126 (ADR 0547 §2 to §4): a node the catalogue places composes its own: its wants name the build alone and its own want
|
||||
|
||||
@@ -16,14 +16,12 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/judge"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/handacts"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
"mesh-delivery-assigner/internal/judge"
|
||||
)
|
||||
|
||||
// The seat mesh-placement's holder in the assigner's process (novox/hq ADR 0567 §2-§8): each pass follows every open
|
||||
@@ -354,67 +352,35 @@ func moved(prop delivery.PlacementProposal, revs map[string]uint64) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// intended is, by proposal and the revisions it was laid over, what Intended computed, kept while this process runs.
|
||||
// intendedOf is, by proposal and the revisions it was laid over, what Intended computed, kept while this process runs.
|
||||
func (p *passer) intendedOf(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) map[string]map[string]bool {
|
||||
key := fmt.Sprint(prop.ID, prop.Base)
|
||||
if got, ok := p.intended[key]; ok {
|
||||
return got
|
||||
}
|
||||
got, _, _ := p.compute(prop, current, cands)
|
||||
got, _ := p.compute(prop, current, cands)
|
||||
p.intended[key] = got
|
||||
return got
|
||||
}
|
||||
|
||||
// compute composes every node with the current placement and with the proposal laid over it (judge.Intended).
|
||||
func (p *passer) compute(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) (map[string]map[string]bool, []string, []string) {
|
||||
// compute is what the proposal changes on each node by its placement, and the nodes that answer it: every node with a
|
||||
// placement or a record (judge.Intended, ADR 0593 §8). The holder composes nothing: each node says what its compositions
|
||||
// change in its answer.
|
||||
func (p *passer) compute(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) (map[string]map[string]bool, []string) {
|
||||
laid := map[string]*delivery.Placement{}
|
||||
for _, node := range prop.Nodes() {
|
||||
if c, ok := cands[node]; ok && !c.Reached {
|
||||
laid[node] = c.Placement
|
||||
}
|
||||
}
|
||||
get := kvGetter(p.kvs[delivery.ManifestsBucket])
|
||||
r := worldReader{w: p.w, manifests: get}
|
||||
// A read that fails is said, never dropped silently (novox/hq ADR 0578 §3); a refused manifest no longer fails it, and
|
||||
// its module is held, named in the input's Refused.
|
||||
var now, with *compose.Input
|
||||
if in, err := frombus.Input(context.Background(), r, frombus.Mesh{}, frombus.Scope{Every: true}); err == nil {
|
||||
now = &in
|
||||
} else if p.logf != nil {
|
||||
p.logf("proposal %s: the current composition input cannot be read, so what it changes is judged without it: %v", prop.ID, err)
|
||||
}
|
||||
if in, err := frombus.Input(context.Background(), r, frombus.Mesh{}, frombus.Scope{Every: true, Placements: laid}); err == nil {
|
||||
with = &in
|
||||
} else if p.logf != nil {
|
||||
p.logf("proposal %s: the composition input with the proposal cannot be read, so what it changes is judged without it: %v", prop.ID, err)
|
||||
}
|
||||
if now != nil && len(now.Refused) > 0 && p.logf != nil {
|
||||
var held []string
|
||||
for _, m := range slices.Sorted(maps.Keys(now.Refused)) {
|
||||
held = append(held, m+" ("+now.Refused[m]+")")
|
||||
}
|
||||
p.logf("proposal %s: held as refused by the indexer's judge: %s", prop.ID, strings.Join(held, "; "))
|
||||
}
|
||||
nodes := map[string]bool{}
|
||||
for n := range current {
|
||||
nodes[n] = true
|
||||
}
|
||||
if records, err := r.Records(context.Background()); err == nil {
|
||||
for n := range records {
|
||||
nodes[n] = true
|
||||
}
|
||||
for n := range p.w.records() {
|
||||
nodes[n] = true
|
||||
}
|
||||
names := make([]string, 0, len(nodes))
|
||||
for n := range nodes {
|
||||
names = append(names, n)
|
||||
}
|
||||
fps := map[string]string{}
|
||||
for name, m := range p.w.trunk() {
|
||||
if m.FP2 != "" {
|
||||
fps[name] = m.FP2
|
||||
}
|
||||
}
|
||||
return judge.Intended(now, with, current, laid, names, fps)
|
||||
return judge.Intended(current, laid, slices.Sorted(maps.Keys(nodes)))
|
||||
}
|
||||
|
||||
// mustAnswer finds the nodes the proposal must hear from (§3), writes a reached candidate for each whose key it does not
|
||||
@@ -425,7 +391,7 @@ func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal
|
||||
return // the verbs write the candidates before the proposal: wait to hear them
|
||||
}
|
||||
}
|
||||
intended, must, notes := p.compute(prop, current, cands)
|
||||
intended, must := p.compute(prop, current, cands)
|
||||
p.intended[fmt.Sprint(prop.ID, prop.Base)] = intended
|
||||
for _, node := range must {
|
||||
if _, ok := cands[node]; ok {
|
||||
@@ -447,9 +413,6 @@ func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal
|
||||
}
|
||||
}
|
||||
prop.Must = must
|
||||
if len(notes) > 0 {
|
||||
prop.Words = strings.TrimPrefix(prop.Words+"; "+strings.Join(notes, "; "), "; ")
|
||||
}
|
||||
p.writeProposal(ctx, prop, rev)
|
||||
}
|
||||
|
||||
@@ -483,7 +446,7 @@ func (p *passer) lays(ctx context.Context) (func(string) bool, func(string) bool
|
||||
held[name] = true
|
||||
}
|
||||
}
|
||||
records, _ := worldReader{w: p.w}.Records(ctx)
|
||||
records := p.w.records()
|
||||
return func(m string) bool { return held[m] }, func(n string) bool { _, ok := records[n]; return ok }
|
||||
}
|
||||
|
||||
|
||||
@@ -367,9 +367,16 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
func wantsVerbs(t *testing.T) {
|
||||
fpOld, fpNew := delivery.SourcePrefix2+strings.Repeat("1", 64), delivery.SourcePrefix2+strings.Repeat("2", 64)
|
||||
st := memStore{}
|
||||
for _, state := range []string{modulesState, manifestsState, policiesState, haveState, putsState, "findings"} {
|
||||
for _, state := range []string{modulesState, manifestsState, policiesState, haveState, putsState, "findings", delivery.ComposedRead} {
|
||||
st[state] = &memKV{values: map[string][]byte{}}
|
||||
}
|
||||
// ADR 0593 §6, §8: what each node published, every node's bucket as one state: anchor's own composition and its web at
|
||||
// a fingerprint, and laptop's web refused there.
|
||||
webKey, _ := delivery.ComposedKey("anchor", "web", fpNew)
|
||||
refusedKey, _ := delivery.ComposedKey("laptop", "web", fpNew)
|
||||
st[delivery.ComposedRead].values["anchor.node"], _ = delivery.EncodeComposed(delivery.Composed{Node: "anchor"})
|
||||
st[delivery.ComposedRead].values[webKey], _ = delivery.EncodeComposed(delivery.Composed{Node: "anchor", Module: "web", FP: fpNew})
|
||||
st[delivery.ComposedRead].values[refusedKey], _ = delivery.EncodeComposed(delivery.Composed{Node: "laptop", Module: "web", FP: fpNew, Refused: "a required value is unset"})
|
||||
manifest := func(name, extra string) {
|
||||
raw := `{"module":"` + name + `","version":"1"` + extra + `}`
|
||||
digest := delivery.ManifestDigest([]byte(raw))
|
||||
@@ -409,6 +416,11 @@ func wantsVerbs(t *testing.T) {
|
||||
got["composed"].(map[string]string)["anchor"] != "sha256:"+strings.Repeat("c", 64) || len(got["composed"].(map[string]string)) != 1 {
|
||||
t.Errorf("compare = %+v", got)
|
||||
}
|
||||
pub, _ := out.(map[string]any)["published"].(map[string][]publishedRow)
|
||||
if a, l := pub["anchor"], pub["laptop"]; len(a) != 2 || a[0].Module != "node" || a[1].Module != "web" || a[1].FP != fpNew ||
|
||||
!strings.HasPrefix(a[1].Digest, "sha256:") || len(l) != 1 || l[0].Refused != "a required value is unset" || l[0].Digest != "" {
|
||||
t.Errorf("published = %+v (unread %v)", pub, out.(map[string]any)["published-unread"])
|
||||
}
|
||||
// Read by watch (issue 1953 L5f): the same answer, each watched state started once, and no get per key of the
|
||||
// large states (the modules, their manifests, the composed record).
|
||||
watches := map[string]int{}
|
||||
|
||||
@@ -20,7 +20,7 @@ import (
|
||||
// The seat mesh-wants' verbs (novox/hq ADR 0565 §7, §8, §9, §15; issue 1953 D4-D6), served through the tool runner beside the
|
||||
// put tools and mesh-placement's verbs. `upgrade` reads and sets a person's upgrade policy, kept in the state `policies`
|
||||
// (each set recorded as a hand act and said as policy-changed), and with compare lists the digest of the composition each node's
|
||||
// own have names (the controller writes no policy any more, so no module's policy differs); `bus` says what a bus upgrade would do. The assigner's process reads
|
||||
// own have names beside every composition each node published (ADR 0593 §8, composed.go); `bus` says what a bus upgrade would do. The assigner's process reads
|
||||
// `policies` when it judges a move and writes the finding policy-split while a split stands.
|
||||
|
||||
// The states the verbs read, as the runtime names them.
|
||||
@@ -158,6 +158,12 @@ func (s *wantsTools) doUpgrade(args map[string]any, by string) (any, error) {
|
||||
}
|
||||
}
|
||||
out := map[string]any{"composed": composed}
|
||||
// And what every node published (ADR 0593 §8), read only to compare.
|
||||
if values, err := s.values(delivery.ComposedRead); err != nil {
|
||||
out["published-unread"] = err.Error()
|
||||
} else {
|
||||
out["published"] = published(values)
|
||||
}
|
||||
// The assigner's findings, its said conditions as the condition keeper reads them (§15).
|
||||
if found, err := s.values(health.FindingsState); err != nil {
|
||||
out["findings-unread"] = err.Error()
|
||||
|
||||
@@ -3,20 +3,14 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
@@ -249,63 +243,11 @@ func (w *world) trunk() map[string]delivery.Module {
|
||||
return out
|
||||
}
|
||||
|
||||
// mesh is the mesh as the trunk holds it, for the composition check.
|
||||
func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh {
|
||||
m := assign.Mesh{Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, Digests: map[string]string{}}
|
||||
m.Assigned = w.assigned()
|
||||
for name, mod := range w.trunk() {
|
||||
if mod.Manifest == "" {
|
||||
continue
|
||||
}
|
||||
m.Digests[name] = mod.Manifest
|
||||
// The indexer's verdict on it, carried in its value (ADR 0578 §3, §5): a refused module is held and named as one
|
||||
// that cannot be read, and its verdict goes into the composer's input.
|
||||
if v := mod.VerdictOf(name); v != nil {
|
||||
if refused, _ := delivery.TakeManifestVerdict(v); refused {
|
||||
if m.Verdicts == nil {
|
||||
m.Verdicts = map[string]delivery.ManifestVerdict{}
|
||||
}
|
||||
m.Verdicts[name] = *v
|
||||
m.Unreadable[name] = v.Why()
|
||||
continue
|
||||
}
|
||||
}
|
||||
p := w.manifest(ctx, manifests, mod.Manifest)
|
||||
if p.err != "" {
|
||||
m.Unreadable[name] = p.err
|
||||
continue
|
||||
}
|
||||
m.Manifests[name] = p.manifest
|
||||
}
|
||||
// Once the catalogue places the nodes, the check composes over the input they compose from (ADR 0547 §2, ADR 0549
|
||||
// §3): one composer's input for the check and the nodes. An input that cannot be read is said, and the nodes and their
|
||||
// assignments alone are composed, as before any placement.
|
||||
if _, seeded := w.placed(); seeded {
|
||||
in, err := frombus.Input(ctx, worldReader{w: w, manifests: manifests}, frombus.Mesh{}, frombus.Scope{Every: true})
|
||||
if err != nil {
|
||||
m.InputUnread = err.Error()
|
||||
} else {
|
||||
m.Input = &in
|
||||
}
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// worldReader is what the assigner watched as frombus reads the bus: the seat's placement, the indexer's modules, each node's
|
||||
// record in its own have key, the manifests by digest; the check reads no config value, own secret or credential (the
|
||||
// composition it asks resolves, and stands in for what it does not carry).
|
||||
type worldReader struct {
|
||||
w *world
|
||||
manifests getter
|
||||
}
|
||||
|
||||
func (r worldReader) Placements(context.Context) (map[string]delivery.Placement, error) {
|
||||
return r.w.placements(), nil
|
||||
}
|
||||
|
||||
func (r worldReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
|
||||
// records is each node's record in its own have key (`<node>.node`): the nodes a placement proposal is answered by,
|
||||
// with those placement names (ADR 0593 §8).
|
||||
func (w *world) records() map[string]delivery.NodeRecord {
|
||||
out := map[string]delivery.NodeRecord{}
|
||||
for key, value := range r.w.read(delivery.HaveBucket) {
|
||||
for key, value := range w.read(delivery.HaveBucket) {
|
||||
node, _, isNode, err := delivery.ParseHaveKey(key)
|
||||
if err != nil || !isNode {
|
||||
continue
|
||||
@@ -314,136 +256,7 @@ func (r worldReader) Records(context.Context) (map[string]delivery.NodeRecord, e
|
||||
out[node] = *h.Record
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r worldReader) Modules(context.Context) (map[string]delivery.Module, error) {
|
||||
return r.w.trunk(), nil
|
||||
}
|
||||
|
||||
func (r worldReader) Manifest(ctx context.Context, digest string) ([]byte, error) {
|
||||
key, err := delivery.ManifestsKey(digest)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
raw, err := r.manifests(ctx, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m, err := delivery.DecodeManifest(raw)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(m.JSON), nil
|
||||
}
|
||||
|
||||
// Values is one module's config values on one node as mesh-config resolved them, each a value its source lays
|
||||
// (catalogue.LaidValue): the holder composes every node a proposal touches with them (ADR 0567 §9).
|
||||
func (r worldReader) Values(_ context.Context, node, module string) (map[string]any, error) {
|
||||
var out map[string]any
|
||||
prefix := node + "." + module + "."
|
||||
for key, raw := range r.w.read(buscompose.ResolvedBucket) {
|
||||
name, ok := strings.CutPrefix(key, prefix)
|
||||
if !ok || strings.Contains(name, ".") {
|
||||
continue
|
||||
}
|
||||
if v, laid := catalogue.LaidValue(raw); laid {
|
||||
if out == nil {
|
||||
out = map[string]any{}
|
||||
}
|
||||
out[name] = v
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (worldReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (worldReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
|
||||
return nil, nil, nil
|
||||
}
|
||||
|
||||
// Build and Archive are read only for a node composing itself (frombus.Scope without Every): the check of every node
|
||||
// composes from the manifests the indexer read, and reads no build and no archive.
|
||||
func (worldReader) Build(context.Context, string, string) (*delivery.BuilderBuild, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (worldReader) Archive(_ context.Context, url, _ string) ([]byte, error) {
|
||||
return nil, fmt.Errorf("the composition check reads no archive (%s)", url)
|
||||
}
|
||||
|
||||
func (r worldReader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) {
|
||||
all, err := r.Values(ctx, node, module)
|
||||
var out map[string]any
|
||||
for _, k := range keys {
|
||||
if v, ok := all[k]; ok {
|
||||
if out == nil {
|
||||
out = map[string]any{}
|
||||
}
|
||||
out[k] = v
|
||||
}
|
||||
}
|
||||
return out, err
|
||||
}
|
||||
|
||||
// change is a candidate's change for the composition check, or why the composition waits for it.
|
||||
func (w *world) change(ctx context.Context, manifests getter, c delivery.Candidate, trunk map[string]delivery.Module) (assign.Change, string) {
|
||||
ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{},
|
||||
Twice: map[string][]string{}, Digests: map[string]string{}, Members: map[string]string{}}
|
||||
heads := map[string]string{}
|
||||
for _, p := range c.Pulls {
|
||||
heads[fmt.Sprintf("%s#%d", p.Repository, p.Number)] = p.Head.Head
|
||||
}
|
||||
ch.Held = assign.HeldOf(w.read(keptBucket), heads)
|
||||
definedIn := map[string]map[string]bool{}
|
||||
for _, p := range c.Pulls {
|
||||
if p.Repository == assign.SDKRepository {
|
||||
ch.SDKHead = p.Head.Head
|
||||
}
|
||||
ch.Members[p.Repository] = p.Head.Head
|
||||
for _, name := range p.Modules {
|
||||
fp, ok := p.Fps[name]
|
||||
if !ok || fp.Manifest == "" {
|
||||
if ok && fp.None2 == noModuleWords {
|
||||
ch.Removed = append(ch.Removed, name)
|
||||
continue
|
||||
}
|
||||
return ch, fmt.Sprintf("the indexer holds no manifest of %s for %s#%d yet", name, p.Repository, p.Number)
|
||||
}
|
||||
if fp.FP2 != "" && fp.FP2 != trunk[name].FP2 {
|
||||
ch.Rebuilt++
|
||||
}
|
||||
got := w.manifest(ctx, manifests, fp.Manifest)
|
||||
if strings.HasPrefix(got.err, "the indexer holds no manifest") {
|
||||
return ch, got.err
|
||||
}
|
||||
ch.Digests[name] = fp.Manifest
|
||||
if got.err != "" {
|
||||
ch.Unreadable[name] = got.err
|
||||
continue
|
||||
}
|
||||
ch.Manifests[got.module] = got.manifest
|
||||
if definedIn[got.module] == nil {
|
||||
definedIn[got.module] = map[string]bool{}
|
||||
}
|
||||
definedIn[got.module][p.Repository] = true
|
||||
if t, ok := trunk[got.module]; ok && t.Repository != "" && t.Repository != p.Repository {
|
||||
definedIn[got.module][t.Repository] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
for name, repos := range definedIn {
|
||||
if len(repos) > 1 {
|
||||
for r := range repos {
|
||||
ch.Twice[name] = append(ch.Twice[name], r)
|
||||
}
|
||||
sort.Strings(ch.Twice[name])
|
||||
}
|
||||
}
|
||||
return ch, ""
|
||||
return out
|
||||
}
|
||||
|
||||
// online is, by node, whether the watchman holds the node online.
|
||||
@@ -594,40 +407,6 @@ func (w *world) changed(bucket, key string) uint64 {
|
||||
return w.changes[bucket+"|"+key]
|
||||
}
|
||||
|
||||
// logWatch follows the builder's log: handle each entry it keeps, then each new one (process.LogReader.Watch).
|
||||
type logWatch func(handle func(stdio.LogEntry) error) error
|
||||
|
||||
// putBuilt takes one entry of the builder's log in: a done trunk build's end (no ref) naming a full commit, newer than the
|
||||
// module's last, becomes the module's running commit. It answers whether the commit changed.
|
||||
func (w *world) putBuilt(raw []byte) bool {
|
||||
e, err := delivery.DecodeBuildLogEntry(raw)
|
||||
if err != nil || e.Event != delivery.BuildEndedEvent || e.Outcome != delivery.BuildDone || e.Ref != "" || len(e.Commit) != 40 {
|
||||
return false
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
have, ok := w.trunkBuilt[e.Module]
|
||||
if ok && !e.At.After(have.At) {
|
||||
return false
|
||||
}
|
||||
w.trunkBuilt[e.Module] = e
|
||||
return !ok || have.Commit != e.Commit
|
||||
}
|
||||
|
||||
// runningCommits is the commit of module's newest done trunk build in the builder's log (ADR 0571 clause 1): the commit a
|
||||
// reader of the manifests runs at (novox/hq ADR 0482 clause 5, issue 1623); none when the log holds no such build.
|
||||
func (w *world) runningCommits(module string) []string {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if e, ok := w.trunkBuilt[module]; ok {
|
||||
return []string{e.Commit}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// keptBucket is mesh-config's `kept`: the approved held values of open pull requests (ADR 0565 §5).
|
||||
var keptBucket = "mesh-config_" + delivery.MeshConfigKeptState
|
||||
|
||||
// ownComposed is a placed node's own composition, as the digest its own have names (`<node>.node` of the haves, Composed):
|
||||
// what rule 7 reads. A node whose have names none is not composed yet.
|
||||
func (w *world) ownComposed(node string) (delivery.ComposedDigest, bool) {
|
||||
|
||||
@@ -2,15 +2,12 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
|
||||
// fakeManifests is the indexer's manifests bucket in memory.
|
||||
@@ -45,76 +42,6 @@ func pull(repo string, number int, modules map[string]delivery.CandidateModule)
|
||||
return p
|
||||
}
|
||||
|
||||
// 1297/B3b: a candidate's change reads each module's manifest by its digest, counts the modules rebuilt, takes a
|
||||
// module with no manifest in the preview as removed, a mesh-sdk member's head, and a module the trunk holds from
|
||||
// another repository as defined twice.
|
||||
func TestTheChangeOfACandidate(t *testing.T) {
|
||||
theWorldCountsTheValuesAKeyTakes(t)
|
||||
f := fakeManifests{}
|
||||
web := f.add(t, "web", `{"module":"web","version":"2"}`)
|
||||
mail := f.add(t, "mail", `{"module":"mail","version":"1"}`)
|
||||
trunk := map[string]delivery.Module{"web": {FP2: fp('a'), Repository: "novox/web"}, "mail": {FP2: fp('c'), Repository: "novox/mail"}}
|
||||
c := delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", Pulls: []delivery.CandidatePull{
|
||||
pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: web}, "old": {None2: noModuleWords}}),
|
||||
pull("novox/other", 2, map[string]delivery.CandidateModule{"mail": {FP2: fp('c'), Manifest: mail}}),
|
||||
pull(assign.SDKRepository, 3, nil),
|
||||
}}
|
||||
ch, waits := newWorld().change(context.Background(), f.get, c, trunk)
|
||||
if waits != "" {
|
||||
t.Fatalf("the change waits: %s", waits)
|
||||
}
|
||||
if ch.Manifests["web"].Version != "2" || ch.Rebuilt != 1 || !slices.Equal(ch.Removed, []string{"old"}) || ch.SDKHead != strings.Repeat("3", 40) {
|
||||
t.Errorf("change = %+v", ch)
|
||||
}
|
||||
if !slices.Equal(ch.Twice["mail"], []string{"novox/mail", "novox/other"}) || len(ch.Twice) != 1 {
|
||||
t.Errorf("twice = %v, want mail in novox/mail and novox/other", ch.Twice)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B3b: a candidate whose module has no manifest digest yet, or whose manifest the indexer does not hold yet, waits,
|
||||
// saying for what; a manifest that does not decode is the change's, not a wait (one with a field the assigner does not
|
||||
// know is decoded with it set aside, novox/hq ADR 0578 §6).
|
||||
func TestAChangeWaitsForItsManifests(t *testing.T) {
|
||||
f := fakeManifests{}
|
||||
bad := f.add(t, "web", `{"module":"web","version":"2","requires":"x"}`)
|
||||
for _, c := range []delivery.CandidatePull{
|
||||
pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {None2: "no preview yet"}}),
|
||||
pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: strings.Repeat("9", 64)}}),
|
||||
} {
|
||||
_, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", Pulls: []delivery.CandidatePull{c}}, nil)
|
||||
if waits == "" {
|
||||
t.Errorf("a change without its manifest %+v does not wait", c.Fps)
|
||||
}
|
||||
}
|
||||
ch, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab",
|
||||
Pulls: []delivery.CandidatePull{pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: bad}})}}, nil)
|
||||
if waits != "" || ch.Unreadable["web"] == "" {
|
||||
t.Errorf("an unreadable manifest: waits %q, change %+v", waits, ch)
|
||||
}
|
||||
}
|
||||
|
||||
// A manifest of the change whose config value falls back to a predefined key (novox/hq ADR 0348 §7) is read, not
|
||||
// unreadable for the catalogue's predefined.json the assigner never holds: the merge gate judges the fallback in the
|
||||
// catalogue, and the indexer reads it the same way (catalogue.ParseManifestForInputs, issue 1751). A group whose
|
||||
// mesh-catalog member touched pacman and xterm failed its composition on this (group check 18df81380db3). The manifest is
|
||||
// xterm's module.json from mesh-catalog at 835bf22b.
|
||||
func TestAManifestFallingBackToAPredefinedKeyIsReadable(t *testing.T) {
|
||||
raw, err := os.ReadFile("testdata/catalogue-835bf22b-xterm.module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(string(raw), `"predefined"`) {
|
||||
t.Fatal("the fixture names no predefined key")
|
||||
}
|
||||
f := fakeManifests{}
|
||||
xterm := f.add(t, "xterm", string(raw))
|
||||
ch, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab",
|
||||
Pulls: []delivery.CandidatePull{pull("novox/mesh-catalog", 1, map[string]delivery.CandidateModule{"xterm": {FP2: fp('b'), Manifest: xterm}})}}, nil)
|
||||
if waits != "" || ch.Unreadable["xterm"] != "" || ch.Manifests["xterm"].Module != "xterm" {
|
||||
t.Errorf("xterm: waits %q, unreadable %q, want read", waits, ch.Unreadable["xterm"])
|
||||
}
|
||||
}
|
||||
|
||||
// Review round 1 (items 3 and 4): the world counts the values a key takes, so a have written again after an answer is seen.
|
||||
func theWorldCountsTheValuesAKeyTakes(t *testing.T) {
|
||||
w := newWorld()
|
||||
|
||||
@@ -9,7 +9,6 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun v0.0.0-00010101000000-000000000000
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
|
||||
github.com/google/go-tpm v0.9.8 // indirect
|
||||
github.com/klauspost/compress v1.20.0 // indirect
|
||||
@@ -22,4 +21,3 @@ require (
|
||||
golang.org/x/time v0.15.0 // indirect
|
||||
)
|
||||
|
||||
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun => ../mesh-delivery-builder/buildrun
|
||||
|
||||
@@ -33,9 +33,11 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
reads := slices.Clone(m.Reads)
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-build-go.steps", "mesh-config.kept", "mesh-config.resolved",
|
||||
"mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-provisioner.provisions", "node-engine.have"}
|
||||
// ADR 0593 §8: it composes nothing, so it reads no candidate, held or resolved value, compose step or build log; it reads
|
||||
// every node's published compositions to compare them.
|
||||
want := []string{"mesh-delivery-builder.builds", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed",
|
||||
"mesh-provisioner.provisions", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
t.Errorf("reads = %v, want exactly %v", reads, want)
|
||||
}
|
||||
@@ -43,7 +45,7 @@ func TestManifest(t *testing.T) {
|
||||
for _, s := range m.State {
|
||||
states[s.Name] = s.History
|
||||
}
|
||||
if want := map[string]int{"asks": 5, "compositions": 5, "differences": 64, "puts": 20, "wants": 20,
|
||||
if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20,
|
||||
"placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5}; !reflect.DeepEqual(states, want) {
|
||||
t.Errorf("state = %v, want %v", states, want)
|
||||
}
|
||||
@@ -53,10 +55,10 @@ func TestManifest(t *testing.T) {
|
||||
if m.SecretsOwner != "mesh-delivery-assigner" {
|
||||
t.Errorf("secrets-owner = %q", m.SecretsOwner)
|
||||
}
|
||||
// 1906/D3: it says its conditions on the operator channel; 1622: it puts its compose asks on the Go build seat, as the
|
||||
// checker puts its checks.
|
||||
if !slices.Equal(m.Uses, []string{"node-build-go", "operator-channel"}) {
|
||||
t.Errorf("uses = %v, want exactly node-build-go and operator-channel", m.Uses)
|
||||
// 1906/D3: it says its conditions on the operator channel; it asks the Go build seat nothing since the composition is
|
||||
// the node-engine's (ADR 0593 §7).
|
||||
if !slices.Equal(m.Uses, []string{"operator-channel"}) {
|
||||
t.Errorf("uses = %v, want exactly operator-channel", m.Uses)
|
||||
}
|
||||
// ADR 0546: it serves its own tools (put, release, puts) from its bundle through the tool runner (the artifact loads the
|
||||
// module), says each put's step as the event put, and calls the hand-act seat and the bus node's backup holder.
|
||||
|
||||
@@ -24,22 +24,17 @@
|
||||
}
|
||||
],
|
||||
"reads": [
|
||||
"mesh-build-go.steps",
|
||||
"mesh-config.kept",
|
||||
"mesh-config.resolved",
|
||||
"mesh-delivery-builder.build-log",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-teller.answers",
|
||||
"mesh-delivery-watchman.presence",
|
||||
"mesh-grants.grants",
|
||||
"mesh-host.composed",
|
||||
"mesh-provisioner.provisions",
|
||||
"node-engine.have"
|
||||
],
|
||||
"uses": [
|
||||
"node-build-go",
|
||||
"operator-channel"
|
||||
],
|
||||
"invokes": [
|
||||
@@ -63,7 +58,6 @@
|
||||
{ "name": "candidates", "history": 5 },
|
||||
{ "name": "pulses", "key-ttl": true },
|
||||
{ "name": "asks", "history": 5 },
|
||||
{ "name": "compositions", "history": 5 },
|
||||
{ "name": "differences", "history": 64 },
|
||||
{ "name": "puts", "history": 20 },
|
||||
{ "name": "wants", "history": 20 },
|
||||
|
||||
Reference in New Issue
Block a user