mesh-catalog/modules/mesh-build-node/cmd/mesh-build-node/module.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"net/http"
	"os"
	"os/exec"
	"sync"
	"time"

	buildrun "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildrun"
	sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/steprun"
	"git.novox.be/novox/mesh-sdk/go/forgekey"
	"git.novox.be/novox/mesh-sdk/go/health"
)

// CheckRound is how often the module reads the installed toolchain again, beside the check before every step and
// the one every health look makes.
const CheckRound = 10 * time.Minute

// Module is the Node build module's one running state: the node it runs on, its recipe, the toolchain guard, the
// toolchain directory and the states it writes (toolchains, steps).
type Module struct {
	// Callers says which seats the build seat's definition lists as callers of an accepted kind (steprun.SeatListCallers when
	// the module pulls through its runtime; nil in a test checks no named asker): design 47 G §543, A-163.
	Callers func(kind string) ([]string, bool)
	// SayEvent publishes an event of the node build seat on a seat subject (stdio.SeatPublish; nil says none, a test's): the
	// puller's ask-taken, step-progress and step-ended (design 47 G §496, §602).
	SayEvent func(subject string, body any, id string) error

	Node string
	// Recipe is the recipe in force. Reload, when set, reads it again at every check (the lock file the mesh
	// writes can change under a running module); a failure to read it is a refusal.
	Recipe Recipe
	Reload func() (Recipe, error)
	// Probe reads the installed Node version (ProbeNode).
	Probe func(ctx context.Context) (string, error)
	// Toolchain installs the toolchain directory; nil in a module that only guards.
	Toolchain *Toolchain
	// Work is the work directory, outside the home: a workspace is <Work>/<step id>/tree. Run runs the commands of a
	// step (a fake in a test). Clock is the time the steps are timed by.
	Work string
	Run  forgekey.Runner
	// RemoteBase is where repositories are fetched from, "ssh://mesh-forge/" when empty; FetchOptions are the options of
	// the SDK's fetch and merge. A test points them at a local repository (steprun.AllowFileRemote).
	RemoteBase   string
	FetchOptions []steprun.Option
	Clock        func() time.Time
	// Toolchains and Steps are the Put of the states of those names.
	Toolchains, Steps steprun.Put
	// ReadSteps and ReadToolchainsState read those two states back for the read tools (issue 1562).
	// Held is the puller whose running asks a step-end tool will not end (issue 1890); nil in a test.
	Held                           *steprun.Puller
	ReadSteps, ReadToolchainsState StateRead
	// Detected and ReadDetected are the Put and the read of the state detected-steps (design 47 G §326, §414).
	Detected     steprun.Put
	ReadDetected func(key string) (json.RawMessage, error)
	// ArtifactStoreFile is the binding of the provision artifact-store, read for a group run's package members (§465).
	ArtifactStoreFile string
	// ForgetDetected deletes one key of detected-steps; RecordAct writes an act in the hand-act log of the seat
	// mesh-hand-acts (§463, §480): a drop writes it first.
	ForgetDetected func(key string) error
	RecordAct      func(what, why string) error
	// Pauses is the Put of the pauses state, which keeps the node's pause key without an age; ReadPauses reads it back.
	Pauses     steprun.Put
	ReadPauses StateRead
	// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
	ForgetPause func(key string) error
	// WatchPauses follows the drops other holders write in the shared pauses state (stdio.State(...).Watch).
	WatchPauses func(key string, handle func(stdio.StateChange) error) error
	// SayToolchain says an event on a seat's subject (stdio.SeatPublish; nil says none, a test's); ToolchainProgram finds the
	// node program whose digest the toolchain's is (exec.LookPath when nil).
	SayToolchain     func(subject string, body any, id string) error
	ToolchainProgram func() (string, error)
	reporterOnce     sync.Once
	reporter         *toolchainReporter
	digestMu         sync.Mutex
	digestFor        string
	progDigest       string
	// Ctx is the process's context, which a step runs under (the tool call's is cut at 30 s, the step's is not).
	Ctx context.Context
	Say func(format string, a ...any)
	// HTTP reads and publishes to the package registry; a client of its own when nil. A test serves its own.
	HTTP *http.Client
	// Call asks a registry seat's verb over the bus, the publish of a done build (publish.go): stdio.Ask when nil; a
	// test's fake holder.
	Call steprun.SeatCall
	// Bubblewrap reads the installed bubblewrap's version (ProbeBubblewrap): a step is refused, and the module is
	// unhealthy, while it is missing or older than BubblewrapLeast. Nil probes nothing.
	Bubblewrap func(ctx context.Context) (string, error)

	// Floor refuses every step while the work or toolchain cache filesystem has less free than it (issue 1839); main
	// sets floorOf(node), and a zero Floor checks nothing.
	Floor steprun.Floor

	// Settings is the module's settings file, which holds the node's build capacity (sdkstep.CapacitySetting); Capacity,
	// when set, answers it instead (a test).
	Settings string
	Capacity func() int

	spacesOnce sync.Once
	workspaces *steprun.Workspaces
	mu         sync.Mutex // guards Recipe
	install    sync.Mutex // one install at a time
}

