fix(cluster): 跨节点同名流 KV 冲突重入死锁 + G.711 音频 channels + gotask v1.0.5 根治

api.go

@@ -14,7 +14,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/config"
 
 	myip "github.com/husanpao/ip"

example/cluster-e2e/Dockerfile

@@ -1,5 +1,7 @@
 # Multi-stage build for m7s cluster e2e.
-FROM golang:1.24-alpine AS builder
+# go.mod requires Go 1.26; using golang:1.26-alpine to match local toolchain
+# (the host in this env has go1.26.0 darwin/amd64).
+FROM golang:1.26-alpine AS builder
 RUN apk add --no-cache git
 WORKDIR /src
 COPY go.mod go.sum ./

example/cluster-e2e/config-node-1.yaml

@@ -8,14 +8,14 @@ global:
     dsn: "postgres://postgres:m7s@postgres:5432/m7s?sslmode=disable"
     dbtype: postgres
   storage:
-    type: s3
-    endpoint: http://minio:9000
-    accesskeyid: admin
-    secretaccesskey: m7sm7sm7s
-    bucket: m7s-records
-    region: us-east-1
-    forcepathstyle: true
-    usessl: false
+    s3:
+      endpoint: http://minio:9000
+      accesskeyid: admin
+      secretaccesskey: m7sm7sm7s
+      bucket: m7s-records
+      region: us-east-1
+      forcepathstyle: true
+      usessl: false
 
 rtmp:
   tcp:

example/cluster-e2e/config-node-2.yaml

@@ -8,14 +8,14 @@ global:
     dsn: "postgres://postgres:m7s@postgres:5432/m7s?sslmode=disable"
     dbtype: postgres
   storage:
-    type: s3
-    endpoint: http://minio:9000
-    accesskeyid: admin
-    secretaccesskey: m7sm7sm7s
-    bucket: m7s-records
-    region: us-east-1
-    forcepathstyle: true
-    usessl: false
+    s3:
+      endpoint: http://minio:9000
+      accesskeyid: admin
+      secretaccesskey: m7sm7sm7s
+      bucket: m7s-records
+      region: us-east-1
+      forcepathstyle: true
+      usessl: false
 
 rtmp:
   tcp:

example/cluster-e2e/config-node-3.yaml

@@ -8,14 +8,14 @@ global:
     dsn: "postgres://postgres:m7s@postgres:5432/m7s?sslmode=disable"
     dbtype: postgres
   storage:
-    type: s3
-    endpoint: http://minio:9000
-    accesskeyid: admin
-    secretaccesskey: m7sm7sm7s
-    bucket: m7s-records
-    region: us-east-1
-    forcepathstyle: true
-    usessl: false
+    s3:
+      endpoint: http://minio:9000
+      accesskeyid: admin
+      secretaccesskey: m7sm7sm7s
+      bucket: m7s-records
+      region: us-east-1
+      forcepathstyle: true
+      usessl: false
 
 rtmp:
   tcp:

example/cluster-e2e/smoke.sh

@@ -34,10 +34,12 @@ $COMPOSE up -d --build
 
 step "Waiting for nodes to be ready (60s timeout)"
 deadline=$(($(date +%s) + 60))
+# m7s core prefixes plugin HTTP handlers with /<pluginName>/, so the cluster
+# plugin's /api/cluster/* handlers live under /cluster/api/cluster/*.
 while [ "$(date +%s)" -lt "$deadline" ]; do
-    if curl -fs http://localhost:8081/api/cluster/nodes >/dev/null 2>&1 \
-       && curl -fs http://localhost:8082/api/cluster/nodes >/dev/null 2>&1 \
-       && curl -fs http://localhost:8083/api/cluster/nodes >/dev/null 2>&1; then
+    if curl -fs http://localhost:8081/cluster/api/cluster/nodes >/dev/null 2>&1 \
+       && curl -fs http://localhost:8082/cluster/api/cluster/nodes >/dev/null 2>&1 \
+       && curl -fs http://localhost:8083/cluster/api/cluster/nodes >/dev/null 2>&1; then
         echo "All 3 nodes responding"
         break
     fi
@@ -46,7 +48,7 @@ done
 
 # Scenario 1: cluster membership
 step "Scenario 1: three nodes in cluster"
