feat: add sequential port allocation with atomic reservations (#584)

Adapt the optional sequential allocation feature from #583 by xigua761. Preserve commercial licensing and serialize port reservations across SQLite/PostgreSQL instances.

Co-authored-by: xigua761 <xigua1166@outlook.com>

.github/workflows/ci-build.yml

@@ -103,10 +103,10 @@ jobs:
       - name: Download dependencies
         run: go mod download
 
-      - name: Run PostgreSQL contract test
+      - name: Run PostgreSQL contract tests
         env:
           FLVX_POSTGRES_TEST_DSN: 'postgres://flux_test:flux_test_pass@127.0.0.1:5432/flux_test?sslmode=disable'
-        run: go test ./tests/contract -run TestPostgresNodeCreateRepairsMissingIDDefaultContract -count=1
+        run: go test ./tests/contract -run 'TestPostgres(NodeCreateRepairsMissingIDDefaultContract|NewRelayTunnelEmitsMatchingPrivateEndpointAuthentication)$' -count=1
 
   agent:
     name: Build Agent

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 {

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

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

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

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(&current).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
 				}

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
 		}

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

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

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,

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

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

go-backend/internal/store/repo/repository_relay_auth.go

@@ -2,6 +2,8 @@ package repo
 
 import (
 	"errors"
+	"strconv"
+
 	"go-backend/internal/relaysecurity"
 	"go-backend/internal/store/model"
 	"gorm.io/gorm"
@@ -22,7 +24,8 @@ func (r *Repository) RelayEndpointCredentials(tx *gorm.DB, tunnelID, nodeID int6
 	err := find()
 	if errors.Is(err, gorm.ErrRecordNotFound) && create {
 		var existing int64
-		if e := db.Model(&model.ChainTunnel{}).Where("tunnel_id = ? AND node_id = ? AND chain_type = ?", tunnelID, nodeID, role).Count(&existing).Error; e != nil {
+		// chain_tunnel stores roles as text; relay_endpoint_auth uses integers.
+		if e := db.Model(&model.ChainTunnel{}).Where("tunnel_id = ? AND node_id = ? AND chain_type = ?", tunnelID, nodeID, strconv.Itoa(role)).Count(&existing).Error; e != nil {
 			return nil, e
 		}
 		if existing > 0 {

go-backend/internal/store/repo/repository_relay_auth_test.go

@@ -0,0 +1,98 @@
+package repo
+
+import (
+	"fmt"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgtype"
+	"gorm.io/gorm"
+
+	"go-backend/internal/relaysecurity"
+	"go-backend/internal/store/model"
+)
+
+func TestRelayEndpointCredentialsBindsLegacyChainTypeAsText(t *testing.T) {
+	for _, role := range []int{2, 3} {
+		t.Run(strconv.Itoa(role), func(t *testing.T) {
+			r, err := Open(filepath.Join(t.TempDir(), "relay.db"))
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() { _ = r.Close() })
+
+			// SQLite coerces integer parameters for this legacy text column.
+			// Check the actual repository query with pgx's text encoder as well,
+			// so the PostgreSQL regression is caught without a running server.
+			checked := 0
+			if err := r.DB().Callback().Query().After("gorm:query").Register("test:relay_chain_text_parameter", func(db *gorm.DB) {
+				if db.Statement.Table != "chain_tunnel" {
+					return
+				}
+				checked++
+				if len(db.Statement.Vars) != 3 {
+					db.AddError(fmt.Errorf("expected three chain lookup arguments, got %d", len(db.Statement.Vars)))
+					return
+				}
+				encoded, err := pgtype.NewMap().Encode(pgtype.TextOID, pgtype.TextFormatCode, db.Statement.Vars[2], nil)
+				if err != nil {
+					db.AddError(fmt.Errorf("encode chain_type for PostgreSQL: %w", err))
+				} else if string(encoded) != strconv.Itoa(role) {
+					db.AddError(fmt.Errorf("unexpected chain_type argument: %q", encoded))
+				}
+			}); err != nil {
+				t.Fatal(err)
+			}
+
+			tunnel := model.Tunnel{Name: "relay", Type: 2, Flow: 1, RelayAuthMode: relaysecurity.Required}
+			if err := r.DB().Create(&tunnel).Error; err != nil {
+				t.Fatal(err)
+			}
+			const nodeID = int64(7)
+			if err := r.DB().Transaction(func(tx *gorm.DB) error {
+				created, err := r.RelayEndpointCredentials(tx, tunnel.ID, nodeID, role, true)
+				if err != nil {
+					return err
+				}
+				for _, create := range []bool{false, true} {
+					loaded, err := r.RelayEndpointCredentials(tx, tunnel.ID, nodeID, role, create)
+					if err != nil {
+						return err
+					}
+					if *created != *loaded {
+						return fmt.Errorf("reloading endpoint changed credentials")
+					}
+				}
+				return nil
+			}); err != nil {
+				t.Fatalf("create and reload endpoint: %v", err)
+			}
+			if checked != 1 {
+				t.Fatalf("expected one chain lookup for a new endpoint, got %d", checked)
+			}
+
+			chain := model.ChainTunnel{TunnelID: tunnel.ID, NodeID: nodeID, ChainType: strconv.Itoa(role)}
+			if err := r.DB().Create(&chain).Error; err != nil {
+				t.Fatal(err)
+			}
+			if err := r.DB().Where("tunnel_id = ?", tunnel.ID).Delete(&model.RelayEndpointAuth{}).Error; err != nil {
+				t.Fatal(err)
+			}
+			if _, err := r.RelayEndpointCredentials(nil, tunnel.ID, nodeID, role, true); err == nil || !strings.Contains(err.Error(), "refusing regeneration") {
+				t.Fatalf("missing protected credentials must fail closed, got %v", err)
+			}
+			if checked != 2 {
+				t.Fatalf("expected chain lookup before refusing regeneration, got %d", checked)
+			}
+			var count int64
+			if err := r.DB().Model(&model.RelayEndpointAuth{}).Count(&count).Error; err != nil {
+				t.Fatal(err)
+			}
+			if count != 0 {
+				t.Fatal("regenerated credentials for an existing protected endpoint")
+			}
+		})
+	}
+}

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 {

go-backend/tests/contract/federation_dual_panel_contract_test.go

@@ -440,13 +440,15 @@ func TestFederationDualPanelRemoteEntryRuntimeContract(t *testing.T) {
 func insertContractNode(t *testing.T, r *repo.Repository, name, ip, portRange, secret string, status int) int64 {
 	t.Helper()
 	now := time.Now().UnixMilli()
-	if err := r.DB().Exec(`
+	var id int64
+	if err := r.DB().Raw(`
 		INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx)
 		VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-	`, name, secret, ip, ip, "", portRange, "", "v1", 1, 1, 1, now, now, status, "[::]", "[::]", 0).Error; err != nil {
+		RETURNING id
+	`, name, secret, ip, ip, "", portRange, "", "v1", 1, 1, 1, now, now, status, "[::]", "[::]", 0).Row().Scan(&id); err != nil {
 		t.Fatalf("insert node %s: %v", name, err)
 	}
-	return mustLastInsertID(t, r, name)
+	return id
 }
 
 func insertPeerShare(t *testing.T, r *repo.Repository, share *repo.PeerShare) int64 {

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)

go-backend/tests/contract/relay_auth_contract_test.go

@@ -11,11 +11,24 @@ import (
 	"time"
 
 	"go-backend/internal/auth"
+	httpserver "go-backend/internal/http"
+	"go-backend/internal/http/handler"
 	"go-backend/internal/store/repo"
 )
 
 func TestNewRelayTunnelEmitsMatchingPrivateEndpointAuthentication(t *testing.T) {
 	router, r := setupContractRouter(t, "relay-auth-contract")
+	assertNewRelayTunnelEndpointAuthentication(t, router, r)
+}
+
+func TestPostgresNewRelayTunnelEmitsMatchingPrivateEndpointAuthentication(t *testing.T) {
+	r := ownerNamesRepository(t, "postgres")()
+	router := httpserver.NewRouter(handler.New(r, "relay-auth-contract"), "relay-auth-contract")
+	assertNewRelayTunnelEndpointAuthentication(t, router, r)
+}
+
+func assertNewRelayTunnelEndpointAuthentication(t *testing.T, router http.Handler, r *repo.Repository) {
+	t.Helper()
 	server := httptest.NewServer(router)
 	defer server.Close()
 	token, err := auth.GenerateToken(1, "admin", 0, "relay-auth-contract")

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<string, string> => {
     "is_commercial",
     "license_expiry",
     "hide_footer_brand",
+    "sequential_port_enabled",
   ];
   const initialConfigs: Record<string, string> = {};