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)) } }