mesh-delivery/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go

// Command mesh-delivery-assigner is delivery's assigner (novox/hq ADR 0429 §1 and §6, ADR 0446, ADR 0528, issue 1297): a
// process on its own bus credential that watches what the indexer, the builder, the controller, the node-engines and
// the watchman write, and writes what each node should run (`want.<node>.<module>`, by the six rules), the
// fingerprints it holds, every difference between a want and what the controller sends, and the composition check's
// verdict of each group the checker judges. In release 5 the wants are only compared: the controller still sends what
// each node runs (ADR 0429 §7). It holds no timer: every pass is started by a change it watched.
package main

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

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

	"git.novox.be/novox/mesh-sdk/go/announce"
	"git.novox.be/novox/mesh-sdk/go/asks"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
	"git.novox.be/novox/mesh-sdk/go/process"
	"git.novox.be/novox/mesh-sdk/go/provider"
)

func main() {
	if os.Getenv("MESH_BROKER_FILE") == "" {
		// Started by the tool runner, with no credential of its own: the put's tools (ADR 0546), served through the runtime,
		// since a process never serves (ADR 0436). The process the manifest runs names its credential.
		if err := serveTools(); err != nil {
			fmt.Fprintln(os.Stderr, "mesh-delivery-assigner:", err)
			os.Exit(1)
		}
		return
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	if err := run(ctx, os.Getenv, os.Stderr); err != nil {
		fmt.Fprintln(os.Stderr, "mesh-delivery-assigner:", err)
		os.Exit(1)
	}
}

// readBuckets are the buckets of other parts the assigner reads, each named in its manifest's reads (the manifest test
// holds the two lists equal), so the controller's broker grants exactly these.
var readBuckets = []string{
	provider.AssignmentsBucket,
	delivery.BuildsBucket,
	delivery.ControllerGrantsBucket,
	delivery.ControllerComposedBucket,
	delivery.AnswersBucket,
	delivery.BuilderBuildsBucket,
	delivery.CandidatesBucket,
	delivery.ManifestsBucket,
	delivery.ModulesBucket,
	delivery.PresenceBucket,
	delivery.HaveBucket,
	delivery.StepsBucket(goBuildModule),
}

// goBuildModule is the Go build module, which takes the assigner's compose asks on its seat's check stream and writes
// their steps in its steps state (novox/hq ADR 0482 clause 4).
const goBuildModule = "mesh-build-go"

// ownBuckets are the assigner's own states, as its manifest declares them.
var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket,
	delivery.AssignerPutsBucket, delivery.AssignerWantsBucket}

// keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view
// (and the node agent from release 6), its differences by the view, its compositions by the checker, its asks by the view
// and the teller, its puts by whoever follows one (ADR 0546).
var keptStates = []announce.State{
	{Name: "asks", Readable: true},
	{Name: "compositions", Readable: true},
	{Name: "differences", Readable: true},
	{Name: "puts", Readable: true},
	{Name: "wants", Readable: true},
}

// buckets is every bucket the assigner opened, by name.
type buckets map[string]jetstream.KeyValue

// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named with its grant (issue 1945),
// since a pass without it would write wants from half the mesh. The puts' bucket is the exception: whatever keeps it from opening (it
// does not exist yet, the credential is not granted it yet, the bus does not answer in time), the assigner writes wants
// without it and takes no put further until it starts again (ADR 0546).
func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any)) (buckets, error) {
	out := buckets{}
	for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
		kv, err := js.KeyValue(ctx, name)
		if name == delivery.AssignerPutsBucket && err != nil {
			logf("the bucket %s cannot be opened (%v): no put is taken further until the assigner starts again", name, err)
			continue
		}
		if err != nil {
			return nil, process.EssentialErr("assigner", name, "write a want from the whole mesh", err)
		}
		out[name] = kv
	}
	return out, nil
}

// run is the program: it returns when ctx ends or the bus cannot be used.
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
	logger := log.New(logOut, "mesh-delivery-assigner: ", log.LstdFlags)
	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
	if path == "" {
		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to read the bus with")
	}
	account, err := process.ReadAccount(path)
	if err != nil {
		return err
	}
	opts := []nats.Option{
		nats.Name(account.Node + "." + account.Module),
		nats.UserInfo(account.User, account.Password),
		nats.CustomInboxPrefix(account.InboxPrefix()),
		nats.MaxReconnects(-1),
		nats.RetryOnFailedConnect(true),
	}
	if strings.TrimSpace(account.Fingerprint) != "" {
		opts = append(opts, nats.Secure(process.PinnedToFingerprint(account.Fingerprint)))
	}
	conn, err := nats.Connect(account.URL, opts...)
	if err != nil {
		return fmt.Errorf("connecting to the bus: %w", err)
	}
	defer conn.Close()
	js, err := jetstream.New(conn)
	if err != nil {
		return err
	}
	kvs, err := open(ctx, js, logger.Printf)
	if err != nil {
		return err
	}
	// What the assigner keeps, said on the bus (ADR 0495 §2-§3): a reader is granted from it. Not fatal when it cannot
	// be said now: process.Conn says it again at every reconnect.
	said, err := process.Connect(account, nats.Name(account.Node+"."+account.Module+".keeps"), nats.RetryOnFailedConnect(true))
	if err != nil {
		return err
	}
	defer said.Close()
	if err := said.Keeps(keptStates...); err != nil {
		logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err)
	}
	logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
	return follow(ctx, js, kvs, sayVia(ctx, js), emitting{said}, logger)
}

