diff --git a/panel-go/handlers.go b/panel-go/handlers.go index 6aa8a3e..5a85c0d 100644 --- a/panel-go/handlers.go +++ b/panel-go/handlers.go @@ -2319,9 +2319,10 @@ func (a *App) handleAdminApi(w http.ResponseWriter, r *http.Request, pathname st continue } rebuilt[node.ID] = true - if err := RebuildNode(d, node, false); err != nil { - return nil, err - } + // Credential rotation is a binding-class change: it goes + // through the coalesced window (item 23), not an + // immediate per-node restart. + markRebuildPending(node, false) } Audit(d, actor, "regenerate_credentials", customer.ID, map[string]any{"nodesRebuilt": len(rebuilt)}) view = customerCredentialsView(d, customer) @@ -2620,6 +2621,9 @@ func (a *App) handleAdminApi(w http.ResponseWriter, r *http.Request, pathname st item.ExitReady = false } if item.Status == "active" || item.Status == "deploying" || item.Status == "failed" || item.Status == "degraded" { + // A node edit ships immediately and consumes any pending + // coalesced mark (item 23). + clearRebuildPending(item) if err := RebuildNode(d, item, true); err != nil { return nil, err } diff --git a/panel-go/main.go b/panel-go/main.go index 24fd982..92de4d0 100644 --- a/panel-go/main.go +++ b/panel-go/main.go @@ -407,6 +407,17 @@ func run() error { app.housekeeping() } }() + // Coalesced rebuild flusher (round-3 item 23): applies pending + // binding-class rebuilds once their fixed window elapses. Pending + // marks are persisted, so this also drains whatever a previous + // process left mid-window. + go func() { + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for range ticker.C { + _, _ = app.orchestrator.FlushPendingRebuilds(time.Now()) + } + }() fmt.Printf("NexusGate v%s listening on http://%s:%d\n", VERSION, host, port) return server.ListenAndServe() } diff --git a/panel-go/orchestrator.go b/panel-go/orchestrator.go index dc52b22..5ebcc79 100644 --- a/panel-go/orchestrator.go +++ b/panel-go/orchestrator.go @@ -388,6 +388,75 @@ func DeriveNodeStatus(node *Node) { node.Error = nil } +// ---------- coalesced rebuilds (round-3 item 23) ---------- + +// RebuildCoalesceWindow is the FIXED coalescing window for binding-class +// rebuilds: it starts at a node's first pending change and is never +// extended by later changes inside it, so a burst of assign/unassign +// operations can never postpone the dispatch indefinitely. When the +// window elapses the flusher applies exactly one rebuild with the +// node's final state. +const RebuildCoalesceWindow = 10 * time.Second + +// markRebuildPending records a binding-class change for the coalesced +// flusher instead of rebuilding (and restarting the node's sing-box) +// immediately. The window is anchored at the first pending change: +// re-marking inside the window only widens the exit flag. +func markRebuildPending(node *Node, includeExit bool) { + if !node.RebuildPending { + node.RebuildPending = true + node.RebuildPendingAt = NowISO() + } + if includeExit { + node.RebuildPendingExit = true + } + node.UpdatedAt = NowISO() +} + +// clearRebuildPending drops a pending coalesced rebuild — used when an +// immediate action (manual deploy, repair, node edit, removal, server +// reconcile) already ships the node's current state or ends its life. +func clearRebuildPending(node *Node) { + node.RebuildPending = false + node.RebuildPendingAt = "" + node.RebuildPendingExit = false +} + +// FlushPendingRebuilds applies one rebuild per node whose fixed window +// has elapsed, with the node's state as of now. Pending marks on nodes +// that were deleted, are being removed, or fell back to draft are +// discarded — a pending apply must never resurrect a node. Called once +// per second from the panel's background loop; because the mark is +// persisted in the data file, a panel restart inside the window simply +// means the first post-boot pass flushes whatever is due. +func (o *Orchestrator) FlushPendingRebuilds(now time.Time) (int, error) { + flushed := 0 + err := o.tx(func(d *Data) error { + for _, node := range d.Nodes { + if !node.RebuildPending { + continue + } + at, ok := ParseISO(node.RebuildPendingAt) + if !ok || now.Sub(at) < RebuildCoalesceWindow { + continue + } + includeExit := node.RebuildPendingExit + clearRebuildPending(node) + if node.PendingDelete || node.Status == "removing" || node.Status == "draft" { + continue + } + if err := RebuildNode(d, node, includeExit); err != nil { + node.Error = StrPtr(err.Error()) + Audit(d, "system", "rebuild_flush_failed", node.ID, map[string]any{"error": err.Error()}) + continue + } + flushed++ + } + return nil + }) + return flushed, err +} + // ---------- customer suspend / resume ---------- func SuspendCustomerResources(d *Data, customer *Customer, actor string) (int, error) { @@ -409,9 +478,7 @@ func SuspendCustomerResources(d *Data, customer *Customer, actor string) (int, e for _, nodeID := range affected { node := orchFindNode(d, nodeID) if node != nil && node.Status != "draft" && node.Status != "removing" { - if err := RebuildNode(d, node, false); err != nil { - return len(affected), err - } + markRebuildPending(node, false) } } Audit(d, actor, "suspend_customer_resources", customer.ID, map[string]any{"nodes": len(affected)}) @@ -462,9 +529,7 @@ func ResumeCustomerResources(d *Data, customer *Customer, actor string) (int, er } } includeExit := forwardNode(node) && !node.ExitReady - if err := RebuildNode(d, node, includeExit); err != nil { - return len(affected), err - } + markRebuildPending(node, includeExit) } Audit(d, actor, "resume_customer_resources", customer.ID, map[string]any{"nodes": len(affected)}) return len(affected), nil @@ -815,6 +880,9 @@ func reconcileServerData(d *Data, serverID, actor string, opts ReconcileOptions) } QueueJob(d, serverID, node.ID, "apply_resource", payloadWithResource(resource, "node-exit", nil)) } + // The reconcile just shipped the node's current state; a + // coalesced pending mark for it is satisfied. + clearRebuildPending(node) node.Status = "deploying" } node.UpdatedAt = NowISO() @@ -898,6 +966,9 @@ func (o *Orchestrator) DeployNode(nodeID, actor string) (*Node, error) { if err := ValidateNodeDeploy(d, node); err != nil { return err } + // Manual deploy is immediate (item 23 constraint 1): it ships the + // current state now and consumes any coalesced pending mark. + clearRebuildPending(node) if err := RebuildNode(d, node, true); err != nil { return err } @@ -919,6 +990,9 @@ func (o *Orchestrator) RemoveNode(nodeID, actor string, deleteRecord bool) (bool } node.PendingDelete = deleteRecord node.AutoRemoving = BoolPtr(false) + // A pending coalesced apply must never resurrect a node that is + // being removed (item 23 constraint 2). + clearRebuildPending(node) if node.Status != "draft" { QueueJob(d, node.ServerID, node.ID, "delete_resource", &JobPayload{ResourceID: node.ResourceID, Kind: "node"}) if forwardNode(node) && node.ExitResourceID != "" { @@ -966,6 +1040,7 @@ func (o *Orchestrator) RepairNode(nodeID, actor string) (bool, error) { if node.Status != "failed" && node.Status != "degraded" { return httpErr(409, "节点状态正常，无需修复") } + clearRebuildPending(node) if err := RebuildNode(d, node, true); err != nil { return err } @@ -1022,9 +1097,7 @@ func assignCustomerData(d *Data, nodeID, customerID, actor string) (*NodeClient, cancelQueuedDeletes(d, node.ID, "已取消：节点恢复服务") node.AutoRemoving = BoolPtr(false) } - if err := RebuildNode(d, node, false); err != nil { - return nil, err - } + markRebuildPending(node, false) } Audit(d, actor, "assign_customer", node.ID, map[string]any{"customerId": customerID, "customer": customer.Name}) return nc, nil @@ -1047,9 +1120,7 @@ func unassignCustomerData(d *Data, nodeID, customerID, actor string) error { } d.NodeClients = append(d.NodeClients[:index], d.NodeClients[index+1:]...) if node.Status != "draft" && node.Status != "removing" { - if err := RebuildNode(d, node, false); err != nil { - return err - } + markRebuildPending(node, false) } Audit(d, actor, "unassign_customer", node.ID, map[string]any{"customerId": customerID}) return nil @@ -1130,9 +1201,7 @@ func AlignCustomerNodes(d *Data, customer *Customer, desiredIds []string, actor cancelQueuedDeletes(d, node.ID, "已取消：节点恢复服务") node.AutoRemoving = BoolPtr(false) } - if err := RebuildNode(d, node, false); err != nil { - return err - } + markRebuildPending(node, false) } return nil } diff --git a/panel-go/rebuild_coalesce_test.go b/panel-go/rebuild_coalesce_test.go new file mode 100644 index 0000000..08e907c --- /dev/null +++ b/panel-go/rebuild_coalesce_test.go @@ -0,0 +1,197 @@ +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)) + } +} diff --git a/panel-go/types.go b/panel-go/types.go index 24b45bf..e1ff430 100644 --- a/panel-go/types.go +++ b/panel-go/types.go @@ -742,6 +742,14 @@ type Node struct { // (MigrateSocksNodeCredentials) mints a pair. SocksUser string `json:"socksUser,omitempty"` SocksPassword string `json:"socksPassword,omitempty"` + // RebuildPending marks a node whose binding-class changes are being + // coalesced (round-3 item 23): the flusher applies ONE rebuild with + // the final state once the fixed window (from RebuildPendingAt, the + // FIRST pending change) elapses. Persisted so a panel restart inside + // the window cannot strand the node un-deployed. + RebuildPending bool `json:"rebuildPending,omitempty"` + RebuildPendingAt string `json:"rebuildPendingAt,omitempty"` + RebuildPendingExit bool `json:"rebuildPendingExit,omitempty"` Reality *RealityConfig `json:"reality,omitempty"` ExitCredentials *ExitCredentials `json:"exitCredentials,omitempty"` UsageCounters *UsageCounters `json:"usageCounters"`