mesh-controller/cmd/mesh-controller/kept_compositions.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/inventory"
)

// **A kept composition follows its inputs** (novox/hq ADR 0545, issue 1945). The controller composes a module only at
// the fingerprint it is registered at, and keeps the compositions of up to five fingerprints (ADR 0539 §4) so that
// going back is one want. A kept one was never composed again, so it carried the config of the moment it was made: on
// 2026-10-25 a want back to an earlier mesh-build-go on g14 applied its capacity of 0, the value from before it was
// set. Now a write of mesh-config's resolved values for a module on a node, heard from that bucket, composes the node
// and then every kept fingerprint of that module on that node again, each from its own build's manifest with today's
// inputs, writing only those whose composition differs (writeComposed). The writes are read from a durable consumer,
// acknowledged once handled, so a change made while no controller ran is delivered to the next one, and a start
// composes again only what was written since the last write handled, never every kept composition.

// configTouched is the node and module a key of mesh-config_resolved (`<node>.<module>.<key>`) names.
func configTouched(key string) (node, module string, ok bool) {
	parts := strings.SplitN(key, ".", 3)
	if len(parts) != 3 || parts[0] == "" || parts[1] == "" {
		return "", "", false
	}
	return parts[0], parts[1], true
}

// keptKey is one kept composition to compose again.
type keptKey struct{ module, fp string }

// keptToFollow is, of a node's composition keys, each of modules (every module when nil) at a fingerprint other than
// the one it is registered at, sorted: the kept compositions. The node's own composition and other nodes' are not.
func keptToFollow(keys []string, node string, modules map[string]bool, registered map[string]string) []keptKey {
	var out []keptKey
	for _, key := range keys {
		n, module, fp, isNode, err := sdkdelivery.ParseComposedKey(key)
		if err != nil || isNode || n != node || (modules != nil && !modules[module]) {
			continue
		}
		if fp == registered[module] || !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
			continue
		}
		out = append(out, keptKey{module, fp})
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].module != out[j].module {
			return out[i].module < out[j].module
		}
		return out[i].fp < out[j].fp
	})
	return out
}

// buildAt is, of a module's builds newest first, the newest that worked at fp and kept its manifest.
func buildAt(builds []inventory.Build, fp string) (inventory.Build, bool) {
	for _, b := range builds {
		if b.Worked() && b.SourceFingerprint == fp && len(b.Manifest) > 0 {
			return b, true
		}
	}
	return inventory.Build{}, false
}

