mesh-tools/node-tools/cmd/node-tools/main.go

// node-tools — the node's tool runtime, in Go (novox/hq ADR 0175, ADR 0193; to-be 38 WP4d).
//
// One process per machine: it connects to the bus on the node's credential, launches every assigned
// module's tools bundle and serves its tools and its seats' verbs; as the node-tools module — or
// wherever MESH_CONSOLE_LISTEN says — it is also the console, MCP over HTTP on loopback.
//
//	MESH_BROKER_FILE       the credential the mesh sealed to this machine for the runtime
//	MESH_LEDGERS_FILE      its own secret ledgers, each module's ledger credential; beside the broker file when unset
//	MESH_TOOL_MODULES      <module>=<entrypoint>,… the bundles to launch
//	MESH_TOOL_ENV          {"<module>": {"<word>": "<value>"}} what each is given (ADR 0192)
//	MESH_OPERATOR_ACCOUNT  whose machine this is, and MESH_OPERATOR_HOME their home
//	MESH_CONSOLE_LISTEN    where the console listens, overriding 127.0.0.1:4270
package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math/rand/v2"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/novox/mesh-tools/node-tools/internal/alive"
	"github.com/novox/mesh-tools/node-tools/internal/approval"
	"github.com/novox/mesh-tools/node-tools/internal/bus"
	"github.com/novox/mesh-tools/node-tools/internal/console"
	"github.com/novox/mesh-tools/node-tools/internal/firstview"
	"github.com/novox/mesh-tools/node-tools/internal/launch"
	"github.com/novox/mesh-tools/node-tools/internal/provide"
	"github.com/novox/mesh-tools/node-tools/internal/runtime"
)

// runtimeModule is the module that is the node's tool runtime; on its credential, serving is also
// the console.
const runtimeModule = "node-tools"

// consoleListen is where the console listens when nothing says otherwise.
const consoleListen = "127.0.0.1:4270"

func main() {
	log.SetFlags(0)
	if len(os.Args) > 1 && os.Args[1] != "serve" {
		fmt.Fprintf(os.Stderr, "node-tools: %q is not a mode; this runtime serves (novox/hq ADR 0193)\n", os.Args[1])
		os.Exit(2)
	}
	until := time.Now().Add(startWait)
	cred := credential(until)
	conn := connectPatiently(cred, until)
	// And the seat list it describes the seats by, read through the seat mesh-catalog (novox/hq ADR 0636 §4).
	seatList := startSeatList(conn)

	served, err := runtime.ServedModulesFrom(os.Getenv(runtime.ToolModules), cred.Module)
	if err != nil {
		log.Fatalf("node-tools: %v", err)
	}
	envs, err := runtime.TakeToolEnvs()
	if err != nil {
		log.Fatalf("node-tools: %v", err)
	}
	// Its own secret ledgers: each module's ledger connection, the only one its history is published on (novox/hq ADR
	// 0597 §9). Followed, so a credential minted or retired is taken with no restart.
	var opts []runtime.Option
	if file := runtime.LedgersFileOf(os.Getenv("MESH_BROKER_FILE")); file != "" {
		ledgers := runtime.FollowLedgerCredentials(file, bus.Dial, log.Printf)
		defer ledgers.Stop()
		opts = append(opts, runtime.WithLedgerWriters(ledgers.Writer))
	}
	// The indexer's and the assigner's states the console indexes the mesh by, and the gate reads its marks from, come to it as
	// first views, asked of the owners' own tools and followed on their views families, never read from their state (novox/hq
	// ADR 0615 §1; design 46 7-approval-mark §5).
	source, stopViews := firstview.Start(context.Background(), conn, log.Printf)
	defer stopViews()
	gate := approval.New(approval.Deps{Node: conn.Node(), Runner: cred.Module, Catalog: source.ApprovalCatalog(definitionsOf(seatList)),
		Findings: source.Findings(), Logf: log.Printf})
	opts = append(opts, runtime.WithGate(gate))
	stop, err := runtime.Run(conn, served, envs, log.Printf, opts...)
	if err != nil {
		log.Fatalf("node-tools: %v", err)
	}
	// And says it is there (novox/hq to-be 45 S11): the machine's runtime, on its machine's name. A
	// runtime on another credential — a module's own, in a test — is not the machine's.
	stopSaying := func() {}
	if cred.Module == runtimeModule && cred.Node != "" {
		stopSaying = alive.Keep(conn, cred.Node, alive.Every, log.Printf)
	}

	// As the machine's runtime it is also the provider of `mcp-endpoint`: it answers its consumers through the
	// SDK's helper wherever it runs, whether or not the module is handed (novox/hq ADR 0349 clause 12, issue 791).
	stopProviding := provide.Start(conn, cred.Module, cred.Node, time.Minute, log.Printf)

	listen := os.Getenv("MESH_CONSOLE_LISTEN")
	if listen == "" && cred.Module == runtimeModule {
		listen = consoleListen
	}
	var up *console.Listening
	if listen != "" {
		node := cred.Node
		if node == "" {
			node = "?"
		}
		who := node + "." + cred.Module
		up, err = console.Serve(console.NewSurface(conn, source, who), listen)
		if err != nil {
			log.Fatalf("node-tools: %v", err)
		}
		log.Printf("mesh console listening on http://%s/mcp as %s", up.Address, who)
	}

	signals := make(chan os.Signal, 1)
	signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
	<-signals
	// First of all: the service manager signalled every bundle with this process, and none is started again
	// (issue 1126).
	launch.ShutDown()
	stopSaying()
	stopProviding()
	stop()
	if up != nil {
		_ = up.Close()
	}
	conn.Close()
}

