diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 500d7e5fd..13944ea1e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -141,6 +141,22 @@ jobs: # internal/web/service runs ~10x slower under -race and overruns the 10m default. go test -race -shuffle=on -count=1 -timeout 25m $(cat /tmp/go-packages.txt) + # A real master and node panel, each its own process, driven through node sync. + # A SKIP here means the binary was never handed over, so it fails the job. + node-e2e: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-go@v7 + with: + go-version-file: go.mod + cache: true + - name: Master + node end to end + run: | + set -o pipefail + make node-e2e 2>&1 | tee /tmp/node-e2e.log + if grep -q -- '--- SKIP' /tmp/node-e2e.log; then echo "node-e2e skipped"; exit 1; fi + # Brief native-fuzz smoke on the security-/parser-critical decoders. Each runs the # generated corpus plus 30s of exploration; a crash here is a real input-handling bug. fuzz-smoke: diff --git a/CLAUDE.md b/CLAUDE.md index cd3298d56..00fc01287 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -160,9 +160,10 @@ file locations when it can answer in one hop. `-race`); `httptest` for HTTP. Keep `database.InitDB` for reopening a file or migrating a hand-built legacy DB. `internal/sub`'s `initSubDB(t)` is the template. - Code must pass `golangci-lint run` (gofumpt + goimports formatting): `make lint`. -- Postgres, xray-gRPC-e2e and scale tests `t.Skip` unless `XUI_TEST_PG_DSN`, - `XUI_DB_TYPE`+`XUI_DB_DSN`, `XRAY_E2E_BINARY` or `XUI_SCALE_TEST` is set — a - green `go test ./...` does not mean those paths ran. +- Postgres, xray-gRPC-e2e, master+node and scale tests `t.Skip` unless + `XUI_TEST_PG_DSN`, `XUI_DB_TYPE`+`XUI_DB_DSN`, `XRAY_E2E_BINARY`, + `XUI_NODE_E2E_BINARY` or `XUI_SCALE_TEST` is set — a green `go test ./...` + does not mean those paths ran. ## Frontend conventions (summary; full version in frontend/CLAUDE.md) - Ant Design 6 only — no Tailwind/shadcn. Targeted tweaks, not rewrites. @@ -191,9 +192,11 @@ reads as a broken repo, not a missing step. Run `make dist-stub` once; every make verify # gen-check + lint + typecheck + test + build + build-storybook That is the *fast* gate, not all of CI. `ci.yml` also runs `make race`, -`make vulncheck`, a live-Postgres job (where a SKIP counts as a failure) and a +`make vulncheck`, a live-Postgres job (where a SKIP counts as a failure), +`make node-e2e` (a real master and node panel, `internal/nodee2e/`) and a 30s fuzz smoke on `FuzzParseLink`/`FuzzDecodeCertPin` — run those locally when -you touch DB/dialect or parser code. +you touch DB/dialect, node sync or parser code. A new node-sync behaviour gets +a cell in `internal/nodee2e/node_sync_test.go`. Common targets: `make gen` (regenerate Zod/OpenAPI), `make lint` (Go + frontend), `make test` (Go `-shuffle=on` + frontend), `make race`, `make build`. See `Makefile`. diff --git a/Makefile b/Makefile index 8b36b610d..26ec51307 100644 --- a/Makefile +++ b/Makefile @@ -58,6 +58,13 @@ test-go: dist-stub ## Go tests (shuffle, no cache) race: dist-stub ## Go tests with the race detector (needs a C compiler) go test -race -shuffle=on -count=1 -timeout 25m $(GO_PKGS) +.PHONY: node-e2e +# Two real panel processes (master + node); test-go only runs nodee2e as a skip. +NODE_E2E_BIN = $(CURDIR)/.cache/node-e2e/x-ui$(shell go env GOEXE) +node-e2e: dist-stub ## Master+node sync end to end with two real panel processes + go build -o $(NODE_E2E_BIN) . + XUI_NODE_E2E_BINARY=$(NODE_E2E_BIN) go test -count=1 -timeout 20m -v ./internal/nodee2e/ + .PHONY: test-fe test-fe: ## Frontend tests (vitest) cd $(FRONTEND) && npm test diff --git a/internal/nodee2e/harness_test.go b/internal/nodee2e/harness_test.go new file mode 100644 index 000000000..16244cc1f --- /dev/null +++ b/internal/nodee2e/harness_test.go @@ -0,0 +1,350 @@ +// Package nodee2e drives a real master panel and a real node panel, each its own +// process, through the node-sync paths. Gated by XUI_NODE_E2E_BINARY. +package nodee2e + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "regexp" + "strconv" + "strings" + "testing" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database" +) + +const settleTimeout = 30 * time.Second + +func panelBinary(t *testing.T) string { + t.Helper() + bin := os.Getenv("XUI_NODE_E2E_BINARY") + if bin == "" { + t.Skip("XUI_NODE_E2E_BINARY not set; run `make node-e2e`") + } + abs, err := filepath.Abs(bin) + if err != nil { + t.Fatalf("resolve %s: %v", bin, err) + } + return abs +} + +type panel struct { + t *testing.T + name string + bin string + dir string + port int + token string + cmd *exec.Cmd + logOut *os.File +} + +func freePort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("free port: %v", err) + } + defer l.Close() + return l.Addr().(*net.TCPAddr).Port +} + +func (p *panel) env() []string { + return append(os.Environ(), + "XUI_DB_FOLDER="+filepath.Join(p.dir, "db"), + "XUI_LOG_FOLDER="+filepath.Join(p.dir, "log"), + "XUI_BIN_FOLDER="+filepath.Join(p.dir, "bin"), + "XUI_ENABLE_FAIL2BAN=false", + "MSYS_NO_PATHCONV=1", + ) +} + +func (p *panel) cli(args ...string) string { + p.t.Helper() + cmd := exec.Command(p.bin, args...) + cmd.Env = p.env() + out, err := cmd.CombinedOutput() + if err != nil { + p.t.Fatalf("%s %v: %v\n%s", p.name, args, err, out) + } + return string(out) +} + +var apiTokenLine = regexp.MustCompile(`(?m)^apiToken:\s*(\S+)`) + +func (p *panel) mintToken(name, scope string) string { + p.t.Helper() + out := p.cli("setting", "-getApiToken", "-tokenName", name, "-tokenScope", scope) + m := apiTokenLine.FindStringSubmatch(out) + if m == nil { + p.t.Fatalf("%s: no apiToken in output:\n%s", p.name, out) + } + return m[1] +} + +// newPanel prepares a panel's database: credentials, a private port, its own +// sub-server port (two panels on one host would race for 2096) and an admin token. +func newPanel(t *testing.T, bin, name string) *panel { + t.Helper() + p := &panel{t: t, name: name, bin: bin, dir: t.TempDir(), port: freePort(t)} + for _, d := range []string{"db", "log", "bin"} { + if err := os.MkdirAll(filepath.Join(p.dir, d), 0o755); err != nil { + t.Fatal(err) + } + } + p.cli("setting", "-username", "e2e", "-password", "e2e-pass", "-port", strconv.Itoa(p.port), "-webBasePath", "/") + if err := database.InitDB(filepath.Join(p.dir, "db", "x-ui.db")); err != nil { + t.Fatalf("%s: open db: %v", name, err) + } + db := database.GetDB() + db.Exec("DELETE FROM settings WHERE key = ?", "subPort") + if err := db.Exec("INSERT INTO settings(key, value) VALUES (?, ?)", "subPort", strconv.Itoa(freePort(t))).Error; err != nil { + t.Fatalf("%s: set subPort: %v", name, err) + } + if err := database.CloseDB(); err != nil { + t.Fatalf("%s: close db: %v", name, err) + } + p.token = p.mintToken("e2e-driver", "admin") + t.Cleanup(p.stop) + return p +} + +func (p *panel) start() { + p.t.Helper() + logOut, err := os.OpenFile(filepath.Join(p.dir, "stdout.log"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + p.t.Fatal(err) + } + p.logOut = logOut + p.cmd = exec.Command(p.bin, "run") + p.cmd.Env = p.env() + p.cmd.Stdout = logOut + p.cmd.Stderr = logOut + if err := p.cmd.Start(); err != nil { + p.t.Fatalf("%s: start: %v", p.name, err) + } + eventually(p.t, settleTimeout, p.name+" answers /server/status", func() (bool, string) { + env, err := p.try(http.MethodGet, "/panel/api/server/status", nil) + if err != nil { + return false, err.Error() + } + return env.Success, env.Msg + }) +} + +func (p *panel) stop() { + if p.cmd == nil || p.cmd.Process == nil { + return + } + _ = p.cmd.Process.Kill() + _, _ = p.cmd.Process.Wait() + p.cmd = nil + if p.logOut != nil { + _ = p.logOut.Close() + p.logOut = nil + } + if p.t.Failed() { + if b, err := os.ReadFile(filepath.Join(p.dir, "stdout.log")); err == nil { + tail := string(b) + if len(tail) > 6000 { + tail = tail[len(tail)-6000:] + } + p.t.Logf("---- %s stdout tail ----\n%s", p.name, tail) + } + } +} + +// deleteInboundRow simulates a node that lost an inbound (restore, reinstall) +// while stopped; it must not run against a live panel. +func (p *panel) deleteInboundRow(id int) { + p.t.Helper() + if p.cmd != nil { + p.t.Fatalf("%s: deleteInboundRow on a running panel", p.name) + } + if err := database.InitDB(filepath.Join(p.dir, "db", "x-ui.db")); err != nil { + p.t.Fatalf("%s: open db: %v", p.name, err) + } + defer func() { _ = database.CloseDB() }() + db := database.GetDB() + for _, q := range []string{"DELETE FROM client_inbounds WHERE inbound_id = ?", "DELETE FROM client_traffics WHERE inbound_id = ?", "DELETE FROM inbounds WHERE id = ?"} { + if err := db.Exec(q, id).Error; err != nil { + p.t.Fatalf("%s: %s: %v", p.name, q, err) + } + } +} + +func (p *panel) url() string { return "http://127.0.0.1:" + strconv.Itoa(p.port) } + +type envelope struct { + Success bool `json:"success"` + Msg string `json:"msg"` + Obj json.RawMessage `json:"obj"` +} + +func (p *panel) try(method, path string, body any) (*envelope, error) { + var rd io.Reader + if body != nil { + b, err := json.Marshal(body) + if err != nil { + return nil, err + } + rd = bytes.NewReader(b) + } + req, err := http.NewRequest(method, p.url()+path, rd) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+p.token) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + resp, err := (&http.Client{Timeout: 20 * time.Second}).Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + raw, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, raw) + } + var env envelope + if err := json.Unmarshal(raw, &env); err != nil { + return nil, fmt.Errorf("decode %s: %w (%s)", path, err, raw) + } + return &env, nil +} + +// call fails the test on transport errors or success:false. +func (p *panel) call(method, path string, body any) json.RawMessage { + p.t.Helper() + env, err := p.try(method, path, body) + if err != nil { + p.t.Fatalf("%s %s %s: %v", p.name, method, path, err) + } + if !env.Success { + p.t.Fatalf("%s %s %s: success=false msg=%q", p.name, method, path, env.Msg) + } + return env.Obj +} + +func eventually(t *testing.T, timeout time.Duration, what string, check func() (bool, string)) { + t.Helper() + deadline := time.Now().Add(timeout) + last := "" + for { + ok, detail := check() + if ok { + return + } + last = detail + if time.Now().After(deadline) { + t.Fatalf("timed out after %s waiting for %s; last: %s", timeout, what, last) + } + time.Sleep(500 * time.Millisecond) + } +} + +// inboundView is the subset of an inbound row the scenarios assert on. +type inboundView struct { + Id int `json:"id"` + Remark string `json:"remark"` + Enable bool `json:"enable"` + Port int `json:"port"` + Tag string `json:"tag"` + NodeID *int `json:"nodeId"` + Settings json.RawMessage `json:"settings"` +} + +type clientEntry map[string]any + +func (c clientEntry) email() string { s, _ := c["email"].(string); return s } + +func (ib inboundView) clients() []clientEntry { + raw := ib.Settings + var asString string + if json.Unmarshal(raw, &asString) == nil { + raw = json.RawMessage(asString) + } + var s struct { + Clients []clientEntry `json:"clients"` + } + _ = json.Unmarshal(raw, &s) + return s.Clients +} + +func (ib inboundView) emails() []string { + out := []string{} + for _, c := range ib.clients() { + out = append(out, c.email()) + } + return out +} + +func (ib inboundView) client(email string) clientEntry { + for _, c := range ib.clients() { + if strings.EqualFold(c.email(), email) { + return c + } + } + return nil +} + +func (p *panel) inbounds() []inboundView { + p.t.Helper() + var list []inboundView + if err := json.Unmarshal(p.call(http.MethodGet, "/panel/api/inbounds/list", nil), &list); err != nil { + p.t.Fatalf("%s: decode inbound list: %v", p.name, err) + } + return list +} + +func (p *panel) inboundOnPort(port int) (inboundView, bool) { + p.t.Helper() + for _, ib := range p.inbounds() { + if ib.Port == port { + return ib, true + } + } + return inboundView{}, false +} + +const tcpStream = `{"network":"tcp","security":"none","tcpSettings":{"header":{"type":"none"}}}` + +func vlessInbound(remark string, port int, nodeID *int, clients ...map[string]any) map[string]any { + if clients == nil { + clients = []map[string]any{} + } + settings, _ := json.Marshal(map[string]any{"clients": clients, "decryption": "none"}) + body := map[string]any{ + "remark": remark, "enable": true, "port": port, "protocol": "vless", + "settings": string(settings), "streamSettings": tcpStream, "sniffing": `{}`, + } + if nodeID != nil { + body["nodeId"] = *nodeID + } + return body +} + +func vlessClient(email string) map[string]any { + return map[string]any{"email": email, "enable": true, "id": newUUID(email)} +} + +// newUUID derives a stable, valid UUID from a label so failures are reproducible. +func newUUID(label string) string { + var b [16]byte + copy(b[:], []byte(label+"________________")) + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]) +} diff --git a/internal/nodee2e/node_sync_test.go b/internal/nodee2e/node_sync_test.go new file mode 100644 index 000000000..c22628521 --- /dev/null +++ b/internal/nodee2e/node_sync_test.go @@ -0,0 +1,343 @@ +package nodee2e + +import ( + "encoding/json" + "fmt" + "net/http" + "slices" + "strconv" + "testing" + "time" +) + +// TestNodeSync walks one master/node pair per enrollment scope through every +// operation that must converge onto the node. Each subtest names its invariant. +func TestNodeSync(t *testing.T) { + bin := panelBinary(t) + for _, scope := range []string{"admin", "node-sync"} { + t.Run("enrolled with "+scope+" token", func(t *testing.T) { + runNodeSyncScenarios(t, bin, scope) + }) + } +} + +type pair struct { + t *testing.T + master *panel + node *panel + nodeID int +} + +func (pr *pair) nodeInbound(port int) (inboundView, bool) { return pr.node.inboundOnPort(port) } + +func (pr *pair) masterInbound(port int) (inboundView, bool) { + for _, ib := range pr.master.inbounds() { + if ib.Port == port && ib.NodeID != nil && *ib.NodeID == pr.nodeID { + return ib, true + } + } + return inboundView{}, false +} + +func (pr *pair) waitNode(what string, port int, ok func(inboundView) bool) { + pr.t.Helper() + eventually(pr.t, settleTimeout, what, func() (bool, string) { + ib, found := pr.nodeInbound(port) + if !found { + return ok(inboundView{}) && false, fmt.Sprintf("node has no inbound on %d", port) + } + return ok(ib), fmt.Sprintf("node inbound %d: enable=%v remark=%q emails=%v", port, ib.Enable, ib.Remark, ib.emails()) + }) +} + +func (pr *pair) waitNodeAbsent(what string, port int) { + pr.t.Helper() + eventually(pr.t, settleTimeout, what, func() (bool, string) { + ib, found := pr.nodeInbound(port) + return !found, fmt.Sprintf("node still has inbound %d with %v", port, ib.emails()) + }) +} + +func (pr *pair) bulkAttach(emails []string, masterInboundID int) { + pr.t.Helper() + var res struct { + Attached []string `json:"attached"` + Errors []string `json:"errors"` + } + obj := pr.master.call(http.MethodPost, "/panel/api/clients/bulkAttach", map[string]any{"emails": emails, "inboundIds": []int{masterInboundID}}) + if err := json.Unmarshal(obj, &res); err != nil { + pr.t.Fatalf("decode bulkAttach: %v", err) + } + if len(res.Errors) != 0 || len(res.Attached) != len(emails) { + pr.t.Fatalf("bulkAttach attached=%d/%d errors=%v", len(res.Attached), len(emails), res.Errors) + } +} + +func emailRange(prefix string, from, to int) []string { + out := make([]string, 0, to-from+1) + for i := from; i <= to; i++ { + out = append(out, prefix+strconv.Itoa(i)) + } + return out +} + +func hasAll(have []string, want ...string) bool { + for _, w := range want { + if !slices.Contains(have, w) { + return false + } + } + return true +} + +func runNodeSyncScenarios(t *testing.T, bin, scope string) { + master := newPanel(t, bin, "master") + node := newPanel(t, bin, "node") + linkToken := node.mintToken("master-link", scope) + master.start() + node.start() + pr := &pair{t: t, master: master, node: node} + + var ( + adoptedPort = freePort(t) + madePort = freePort(t) + lostPort = freePort(t) + droppedPort = freePort(t) + offlinePort = freePort(t) + unmanagedPort = freePort(t) + localPort = freePort(t) + ) + node.call(http.MethodPost, "/panel/api/inbounds/add", vlessInbound("pre-existing", adoptedPort, nil)) + + local := master.call(http.MethodPost, "/panel/api/inbounds/add", vlessInbound("local-pool", localPort, nil)) + var localIb inboundView + _ = json.Unmarshal(local, &localIb) + for _, email := range emailRange("p", 1, 45) { + master.call(http.MethodPost, "/panel/api/clients/add", map[string]any{ + "client": map[string]any{"email": email, "enable": true}, "inboundIds": []int{localIb.Id}, + }) + } + + var nodeView struct { + Id int `json:"id"` + } + obj := master.call(http.MethodPost, "/panel/api/nodes/add", map[string]any{ + "name": "n1", "scheme": "http", "address": "127.0.0.1", "port": node.port, "basePath": "/", + "apiToken": linkToken, "enable": true, "allowPrivateAddress": true, + }) + if err := json.Unmarshal(obj, &nodeView); err != nil || nodeView.Id == 0 { + t.Fatalf("decode node add: %v (%s)", err, obj) + } + pr.nodeID = nodeView.Id + + var adoptedID, madeID int + t.Run("an inbound already on the node is adopted by the master", func(t *testing.T) { + pr.t = t + eventually(t, settleTimeout, "master adopts the node inbound", func() (bool, string) { + ib, ok := pr.masterInbound(adoptedPort) + adoptedID = ib.Id + return ok, "not adopted yet" + }) + }) + if adoptedID == 0 { + t.Fatal("no adopted inbound; later scenarios depend on it") + } + + t.Run("an inbound created on the master for the node lands there with its clients", func(t *testing.T) { + pr.t = t + obj := master.call(http.MethodPost, "/panel/api/inbounds/add", + vlessInbound("made-on-master", madePort, &pr.nodeID, vlessClient("m1"), vlessClient("m2"))) + var ib inboundView + _ = json.Unmarshal(obj, &ib) + madeID = ib.Id + pr.waitNode("node holds the master-made inbound", madePort, func(ib inboundView) bool { + return hasAll(ib.emails(), "m1", "m2") + }) + }) + + t.Run("editing the inbound on the master updates the node and keeps its clients", func(t *testing.T) { + pr.t = t + body := vlessInbound("renamed-on-master", madePort, &pr.nodeID) + body["settings"] = `{"decryption":"none"}` + master.call(http.MethodPost, "/panel/api/inbounds/update/"+strconv.Itoa(madeID), body) + pr.waitNode("node shows the new remark with both clients", madePort, func(ib inboundView) bool { + return ib.Remark == "renamed-on-master" && hasAll(ib.emails(), "m1", "m2") + }) + }) + + t.Run("attaching a few existing clients reaches the node", func(t *testing.T) { + pr.t = t + pr.bulkAttach([]string{"p1", "p2", "p3"}, madeID) + pr.waitNode("node holds p1..p3", madePort, func(ib inboundView) bool { + return hasAll(ib.emails(), "m1", "m2", "p1", "p2", "p3") + }) + }) + + t.Run("attaching more clients than the per-client push limit reaches the node", func(t *testing.T) { + pr.t = t + emails := emailRange("p", 4, 43) + pr.bulkAttach(emails, adoptedID) + pr.waitNode("node holds all 40", adoptedPort, func(ib inboundView) bool { + return hasAll(ib.emails(), emails...) + }) + }) + + t.Run("disabling a client on the master disables it on the node", func(t *testing.T) { + pr.t = t + mib, _ := pr.masterInbound(madePort) + entry := mib.client("p1") + if entry == nil { + t.Fatalf("master inbound has no p1: %v", mib.emails()) + } + entry["enable"] = false + master.call(http.MethodPost, "/panel/api/clients/update/p1", entry) + pr.waitNode("node p1 disabled", madePort, func(ib inboundView) bool { + c := ib.client("p1") + return c != nil && c["enable"] == false + }) + }) + + t.Run("detaching a client from the node inbound removes it there", func(t *testing.T) { + pr.t = t + master.call(http.MethodPost, "/panel/api/clients/p2/detach", map[string]any{"inboundIds": []int{madeID}}) + pr.waitNode("node drops p2", madePort, func(ib inboundView) bool { + return ib.client("p2") == nil && ib.client("p3") != nil + }) + }) + + t.Run("deleting a client on the master removes it from the node", func(t *testing.T) { + pr.t = t + master.call(http.MethodPost, "/panel/api/clients/del/p4", nil) + pr.waitNode("node drops p4", adoptedPort, func(ib inboundView) bool { + return ib.client("p4") == nil && ib.client("p5") != nil + }) + }) + + t.Run("switching the inbound off on the master switches it off on the node", func(t *testing.T) { + pr.t = t + master.call(http.MethodPost, "/panel/api/inbounds/setEnable/"+strconv.Itoa(madeID), map[string]any{"enable": false}) + pr.waitNode("node inbound disabled", madePort, func(ib inboundView) bool { return !ib.Enable }) + }) + + t.Run("node traffic reaches the master and a master reset clears the node", func(t *testing.T) { + pr.t = t + node.call(http.MethodPost, "/panel/api/clients/updateTraffic/p5", map[string]any{"upload": 1000, "download": 2000}) + usage := func(p *panel) int64 { + var tr struct{ Up, Down int64 } + _ = json.Unmarshal(p.call(http.MethodGet, "/panel/api/clients/traffic/p5", nil), &tr) + return tr.Up + tr.Down + } + eventually(t, settleTimeout, "master sees p5's node traffic", func() (bool, string) { + u := usage(master) + return u == 3000, fmt.Sprintf("master p5 usage %d", u) + }) + master.call(http.MethodPost, "/panel/api/clients/resetTraffic/p5", nil) + eventually(t, settleTimeout, "node p5 usage reset", func() (bool, string) { + u := usage(node) + return u == 0, fmt.Sprintf("node p5 usage %d", u) + }) + time.Sleep(12 * time.Second) + if u := usage(master); u != 0 { + t.Fatalf("master p5 usage %d after reset settled, want 0", u) + } + }) + + t.Run("an inbound deleted on the node is removed from the master too", func(t *testing.T) { + pr.t = t + master.call(http.MethodPost, "/panel/api/inbounds/add", vlessInbound("deleted-on-node", lostPort, &pr.nodeID, vlessClient("l1"))) + pr.waitNode("node holds the inbound", lostPort, func(ib inboundView) bool { return ib.client("l1") != nil }) + nib, _ := pr.nodeInbound(lostPort) + node.call(http.MethodPost, "/panel/api/inbounds/del/"+strconv.Itoa(nib.Id), nil) + eventually(t, settleTimeout, "master mirrors the node-side delete (#6219)", func() (bool, string) { + _, still := pr.masterInbound(lostPort) + return !still, "master still has the inbound" + }) + }) + + t.Run("deleting the inbound on the master removes it from the node", func(t *testing.T) { + pr.t = t + obj := master.call(http.MethodPost, "/panel/api/inbounds/add", vlessInbound("deleted-on-master", droppedPort, &pr.nodeID, vlessClient("d1"))) + var ib inboundView + _ = json.Unmarshal(obj, &ib) + pr.waitNode("node holds the inbound", droppedPort, func(ib inboundView) bool { return ib.client("d1") != nil }) + master.call(http.MethodPost, "/panel/api/inbounds/del/"+strconv.Itoa(ib.Id), nil) + pr.waitNodeAbsent("node drops the inbound", droppedPort) + }) + + t.Run("a change made while the node is down reaches it once it is back", func(t *testing.T) { + pr.t = t + node.stop() + pr.bulkAttach([]string{"p44"}, adoptedID) + node.start() + pr.waitNode("node holds p44 after restart", adoptedPort, func(ib inboundView) bool { + return ib.client("p44") != nil + }) + }) + + t.Run("an inbound the node lost while down is re-created with the master's pending change", func(t *testing.T) { + pr.t = t + nib, ok := pr.nodeInbound(adoptedPort) + if !ok { + t.Fatal("node has no adopted inbound to lose") + } + node.stop() + node.deleteInboundRow(nib.Id) + pr.bulkAttach([]string{"p45"}, adoptedID) + node.start() + pr.waitNode("node re-creates the inbound with p44 and p45", adoptedPort, func(ib inboundView) bool { + return ib.client("p44") != nil && ib.client("p45") != nil + }) + }) + + t.Run("an inbound created and edited while the node is down lands once it is back", func(t *testing.T) { + pr.t = t + node.stop() + obj := master.call(http.MethodPost, "/panel/api/inbounds/add", vlessInbound("made-while-down", offlinePort, &pr.nodeID, vlessClient("o1"))) + var ib inboundView + _ = json.Unmarshal(obj, &ib) + body := vlessInbound("edited-while-down", offlinePort, &pr.nodeID) + body["settings"] = `{"decryption":"none"}` + master.call(http.MethodPost, "/panel/api/inbounds/update/"+strconv.Itoa(ib.Id), body) + node.start() + pr.waitNode("node holds the edited inbound with o1", offlinePort, func(ib inboundView) bool { + return ib.Remark == "edited-while-down" && ib.client("o1") != nil + }) + }) + + t.Run("a change made while the node is disabled on the master lands once it is re-enabled", func(t *testing.T) { + pr.t = t + nodePath := "/panel/api/nodes/setEnable/" + strconv.Itoa(pr.nodeID) + master.call(http.MethodPost, nodePath, map[string]any{"enable": false}) + pr.bulkAttach([]string{"p6"}, madeID) + time.Sleep(6 * time.Second) + if ib, _ := pr.nodeInbound(madePort); ib.client("p6") != nil { + t.Fatal("a disabled node received a push") + } + master.call(http.MethodPost, nodePath, map[string]any{"enable": true}) + pr.waitNode("node holds p6 after re-enable", madePort, func(ib inboundView) bool { return ib.client("p6") != nil }) + }) + + t.Run("selected sync mode leaves the node's unselected inbounds alone", func(t *testing.T) { + pr.t = t + var selected []string + for _, ib := range master.inbounds() { + if ib.NodeID != nil && *ib.NodeID == pr.nodeID { + selected = append(selected, ib.Tag) + } + } + master.call(http.MethodPost, "/panel/api/nodes/update/"+strconv.Itoa(pr.nodeID), map[string]any{ + "name": "n1", "scheme": "http", "address": "127.0.0.1", "port": node.port, "basePath": "/", + "enable": true, "allowPrivateAddress": true, "inboundSyncMode": "selected", "inboundTags": selected, + }) + node.call(http.MethodPost, "/panel/api/inbounds/add", vlessInbound("node-only", unmanagedPort, nil, vlessClient("u1"))) + pr.bulkAttach([]string{"p7"}, madeID) + pr.waitNode("selected inbound still converges", madePort, func(ib inboundView) bool { return ib.client("p7") != nil }) + time.Sleep(12 * time.Second) + if _, adopted := pr.masterInbound(unmanagedPort); adopted { + t.Fatal("master adopted an unselected node inbound") + } + if ib, ok := pr.nodeInbound(unmanagedPort); !ok || ib.client("u1") == nil { + t.Fatal("reconcile swept or rewrote an unselected node inbound") + } + }) +}