mesh-controller/cmd/mesh-controller/new_modules.go

package main

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"

	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/novox/mesh-controller/internal/builderkeys"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/inventory"
)

// A new module is registered from its first done key (novox/hq ADR 0501 §1, §6; issue 1851, A5).
//
// Nothing asks a new module's first build: the builder builds every fingerprint the indexer writes, a new module's
// included (ADR 0429 §5). The controller registers a module the catalogue does not hold from the builder's done key of the
// fingerprint the indexer's trunk value names, under the same rules of who may register as before (ADR 0501 §3): from a
// repository the catalogue builds another module from, or on the operator's Approve of the merge the trunk value names
// (a first merge, kept with mesh-delivery's word), or at the terminal (ADR 0501 §4). It is assigned nowhere.
//
// **No timer** (ADR 0426): a write of the builder or the indexer wakes it for that module (builderkeys.Watch), a first
// merge's word wakes it for that repository, and a controller that starts looks at every module once (ADR 0501 §7: a
// manifest the controller before could not read is registered when the controller that reads it runs).

// newModulesMu keeps one take-in of new modules at a time in this process.
var newModulesMu sync.Mutex

// newModuleWait is a new module the controller could not register yet, and why: what the first-builds condition names.
type newModuleWait struct {
	Module, Repository, Commit, Why string
	// Word is the manifest word this controller does not read, when that is why (ADR 0501 §7).
	Word  string
	Since time.Time
}

// newModulesWaiting are the new modules waiting, by module, in this process: a restart looks at every module again.
var newModulesWaiting = struct {
	sync.Mutex
	by map[string]newModuleWait
}{by: map[string]newModuleWait{}}

// takeNewModules looks at the modules the indexer holds that the catalogue does not, all of them or those named, and
// registers each whose done key may register it. Each outcome is said once: a module still waiting for the same reason
// is not said again.
func takeNewModules(ctx context.Context, open *stores, only ...string) {
	newModulesMu.Lock()
	defer newModulesMu.Unlock()
	keys, closeKeys, err := openBuilderKeys()
	if err != nil {
		fmt.Printf("new modules: the builder's keys cannot be read: %v\n", err)
		return
	}
	defer closeKeys()
	for _, module := range takeNewModulesWith(ctx, open.inventory, keys, only...) {
		// An assignment waiting for it is made now (ADR 0501 §6; issue 1851, A10).
		for _, line := range settlePendingOnKey(ctx, open, module) {
			fmt.Println(line)
		}
	}
}

// takeNewModulesWith is takeNewModules on the buckets handed in; it answers the modules it registered.
func takeNewModulesWith(ctx context.Context, inv *inventory.Inventory, keys newModuleReads, only ...string) []string {
	entries, err := inv.Catalogued(ctx)
	if err != nil {
		fmt.Printf("new modules: the catalogue cannot be read: %v\n", err)
		// empty-on-error: said above; nothing is registered this time, and the next write of a key looks again
		return nil
	}
	held := map[string]bool{}
	for _, e := range entries {
		held[e.Manifest.Module] = true
	}
	values, unread, err := keys.Modules(ctx)
	if err != nil {
		fmt.Printf("new modules: the indexer's modules cannot be read: %v\n", err)
		// empty-on-error: said above; nothing is registered this time, and the next write of a key looks again
		return nil
	}
	for module, err := range unread {
		if !held[module] && (len(only) == 0 || slices.Contains(only, module)) {
			sayNewModuleWait(newModuleWait{Module: module, Why: "the indexer's value of it cannot be read: " + err.Error()})
		}
	}
	names := make([]string, 0, len(values))
	for module := range values {
		if !held[module] && (len(only) == 0 || slices.Contains(only, module)) {
			names = append(names, module)
		}
	}
	sort.Strings(names)
	var registered []string
	for _, module := range names {
		v := values[module]
		_, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: v, Ask: registrationAsk{Merge: v.Commit}})
		switch {
		case err == nil:
			forgetNewModuleWait(module)
			registered = append(registered, module)
			fmt.Printf("%s is new in %s at %.8s: registered from the builder's done key, assigned nowhere\n", module,
				v.Repository, v.Commit)
		case errors.Is(err, errKeyWaits):
			w := newModuleWait{Module: module, Repository: v.Repository, Commit: v.Commit, Why: err.Error()}
			if word := wordIn(err.Error()); word != "" {
				w.Word = word
			}
			sayNewModuleWait(w)
		default:
			sayNewModuleWait(newModuleWait{Module: module, Repository: v.Repository, Commit: v.Commit,
				Why: "not registered: " + err.Error()})
		}
	}
	return registered
}

// newModuleReads is what the take-in of new modules reads of the builder and the indexer.
type newModuleReads interface {
	keyReads
	Modules(ctx context.Context) (map[string]sdkdelivery.Module, map[string]error, error)
}

var _ newModuleReads = builderkeys.Reader{}

// wordIn is the manifest word a take-in's words say this controller does not read.
func wordIn(said string) string {
	_, rest, ok := strings.Cut(said, "uses the word \"")
	if !ok {
		return ""
	}
	word, _, _ := strings.Cut(rest, "\"")
	return word
}

