mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-09-29 07:56:37 +08:00
fix: add forward port occupancy validation and runtime residual cleanup (#278)
* fix: tolerate service not found during forward deletion - Refactor deleteForwardServicesOnNode to handle not-found errors gracefully - Extract deleteForwardServiceCandidates helper for reuse - Add tests for not-found tolerance scenarios - Ensures compatibility with legacy node versions Entire-Checkpoint: a3bacf836c57 * fix: add forward port occupancy validation and runtime residual cleanup - Add forward port occupancy validation on create/update paths - Extend self-occupy recovery to clean residual candidate service names - Add regression tests for address-in-use recovery with legacy runtime residue Fixes port conflict issues when upgrading from 2.1.6 to later versions Entire-Checkpoint: fb0a2aee4cb5
This commit is contained in:
@@ -359,7 +359,12 @@ func (h *Handler) rebindForwardServiceOnSelfOccupiedPort(forward *forwardRecord,
|
||||
return fmt.Errorf("端口 %d 已被其他转发占用", port)
|
||||
}
|
||||
|
||||
if err := h.deleteForwardServicesOnNode(forward, node.ID); err != nil {
|
||||
bases, err := h.forwardServiceBaseCandidates(forward)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := h.deleteForwardServiceBasesOnNode(node.ID, bases); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -377,24 +382,38 @@ func (h *Handler) deleteForwardServicesOnNode(forward *forwardRecord, nodeID int
|
||||
if h == nil || forward == nil {
|
||||
return errors.New("invalid forward delete context")
|
||||
}
|
||||
|
||||
userTunnelID, _, _, err := h.resolveUserTunnelAndLimiter(forward.UserID, forward.TunnelID)
|
||||
bases, err := h.forwardServiceBaseCandidates(forward)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return h.deleteForwardServiceBasesOnNode(nodeID, bases)
|
||||
|
||||
}
|
||||
|
||||
func (h *Handler) forwardServiceBaseCandidates(forward *forwardRecord) ([]string, error) {
|
||||
if h == nil || forward == nil {
|
||||
return nil, errors.New("invalid forward service base context")
|
||||
}
|
||||
userTunnelID, _, _, err := h.resolveUserTunnelAndLimiter(forward.UserID, forward.TunnelID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
userTunnelIDs, err := h.listUserTunnelIDs(forward.UserID, forward.TunnelID)
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
allUserTunnelIDs, err := h.listUserTunnelIDsByUser(forward.UserID)
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
candidateTunnelIDs := make([]int64, 0, len(userTunnelIDs)+len(allUserTunnelIDs))
|
||||
candidateTunnelIDs = append(candidateTunnelIDs, userTunnelIDs...)
|
||||
candidateTunnelIDs = append(candidateTunnelIDs, allUserTunnelIDs...)
|
||||
bases := buildForwardServiceBaseCandidates(forward.ID, forward.UserID, userTunnelID, candidateTunnelIDs)
|
||||
return buildForwardServiceBaseCandidates(forward.ID, forward.UserID, userTunnelID, candidateTunnelIDs), nil
|
||||
|
||||
}
|
||||
|
||||
func (h *Handler) deleteForwardServiceBasesOnNode(nodeID int64, bases []string) error {
|
||||
return deleteForwardServiceCandidates(bases, func(name string) error {
|
||||
payload := map[string]interface{}{
|
||||
"services": []string{name},
|
||||
|
||||
@@ -4,6 +4,8 @@ import (
|
||||
"errors"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"go-backend/internal/store/repo"
|
||||
)
|
||||
|
||||
func TestBuildForwardControlServiceNamesPauseResume(t *testing.T) {
|
||||
@@ -151,6 +153,62 @@ func TestDeleteForwardServiceCandidatesTreatsAllMissingAsSuccess(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestForwardServiceBaseCandidatesIncludesResolvedAndLegacyZero(t *testing.T) {
|
||||
bases := buildForwardServiceBaseCandidates(46, 9, 123, []int64{123, 77, 0})
|
||||
want := []string{"46_9_123", "46_9_77", "46_9_0"}
|
||||
if !reflect.DeepEqual(bases, want) {
|
||||
t.Fatalf("expected %v, got %v", want, bases)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteForwardServiceBasesOnNodeRetriesLegacyZeroResidue(t *testing.T) {
|
||||
bases := []string{"46_9_123", "46_9_0"}
|
||||
called := make([]string, 0)
|
||||
err := deleteForwardServiceCandidates(bases, func(name string) error {
|
||||
called = append(called, name)
|
||||
if name == "46_9_0_tcp" || name == "46_9_0_udp" {
|
||||
return nil
|
||||
}
|
||||
return errors.New("service " + name + " not found")
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
want := []string{"46_9_123_tcp", "46_9_123_udp", "46_9_123", "46_9_0_tcp", "46_9_0_udp", "46_9_0"}
|
||||
if !reflect.DeepEqual(called, want) {
|
||||
t.Fatalf("expected calls %v, got %v", want, called)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateForwardPortAvailabilityRejectsOtherForwardOccupancy(t *testing.T) {
|
||||
h := &Handler{repo: nil}
|
||||
node := &nodeRecord{ID: 9, Name: "test-node"}
|
||||
_ = h
|
||||
_ = node
|
||||
|
||||
rawRepo, err := repo.Open(":memory:")
|
||||
if err != nil {
|
||||
t.Fatalf("open repo: %v", err)
|
||||
}
|
||||
h = &Handler{repo: rawRepo}
|
||||
if err := rawRepo.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(1, 9, 2000)`).Error; err != nil {
|
||||
t.Fatalf("insert forward port: %v", err)
|
||||
}
|
||||
|
||||
err = h.validateForwardPortAvailability(&nodeRecord{ID: 9, Name: "test-node"}, 2000, 2)
|
||||
if err == nil {
|
||||
t.Fatalf("expected occupancy error")
|
||||
}
|
||||
if err.Error() != "节点 test-node 端口 2000 已被其他转发占用" {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
|
||||
err = h.validateForwardPortAvailability(&nodeRecord{ID: 9, Name: "test-node"}, 2000, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("same forward should be allowed, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestControlForwardServiceCommandReturnsHardError(t *testing.T) {
|
||||
bases := []string{"12_34_56"}
|
||||
handled, lastNotFoundErr, err := controlForwardServiceCommand(bases, "PauseService", func(name string) error {
|
||||
|
||||
@@ -1191,6 +1191,10 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) {
|
||||
response.WriteJSON(w, response.ErrDefault(err.Error()))
|
||||
return
|
||||
}
|
||||
if err := h.validateForwardPortAvailability(node, port, 0); err != nil {
|
||||
response.WriteJSON(w, response.ErrDefault(err.Error()))
|
||||
return
|
||||
}
|
||||
}
|
||||
now := time.Now().UnixMilli()
|
||||
inx := h.repo.NextIndex("forward")
|
||||
@@ -1326,6 +1330,10 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) {
|
||||
response.WriteJSON(w, response.ErrDefault(err.Error()))
|
||||
return
|
||||
}
|
||||
if err := h.validateForwardPortAvailability(node, port, id); err != nil {
|
||||
response.WriteJSON(w, response.ErrDefault(err.Error()))
|
||||
return
|
||||
}
|
||||
}
|
||||
now := time.Now().UnixMilli()
|
||||
if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID); err != nil {
|
||||
@@ -3532,6 +3540,20 @@ func validateLocalNodePort(node *nodeRecord, port int) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *Handler) validateForwardPortAvailability(node *nodeRecord, port int, currentForwardID int64) error {
|
||||
if h == nil || h.repo == nil || node == nil || port <= 0 {
|
||||
return nil
|
||||
}
|
||||
occupied, err := h.repo.HasOtherForwardOnNodePort(node.ID, port, currentForwardID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if occupied {
|
||||
return fmt.Errorf("节点 %s 端口 %d 已被其他转发占用", node.Name, port)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func parsePortRangeMinMax(input string) (int, int) {
|
||||
input = strings.TrimSpace(input)
|
||||
if input == "" {
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
# 014 Forward Port Occupancy Validation
|
||||
|
||||
## Checklist
|
||||
|
||||
- [x] Confirm current forward create/update only validates node port range and misses DB-backed occupancy checks for local nodes.
|
||||
- [x] Add shared forward port occupancy validation for create/update paths before runtime dispatch.
|
||||
- [x] Add focused tests covering create/update validation when another forward already uses the same node+port.
|
||||
- [x] Run focused backend handler tests and record the result.
|
||||
|
||||
## Test Record
|
||||
|
||||
- Command: `cd go-backend && go test ./internal/http/handler/...`
|
||||
- Result: passed.
|
||||
@@ -0,0 +1,13 @@
|
||||
# 015 Forward Runtime Port Residual Cleanup
|
||||
|
||||
## Checklist
|
||||
|
||||
- [x] Confirm 2.1.6 used service names with `_0` runtime base while later versions may target resolved `user_tunnel_id`, leaving old runtime services behind after direct upgrade.
|
||||
- [x] Extend self-occupy recovery to clean residual candidate service names and retry update/add when the port is only occupied by self-owned legacy runtime services.
|
||||
- [x] Add regression tests covering address-in-use recovery with legacy `_0` runtime residue.
|
||||
- [x] Run focused backend handler tests and record the result.
|
||||
|
||||
## Test Record
|
||||
|
||||
- Command: `cd go-backend && go test ./internal/http/handler/...`
|
||||
- Result: passed.
|
||||
Reference in New Issue
Block a user