mesh-host/internal/agent/kept_test.go

package agent

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

	"git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/novox/mesh-host/internal/compositions"
)

// A recovery module that answers as scripted: each ask takes the next answer and says the events it carries after it.
type recoveryFake struct {
	mu      sync.Mutex
	asks    []string // verb and its input, in order
	answers []fakeAnswer
	points  []meshrecovery.Point
	events  chan Ending
	changes []catalogue.FormChange
	calls   []string // stop, ask, apply, catch-up, in order
	catchUp []meshrecovery.CatchUp
	expired chan struct{}
}

type fakeAnswer struct {
	accepted, busy, errWords string
	then                     []Ending
}

func (f *recoveryFake) next(verb string, in any) (string, string, string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.asks = append(f.asks, fmt.Sprintf("%s %s", verb, asJSON(in)))
	f.calls = append(f.calls, verb)
	if len(f.answers) == 0 {
		return "", "", "no answer scripted"
	}
	a := f.answers[0]
	f.answers = f.answers[1:]
	for _, e := range a.then {
		f.events <- e
	}
	return a.accepted, a.busy, a.errWords
}

func (f *recoveryFake) env() *Recovery {
	return &Recovery{
		FormChanges: func(context.Context, string, string, string) ([]catalogue.FormChange, error) { return f.changes, nil },
		SnapshotKept: func(_ context.Context, in meshrecovery.SnapshotKeptInput) (meshrecovery.SnapshotKeptAnswer, error) {
			a, b, e := f.next("snapshot-kept", in)
			return meshrecovery.SnapshotKeptAnswer{Accepted: a, Busy: b, Error: e}, nil
		},
		RestoreKept: func(_ context.Context, in meshrecovery.RestoreKeptInput) (meshrecovery.RestoreKeptAnswer, error) {
			a, b, e := f.next("restore-kept", in)
			return meshrecovery.RestoreKeptAnswer{Accepted: a, Busy: b, Error: e}, nil
		},
		KeptPoints: func(context.Context, meshrecovery.KeptPointsInput) (meshrecovery.KeptPointsAnswer, error) {
			f.mu.Lock()
			defer f.mu.Unlock()
			f.calls = append(f.calls, "kept-points")
			return meshrecovery.KeptPointsAnswer{Points: f.points}, nil
		},
		Hear: func(context.Context, string, string, string) (<-chan Ending, func(), error) {
			return f.events, func() {}, nil
		},
		Bound: func(context.Context, string, string, string, time.Time, time.Duration) (<-chan struct{}, func(), error) {
			return f.expired, func() {}, nil
		},
		CatchUp: func(_ context.Context, _ string, c meshrecovery.CatchUp) error {
			f.mu.Lock()
			defer f.mu.Unlock()
			f.calls = append(f.calls, "catch-up")
			f.catchUp = append(f.catchUp, c)
			return nil
		},
		Now: time.Now,
	}
}

func (f *recoveryFake) seen() []string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return slices.Clone(f.calls)
}

// moveDB runs a data module at fp "a", then wants it at fp "b", and answers the bed once the second want is settled or held.
func moveDB(t *testing.T, f *recoveryFake, ranFrom, to string, settle func(b *bed) bool, before ...func(b *bed, dig string)) *bed {
	t.Helper()
	b := newBed(t)
	inner, applies := b.env.Apply, 0
	b.env.Apply = func(ctx context.Context, c compositions.Composed, others []compositions.Composed) Result {
		if applies++; applies >= 2 {
			f.mu.Lock()
			f.calls = append(f.calls, "apply")
			f.mu.Unlock()
		}
		return inner(ctx, c, others)
	}
	f.events, f.expired = make(chan Ending, 16), make(chan struct{})
	b.env.Recovery = f.env()
	b.env.Stop = func(context.Context, string) Result {
		f.mu.Lock()
		f.calls = append(f.calls, "stop")
		f.mu.Unlock()
		return Result{}
	}
	stop := b.run()
	t.Cleanup(stop)
	fp1, fp2 := fpOf(ranFrom), fpOf(to)
	ck, craw, dig := composition(t, "db", fp1, true, file("db", "x", "1"))
	wk, wraw := wantFor(t, "db", fp1, dig, "")
	b.comps.put(ck, craw)
	b.wants.put(wk, wraw)
	eventually(t, "the first build is not applied", func() bool { return len(b.appliedList()) == 1 })
	b.idle()
	ck2, craw2, dig2 := composition(t, "db", fp2, true, file("db", "x", "2"))
	for _, h := range before {
		h(b, dig)
	}
	wk2, wraw2 := wantFor(t, "db", fp2, dig2, "")
	b.comps.put(ck2, craw2)
	b.wants.put(wk2, wraw2)
	eventually(t, "the move is neither applied nor held", func() bool { return settle(b) })
	return b
}

