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

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"slices"
	"strings"
	"sync/atomic"
	"testing"
	"time"

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

// fakeState is a state as stored: key to raw value.
type fakeState map[string]json.RawMessage

func (f fakeState) read() StateRead {
	return StateRead{
		Keys: func() ([]string, error) {
			var k []string
			for key := range f {
				k = append(k, key)
			}
			return k, nil
		},
		Get: func(key string) (json.RawMessage, error) { return f[key], nil },
	}
}

func putStep(t *testing.T, f fakeState, s delivery.Step) {
	t.Helper()
	b, err := delivery.EncodeStep(s)
	if err != nil {
		t.Fatal(err)
	}
	key, _ := delivery.StepKey(s.Node, s.ID)
	f[key] = b
}

func readModule(steps, toolchains fakeState) *Module {
	m := moduleWith(newStates(), "v22.1.0")
	m.ReadSteps, m.ReadToolchainsState = steps.read(), toolchains.read()
	return m
}

func tool(t *testing.T, m *Module, name string) func(map[string]any) (any, error) {
	t.Helper()
	for _, tl := range m.Tools(m.StepTools()) {
		if tl.Name == name {
			return tl.Run
		}
	}
	t.Fatalf("no tool %s", name)
	return nil
}

// 1562/AC1: build_node_step answers the stored value of a step: state, reason and result (a fetch's workspace id).
func TestStepToolAnswersTheStoredStep(t *testing.T) {
	steps := fakeState{}
	t0 := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	putStep(t, steps, delivery.Step{ID: "abc1", Tool: "build_node_fetch", Node: "ace", State: delivery.StepPassed, Started: t0, Ended: t0.Add(time.Second),
		Result: json.RawMessage(`{"workspace":"w42"}`)})
	putStep(t, steps, delivery.Step{ID: "abc2", Tool: "build_node_test", Node: "ace", State: delivery.StepRefused, Started: t0, Reason: "node cannot be run"})
	m := readModule(steps, fakeState{})
	for _, id := range []string{"abc1", "ace.abc1"} {
		out, err := tool(t, m, StepReadTool)(map[string]any{"id": id})
		if err != nil {
			t.Fatal(err)
		}
		st := out.(delivery.Step)
		if st.State != delivery.StepPassed || string(st.Result) != `{"workspace":"w42"}` || st.Tool != "build_node_fetch" {
			t.Errorf("id %s: step = %+v", id, st)
		}
	}
	out, _ := tool(t, m, StepReadTool)(map[string]any{"id": "abc2"})
	if st := out.(delivery.Step); st.State != delivery.StepRefused || st.Reason != "node cannot be run" {
		t.Errorf("refused step = %+v", st)
	}
}

func TestStepToolRefusesAnUnknownOrForeignOrBadID(t *testing.T) {
	m := readModule(fakeState{}, fakeState{})
	for id, want := range map[string]string{"nothere": "no step ace.nothere", "g14.abc1": "of node", "Bad Id": "lower-case"} {
		_, err := tool(t, m, StepReadTool)(map[string]any{"id": id})
		if err == nil || !strings.Contains(err.Error(), want) {
			t.Errorf("id %q: err = %v, want it to say %q", id, err, want)
		}
	}
}

// 1562: build_node_steps lists this node's steps newest first without results, at most limit.
func TestStepsToolListsRecentStepsOfThisNode(t *testing.T) {
	steps := fakeState{}
	t0 := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	putStep(t, steps, delivery.Step{ID: "a", Tool: "build_node_fetch", Node: "ace", State: delivery.StepPassed, Started: t0, Ended: t0, Result: json.RawMessage(`1`)})
	putStep(t, steps, delivery.Step{ID: "b", Tool: "build_node_test", Node: "ace", State: delivery.StepRunning, Started: t0.Add(time.Minute)})
	putStep(t, steps, delivery.Step{ID: "c", Tool: "build_node_build", Node: "ace", State: delivery.StepFailed, Started: t0.Add(2 * time.Minute), Reason: "boom"})
	putStep(t, steps, delivery.Step{ID: "d", Tool: "build_node_build", Node: "g14", State: delivery.StepPassed, Started: t0.Add(time.Hour)})
	m := readModule(steps, fakeState{})
	out, err := tool(t, m, StepsReadTool)(map[string]any{})
	if err != nil {
		t.Fatal(err)
	}
	lines := out.(map[string]any)["steps"].([]StepLine)
	var ids []string
	for _, l := range lines {
		ids = append(ids, l.ID)
	}
	if !slices.Equal(ids, []string{"c", "b", "a"}) {
		t.Errorf("ids = %v, want c b a", ids)
	}
	if lines[0].Reason != "boom" || lines[0].State != delivery.StepFailed {
		t.Errorf("first = %+v", lines[0])
	}
	out, _ = tool(t, m, StepsReadTool)(map[string]any{"limit": float64(1)})
	if n := len(out.(map[string]any)["steps"].([]StepLine)); n != 1 {
		t.Errorf("limit 1 answered %d steps", n)
	}
}