// sayNewModuleWait keeps a new module waiting and says it, once for each reason.
func sayNewModuleWait(w newModuleWait) {
	newModulesWaiting.Lock()
	defer newModulesWaiting.Unlock()
	was, known := newModulesWaiting.by[w.Module]
	if known && was.Why == w.Why && was.Commit == w.Commit {
		return
	}
	w.Since = time.Now().UTC()
	if known && was.Commit == w.Commit {
		w.Since = was.Since
	}
	newModulesWaiting.by[w.Module] = w
	fmt.Printf("%s is new at %.8s and not registered yet: %s\n", w.Module, w.Commit, firstLine(w.Why))
}

func forgetNewModuleWait(module string) {
	newModulesWaiting.Lock()
	defer newModulesWaiting.Unlock()
	delete(newModulesWaiting.by, module)
}

// kindNewModuleWaits is what a new module waiting for this controller's reader of its manifest, or waiting past its
// bound, raises (ADR 0501 §6, §7: the first-builds condition reads the key, not a build request).
const kindNewModuleWaits = "new-module-waits"

// newModuleWaitBound is how long a new module waits for its key before it is said: the builder builds a fingerprint in
// minutes. Not a delay: its key's write registers it the moment it is done.
const newModuleWaitBound = 30 * time.Minute

// newModuleObservations names the new modules waiting: one whose manifest uses a word this controller does not read at
// once (nothing is lost, but the operator sees what waits for which controller), any other past the bound.
func newModuleObservations(now time.Time) []conditions.Observation {
	newModulesWaiting.Lock()
	defer newModulesWaiting.Unlock()
	var out []conditions.Observation
	for _, w := range newModulesWaiting.by {
		if w.Word == "" && now.Sub(w.Since) <= newModuleWaitBound {
			continue
		}
		head := fmt.Sprintf("%s is not registered: %s", w.Module, firstLine(w.Why))
		resolved := "its key is done and its take-in registers it"
		if w.Word != "" {
			head = fmt.Sprintf("%s waits for a controller that reads the manifest word %q", w.Module, w.Word)
			resolved = "a controller that reads the word runs and registers it from the same key, with no second build"
		}
		out = append(out, conditions.Observation{Scope: conditions.ScopeMerge,
			ID: fmt.Sprintf("%s_%s", w.Module, short(w.Commit)), Kind: kindNewModuleWaits, Severity: conditions.Warning,
			Summary:  head,
			Said:     fmt.Sprintf("%s at %s waits since %s: %s", w.Module, w.Commit, w.Since.Format(time.RFC3339), w.Why),
			Headline: fmt.Sprintf("The new module %s is not registered yet", w.Module),
			Explanation: fmt.Sprintf("The merge %.8s of %s adds %s. The mesh registers it from the builder's done key of its "+
				"fingerprint, and has not yet: %s", w.Commit, w.Repository, w.Module, w.Why),
			Resolved: resolved,
		})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
	return out
}

// watchNewModules wakes the take-in of new modules for each module the builder or the indexer writes, until ctx ends, and
// looks at every module once now. A bucket not there yet is said: the look at start and a first merge's word still work.
func watchNewModules(ctx context.Context, open *stores) {
	inBackground(func() {
		answerHeldMerges(ctx, open.inventory)
		takeNewModules(ctx, open)
	})
	keys, closeKeys, err := openBuilderKeys()
	if err != nil {
		fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
		return
	}
	if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
		wakeNewModule(ctx, open, module)
	}); err != nil {
		closeKeys()
		fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
		return
	}
	go func() {
		<-ctx.Done()
		closeKeys()
	}()
}

// wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
// draining.
func wakeNewModule(ctx context.Context, open *stores, module string) {
	inBackground(func() {
		entries, err := open.inventory.Catalogued(ctx)
		if err != nil {
			return
		}
		for _, e := range entries {
			if e.Manifest.Module == module {
				// Known: a plan waiting on its key is woken by the plans' own look.
				askToLook()
				return
			}
		}
		takeNewModules(ctx, open, module)
	})
}

// background is the work the take-ins of keys and of announcements start beside the event that woke them, so a test
// ends only once the work its mesh started has.
var background sync.WaitGroup

// inBackground runs work beside the caller, counted in background. A variable so a test of what a walk does around a
// take-in may run it in the caller's turn and need not wait for it.
var inBackground = func(work func()) {
	background.Add(1)
	go func() {
		defer background.Done()
		work()
	}()
}

// newModulesOf are the modules the indexer holds in a repository: what a first merge's word wakes.
func newModulesOf(ctx context.Context, repository string) ([]string, error) {
	keys, closeKeys, err := openBuilderKeys()
	if err != nil {
		return nil, err
	}
	defer closeKeys()
	values, _, err := keys.Modules(ctx)
	if err != nil {
		return nil, err
	}
	var out []string
	for module, v := range values {
		if strings.EqualFold(v.Repository, repository) {
			out = append(out, module)
		}
	}
	sort.Strings(out)
	return out, nil
}