mesh-delivery/modules/mesh-delivery-builder/buildrun/standing.go

package buildrun

import (
	"context"
	"errors"
	"sort"
	"sync"
	"time"

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

	"git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
	meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
	"git.novox.be/novox/mesh-sdk/go/process"
)

// PlacedNodes holds which modules the catalogue places on which node: the seat mesh-module-assigner's state `assignments`,
// held by the assigner (AssignmentsBucket, read by meshmoduleassignerclient.Placed; ADR 0567), or the assignment half
// (meshwantsclient.Placed, fed by Revision). The builder counts the nodes it names (Nodes; ADR 0624 §1). It holds no
// toolchain: the builder keeps the toolchains it heard in its state toolchains-heard and forgets one when the node-engine
// seat says the build module stopped on the node, never on a placement (design 47 G §546).
type PlacedNodes struct {
	mu     sync.Mutex
	logf   func(format string, a ...any)
	placed placements
	known  bool // the assignments is settled
	// OnPlaced is told, outside the lock, once the assignments's history is read and after each revision of it
	// FollowAssignments hears from then on (the counted nodes, Nodes, may have changed). Nil tells nobody.
	OnPlaced func()
}

// placements is where the part reads which modules stand on which node.
type placements interface {
	Seeded() bool
	ByNode() map[string][]string
}

// NewPlacedNodes is the placements of the builder, saying with logf what it cannot read.
func NewPlacedNodes(logf func(string, ...any)) *PlacedNodes {
	return &PlacedNodes{logf: logf, placed: meshmoduleassignerclient.NewPlaced()}
}

// Assignments takes one revision of the seat mesh-module-assigner's assignments in (at is when the bus wrote it) and
// answers whether the nodes the modules stand on changed. A value that does not read is said, and places nothing.
func (g *PlacedNodes) Assignments(key string, value []byte, deleted bool, at time.Time) bool {
	g.mu.Lock()
	defer g.mu.Unlock()
	placed, ok := g.placed.(*meshmoduleassignerclient.Placed)
	if !ok {
		g.logf("%s: %s is not read, the placements take the assignment half (design 47)", AssignmentsBucket, key)
		return false
	}
	was := placed.Seeded()
	_, changed, err := placed.Put(key, value, deleted, at)
	if err != nil {
		g.logf("%s: %v (issue 1940)", AssignmentsBucket, err)
	}
	if !g.known {
		return false
	}
	return was != g.placed.Seeded() || len(changed) > 0
}

// Revision takes one assignment graph revision the part's view holds in (design 47 *Each part acts on its own entries*:
// the placements come from the seat mesh-wants' assignment half, never the assigner's state) and answers whether the
// nodes the modules stand on changed. The first revision taken settles it. Placements fed by Revision read no Assignments.
func (g *PlacedNodes) Revision(r meshwants.AssignmentRevision) bool {
	g.mu.Lock()
	defer g.mu.Unlock()
	placed, ok := g.placed.(*meshwantsclient.Placed)
	if !ok {
		placed = meshwantsclient.NewPlaced()
		g.placed = placed
	}
	was := placed.Seeded()
	changed := placed.Take(r)
	if changed == nil {
		return false
	}
	first := !g.known
	g.known = true
	return first || was != placed.Seeded() || len(changed) > 0
}

// Nodes are the nodes the assignments places any module on, sorted: the nodes the mesh counts (a node leaves the count
// when the assignments no longer names it; ADR 0624 §1). None until the assignments is read.
func (g *PlacedNodes) 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 assignments's history is read.
func (g *PlacedNodes) Settled() {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.known = true
}

// assignmentsFollowed is what FollowAssignments feeds: the builder's PlacedNodes, or the ToolchainGate of the checker and the
// indexer. settle says the history is read and answers whether what the part passes on changed; placedTold tells its OnPlaced.
type assignmentsFollowed interface {
	Assignments(key string, value []byte, deleted bool, at time.Time) bool
	settle() bool
	placedTold()
}

func (g *PlacedNodes) settle() bool { g.Settled(); return true }
func (g *PlacedNodes) placedTold() {
	if g.OnPlaced != nil {
		g.OnPlaced()
	}
}

func (g *ToolchainGate) settle() bool { return g.Settled() }
func (g *ToolchainGate) placedTold() {
	if g.OnPlaced != nil {
		g.OnPlaced()
	}
}

// FollowAssignments reads the seat mesh-module-assigner's assignments 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 FollowAssignments(ctx context.Context, js process.BucketOpener, gate assignmentsFollowed, changed func(context.Context), logf func(string, ...any)) (<-chan error, error) {
	errs := make(chan error, 1)
	kv, err := js.KeyValue(ctx, AssignmentsBucket)
	if err != nil {
		logf("the state %s cannot be read, so no node is counted by the assignments: %v", AssignmentsBucket, err)
		return errs, nil
	}
	w, err := kv.WatchAll(ctx, jetstream.IncludeHistory())
	if err != nil {
		logf("the state %s cannot be watched, so no node is counted by the assignments: %v", AssignmentsBucket, err)
		return errs, nil
	}
	put := func(e jetstream.KeyValueEntry) bool {
		return gate.Assignments(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 " + AssignmentsBucket + " ended before it settled")
			}
			if e == nil {
				settled = true
				continue
			}
			put(e)
		}
	}
	if gate.settle() {
		changed(ctx)
	}
	gate.placedTold()
	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 " + AssignmentsBucket + " ended")
					return
				}
				if e != nil && put(e) {
					changed(ctx)
				}
				if e != nil {
					gate.placedTold()
				}
			}
		}
	}()
	return errs, nil
}