// followKept composes again, on node, every kept composition of modules (every module when nil) with today's inputs,
// and writes those that differ. It answers the keys written; a fingerprint whose build's manifest is gone is said and
// left as it is.
func followKept(ctx context.Context, open *stores, node string, modules map[string]bool) ([]string, error) {
	if _, err := theLease.epoch(ctx); err != nil {
		return nil, err
	}
	bus, err := aBus()
	if err != nil {
		return nil, err
	}
	defer bus.Close()
	js, err := jetstream.New(bus.Conn())
	if err != nil {
		return nil, err
	}
	inv := open.inventory
	registered, err := currentFingerprints(ctx, inv)
	if err != nil {
		return nil, err
	}
	at, err := openComposedPlace(ctx, js, bus.Conn(), node)
	if err != nil {
		return nil, err
	}
	lister, err := at.bucket.ListKeys(ctx)
	if err != nil {
		if errors.Is(err, jetstream.ErrNoKeysFound) {
			return nil, nil // empty-on-error: a node with no compositions keeps none to compose again
		}
		return nil, err
	}
	var keys []string
	for k := range lister.Keys() {
		keys = append(keys, k)
	}
	kept := keptToFollow(keys, node, modules, registered)
	if len(kept) == 0 {
		return nil, nil
	}
	sign, err := controllerSigner(ctx, open)
	if err != nil {
		return nil, err
	}
	upgrades, err := inv.Upgrades(ctx)
	if err != nil {
		return nil, err
	}
	at.sign = sign
	at.policy = func(module string) (string, string, string) {
		u, ok := upgrades[module]
		if !ok {
			return "", "", ""
		}
		return u.Policy(), u.From, u.Why
	}
	in, err := inputFor(ctx, open)
	if err != nil {
		return nil, err
	}
	var wrote []string
	var errs []error
	builds := map[string][]inventory.Build{}
	for _, k := range kept {
		if ctx.Err() != nil {
			return wrote, errors.Join(append(errs, ctx.Err())...) // stopped: the rest is the next controller's
		}
		if _, read := builds[k.module]; !read {
			if builds[k.module], err = inv.Builds(ctx, k.module, 20); err != nil {
				errs = append(errs, fmt.Errorf("%s's builds: %w", k.module, err))
				continue
			}
		}
		b, found := buildAt(builds[k.module], k.fp)
		if !found {
			fmt.Printf("compositions: %s's kept %s on %s has no build whose manifest is kept, and is not composed again\n",
				k.module, short(k.fp), node)
			continue
		}
		var m catalogue.Manifest
		if err := json.Unmarshal(b.Manifest, &m); err != nil {
			errs = append(errs, fmt.Errorf("the manifest of %s's build %s: %w", k.module, b.ID, err))
			continue
		}
		c, err := keptComposition(in.Input, node, k.module, k.fp, m)
		if err != nil {
			errs = append(errs, fmt.Errorf("%s at %s on %s: %w", k.module, short(k.fp), node, err))
			continue
		}
		key, _ := sdkdelivery.ComposedKey(node, k.module, k.fp)
		// One write on the node's keys at a time in this process, held only for the write: composing in memory holds
		// nothing, so a sweep never keeps a take-in from the node (the review of ADR 0545).
		unlock := lockComposing(node)
		w, err := writeComposed(ctx, at, key, c)
		unlock()
		if err != nil {
			errs = append(errs, err)
			continue
		}
		if w {
			wrote = append(wrote, key)
		}
	}
	if len(wrote) > 0 {
		fmt.Printf("compositions: %s: kept compositions composed again with today's inputs: %s\n", node, strings.Join(wrote, ", "))
	}
	return wrote, errors.Join(errs...)
}

// configFollowGather is how long the config consumer waits after a write for more before it composes: mesh-config writes
// a module's keys one by one.
var configFollowGather = time.Second

// configFollowConsumer is the controller's durable consumer of mesh-config's resolved values (ADR 0545 §1, §3).
const configFollowConsumer = broker.ConfigFollowConsumer

// configFollowConfig is that consumer: each resolved key's last value once when it is made, then each write,
// acknowledged after the write is handled, so what a stop cut is delivered to the next controller. The first start of
// the release that makes it composes again the kept compositions of every module with config, through the same gather,
// lock and acknowledgement: mesh-config never writes an unchanged value again, so a value set before this release (g14's
// mesh-build-go capacity of 2, against kept compositions carrying 0) would otherwise never reach them. Its deliver
// policy is set before its first delivery; an existing durable's cannot be changed.
func configFollowConfig() jetstream.ConsumerConfig {
	return jetstream.ConsumerConfig{Durable: configFollowConsumer, FilterSubject: "$KV." + broker.ConfigResolvedBucket + ".>",
		DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait,
		MaxAckPending: takeInPending, Description: "the controller's following of config into kept compositions (novox/hq ADR 0545)"}
}

// kindConfigNotFollowed is the condition of a config consumer that cannot be read: a change of config composes nothing
// until it is.
const kindConfigNotFollowed = "config-not-followed"

func configNotFollowedObservation(err error) conditions.Observation {
	return conditions.Observation{Scope: conditions.ScopeCore, ID: "config-follow", Token: kindConfigNotFollowed,
		Kind: kindConfigNotFollowed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "config",
		Summary: fmt.Sprintf("mesh-config's resolved values are not heard, so a change of config composes nothing: %v", err)}
}

