mesh-controller/cmd/mesh-controller/composed_keys.go

package main

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"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/builderkeys"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/inventory"
)

// A new fingerprint of a registered module, from its done key to its compositions (novox/hq ADR 0539 clauses 5 and 7,
// ADR 0501 §2). With no delivery plan left to take a module's done key in, a write of the builder's or the indexer's
// keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names (registers the
// module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what a module's
// new manifest needs, a credential among it, as a send made it), composes the node holding the bus (its user list now
// carries the module's permissions at the new fingerprint: a new revision of the bus module's composition alone,
// written into its want at once by the assigner), and writes `grants.<module>.<fp>`. The assigner's sixth and seventh
// rules then let a want name the fingerprint. **No timer**: woken by the keys, and a starting controller looks at every
// module once (doneKeysAtStart).

// doneKeySteps are what taking a module's done key in does, each a step a test may stand in for.
type doneKeySteps struct {
	// held says the catalogue holds the module (a new one is the new-module take-in's, new_modules.go).
	held func(ctx context.Context, module string) (bool, error)
	// trunk is the indexer's trunk value of the module.
	trunk func(ctx context.Context, module string) (sdkdelivery.Module, error)
	// registeredAt says the module is registered at the commit.
	registeredAt func(ctx context.Context, module, commit string) (bool, error)
	// takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey); errKeyWaits when
	// the key is not done yet.
	takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error
	// nodesOf are the nodes the module is assigned to; busHolder the node holding the bus, "" for none.
	nodesOf   func(ctx context.Context, module string) ([]string, error)
	busHolder func(ctx context.Context) (string, error)
	// compose writes the nodes' compositions; grants writes the grants issued.
	compose func(ctx context.Context, nodes []string) error
	grants  func(ctx context.Context) error
}

// take takes module's done key in and composes what it changed. It answers what it did in a line, "" when there was
// nothing to do.
func (s doneKeySteps) take(ctx context.Context, module string) (string, error) {
	held, err := s.held(ctx, module)
	if err != nil || !held {
		return "", err
	}
	trunk, err := s.trunk(ctx, module)
	if errors.Is(err, builderkeys.ErrNoValue) {
		return "", nil
	}
	if err != nil {
		return "", err
	}
	if !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) {
		return "", nil
	}
	at, err := s.registeredAt(ctx, module, trunk.Commit)
	if err != nil {
		return "", err
	}
	said := ""
	if !at {
		if err := s.takeIn(ctx, module, trunk); err != nil {
			return "", err
		}
		said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", module, short(trunk.FP2), short(trunk.Commit))
	}
	nodes, err := s.nodesOf(ctx, module)
	if err != nil {
		return said, err
	}
	if len(nodes) == 0 {
		return said, nil
	}
	// The module's nodes first: composing them makes what the new manifest needs, its credentials among it.
	if err := s.compose(ctx, nodes); err != nil {
		return said, err
	}
	// Then the bus's user list, which carries the module's permissions at its new fingerprint.
	holder, err := s.busHolder(ctx)
	if err != nil {
		return said, err
	}
	if holder != "" && !inList(nodes, holder) {
		if err := s.compose(ctx, []string{holder}); err != nil {
			return said, err
		}
	}
	// Then the grant, which with the composition lets a want name the fingerprint (rules 6 and 7).
	if err := s.grants(ctx); err != nil {
		return said, err
	}
	if said == "" {
		said = fmt.Sprintf("%s at %s", module, short(trunk.FP2))
	}
	return said + "; composed for " + strings.Join(nodes, ", ") + ", its grant written", nil
}

func inList(list []string, s string) bool {
	for _, v := range list {
		if v == s {
			return true
		}
	}
	return false
}

// doneKeyTakeOnce is one take-in; a variable so a test may stand in for it. doneKeysSay says a line of the take-ins.
var (
	doneKeyTakeOnce = takeDoneKeyOnce
	doneKeysSay     = func(s string) { fmt.Println(s) }
)

// doneKeysBeforeLetGo is called just before a follower lets go of its module; a variable so a test can wake the module
// at that moment. Nil does nothing.
var doneKeysBeforeLetGo func(module string)

// doneKeysLookAgain marks a module woken while a take-in of it ran: that take-in goes round once more.
var doneKeysLookAgain sync.Map

// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module.
var doneKeysMu sync.Map

