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

package main

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

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

	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildnode"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
)

// toolchainReportedPrefixes are the subjects the three build seats say toolchain-reported on, a node its last token (design
// 47 G §492). The builder hears them through the consumer of its manifest's consumes, and keeps what it heard in its
// state toolchains-heard (§604).
var toolchainReportedPrefixes = []string{
	strings.TrimSuffix(nodebuildgo.SubjectToolchainReportedAll, "*"),
	strings.TrimSuffix(nodebuildnode.SubjectToolchainReportedAll, "*"),
}

// heardKept is the state toolchains-heard: one key <node>.<language> (design 47 G §604).
type heardKept interface {
	Put(key string, value []byte) error
	Delete(key string) error
}

// heardState is the state as the builder reads it back at its start.
type heardState interface {
	heardKept
	Keys() ([]string, error)
	Get(key string) ([]byte, error)
}

// kvHeard is heardState over the bound bucket.
type kvHeard struct {
	kvKeeper
}

func (k kvHeard) Keys() ([]string, error) {
	keys, err := k.kv.Keys(k.ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	return keys, err
}

func (k kvHeard) Get(key string) ([]byte, error) {
	e, err := k.kv.Get(k.ctx, key)
	if err != nil {
		return nil, err
	}
	return e.Value(), nil
}

// toolchainsHeard keeps the toolchains the build seats said (design 47 G §492, §604), forgets a node's on the node-engine
// seat's stopped (§546), and says toolchain-agreed with its digests when a language's agreed version changes (§472, §574).
type toolchainsHeard struct {
	set     *buildrun.ToolchainSet
	changed func(context.Context)
	say     func(ctx context.Context, e meshbuilds.ToolchainAgreed) error
	now     func() time.Time
	logf    func(string, ...any)

	mu     sync.Mutex
	kept   heardKept            // nil until the state is bound: what was heard is kept in memory only, and written then
	agreed map[string]string    // language -> the version last said
	runAt  map[string]time.Time // <node>.<module> -> the `at` of the newest running read
}

func newToolchainsHeard(set *buildrun.ToolchainSet, changed func(context.Context), say func(context.Context, meshbuilds.ToolchainAgreed) error, logger *log.Logger) *toolchainsHeard {
	return &toolchainsHeard{set: set, changed: changed, say: say, now: time.Now, logf: logger.Printf, agreed: map[string]string{}, runAt: map[string]time.Time{}}
}

// bind takes the bound state in: the toolchains it holds are read back (a restart needs no node to report again), and
// what was heard before it was bound is written.
func (h *toolchainsHeard) bind(ctx context.Context, state heardState) {
	keys, err := state.Keys()
	if err != nil {
		h.logf("%s cannot be read, so the toolchains of before this start are not known: %v", buildrun.ToolchainsHeardState, err)
	}
	for _, k := range keys {
		value, err := state.Get(k)
		if err != nil {
			continue
		}
		node, language, err := buildstep.ParseToolchainKey(k)
		tc, derr := buildstep.DecodeToolchain(value)
		if err != nil || derr != nil || tc.Language != language {
			h.logf("%s.%s holds what is not a toolchain, and is left: %v %v", buildrun.ToolchainsHeardState, k, err, derr)
			continue
		}
		h.set.Set(node, tc)
	}
	h.mu.Lock()
	h.kept = state
	h.mu.Unlock()
	for _, language := range h.set.Seats.Languages() {
		for _, e := range h.set.Entries(language) {
			h.keep(e.Node, e.Toolchain)
		}
	}
	h.agree(ctx)
	h.changed(ctx)
}

func (h *toolchainsHeard) keep(node string, tc buildstep.Toolchain) {
	h.mu.Lock()
	kept := h.kept
	h.mu.Unlock()
	if kept == nil {
		return
	}
	key, err := buildstep.ToolchainKey(node, tc.Language)
	if err != nil {
		return
	}
	data, err := buildstep.EncodeToolchain(tc)
	if err == nil {
		err = kept.Put(key, data)
	}
	if err != nil {
		h.logf("%s.%s could not be written, so it is kept in memory only: %v", buildrun.ToolchainsHeardState, key, err)
	}
}

// Reported takes one toolchain-reported of a build seat in; it answers whether the message was one. The toolchain is
// kept in the state, and builds that waited for the nodes to agree are looked at again.
func (h *toolchainsHeard) Reported(ctx context.Context, m jetstream.Msg) bool {
	subject := m.Subject()
	var prefix string
	for _, p := range toolchainReportedPrefixes {
		if strings.HasPrefix(subject, p) {
			prefix = p
		}
	}
	if prefix == "" {
		return false
	}
	_ = m.Ack()
	var e nodebuildgo.ToolchainReported
	if err := json.Unmarshal(m.Data(), &e); err != nil || e.Node == "" || e.Node != strings.TrimPrefix(subject, prefix) {
		h.logf("%s holds no toolchain of its node, and is dropped: %v", subject, err)
		return true
	}
	tc := buildstep.Toolchain{Language: e.Language, Version: e.Version, Recipe: e.Recipe, Digest: e.Digest, At: h.now().UTC()}
	if _, err := buildstep.EncodeToolchain(tc); err != nil {
		h.logf("%s is no toolchain, and is dropped: %v", subject, err)
		return true
	}
	if h.set.Set(e.Node, tc) {
		h.keep(e.Node, tc)
		h.agree(ctx)
		h.changed(ctx)
	}
	return true
}

// Running takes one running of the node-engine seat in (design 47 G §546); it answers whether the message was one. A
// running with no fingerprint is stopped (§365): the key <node>.<language> of the build module of that language on that
// node is deleted. An event older by `at` than one taken before it changes nothing.
func (h *toolchainsHeard) Running(ctx context.Context, m jetstream.Msg) bool {
	if !strings.HasPrefix(m.Subject(), strings.TrimSuffix(nodeengine.SubjectRunningAll, "*.*")) {
		return false
	}
	_ = m.Ack()
	h.running(ctx, m.Data())
	return true
}

func (h *toolchainsHeard) running(ctx context.Context, data []byte) {
	var r nodeengine.Running
	if err := json.Unmarshal(data, &r); err != nil {
		h.logf("a running that does not read is dropped: %v", err)
		return
	}
	id := r.Node + "." + r.Module
	h.mu.Lock()
	if prev, ok := h.runAt[id]; ok && r.At.Before(prev) {
		h.mu.Unlock()
		return
	}
	h.runAt[id] = r.At
	kept := h.kept
	h.mu.Unlock()
	language, build := h.set.Seats.LanguageOf(r.Module)
	if !build || r.Fingerprint != "" {
		return
	}
	if !h.set.Delete(r.Node, language) {
		return
	}
	key, _ := buildstep.ToolchainKey(r.Node, language)
	if kept != nil {
		if err := kept.Delete(key); err != nil {
			h.logf("%s.%s could not be deleted: %v", buildrun.ToolchainsHeardState, key, err)
		}
	}
	h.agree(ctx)
	h.changed(ctx)
}

// Dropped is the key's drop by a person (builder_toolchains_drop), taken from the state's watch.
func (h *toolchainsHeard) Dropped(ctx context.Context, key string) {
	node, language, err := buildstep.ParseToolchainKey(key)
	if err != nil {
		return
	}
	if h.set.Delete(node, language) {
		h.agree(ctx)
		h.changed(ctx)
	}
}

// agree says toolchain-agreed for each language whose agreed version is not the one last said (design 47 G §472: said
// only when the agreed version changes, ADR 0584 §1).
func (h *toolchainsHeard) agree(ctx context.Context) {
	for _, language := range h.set.Seats.Languages() {
		entries := h.set.Entries(language)
		tc, none := buildstep.AgreedToolchain(entries)
		if none != "" {
			continue
		}
		h.mu.Lock()
		same := h.agreed[language] == tc.Version
		if !same {
			h.agreed[language] = tc.Version
		}
		h.mu.Unlock()
		if same {
			continue
		}
		at := h.now().UTC()
		e := meshbuilds.ToolchainAgreed{V: 1, At: &at, Language: language, Version: tc.Version, Digests: []meshbuilds.ToolchainAgreedDigestsItem{}}
		for _, d := range buildrun.Digests(entries, tc) {
			e.Digests = append(e.Digests, meshbuilds.ToolchainAgreedDigestsItem{Digest: d.Digest, Nodes: d.Nodes})
		}
		if err := h.say(ctx, e); err != nil {
			h.logf("saying %s of %s %s: %v", meshbuilds.EventToolchainAgreed, language, tc.Version, err)
			h.mu.Lock()
			delete(h.agreed, language) // said again at the next change
			h.mu.Unlock()
		}
	}
}

// lastPerSubject reads the last message of every subject of a stream kept last per subject, through an ordered consumer
// made for that read (design 47 G §556), and hands each to each with its stream sequence; it answers when the last of them was handed.
func lastPerSubject(ctx context.Context, js jetstream.JetStream, stream, filter string, each func(subject string, data []byte, sequence uint64)) error {
	cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy})
	if err != nil {
		return fmt.Errorf("reading %s: %w", stream, err)
	}
	info, err := cons.Info(ctx)
	if err != nil {
		return fmt.Errorf("reading %s: %w", stream, err)
	}
	pending := int(info.NumPending)
	if pending == 0 {
		return nil
	}
	msgs, err := cons.Messages()
	if err != nil {
		return fmt.Errorf("reading %s: %w", stream, err)
	}
	defer msgs.Stop()
	for ; pending > 0; pending-- {
		m, err := msgs.Next()
		if err != nil {
			return fmt.Errorf("reading %s: %w", stream, err)
		}
		seq := uint64(0)
		if meta, err := m.Metadata(); err == nil {
			seq = meta.Sequence.Stream
		}
		each(m.Subject(), m.Data(), seq)
	}
	return nil
}

// readRunning is the start's read of every node's module's last running (§556), before the consumer's.
func (h *toolchainsHeard) readRunning(ctx context.Context, js jetstream.JetStream) {
	if err := lastPerSubject(ctx, js, nodeengine.StreamRunning, nodeengine.SubjectRunningAll, func(_ string, data []byte, _ uint64) { h.running(ctx, data) }); err != nil {
		h.logf("the running of the node-engine seat is not read at the start, so a build module stopped while this builder was away is forgotten at its next running: %v", err)
	}
}