// startWait is how long a runtime waits, from its start, for its broker credential to be on the machine and for the bus
// to accept the account it names (issue 1496): the node-engine starts a module's runtime before the provisioner's client
// fills the credential, and the bus carries a new account only after it reloads its users. Past it the runtime gives up,
// loudly; inside it the wait is said once.
var startWait = 5 * time.Minute

// say is a line in the runtime's journal.
func say(format string, args ...any) { fmt.Fprintf(os.Stderr, "node-tools: "+format+"\n", args...) }

// credential reads the credential the mesh delivered, waiting for it within the start's bound (issue 1496); one that can
// never be a credential (no url) is a fault of configuration, said and final.
func credential(until time.Time) bus.Credential {
	file := os.Getenv("MESH_BROKER_FILE")
	if file == "" {
		if url := os.Getenv("MESH_BROKER_URL"); url != "" {
			return bus.Credential{URL: url, Module: runtimeModule}
		}
		say("set MESH_BROKER_FILE (a sealed credential) or MESH_BROKER_URL — there is no broker to reach")
		os.Exit(1)
	}
	cred, err := awaitCredential(file, until, say)
	if err != nil {
		say("%v", err)
		os.Exit(1)
	}
	return cred
}

// errNotCredential is a file that is there and readable and still no broker credential: waiting will not fix it.
var errNotCredential = errors.New("it carries no url — it is not a broker credential")

// readCredential reads the credential at file. A file not there yet, empty, or not yet whole JSON is one the node-engine
// has still to write: the error says so and waiting may fix it. errNotCredential is final.
func readCredential(file string) (bus.Credential, error) {
	raw, err := os.ReadFile(file)
	if err != nil {
		return bus.Credential{}, err
	}
	if len(bytes.TrimSpace(raw)) == 0 {
		return bus.Credential{}, errors.New("the file is empty")
	}
	var cred bus.Credential
	if err := json.Unmarshal(raw, &cred); err != nil {
		return bus.Credential{}, err
	}
	if cred.URL == "" {
		return bus.Credential{}, errNotCredential
	}
	return cred, nil
}

