mesh-delivery-builder: a done key keeps each reference form a counted node's engine reads, the oldest in its single reference, and names its needs; what each counted node reads is asked through the node-engine seat's verb and kept in the state engine-reads (ADR 0618 §1, §2, §4, ADR 0624 §1, §3, issue 774)
A counted node is one the placement names; a node with no statement reads as the baseline's first forms, and no node record is read in its place. builder_engine_reads lists, asks again and drops. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
@@ -6,6 +6,7 @@ import (
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -42,6 +43,9 @@ type ToolchainGate struct {
|
||||
known bool // the placement is settled
|
||||
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
|
||||
left map[string]string // <module>/<key> -> why it was said left out, so each reason is said once
|
||||
// OnPlaced is told, outside the gate's lock, once the placement's history is read and after each revision of it
|
||||
// FollowPlacement hears from then on (the counted nodes, Nodes, may have changed). Nil tells nobody.
|
||||
OnPlaced func()
|
||||
}
|
||||
|
||||
// NewToolchainGate is a gate over the toolchains of modules, passing what stands on to forward and saying with logf what
|
||||
@@ -90,6 +94,19 @@ func (g *ToolchainGate) Placement(key string, value []byte, deleted bool, at tim
|
||||
return g.again(node, changed)
|
||||
}
|
||||
|
||||
// Nodes are the nodes the placement places any module on, sorted: the nodes the mesh counts (a node leaves the count
|
||||
// when the placement no longer names it; ADR 0624 §1). None until the placement is read.
|
||||
func (g *ToolchainGate) Nodes() []string {
|
||||
var out []string
|
||||
for node, modules := range g.placed.ByNode() {
|
||||
if len(modules) > 0 {
|
||||
out = append(out, node)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// Settled says the placement's history is read: from now on only the reports of placed nodes pass, once a node's
|
||||
// placement is held. It answers whether what it passed on changed.
|
||||
func (g *ToolchainGate) Settled() bool {
|
||||
@@ -203,6 +220,9 @@ func FollowPlacement(ctx context.Context, js process.BucketOpener, gate *Toolcha
|
||||
if gate.Settled() {
|
||||
changed(ctx)
|
||||
}
|
||||
if gate.OnPlaced != nil {
|
||||
gate.OnPlaced()
|
||||
}
|
||||
go func() {
|
||||
defer func() { _ = w.Stop() }()
|
||||
for {
|
||||
@@ -218,6 +238,9 @@ func FollowPlacement(ctx context.Context, js process.BucketOpener, gate *Toolcha
|
||||
if e != nil && put(e) {
|
||||
changed(ctx)
|
||||
}
|
||||
if e != nil && gate.OnPlaced != nil {
|
||||
gate.OnPlaced()
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -18,9 +18,14 @@ import (
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
|
||||
|
||||
"mesh-delivery-builder/internal/build"
|
||||
)
|
||||
|
||||
// EngineReadsTool lists, asks again and drops what each counted node's engine reads, as the builder keeps it.
|
||||
const EngineReadsTool = "builder_engine_reads"
|
||||
|
||||
// queueAsker writes a queues value and waits for the builder's answer to it.
|
||||
type queueAsker func(key string, value []byte) (json.RawMessage, error)
|
||||
|
||||
@@ -264,7 +269,11 @@ func main() {
|
||||
a, err := build.DecodeReAsk(v)
|
||||
return err == nil && !a.Answered.IsZero()
|
||||
})
|
||||
if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now)); err != nil {
|
||||
tools := Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now)
|
||||
// What each counted node's engine reads, as the builder keeps it (ADR 0618 §2, ADR 0624 §1): listed, one node asked
|
||||
// again, or one node's key dropped.
|
||||
tools = append(tools, enginereads.Tool(EngineReadsTool, "the builder", enginereads.KeptState(stdio.State("engine-reads")), true, time.Now))
|
||||
if err := stdio.Serve("", tools); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
@@ -439,6 +439,10 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
|
||||
if v.Toolchains["go"].Version != "go1.27.1" || v.Toolchains["typescript"].Version != "tsc5" {
|
||||
t.Errorf("toolchains = %+v", v.Toolchains)
|
||||
}
|
||||
// ADR 0618 §1: the done key names the reference forms it is written in.
|
||||
if len(v.Needs) != 1 || v.Needs[0] != "form:artifact-store" {
|
||||
t.Errorf("needs = %v, want the first form", v.Needs)
|
||||
}
|
||||
// 1955/L7 (ADR 0571 §1): the build's start and end are in the module log build-log under the module's key, read as a
|
||||
// reader with the log's read reads them, the end naming the manifest and what was made.
|
||||
ended := b.logged(t, "netcheck")
|
||||
|
||||
@@ -126,7 +126,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision
|
||||
// is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it
|
||||
// cannot be read the trunk's keys are built and the candidates wait (issue 1945).
|
||||
errs := make(chan error, 6)
|
||||
errs := make(chan error, 7)
|
||||
trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: b.TrunkMoved, errs: errs, logf: logger.Printf}
|
||||
trunks.start()
|
||||
b.Trunk = trunks.head
|
||||
@@ -165,6 +165,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// placement is settled first, so the toolchains settle through the gate already knowing it.
|
||||
gate := buildrun.NewToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
|
||||
watch.put = gate.Toolchain
|
||||
// What each counted node's engine reads (ADR 0618 §2, ADR 0624 §1): asked through the node-engine seat's verb as the
|
||||
// builder's own account, for every node the placement names, and kept in the state engine-reads once it is bound.
|
||||
reads := newEngineReads(calls.CallOn, gate, logger.Printf)
|
||||
gate.OnPlaced = reads.placed
|
||||
b.Forms = reads.forms
|
||||
placementErrs, err := buildrun.FollowPlacement(ctx, js, gate, b.NodesChanged, logger.Printf)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -196,7 +201,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v",
|
||||
cred.Node, cred.Module, eventsStream, delivery.ArtifactsGoneEvent, err)
|
||||
default:
|
||||
cc, err := process.HearEvents(cons, func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) }, logger.Printf)
|
||||
cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
|
||||
if reads.heard(m, logger.Printf) { // the node-engine seat's manifest-reads-changed (ADR 0618 §4)
|
||||
return
|
||||
}
|
||||
heardGone(ctx, b, m, logger.Printf)
|
||||
}, logger.Printf)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -254,6 +264,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
go func() { errs <- follow(ctx, qw, eachRead) }()
|
||||
}
|
||||
})
|
||||
// The state engine-reads, new with ADR 0618 (issue 1790's kind): until it is on the bus each statement is kept in memory
|
||||
// only, and builder_engine_reads answers nothing.
|
||||
process.AwaitBucket(ctx, js, conn, engineReadsBucket, "what each node's engine reads is kept in memory only", logger.Printf).Ready(func(kv jetstream.KeyValue) {
|
||||
reads.bind(ctx, kv, errs, logger.Printf)
|
||||
})
|
||||
logger.Printf("started on %s", cred.Node)
|
||||
go func() { errs <- follow(ctx, w, each) }()
|
||||
select {
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"slices"
|
||||
"sync"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
|
||||
)
|
||||
|
||||
// EngineReadsState is the builder's state of what each counted node's engine states it reads (novox/hq ADR 0618 §2, ADR
|
||||
// 0624 §1): one key `<node>`, written from the node-engine seat's verb manifest-reads at the builder's start, on the
|
||||
// seat's event manifest-reads-changed for that node, and when a person asks again (builder_engine_reads); a node that
|
||||
// is away keeps its last statement, and its key is removed when the placement no longer names the node. A counted node
|
||||
// with no key reads as stating the baseline: its first forms. No node's record is read in its place.
|
||||
const EngineReadsState = "engine-reads"
|
||||
|
||||
// engineReadsBucket is that state's bucket.
|
||||
const engineReadsBucket = delivery.BuilderModule + "_" + EngineReadsState
|
||||
|
||||
// engineReads keeps the statements and the counted nodes for the builder's forms (build.Builder.Forms).
|
||||
type engineReads struct {
|
||||
keeper *enginereads.Keeper
|
||||
gate *buildrun.ToolchainGate
|
||||
mu sync.Mutex
|
||||
counted []string
|
||||
}
|
||||
|
||||
// newEngineReads asks through call (process.Conn.CallOn) and keeps each answer in the state once it is bound.
|
||||
func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, logf func(string, ...any)) *engineReads {
|
||||
return &engineReads{gate: gate, keeper: &enginereads.Keeper{Call: call, Logf: logf}}
|
||||
}
|
||||
|
||||
// bind keeps the statements in kv and hears it: every node's last statement at the start, a person's ask again and drop
|
||||
// later. Bound once the state is on the bus (process.AwaitBucket).
|
||||
func (r *engineReads) bind(ctx context.Context, kv jetstream.KeyValue, errs chan<- error, logf func(string, ...any)) {
|
||||
r.keeper.Put = func(node string, v []byte) error { _, err := kv.Put(ctx, node, v); return err }
|
||||
r.keeper.Delete = func(node string) error { return kv.Delete(ctx, node) }
|
||||
each := func(e jetstream.KeyValueEntry) {
|
||||
r.keeper.Heard(e.Key(), e.Value(), e.Operation() != jetstream.KeyValuePut)
|
||||
}
|
||||
w, err := settle(ctx, kv, each)
|
||||
if err != nil {
|
||||
process.NotWatched(logf, engineReadsBucket, err, "builder_engine_reads is not answered until the builder starts again")
|
||||
return
|
||||
}
|
||||
go func() { errs <- follow(ctx, w, each) }()
|
||||
// Every counted node asked again, so what was asked before the state was bound is kept in it too.
|
||||
r.mu.Lock()
|
||||
nodes := slices.Clone(r.counted)
|
||||
r.mu.Unlock()
|
||||
for _, node := range nodes {
|
||||
r.keeper.AskNode(node, "the state engine-reads is bound")
|
||||
}
|
||||
}
|
||||
|
||||
// placed takes the placement in: a node it names newly is asked, one it no longer names is forgotten, key and all. The
|
||||
// first call asks every counted node (the builder's start).
|
||||
func (r *engineReads) placed() {
|
||||
nodes := r.gate.Nodes()
|
||||
r.mu.Lock()
|
||||
before, first := r.counted, r.counted == nil
|
||||
r.counted = nodes
|
||||
r.mu.Unlock()
|
||||
for _, node := range nodes {
|
||||
if first || !slices.Contains(before, node) {
|
||||
r.keeper.AskNode(node, "counted: the placement names it")
|
||||
}
|
||||
}
|
||||
for _, node := range before {
|
||||
if !slices.Contains(nodes, node) {
|
||||
r.keeper.Forget(node)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// heard takes an event of the builder's consumer on EVENTS: manifest-reads-changed asks its node again, and answers true.
|
||||
func (r *engineReads) heard(m jetstream.Msg, logf func(string, ...any)) bool {
|
||||
node, ours, err := enginereads.NodeOf(m.Subject(), m.Data())
|
||||
if !ours {
|
||||
return false
|
||||
}
|
||||
if ackErr := m.Ack(); ackErr != nil {
|
||||
logf("acknowledging %s: %v", m.Subject(), ackErr)
|
||||
}
|
||||
if err != nil {
|
||||
logf("%s holds what is not %s, and %s is asked again all the same: %v", m.Subject(), enginereads.Event, node, err)
|
||||
}
|
||||
r.keeper.AskNode(node, enginereads.Event)
|
||||
return true
|
||||
}
|
||||
|
||||
// forms is build.Builder.Forms: the counted nodes' statements, and those of the nodes whose engine answered.
|
||||
func (r *engineReads) forms() (present, counted map[string][]string) {
|
||||
r.mu.Lock()
|
||||
nodes := slices.Clone(r.counted)
|
||||
r.mu.Unlock()
|
||||
return r.keeper.Present(nodes), r.keeper.Reads(nodes)
|
||||
}
|
||||
@@ -3,16 +3,23 @@ package main
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
@@ -45,6 +52,100 @@ func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) {
|
||||
if src.opened {
|
||||
t.Error("a refused source counts as opened, so it is never opened again")
|
||||
}
|
||||
t.Run("ADR 0624: the counted nodes' engine reads", theCountedNodesEngineReads)
|
||||
}
|
||||
|
||||
// theCountedNodesEngineReads: the builder asks the node-engine seat's verb of every node the placement names; a node away
|
||||
// and never asked has no statement and reads as the baseline's first forms, with no node record read in its place; an
|
||||
// away node keeps its last statement; a node the placement no longer names leaves the count, its key removed; and
|
||||
// builder_engine_reads drops one node's key.
|
||||
func theCountedNodesEngineReads(t *testing.T) {
|
||||
gate := buildrun.NewToolchainGate(nil, func(string, string, []byte, bool) bool { return false }, t.Logf)
|
||||
place := func(nodes ...string) {
|
||||
for _, node := range []string{"n1", "n2"} {
|
||||
pl := delivery.Placement{Node: node}
|
||||
if slices.Contains(nodes, node) {
|
||||
pl.Modules = []delivery.PlacedModule{{Module: "web"}}
|
||||
}
|
||||
values, err := delivery.PlacementValues(pl)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for key, raw := range values {
|
||||
gate.Placement(key, raw, false, time.Now())
|
||||
}
|
||||
if !slices.Contains(nodes, node) {
|
||||
key, _ := delivery.PlacementModuleKey(node, "web")
|
||||
gate.Placement(key, nil, true, time.Now())
|
||||
}
|
||||
}
|
||||
}
|
||||
away := map[string]bool{"n2": true}
|
||||
asked := map[string]int{}
|
||||
r := newEngineReads(func(node, key string, _ any) (json.RawMessage, error) {
|
||||
asked[node]++
|
||||
if away[node] {
|
||||
return nil, nats.ErrNoResponders
|
||||
}
|
||||
return delivery.EncodeManifestReadsAnswer(delivery.ManifestReadsAnswer{Node: node, ManifestReads: []string{"form:new"}, StatedAt: time.Now()})
|
||||
}, gate, t.Logf)
|
||||
r.keeper.Sync = true
|
||||
kept := map[string][]byte{}
|
||||
r.keeper.Put = func(node string, v []byte) error { kept[node] = v; return nil }
|
||||
r.keeper.Delete = func(node string) error { delete(kept, node); return nil }
|
||||
place("n1", "n2")
|
||||
r.placed()
|
||||
present, counted := r.forms()
|
||||
if !slices.Equal(present["n1"], []string{"form:new"}) || len(present) != 1 || len(counted) != 2 || counted["n2"] != nil {
|
||||
t.Errorf("away and never asked: present %v, counted %v", present, counted)
|
||||
}
|
||||
if s := r.keeper.Statements()["n2"]; s.From != enginereads.FromNone || s.Answered || asked["n2"] != 1 {
|
||||
t.Errorf("n2, away and never asked, is %+v (asked %d)", s, asked["n2"])
|
||||
}
|
||||
// n2 back, stating {word:module}, then away again: it keeps that statement.
|
||||
away["n2"] = false
|
||||
r.keeper.Heard("n2", mustStatement(t, enginereads.Statement{Node: "n2", From: enginereads.FromVerb, Answered: true,
|
||||
AskAgain: &enginereads.Again{By: "n1.ada", Why: "back"}}), false)
|
||||
away["n2"] = true
|
||||
r.keeper.AskNode("n2", "its event")
|
||||
if _, counted = r.forms(); !slices.Equal(counted["n2"], []string{"form:new"}) || r.keeper.Statements()["n2"].Answered {
|
||||
t.Errorf("away again, n2 reads %v (%+v)", counted["n2"], r.keeper.Statements()["n2"])
|
||||
}
|
||||
// n2 retired: it leaves the placement, and its key goes.
|
||||
place("n1")
|
||||
r.placed()
|
||||
if _, counted = r.forms(); len(counted) != 1 || kept["n2"] != nil {
|
||||
t.Errorf("retired n2: counted %v, its key %s", counted, kept["n2"])
|
||||
}
|
||||
// builder_engine_reads drops one key.
|
||||
st := &memReads{v: map[string][]byte{"n1": kept["n1"]}}
|
||||
tool := enginereads.Tool("builder_engine_reads", "the builder", st, true, time.Now)
|
||||
if _, err := tool.Run(map[string]any{"node": "n1", "drop": "true", "why": "a test"}); err != nil || st.v["n1"] != nil {
|
||||
t.Errorf("drop: %v, the key %s", err, st.v["n1"])
|
||||
}
|
||||
}
|
||||
|
||||
func mustStatement(t *testing.T, s enginereads.Statement) []byte {
|
||||
t.Helper()
|
||||
raw, err := enginereads.Encode(s)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
// memReads is the builder's state of engine reads in memory, as the tool reads it.
|
||||
type memReads struct{ v map[string][]byte }
|
||||
|
||||
func (m *memReads) Keys() ([]string, error) { return slices.Collect(maps.Keys(m.v)), nil }
|
||||
func (m *memReads) Get(key string) ([]byte, uint64, bool, error) {
|
||||
v, ok := m.v[key]
|
||||
return v, 1, ok, nil
|
||||
}
|
||||
func (m *memReads) Put(key string, value []byte) (uint64, error) { m.v[key] = value; return 2, nil }
|
||||
func (m *memReads) Delete(key string) error { delete(m.v, key); return nil }
|
||||
func (m *memReads) Await(string, uint64, func([]byte) bool, time.Duration) ([]byte, error) {
|
||||
return nil, errors.New("no part answers here")
|
||||
}
|
||||
|
||||
// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err.
|
||||
|
||||
@@ -25,3 +25,9 @@ require (
|
||||
)
|
||||
|
||||
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun => ./buildrun
|
||||
|
||||
replace git.novox.be/novox/mesh-sdk/go => ../../../0618-sdk/go
|
||||
|
||||
require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000
|
||||
|
||||
replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk
|
||||
|
||||
@@ -1,7 +1,5 @@
|
||||
git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep v0.0.0-20261102193506-d221c633799a h1:oxgksCiq272/afXbUION/dxtAS5MPGnmyq6+AZQxmXg=
|
||||
git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep v0.0.0-20261102193506-d221c633799a/go.mod h1:kpsX4Pgss5EeuBvel1zWa2ogj7MkR3jZso6oa/J0YUQ=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 h1:XEbIVImp6yGjA91rp0iZBoDdL5KfamjOuDLwWvqh0hc=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8/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/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/forms"
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
@@ -79,6 +80,11 @@ type Builder struct {
|
||||
// time.Now. Node is the node the builder runs on, which asked the build.
|
||||
Now func() time.Time
|
||||
Node string
|
||||
// Forms answers what each node's engine states it reads (ADR 0618 §2, ADR 0624 §1): present, the nodes whose engine
|
||||
// answered when last asked, and counted, every node the placement names, present or away, a node with no statement
|
||||
// stating nothing (the baseline's first forms). A done key names each artifact in the forms these keep, the oldest in
|
||||
// its single reference, and its needs. Nil: no node states anything, and the first forms are kept.
|
||||
Forms func() (present, counted map[string][]string)
|
||||
|
||||
once sync.Once
|
||||
sem chan struct{}
|
||||
@@ -606,6 +612,10 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p
|
||||
return failed("%s", keyError(errs))
|
||||
}
|
||||
var artifacts []delivery.BuiltArtifact
|
||||
var present, counted map[string][]string
|
||||
if b.Forms != nil {
|
||||
present, counted = b.Forms()
|
||||
}
|
||||
keptModule := map[string]bool{}
|
||||
for _, bm := range kept {
|
||||
keptModule[bm] = true
|
||||
@@ -634,6 +644,7 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p
|
||||
return failed("%s", err)
|
||||
}
|
||||
built := delivery.BuiltArtifact{Name: t.Artifact, Kind: t.Kind, Digest: d}
|
||||
var answered []string
|
||||
for _, l := range lanes {
|
||||
if l.module != t.Module {
|
||||
continue
|
||||
@@ -645,10 +656,11 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p
|
||||
return failed("%s%s on %s: the build answered no reference, so its push to the artifact store was not answered", delivery.NotStored, t.Artifact, l.node)
|
||||
}
|
||||
if r != "" {
|
||||
built.Reference = r
|
||||
answered = append(answered, r)
|
||||
}
|
||||
}
|
||||
artifacts = append(artifacts, built)
|
||||
// The forms a counted node's engine reads are kept, the oldest in the single reference (ADR 0624 §1, §3).
|
||||
artifacts = append(artifacts, forms.Current.Choose(built, answered, present, counted))
|
||||
}
|
||||
var steps []string
|
||||
for _, l := range lanes {
|
||||
@@ -670,7 +682,7 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p
|
||||
used = nil
|
||||
}
|
||||
done := delivery.BuilderBuild{State: delivery.BuildDone, By: b.Claimer.By, Manifest: m.Manifest, Step: step, Artifacts: artifacts, Toolchains: used,
|
||||
Imports: importsOf(lanes)}
|
||||
Imports: importsOf(lanes), Needs: forms.Current.Needs(artifacts)} // the forms it is written in (ADR 0618 §1)
|
||||
if _, err := delivery.EncodeBuilderBuild(done); err != nil {
|
||||
// A reference the build module answered that does not name the artifact and its digest is no reference.
|
||||
return failed("%s%v", delivery.NotStored, err)
|
||||
@@ -765,7 +777,7 @@ func (n *askNotes) note(ctx context.Context, language, id, step string) error {
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
v := n.cl.Build
|
||||
v.State, v.By, v.Step, v.Error, v.Artifacts = delivery.BuildBuilding, n.b.Claimer.By, step, "", nil
|
||||
v.State, v.By, v.Step, v.Error, v.Artifacts, v.Needs = delivery.BuildBuilding, n.b.Claimer.By, step, "", nil, nil
|
||||
v.Asks = maps.Clone(v.Asks)
|
||||
if v.Asks == nil {
|
||||
v.Asks = map[string]string{}
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -14,6 +15,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/forms"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
@@ -222,6 +224,7 @@ var (
|
||||
|
||||
func TestABundleAndAnArchiveAreBuiltOnOneNodeAndDone(t *testing.T) { // 1607/AC7, ADR 0478 clauses 1, 4
|
||||
t.Run("the imports a build answers reach the done key and the log", importsABuildAnswersReachTheDoneKeyAndTheLog)
|
||||
t.Run("ADR 0624: a done key keeps each form a counted node reads", aDoneKeyKeepsEachFormACountedNodeReads)
|
||||
r := newRig(t, goBundle, archiveA, tsBundle)
|
||||
// ADR 0584: a third node agrees with novox and ace's typescript is the odd one; the build is done and its step says ace sits out.
|
||||
r.tools.Put(buildrun.NodeBuildModule, "g14.typescript", tcValue(t, "typescript", "tsc5"), false)
|
||||
@@ -278,6 +281,11 @@ func TestABundleAndAnArchiveAreBuiltOnOneNodeAndDone(t *testing.T) { // 1607/AC7
|
||||
if v.Artifacts[0].Kind != "archive" || v.Artifacts[1].Kind != "bundle" {
|
||||
t.Errorf("kinds = %+v", v.Artifacts)
|
||||
}
|
||||
// ADR 0618 §1: the done key names the forms its references are written in, here the first form alone, and keeps no list
|
||||
// beside the single reference while no counted node needs another form.
|
||||
if !slices.Equal(v.Needs, []string{"form:artifact-store"}) || v.Artifacts[0].References != nil {
|
||||
t.Errorf("needs = %v, references = %v", v.Needs, v.Artifacts[0].References)
|
||||
}
|
||||
if got := v.Toolchains["go"]; got.Version != "go1.27.1" || got.Recipe != "1" {
|
||||
t.Errorf("toolchains = %+v, want go and typescript named", v.Toolchains)
|
||||
}
|
||||
@@ -1234,3 +1242,71 @@ type silentWatch struct{}
|
||||
func (silentWatch) WatchAsk(context.Context, string, string) (<-chan delivery.Step, error) {
|
||||
return make(chan delivery.Step), nil
|
||||
}
|
||||
|
||||
// aDoneKeyKeepsEachFormACountedNodeReads holds the builder's rows of ADR 0624's check: every build keeps each form a
|
||||
// counted node's engine reads, present or away, a node stating nothing reading the baseline's first form; the oldest kept
|
||||
// form is the single reference; a retired node leaves the count. old is the first form, new a fixture second form read
|
||||
// only by an engine stating form:new.
|
||||
func aDoneKeyKeepsEachFormACountedNodeReads(t *testing.T) {
|
||||
fs := forms.ArtifactForms{
|
||||
{Name: "old", First: true, Kinds: []string{"bundle"}, Is: func(r string) bool { return strings.HasPrefix(r, "artifact-store://") }},
|
||||
{Name: "new", Kinds: []string{"bundle"}, Is: func(r string) bool { return strings.HasPrefix(r, "10.0.0.1:5000/") }},
|
||||
}
|
||||
both, nothing, onlyNew := []string{"form:new", "word:module"}, []string(nil), []string{"form:new"}
|
||||
oldRef, newRef := "artifact-store://m/code/blobs/"+digest, "10.0.0.1:5000/m/code@"+digest
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
present, counted map[string][]string
|
||||
kept []string
|
||||
reference string
|
||||
references []string
|
||||
}{
|
||||
{"A present {old,new}, B away {old}: both forms, the old in the single reference",
|
||||
map[string][]string{"a": both}, map[string][]string{"a": both, "b": {"word:module"}}, []string{"old", "new"}, oldRef, []string{oldRef, newRef}},
|
||||
{"A present {new}, B away stating nothing: the baseline's form stays",
|
||||
map[string][]string{"a": onlyNew}, map[string][]string{"a": onlyNew, "b": nothing}, []string{"old", "new"}, oldRef, []string{oldRef, newRef}},
|
||||
{"A present {old,new}, B away and never asked (no key): the baseline's form stays",
|
||||
map[string][]string{"a": both}, map[string][]string{"a": both, "b": nil}, []string{"old", "new"}, oldRef, []string{oldRef, newRef}},
|
||||
{"B retired: the new form only", map[string][]string{"a": both}, map[string][]string{"a": both}, []string{"new"}, newRef, nil},
|
||||
{"B present stating nothing: the new form is not written",
|
||||
map[string][]string{"a": both, "b": nothing}, map[string][]string{"a": both, "b": nothing}, []string{"old"}, oldRef, nil},
|
||||
{"no node present: the first form", nil, map[string][]string{"a": both}, []string{"old"}, oldRef, nil},
|
||||
} {
|
||||
if got := fs.Kept("bundle", row.present, row.counted); !slices.Equal(got, row.kept) {
|
||||
t.Errorf("%s: kept %v, want %v", row.name, got, row.kept)
|
||||
}
|
||||
a := fs.Choose(delivery.BuiltArtifact{Name: "code", Kind: "bundle", Digest: digest}, []string{newRef, oldRef}, row.present, row.counted)
|
||||
if a.Reference != row.reference || !slices.Equal(a.References, row.references) {
|
||||
t.Errorf("%s: %q %v, want %q %v", row.name, a.Reference, a.References, row.reference, row.references)
|
||||
}
|
||||
// An engine that predates the list reads the single reference, in a form every counted node reads (ADR 0624 §3).
|
||||
if form, _ := fs.FormOf("bundle", a.Reference); form != row.kept[0] {
|
||||
t.Errorf("%s: the single reference is in %q, want the oldest kept %q", row.name, form, row.kept[0])
|
||||
}
|
||||
}
|
||||
// An answered reference no kept form names still names where the artifact is (ADR 0498 clause 1).
|
||||
if a := fs.Choose(delivery.BuiltArtifact{Name: "code", Kind: "bundle", Digest: digest}, []string{oldRef}, map[string][]string{"a": both},
|
||||
map[string][]string{"a": both}); a.Reference != oldRef {
|
||||
t.Errorf("a lone answered reference was dropped: %+v", a)
|
||||
}
|
||||
withBoth := []delivery.BuiltArtifact{{Name: "code", Kind: "bundle", Digest: digest, Reference: oldRef, References: []string{oldRef, newRef}}}
|
||||
if got := fs.Needs(withBoth); !slices.Equal(got, []string{"form:new", "form:old"}) {
|
||||
t.Errorf("needs %v", got)
|
||||
}
|
||||
// What the assigner holds a want by (ADR 0624 §2), from the same forms.
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
needs []string
|
||||
stated []string
|
||||
held bool
|
||||
}{
|
||||
{"a node stating {old}, a build carrying {old,new}: moves", []string{"form:new", "form:old"}, []string{"word:module"}, false},
|
||||
{"a node stating nothing, a build carrying the baseline's form: moves", []string{"form:old"}, nil, false},
|
||||
{"a build carrying {new} only, the node stating {old}: held", []string{"form:new"}, []string{"word:module"}, true},
|
||||
{"a build naming no form: moves", nil, nil, false},
|
||||
} {
|
||||
if got := fs.Held(row.needs, row.stated); got != row.held {
|
||||
t.Errorf("%s: held %v", row.name, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -93,7 +93,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
if m.Module != "mesh-delivery-builder" {
|
||||
t.Errorf("module = %q", m.Module)
|
||||
}
|
||||
if want := []state{{Name: "builds", History: 5}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
|
||||
if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
|
||||
@@ -108,13 +108,14 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
// 1937/AC3: the builder asks the controller whether the artifact store still holds a done key's blobs (held), and no
|
||||
// other tool; it holds no capacity.
|
||||
// 1955/L7 (ADR 0571 §2): the seat mesh-artifact-sweep's held, the controller's `held` is no longer asked (issue 1955).
|
||||
if want := []string{"seat:mesh-artifact-sweep.held"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
|
||||
if want := []string{"seat:mesh-artifact-sweep.held", "seat:node-engine.manifest-reads"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
|
||||
t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want)
|
||||
}
|
||||
// The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it
|
||||
// starts; and by the seat mesh-artifact-sweep's artifacts-gone, a blob the sweep found gone (1937/AC4, 1955/L7), the
|
||||
// controller's no longer (1126/B9, ADR 0590).
|
||||
if want := []string{"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
if want := []string{"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported",
|
||||
"node-engine.manifest-reads-changed"}; !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" {
|
||||
@@ -124,7 +125,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
t.Errorf("secrets-owner = %q", m.SecretsOwner)
|
||||
}
|
||||
// 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again.
|
||||
if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_seat_queue", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) {
|
||||
if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_engine_reads", "builder_seat_queue", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) {
|
||||
t.Errorf("tools = %v, want %v", m.Tools, want)
|
||||
}
|
||||
if want := []string{"build.asked-again"}; !reflect.DeepEqual(m.Emits, want) {
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
"builder_ask_again",
|
||||
"builder_build_again",
|
||||
"builder_builds_since",
|
||||
"builder_engine_reads",
|
||||
"builder_seat_queue",
|
||||
"builder_stuck"
|
||||
],
|
||||
@@ -16,13 +17,16 @@
|
||||
"consumes": [
|
||||
"mesh-artifact-sweep.artifacts-gone",
|
||||
"mesh-build-go.toolchain.reported",
|
||||
"mesh-build-node.toolchain.reported"
|
||||
"mesh-build-node.toolchain.reported",
|
||||
"node-engine.manifest-reads-changed"
|
||||
],
|
||||
"invokes": [
|
||||
"seat:mesh-artifact-sweep.held"
|
||||
"seat:mesh-artifact-sweep.held",
|
||||
"seat:node-engine.manifest-reads"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "builds", "history": 5 },
|
||||
{ "name": "engine-reads", "history": 1 },
|
||||
{ "name": "hearing", "history": 1 },
|
||||
{ "name": "instances", "history": 1 },
|
||||
{ "name": "queues", "history": 1 },
|
||||
|
||||
Reference in New Issue
Block a user