func (m *Module) recipe() Recipe {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.Recipe
}

// guard is the SDK guard for the version the probe just read: the module names no Node version (ADR 0477 §2), so the
// version wanted is the one installed and the guard refuses only when node cannot be run, writing the installed
// version to <node>.typescript of the toolchains state every time.
func (m *Module) guard(r Recipe, installed string, probeErr error) steprun.Guard {
	return steprun.Guard{Language: "typescript", Want: installed, Recipe: r.Version(), Node: m.Node, Put: m.reportingPut(),
		Probe: func(context.Context) (string, error) { return installed, probeErr }}
}

// Check reads the installed toolchain, writes <node>.typescript of the toolchains state and answers nil, or the
// refusal when node cannot be run (steprun.Guard.Check); another version than another node's is no refusal (ADR 0477
// §2). It runs at start, before every step, at every health look and every CheckRound.
func (m *Module) Check(ctx context.Context) error {
	_, err := m.check(ctx)
	return err
}

// check is Check that also answers the version it found installed.
func (m *Module) check(ctx context.Context) (string, error) {
	if m.Reload != nil {
		r, err := m.Reload()
		if err != nil {
			return "", &steprun.Refusal{Text: err.Error()}
		}
		m.mu.Lock()
		m.Recipe = r
		m.mu.Unlock()
	}
	installed, err := m.Probe(ctx)
	if err := m.guard(m.recipe(), installed, err).Check(ctx); err != nil {
		return installed, err
	}
	return installed, m.checkBubblewrap(ctx)
}

// Prepare is Check and then the toolchain directory of the recipe in force: what every step runs behind.
func (m *Module) Prepare(ctx context.Context) error {
	if err := m.Check(ctx); err != nil {
		return err
	}
	if m.Toolchain == nil {
		return nil
	}
	m.install.Lock()
	defer m.install.Unlock()
	return m.Toolchain.Ensure(ctx, m.recipe().Version())
}

// Start prepares once, as the module starts, and then every CheckRound until ctx ends.
func (m *Module) Start(ctx context.Context) {
	for {
		if err := m.Prepare(ctx); err != nil && m.Say != nil {
			m.Say("the toolchain: %v", err)
		}
		m.CleanWorkspaces(m.now())
		buildrun.EndStaleRound(m.stepScan(), m.Node, m.Say) // asks past the step bound, on the module's own round (issue 1890)
		select {
		case <-ctx.Done():
			return
		case <-time.After(CheckRound):
		}
	}
}

// Health is the module's health (the runtime-health tool, ADR 0366): unhealthy only while node cannot be run (or the lock cannot
// be read), saying why, so the controller raises the node's condition; healthy it names the installed version. It looks again each time it is
// asked.
func (m *Module) Health(ctx context.Context) health.Answer {
	installed, err := m.check(ctx)
	if err != nil {
		return health.Unhealthy(err.Error())
	}
	if err := m.room(); err != nil {
		return health.Unhealthy(err.Error())
	}
	note := ""
	if m.Held != nil {
		why, fault := m.Held.CapacityNote()
		if fault {
			return health.Unhealthy(why)
		}
		if why != "" {
			note = "; " + why // a capacity set to 0 is a choice, said so a starved builder is seen
		}
	}
	return health.Healthy(fmt.Sprintf("node %s installed, the version the module builds with (recipe %s)", installed, m.recipe().Version()) + note)
}

