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>
This commit is contained in:
sagit
2026-02-25 14:10:24 +08:00
committed by GitHub
parent 0eed74fe10
commit 7ba90e8696
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)
}
}