NexusGate/panel-go/rebuild_coalesce_test.go

package main

// Tests for round-3 item 23: binding-class rebuilds are coalesced in a
// FIXED 10-second window (anchored at the first pending change, never
// extended), manual deploy stays immediate, and a pending apply never
// resurrects a node removed inside the window.

import (
	"encoding/json"
	"testing"
	"time"
)

func coalesceFixture(t *testing.T) (*Store, *Orchestrator, *Node) {
	t.Helper()
	store, _ := newTestStore(t)
	d := store.Data
	d.Servers = []*Server{{ID: "srv_1", Name: "S1", PublicAddress: "entry.example.com",
		PortRangeStart: 20000, PortRangeEnd: 50000, Status: "online", LastSeenAt: StrPtr(NowISO())}}
	for i := 1; i <= 5; i++ {
		d.Customers = append(d.Customers, &Customer{
			ID: "cus_" + string(rune('0'+i)), Name: "C", Status: "active",
			Credentials: &Credentials{ClientID: "11111111-2222-3333-4444-00000000000" + string(rune('0'+i)), RelayPassword: "FRYXGBkaGxwdHh8gISIjJA==", ShortID: "abcd000" + string(rune('0'+i))},
		})
	}
	node := &Node{ID: "nod_1", Name: "N1", Protocol: "vless-reality-vision", ServerID: "srv_1", Topology: "direct",
		Status: "active", Port: 23456, ResourceID: "res_1", EntryReady: true,
		Reality: &RealityConfig{KeyID: "reality-nod_1", ServerName: "www.tesla.com", DestPort: 443}}
	d.Nodes = []*Node{node}
	return store, NewOrchestrator(store), node
}

func applyJobsFor(d *Data, nodeID string) []*Job {
	out := []*Job{}
	for _, job := range d.Jobs {
		if job.NodeID == nodeID && job.Action == "apply_resource" {
			out = append(out, job)
		}
	}
	return out
}

func resourceUsers(t *testing.T, job *Job) []any {
	t.Helper()
	var resource struct {
		Inbounds []struct {
			Users []any `json:"users"`
		} `json:"inbounds"`
	}
	if err := json.Unmarshal(job.Payload.Resource, &resource); err != nil {
		t.Fatal(err)
	}
	if len(resource.Inbounds) != 1 {
		t.Fatalf("inbounds = %d, want 1", len(resource.Inbounds))
	}
	return resource.Inbounds[0].Users
}

func TestCoalescedAssignsFlushOnceWithFinalState(t *testing.T) {
	store, orch, node := coalesceFixture(t)
	var firstMark string
	for i := 1; i <= 5; i++ {
		if _, err := orch.AssignCustomer(node.ID, "cus_"+string(rune('0'+i)), "admin"); err != nil {
			t.Fatal(err)
		}
		snap := store.Snapshot()
		n := snap.Nodes[0]
		if !n.RebuildPending {
			t.Fatal("assign must mark the node pending, not rebuild immediately")
		}
		if i == 1 {
			firstMark = n.RebuildPendingAt
		} else if n.RebuildPendingAt != firstMark {
			t.Fatalf("fixed window violated: mark moved from %s to %s", firstMark, n.RebuildPendingAt)
		}
		if got := applyJobsFor(snap, node.ID); len(got) != 0 {
			t.Fatalf("after assign %d: %d apply jobs, want 0 inside the window", i, len(got))
		}
	}
	// Flushing before the window elapses is a no-op.
	markTime, ok := ParseISO(firstMark)
	if !ok {
		t.Fatalf("unparseable pending mark %q", firstMark)
	}
	if n, err := orch.FlushPendingRebuilds(markTime.Add(5 * time.Second)); err != nil || n != 0 {
		t.Fatalf("early flush = %d, %v; want 0, nil", n, err)
	}
	// At window end exactly one apply ships, carrying all five users.
	flushed, err := orch.FlushPendingRebuilds(markTime.Add(RebuildCoalesceWindow + time.Second))
	if err != nil || flushed != 1 {
		t.Fatalf("flush = %d, %v; want 1, nil", flushed, err)
	}
	snap := store.Snapshot()
	jobs := applyJobsFor(snap, node.ID)
	if len(jobs) != 1 {
		t.Fatalf("apply jobs = %d, want exactly 1", len(jobs))
	}
	if users := resourceUsers(t, jobs[0]); len(users) != 5 {
		t.Fatalf("flushed resource has %d users, want all 5", len(users))
	}
	if snap.Nodes[0].RebuildPending {
		t.Fatal("pending mark must be consumed by the flush")
	}
	// A second flush ships nothing more.
	if n, _ := orch.FlushPendingRebuilds(markTime.Add(time.Hour)); n != 0 {
		t.Fatalf("second flush = %d, want 0", n)
	}
	if got := applyJobsFor(store.Snapshot(), node.ID); len(got) != 1 {
		t.Fatalf("apply jobs after second flush = %d, want 1", len(got))
	}
}

