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 (the holder's `<holder>_assignments`, 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
	// FollowAssignmentsIn 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("the assignments: %s is not read, the placements take the assignment half (design 47)", key)
		return false
	}
	was := placed.Seeded()
	_, changed, err := placed.Put(key, value, deleted, at)
	if err != nil {
		g.logf("the assignments: %v (issue 1940)", 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 FollowAssignmentsIn 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()
	}
}

// FollowAssignmentsIn reads the seat mesh-module-assigner's assignments, the state `bucket` of its holder (found at run time, design 47
// G A-119, A-126: AssignmentsOf), 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 FollowAssignmentsIn(ctx context.Context, js process.BucketOpener, bucket string, gate assignmentsFollowed, changed func(context.Context), logf func(string, ...any)) (<-chan error, error) {
	errs := make(chan error, 1)
	kv, err := js.KeyValue(ctx, bucket)
	if err != nil {
		logf("the state %s cannot be read, so no node is counted by the assignments: %v", bucket, 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", bucket, 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 " + bucket + " 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 " + bucket + " ended")
					return
				}
				if e != nil && put(e) {
					changed(ctx)
				}
				if e != nil {
					gate.placedTold()
				}
			}
		}
	}()
	return errs, nil
}