Compare commits

...

1 Commits

Author SHA1 Message Date
sagit 7ba90e8696 fix(backend): resolve federation forward traffic stats and listener disappearance (#208)
* fix(backend): add repository methods for federation forward runtime management

- GetActiveForwardPeerShareRuntimeByServiceName: lookup runtime by share_id and service_name
- MarkForwardPeerShareRuntimeReleasedByServiceName: release runtime by service_name
- ListActiveForwardPeerShareRuntimesByNodeAndServiceName: node-scoped query for flow processing

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

* fix(backend): bind and release federation forward runtimes on service commands

- bindPeerShareForwardRuntimeServices: create runtime if missing, update ServiceName/Port/Applied/Status
- releasePeerShareForwardRuntimeServices: handle deleteservice command to mark runtime released
- parseFederationForwardServiceNamesForRelease: extract service names from delete payload
- Tests: bind creates runtime when missing, release marks runtime as released

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

* fix(backend): scope federation flow lookup by node to avoid cross-share collisions

- flowUpload: use GetNodeBySecret to extract nodeID for flow processing
- processFlowItem: accept nodeID parameter and pass to flow handlers
- processPeerShareFlowByServiceName: try node-scoped query first, fallback to global
- Add warning log when multiple runtimes match (ambiguous)
- Tests: update all processFlowItem calls with nodeID parameter

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

* test(contract): adjust federation dual panel contract expectations

Update assertion for entry share runtime binding behavior after fix

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

---------

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
2026-02-25 14:10:24 +08:00
7 changed files with 328 additions and 25 deletions
+97 -3
View File
@@ -1268,6 +1268,8 @@ func (h *Handler) federationRuntimeCommand(w http.ResponseWriter, r *http.Reques
}
if strings.EqualFold(cmd, "addservice") || strings.EqualFold(cmd, "updateservice") {
h.bindPeerShareForwardRuntimeServices(share, req.Data)
} else if strings.EqualFold(cmd, "deleteservice") {
h.releasePeerShareForwardRuntimeServices(share, req.Data)
}
response.WriteJSON(w, response.OK(res))
}
@@ -1326,6 +1328,45 @@ func parseFederationForwardServiceBindings(data interface{}) []federationForward
return bindings
}
func parseFederationForwardServiceNamesForRelease(data interface{}) []string {
names := make(map[string]struct{})
appendName := func(raw string) {
name := normalizeForwardRuntimeServiceName(raw)
if name == "" {
return
}
if _, _, _, ok := parseFlowServiceIDs(name); !ok {
return
}
names[name] = struct{}{}
}
for _, svcMap := range extractFederationServiceEntries(data) {
appendName(asString(svcMap["name"]))
}
if dataMap, ok := data.(map[string]interface{}); ok {
for _, item := range asAnySlice(dataMap["services"]) {
appendName(asString(item))
}
}
for _, item := range asAnySlice(data) {
appendName(asString(item))
}
if len(names) == 0 {
return nil
}
out := make([]string, 0, len(names))
for name := range names {
out = append(out, name)
}
sort.Strings(out)
return out
}
func (h *Handler) bindPeerShareForwardRuntimeServices(share *repo.PeerShare, data interface{}) {
if h == nil || h.repo == nil || share == nil {
return
@@ -1338,13 +1379,66 @@ func (h *Handler) bindPeerShareForwardRuntimeServices(share *repo.PeerShare, dat
now := time.Now().UnixMilli()
for _, binding := range bindings {
runtime, err := h.repo.GetActiveForwardPeerShareRuntimeByPort(share.ID, binding.Port)
if err != nil || runtime == nil || runtime.Status != 1 {
if err != nil {
continue
}
if runtime.ServiceName == binding.Name && runtime.Applied == 1 {
if runtime == nil {
runtime, err = h.repo.GetActiveForwardPeerShareRuntimeByServiceName(share.ID, binding.Name)
if err != nil {
continue
}
}
if runtime == nil {
_ = h.repo.CreatePeerShareRuntime(&repo.PeerShareRuntime{
ShareID: share.ID,
NodeID: share.NodeID,
ReservationID: randomToken(24),
ResourceKey: fmt.Sprintf("forward-runtime:%d:%s:%d:%s", share.ID, binding.Name, binding.Port, randomToken(8)),
BindingID: "",
Role: "forward",
ChainName: "",
ServiceName: binding.Name,
Protocol: "tcp",
Strategy: "fifo",
Port: binding.Port,
Target: "",
Applied: 1,
Status: 1,
CreatedTime: now,
UpdatedTime: now,
})
continue
}
_ = h.repo.UpdatePeerShareRuntimeServiceName(runtime.ID, binding.Name, now)
if runtime.ServiceName == binding.Name && runtime.Applied == 1 && runtime.Port == binding.Port && runtime.Status == 1 {
continue
}
runtime.ServiceName = binding.Name
runtime.Port = binding.Port
runtime.Applied = 1
runtime.Status = 1
runtime.UpdatedTime = now
if strings.TrimSpace(runtime.Protocol) == "" {
runtime.Protocol = "tcp"
}
if strings.TrimSpace(runtime.Strategy) == "" {
runtime.Strategy = "fifo"
}
_ = h.repo.UpdatePeerShareRuntime(runtime)
}
}
func (h *Handler) releasePeerShareForwardRuntimeServices(share *repo.PeerShare, data interface{}) {
if h == nil || h.repo == nil || share == nil {
return
}
names := parseFederationForwardServiceNamesForRelease(data)
if len(names) == 0 {
return
}
now := time.Now().UnixMilli()
for _, name := range names {
_ = h.repo.MarkForwardPeerShareRuntimeReleasedByServiceName(share.ID, name, now)
}
}
@@ -722,6 +722,123 @@ func TestBindPeerShareForwardRuntimeServicesAcceptsTopLevelServiceArray(t *testi
}
}
func TestBindPeerShareForwardRuntimeServicesCreatesRuntimeWhenMissing(t *testing.T) {
r, err := repo.Open(filepath.Join(t.TempDir(), "panel-bind-create-runtime.db"))
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
t.Cleanup(func() { _ = r.Close() })
h := New(r, "test-jwt-secret")
now := time.Now().UnixMilli()
if err := r.CreatePeerShare(&repo.PeerShare{
Name: "bind-create-runtime-share",
NodeID: 1,
Token: "bind-create-runtime-token",
MaxBandwidth: 0,
CurrentFlow: 0,
PortRangeStart: 26300,
PortRangeEnd: 26320,
IsActive: 1,
CreatedTime: now,
UpdatedTime: now,
}); err != nil {
t.Fatalf("create share: %v", err)
}
share, err := r.GetPeerShareByToken("bind-create-runtime-token")
if err != nil || share == nil {
t.Fatalf("load share: %v", err)
}
h.bindPeerShareForwardRuntimeServices(share, map[string]interface{}{
"services": []interface{}{
map[string]interface{}{"name": "55_2_10_tcp", "addr": "[::]:26301"},
},
})
var count int64
if err := r.DB().Raw(`SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND role = ? AND status = 1`, share.ID, "forward").Scan(&count).Error; err != nil {
t.Fatalf("query runtime count: %v", err)
}
if count != 1 {
t.Fatalf("expected 1 active forward runtime row, got %d", count)
}
var serviceName string
var port int
var applied int
if err := r.DB().Raw(`SELECT service_name, port, applied FROM peer_share_runtime WHERE share_id = ? AND role = ? ORDER BY id DESC LIMIT 1`, share.ID, "forward").Row().Scan(&serviceName, &port, &applied); err != nil {
t.Fatalf("query created runtime: %v", err)
}
if serviceName != "55_2_10" {
t.Fatalf("expected service_name=55_2_10, got %q", serviceName)
}
if port != 26301 {
t.Fatalf("expected port=26301, got %d", port)
}
if applied != 1 {
t.Fatalf("expected applied=1, got %d", applied)
}
}
func TestReleasePeerShareForwardRuntimeServicesMarksRuntimeReleased(t *testing.T) {
r, err := repo.Open(filepath.Join(t.TempDir(), "panel-release-runtime.db"))
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
t.Cleanup(func() { _ = r.Close() })
h := New(r, "test-jwt-secret")
now := time.Now().UnixMilli()
if err := r.CreatePeerShare(&repo.PeerShare{
Name: "release-runtime-share",
NodeID: 1,
Token: "release-runtime-token",
MaxBandwidth: 0,
CurrentFlow: 0,
PortRangeStart: 26400,
PortRangeEnd: 26420,
IsActive: 1,
CreatedTime: now,
UpdatedTime: now,
}); err != nil {
t.Fatalf("create share: %v", err)
}
share, err := r.GetPeerShareByToken("release-runtime-token")
if err != nil || share == nil {
t.Fatalf("load share: %v", err)
}
if err := r.DB().Exec(`
INSERT INTO peer_share_runtime(share_id, node_id, reservation_id, resource_key, binding_id, role, chain_name, service_name, protocol, strategy, port, target, applied, status, created_time, updated_time)
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, share.ID, share.NodeID, "release-r1", "release-rk1", "", "forward", "", "77_2_10", "tcp", "fifo", 26401, "", 1, 1, now, now).Error; err != nil {
t.Fatalf("insert runtime: %v", err)
}
h.releasePeerShareForwardRuntimeServices(share, map[string]interface{}{
"services": []interface{}{"77_2_10_tcp"},
})
var status int
var applied int
var serviceName string
if err := r.DB().Raw(`SELECT status, applied, service_name FROM peer_share_runtime WHERE share_id = ? AND role = ? ORDER BY id DESC LIMIT 1`, share.ID, "forward").Row().Scan(&status, &applied, &serviceName); err != nil {
t.Fatalf("query released runtime: %v", err)
}
if status != 0 {
t.Fatalf("expected status=0 after release, got %d", status)
}
if applied != 0 {
t.Fatalf("expected applied=0 after release, got %d", applied)
}
if serviceName != "" {
t.Fatalf("expected service_name cleared after release, got %q", serviceName)
}
}
func TestValidateFederationCommandPortsAcceptsTopLevelServiceArray(t *testing.T) {
share := &repo.PeerShare{
PortRangeStart: 26200,
+41 -14
View File
@@ -2,9 +2,12 @@ package handler
import (
"encoding/json"
"log"
"strconv"
"strings"
"time"
"go-backend/internal/store/model"
)
const bytesPerGB int64 = 1024 * 1024 * 1024
@@ -30,7 +33,7 @@ type namedConfigItem struct {
Name string `json:"name"`
}
func (h *Handler) processFlowItem(item flowItem) {
func (h *Handler) processFlowItem(nodeID int64, item flowItem) {
serviceName := strings.TrimSpace(item.N)
if serviceName == "" || serviceName == "web_api" {
return
@@ -40,7 +43,7 @@ func (h *Handler) processFlowItem(item flowItem) {
if ok {
inFlow, outFlow := h.scaleFlowByTunnel(forwardID, item.D, item.U)
_ = h.repo.AddFlow(forwardID, userID, userTunnelID, inFlow, outFlow)
h.processPeerShareFlowFromForward(forwardID, serviceName, item)
h.processPeerShareFlowFromForward(forwardID, nodeID, serviceName, item)
if userTunnelID > 0 {
h.enforceFlowPolicies(userID, userTunnelID)
@@ -153,7 +156,7 @@ func (h *Handler) processPeerShareFlow(runtimeID int64, item flowItem) {
h.enforcePeerShareFlowLimit(share.ID)
}
func (h *Handler) processPeerShareFlowFromForward(forwardID int64, serviceName string, item flowItem) {
func (h *Handler) processPeerShareFlowFromForward(forwardID int64, nodeID int64, serviceName string, item flowItem) {
if h == nil || h.repo == nil || forwardID <= 0 {
return
}
@@ -167,22 +170,22 @@ func (h *Handler) processPeerShareFlowFromForward(forwardID int64, serviceName s
if err != nil || forward == nil {
// Forward not found in local database - might be a federation port-forward
// Try to find by service name in peer_share_runtime
h.processPeerShareFlowByServiceName(serviceName, item)
h.processPeerShareFlowByServiceName(nodeID, serviceName, item)
return
}
tunnelName, err := h.repo.GetTunnelName(forward.TunnelID)
if err != nil {
h.processPeerShareFlowByServiceName(serviceName, item)
h.processPeerShareFlowByServiceName(nodeID, serviceName, item)
return
}
shareID, ok := parsePeerShareIDFromFederationTunnelName(tunnelName)
if !ok {
h.processPeerShareFlowByServiceName(serviceName, item)
h.processPeerShareFlowByServiceName(nodeID, serviceName, item)
return
}
if err := h.repo.AddPeerShareCurrentFlow(shareID, delta); err != nil {
h.processPeerShareFlowByServiceName(serviceName, item)
h.processPeerShareFlowByServiceName(nodeID, serviceName, item)
return
}
@@ -207,7 +210,7 @@ func normalizeForwardRuntimeServiceName(serviceName string) string {
return name
}
func (h *Handler) processPeerShareFlowByServiceName(serviceName string, item flowItem) {
func (h *Handler) processPeerShareFlowByServiceName(nodeID int64, serviceName string, item flowItem) {
if h == nil || h.repo == nil || strings.TrimSpace(serviceName) == "" {
return
}
@@ -218,17 +221,41 @@ func (h *Handler) processPeerShareFlowByServiceName(serviceName string, item flo
}
normalized := normalizeForwardRuntimeServiceName(serviceName)
runtimes, err := h.repo.ListActiveForwardPeerShareRuntimesByServiceName(normalized)
if err != nil {
return
}
if len(runtimes) == 0 && normalized != serviceName {
runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByServiceName(serviceName)
var runtimes []model.PeerShareRuntime
var err error
// Try node-scoped query first if nodeID is valid
if nodeID > 0 {
runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByNodeAndServiceName(nodeID, normalized)
if err != nil {
return
}
if len(runtimes) == 0 && normalized != serviceName {
runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByNodeAndServiceName(nodeID, serviceName)
if err != nil {
return
}
}
}
// Fallback to global query if node-scoped query returned nothing or nodeID is invalid
if len(runtimes) == 0 {
runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByServiceName(normalized)
if err != nil {
return
}
if len(runtimes) == 0 && normalized != serviceName {
runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByServiceName(serviceName)
if err != nil {
return
}
}
}
if len(runtimes) != 1 {
if len(runtimes) > 1 {
log.Printf("WARN: ambiguous peer share runtime match for service=%s nodeID=%d count=%d", serviceName, nodeID, len(runtimes))
}
return
}
runtime := runtimes[0]
@@ -44,7 +44,7 @@ func TestProcessFlowItemTracksPeerShareFlowAndEnforcesLimit(t *testing.T) {
}
h := &Handler{repo: r}
h.processFlowItem(flowItem{N: "fed_svc_17", U: 1200, D: 900})
h.processFlowItem(1, flowItem{N: "fed_svc_17", U: 1200, D: 900})
updatedShare, err := r.GetPeerShare(share.ID)
if err != nil || updatedShare == nil {
@@ -120,7 +120,7 @@ func TestProcessFlowItemTracksPeerShareFlowForFederationPortForward(t *testing.T
}
h := &Handler{repo: r}
h.processFlowItem(flowItem{N: "20_2_10", U: 120, D: 80})
h.processFlowItem(1, flowItem{N: "20_2_10", U: 120, D: 80})
updatedShare, err := r.GetPeerShare(share.ID)
if err != nil || updatedShare == nil {
@@ -166,7 +166,7 @@ func TestProcessFlowItemTracksPeerShareFlowByForwardServiceName(t *testing.T) {
}
h := &Handler{repo: r}
h.processFlowItem(flowItem{N: "20_2_10_tcp", U: 120, D: 80})
h.processFlowItem(1, flowItem{N: "20_2_10_tcp", U: 120, D: 80})
updatedShare, err := r.GetPeerShare(share.ID)
if err != nil || updatedShare == nil {
@@ -226,7 +226,7 @@ func TestProcessFlowItemFallsBackToServiceNameWhenForwardIDCollidesAcrossPanels(
}
h := &Handler{repo: r}
h.processFlowItem(flowItem{N: "20_2_10_tcp", U: 120, D: 80})
h.processFlowItem(1, flowItem{N: "20_2_10_tcp", U: 120, D: 80})
updatedShare, err := r.GetPeerShare(share.ID)
if err != nil || updatedShare == nil {
@@ -288,7 +288,7 @@ func TestProcessFlowItemSkipsPeerShareFlowWhenServiceNameIsAmbiguous(t *testing.
}
h := &Handler{repo: r}
h.processFlowItem(flowItem{N: "99_2_10_tcp", U: 120, D: 80})
h.processFlowItem(1, flowItem{N: "99_2_10_tcp", U: 120, D: 80})
updatedA, _ := r.GetPeerShare(shareA.ID)
updatedB, _ := r.GetPeerShare(shareB.ID)
+3 -2
View File
@@ -704,7 +704,8 @@ func (h *Handler) flowConfig(w http.ResponseWriter, r *http.Request) {
func (h *Handler) flowUpload(w http.ResponseWriter, r *http.Request) {
secret := r.URL.Query().Get("secret")
if ok, _ := h.repo.NodeExistsBySecret(secret); !ok {
node, _ := h.repo.GetNodeBySecret(secret)
if node == nil {
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
_, _ = w.Write([]byte("ok"))
return
@@ -715,7 +716,7 @@ func (h *Handler) flowUpload(w http.ResponseWriter, r *http.Request) {
var items []flowItem
if json.Unmarshal([]byte(raw), &items) == nil {
for _, item := range items {
h.processFlowItem(item)
h.processFlowItem(node.ID, item)
}
}
}
@@ -1287,6 +1287,28 @@ func (r *Repository) ListActiveForwardPeerShareRuntimesByServiceName(serviceName
return items, nil
}
func (r *Repository) ListActiveForwardPeerShareRuntimesByNodeAndServiceName(nodeID int64, serviceName string) ([]model.PeerShareRuntime, error) {
if r == nil || r.db == nil {
return nil, errors.New("repository not initialized")
}
serviceName = strings.TrimSpace(serviceName)
if serviceName == "" {
return []model.PeerShareRuntime{}, nil
}
var items []model.PeerShareRuntime
err := r.db.Where("node_id = ? AND service_name = ? AND status = 1 AND role = ?", nodeID, serviceName, "forward").
Order("id ASC").
Find(&items).Error
if err != nil {
return nil, err
}
if items == nil {
items = make([]model.PeerShareRuntime, 0)
}
return items, nil
}
func (r *Repository) ListActiveForwardPeerShareRuntimeServiceNamesByNode(nodeID int64) ([]string, error) {
if r == nil || r.db == nil {
return nil, errors.New("repository not initialized")
@@ -1333,6 +1355,27 @@ func (r *Repository) GetActiveForwardPeerShareRuntimeByPort(shareID int64, port
return &item, nil
}
func (r *Repository) GetActiveForwardPeerShareRuntimeByServiceName(shareID int64, serviceName string) (*model.PeerShareRuntime, error) {
if r == nil || r.db == nil {
return nil, errors.New("repository not initialized")
}
serviceName = strings.TrimSpace(serviceName)
if shareID <= 0 || serviceName == "" {
return nil, nil
}
var item model.PeerShareRuntime
err := r.db.Where("share_id = ? AND service_name = ? AND status = 1 AND role = ?", shareID, serviceName, "forward").
Order("id ASC").
First(&item).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
if err != nil {
return nil, err
}
return &item, nil
}
func (r *Repository) ExistsActivePeerShareRuntimeOnNodePort(nodeID int64, port int) (bool, error) {
if r == nil || r.db == nil {
return false, errors.New("repository not initialized")
@@ -1376,6 +1419,27 @@ func (r *Repository) MarkPeerShareRuntimeReleasedByPort(shareID int64, port int,
}).Error
}
func (r *Repository) MarkForwardPeerShareRuntimeReleasedByServiceName(shareID int64, serviceName string, updatedTime int64) error {
if r == nil || r.db == nil {
return errors.New("repository not initialized")
}
serviceName = strings.TrimSpace(serviceName)
if shareID <= 0 || serviceName == "" {
return nil
}
if updatedTime <= 0 {
updatedTime = unixMilliNow()
}
return r.db.Model(&model.PeerShareRuntime{}).
Where("share_id = ? AND status = 1 AND role = ? AND service_name = ?", shareID, "forward", serviceName).
Updates(map[string]interface{}{
"status": 0,
"applied": 0,
"service_name": "",
"updated_time": updatedTime,
}).Error
}
// ─── FederationTunnelBinding ─────────────────────────────────────────
func (r *Repository) UpsertFederationTunnelBinding(item *model.FederationTunnelBinding) error {
@@ -166,7 +166,7 @@ func TestFederationDualPanelMiddleExitAutoPortContract(t *testing.T) {
assertCount(t, providerRepo, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND status = 1 AND applied = 1`, middleShareID, 1)
assertCount(t, providerRepo, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND status = 1 AND applied = 1`, exitShareID, 1)
assertCount(t, providerRepo, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ?`, entryShareID, 0)
assertCount(t, providerRepo, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND status = 1 AND applied = 1`, entryShareID, 1)
}
func TestFederationDualPanelRemoteDiagnosisContract(t *testing.T) {