// takeDoneKey is a woken module's take-in on the live stores and buckets, said. A take-in that fails (its compositions
// or its grant not written) is tried again every doneKeyRetry while it fails, loudly: printed each time, and an urgent
// condition raised by the first until it has worked, as the chain's own moves are (layBuildsOutSoon). A key not done
// yet is no failure: it is said once for each reason it waits for and looked at again every doneKeyRetry.
func takeDoneKey(ctx context.Context, open *stores, module string) {
	mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
	for {
		if !mu.(*sync.Mutex).TryLock() {
			// A take-in of the module already follows it: it is told to look again once it is done, so a newer key
			// written meanwhile is never lost (the review of issue 1945).
			doneKeysLookAgain.Store(module, true)
			return
		}
		followDoneKey(ctx, open, module)
		if doneKeysBeforeLetGo != nil {
			doneKeysBeforeLetGo(module)
		}
		mu.(*sync.Mutex).Unlock()
		// A wake between this follower's last look and its letting go failed its TryLock and only marked the module:
		// looked at again now that it is let go, and taken once more (the review of issue 1945).
		if _, again := doneKeysLookAgain.LoadAndDelete(module); !again || ctx.Err() != nil {
			return
		}
	}
}

// followDoneKey takes a module's done key in, under the module's lock, until it is taken in or ctx ends: a failure or a
// wait is tried again every doneKeyRetry.
func followDoneKey(ctx context.Context, open *stores, module string) {
	failed := false
	saidWait := ""
	for {
		doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
		err := doneKeyTakeOnce(ctx, open, module)
		o := doneKeyObservation(module)
		if err == nil {
			if keeper := conditionsFrom; keeper != nil && failed {
				_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
				failed = false
			}
			if _, again := doneKeysLookAgain.LoadAndDelete(module); again {
				continue // a wake came while this one ran: once more, for the key written meanwhile
			}
			return
		}
		if errors.Is(err, errKeyWaits) {
			// A take-in that waits is said once for each reason, and tried again every doneKeyRetry: the key write that
			// would wake it may never come (issue 1945, live on 2026-10-25: mesh-host's take-in waited unsaid).
			if why := err.Error(); why != saidWait {
				doneKeysSay(fmt.Sprintf("done keys: %s waits: %s; looked at again every %s", module,
					strings.TrimPrefix(why, errKeyWaits.Error()+": "), doneKeyRetry))
				saidWait = why
			}
		} else {
			doneKeysSay(fmt.Sprintf("done keys: %s: %v; tried again in %s", module, err, doneKeyRetry))
		}
		if errors.Is(err, errKeyWaits) {
			select {
			case <-ctx.Done():
				return
			case <-time.After(doneKeyRetry):
			}
			continue
		}
		if keeper := conditionsFrom; keeper != nil && !failed {
			o.Summary += err.Error()
			if _, oerr := keeper.Observe(ctx, o); oerr == nil {
				failed = true
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(doneKeyRetry):
		}
	}
}

// doneKeyRetry is how long a failed take-in of a done key waits before it tries again.
var doneKeyRetry = 30 * time.Second

// kindNotComposed is the condition of a module's new fingerprint whose compositions or grant are not written: until
// they are, no want can name it.
const kindNotComposed = "fingerprint-not-composed"

func doneKeyObservation(module string) conditions.Observation {
	return conditions.Observation{Scope: conditions.ScopeBuild, ID: module, Token: kindNotComposed, Kind: kindNotComposed,
		Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "done keys",
		Summary: module + "'s new fingerprint is not composed or granted, so no want can name it: "}
}

// takeDoneKeyOnce is one take-in of a module's done key.
func takeDoneKeyOnce(ctx context.Context, open *stores, module string) error {
	keys, closeKeys, err := openBuilderKeys()
	if err != nil {
		return fmt.Errorf("the builder's keys cannot be read: %w", err)
	}
	defer closeKeys()
	said, err := liveDoneKeySteps(open, keys).take(ctx, module)
	if err == nil && said != "" {
		fmt.Printf("done keys: %s\n", said)
	}
	return err
}

// liveDoneKeySteps are the steps on the controller's own stores and the bus.
func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
	inv := open.inventory
	return doneKeySteps{
		held: func(ctx context.Context, module string) (bool, error) {
			_, err := inv.SourceOf(ctx, module)
			if errors.Is(err, inventory.ErrNoSuchModule) {
				return false, nil
			}
			return err == nil, err
		},
		trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) {
			return keys.ModuleAt(ctx, module, "")
		},
		registeredAt: func(ctx context.Context, module, commit string) (bool, error) {
			src, err := inv.SourceOf(ctx, module)
			if err != nil {
				return false, err
			}
			return sameCommitID(src.BuiltFrom, commit), nil
		},
		takeIn: func(ctx context.Context, module string, trunk sdkdelivery.Module) error {
			_, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: trunk, Ask: registrationAsk{Merge: trunk.Commit}})
			if errors.Is(err, inventory.ErrSuperseded) {
				return nil
			}
			return err
		},
		nodesOf: func(ctx context.Context, module string) ([]string, error) {
			entries, err := inv.Catalogued(ctx)
			if err != nil {
				return nil, err
			}
			for _, e := range entries {
				if e.Manifest.Module == module {
					return e.On, nil
				}
			}
			return nil, nil
		},
		busHolder: func(ctx context.Context) (string, error) {
			holders, err := seatHolders(ctx, inv)
			if err != nil {
				return "", err
			}
			return holders[catalogue.BrokerSeat].Node, nil
		},
		// The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the
		// grants once (issue 1945, composer.go).
		compose: func(ctx context.Context, nodes []string) error {
			return composerFor(open).compose(ctx, nodes, false)
		},
		grants: func(ctx context.Context) error {
			return composerFor(open).compose(ctx, nil, true)
		},
	}
}