func TestFixedWindowDoesNotSlide(t *testing.T) {
	store, orch, node := coalesceFixture(t)
	if _, err := orch.AssignCustomer(node.ID, "cus_1", "admin"); err != nil {
		t.Fatal(err)
	}
	firstMark := store.Snapshot().Nodes[0].RebuildPendingAt
	markTime, _ := ParseISO(firstMark)
	// A change late in the window must not push the deadline out: the
	// flush still happens at first-mark + window.
	if _, err := orch.AssignCustomer(node.ID, "cus_2", "admin"); err != nil {
		t.Fatal(err)
	}
	flushed, err := orch.FlushPendingRebuilds(markTime.Add(RebuildCoalesceWindow + 100*time.Millisecond))
	if err != nil || flushed != 1 {
		t.Fatalf("flush at first-mark+window = %d, %v; want 1 (fixed window)", flushed, err)
	}
}

func TestPendingApplyDiscardedWhenNodeRemoved(t *testing.T) {
	store, orch, node := coalesceFixture(t)
	if _, err := orch.AssignCustomer(node.ID, "cus_1", "admin"); err != nil {
		t.Fatal(err)
	}
	markTime, _ := ParseISO(store.Snapshot().Nodes[0].RebuildPendingAt)
	if _, err := orch.RemoveNode(node.ID, "admin", true); err != nil {
		t.Fatal(err)
	}
	flushed, err := orch.FlushPendingRebuilds(markTime.Add(time.Hour))
	if err != nil || flushed != 0 {
		t.Fatalf("flush after removal = %d, %v; want 0", flushed, err)
	}
	if got := applyJobsFor(store.Snapshot(), node.ID); len(got) != 0 {
		t.Fatalf("removed node got %d apply jobs; a pending apply must never resurrect it", len(got))
	}
}

func TestManualDeployImmediateAndConsumesPending(t *testing.T) {
	store, orch, node := coalesceFixture(t)
	if _, err := orch.AssignCustomer(node.ID, "cus_1", "admin"); err != nil {
		t.Fatal(err)
	}
	if _, err := orch.DeployNode(node.ID, "admin"); err != nil {
		t.Fatal(err)
	}
	snap := store.Snapshot()
	if got := applyJobsFor(snap, node.ID); len(got) != 1 {
		t.Fatalf("manual deploy must apply immediately: %d jobs", len(got))
	}
	if snap.Nodes[0].RebuildPending {
		t.Fatal("manual deploy must consume the pending mark")
	}
	if n, _ := orch.FlushPendingRebuilds(time.Now().Add(time.Hour)); n != 0 {
		t.Fatalf("flush after manual deploy = %d, want 0", n)
	}
	if got := applyJobsFor(store.Snapshot(), node.ID); len(got) != 1 {
		t.Fatalf("apply jobs = %d, want still 1", len(got))
	}
}

func TestPendingMarkSurvivesStoreReload(t *testing.T) {
	store, orch, node := coalesceFixture(t)
	if _, err := orch.AssignCustomer(node.ID, "cus_1", "admin"); err != nil {
		t.Fatal(err)
	}
	// Simulate a panel restart inside the window: a fresh Store over the
	// same file must still see the pending mark and flush it when due —
	// the node can never be stranded un-deployed.
	reloaded := NewStore(store.File)
	if err := reloaded.Init(); err != nil {
		t.Fatal(err)
	}
	n := reloaded.Data.Nodes[0]
	if !n.RebuildPending {
		t.Fatal("pending mark must be persisted in the data file")
	}
	markTime, _ := ParseISO(n.RebuildPendingAt)
	orch2 := NewOrchestrator(reloaded)
	flushed, err := orch2.FlushPendingRebuilds(markTime.Add(RebuildCoalesceWindow + time.Second))
	if err != nil || flushed != 1 {
		t.Fatalf("post-restart flush = %d, %v; want 1", flushed, err)
	}
	if got := applyJobsFor(reloaded.Snapshot(), node.ID); len(got) != 1 {
		t.Fatalf("apply jobs after restart flush = %d, want 1", len(got))
	}
}