// Long is a tool that runs a long step (ADR 0469 §2): it answers at once with the step's id, and the step ends on
// its key in the steps state. The guard runs inside the step, before run, so a node whose node cannot be run ends every
// step refused with the cause named.
func (m *Module) Long(tool string, run func(ctx context.Context, args map[string]any) (any, error)) func(map[string]any) (any, error) {
	return func(args map[string]any) (any, error) {
		id, err := steprun.Start(m.Ctx, m.Steps, m.Node, tool, func(ctx context.Context) (any, error) {
			if err := m.Prepare(ctx); err != nil {
				return nil, err
			}
			if err := m.room(); err != nil {
				return nil, err
			}
			defer m.holdArg(args)()
			return run(ctx, args)
		})
		if err != nil {
			return nil, err
		}
		key, _ := sdkstep.StepKey(m.Node, id)
		return map[string]any{"step": id, "key": key, "state": sdkstep.StepRunning}, nil
	}
}

// Short is a tool that answers in its own call; the guard runs before it, and a refusal is the call's error.
func (m *Module) Short(run func(ctx context.Context, args map[string]any) (any, error)) func(map[string]any) (any, error) {
	return func(args map[string]any) (any, error) {
		if err := m.Prepare(m.Ctx); err != nil {
			return nil, err
		}
		if err := m.room(); err != nil {
			return nil, err
		}
		defer m.holdArg(args)()
		return run(m.Ctx, args)
	}
}

// Status is what the module reports of itself: the node, the recipe, the toolchain found and why it refuses, if it
// does.
func (m *Module) Status(ctx context.Context) map[string]any {
	err := m.Check(ctx)
	installed, _ := m.Probe(ctx)
	r := m.recipe()
	out := map[string]any{"node": m.Node, "recipe": r.Version(), "installed": installed, "refusing": err != nil,
		"cpu": m.cpuStatus()}
	if m.Toolchain != nil {
		out["toolchain-dir"] = m.Toolchain.Paths.Dir(r.Version())
		out["toolchain-installed"] = m.Toolchain.Present(r.Version())
	}
	if err != nil {
		out["why"] = err.Error()
	}
	return out
}

func (m *Module) now() time.Time {
	if m.Clock != nil {
		return m.Clock()
	}
	return time.Now()
}

// defaultPath is where a step looks for node, npm, git and ssh when this process was given no PATH.
const defaultPath = "/usr/local/bin:/usr/bin:/bin"

// baseEnv is the environment of a step's command beyond what the step adds: PATH, the account's own HOME (its ssh
// files and git configuration) and a fixed locale. A command is given these and its own words, never this process's
// environment, which holds the module's bus credentials (ADR 0466 §4).
func (m *Module) baseEnv() []string {
	return append([]string{"PATH=" + pathOrDefault(), "HOME=" + m.Toolchain.homeOrEmpty(), "LC_ALL=C"}, m.threadEnv()...)
}

func pathOrDefault() string {
	if path := os.Getenv("PATH"); path != "" {
		return path
	}
	return defaultPath
}

// reportingPut is the toolchains Put that then says toolchain-reported on the Node build seat's subject of this node when
// the toolchain changed (design 47 G §492): at the module's start and on each change of the key. A write that failed
// says nothing; an event that failed is said and does not refuse the step.
func (m *Module) reportingPut() steprun.Put {
	m.reporterOnce.Do(func() {
		m.reporter = &toolchainReporter{AskLanguage: sdkstep.AskNode, Node: m.Node, Publish: m.SayToolchain}
	})
	return putReporting(m.Toolchains, m.reporter, m.programDigest, m.Say)
}

// programDigest is the digest of the node program this node's node-toolchain provides, the `node` found on the path: read
// again only when the installed version changed (design 47 G §472; the lead's answer A-152).
func (m *Module) programDigest(version string) string {
	m.digestMu.Lock()
	defer m.digestMu.Unlock()
	if m.digestFor == version && m.progDigest != "" {
		return m.progDigest
	}
	find := m.ToolchainProgram
	if find == nil {
		find = func() (string, error) { return exec.LookPath("node") }
	}
	path, err := find()
	if err == nil {
		var d string
		if d, err = programFileDigest(path); err == nil {
			m.digestFor, m.progDigest = version, d
			return d
		}
	}
	if m.Say != nil {
		m.Say("the digest of the node toolchain program could not be read: %v", err)
	}
	return ""
}