// watchConfigFollow reads the writes of mesh-config's resolved values from the controller's durable consumer and, for
// each node and module they touch, composes the node and the module's kept compositions there again, until ctx ends
// (ADR 0545). A consumer that cannot be made or read raises config-not-followed and is tried again a minute later; it is
// cleared once it is read.
func watchConfigFollow(ctx context.Context, open *stores) {
	goServing(ctx, func() {
		for {
			err := followConfigOnce(ctx, open)
			if ctx.Err() != nil {
				return
			}
			fmt.Printf("config: %v; tried again in a minute\n", err)
			if keeper := conditionsFrom; keeper != nil {
				_, _ = keeper.Observe(ctx, configNotFollowedObservation(err))
			}
			select {
			case <-ctx.Done():
				return
			case <-time.After(time.Minute):
			}
		}
	})
}

// followConfigOnce reads the consumer until it fails or ctx ends.
func followConfigOnce(ctx context.Context, open *stores) error {
	bus, err := aBus()
	if err != nil {
		return err
	}
	defer bus.Close()
	js, err := jetstream.New(bus.Conn())
	if err != nil {
		return err
	}
	c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+broker.ConfigResolvedBucket, configFollowConfig())
	if err != nil {
		return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err)
	}
	it, err := c.Messages(jetstream.PullMaxMessages(startTakeIns))
	if err != nil {
		return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err)
	}
	defer it.Stop()
	if keeper := conditionsFrom; keeper != nil {
		_, _ = keeper.Clear(ctx, configNotFollowedObservation(nil).Key(), "mesh-config's resolved values are heard")
	}
	type next struct {
		m   jetstream.Msg
		err error
	}
	got := make(chan next)
	go func() {
		for {
			m, err := it.Next()
			select {
			case got <- next{m, err}:
			case <-ctx.Done():
				if m != nil {
					_ = m.Nak()
				}
				return
			}
			if err != nil {
				return
			}
		}
	}()
	var held []jetstream.Msg
	touched := map[string]map[string]bool{}
	var gather <-chan time.Time
	for {
		select {
		case <-ctx.Done():
			for _, m := range held {
				_ = m.Nak() // delivered at once to the next controller
			}
			return nil
		case n := <-got:
			if n.err != nil {
				for _, m := range held {
					_ = m.Nak()
				}
				return fmt.Errorf("the consumer of %s stopped: %w", broker.ConfigResolvedBucket, n.err)
			}
			held = append(held, n.m)
			key := strings.TrimPrefix(n.m.Subject(), "$KV."+broker.ConfigResolvedBucket+".")
			if node, module, ok := configTouched(key); ok {
				if touched[node] == nil {
					touched[node] = map[string]bool{}
				}
				touched[node][module] = true
			}
			if gather == nil {
				gather = time.After(configFollowGather)
			}
		case <-gather:
			gather = nil
			now, msgs := touched, held
			touched, held = map[string]map[string]bool{}, nil
			forgetResolved() // the next composition reads the values just written, not a read of the last 5 s
			failed := false
			for node, modules := range now {
				if ctx.Err() != nil {
					failed = true // stopped: the writes are refused below and delivered to the next controller
					break
				}
				if err := composerFor(open).compose(ctx, []string{node}, false); err != nil {
					fmt.Printf("config: %s is not composed after its config changed: %v\n", node, err)
					failed = true
				}
				if _, err := followKept(ctx, open, node, modules); err != nil {
					fmt.Printf("config: %s's kept compositions are not composed again: %v\n", node, err)
					failed = true
				}
			}
			for _, m := range msgs {
				if failed && ctx.Err() != nil {
					_ = m.Nak() // stopped: delivered at once to the next controller
				} else if failed {
					_ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again
				} else {
					acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
					if err := m.DoubleAck(acking); err != nil {
						fmt.Printf("config: the bus did not take the acknowledgement of %s: %v\n", m.Subject(), err)
					}
					cancel()
				}
			}
		}
	}
}

// forgetResolved drops the last read of mesh-config's resolved values.
func forgetResolved() {
	theResolved.mu.Lock()
	theResolved.entries = nil
	theResolved.mu.Unlock()
}