// 1562 (1483/AC5): build_node_toolchains answers each toolchains key as stored.
func TestToolchainsToolAnswersTheStoredKeys(t *testing.T) {
	tcs := fakeState{}
	b, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: "v22.1.0", Recipe: "1", At: time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)})
	if err != nil {
		t.Fatal(err)
	}
	tcs["ace.typescript"] = b
	m := readModule(fakeState{}, tcs)
	out, err := tool(t, m, ToolchainsReadTool)(map[string]any{})
	if err != nil {
		t.Fatal(err)
	}
	got := out.(map[string]any)["toolchains"].(map[string]delivery.Toolchain)
	if got["ace.typescript"].Version != "v22.1.0" || got["ace.typescript"].Recipe != "1" {
		t.Errorf("toolchains = %+v", got)
	}
}

func TestReadToolsAreInTheManifestAndNeedNoToolchain(t *testing.T) {
	m := readModule(fakeState{}, fakeState{})
	m.Probe = func(context.Context) (string, error) { return "", context.DeadlineExceeded }
	// a node whose go cannot be run can still be asked what its steps said
	if _, err := tool(t, m, StepsReadTool)(map[string]any{}); err != nil {
		t.Errorf("the read refused: %v", err)
	}
}

// 1827: on the live shape (4 nodes, 600 steps each; a steps state of 1164 keys / 20 MB timed out at 30 s) the steps tool
// lists, then reads only this node's keys, in parallel, at most MaxStateReads of them, and says when it cut.
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
	t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
	for _, row := range []struct {
		perNode, limit int
		reads          int64
		steps          int
		cut            bool
	}{
		{20, 0, 20, 10, false},
		{600, 0, 100, 10, true},
		{1500, 300, MaxStateReads, 300, true},
	} {
		var reads, inFlight, maxFlight atomic.Int64
		stored := map[string][]byte{}
		for _, n := range []string{"ace", "g14", "novox", "shanks"} {
			for i := 0; i < row.perNode; i++ {
				b, _ := delivery.EncodeStep(delivery.Step{ID: fmt.Sprintf("s%04d", i), Tool: "t", Node: n, State: delivery.StepPassed, Started: t0.Add(time.Duration(i) * time.Second)})
				stored[n+".s"+fmt.Sprintf("%04d", i)] = b
			}
		}
		m := &Module{Node: "shanks"}
		m.ReadSteps = StateRead{
			Keys: func() ([]string, error) {
				var k []string
				for key := range stored {
					k = append(k, key)
				}
				return k, nil
			},
			Get: func(key string) (json.RawMessage, error) {
				reads.Add(1)
				now := inFlight.Add(1)
				for o := maxFlight.Load(); now > o && !maxFlight.CompareAndSwap(o, now); o = maxFlight.Load() {
				}
				time.Sleep(time.Millisecond)
				inFlight.Add(-1)
				if !strings.HasPrefix(key, "shanks.") {
					t.Errorf("read %s, a key of another node", key)
				}
				return stored[key], nil
			},
		}
		start := time.Now()
		lines, cut, err := m.RecentSteps(row.limit)
		if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads {
			t.Errorf("%d a node: %d steps, cut %v, %d reads, err %v; want %d, %v, %d", row.perNode, len(lines), cut, reads.Load(), err, row.steps, row.cut, row.reads)
		}
		if row.cut && len(lines) > 0 && lines[0].ID != fmt.Sprintf("s%04d", row.perNode-1) {
			t.Errorf("%d a node: newest answered %s", row.perNode, lines[0].ID)
		}
		if maxFlight.Load() < 2 {
			t.Errorf("%d a node: values read one at a time", row.perNode)
		}
		t.Logf("%d a node: %d reads in %v (one after the other: %v)", row.perNode, reads.Load(), time.Since(start), time.Duration(reads.Load())*time.Millisecond)
	}
}