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", What: "check ask, novox/x@abc"}) 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].What != "check ask, novox/x@abc" || 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) } }