// watchDoneKeys wakes the take-in of a module's done key for each module the builder or the indexer writes, until ctx
// ends: never on the watch's own goroutine.
func watchDoneKeys(ctx context.Context, open *stores) {
	keys, closeKeys, err := openBuilderKeys()
	if err != nil {
		fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
		return
	}
	if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
		inBackground(func() { takeDoneKey(ctx, open, module) })
	}); err != nil {
		closeKeys()
		fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
		return
	}
	go func() {
		<-ctx.Done()
		closeKeys()
	}()
	// And every module once now, after the watch is open so no write falls between the two.
	inBackground(func() {
		doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
	})
}

// doneKeysAtStart takes in the done key of every catalogued module once, one after another: the start's look the watch of
// new writes cannot give (it hears updates only). A done key written while no controller served, or whose take-in the
// controller this one replaced was stopped in, is taken here; without it the module stays in the node's own composition
// with no credential until its keys are written again (issue 1945, live on 2026-10-25 at 02:34). A take-in with nothing
// to do does nothing; a catalogue that cannot be read is said.
func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]string, error), take func(module string)) {
	names, err := modules(ctx)
	if err != nil {
		fmt.Printf("done keys: the modules are not looked at once at the start: %v\n", err)
		return
	}
	sort.Strings(names)
	// Side by side, a few at a time: their compositions meet in the composer's passes, each node composed once per pass
	// (issue 1945), so the start's look is not one module's composition after another.
	sem := make(chan struct{}, startTakeIns)
	var wg sync.WaitGroup
	for _, m := range names {
		if ctx.Err() != nil {
			break
		}
		sem <- struct{}{}
		wg.Add(1)
		go func() {
			defer func() { <-sem; wg.Done() }()
			take(m)
		}()
	}
	wg.Wait()
}

// startTakeIns is how many take-ins the start's look runs at once.
const startTakeIns = 16

// takeDoneKeyAtStart is one take-in of the start's look, under the module's lock; one that fails is handed to takeDoneKey,
// which tries again every doneKeyRetry, loudly, so a failing module never holds up the look at the others.
func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
	mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
	if !mu.(*sync.Mutex).TryLock() {
		// A take-in already follows the module: told to look again, it takes what the start would have (issue 1945).
		doneKeysLookAgain.Store(module, true)
		return
	}
	doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
	err := doneKeyTakeOnce(ctx, open, module)
	mu.(*sync.Mutex).Unlock()
	// Read after letting go, as every holder of a module's lock does: a wake during this look only marked the module
	// (the review of issue 1945).
	_, again := doneKeysLookAgain.LoadAndDelete(module)
	if err != nil || again {
		// A failure, a wait or a wake meanwhile is followed by takeDoneKey, which says it and looks again.
		inBackground(func() { takeDoneKey(ctx, open, module) })
	}
}

// catalogedModules are the names of every module the catalogue holds.
func catalogedModules(open *stores) func(context.Context) ([]string, error) {
	return func(ctx context.Context) ([]string, error) {
		entries, err := open.inventory.Catalogued(ctx)
		if err != nil {
			return nil, err
		}
		out := make([]string, 0, len(entries))
		for _, e := range entries {
			out = append(out, e.Manifest.Module)
		}
		return out, nil
	}
}

// doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test
// runs several meshes), each started at its first use and run for as long as the process serves.
var doneKeysComposers sync.Map

// composerFor is the composer of the stores, its passes composing on them and writing the grants on the bus.
func composerFor(open *stores) *composer {
	if c, ok := doneKeysComposers.Load(open); ok {
		return c.(*composer)
	}
	c := newComposer(func(ctx context.Context, nodes []string) map[string]error {
		_, failed := composeModulesByNode(ctx, open, nodes)
		return failed
	}, func(ctx context.Context) error {
		bus, err := aBus()
		if err != nil {
			return err
		}
		defer bus.Close()
		js, err := jetstream.New(bus.Conn())
		if err != nil {
			return err
		}
		return grantsIssued(ctx, open.inventory, js, "done keys: ")
	})
	if was, loaded := doneKeysComposers.LoadOrStore(open, c); loaded {
		return was.(*composer)
	}
	go c.run(context.Background())
	return c
}