// emitting is the process connection as the puts' caller: its calls, and its events with no options.
type emitting struct{ *process.Conn }

func (e emitting) Emit(eventType string, body any) error { return e.Conn.Emit(eventType, body) }

// watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and
// watched only for keys, so a manifest that arrives wakes a composition that waits for it.
var watched = []string{
	provider.AssignmentsBucket,
	delivery.ControllerGrantsBucket,
	delivery.ControllerComposedBucket,
	delivery.AnswersBucket,
	delivery.BuilderBuildsBucket,
	delivery.CandidatesBucket,
	delivery.ModulesBucket,
	delivery.PresenceBucket,
	delivery.HaveBucket,
	delivery.AssignerAsksBucket,
	delivery.AssignerCompositionsBucket,
	delivery.AssignerDifferencesBucket,
	delivery.AssignerWantsBucket,
}

// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends.
func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(asks.Condition) error, bus caller, logger *log.Logger) error {
	w := newWorld()
	dirty := make(chan struct{}, 1)
	wake := func() {
		select {
		case dirty <- struct{}{}:
		default:
		}
	}
	errs := make(chan error, len(watched)+2)
	settled := make(chan struct{}, len(watched)+2)
	watch := func(name string, opts ...jetstream.WatchOpt) error {
		kw, err := kvs[name].WatchAll(ctx, opts...)
		if err != nil {
			return fmt.Errorf("the bucket %s cannot be watched: %w", name, err)
		}
		// The puts' watch never stops the wants (ADR 0546): when it ends, the puts are no longer heard, said in the journal.
		optional := name == delivery.AssignerPutsBucket
		go func() {
			defer kw.Stop()
			for {
				select {
				case <-ctx.Done():
					errs <- nil
					return
				case e, ok := <-kw.Updates():
					if !ok {
						if optional {
							logger.Printf("the watch of %s ended: no put is taken further until the assigner starts again", name)
							return
						}
						errs <- fmt.Errorf("the watch of %s ended", name)
						return
					}
					if e == nil {
						if !optional { // the puts never gate the first pass, so never the wants (ADR 0546)
							settled <- struct{}{}
						}
						continue
					}
					var value []byte
					if e.Operation() == jetstream.KeyValuePut {
						value = e.Value()
					}
					w.put(name, e.Key(), value)
					wake()
				}
			}
		}()
		return nil
	}
	// The puts' watch is not waited for before the first pass: a watch of them that ends before it handed what stands only
	// stops the puts, never the wants.
	if kvs[delivery.AssignerPutsBucket] != nil {
		if err := watch(delivery.AssignerPutsBucket); err != nil {
			logger.Printf("%v: no put is taken further until it is opened again", err)
			delete(kvs, delivery.AssignerPutsBucket)
		}
	}
	// The puts' bucket is made by the controller from the manifest's states at its start (ADR 0201), and the assigner's
	// credential reads it once the bus's user list carries the grant: neither waits for the assigner. So while it is not open,
	// a pass started by a change opens it again, at most once a minute, and from then on the puts are watched and taken
	// further, without a restart.
	var triedPuts time.Time
	reopenPuts := func(p *passer) {
		if kvs[delivery.AssignerPutsBucket] != nil || time.Since(triedPuts) < PutsReopenEvery {
			return
		}
		triedPuts = time.Now()
		kv, err := js.KeyValue(ctx, delivery.AssignerPutsBucket)
		if err != nil {
			return
		}
		kvs[delivery.AssignerPutsBucket] = kv
		if err := watch(delivery.AssignerPutsBucket); err != nil {
			logger.Printf("%v: no put is taken further until it is opened again", err)
			delete(kvs, delivery.AssignerPutsBucket)
			return
		}
		if bus != nil {
			p.acts = newPutActs(bus, wake, logger.Printf)
		}
		logger.Printf("the bucket %s is open: puts are taken further", delivery.AssignerPutsBucket)
	}
	for _, name := range watched {
		var opts []jetstream.WatchOpt
		if name == delivery.HaveBucket {
			opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back
		}
		if err := watch(name, opts...); err != nil {
			return err
		}
	}
	if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil {
		return err
	}
	for range len(watched) + 1 {
		select {
		case <-ctx.Done():
			return nil
		case err := <-errs:
			return err
		case <-settled:
		}
	}
	p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, said: map[string]bool{}, answeredAt: map[string]uint64{}}
	p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf)
	if bus != nil && kvs[delivery.AssignerPutsBucket] != nil {
		p.acts = newPutActs(bus, wake, logger.Printf)
	}
	wake()
	for {
		select {
		case <-ctx.Done():
			return nil
		case err := <-errs:
			return err
		case <-settled:
		case <-dirty:
			reopenPuts(p)
			p.pass(ctx)
		}
	}
}

// PutsReopenEvery is how often, at most, a pass opens the puts' bucket again while it is not open.
var PutsReopenEvery = time.Minute

// sayVia publishes a condition on the operator-channel seat's one subject for conditions, under the assigner's own name
// (ADR 0437 clause 9).
func sayVia(ctx context.Context, js jetstream.JetStream) func(asks.Condition) error {
	return func(c asks.Condition) error {
		subject, err := asks.ConditionSubject(delivery.AssignerModule)
		if err != nil {
			return err
		}
		data, err := asks.EncodeCondition(c)
		if err != nil {
			return err
		}
		_, err = js.Publish(ctx, subject, data)
		return err
	}
}