mesh-delivery/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go

package main

import (
	"context"
	"slices"
	"sync"

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

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/process"

	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
	"git.novox.be/novox/mesh-delivery/sdk/enginereads"
)

// EngineReadsState is the builder's state of what each counted node's engine states it reads (novox/hq ADR 0618 §2, ADR
// 0624 §1): one key `<node>`, written from the node-engine seat's verb manifest-reads at the builder's start, on the
// seat's event manifest-reads-changed for that node, and when a person asks again (builder_engine_reads); a node that
// is away keeps its last statement, and its key is removed when the placement no longer names the node. A counted node
// with no key reads as stating the baseline: its first forms. No node's record is read in its place.
const EngineReadsState = "engine-reads"

// engineReadsBucket is that state's bucket.
const engineReadsBucket = delivery.BuilderModule + "_" + EngineReadsState

// engineReads keeps the statements and the counted nodes for the builder's forms (build.Builder.Forms).
type engineReads struct {
	keeper  *enginereads.Keeper
	gate    *buildrun.ToolchainGate
	mu      sync.Mutex
	counted []string
	// window is the old reference form's window over these statements (DLV-B9); nil keeps none.
	window *oldForm
	ctx    context.Context
}

// newEngineReads asks through call (process.Conn.CallOn) and keeps each answer in the state once it is bound.
func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, logf func(string, ...any)) *engineReads {
	r := &engineReads{gate: gate, keeper: &enginereads.Keeper{Call: call, Logf: logf}}
	r.keeper.Changed = r.stepWindow
	return r
}

// stepWindow moves the old form's window to what the counted nodes state now (design 51 *Every build keeps the old
// reference form*): on a statement's change, the placement's and a hand drop, and at each build key written.
func (r *engineReads) stepWindow() {
	if r.window == nil {
		return
	}
	ctx := r.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	_, counted := r.forms()
	r.window.step(ctx, counted)
}

// bind keeps the statements in kv and hears it: every node's last statement at the start, a person's ask again and drop
// later. Bound once the state is on the bus (process.AwaitBucket).
func (r *engineReads) bind(ctx context.Context, kv jetstream.KeyValue, errs chan<- error, logf func(string, ...any)) {
	r.keeper.Put = func(node string, v []byte) error { _, err := kv.Put(ctx, node, v); return err }
	r.keeper.Delete = func(node string) error { return kv.Delete(ctx, node) }
	each := func(e jetstream.KeyValueEntry) {
		r.keeper.Heard(e.Key(), e.Value(), e.Operation() != jetstream.KeyValuePut)
	}
	w, err := settle(ctx, kv, each)
	if err != nil {
		process.NotWatched(logf, engineReadsBucket, err, "builder_engine_reads is not answered until the builder starts again")
		return
	}
	go func() { errs <- follow(ctx, w, each) }()
	// Every counted node asked again, so what was asked before the state was bound is kept in it too.
	r.mu.Lock()
	nodes := slices.Clone(r.counted)
	r.mu.Unlock()
	for _, node := range nodes {
		r.keeper.AskNode(node, "the state engine-reads is bound")
	}
}

// placed takes the placement in: a node it names newly is asked, one it no longer names is forgotten, key and all. The
// first call asks every counted node (the builder's start).
func (r *engineReads) placed() {
	nodes := r.gate.Nodes()
	r.mu.Lock()
	before, first := r.counted, r.counted == nil
	r.counted = nodes
	r.mu.Unlock()
	for _, node := range nodes {
		if first || !slices.Contains(before, node) {
			r.keeper.AskNode(node, "counted: the placement names it")
		}
	}
	for _, node := range before {
		if !slices.Contains(nodes, node) {
			r.keeper.Forget(node)
		}
	}
	r.stepWindow()
}

// heard takes an event of the builder's consumer on EVENTS: manifest-reads-changed asks its node again, and answers true.
func (r *engineReads) heard(m jetstream.Msg, logf func(string, ...any)) bool {
	node, ours, err := enginereads.NodeOf(m.Subject(), m.Data())
	if !ours {
		return false
	}
	if ackErr := m.Ack(); ackErr != nil {
		logf("acknowledging %s: %v", m.Subject(), ackErr)
	}
	if err != nil {
		logf("%s holds what is not %s, and %s is asked again all the same: %v", m.Subject(), enginereads.Event, node, err)
	}
	r.keeper.AskNode(node, enginereads.Event)
	return true
}

// forms is the counted nodes' statements, and those of the nodes whose engine answered; a retired node dropped by hand
// (builder_old_form) counts no more.
func (r *engineReads) forms() (present, counted map[string][]string) {
	r.mu.Lock()
	nodes := slices.Clone(r.counted)
	r.mu.Unlock()
	if r.window != nil {
		dropped := r.window.dropped()
		nodes = slices.DeleteFunc(nodes, func(n string) bool { return dropped[n] })
	}
	return r.keeper.Present(nodes), r.keeper.Reads(nodes)
}

// built is build.Builder.Forms: forms, and the window stepped with them, so a node's seven days are seen at the next
// build key at the latest.
func (r *engineReads) built() (present, counted map[string][]string) {
	present, counted = r.forms()
	if r.window != nil {
		ctx := r.ctx
		if ctx == nil {
			ctx = context.Background()
		}
		r.window.step(ctx, counted)
	}
	return present, counted
}