// awaitCredential is the credential at file, read at once and, while it is not there or not whole, again each time its
// directory changes (runtime.WatchFile, inotify), until the time until. The wait is said once; at until it ends with an
// error naming the file, how long it waited and what the last read found. A file that is not a credential ends it at once.
func awaitCredential(file string, until time.Time, say func(string, ...any)) (bus.Credential, error) {
	cred, err := readCredential(file)
	if err == nil {
		return cred, nil
	}
	if errors.Is(err, errNotCredential) {
		return bus.Credential{}, fmt.Errorf("%s: %w", file, err)
	}
	began := time.Now()
	say("waits for its broker credential at %s (%v): the node-engine writes it once the provisioner's client fills it; "+
		"it gives up at %s", file, err, until.Format("15:04:05"))
	changed := make(chan struct{}, 1)
	stop, werr := runtime.WatchFile(file, func() {
		select {
		case changed <- struct{}{}:
		default:
		}
	}, func(format string, args ...any) { say(format, args...) },
		fmt.Sprintf("the credential is read again every %s instead", rereadEvery))
	if werr != nil {
		say("cannot watch %s (%v): it is read again every %s instead", file, werr, rereadEvery)
	} else {
		defer stop()
	}
	for {
		left := time.Until(until)
		if left <= 0 {
			return bus.Credential{}, fmt.Errorf("cannot read the broker credential at %s: waited %s for it, and the last read "+
				"found: %v", file, time.Since(began).Round(time.Second), err)
		}
		// The watch wakes the read; the clock is only the backstop for a watch that could not be made or missed a change.
		select {
		case <-changed:
		case <-time.After(min(left, rereadEvery)):
		}
		if cred, err = readCredential(file); err == nil {
			say("read its broker credential at %s after %s", file, time.Since(began).Round(time.Second))
			return cred, nil
		}
		if errors.Is(err, errNotCredential) {
			return bus.Credential{}, fmt.Errorf("%s: %w", file, err)
		}
	}
}

// rereadEvery is the backstop of the wait for the credential, beside the watch of its directory.
var rereadEvery = 30 * time.Second

// connectPatiently retries while the bus is merely not reachable yet — the normal case at startup — and, until the start's
// bound, while the bus refuses the account: the bus carries a new account only after it reloads its users, which may come
// after the credential is on the machine (issue 1496, issue 1423). What waiting cannot fix (a pin that does not match, an
// address that is no URL) gives up at once (issue 058), and so does a refusal past the bound.
func connectPatiently(cred bus.Credential, until time.Time) *bus.Conn {
	conn, err := connectWithin(cred, until, bus.Connect, time.Sleep, say)
	if err != nil {
		say("%v", err)
		os.Exit(1)
	}
	return conn
}

// connectWithin is connectPatiently with its dial, its sleep and its words given, as a test gives them.
func connectWithin(cred bus.Credential, until time.Time, dial func(bus.Credential) (*bus.Conn, error), sleep func(time.Duration),
	say func(string, ...any)) (*bus.Conn, error) {
	refusedSaid := false
	for delay := 2 * time.Second; ; delay = min(delay*2, 30*time.Second) {
		conn, err := dial(cred)
		if err == nil {
			if refusedSaid {
				say("the bus accepts the account now")
			}
			return conn, nil
		}
		wait := delay + time.Duration(rand.IntN(1000))*time.Millisecond
		switch fatal := bus.Fatal(err); {
		case bus.Refused(err) && time.Now().Add(wait).Before(until):
			if !refusedSaid {
				say("the bus refuses this account (%v): it may not carry it yet, since it loads a new account by a reload; "+
					"retrying until %s", err, until.Format("15:04:05"))
				refusedSaid = true
			}
		case bus.Refused(err) && refusedSaid:
			return nil, fmt.Errorf("the bus still refuses this account (%v) at the end of the start's bound, %s; giving up", err,
				until.Format("15:04:05"))
		case fatal != "":
			return nil, fmt.Errorf("%s — waiting will not fix this; giving up", fatal)
		default:
			say("the broker is not reachable yet (%v); retrying in %ds", err, int(wait.Round(time.Second)/time.Second))
		}
		sleep(wait)
	}
}