diff --git a/go-backend/internal/http/handler/handler.go b/go-backend/internal/http/handler/handler.go index 0ddad66..513871e 100644 --- a/go-backend/internal/http/handler/handler.go +++ b/go-backend/internal/http/handler/handler.go @@ -1123,6 +1123,14 @@ func normalizeAndValidateConfigValue(key, value string) (string, error) { } return pngDataURLPrefix + payload, nil + case sequentialPortConfigKey: + normalized := strings.TrimSpace(strings.ToLower(value)) + switch normalized { + case "true", "false": + return normalized, nil + default: + return "", fmt.Errorf("顺次分配端口开关配置值无效") + } case monitorTunnelQualityEnabledConfigKey: normalized := strings.TrimSpace(strings.ToLower(value)) switch normalized { diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 23162b1..b14802e 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -9,7 +9,6 @@ import ( "errors" "fmt" "log" - "math/big" "net" "net/http" "sort" @@ -1315,13 +1314,13 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) { return } + var runtimePending error newEntryNodeIDs, _ = h.tunnelEntryNodeIDs(id) if !sameInt64Set(oldEntryNodeIDs, newEntryNodeIDs) { h.cleanupTunnelForwardRuntimesOnRemovedEntryNodes(id, oldEntryNodeIDs, newEntryNodeIDs) - h.syncTunnelForwardsEntryPorts(id, newEntryNodeIDs) + runtimePending = h.syncTunnelForwardsEntryPorts(id, newEntryNodeIDs) } - var runtimePending error if typeVal == 2 { applyRuntime := h.applyTunnelRuntime if oldTunnel != nil && oldTunnel.Type == 2 { @@ -1329,9 +1328,9 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) { } createdChains, createdServices, applyErr := applyRuntime(runtimeState) if isTunnelRuntimePending(applyErr) { - runtimePending = applyErr + runtimePending = errors.Join(runtimePending, applyErr) } - if applyErr != nil && runtimePending == nil { + if applyErr != nil && !isTunnelRuntimePending(applyErr) { updateProtocol := "tls" if len(runtimeState.InNodes) > 0 && strings.TrimSpace(runtimeState.InNodes[0].Protocol) != "" { updateProtocol = strings.TrimSpace(runtimeState.InNodes[0].Protocol) @@ -1597,19 +1596,20 @@ func (h *Handler) validateTunnelEntryPortConflictsForNewEntriesTx(tx *gorm.DB, t return nil } -func (h *Handler) syncTunnelForwardsEntryPorts(tunnelID int64, entryNodeIDs []int64) { +func (h *Handler) syncTunnelForwardsEntryPorts(tunnelID int64, entryNodeIDs []int64) error { if h == nil || h.repo == nil || tunnelID <= 0 { - return + return nil } entryNodeIDs = uniqueInt64s(entryNodeIDs) if len(entryNodeIDs) == 0 { - return + return nil } forwards, err := h.listForwardsByTunnel(tunnelID) - if err != nil || len(forwards) == 0 { - return + if err != nil { + return err } + var syncErrors []error allowInIP := len(entryNodeIDs) == 1 for i := range forwards { @@ -1619,6 +1619,7 @@ func (h *Handler) syncTunnelForwardsEntryPorts(tunnelID int64, entryNodeIDs []in } oldPorts, err := h.listForwardPorts(f.ID) if err != nil { + syncErrors = append(syncErrors, err) continue } referencePort := pickForwardPortFromRecords(oldPorts) @@ -1660,14 +1661,17 @@ func (h *Handler) syncTunnelForwardsEntryPorts(tunnelID int64, entryNodeIDs []in } entries = append(entries, forwardPortReplaceEntry{NodeID: nid, Port: port, InIP: inIP}) } - _ = h.repo.ReplaceForwardPorts(f.ID, entries) + if err := h.repo.ReplaceForwardPortsIndividually(f.ID, entries, h.isSequentialPortEnabled()); err != nil { + syncErrors = append(syncErrors, fmt.Errorf("转发 %s 入口端口更新失败: %w", f.Name, err)) + } } + return errors.Join(syncErrors...) } // resolvePortForNewEntryNode determines the port for a forward on a newly added // entry node. It tries to reuse referencePort (from existing entries); if that -// port is out of range or already occupied, it picks a random available port -// for this specific node. +// port is out of range or already occupied, zero requests atomic allocation +// when ReplaceForwardPorts persists the new entry. Explicit ports are rechecked there. func (h *Handler) resolvePortForNewEntryNode(nodeID int64, referencePort int, forwardID int64) int { node, err := h.getNodeRecord(nodeID) if err != nil { @@ -1684,48 +1688,23 @@ func (h *Handler) resolvePortForNewEntryNode(nodeID int64, referencePort int, fo } } - // referencePort doesn't work for this node; pick a random one. - newPort := h.pickRandomPortForNode(nodeID) - if newPort > 0 { - return newPort - } - return referencePort // last resort fallback + // Defer fallback selection to the same transaction that saves the entries. + return 0 } -// pickRandomPortForNode picks a random available port from a single node's -// port range, excluding ports already occupied by other forwards or chains. -func (h *Handler) pickRandomPortForNode(nodeID int64) int { - portRange, err := h.repo.GetNodePortRange(nodeID) +// sequentialPortConfigKey 是「未指定端口时按端口范围顺次分配」开关的配置键。 +const sequentialPortConfigKey = "sequential_port_enabled" + +// isSequentialPortEnabled 返回是否开启了顺次分配端口（默认关闭，保持随机分配）。 +func (h *Handler) isSequentialPortEnabled() bool { + if h == nil || h.repo == nil { + return false + } + value, err := h.repo.GetViteConfigValue(sequentialPortConfigKey) if err != nil { - return 0 + return false } - if portRange == "" { - portRange = "1000-65535" - } - - nodePorts, err := parsePorts(portRange) - if err != nil || len(nodePorts) == 0 { - return 0 - } - - used, err := h.getUsedPorts(nodeID) - if err != nil { - return 0 - } - - var available []int - for _, p := range nodePorts { - if !used[p] { - available = append(available, p) - } - } - - if len(available) == 0 { - return 0 - } - - idx, _ := rand.Int(rand.Reader, big.NewInt(int64(len(available)))) - return available[idx.Int64()] + return strings.EqualFold(strings.TrimSpace(value), "true") } func (h *Handler) tunnelDelete(w http.ResponseWriter, r *http.Request) { @@ -2290,12 +2269,7 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) { return } port := asInt(req["inPort"], 0) - if port <= 0 { - port = h.pickTunnelPort(tunnelID) - } - if port <= 0 { - port = 10000 - } + // A zero port is allocated and persisted atomically by the repository. entryNodes, _ := h.tunnelEntryNodeIDs(tunnelID) isNftTunnel, _, err := h.tunnelUsesNftables(tunnelID) if err != nil { @@ -2314,6 +2288,9 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) { return } for _, nodeID := range entryNodes { + if port <= 0 { + break // Repository validates all nodes while allocating atomically. + } node, nodeErr := h.getNodeRecord(nodeID) if nodeErr != nil { continue @@ -2346,7 +2323,7 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) { proxyProtocolReceive := asInt(req["proxyProtocolReceive"], 0) proxyProtocolSend := asInt(req["proxyProtocolSend"], proxyProtocol) - forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn, ipMaxConn, nullableInt(ipSpeedID), proxyProtocol, proxyProtocolReceive, proxyProtocolSend) + forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn, ipMaxConn, nullableInt(ipSpeedID), proxyProtocol, proxyProtocolReceive, proxyProtocolSend, h.isSequentialPortEnabled()) if err != nil { response.WriteJSON(w, response.Err(-2, err.Error())) return @@ -2484,9 +2461,6 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) { if minPort.Valid { port = int(minPort.Int64) } - if port <= 0 { - port = h.pickTunnelPort(tunnelID) - } } hasInIP := false inIp := "" @@ -2511,6 +2485,9 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) { keptNodeIDs = diffInt64s(oldNodeIDs, removedNodeIDs) } for _, nodeID := range fwdEntryNodes { + if port <= 0 { + break // Repository validates all nodes while allocating atomically. + } node, nodeErr := h.getNodeRecord(nodeID) if nodeErr != nil { continue @@ -3033,9 +3010,6 @@ func (h *Handler) forwardBatchChangeTunnel(w http.ResponseWriter, r *http.Reques if port.Valid { p = int(port.Int64) } - if p <= 0 { - p = h.pickTunnelPort(req.TargetTunnelID) - } bctEntryNodes, _ := h.tunnelEntryNodeIDs(req.TargetTunnelID) newNodeIDs := uniqueInt64s(bctEntryNodes) removedNodeIDs := diffInt64s(oldNodeIDs, newNodeIDs) @@ -3043,6 +3017,9 @@ func (h *Handler) forwardBatchChangeTunnel(w http.ResponseWriter, r *http.Reques portRangeOk := true var portRangeErr error for _, nid := range bctEntryNodes { + if p <= 0 { + break // Repository validates all nodes while allocating atomically. + } nd, ndErr := h.getNodeRecord(nid) if ndErr != nil { portRangeErr = ndErr @@ -4972,75 +4949,6 @@ func (h *Handler) tunnelEntryNodeIDs(tunnelID int64) ([]int64, error) { return h.repo.TunnelEntryNodeIDs(tunnelID) } -func (h *Handler) pickTunnelPort(tunnelID int64) int { - entryNodes, err := h.tunnelEntryNodeIDs(tunnelID) - if err != nil || len(entryNodes) == 0 { - return 10000 - } - - var commonAvailable []int - firstNode := true - - for _, nodeID := range entryNodes { - portRange, err := h.repo.GetNodePortRange(nodeID) - if err != nil { - continue - } - if portRange == "" { - portRange = "1000-65535" - } - - nodePorts, err := parsePorts(portRange) - if err != nil { - continue - } - - used, err := h.getUsedPorts(nodeID) - if err != nil { - continue - } - - var available []int - for _, p := range nodePorts { - if !used[p] { - available = append(available, p) - } - } - - if firstNode { - commonAvailable = available - firstNode = false - } else { - set := make(map[int]bool) - for _, p := range available { - set[p] = true - } - var newCommon []int - for _, p := range commonAvailable { - if set[p] { - newCommon = append(newCommon, p) - } - } - commonAvailable = newCommon - } - - if len(commonAvailable) == 0 { - break - } - } - - if len(commonAvailable) > 0 { - idx, _ := rand.Int(rand.Reader, big.NewInt(int64(len(commonAvailable)))) - return commonAvailable[idx.Int64()] - } - - return 10000 -} - -func (h *Handler) getUsedPorts(nodeID int64) (map[int]bool, error) { - return h.repo.GetUsedPortsOnNodeAsMap(nodeID) -} - func parsePorts(portRange string) ([]int, error) { var ports []int parts := strings.Split(portRange, ",") @@ -5113,7 +5021,7 @@ func (h *Handler) replaceForwardPorts(forwardID, tunnelID int64, port int, inIp for i, nid := range entryNodes { entries[i] = forwardPortReplaceEntry{NodeID: nid, Port: port, InIP: inIp} } - return h.repo.ReplaceForwardPorts(forwardID, entries) + return h.repo.ReplaceForwardPorts(forwardID, entries, h.isSequentialPortEnabled()) } func (h *Handler) replaceForwardPortsPreservingInIP(forwardID, tunnelID int64, port int, oldPorts []forwardPortRecord) error { @@ -5122,7 +5030,7 @@ func (h *Handler) replaceForwardPortsPreservingInIP(forwardID, tunnelID int64, p return err } entries := buildForwardPortEntriesWithPreservedInIP(entryNodes, oldPorts, port) - return h.repo.ReplaceForwardPorts(forwardID, entries) + return h.repo.ReplaceForwardPorts(forwardID, entries, h.isSequentialPortEnabled()) } func (h *Handler) replaceForwardPortsWithRecords(forwardID int64, ports []forwardPortRecord) error { @@ -5130,7 +5038,7 @@ func (h *Handler) replaceForwardPortsWithRecords(forwardID int64, ports []forwar for i, fp := range ports { entries[i] = forwardPortReplaceEntry{NodeID: fp.NodeID, Port: fp.Port, InIP: fp.InIP} } - return h.repo.ReplaceForwardPorts(forwardID, entries) + return h.repo.ReplaceForwardPorts(forwardID, entries, h.isSequentialPortEnabled()) } func (h *Handler) rollbackForwardMutation(oldForward *forwardRecord, oldPorts []forwardPortRecord) { diff --git a/go-backend/internal/http/handler/sequential_port_test.go b/go-backend/internal/http/handler/sequential_port_test.go new file mode 100644 index 0000000..4575889 --- /dev/null +++ b/go-backend/internal/http/handler/sequential_port_test.go @@ -0,0 +1,34 @@ +package handler + +import "testing" + +func TestNormalizeSequentialPortConfigValue(t *testing.T) { + for in, want := range map[string]string{"true": "true", " FALSE ": "false"} { + got, err := normalizeAndValidateConfigValue(sequentialPortConfigKey, in) + if err != nil || got != want { + t.Fatalf("value %q: got %q err=%v, want %q", in, got, err, want) + } + } + if _, err := normalizeAndValidateConfigValue(sequentialPortConfigKey, "maybe"); err == nil { + t.Fatal("expected invalid value to be rejected") + } +} + +func TestSequentialPortConfigReadsRepositoryAndDefaultsOff(t *testing.T) { + _, r := setupConfigAccessTestRouter(t) + h := &Handler{repo: r} + if h.isSequentialPortEnabled() { + t.Fatal("missing config must preserve random allocation") + } + for _, tc := range []struct { + value string + want bool + }{ + {"true", true}, {" TRUE ", true}, {"false", false}, {"invalid", false}, + } { + seedConfigValue(t, r, sequentialPortConfigKey, tc.value) + if got := h.isSequentialPortEnabled(); got != tc.want { + t.Fatalf("config %q: got %v, want %v", tc.value, got, tc.want) + } + } +} diff --git a/go-backend/internal/http/handler/tunnel_delete_resolution.go b/go-backend/internal/http/handler/tunnel_delete_resolution.go index 49bba2e..08e4d71 100644 --- a/go-backend/internal/http/handler/tunnel_delete_resolution.go +++ b/go-backend/internal/http/handler/tunnel_delete_resolution.go @@ -500,10 +500,7 @@ func (h *Handler) planSingleTunnelDeleteForwardMigration(forward *forwardRecord, targetPort = int(minPort.Int64) } if targetPort <= 0 { - targetPort = h.pickTunnelPort(targetTunnelID) - } - if targetPort <= 0 { - targetPort = 10000 + return tunnelForwardMigrationPlan{}, errors.New("转发入口端口无效，请先修复该规则") } hasCustomInIP := false diff --git a/go-backend/internal/store/repo/repository.go b/go-backend/internal/store/repo/repository.go index 00381a0..dced279 100644 --- a/go-backend/internal/store/repo/repository.go +++ b/go-backend/internal/store/repo/repository.go @@ -1609,7 +1609,15 @@ func (r *Repository) CreatePeerShareRuntime(item *model.PeerShareRuntime) error if item == nil { return errors.New("runtime item is nil") } - return r.db.Create(item).Error + return r.db.Transaction(func(tx *gorm.DB) error { + if err := lockPortAllocationTx(tx); err != nil { + return err + } + if err := r.checkPeerRuntimePortTx(tx, item); err != nil { + return err + } + return tx.Create(item).Error + }) } func (r *Repository) UpdatePeerShareRuntime(item *model.PeerShareRuntime) error { @@ -1619,17 +1627,34 @@ func (r *Repository) UpdatePeerShareRuntime(item *model.PeerShareRuntime) error if item == nil { return errors.New("runtime item is nil") } - return r.db.Model(&model.PeerShareRuntime{}).Where("id = ?", item.ID).Updates(map[string]interface{}{ - "reservation_id": item.ReservationID, - "relay_auth": item.RelayAuth, - "relay_auth_version": item.RelayAuthVersion, - "binding_id": item.BindingID, "role": item.Role, - "chain_name": item.ChainName, "service_name": item.ServiceName, - "protocol": item.Protocol, "strategy": item.Strategy, - "port": item.Port, "target": item.Target, - "applied": item.Applied, "status": item.Status, - "updated_time": item.UpdatedTime, - }).Error + return r.db.Transaction(func(tx *gorm.DB) error { + if err := lockPortAllocationTx(tx); err != nil { + return err + } + var current model.PeerShareRuntime + if err := tx.Where("id = ?", item.ID).First(¤t).Error; err != nil { + return err + } + if current.NodeID != item.NodeID { + return errors.New("runtime node cannot change") + } + if !(current.Status == 1 && current.Port == item.Port) { + if err := r.checkPeerRuntimePortTx(tx, item); err != nil { + return err + } + } + return tx.Model(&model.PeerShareRuntime{}).Where("id = ?", item.ID).Updates(map[string]interface{}{ + "reservation_id": item.ReservationID, + "relay_auth": item.RelayAuth, + "relay_auth_version": item.RelayAuthVersion, + "binding_id": item.BindingID, "role": item.Role, + "chain_name": item.ChainName, "service_name": item.ServiceName, + "protocol": item.Protocol, "strategy": item.Strategy, + "port": item.Port, "target": item.Target, + "applied": item.Applied, "status": item.Status, + "updated_time": item.UpdatedTime, + }).Error + }) } func (r *Repository) MarkPeerShareRuntimeReleased(id int64, updatedTime int64) error { @@ -2664,6 +2689,15 @@ func importTunnels(tx *gorm.DB, tunnels []model.TunnelBackup, now int64) (int, e return count, err } for _, ct := range t.ChainTunnels { + if ct.Port > 0 { + used, err := (&Repository{}).usedPortSetTx(tx, ct.NodeID, 0, ct.TunnelID) + if err != nil { + return count, err + } + if used[ct.Port] { + return count, errors.New("导入链路端口已占用") + } + } chainItem := model.ChainTunnel{ ID: ct.ID, TunnelID: ct.TunnelID, @@ -2735,6 +2769,13 @@ func importForwards(tx *gorm.DB, forwards []model.ForwardBackup, now int64) (int return count, err } for _, fp := range *f.ForwardPorts { + used, err := (&Repository{}).usedPortSetTx(tx, fp.NodeID, f.ID, 0) + if err != nil { + return count, err + } + if used[fp.Port] { + return count, errors.New("导入转发端口已占用") + } if err := tx.Create(&model.ForwardPort{ForwardID: f.ID, NodeID: fp.NodeID, Port: fp.Port}).Error; err != nil { return count, err } diff --git a/go-backend/internal/store/repo/repository_backup_postgres.go b/go-backend/internal/store/repo/repository_backup_postgres.go index 238437a..83f916e 100644 --- a/go-backend/internal/store/repo/repository_backup_postgres.go +++ b/go-backend/internal/store/repo/repository_backup_postgres.go @@ -18,7 +18,12 @@ var errBackupImportBusy = errors.New("database is busy; retry the backup import" func backupImportTransaction(db *gorm.DB, importRows func(*gorm.DB) error) error { var err error for attempt := 0; attempt < 10; attempt++ { - err = db.Transaction(importRows) + err = db.Transaction(func(tx *gorm.DB) error { + if err := lockPortAllocationTx(tx); err != nil { + return err + } + return importRows(tx) + }) if !errors.Is(err, errBackupImportBusy) { return err } diff --git a/go-backend/internal/store/repo/repository_federation.go b/go-backend/internal/store/repo/repository_federation.go index effe63f..99d096f 100644 --- a/go-backend/internal/store/repo/repository_federation.go +++ b/go-backend/internal/store/repo/repository_federation.go @@ -146,19 +146,13 @@ func (r *Repository) CreateFederationTunnel(name string, tunnelType int, protoco InIP: sql.NullString{String: "", Valid: false}, } err := r.db.Transaction(func(tx *gorm.DB) error { + if err := lockPortAllocationTx(tx); err != nil { + return err + } if err := tx.Create(&tunnel).Error; err != nil { return err } - ct := model.ChainTunnel{ - TunnelID: tunnel.ID, - ChainType: "1", - NodeID: nodeID, - Port: sql.NullInt64{Int64: int64(remotePort), Valid: true}, - Strategy: sql.NullString{String: "fifo", Valid: true}, - Inx: sql.NullInt64{Int64: 0, Valid: true}, - Protocol: sql.NullString{String: protocol, Valid: true}, - } - if err := tx.Create(&ct).Error; err != nil { + if err := r.CreateChainTunnelTx(tx, tunnel.ID, "1", nodeID, sql.NullInt64{Int64: int64(remotePort), Valid: true}, "fifo", 0, protocol, ""); err != nil { return err } return nil diff --git a/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go b/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go index 76715b7..e848d3d 100644 --- a/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go +++ b/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go @@ -160,6 +160,9 @@ func TestForwardRepositoryPersistsPerIPLimits(t *testing.T) { defer r.Close() now := time.Now().UnixMilli() + if err := r.db.Create(&model.Node{ID: 3, Port: "1000-65535"}).Error; err != nil { + t.Fatal(err) + } forwardID, err := r.CreateForwardTx(1, "admin", "per-ip-forward", 2, "1.1.1.1:443", "fifo", now, 1, []int64{3}, 24000, "", nil, 0, 5, int64(21), 0, 0, 0) if err != nil { t.Fatalf("CreateForwardTx: %v", err) @@ -224,6 +227,9 @@ func TestForwardRepositoryPersistsProxyProtocolReceiveAndSend(t *testing.T) { defer r.Close() now := time.Now().UnixMilli() + if err := r.db.Create(&model.Node{ID: 3, Port: "1000-65535"}).Error; err != nil { + t.Fatal(err) + } forwardID, err := r.CreateForwardTx(1, "admin", "proxy-protocol-forward", 2, "1.1.1.1:443", "fifo", now, 1, []int64{3}, 24000, "", nil, 0, 0, nil, 0, 1, 2) if err != nil { t.Fatalf("CreateForwardTx: %v", err) @@ -301,7 +307,10 @@ func TestRollbackForwardFieldsRestoresPerIPLimits(t *testing.T) { defer r.Close() now := time.Now().UnixMilli() - forwardID, err := r.CreateForwardTx(1, "admin", "rollback-per-ip-forward", 2, "1.1.1.1:443", "fifo", now, 1, nil, 0, "", nil, 7, 5, int64(21), 2, 0, 2) + if err := r.db.Create(&model.Node{ID: 3, Port: "1000-65535"}).Error; err != nil { + t.Fatal(err) + } + forwardID, err := r.CreateForwardTx(1, "admin", "rollback-per-ip-forward", 2, "1.1.1.1:443", "fifo", now, 1, []int64{3}, 24000, "", nil, 7, 5, int64(21), 2, 0, 2) if err != nil { t.Fatalf("CreateForwardTx: %v", err) } diff --git a/go-backend/internal/store/repo/repository_mutations.go b/go-backend/internal/store/repo/repository_mutations.go index 5618a6f..b956091 100644 --- a/go-backend/internal/store/repo/repository_mutations.go +++ b/go-backend/internal/store/repo/repository_mutations.go @@ -448,7 +448,14 @@ func (r *Repository) BeginTx() *gorm.DB { if r == nil || r.db == nil { return nil } - return r.db.Begin() + tx := r.db.Begin() + if tx.Error == nil { + if err := lockPortAllocationTx(tx); err != nil { + tx.Rollback() + tx.AddError(err) + } + } + return tx } func (r *Repository) UpdateTunnelOrder(tunnelID int64, inx int, now int64) { @@ -492,6 +499,18 @@ func (r *Repository) CreateChainTunnelTx(tx *gorm.DB, tunnelID int64, chainType if tx == nil { return errors.New("database unavailable") } + if port.Valid && port.Int64 > 0 { + if err := r.LockPortNodesTx(tx, []int64{nodeID}); err != nil { + return err + } + used, err := r.usedPortSetTx(tx, nodeID, 0, tunnelID) + if err != nil { + return err + } + if used[int(port.Int64)] { + return errors.New("端口已占用") + } + } ct := model.ChainTunnel{ TunnelID: tunnelID, ChainType: chainType, @@ -538,6 +557,9 @@ func (r *Repository) pickNodePortTx(tx *gorm.DB, nodeID int64, allocated map[int if nodeID <= 0 { return 0, errors.New("节点不存在") } + if err := r.LockPortNodesTx(tx, []int64{nodeID}); err != nil { + return 0, err + } if port, ok := allocated[nodeID]; ok && port > 0 { return port, nil } @@ -556,47 +578,16 @@ func (r *Repository) pickNodePortTx(tx *gorm.DB, nodeID int64, allocated map[int return 0, errors.New("节点端口已满，无可用端口") } - used := make(map[int]struct{}) - - chainQuery := tx.Model(&model.ChainTunnel{}).Where("node_id = ? AND port > 0", nodeID) - if excludeTunnelID > 0 { - chainQuery = chainQuery.Where("tunnel_id != ?", excludeTunnelID) - } - var chainPorts []int - if err := chainQuery.Pluck("port", &chainPorts).Error; err != nil { - return 0, err - } - for _, p := range chainPorts { - if p > 0 { - used[p] = struct{}{} - } - } - - var forwardPorts []int - if err := tx.Model(&model.ForwardPort{}). - Where("node_id = ? AND port > 0", nodeID). - Pluck("port", &forwardPorts).Error; err != nil { - return 0, err - } - for _, p := range forwardPorts { - if p > 0 { - used[p] = struct{}{} - } - } - - pendingPorts, err := r.PendingTunnelCleanupPortsTx(tx, nodeID, excludeTunnelID) + used, err := r.usedPortSetTx(tx, nodeID, 0, excludeTunnelID) if err != nil { return 0, err } - for _, port := range pendingPorts { - used[port] = struct{}{} - } var available []int for _, candidate := range candidates { if candidate <= 0 { continue } - if _, ok := used[candidate]; ok { + if used[candidate] { continue } available = append(available, candidate) @@ -836,11 +827,64 @@ func (r *Repository) ReplaceForwardPorts(forwardID int64, entries []struct { NodeID int64 Port int InIP string -}) error { +}, sequential ...bool) error { + return r.replaceForwardPorts(forwardID, entries, len(sequential) > 0 && sequential[0], false) +} + +// ReplaceForwardPortsIndividually preserves independent ports when adding entry nodes. +func (r *Repository) ReplaceForwardPortsIndividually(forwardID int64, entries []struct { + NodeID int64 + Port int + InIP string +}, sequential bool) error { + return r.replaceForwardPorts(forwardID, entries, sequential, true) +} + +func (r *Repository) replaceForwardPorts(forwardID int64, entries []struct { + NodeID int64 + Port int + InIP string +}, sequential, individually bool) error { if r == nil || r.db == nil { return errors.New("repository not initialized") } return r.db.Transaction(func(tx *gorm.DB) error { + ids := make([]int64, 0, len(entries)) + autoIDs := []int64{} + for _, e := range entries { + ids = append(ids, e.NodeID) + if e.Port <= 0 { + autoIDs = append(autoIDs, e.NodeID) + } + } + if err := r.LockPortNodesTx(tx, ids); err != nil { + return err + } + autoPort := 0 + if len(autoIDs) > 0 && !individually { + var err error + autoPort, err = r.selectForwardPortTx(tx, autoIDs, 0, forwardID, sequential) + if err != nil { + return err + } + } + entries = append(entries[:0:0], entries...) + for i := range entries { + if entries[i].Port <= 0 { + if individually { + var err error + entries[i].Port, err = r.selectForwardPortTx(tx, []int64{entries[i].NodeID}, 0, forwardID, sequential) + if err != nil { + return err + } + } else { + entries[i].Port = autoPort + } + } + if _, err := r.selectForwardPortTx(tx, []int64{entries[i].NodeID}, entries[i].Port, forwardID, false); err != nil { + return err + } + } if err := tx.Where("forward_id = ?", forwardID).Delete(&model.ForwardPort{}).Error; err != nil { return err } @@ -1362,13 +1406,21 @@ func (r *Repository) EnsureUserTunnelGrant(userID, tunnelID int64) (int64, bool, return ut.ID, true, nil } -func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, now int64, inx int, entryNodeIDs []int64, port int, inIp string, speedID interface{}, maxConn int, ipMaxConn int, ipSpeedID interface{}, proxyProtocol int, proxyProtocolReceive int, proxyProtocolSend int) (int64, error) { +func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, now int64, inx int, entryNodeIDs []int64, port int, inIp string, speedID interface{}, maxConn int, ipMaxConn int, ipSpeedID interface{}, proxyProtocol int, proxyProtocolReceive int, proxyProtocolSend int, sequential ...bool) (int64, error) { if r == nil || r.db == nil { return 0, errors.New("repository not initialized") } _, proxyProtocolSend = normalizeForwardProxyProtocol(proxyProtocol, proxyProtocolReceive, proxyProtocolSend) var forwardID int64 err := r.db.Transaction(func(tx *gorm.DB) error { + if err := r.LockPortNodesTx(tx, entryNodeIDs); err != nil { + return err + } + selectedPort, err := r.selectForwardPortTx(tx, entryNodeIDs, port, 0, len(sequential) > 0 && sequential[0]) + if err != nil { + return err + } + port = selectedPort fwd := model.Forward{ UserID: userID, UserName: userName, diff --git a/go-backend/internal/store/repo/repository_port_allocation.go b/go-backend/internal/store/repo/repository_port_allocation.go new file mode 100644 index 0000000..cca7bd7 --- /dev/null +++ b/go-backend/internal/store/repo/repository_port_allocation.go @@ -0,0 +1,184 @@ +package repo + +import ( + "crypto/rand" + "encoding/json" + "errors" + "fmt" + "go-backend/internal/store/model" + "gorm.io/gorm" + "math/big" + "strings" +) + +// lockPortAllocationTx serializes reservation mutations across processes sharing a database. +// PostgreSQL uses a transaction-scoped advisory lock; SQLite acquires its writer +// lock without changing any row, including when the node table is empty. +// All panel instances sharing this database must use this protocol; mixed old +// binaries and out-of-band SQL writes are not protected by advisory locks. +func lockPortAllocationTx(tx *gorm.DB) error { + if tx.Dialector.Name() == "postgres" { + return tx.Exec("SELECT pg_advisory_xact_lock(?)", int64(0x464c5658504f5254)).Error + } + if tx.Dialector.Name() == "sqlite" { + return tx.Model(&model.Node{}).Where("1 = 0").UpdateColumn("id", gorm.Expr("id")).Error + } + return errors.New("unsupported database for atomic port allocation") +} + +// LockPortNodesTx must precede occupancy reads and reservation writes. +func (r *Repository) LockPortNodesTx(tx *gorm.DB, ids []int64) error { + if err := lockPortAllocationTx(tx); err != nil { + return err + } + for _, id := range ids { + if id <= 0 { + return errors.New("节点不存在") + } + } + return nil +} + +func (r *Repository) usedPortSetTx(tx *gorm.DB, nodeID, excludeForwardID, excludeTunnelID int64, excludeRuntimeIDs ...int64) (map[int]bool, error) { + used := map[int]bool{} + var ports []int + if err := tx.Model(&model.ForwardPort{}).Where("node_id = ? AND forward_id <> ?", nodeID, excludeForwardID).Pluck("port", &ports).Error; err != nil { + return nil, err + } + for _, p := range ports { + used[p] = true + } + query := tx.Model(&model.ChainTunnel{}).Where("node_id = ? AND port > 0", nodeID) + if excludeTunnelID > 0 { + query = query.Where("tunnel_id <> ?", excludeTunnelID) + } + if err := query.Pluck("port", &ports).Error; err != nil { + return nil, err + } + for _, p := range ports { + used[p] = true + } + runtimeQuery := tx.Model(&model.PeerShareRuntime{}).Where("node_id = ? AND (status = 1 OR release_pending <> 0) AND port > 0", nodeID) + if len(excludeRuntimeIDs) > 0 { + runtimeQuery = runtimeQuery.Where("id <> ?", excludeRuntimeIDs[0]) + } + if err := runtimeQuery.Pluck("port", &ports).Error; err != nil { + return nil, err + } + for _, p := range ports { + used[p] = true + } + pending, err := r.PendingTunnelCleanupPortsTx(tx, nodeID, excludeTunnelID) + if err != nil { + return nil, err + } + for _, p := range pending { + used[p] = true + } + return used, nil +} + +// Selection and insert must be in the same transaction after locking all nodes. +func (r *Repository) selectForwardPortTx(tx *gorm.DB, ids []int64, requested int, excludeForwardID int64, sequential bool) (int, error) { + if len(ids) == 0 { + return 0, errors.New("隧道没有可用入口节点") + } + var common []int + for i, id := range ids { + var node model.Node + if err := tx.Where("id = ?", id).First(&node).Error; err != nil { + return 0, err + } + spec := strings.TrimSpace(node.Port) + if spec == "" { + spec = "1000-65535" + if requested > 0 { + spec = "1-65535" + } + } + candidates := parsePortRangeSpec(spec) + used, err := r.usedPortSetTx(tx, id, excludeForwardID, 0) + if err != nil { + return 0, err + } + var remote struct { + Start int `json:"portRangeStart"` + End int `json:"portRangeEnd"` + } + if node.IsRemote == 1 { + _ = json.Unmarshal([]byte(node.RemoteConfig.String), &remote) + } + available := map[int]bool{} + for _, p := range candidates { + if p <= 65535 && !used[p] && (remote.Start <= 0 || remote.End <= 0 || (p >= remote.Start && p <= remote.End)) { + available[p] = true + } + } + if requested > 0 { + if !available[requested] { + return 0, errors.New("端口已占用或不在节点允许范围内") + } + continue + } + if i == 0 { + for _, p := range candidates { + if available[p] { + common = append(common, p) + } + } + } else { + filtered := common[:0] + for _, p := range common { + if available[p] { + filtered = append(filtered, p) + } + } + common = filtered + } + } + if requested > 0 { + return requested, nil + } + if len(common) == 0 { + return 0, errors.New("节点端口已满，无可用端口") + } + if sequential { + return common[0], nil + } + idx, err := rand.Int(rand.Reader, big.NewInt(int64(len(common)))) + if err != nil { + return 0, err + } + return common[idx.Int64()], nil +} + +// checkPeerRuntimePortTx permits only a verified federation-forward attachment +// to share its owning chain reservation. Other new reservations must be free. +func (r *Repository) checkPeerRuntimePortTx(tx *gorm.DB, item *model.PeerShareRuntime) error { + if item.Port <= 0 || (item.Status != 1 && item.ReleasePending == 0) { + return nil + } + var ownedTunnel int64 + if item.Role == "forward" { + var shareID, tunnelID int64 + var port int + _, err := fmt.Sscanf(item.ResourceKey, "federation-forward-%d-%d-%d", &shareID, &tunnelID, &port) + if err == nil && shareID == item.ShareID && port == item.Port && item.ResourceKey == fmt.Sprintf("federation-forward-%d-%d-%d", shareID, tunnelID, port) { + var count int64 + if err := tx.Model(&model.ChainTunnel{}).Where("tunnel_id = ? AND node_id = ? AND port = ?", tunnelID, item.NodeID, item.Port).Count(&count).Error; err != nil { + return err + } + if count > 0 { + ownedTunnel = tunnelID + } + } + } + used, err := r.usedPortSetTx(tx, item.NodeID, 0, ownedTunnel, item.ID) + if err != nil { + return err + } + if used[item.Port] { + return errors.New("端口已占用") + } + return nil +} diff --git a/go-backend/internal/store/repo/repository_port_allocation_test.go b/go-backend/internal/store/repo/repository_port_allocation_test.go new file mode 100644 index 0000000..c271759 --- /dev/null +++ b/go-backend/internal/store/repo/repository_port_allocation_test.go @@ -0,0 +1,326 @@ +package repo + +import ( + "database/sql" + "fmt" + "net/url" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "go-backend/internal/store/model" + "gorm.io/gorm" +) + +func portTestCreate(r *Repository, port int, nodes ...int64) (int64, error) { + return r.CreateForwardTx(1, "admin", "concurrent", 1, "127.0.0.1:80", "fifo", time.Now().UnixMilli(), 0, nodes, port, "", nil, 0, 0, nil, 0, 0, 0, true) +} +func TestAtomicPortAllocationAcrossRepositories(t *testing.T) { + path := filepath.Join(t.TempDir(), "ports.db") + a, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer a.Close() + b, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer b.Close() + runConcurrentPortAllocation(t, a, b) +} + +func runConcurrentPortAllocation(t *testing.T, a, b *Repository) { + for _, id := range []int64{1, 2} { + if err := a.db.Create(&model.Node{ID: id, Name: fmt.Sprint(id), Port: "20000-20031"}).Error; err != nil { + t.Fatal(err) + } + } + const count = 16 + start := make(chan struct{}) + errs := make(chan error, count) + var wg sync.WaitGroup + for i := 0; i < count; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + r := a + if i%2 == 1 { + r = b + } + _, err := portTestCreate(r, 0, 1, 2) + errs <- err + }(i) + } + close(start) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + var rows []model.ForwardPort + if err := a.db.Order("port").Find(&rows).Error; err != nil { + t.Fatal(err) + } + if len(rows) != count*2 { + t.Fatalf("got %d reservations", len(rows)) + } + seen := map[string]bool{} + for _, row := range rows { + k := fmt.Sprintf("%d:%d", row.NodeID, row.Port) + if seen[k] { + t.Fatalf("duplicate %s", k) + } + seen[k] = true + } + explicitStart := make(chan struct{}) + explicitResults := make(chan error, 2) + for _, r := range []*Repository{a, b} { + go func(r *Repository) { <-explicitStart; _, err := portTestCreate(r, 20031, 1, 2); explicitResults <- err }(r) + } + close(explicitStart) + success := 0 + for i := 0; i < 2; i++ { + if <-explicitResults == nil { + success++ + } + } + if success != 1 { + t.Fatalf("explicit port race: %d successes, want exactly 1", success) + } + for p := 20000; p < 20000+count; p++ { + for _, n := range []int{1, 2} { + if !seen[fmt.Sprintf("%d:%d", n, p)] { + t.Fatalf("gap at %d:%d", n, p) + } + } + } +} +func TestAtomicPortAllocationRejectsConflictsAndExhaustion(t *testing.T) { + r, err := Open(":memory:") + if err != nil { + t.Fatal(err) + } + defer r.Close() + if err := r.db.Create(&model.Node{ID: 1, Port: "21000-21001"}).Error; err != nil { + t.Fatal(err) + } + id, err := portTestCreate(r, 21000, 1) + if err != nil { + t.Fatal(err) + } + if _, err := portTestCreate(r, 21000, 1); err == nil { + t.Fatal("explicit collision accepted") + } + if err := r.db.Transaction(func(tx *gorm.DB) error { + return r.CreateChainTunnelTx(tx, 2, "2", 1, sql.NullInt64{Int64: 21000, Valid: true}, "", 0, "", "") + }); err == nil { + t.Fatal("chain collision accepted") + } + if _, err := portTestCreate(r, 0, 1); err != nil { + t.Fatal(err) + } + if _, err := portTestCreate(r, 0, 1); err == nil { + t.Fatal("exhausted range accepted") + } + entries := []struct { + NodeID int64 + Port int + InIP string + }{{1, 21001, ""}} + if err := r.ReplaceForwardPorts(id, entries, true); err == nil { + t.Fatal("replacement collision accepted") + } + var old model.ForwardPort + if err := r.db.Where("forward_id = ?", id).First(&old).Error; err != nil { + t.Fatal(err) + } + if old.Port != 21000 { + t.Fatal("failed replacement lost original reservation") + } +} + +func TestAtomicPortAllocationPostgresAcrossRepositories(t *testing.T) { + dsn := os.Getenv("FLVX_POSTGRES_TEST_DSN") + if dsn == "" { + t.Skip("set FLVX_POSTGRES_TEST_DSN for isolated PostgreSQL allocation test") + } + admin, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatal(err) + } + schema := fmt.Sprintf("port_allocation_%d", time.Now().UnixNano()) + if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { admin.Exec("DROP SCHEMA " + schema + " CASCADE"); admin.Close() }) + scoped := dsn + " search_path=" + schema + if strings.Contains(dsn, "://") { + u, err := url.Parse(dsn) + if err != nil { + t.Fatal(err) + } + q := u.Query() + q.Set("search_path", schema) + u.RawQuery = q.Encode() + scoped = u.String() + } + a, err := OpenPostgres(scoped) + if err != nil { + t.Fatal(err) + } + defer a.Close() + b, err := OpenPostgres(scoped) + if err != nil { + t.Fatal(err) + } + defer b.Close() + runConcurrentPortAllocation(t, a, b) +} + +func TestAtomicPortAllocationRespectsRangesAndPendingCleanup(t *testing.T) { + r, err := Open(":memory:") + if err != nil { + t.Fatal(err) + } + defer r.Close() + nodes := []model.Node{{ID: 1, Port: "23004,23000-23002", IsRemote: 1, RemoteConfig: sql.NullString{String: `{"portRangeStart":23001,"portRangeEnd":23004}`, Valid: true}}, {ID: 2, Port: "23001-23004"}} + if err := r.db.Create(&nodes).Error; err != nil { + t.Fatal(err) + } + if err := r.db.Create(&model.TunnelRuntimeCleanup{TunnelID: 2, NodeID: 2, Kind: "service", Port: 23001}).Error; err != nil { + t.Fatal(err) + } + id, err := portTestCreate(r, 0, 1, 2) + if err != nil { + t.Fatal(err) + } + if got := r.GetMinForwardPort(id).Int64; got != 23002 { + t.Fatalf("got %d, want23002", got) + } + id2, err := portTestCreate(r, 0, 1, 2) + if err != nil { + t.Fatal(err) + } + if got := r.GetMinForwardPort(id2).Int64; got != 23004 { + t.Fatalf("got %d, want23004", got) + } + if err := r.DeleteForwardCascade(id); err != nil { + t.Fatal(err) + } + id3, err := portTestCreate(r, 0, 1, 2) + if err != nil { + t.Fatal(err) + } + if got := r.GetMinForwardPort(id3).Int64; got != 23002 { + t.Fatalf("released gap not reused: %d", got) + } +} + +func TestAtomicPortAllocationIndependentNewEntries(t *testing.T) { + r, err := Open(":memory:") + if err != nil { + t.Fatal(err) + } + defer r.Close() + if err := r.db.Create(&[]model.Node{{ID: 1, Port: "24000-24001"}, {ID: 2, Port: "25000-25001"}}).Error; err != nil { + t.Fatal(err) + } + entries := []struct { + NodeID int64 + Port int + InIP string + }{{1, 0, ""}, {2, 0, ""}} + if err := r.ReplaceForwardPortsIndividually(1, entries, true); err != nil { + t.Fatal(err) + } + var rows []model.ForwardPort + if err := r.db.Order("node_id").Find(&rows).Error; err != nil { + t.Fatal(err) + } + if len(rows) != 2 || rows[0].Port != 24000 || rows[1].Port != 25000 { + t.Fatalf("unexpected independent allocation: %+v", rows) + } + if _, err := r.CreateFederationTunnel("conflict", 1, "tcp", 0, 1, 24000); err == nil { + t.Fatal("federation collision accepted") + } +} + +func TestAtomicPortPeerRuntimeConflicts(t *testing.T) { + path := filepath.Join(t.TempDir(), "mixed.db") + a, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer a.Close() + b, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer b.Close() + if err := a.db.Create(&model.Node{ID: 1, Port: "25000-25010"}).Error; err != nil { + t.Fatal(err) + } + runtime := func(key string, port int) *model.PeerShareRuntime { + return &model.PeerShareRuntime{ShareID: 1, NodeID: 1, Port: port, ResourceKey: key, ReservationID: key, Status: 1} + } + start := make(chan struct{}) + results := make(chan error, 2) + go func() { <-start; _, err := portTestCreate(a, 25000, 1); results <- err }() + go func() { <-start; results <- b.CreatePeerShareRuntime(runtime("peer", 25000)) }() + close(start) + success := 0 + for i := 0; i < 2; i++ { + if <-results == nil { + success++ + } + } + if success != 1 { + t.Fatalf("mixed reservation race succeeded %d times", success) + } + first := runtime("first", 25001) + if err := a.CreatePeerShareRuntime(first); err != nil { + t.Fatal(err) + } + if err := b.CreatePeerShareRuntime(runtime("second", 25001)); err == nil { + t.Fatal("duplicate peer port accepted") + } + first.Role = "middle" + first.Applied = 1 + if err := a.UpdatePeerShareRuntime(first); err != nil { + t.Fatalf("own metadata update: %v", err) + } + owned := runtime("federation-forward-1-22-25002", 25002) + owned.Role = "forward" + if err := a.db.Transaction(func(tx *gorm.DB) error { + return a.CreateChainTunnelTx(tx, 22, "1", 1, sql.NullInt64{Int64: 25002, Valid: true}, "fifo", 0, "tcp", "") + }); err != nil { + t.Fatal(err) + } + if err := a.CreatePeerShareRuntime(runtime("federation-forward-1-22-25002", 25002)); err == nil { + t.Fatal("generic spoof of ownership accepted") + } + if err := a.CreatePeerShareRuntime(owned); err != nil { + t.Fatalf("valid attachment rejected: %v", err) + } + other := runtime("other-forward", 25002) + other.Role = "forward" + if err := b.CreatePeerShareRuntime(other); err == nil { + t.Fatal("unowned forward attachment accepted") + } + pending := runtime("pending", 25003) + pending.Status = 0 + pending.ReleasePending = 1 + if err := a.db.Create(pending).Error; err != nil { + t.Fatal(err) + } + if _, err := portTestCreate(b, 25003, 1); err == nil { + t.Fatal("pending runtime reservation reused") + } +} diff --git a/go-backend/internal/store/repo/repository_tunnel_cleanup.go b/go-backend/internal/store/repo/repository_tunnel_cleanup.go index d9d00d9..1133367 100644 --- a/go-backend/internal/store/repo/repository_tunnel_cleanup.go +++ b/go-backend/internal/store/repo/repository_tunnel_cleanup.go @@ -7,6 +7,9 @@ import ( ) func (r *Repository) QueueTunnelRuntimeCleanupTx(tx *gorm.DB, items []model.TunnelRuntimeCleanup) error { + if err := lockPortAllocationTx(tx); err != nil { + return err + } for _, item := range items { var node model.Node if err := tx.Select("id", "is_remote").Where("id = ?", item.NodeID).First(&node).Error; err != nil { diff --git a/go-backend/tests/contract/postgres_backup_sequences_test.go b/go-backend/tests/contract/postgres_backup_sequences_test.go index 35c6389..a78d562 100644 --- a/go-backend/tests/contract/postgres_backup_sequences_test.go +++ b/go-backend/tests/contract/postgres_backup_sequences_test.go @@ -190,7 +190,7 @@ func TestPostgresBackupImportAdvancesEveryExplicitIDTable(t *testing.T) { {"tunnel", func() error { _, err := r.CreateTunnelTx(r.DB(), "new", 1, 1, 0, 0, 1, nil, 0, "", "", 0); return err }}, {"chain tunnel", func() error { return r.CreateChainTunnelTx(r.DB(), 1, "1", 2, sql.NullInt64{}, "", 1, "tls", "") }}, {"forward", func() error { - _, err := r.CreateForwardTx(1, "old", "new", 1, "127.0.0.1:80", "", 0, 0, nil, 0, "", nil, 0, 0, nil, 0, 0, 0) + _, err := r.CreateForwardTx(1, "old", "new", 1, "127.0.0.1:80", "", 0, 0, []int64{1}, 12346, "", nil, 0, 0, nil, 0, 0, 0) return err }}, {"user tunnel", func() error { return r.InsertUserTunnel(2, 1, nil, 0, 0, 0, 0, 1) }}, @@ -428,6 +428,9 @@ func TestPostgresBackupImportAvoidsCascadeDeleteDeadlock(t *testing.T) { if err != nil { t.Fatal(err) } + if err := r.DB().Create(&model.Node{ID: 1, Name: "entry", Port: "10000-20000"}).Error; err != nil { + t.Fatal(err) + } forward, err := r.CreateForwardTx(user, "delete-me", "old", 1, "127.0.0.1:80", "", 0, 0, []int64{1}, 12345, "", nil, 0, 0, nil, 0, 0, 0) if err != nil { t.Fatal(err) diff --git a/vite-frontend/src/pages/config.tsx b/vite-frontend/src/pages/config.tsx index 4d5006f..e0bc4dc 100644 --- a/vite-frontend/src/pages/config.tsx +++ b/vite-frontend/src/pages/config.tsx @@ -231,6 +231,13 @@ const CONFIG_ITEMS: ConfigItem[] = [ "开启后，普通用户创建或编辑规则时可将目标地址指向 127.0.0.1、10.x.x.x、172.16-31.x.x、192.168.x.x 等本地或内网地址。默认关闭以降低开放代理风险。", type: "switch", }, + { + key: "sequential_port_enabled", + label: "顺次分配入口端口", + description: + "开启后，创建规则时如果不指定入口端口，将从节点端口范围内按从小到大的顺序分配第一个空闲端口（例如范围 100-1000，已用 100、101、103，则下一个为 102，再下一个为 104）。关闭时为随机分配。", + type: "switch", + }, ]; const BACKUP_TYPE_OPTIONS = [ @@ -274,6 +281,7 @@ const getInitialConfigs = (): Record => { "is_commercial", "license_expiry", "hide_footer_brand", + "sequential_port_enabled", ]; const initialConfigs: Record = {};