// testAMoveAcrossAFormChange is design 47 G §296, §297, §554, §560, §630 to §632 and A-98, A-108 at the node agent.
func testAMoveAcrossAFormChange(t *testing.T) {
	up := []catalogue.FormChange{{Name: "cache", Kind: catalogue.KeptKindState, Running: &catalogue.KeptLayout{Form: 1},
		Applied: &catalogue.KeptLayout{Form: 2, CatchUp: &catalogue.CatchUp{Tool: "rebuild", Arguments: map[string]any{"all": true}}}}}
	down := []catalogue.FormChange{{Name: "cache", Kind: catalogue.KeptKindState, Running: &catalogue.KeptLayout{Form: 2},
		Applied: &catalogue.KeptLayout{Form: 1}}}
	fp := func(c string) string { return fpOf(c) }
	done := func(act string) Ending { return Ending{Act: act, Outcome: meshrecovery.EndedEventOutcomeDone} }
	failed := func(act, why string) Ending {
		return Ending{Act: act, Outcome: meshrecovery.EndedEventOutcomeFailed, Error: why}
	}
	applied := func(b *bed) bool { return len(b.appliedList()) == 2 }
	heldHave := func(b *bed) bool { return b.haves.last("db").Phase == "held" }

	t.Run("a form that rises is taken as a restore point before the move, with the why and the catch-up, and then applied", func(t *testing.T) {
		f := &recoveryFake{changes: up, answers: []fakeAnswer{{accepted: "a1", then: []Ending{done("a1")}}}}
		moveDB(t, f, "a", "b", applied)
		if got := f.seen(); !slices.Equal(got, []string{"snapshot-kept", "apply"}) {
			t.Fatalf("calls %v", got)
		}
		want := fmt.Sprintf("snapshot-kept %s", asJSON(meshrecovery.SnapshotKeptInput{Module: "db", Kept: "cache", Form: 1, Fingerprint: fp("b"),
			Why: "form change of db's cache to " + fp("b"), CatchUp: &meshrecovery.CatchUp{Tool: "rebuild", Arguments: map[string]any{"all": true}}}))
		if f.asks[0] != want {
			t.Fatalf("asked %s, want %s", f.asks[0], want)
		}
	})

	t.Run("busy waits for the other act's ending and asks again, and a stopped act is asked again; another act's failure wakes nothing", func(t *testing.T) {
		f := &recoveryFake{changes: up, answers: []fakeAnswer{
			{busy: "x1", then: []Ending{failed("zz", "cap"), done("x1")}},
			{accepted: "a2", then: []Ending{failed("a2", "stopped")}},
			{accepted: "a3", then: []Ending{done("a3")}},
		}}
		moveDB(t, f, "a", "b", applied)
		if got := f.seen(); !slices.Equal(got, []string{"snapshot-kept", "snapshot-kept", "snapshot-kept", "apply"}) {
			t.Fatalf("calls %v", got)
		}
	})

	for _, row := range []struct{ error, held string }{
		{"cancelled", "cancelled"},
		{"cap", "cap"},
		{"restore point damaged", "restore point damaged"},
		{"disk full", "disk full"},
	} {
		t.Run("an ending failed with "+row.error+" holds the move and asks nothing again", func(t *testing.T) {
			f := &recoveryFake{changes: up, answers: []fakeAnswer{{accepted: "a1", then: []Ending{failed("a1", row.error)}}}}
			b := moveDB(t, f, "a", "b", heldHave)
			if h := b.haves.last("db"); h.FP != fp("a") || !contains(h.Detail, row.held) || h.Wait == nil || h.Wait.Reason == "" {
				t.Fatalf("the have %+v", h)
			}
			if got := f.seen(); len(got) != 1 {
				t.Fatalf("asked again: %v", got)
			}
			if n := len(b.appliedList()); n != 1 {
				t.Fatalf("applied %d", n)
			}
		})
	}

	t.Run("the bound passed says the wait in the have and the wait goes on to the ending", func(t *testing.T) {
		f := &recoveryFake{changes: up, answers: []fakeAnswer{{accepted: "a1"}}}
		b := moveDB(t, f, "a", "b", func(b *bed) bool {
			b.haves.mu.Lock()
			defer b.haves.mu.Unlock()
			for _, h := range b.haves.log {
				if h.Wait != nil && h.Wait.Words != "" && h.Phase == "backing-up" {
					return true
				}
			}
			return false
		}, func(*bed, string) { close(f.expired) })
		f.events <- done("a1")
		eventually(t, "the move went on after the ending", func() bool { return applied(b) })
	})

	t.Run("a revert stops the newer build, restores the point of the form it reverts from, applies, then calls its catch-up", func(t *testing.T) {
		f := &recoveryFake{changes: down,
			points: []meshrecovery.Point{
				{Module: "db", Kept: "cache", Form: 1, Fingerprint: fp("z"), Snapshot: "other", At: time.Now()},
				{Module: "db", Kept: "cache", Form: 1, Fingerprint: fp("a"), Snapshot: "snap-1", At: time.Now(), CatchUp: &meshrecovery.CatchUp{Tool: "rebuild"}}},
			answers: []fakeAnswer{{accepted: "r1", then: []Ending{done("r1")}}}}
		moveDB(t, f, "a", "b", func(b *bed) bool { return len(f.seen()) >= 5 })
		if got := f.seen(); !slices.Equal(got, []string{"stop", "kept-points", "restore-kept", "apply", "catch-up"}) {
			t.Fatalf("calls %v", got)
		}
		if f.catchUp[0].Tool != "rebuild" {
			t.Fatalf("catch-up %+v", f.catchUp)
		}
		want := fmt.Sprintf("restore-kept %s", asJSON(meshrecovery.RestoreKeptInput{Module: "db", Kept: "cache", Form: 1, Fingerprint: fp("a"),
			Snapshot: "snap-1", Why: "revert of db's cache from " + fp("a")}))
		if f.asks[0] != want {
			t.Fatalf("asked %s, want %s", f.asks[0], want)
		}
	})

	t.Run("a revert with no point of its form holds as a restore point missing", func(t *testing.T) {
		f := &recoveryFake{changes: down}
		b := moveDB(t, f, "a", "b", heldHave)
		if h := b.haves.last("db"); !contains(h.Detail, "restore point missing") {
			t.Fatalf("the have %+v", h)
		}
	})

	t.Run("a restart waits for the act its have names and asks nothing again", func(t *testing.T) {
		f := &recoveryFake{changes: up}
		b := moveDB(t, f, "a", "b", applied, func(b *bed, dig string) {
			at := time.Now().Add(-time.Minute)
			b.haves.mu.Lock()
			b.haves.stored["db"] = Have{Module: "db", FP: fp("a"), Phase: "backing-up", Wait: &nodeengine.Wait{
				On: "mesh-recovery.ended.db.cache", AskAt: &at, Act: "a1", Since: at}}
			b.haves.mu.Unlock()
			f.events <- done("a1")
		})
		_ = b
		if got := f.seen(); !slices.Equal(got, []string{"apply"}) {
			t.Fatalf("calls %v", got)
		}
	})
}

func contains(s, sub string) bool { return strings.Contains(s, sub) }

func asJSON(v any) string { raw, _ := json.Marshal(v); return string(raw) }