mesh-sdk/go/delivery/buildrun/standing.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
}