mesh-controller/cmd/mesh-controller/data_integration_test.go

//go:build integration

package main

import (
	"context"
	"strings"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/testbus"
)

func TestAMovedPathIsNoShrinkAndAShrinkThereStillIs(t *testing.T) {
	inv := inventory.ForTest(t)
	ctx := t.Context()
	shelf := shelfFor(t, houseManifest)
	declared := []inventory.DeclaredData{{Module: "house", Item: "config", Class: "irreplaceable", Owned: true}}
	start := time.Now().Add(-6 * time.Hour)
	measure := func(at time.Time, path string, size int64) []conditions.Observation {
		t.Helper()
		if _, err := inv.RecordData(ctx, "home", declared, map[string]map[string]inventory.Measurement{"house": {
			"config": {Path: path, Size: bytesOf(size), MeasuredAt: when(at), LastWrite: when(at),
				LastBackup: when(at)}}}, "", at); err != nil {
			t.Fatal(err)
		}
		records, err := inv.Data(ctx)
		if err != nil {
			t.Fatal(err)
		}
		peaks, err := inv.DataPeaks(ctx, at.Add(-shrinkWindow))
		if err != nil {
			t.Fatal(err)
		}
		return dataFindings(records, peaks, shelf, nil, nil, at)
	}
	measure(start, "/home/operator/.claude", 94_700_000)
	if got := findingsByKind(measure(start.Add(10*time.Minute), "/home/agent/.claude", 490)); got[kindDataShrank].Kind != "" {
		t.Fatalf("a moved path raised a shrink: %+v", got[kindDataShrank])
	}
	measure(start.Add(2*time.Hour), "/home/agent/.claude", 300<<20)
	got := findingsByKind(measure(start.Add(4*time.Hour), "/home/agent/.claude", 1<<20))[kindDataShrank]
	if got.Severity != conditions.Urgent || !strings.Contains(got.Summary, "shrank") {
		t.Fatalf("a genuine shrink at the new path was not raised: %+v", got)
	}
}

func TestNatsUnassigningIrreplaceableDataRetiresItAndAnEmptyReplacementIsUrgent(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	inv := open.inventory
	if _, err := inv.SeedSeats(ctx, catalogue.DefaultSeats()); err != nil {
		t.Fatal(err)
	}
	// Built from a commit, so that a send carries them: a holder is asked only of what the machine was sent (issue 1109).
	built := inventory.Source{BuiltFrom: "c1"}
	if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "keeper", Version: "1",
		Claims: []catalogue.Claim{{Name: catalogue.BackupSeat, Scope: catalogue.ScopeNode, Serves: []string{"backed-up", "now", "restore"}}}}, built); err != nil {
		t.Fatal(err)
	}
	house, err := catalogue.ParseManifest([]byte(houseManifest))
	if err != nil {
		t.Fatal(err)
	}
	if err := inv.RegisterModule(ctx, house, built); err != nil {
		t.Fatal(err)
	}
	if _, err := assign(ctx, open, "laptop", "house"); err == nil {
		t.Fatal("irreplaceable data was assigned to a machine with nothing to back it up")
	}
	for _, node := range []string{"laptop", "anchor"} {
		if _, err := assign(ctx, open, node, "keeper"); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := assign(ctx, open, "laptop", "house"); err != nil {
		t.Fatal(err)
	}
	// Sent, and applied: a holder is asked only once it has been (novox/hq issue 275).
	for _, node := range []string{"laptop", "anchor"} {
		pushedAndApplied(t, open, node)
	}

	conn := onATestBus(t)
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	laptop, anchor := &fakeHolder{}, &fakeHolder{}
	laptop.serve(t, conn, "laptop")
	anchor.serve(t, conn, "anchor")
	now := time.Now()
	heard := &watchdogs{last: &signalFacts{now: now, machines: []machineFacts{
		{name: "laptop", lastHeard: now}, {name: "anchor", lastHeard: now}}}}
	d := &doctor{open: open, js: js, watchdogs: heard}
	var d13 probe
	for _, p := range probeRegistry {
		if p.ID == probeDataID {
			d13 = p
		}
	}
	run := func() []conditions.Observation {
		t.Helper()
		probing := context.WithValue(ctx, probeAsksKey{}, d13)
		obs, err := probeData(probing, d)
		if err != nil {
			t.Fatal(err)
		}
		return obs
	}

	laptop.set(measuredHouse("/var/lib/house/config", 2<<30, now))
	if obs := run(); len(obs) != 0 {
		t.Fatalf("a measured, backed-up item raised %+v", obs)
	}
	r, err := inv.DataOf(ctx, "laptop", "house", "config")
	if err != nil || r.Path != "/var/lib/house/config" || r.Size == nil || *r.Size != 2<<30 || r.LastBackup == nil {
		t.Fatalf("what the holder measured was not kept: %+v, %v", r, err)
	}

	said, err := unassign(ctx, open, "laptop", "house")
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(said, "house's config (/var/lib/house/config, 2.0 GB) stays where it is") {
		t.Fatalf("the unassignment does not say the data stays:\n%s", said)
	}
	laptop.set()
	run()
	r, err = inv.DataOf(ctx, "laptop", "house", "config")
	if err != nil || !r.Retired() || r.Path != "/var/lib/house/config" {
		t.Fatalf("unassigned, the irreplaceable item is not kept retired: %+v, %v", r, err)
	}
	records, _ := inv.Data(ctx)
	if rows := retiredData(records, time.Now()); len(rows) != 1 || rows[0].Path != "/var/lib/house/config" {
		t.Fatalf("cleanup list: %+v", rows)
	}

	// Assigned on the anchor, onto an empty directory.
	if _, err := assign(ctx, open, "anchor", "house"); err != nil {
		t.Fatal(err)
	}
	anchor.set(measuredHouse("/var/lib/house/config", 300<<10, time.Now()))
	obs := findingsByKind(run())
	o, ok := obs[kindEmptyReplacement]
	if !ok || o.Severity != conditions.Urgent || o.Machine != "anchor" || o.Also[0] != "laptop" {
		t.Fatalf("an empty replacement of a module's data was not urgent: %+v", obs)
	}
}