-count=$(curl -fs http://localhost:8081/api/cluster/nodes | jq '.peers | length')
+count=$(curl -fs http://localhost:8081/cluster/api/cluster/nodes | jq '.peers | length')
 [ "$count" = "3" ] || fail "expected 3 peers, got $count"
 echo "PASS"
 
@@ -96,7 +98,7 @@ echo "records visible from node-2: $records (manual verification — should incl
 # Scenarios 7-8 require pickup of record id + 302 inspection.
 # Scenario 7 (manual): GET /download/<streamPath> from node-2 should 302 to node-1 if file is on node-1.
 step "Scenario 7: download redirect (manual check)"
-echo "Run: curl -v http://localhost:8082/download/live/foo  (expect 302 → node-1)"
+echo "Run: curl -v http://localhost:8082/mp4/download/live/foo  (expect 302 → node-1)"
 
 # Scenario 8: kill node-1, expect m7s/streams/live/foo to disappear within 12s
 step "Scenario 8: kill node-1, expect m7s/streams/live/foo to vanish in <12s"
@@ -121,7 +123,7 @@ echo "Manual: push same live/bar to node-1 and node-2 simultaneously. Expect one
 
 # Scenario 10: lb-suggest works
 step "Scenario 10: /api/cluster/lb-suggest"
-sug=$(curl -fs "http://localhost:8082/api/cluster/lb-suggest?excludeSelf=false" 2>/dev/null || echo '{"suggested":""}')
+sug=$(curl -fs "http://localhost:8082/cluster/api/cluster/lb-suggest?excludeSelf=false" 2>/dev/null || echo '{"suggested":""}')
 echo "lb-suggest from node-2 (with node-1 down): $sug"
 suggested=$(echo "$sug" | jq -r '.suggested // empty' 2>/dev/null || echo "")
 [ -n "$suggested" ] || fail "lb-suggest returned no suggested node"

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/cloudwego/goref v0.0.0-20240724113447-685d2a9523c8
 	github.com/deepch/vdk v0.0.27
 	github.com/disintegration/imaging v1.6.2
+	github.com/eanfs/gotask v1.0.5
 	github.com/emiago/sipgo v1.0.0-alpha
 	github.com/getlantern/systray v1.2.2
 	github.com/go-delve/delve v1.23.0
@@ -25,7 +26,6 @@ require (
 	github.com/icholy/digest v1.1.0
 	github.com/jinzhu/copier v0.4.0
 	github.com/kerberos-io/onvif v1.0.0
-	github.com/langhuihui/gotask v1.0.4
 	github.com/mark3labs/mcp-go v0.27.0
 	github.com/mattn/go-sqlite3 v1.14.24
 	github.com/mcuadros/go-defaults v1.2.0

go.sum

@@ -84,6 +84,8 @@ github.com/deepch/vdk v0.0.27 h1:j/SHaTiZhA47wRpaue8NRp7P9xwOOO/lunxrDJBwcao=
 github.com/deepch/vdk v0.0.27/go.mod h1:JlgGyR2ld6+xOIHa7XAxJh+stSDBAkdNvIPkUIdIywk=
 github.com/disintegration/imaging v1.6.2 h1:w1LecBlG2Lnp8B3jk5zSuNqd7b4DXhcjwek1ei82L+c=
 github.com/disintegration/imaging v1.6.2/go.mod h1:44/5580QXChDfwIclfc/PCwrr44amcmDAg8hxG0Ewe4=
+github.com/eanfs/gotask v1.0.5 h1:LjesxJj59+2iahTwxcRahmGDUn85cuAUNg3desQ2oAE=
+github.com/eanfs/gotask v1.0.5/go.mod h1:EmeY+rFRLYcEXHK6Ebx/Zi9Cr5k/WgY+eb7fJ/Yt0TA=
 github.com/elgs/gostrgen v0.0.0-20220325073726-0c3e00d082f6 h1:x9TA+vnGEyqmWY+eA9HfgxNRkOQqwiEpFE9IPXSGuEA=
 github.com/elgs/gostrgen v0.0.0-20220325073726-0c3e00d082f6/go.mod h1:wruC5r2gHdr/JIUs5Rr1V45YtsAzKXZxAnn/5rPC97g=
 github.com/emiago/sipgo v1.0.0-alpha h1:w98VF4Qq3o+CcKPNe6PIouYy/mQdI66yeQGhYrwXX5Y=
@@ -253,8 +255,6 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/langhuihui/gomem v0.0.0-20251001011839-023923cf7683 h1:lITBgMb71ad6OUU9gycsheCw9PpMbXy3/QA8T0V0dVM=
 github.com/langhuihui/gomem v0.0.0-20251001011839-023923cf7683/go.mod h1:BTPq1+4YUP4i7w8VHzs5AUIdn3T5gXjIUXbxgHW9TIQ=
-github.com/langhuihui/gotask v1.0.4 h1:dYVSR6x+/INXvziuhCLeHhj+5SwmCM+AEHZEAdvsk4A=
-github.com/langhuihui/gotask v1.0.4/go.mod h1:2zNqwV8M1pHoO0b5JC/A37oYpdtXrfL10Qof9AvR5IE=
 github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80 h1:6Yzfa6GP0rIo/kULo2bwGEkFvCePZ3qHDDTC3/J9Swo=
 github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80/go.mod h1:imJHygn/1yfhB7XSJJKlFZKl/J+dCPAknuiaGOshXAs=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4=

pkg/av_reader.go

@@ -5,7 +5,7 @@ import (
 	"log/slog"
 	"time"
 
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/codec"
 	"m7s.live/v5/pkg/config"
 )

pkg/config/quic.go

@@ -5,7 +5,7 @@ import (
 	"crypto/tls"
 	"log/slog"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/quic-go/quic-go"
 )
 

pkg/config/tcp.go

@@ -8,7 +8,7 @@ import (
 	"runtime"
 	"time"
 
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 )
 
 //go:embed local.monibuca.com_bundle.pem

pkg/config/udp.go

@@ -6,7 +6,7 @@ import (
 	"net"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 type UDP struct {

pkg/http_server_fasthttp.go

@@ -6,7 +6,7 @@ import (
 	"crypto/tls"
 	"log/slog"
 
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 	"github.com/valyala/fasthttp"
 	"github.com/valyala/fasthttp/fasthttpadaptor"
 	"m7s.live/v5/pkg/config"

pkg/http_server_std.go

@@ -7,7 +7,7 @@ import (
 	"log/slog"
 	"net/http"
 
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/config"
 )
 

pkg/log.go

@@ -6,7 +6,7 @@ import (
 	"slices"
 	"sync"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 var _ slog.Handler = (*MultiLogHandler)(nil)

pkg/ring-writer.go

@@ -5,7 +5,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/util"
 )
 

pkg/track.go

@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/codec"
 	"m7s.live/v5/pkg/config"
 

pkg/util/manager.go

@@ -1,7 +1,7 @@
 package util
 
 import (
-	. "github.com/langhuihui/gotask"
+	. "github.com/eanfs/gotask"
 )
 
 type Manager[K comparable, T ManagerItem[K]] struct {

plugin.go

@@ -28,7 +28,7 @@ import (
 	"google.golang.org/grpc"
 	"gorm.io/gorm"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	. "m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/db"

plugin/cascade/client.go

@@ -11,7 +11,7 @@ import (
 	"m7s.live/v5/pkg/util"
 	cascade "m7s.live/v5/plugin/cascade/pkg"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/quic-go/quic-go"
 )
 

plugin/cascade/clientManager.go

@@ -14,7 +14,7 @@ import (
 	m7s "m7s.live/v5"
 	cascadepkg "m7s.live/v5/plugin/cascade/pkg"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 // {{ AURA-X: PullProxy 拉流代理数据结构 }}

plugin/cascade/pkg/quic-http.go

@@ -10,7 +10,7 @@ import (
 	"net/http"
 	"strings"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	flv "m7s.live/v5/plugin/flv/pkg"
 
 	"github.com/quic-go/quic-go"

plugin/cascade/server.go

@@ -8,7 +8,7 @@ import (
 	"strings"
 	"sync"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg"

plugin/cluster/apiroute_sync.go

@@ -6,7 +6,7 @@ import (
 	"net/url"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	cfg "m7s.live/v5/pkg/config"
 )
 

plugin/cluster/cluster_helper_test.go

@@ -11,7 +11,7 @@ import (
 	"time"
 
 	consulapi "github.com/hashicorp/consul/api"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	m7s "m7s.live/v5"
 )
 

plugin/cluster/index.go

@@ -126,16 +126,9 @@ func (p *ClusterPlugin) setupRelayHooks() {
 			}
 			p.stopRelayPullProxy(streamPath, ErrOriginLost)
 		})
-		p.streamRegistry.SetOnStopPublisher(func(streamPath string, reason error) {
-			if p.Server == nil {
-				return
-			}
-			pub, ok := p.Server.Streams.SafeGet(streamPath)
-			if !ok {
-				return
-			}
-			pub.Stop(reason)
-		})
+		// 注:first-write-wins 冲突停流已改为 StreamRegistry.OnPublish 直接绑定
+		// pub.Stop(见 streamregistry.go),不再需要 SetOnStopPublisher + SafeGet 反查
+		// (SafeGet 会重入 Server.Streams 事件循环导致死锁,RC1)。
 	})
 }
 
@@ -147,6 +140,16 @@ func (p *ClusterPlugin) OnPublish(pub *m7s.Publisher) {
 	}
 }
 
+// isActiveRelay 报告 streamPath 是否是本节点上一条 cluster-relay 派生的流。
+// ensureRelay 成功后会把 streamPath 写进 activeRelays;StreamRegistry.OnPublish
+// 用它作为 relay publisher 的权威判据(避免依赖易丢的 PullProxyConfig.Description)。
+func (p *ClusterPlugin) isActiveRelay(streamPath string) bool {
+	p.activeRelaysMu.Lock()
+	defer p.activeRelaysMu.Unlock()
+	_, ok := p.activeRelays[streamPath]
+	return ok
+}
+
 // Membership 暴露给同包内其它模块(Phase 3/4)读取 peers 与 sessionID。
 func (p *ClusterPlugin) Membership() *Membership { return p.membership }
 

plugin/cluster/membership.go

@@ -11,7 +11,7 @@ import (
 	"time"
 
 	consulapi "github.com/hashicorp/consul/api"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 // PeerInfo 是 Consul KV `m7s/nodes/<nodeID>` 的 JSON 序列化结构。

plugin/cluster/metrics.go

@@ -6,7 +6,7 @@ import (
 	"runtime"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 // LoadReporter 周期把本节点指标写到 m7s/nodes/<self> 的 Metrics 字段。

plugin/cluster/metrics_test.go

@@ -9,7 +9,7 @@ import (
 	"time"
 
 	consulapi "github.com/hashicorp/consul/api"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 // TestLoadReporter_UpdatesMetricsField: 启动 LoadReporter,等一个 tick,