mesh-catalog/modules/disk-load/cmd/disk-load/data_test.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"strings"
	"testing"
	"time"

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

// ADR 0577 clauses 2, 3 (and the disk-load check of clause 5): the items measured come from the node-engine's keys of
// this machine as the watch hands them over, and none is known (none taken for gone) before the watch caught up; a key
// that changes after the watch began has its module measured again, and one deleted is measured no more; an item whose
// measurement fails is a failed item of the round, which measures the rest, and the round's key says so.
func TestTheItemsComeFromTheEnginesKeysAndARoundSaysHowFarItGot(t *testing.T) {
	key := func(module string) string { k, _ := delivery.EngineDataKey("ace", module); return k }
	value := func(module string, items ...catalogue.PlacedDataItem) json.RawMessage {
		raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: "ace", Module: module, Digest: "sha256:" + module, Items: items})
		if err != nil {
			t.Fatal(err)
		}
		return raw
	}
	pg := catalogue.PlacedDataItem{Module: "pg", ID: "data", Class: "valuable", Path: "/srv/pg", Measure: catalogue.MeasureShallow}
	gone := catalogue.PlacedDataItem{Module: "pg", ID: "wal", Class: "valuable", Path: "/srv/missing", Measure: catalogue.MeasureShallow}
	web := catalogue.PlacedDataItem{Module: "web", ID: "uploads", Class: "valuable", Path: "/srv/web", Measure: catalogue.MeasureShallow}
	measured := map[string]int{}
	run := func(_ context.Context, name string, args ...string) (string, error) {
		switch {
		case name == "test" && args[1] == "/srv/missing":
			return "", errors.New("no such file")
		case name == "test":
			return "", nil
		case name == "bash" && strings.Contains(args[1], "-maxdepth 1"):
			measured[strings.Split(args[1], "'")[1]]++
			return "1760000000 2\n", nil
		}
		return "", os.ErrNotExist
	}
	now := time.Date(2026, 11, 1, 0, 30, 0, 0, time.UTC)
	readings, rounds := &memStore{m: map[string]json.RawMessage{}}, &memStore{m: map[string]json.RawMessage{}}
	w := &Watched{Node: "ace"}
	var said []string
	k := &Keeper{Node: "ace", Items: w.All, Store: readings, Rounds: rounds,
		Say:      func(f string, a ...any) { said = append(said, f) },
		Measurer: Measurer{Run: run, Now: func() time.Time { return now }}}
	round := func() Round {
		var r Round
		raw, _ := rounds.Get("ace")
		if err := json.Unmarshal(raw, &r); err != nil {
			t.Fatalf("the round's key: %v", err)
		}
		return r
	}

	// Before the watch caught up nothing is known, and a round says so in its key rather than retiring every reading.
	if _, _, err := k.Round(context.Background(), false); !errors.Is(err, ErrNotWatchedYet) {
		t.Fatalf("a round before the watch: %v", err)
	}
	if r := round(); r.Ended == nil || !strings.Contains(r.Error, "not watched yet") {
		t.Fatalf("the round before the watch is kept as %+v", r)
	}
	refusedAt := round().Started
	for _, step := range []struct {
		name    string
		change  stdio.StateChange
		module  string
		changed bool
		err     bool
	}{
		{"a current key", stdio.StateChange{Key: key("pg"), Op: stdio.OpPut, Value: value("pg", pg, gone), Current: true}, "pg", true, false},
		{"another current key", stdio.StateChange{Key: key("web"), Op: stdio.OpPut, Value: value("web", web), Current: true}, "web", true, false},
		{"the same items again", stdio.StateChange{Key: key("web"), Op: stdio.OpPut, Value: value("web", web)}, "web", false, false},
		{"another machine's items", stdio.StateChange{Key: key("x"), Op: stdio.OpPut, Value: json.RawMessage(
			`{"v":1,"node":"g14","module":"x","digest":"d","items":[]}`)}, "", false, true},
		{"a delete of a key never seen", stdio.StateChange{Key: key("y"), Op: stdio.OpDelete}, "", false, false},
	} {
		module, changed, err := w.Take(step.change)
		if module != step.module || changed != step.changed || (err != nil) != step.err {
			t.Errorf("%s: taken as %q, changed %v, %v", step.name, module, changed, err)
		}
	}
	w.Ready()

	// The round refused before the watch caught up runs once it has (Watched.Follow), not at the next hour: it measures
	// every item it was handed; the missing one fails, and the round goes on and says it.
	now = now.Add(time.Minute)
	if !k.RoundIfRefused(context.Background()) || k.RoundIfRefused(context.Background()) {
		t.Fatal("the round refused before the watch was not run once when the watch caught up")
	}
	r := round()
	if !r.Started.After(refusedAt) {
		t.Fatalf("the round's key still says the refused round: %+v", r)
	}
	if r.Ended == nil || r.Items != 3 || r.Measured != 3 || len(r.Failed) != 1 || r.Failed[0].Item != "pg.wal" ||
		!strings.Contains(r.Failed[0].Why, "does not exist") || measured["/srv/pg"] != 1 || measured["/srv/web"] != 1 {
		t.Fatalf("the round's key says %+v; measured %v", r, measured)
	}
	if !strings.Contains(strings.Join(said, "\n"), "the round ended") || !strings.Contains(strings.Join(said, "\n"), "starts") {
		t.Errorf("the journal said %v", said)
	}

	// A key that changes after the watch began has its module measured again, at once, by Work.
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go k.Work(ctx)
	moved := web
	moved.Path = "/srv/web2"
	if module, changed, err := w.Take(stdio.StateChange{Key: key("web"), Op: stdio.OpPut, Value: value("web", moved)}); err != nil || !changed {
		t.Fatalf("a changed key: %s %v %v", module, changed, err)
	}
	k.Due("web")
	waitFor(t, func() bool {
		raw := readingOf(readings, "ace.web.uploads")
		return raw != nil && raw.Path == "/srv/web2"
	})
	// A key deleted: its module's items are measured no more, and its reading is retired.
	if module, changed, _ := w.Take(stdio.StateChange{Key: key("web"), Op: stdio.OpDelete}); module != "web" || !changed {
		t.Fatalf("a deleted key was taken as %q, %v", module, changed)
	}
	k.Due("web")
	waitFor(t, func() bool { r := readingOf(readings, "ace.web.uploads"); return r != nil && !r.RetiredAt.IsZero() })
	before := measured["/srv/web2"]
	if _, _, err := k.Round(context.Background(), false); err != nil {
		t.Fatal(err)
	}
	if measured["/srv/web2"] != before || round().Items != 2 {
		t.Errorf("an item whose key was deleted was measured again: %d times, round %+v", measured["/srv/web2"]-before, round())
	}

	// A restart: oldie's key was deleted while the watch was down, so no delete arrives. Once the watch handed over its
	// keys, oldie's reading is retired and pg's kept; an empty set retires nothing (the engine has not written yet).
	_ = readings.Put("ace.oldie.x", health.Reading{V: health.ReadingVersion, Node: "ace",
		PlacedDataItem: catalogue.PlacedDataItem{Module: "oldie", ID: "x", Path: "/srv/oldie"}})
	for _, restart := range []struct {
		name    string
		keys    []stdio.StateChange
		retired bool
	}{
		{"no key yet", nil, false},
		{"pg's key alone", []stdio.StateChange{{Key: key("pg"), Op: stdio.OpPut, Value: value("pg", pg, gone), Current: true}}, true},
	} {
		w2 := &Watched{Node: "ace"}
		said = nil
		k2 := &Keeper{Node: "ace", Items: w2.All, Store: readings, Say: k.Say, Measurer: k.Measurer}
		for _, ch := range restart.keys {
			if _, _, err := w2.Take(ch); err != nil {
				t.Fatal(err)
			}
		}
		w2.Ready()
		k2.RetireUnkeyed(context.Background())
		oldie, pgData := readingOf(readings, "ace.oldie.x"), readingOf(readings, "ace.pg.data")
		if oldie == nil || pgData == nil || oldie.RetiredAt.IsZero() == restart.retired || !pgData.RetiredAt.IsZero() {
			t.Errorf("%s: oldie %+v, pg %+v; oldie retired: want %v, pg never", restart.name, oldie, pgData, restart.retired)
		}
		if !restart.retired && !strings.Contains(strings.Join(said, "\n"), "holds no key") {
			t.Errorf("%s: the skip was not said: %v", restart.name, said)
		}
	}
}

func readingOf(s *memStore, key string) *health.Reading {
	raw, _ := s.Get(key)
	if raw == nil {
		return nil
	}
	r, err := health.DecodeReading(raw)
	if err != nil {
		return nil
	}
	return &r
}

func waitFor(t *testing.T, done func() bool) {
	t.Helper()
	for end := time.Now().Add(5 * time.Second); time.Now().Before(end); time.Sleep(5 * time.Millisecond) {
		if done() {
			return
		}
	}
	t.Fatal("not done within 5 s")
}