226 lines
8.8 KiB
Go
226 lines
8.8 KiB
Go
package buildrun
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"maps"
|
|
"slices"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/process"
|
|
)
|
|
|
|
// A build module writes <node>.<language> of its toolchains state on every node it runs on, and nothing deletes the key
|
|
// when the module leaves a node: no part writes another module's state, and the module is gone. A report of a node the
|
|
// build module no longer runs on is therefore no builder's (issue 1940: mesh-build-node taken off ace on 2026-10-23 left
|
|
// ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree").
|
|
//
|
|
// ToolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a
|
|
// ToolchainSet): an entry is passed on only while the placement (the seat mesh-placement's state `placement`, held by
|
|
// the assigner, delivery.PlacementBucket, read by delivery.Placed; ADR 0567, ADR 0549 §3) places its build module on
|
|
// its node, and only once it was written since that placement; it is passed on as deleted otherwise, said once, and
|
|
// passed on when its node writes it again. A report written before the placement is one an earlier run of the module
|
|
// left (issue 1940, 2026-10-25 20:56: mesh-build-node put on ace again brought ace.typescript v26.12.0 of 2026-10-23
|
|
// back into the agreement before the module ran there). Until the placement is settled, and while no node's placement
|
|
// is held (before the assigner's adopt writes the first one, or where the state cannot be read), every entry is passed
|
|
// on. Builders that run and report different versions still disagree: that refusal is the guard of novox/hq ADR 0465,
|
|
// and it stays.
|
|
//
|
|
// This replaces the copies of the gate the indexer, the builder and the checker each kept over the controller's
|
|
// assignments (issue 1940's subtask, issue 774: those go with the controller's lease, ADR 0549 §6).
|
|
type ToolchainGate struct {
|
|
mu sync.Mutex
|
|
modules map[string]bool // the build modules whose toolchains pass the gate
|
|
forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds
|
|
logf func(format string, a ...any)
|
|
placed *delivery.Placed
|
|
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
|
|
}
|
|
|
|
// NewToolchainGate is a gate over the toolchains of modules, passing what stands on to forward and saying with logf what
|
|
// it leaves out.
|
|
func NewToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *ToolchainGate {
|
|
g := &ToolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, placed: delivery.NewPlaced(),
|
|
raw: map[string]map[string][]byte{}, left: map[string]string{}}
|
|
for _, m := range modules {
|
|
g.modules[m] = true
|
|
}
|
|
return g
|
|
}
|
|
|
|
// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed.
|
|
func (g *ToolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool {
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
if g.raw[module] == nil {
|
|
g.raw[module] = map[string][]byte{}
|
|
}
|
|
if deleted {
|
|
delete(g.raw[module], key)
|
|
} else {
|
|
g.raw[module][key] = value
|
|
}
|
|
return g.pass(module, key)
|
|
}
|
|
|
|
// Placement takes one revision of the seat mesh-placement's placement in (at is when the bus wrote it) and answers whether what it
|
|
// passed on changed. A value that does not read is said, and places nothing on its node.
|
|
func (g *ToolchainGate) Placement(key string, value []byte, deleted bool, at time.Time) bool {
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
was := g.placed.Seeded()
|
|
node, changed, err := g.placed.Put(key, value, deleted, at)
|
|
if err != nil {
|
|
g.logf("%s: %v (issue 1940)", delivery.PlacementBucket, err)
|
|
}
|
|
if !g.known {
|
|
return false
|
|
}
|
|
if was != g.placed.Seeded() {
|
|
// The first placement came, or the last went: every report's standing changed.
|
|
return g.all()
|
|
}
|
|
return g.again(node, changed)
|
|
}
|
|
|
|
// 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 {
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
g.known = true
|
|
return g.all()
|
|
}
|
|
|
|
// all passes on every report again. The caller holds mu.
|
|
func (g *ToolchainGate) all() bool {
|
|
changed := false
|
|
for _, module := range slices.Sorted(maps.Keys(g.raw)) {
|
|
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
|
changed = g.pass(module, k) || changed
|
|
}
|
|
}
|
|
return changed
|
|
}
|
|
|
|
// again passes on every report of node by a build module among modules. The caller holds mu.
|
|
func (g *ToolchainGate) again(node string, modules []string) bool {
|
|
changed := false
|
|
for _, module := range modules {
|
|
if !g.modules[module] {
|
|
continue
|
|
}
|
|
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
|
if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node {
|
|
changed = g.pass(module, k) || changed
|
|
}
|
|
}
|
|
}
|
|
return changed
|
|
}
|
|
|
|
// pass hands one key on as it stands: its value when its node is placed and wrote it since (or the placement is not
|
|
// known), deleted otherwise. The caller holds mu.
|
|
func (g *ToolchainGate) pass(module, key string) bool {
|
|
value, held := g.raw[module][key]
|
|
if !held {
|
|
return g.forward(module, key, nil, true)
|
|
}
|
|
node, _, err := delivery.ParseToolchainKey(key)
|
|
why := ""
|
|
if err == nil && g.known && g.placed.Seeded() {
|
|
why = g.leftOut(module, node, value)
|
|
}
|
|
if why == "" {
|
|
delete(g.left, module+"/"+key)
|
|
return g.forward(module, key, value, false)
|
|
}
|
|
if g.left[module+"/"+key] != why {
|
|
g.left[module+"/"+key] = why
|
|
g.logf("%s.%s is left out of the agreed toolchain: %s (issue 1940)", delivery.ToolchainsBucket(module), key, why)
|
|
}
|
|
return g.forward(module, key, nil, true)
|
|
}
|
|
|
|
// leftOut is why a node's report is no builder's, or empty when it counts: the catalogue does not place the module on its
|
|
// node, or the report was written before it did. A value that is no toolchain is passed on for its reader to refuse.
|
|
// The caller holds mu.
|
|
func (g *ToolchainGate) leftOut(module, node string, value []byte) string {
|
|
since, placed := g.placed.Since(node, module)
|
|
if !placed {
|
|
return fmt.Sprintf("the catalogue does not place %s on %s, so it is no builder", module, node)
|
|
}
|
|
tc, err := delivery.DecodeToolchain(value)
|
|
if err == nil && tc.At.Before(since) {
|
|
return fmt.Sprintf("it was written at %s, before %s was placed on %s at %s: it counts once %s writes it again",
|
|
tc.At.UTC().Format(time.RFC3339), module, node, since.UTC().Format(time.RFC3339), node)
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// FollowPlacement reads the seat mesh-placement's placement with its history into gate, says it settled, and follows it; changed
|
|
// is told whenever what the gate passes on changed. Where the state cannot be read or watched (not there, or this part
|
|
// may not read it) it says so and the gate passes every report on: not fatal, since an indexer, builder or checker that
|
|
// stops for it stops every delivery (issue 1945). An error of the watch after that ends on the channel.
|
|
func FollowPlacement(ctx context.Context, js process.BucketOpener, gate *ToolchainGate, changed func(context.Context), logf func(string, ...any)) (<-chan error, error) {
|
|
errs := make(chan error, 1)
|
|
kv, err := js.KeyValue(ctx, delivery.PlacementBucket)
|
|
if err != nil {
|
|
logf("the state %s cannot be read, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", delivery.PlacementBucket, err)
|
|
return errs, nil
|
|
}
|
|
w, err := kv.WatchAll(ctx, jetstream.IncludeHistory())
|
|
if err != nil {
|
|
logf("the state %s cannot be watched, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", delivery.PlacementBucket, err)
|
|
return errs, nil
|
|
}
|
|
put := func(e jetstream.KeyValueEntry) bool {
|
|
return gate.Placement(e.Key(), e.Value(), e.Operation() != jetstream.KeyValuePut, e.Created())
|
|
}
|
|
for settled := false; !settled; {
|
|
select {
|
|
case <-ctx.Done():
|
|
_ = w.Stop()
|
|
return errs, ctx.Err()
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
return errs, errors.New("the watch of " + delivery.PlacementBucket + " ended before it settled")
|
|
}
|
|
if e == nil {
|
|
settled = true
|
|
continue
|
|
}
|
|
put(e)
|
|
}
|
|
}
|
|
if gate.Settled() {
|
|
changed(ctx)
|
|
}
|
|
go func() {
|
|
defer func() { _ = w.Stop() }()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
errs <- ctx.Err()
|
|
return
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
errs <- errors.New("the watch of " + delivery.PlacementBucket + " ended")
|
|
return
|
|
}
|
|
if e != nil && put(e) {
|
|
changed(ctx)
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
return errs, nil
|
|
}
|