mesh-delivery/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go

package main

// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves
// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very
// toolchain agreement this file repairs.

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"maps"
	"slices"
	"strings"
	"sync"
	"time"

	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/provider"
)

// A build module writes <node>.<language> of its toolchains state on every node it runs on, and nothing deletes the key
// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of
// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from 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
// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments,
// key <node>.<module>) hold an open assignment of its build module on its node, and only once it was written since that
// assignment; it is passed on as deleted otherwise, said once, and passed on when its node writes it again. A report
// written before the assignment is the one an earlier run of the module left (issue 1940, 2026-10-25 20:56: mesh-build-node
// assigned to ace again brought ace.typescript v26.12.0 of 2026-10-23 back into the agreement, before the module ran
// there, and the controller's take-in refused every module "builders disagree"). Until the assignments are settled (or where they cannot
// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that
// refusal is the guard of novox/hq ADR 0465, and it stays.
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)
	known    bool                         // the assignments are settled
	standing map[string]time.Time         // <node>.<module> with an open assignment -> since when it stands
	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
}

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, standing: map[string]time.Time{},
		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)
}

// Assignment takes one change of the controller's assignments in (at is when the bus wrote that revision) and answers
// whether what it passed on changed.
func (g *toolchainGate) Assignment(key string, value []byte, deleted bool, at time.Time) bool {
	node, module, ok := strings.Cut(key, ".")
	if !ok || !g.modules[module] {
		return false
	}
	assigned, open := openAssignment(value)
	open = open && !deleted
	g.mu.Lock()
	defer g.mu.Unlock()
	was, stood := g.standing[key]
	since := assigned
	if since.IsZero() {
		// A value with no time of its own (an assignment a module pulled in) stands since the revision that opened it.
		since = at
		if stood {
			since = was
		}
	}
	if open == stood && (!open || since.Equal(was)) {
		return false
	}
	if open {
		g.standing[key] = since
	} else {
		delete(g.standing, key)
	}
	if !g.known {
		return false
	}
	changed := false
	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
}

// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it
// passed on changed.
func (g *toolchainGate) Settled() bool {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.known = true
	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
}

// pass hands one key on as it stands: its value when its node is assigned and wrote it since (or the assignments are 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 {
		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: its node has no open assignment of the module,
// or the report was written before that assignment. 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, open := g.standing[node+"."+module]
	if !open {
		return fmt.Sprintf("%s has no assignment 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's assignment 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 ""
}

// openAssignment is whether a value of the controller's assignments holds an assignment that stands (one that has not
// ended and has no state: a pending one is not on the machine yet; mesh-controller internal/chain Assignment), and when
// it was assigned, zero where the value does not say.
func openAssignment(value []byte) (time.Time, bool) {
	var a struct {
		Assigned time.Time `json:"assigned"`
		Ended    time.Time `json:"ended"`
		State    string    `json:"state"`
	}
	if json.Unmarshal(value, &a) != nil {
		return time.Time{}, false
	}
	return a.Assigned, a.Ended.IsZero() && a.State == ""
}

// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is
// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read
// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel.
func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) {
	errs := make(chan error, 1)
	kv, err := js.KeyValue(ctx, provider.AssignmentsBucket)
	if err != nil {
		// Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery).
		logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
		return errs, nil
	}
	each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) }
	w, err := settle(ctx, kv, each)
	if err != nil {
		return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
	}
	if gate.Settled() {
		changed(ctx)
	}
	go func() {
		errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) {
			if gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) {
				changed(ctx)
			}
		})
	}()
	return errs, nil
}