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

package main

// The module's read tools (novox/hq issue 1562): what the module wrote to its own `steps` and `toolchains` states,
// answered as stored. A long step ends on its key in the steps state (ADR 0469 §2) and its result, such as the
// workspace a fetch made, is only there; a module reading its own bucket needs no grant.

import (
	"encoding/json"
	"fmt"
	"log"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// The read tools, named as the module is: build_node_ and the verb.
const (
	StepReadTool       = "build_node_step"
	StepsReadTool      = "build_node_steps"
	ToolchainsReadTool = "build_node_toolchains"
)

// StateRead reads one of the module's own states: its keys and a key's stored value (nil for a key with none).
type StateRead struct {
	Keys func() ([]string, error)
	Get  func(key string) (json.RawMessage, error)
}

// readOf is the read of a state the process keeps.
func readOf(s stdio.KeptState) StateRead {
	return StateRead{Keys: s.Keys, Get: func(key string) (json.RawMessage, error) {
		e, err := s.Get(key)
		if err != nil || e == nil {
			return nil, err
		}
		return e.Value, nil
	}}
}

// MaxStateReads is the most values one read tool reads from a state in a call, and stateReadParallel how many of them
// are in flight at once. Every value is a round trip from the tool through the runtime to the bus, so a call that read
// every key of a large state one after the other outlasted the call's 30 s (issue 1827: 1164 keys, 20 MB).
const (
	MaxStateReads     = 1000
	stateReadParallel = 16
)

// readValues reads the values of keys, stateReadParallel at a time, in the order of keys (nil for a key with none). It
// stops starting reads after the first error and returns it.
func readValues(keys []string, get func(string) (json.RawMessage, error)) ([]json.RawMessage, error) {
	out := make([]json.RawMessage, len(keys))
	var (
		wg   sync.WaitGroup
		mu   sync.Mutex
		fail error
		next = make(chan int)
	)
	for w := 0; w < stateReadParallel && w < len(keys); w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := range next {
				raw, err := get(keys[i])
				mu.Lock()
				if err != nil && fail == nil {
					fail = err
				}
				mu.Unlock()
				out[i] = raw
			}
		}()
	}
	for i := range keys {
		mu.Lock()
		stop := fail != nil
		mu.Unlock()
		if stop {
			break
		}
		next <- i
	}
	close(next)
	wg.Wait()
	return out, fail
}

// stepKey is the key of a step of this node: the id alone, or the whole key <node>.<id> as a step tool answered it.
func (m *Module) stepKey(id string) (string, error) {
	id = strings.TrimSpace(id)
	if node, rest, ok := strings.Cut(id, "."); ok {
		if node != m.Node {
			return "", fmt.Errorf("the step %q is of node %q, this module runs on %q", id, node, m.Node)
		}
		id = rest
	}
	return delivery.StepKey(m.Node, id)
}

// ReadStep is the stored value of this node's step id (state, reason, result).
func (m *Module) ReadStep(id string) (delivery.Step, error) {
	key, err := m.stepKey(id)
	if err != nil {
		return delivery.Step{}, err
	}
	if m.ReadSteps.Get == nil {
		return delivery.Step{}, fmt.Errorf("the steps state cannot be read here")
	}
	raw, err := m.ReadSteps.Get(key)
	if err != nil {
		return delivery.Step{}, err
	}
	if len(raw) == 0 {
		return delivery.Step{}, fmt.Errorf("no step %s in the steps state", key)
	}
	return delivery.DecodeStep(raw)
}

// StepLine is a step without its result: the list answers these.
type StepLine struct {
	ID      string `json:"id"`
	Tool    string `json:"tool"`
	State   string `json:"state"`
	Started string `json:"started"`
	Ended   string `json:"ended,omitempty"`
	Reason  string `json:"reason,omitempty"`
}

// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only
// the node's own keys (<node>.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's
// id is base-36 milliseconds and random digits, so its string order is its start order. cut says older steps were not
// read. The values still order the lines, since an ask step's id carries the ask time.
func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) {
	if limit <= 0 {
		limit = 10
	}
	if m.ReadSteps.Keys == nil || m.ReadSteps.Get == nil {
		return nil, false, fmt.Errorf("the steps state cannot be read here")
	}
	start := time.Now()
	keys, err := m.ReadSteps.Keys()
	if err != nil {
		return nil, false, err
	}
	own := keys[:0:0]
	for _, k := range keys {
		if strings.HasPrefix(k, m.Node+".") {
			if node, _, err := delivery.ParseStepKey(k); err == nil && node == m.Node {
				own = append(own, k)
			}
		}
	}
	sort.Sort(sort.Reverse(sort.StringSlice(own)))
	if n := min(max(limit*4, 100), MaxStateReads); len(own) > n {
		own, cut = own[:n], true
	}
	raws, err := readValues(own, m.ReadSteps.Get)
	log.Printf("steps read: %d of %d keys listed read in %v (cut %v)", len(own), len(keys), time.Since(start).Round(time.Millisecond), cut)
	if err != nil {
		return nil, cut, err
	}
	var steps []delivery.Step
	for _, raw := range raws {
		if len(raw) == 0 {
			continue
		}
		s, err := delivery.DecodeStep(raw)
		if err != nil {
			continue // a value this module cannot read is not one of its own steps
		}
		steps = append(steps, s)
	}
	sort.SliceStable(steps, func(i, j int) bool { return steps[i].Started.After(steps[j].Started) })
	if len(steps) > limit {
		steps = steps[:limit]
	}
	out := make([]StepLine, 0, len(steps))
	for _, s := range steps {
		l := StepLine{ID: s.ID, Tool: s.Tool, State: s.State, Started: s.Started.UTC().Format("2006-01-02T15:04:05Z"), Reason: s.Reason}
		if !s.Ended.IsZero() {
			l.Ended = s.Ended.UTC().Format("2006-01-02T15:04:05Z")
		}
		out = append(out, l)
	}
	return out, cut, nil
}

// ReadToolchains is every key of the toolchains state (<node>.<language>) with its stored value.
func (m *Module) ReadToolchains() (map[string]delivery.Toolchain, error) {
	if m.ReadToolchainsState.Keys == nil || m.ReadToolchainsState.Get == nil {
		return nil, fmt.Errorf("the toolchains state cannot be read here")
	}
	start := time.Now()
	keys, err := m.ReadToolchainsState.Keys()
	if err != nil {
		return nil, err
	}
	cut := len(keys) > MaxStateReads
	if cut {
		keys = keys[:MaxStateReads] // a node has a key per language: far fewer in practice
	}
	raws, err := readValues(keys, m.ReadToolchainsState.Get)
	log.Printf("toolchains read: %d keys read in %v (cut %v: keys past %d were not read)", len(keys), time.Since(start).Round(time.Millisecond), cut, MaxStateReads)
	if err != nil {
		return nil, err
	}
	out := map[string]delivery.Toolchain{}
	for i, raw := range raws {
		if len(raw) == 0 {
			continue
		}
		t, err := delivery.DecodeToolchain(raw)
		if err != nil {
			return nil, fmt.Errorf("%s: %w", keys[i], err)
		}
		out[keys[i]] = t
	}
	return out, nil
}

// ReadTools are the module's read tools. All are read-only.
func (m *Module) ReadTools() []stdio.Tool {
	return []stdio.Tool{
		{Name: StepReadTool,
			Description: "What one step of this machine ended with, as stored in the module's steps state: its tool, state " +
				"(running, passed, failed, refused), when it started and ended, why it failed or was refused, and its result " +
				"(a fetch's workspace id is there). id is the step id a step tool answered, or the whole key <node>.<id>. Read-only.",
			Input: map[string]any{"type": "object", "required": []string{"id"}, "properties": map[string]any{
				"id": map[string]any{"type": "string", "description": "the step id, or the key <node>.<id>"}}},
			Run: func(args map[string]any) (any, error) {
				id, _ := args["id"].(string)
				return m.ReadStep(id)
			}},
		{Name: StepsReadTool,
			Description: "This machine's latest steps from the module's steps state, newest first, each with its id, tool, state, " +
				"times and reason but not its result (read one with " + StepReadTool + "). Read-only.",
			Input: map[string]any{"type": "object", "properties": map[string]any{
				"limit": map[string]any{"type": "integer", "description": "how many, at most; 10 when left out"}}},
			Run: func(args map[string]any) (any, error) {
				steps, cut, err := m.RecentSteps(intArg(args["limit"]))
				ans := map[string]any{"node": m.Node, "steps": steps}
				if cut {
					ans["cut"] = true
					ans["note"] = fmt.Sprintf("this node has more steps stored than one call reads; older steps were not read (ask for a larger limit, at most %d reads)", MaxStateReads)
				}
				return ans, err
			}},
		{Name: ToolchainsReadTool,
			Description: "What the module's toolchains state holds: each key <node>.<language> with the version installed, the " +
				"recipe version and when the module saw them. Read-only.",
			Run: func(map[string]any) (any, error) {
				keys, err := m.ReadToolchains()
				return map[string]any{"node": m.Node, "toolchains": keys}, err
			}},
	}
}

// intArg is a tool argument that is a number, or a string of one; zero otherwise.
func intArg(v any) int {
	switch n := v.(type) {
	case float64:
		return int(n)
	case int:
		return n
	case string:
		i, _ := strconv.Atoi(n)
		return i
	}
	return 0
}