From 6a9cf4d7068d18ec8a4819ef3a3a39b4d9b841db Mon Sep 17 00:00:00 2001
From: "richard.li" <lirichen32@gmail.com>
Date: Wed, 3 Jun 2026 12:47:42 +0800
Subject: [PATCH 1/4] =?UTF-8?q?fix(cluster):=20=E4=BF=AE=E5=A4=8D=E8=B7=A8?=
 =?UTF-8?q?=E8=8A=82=E7=82=B9=E5=90=8C=E5=90=8D=E6=B5=81=20KV=20=E5=86=B2?=
 =?UTF-8?q?=E7=AA=81=E5=AF=BC=E8=87=B4=20Streams=20=E4=BA=8B=E4=BB=B6?=
 =?UTF-8?q?=E5=BE=AA=E7=8E=AF=E9=87=8D=E5=85=A5=E6=AD=BB=E9=94=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

非属主节点对与属主同名的 streamPath 拉流/订阅时,first-write-wins 冲突处理
在 Server.Streams 事件循环上同步执行 acquire(阻塞 Consul I/O)并经
Server.Streams.SafeGet(m.Call) 重入同一循环,gotask 的 Call 不识别已在本
循环 goroutine,入队+阻塞导致永久死锁,stream-list 与新订阅/发布全挂。

- RC1: OnPublish 直接绑定 pub.Stop 闭包,不再 SafeGet 反查(消除重入)
- RC2: acquire+冲突停流经 sr.dispatch(一次性 callbackGoTask)异步挪到
  StreamRegistry 自己的事件循环,离开 Server.Streams 调用栈
- RC3: relay publisher 判据改用 cluster 插件权威自持的 activeRelays
  (isActiveRelay),而非易丢的 PullProxyConfig.Description(EnsurePullProxy
  命中已存在 pull-proxy 会丢弃新 conf 的 cluster-relay 标记)
- 次生: 加 acquired 标志,冲突未拥有键时不 release,避免删掉属主 peer 的 KV 键

新增回归守卫: ConflictHandlingOffloaded / ConflictDoesNotReleasePeerKey /
SuccessfulAcquireReleasesOnDispose / IsActiveRelay。
真机回归(3 节点活集群): incident2 同名拉流干净停止+不死锁+属主保留;
incident1 auto-relay 无 acquire 失败+15s/147帧持续供流。
---
 plugin/cluster/index.go               |  23 +++--
 plugin/cluster/streamregistry.go      | 112 +++++++++++++++--------
 plugin/cluster/streamregistry_test.go | 126 +++++++++++++++++++++++---
 3 files changed, 202 insertions(+), 59 deletions(-)

diff --git a/plugin/cluster/index.go b/plugin/cluster/index.go
index deae3dde..c405ad49 100644
--- a/plugin/cluster/index.go
+++ b/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 }
 
diff --git a/plugin/cluster/streamregistry.go b/plugin/cluster/streamregistry.go
index f5bbef7c..3c5737e3 100644
--- a/plugin/cluster/streamregistry.go
+++ b/plugin/cluster/streamregistry.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	consulapi "github.com/hashicorp/consul/api"
@@ -18,7 +19,9 @@ import (
 // pull-proxy 是 cluster relay 派生的(而非用户配置的常规拉流代理)。
 //
 // Phase 3 创建 cluster-relay PullProxy 时会把 Description 设置成
-//   "cluster-relay:<originNodeID>"
+//
+//	"cluster-relay:<originNodeID>"
+//
 // StreamRegistry.OnPublish 看到带这个前缀的 publisher 直接跳过——
 // 避免环回(B 从 A 拉到流后又把流写进 Consul,导致 C 上的订阅者去 B 拉)。
 const ClusterRelayDescPrefix = "cluster-relay:"
@@ -41,16 +44,30 @@ type StreamRegistry struct {
 	onStreamRemovedMu sync.Mutex
 	onStreamRemoved   []func(streamPath string)
 
-	onStopPublisherMu sync.Mutex
-	onStopPublisher   func(streamPath string, reason error)
+	// acquireFn/releaseFn 默认指向 sr.acquire/sr.release;抽成字段是为了单测能注入
+	// fake,不依赖真实 Consul。
+	acquireFn func(streamPath string) error
+	releaseFn func(streamPath string) error
+
+	// dispatch 把一段工作异步投递到 StreamRegistry 自己的事件循环(默认 = 一次性
+	// callbackGoTask)。OnPublish 用它把阻塞式 KV acquire 与 first-write-wins 冲突
+	// 停流挪出 Server.Streams 事件循环 —— 避免重入 Streams.Call 死锁(RC1)与阻塞
+	// Streams 事件循环(RC2)。单测可注入同步/捕获版。
+	dispatch func(func())
 }
 
 func newStreamRegistry(p *ClusterPlugin) *StreamRegistry {
-	return &StreamRegistry{
+	sr := &StreamRegistry{
 		plugin:       p,
 		streams:      make(map[string]string),
 		localStreams: make(map[string]struct{}),
 	}
+	sr.acquireFn = sr.acquire
+	sr.releaseFn = sr.release
+	sr.dispatch = func(fn func()) {
+		sr.AddTask(&callbackGoTask{fn: fn})
+	}
+	return sr
 }
 
 func (sr *StreamRegistry) Start() error {
@@ -67,15 +84,31 @@ func (sr *StreamRegistry) Start() error {
 // OnPublish 在 ClusterPlugin.OnPublish 转发过来时被调用。
 // 把 *m7s.Publisher 解耦成简单参数后交给 handleLocalPublish。
 func (sr *StreamRegistry) OnPublish(pub *m7s.Publisher) {
-	isClusterRelay := pub.PullProxyConfig != nil &&
-		strings.HasPrefix(pub.PullProxyConfig.Description, ClusterRelayDescPrefix)
-	sr.handleLocalPublish(pub.StreamPath, isClusterRelay, pub.OnDispose)
+	// RC3 修复:relay publisher 的判据不能只看 pub.PullProxyConfig.Description ——
+	// 该 Description 不可靠(EnsurePullProxy 命中已存在的同 streamPath pull-proxy 会
+	// 直接返回、丢弃新 conf 的 cluster-relay 标记,见 pull_proxy.go EnsurePullProxy)。
+	// 改以 cluster 插件权威自持的 activeRelays(ensureRelay 同步写入)为准,
+	// Description 仅作兜底信号。
+	isClusterRelay := (pub.PullProxyConfig != nil &&
+		strings.HasPrefix(pub.PullProxyConfig.Description, ClusterRelayDescPrefix)) ||
+		(sr.plugin != nil && sr.plugin.isActiveRelay(pub.StreamPath))
+	// stop 直接绑定到该 publisher 的 Stop —— 不再经 Server.Streams.SafeGet 反查。
+	// SafeGet 会重入 Server.Streams 事件循环(m.Call),而 OnPublish 本身就跑在该
+	// 事件循环上,重入会永久死锁(RC1)。
+	sr.handleLocalPublish(pub.StreamPath, isClusterRelay, func(reason error) {
+		pub.Stop(reason)
+	}, pub.OnDispose)
 }
 
 // handleLocalPublish 是 OnPublish 的可测试核心。
 //   - isClusterRelay: 来自 cluster-relay 派生的 publisher(Q2)直接跳过,避免环回
+//   - stop: 停掉该 publisher(生产 = pub.Stop),first-write-wins 冲突时调用
 //   - registerOnDispose: 通常是 Publisher.OnDispose,测试可传 nil 自行管理生命周期
-func (sr *StreamRegistry) handleLocalPublish(streamPath string, isClusterRelay bool, registerOnDispose func(func())) {
+//
+// ⚠️ 调用方是 Server.Streams 事件循环 goroutine。KV acquire 是阻塞式 Consul I/O,
+// 冲突停流又可能重入 Streams —— 两者都必须经 sr.dispatch 异步挪走,绝不能在本函数
+// 的同步调用栈里执行(否则死锁 Streams 事件循环,RC1/RC2)。
+func (sr *StreamRegistry) handleLocalPublish(streamPath string, isClusterRelay bool, stop func(error), registerOnDispose func(func())) {
 	if isClusterRelay || streamPath == "" {
 		return
 	}
@@ -84,28 +117,39 @@ func (sr *StreamRegistry) handleLocalPublish(streamPath string, isClusterRelay b
 	sr.localStreams[streamPath] = struct{}{}
 	sr.localMu.Unlock()
 
-	if err := sr.acquire(streamPath); err != nil {
-		sr.Warn("acquire stream key failed", "streamPath", streamPath, "error", err)
-		// §4.3 first-write-wins: 失败意味着另一个 peer 已经拥有该 streamPath。
-		// 主动 Stop 本地 publisher,reason = ErrStreamPathTaken。
-		// 注:网络抖动也可能导致 acquire 失败,本 spec v1 把所有失败都当 "key 已被占",
-		// 实际生产里这是保守做法 —— 真的撞键 vs 网络抖动,效果都是 publisher 重连。
-		sr.fireStopPublisher(streamPath, ErrStreamPathTaken)
-		// localStreams 已记录,session 重建时仍会试 rebind;但我们已经请求 Stop,
-		// publisher 不久后会消失,localStreams 在 dispose hook 里也会被清。
-		// 不主动从 localStreams 删,避免与 dispose 钩子竞争。
-	}
+	// acquired 标记我们是否真正抢到了这个 streamPath 的 KV 键。只有抢到了,dispose
+	// 时才能 release —— 否则 release 的 KV.Delete 会把属主 peer 的键删掉。
+	var acquired atomic.Bool
 
 	if registerOnDispose != nil {
 		registerOnDispose(func() {
 			sr.localMu.Lock()
 			delete(sr.localStreams, streamPath)
 			sr.localMu.Unlock()
-			if err := sr.release(streamPath); err != nil {
-				sr.Warn("release stream key failed", "streamPath", streamPath, "error", err)
+			if acquired.Load() {
+				if err := sr.releaseFn(streamPath); err != nil {
+					sr.Warn("release stream key failed", "streamPath", streamPath, "error", err)
+				}
 			}
 		})
 	}
+
+	// RC1+RC2 修复:KV acquire 阻塞式 Consul I/O + first-write-wins 失败停流,异步
+	// dispatch 到 StreamRegistry 自己的事件循环,彻底离开 Server.Streams 调用栈。
+	sr.dispatch(func() {
+		if err := sr.acquireFn(streamPath); err != nil {
+			sr.Warn("acquire stream key failed", "streamPath", streamPath, "error", err)
+			// §4.3 first-write-wins: 失败意味着另一个 peer 已经拥有该 streamPath。
+			// 主动 Stop 本地 publisher,reason = ErrStreamPathTaken。
+			// 注:网络抖动也可能导致 acquire 失败,本 spec v1 把所有失败都当 "key 已被占",
+			// 实际生产里这是保守做法 —— 真的撞键 vs 网络抖动,效果都是 publisher 重连。
+			if stop != nil {
+				stop(ErrStreamPathTaken)
+			}
+			return
+		}
+		acquired.Store(true)
+	})
 }
 
 // Lookup 给 Phase 3 (relay) / Phase 4 (StreamRouter) 用,
@@ -135,25 +179,17 @@ func (sr *StreamRegistry) fireStreamRemoved(streamPath string) {
 	}
 }
 
-// SetOnStopPublisher 给 cluster 主插件设置一个回调:当 handleLocalPublish 发现
-// streamPath 已被别的 peer 占据(§4.3 first-write-wins 失败),用这个回调主动
-// Stop 本节点的 publisher 并报 ErrStreamPathTaken。
-//
-// 单一回调而非多 listener: 一个 streamPath 在一个进程里只对应一个 publisher,
-// 也只有 cluster 主插件知道怎么 Stop 它(通过 m7s API)。
-func (sr *StreamRegistry) SetOnStopPublisher(f func(streamPath string, reason error)) {
-	sr.onStopPublisherMu.Lock()
-	defer sr.onStopPublisherMu.Unlock()
-	sr.onStopPublisher = f
+// callbackGoTask 是一次性任务:在 StreamRegistry 自己的事件循环里(经 Go() 的独立
+// goroutine)跑一段回调然后完成退出。用于把 OnPublish 的阻塞式 KV acquire 与冲突
+// 停流挪出 Server.Streams 事件循环(RC1/RC2)。
+type callbackGoTask struct {
+	task.Task
+	fn func()
 }
 
-func (sr *StreamRegistry) fireStopPublisher(streamPath string, reason error) {
-	sr.onStopPublisherMu.Lock()
-	cb := sr.onStopPublisher
-	sr.onStopPublisherMu.Unlock()
-	if cb != nil {
-		cb(streamPath, reason)
-	}
+func (t *callbackGoTask) Go() error {
+	t.fn()
+	return task.ErrTaskComplete
 }
 
 // Streams 返回当前 streams map 的快照,主要给 /api/cluster/streams 用。
diff --git a/plugin/cluster/streamregistry_test.go b/plugin/cluster/streamregistry_test.go
index 2abe988a..4b408234 100644
--- a/plugin/cluster/streamregistry_test.go
+++ b/plugin/cluster/streamregistry_test.go
@@ -17,7 +17,7 @@ import (
 // 纯单元测试,不需要 consul。
 func TestStreamRegistry_HandleLocalPublishSkipsClusterRelay(t *testing.T) {
 	sr := &StreamRegistry{localStreams: make(map[string]struct{})}
-	sr.handleLocalPublish("live/foo", true, func(f func()) {
+	sr.handleLocalPublish("live/foo", true, nil, func(f func()) {
 		t.Fatalf("registerOnDispose must not be called for cluster-relay publisher")
 	})
 	if len(sr.localStreams) != 0 {
@@ -29,7 +29,7 @@ func TestStreamRegistry_HandleLocalPublishSkipsClusterRelay(t *testing.T) {
 // 为空时(理论不该发生)整条路径必须 no-op,不能 panic、不能写 consul。
 func TestStreamRegistry_HandleLocalPublishSkipsEmptyStreamPath(t *testing.T) {
 	sr := &StreamRegistry{localStreams: make(map[string]struct{})}
-	sr.handleLocalPublish("", false, func(f func()) {
+	sr.handleLocalPublish("", false, nil, func(f func()) {
 		t.Fatalf("registerOnDispose must not be called for empty streamPath")
 	})
 	if len(sr.localStreams) != 0 {
@@ -219,16 +219,14 @@ func TestStreamRegistry_HandleLocalPublishStopsPublisherWhenKeyOwnedByPeer(t *te
 		t.Fatalf("seed acquire ok=%v err=%v", ok, err)
 	}
 
-	// 注入一个 onStopReason 通道。
+	// 注入一个 stop 通道(新设计:停流回调由 OnPublish 直接传入 = pub.Stop,不再 SafeGet)。
 	stopCh := make(chan error, 1)
-	sr.SetOnStopPublisher(func(sp string, reason error) {
-		if sp == streamPath {
-			stopCh <- reason
-		}
-	})
+	stop := func(reason error) { stopCh <- reason }
 
-	// 模拟 OnPublish: 不是 cluster-relay,非空 streamPath,registerOnDispose=nil。
-	sr.handleLocalPublish(streamPath, false, nil)
+	// 模拟 OnPublish: 不是 cluster-relay,非空 streamPath,stop=spy,registerOnDispose=nil。
+	// 真实路径下 acquire+停流被 dispatch 到 StreamRegistry 自身事件循环异步执行
+	// (sr 已 start),所以这里 stop 会异步收到 ErrStreamPathTaken。
+	sr.handleLocalPublish(streamPath, false, stop, nil)
 
 	select {
 	case got := <-stopCh:
@@ -236,6 +234,112 @@ func TestStreamRegistry_HandleLocalPublishStopsPublisherWhenKeyOwnedByPeer(t *te
 			t.Fatalf("stop reason = %v, want ErrStreamPathTaken", got)
 		}
 	case <-time.After(2 * time.Second):
-		t.Fatalf("onStopPublisher not fired within 2s")
+		t.Fatalf("stop not fired within 2s")
+	}
+}
+
+// TestClusterPlugin_IsActiveRelay 验证 RC3 修复的权威 relay 判据:ensureRelay 写入
+// activeRelays 后,isActiveRelay 必须返回 true;未注册的返回 false。
+func TestClusterPlugin_IsActiveRelay(t *testing.T) {
+	p := &ClusterPlugin{}
+	if p.isActiveRelay("live/x") {
+		t.Fatal("empty activeRelays must report false")
+	}
+	p.activeRelaysMu.Lock()
+	p.activeRelays = map[string]struct{}{"live/relayed": {}}
+	p.activeRelaysMu.Unlock()
+	if !p.isActiveRelay("live/relayed") {
+		t.Fatal("registered active relay must report true")
+	}
+	if p.isActiveRelay("live/other") {
+		t.Fatal("unregistered streamPath must report false")
+	}
+}
+
+// TestStreamRegistry_ConflictHandlingOffloaded 是这次死锁修复的核心回归守卫(RC1+RC2)。
+//
+// 缺陷:OnPublish 在 Server.Streams 事件循环 goroutine 上同步跑 acquire(阻塞 Consul I/O)
+// 和冲突停流,停流又重入 Server.Streams.SafeGet(m.Call)→ 永久死锁整个 Streams 事件循环。
+//
+// 契约:handleLocalPublish 必须把 acquire+冲突停流 **dispatch 出去异步执行**,
+// 绝不在调用方 goroutine 上同步跑 acquire / stop。
+//
+// 纯单元测试,不依赖 consul / 事件循环:注入 dispatch 捕获被 offload 的闭包。
+func TestStreamRegistry_ConflictHandlingOffloaded(t *testing.T) {
+	sr := &StreamRegistry{localStreams: make(map[string]struct{})}
+
+	acquireCalled := false
+	sr.acquireFn = func(string) error {
+		acquireCalled = true
+		return errors.New("kv acquire returned false; held by peer")
+	}
+	var captured func()
+	sr.dispatch = func(fn func()) { captured = fn } // 捕获,不同步执行
+
+	stopCalled := false
+	var stopReason error
+	sr.handleLocalPublish("live/contested", false, func(r error) {
+		stopCalled = true
+		stopReason = r
+	}, nil)
+
+	// 关键断言:绝不能在调用方 goroutine 上同步跑 acquire / stop(否则就是会死锁的旧行为)。
+	if acquireCalled {
+		t.Fatal("acquire ran synchronously on caller goroutine — must be offloaded (RC1/RC2)")
+	}
+	if stopCalled {
+		t.Fatal("stop ran synchronously on caller goroutine — re-entrancy deadlock risk (RC1)")
+	}
+	if captured == nil {
+		t.Fatal("conflict handling was not dispatched off the caller goroutine")
+	}
+
+	// 执行被 offload 的闭包,验证它会 acquire 并在冲突时用 ErrStreamPathTaken 停流。
+	captured()
+	if !acquireCalled {
+		t.Fatal("dispatched closure did not call acquire")
+	}
+	if !stopCalled || !errors.Is(stopReason, ErrStreamPathTaken) {
+		t.Fatalf("dispatched closure must stop with ErrStreamPathTaken; stopCalled=%v reason=%v", stopCalled, stopReason)
+	}
+}
+
+// TestStreamRegistry_ConflictDoesNotReleasePeerKey 守护一个次生 bug:
+// acquire 冲突(键属于别的 peer)时,本节点 publisher 被停;其 dispose hook
+// 绝不能调 release —— release 会 KV.Delete 这个键,把属主 peer 的所有权删掉。
+func TestStreamRegistry_ConflictDoesNotReleasePeerKey(t *testing.T) {
+	sr := &StreamRegistry{localStreams: make(map[string]struct{})}
+	sr.acquireFn = func(string) error { return errors.New("held by peer") }
+	releaseCalled := false
+	sr.releaseFn = func(string) error { releaseCalled = true; return nil }
+	sr.dispatch = func(fn func()) { fn() } // 同步执行便于断言
+
+	var disposeHook func()
+	sr.handleLocalPublish("live/x", false, func(error) {}, func(h func()) { disposeHook = h })
+
+	if disposeHook == nil {
+		t.Fatal("dispose hook should still be registered")
+	}
+	disposeHook() // 模拟 publisher dispose
+	if releaseCalled {
+		t.Fatal("must NOT release/delete KV key on conflict — it belongs to the peer node")
+	}
+}
+
+// TestStreamRegistry_SuccessfulAcquireReleasesOnDispose 正向:成功 acquire 后,
+// dispose 时必须 release 自己拥有的键。
+func TestStreamRegistry_SuccessfulAcquireReleasesOnDispose(t *testing.T) {
+	sr := &StreamRegistry{localStreams: make(map[string]struct{})}
+	sr.acquireFn = func(string) error { return nil } // 成功
+	releaseCalled := false
+	sr.releaseFn = func(string) error { releaseCalled = true; return nil }
+	sr.dispatch = func(fn func()) { fn() }
+
+	var disposeHook func()
+	sr.handleLocalPublish("live/x", false, func(error) {}, func(h func()) { disposeHook = h })
+
+	disposeHook()
+	if !releaseCalled {
+		t.Fatal("must release acquired key on dispose")
 	}
 }

From 5c9eb04366216c50159de177902f9e962860a5a6 Mon Sep 17 00:00:00 2001
From: "richard.li" <lirichen32@gmail.com>
Date: Wed, 3 Jun 2026 13:16:06 +0800
Subject: [PATCH 2/4] =?UTF-8?q?fix(rtsp):=20G.711(PCMA/PCMU)=E9=9F=B3?=
 =?UTF-8?q?=E9=A2=91=E5=BD=95=E5=88=B6=20channels=3D0=20=E5=AF=BC=E8=87=B4?=
 =?UTF-8?q?=20mp4=20=E5=85=83=E6=95=B0=E6=8D=AE=E9=9D=9E=E6=B3=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

RTSP 收 G.711 时用空 codec.PCMACtx{}/PCMUCtx{}(Channels/SampleSize=0),
而 G.711 的 SDP 'PCMA/8000' 不带声道数,RTPCodecParameters.Channels=0 时
直接赋值会写出 channels=0,录制的 mp4 音频轨元数据非法,ffprobe 无法解析
整文件(Failed to open codec / 0 channels)。

- 改用 codec.NewPCMACtx()/NewPCMUCtx() 作默认(G.711 mono: SampleRate=8000,
  Channels=1, SampleSize=16),仅当 SDP 给出有效值(>0)才覆盖
- 真机验证: node-3 录 camera27 → ffprobe channels=1 + 全文件解码 exit=0

附带: EnsurePullProxy 命中已存在 pull-proxy 复用、丢弃新 conf 时加非静默
Debug 日志(行为不变,正确语义不覆盖已有配置;relay 标记已由 activeRelays
权威判定)。
---
 plugin/rtsp/pkg/transceiver.go | 24 ++++++++++++++++++------
 pull_proxy.go                  | 10 ++++++++++
 2 files changed, 28 insertions(+), 6 deletions(-)

diff --git a/plugin/rtsp/pkg/transceiver.go b/plugin/rtsp/pkg/transceiver.go
index 5ac8534f..0e2d0b47 100644
--- a/plugin/rtsp/pkg/transceiver.go
+++ b/plugin/rtsp/pkg/transceiver.go
@@ -473,18 +473,30 @@ func (r *Receiver) Receive() (err error) {
 			writer.AudioFrame.ICodecCtx = &ctx
 		case webrtc.MimeTypePCMA:
 			var ctx mrtp.PCMACtx
-			ctx.PCMACtx = &codec.PCMACtx{}
+			// NewPCMACtx 给出 G.711 默认值(SampleRate=8000, Channels=1, SampleSize=16)。
+			// 空 struct 会让 Channels/SampleSize=0:G.711 的 SDP "PCMA/8000" 不带声道数,
+			// RTPCodecParameters.Channels=0 时若直接赋值会写出 channels=0,导致录制 mp4
+			// 音频轨元数据非法、ffprobe 无法解析整文件。
+			ctx.PCMACtx = codec.NewPCMACtx()
 			ctx.ParseFmtpLine(r.AudioCodecParameters)
-			ctx.AudioCtx.SampleRate = int(r.AudioCodecParameters.ClockRate)
-			ctx.AudioCtx.Channels = int(ctx.RTPCodecParameters.Channels)
+			if r.AudioCodecParameters.ClockRate > 0 {
+				ctx.AudioCtx.SampleRate = int(r.AudioCodecParameters.ClockRate)
+			}
+			if ctx.RTPCodecParameters.Channels > 0 {
+				ctx.AudioCtx.Channels = int(ctx.RTPCodecParameters.Channels)
+			}
 			audioClockRate = ctx.RTPCodecParameters.ClockRate
 			writer.AudioFrame.ICodecCtx = &ctx
 		case webrtc.MimeTypePCMU:
 			var ctx mrtp.PCMUCtx
-			ctx.PCMUCtx = &codec.PCMUCtx{}
+			ctx.PCMUCtx = codec.NewPCMUCtx()
 			ctx.ParseFmtpLine(r.AudioCodecParameters)
-			ctx.AudioCtx.SampleRate = int(r.AudioCodecParameters.ClockRate)
-			ctx.AudioCtx.Channels = int(ctx.RTPCodecParameters.Channels)
+			if r.AudioCodecParameters.ClockRate > 0 {
+				ctx.AudioCtx.SampleRate = int(r.AudioCodecParameters.ClockRate)
+			}
+			if ctx.RTPCodecParameters.Channels > 0 {
+				ctx.AudioCtx.Channels = int(ctx.RTPCodecParameters.Channels)
+			}
 			audioClockRate = ctx.RTPCodecParameters.ClockRate
 			writer.AudioFrame.ICodecCtx = &ctx
 		case "audio/MP4A-LATM":
diff --git a/pull_proxy.go b/pull_proxy.go
index aeefac7f..38033115 100644
--- a/pull_proxy.go
+++ b/pull_proxy.go
@@ -296,6 +296,16 @@ func (s *Server) EnsurePullProxy(conf *PullProxyConfig) (pullProxy IPullProxy, c
 	if existing, ok := s.PullProxies.Find(func(pullProxy IPullProxy) bool {
 		return pullProxy.GetStreamPath() == streamPath
 	}); ok {
+		// 复用已存在的 pull-proxy,丢弃本次 conf。这对常规调用是正确语义(不覆盖
+		// 已有配置),但会丢掉新 conf 的 Description 等字段 —— 历史上曾导致 cluster
+		// relay 标记丢失(现已由 ClusterPlugin.activeRelays 权威判定 relay,不再依赖
+		// 此处的 Description)。保留一条非静默日志,便于以后排查配置不一致。
+		if conf.Description != "" && existing.GetConfig().Description != conf.Description {
+			s.Debug("EnsurePullProxy reuse existing, dropping incoming conf",
+				"streamPath", streamPath,
+				"existingDesc", existing.GetConfig().Description,
+				"droppedDesc", conf.Description)
+		}
 		return existing, false, nil
 	}
 

From 8af468fd05bccb70f72ae5e4308bb516ac33b6a5 Mon Sep 17 00:00:00 2001
From: "richard.li" <lirichen32@gmail.com>
Date: Wed, 3 Jun 2026 14:42:40 +0800
Subject: [PATCH 3/4] =?UTF-8?q?chore(deps):=20=E5=88=87=E6=8D=A2=20gotask?=
 =?UTF-8?q?=20=E4=BE=9D=E8=B5=96=E5=88=B0=20github.com/eanfs/gotask=20v1.0?=
 =?UTF-8?q?.5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

eanfs/gotask v1.0.5 根治了 Job.Call 同 goroutine 重入死锁(本次 cluster
Streams 事件循环死锁的底层原因)。因 fork 模块路径已改为 github.com/eanfs/gotask
且内部子包自引用同步更新,replace 方式会触发 'two different module paths' 冲突,
故全仓 import 由 langhuihui/gotask 改为 eanfs/gotask。

- 91 个 .go 文件 import 路径更新 + go.mod/go.sum
- 二进制 BuildInfo 确认链接 eanfs/gotask v1.0.5
- cluster 测试通过
---
 api.go                                | 2 +-
 go.mod                                | 2 +-
 go.sum                                | 4 ++--
 pkg/av_reader.go                      | 2 +-
 pkg/config/quic.go                    | 2 +-
 pkg/config/tcp.go                     | 2 +-
 pkg/config/udp.go                     | 2 +-
 pkg/http_server_fasthttp.go           | 2 +-
 pkg/http_server_std.go                | 2 +-
 pkg/log.go                            | 2 +-
 pkg/ring-writer.go                    | 2 +-
 pkg/track.go                          | 2 +-
 pkg/util/manager.go                   | 2 +-
 plugin.go                             | 2 +-
 plugin/cascade/client.go              | 2 +-
 plugin/cascade/clientManager.go       | 2 +-
 plugin/cascade/pkg/quic-http.go       | 2 +-
 plugin/cascade/server.go              | 2 +-
 plugin/cluster/apiroute_sync.go       | 2 +-
 plugin/cluster/cluster_helper_test.go | 2 +-
 plugin/cluster/membership.go          | 2 +-
 plugin/cluster/metrics.go             | 2 +-
 plugin/cluster/metrics_test.go        | 2 +-
 plugin/cluster/streamregistry.go      | 2 +-
 plugin/crontab/crontab.go             | 2 +-
 plugin/crontab/index.go               | 2 +-
 plugin/crontab/retry_watcher.go       | 2 +-
 plugin/debug/chart.go                 | 2 +-
 plugin/debug/index.go                 | 2 +-
 plugin/flv/api.go                     | 2 +-
 plugin/flv/pkg/pull-recorder.go       | 2 +-
 plugin/flv/pkg/record.go              | 2 +-
 plugin/gb28181/alarmsub.go            | 2 +-
 plugin/gb28181/catalogsub.go          | 2 +-
 plugin/gb28181/channel.go             | 2 +-
 plugin/gb28181/client.go              | 2 +-
 plugin/gb28181/device.go              | 2 +-
 plugin/gb28181/dialog.go              | 2 +-
 plugin/gb28181/downloaddialog.go      | 2 +-
 plugin/gb28181/forwarddialog.go       | 2 +-
 plugin/gb28181/index.go               | 2 +-
 plugin/gb28181/pkg/forwarder.go       | 2 +-
 plugin/gb28181/pkg/single_port.go     | 2 +-
 plugin/gb28181/platform.go            | 2 +-
 plugin/gb28181/positionsub.go         | 2 +-
 plugin/gb28181/register.go            | 2 +-
 plugin/gb28181/registerhandler.go     | 2 +-
 plugin/gb28181/talk.go                | 2 +-
 plugin/hiksdk/client.go               | 2 +-
 plugin/hiksdk/device.go               | 2 +-
 plugin/hiksdk/pkg/transceiver.go      | 2 +-
 plugin/hls/index.go                   | 2 +-
 plugin/hls/pkg/pull.go                | 2 +-
 plugin/hls/pkg/writer.go              | 2 +-
 plugin/mp4/api.go                     | 2 +-
 plugin/mp4/exception.go               | 2 +-
 plugin/mp4/pkg/pull-recorder.go       | 2 +-
 plugin/mp4/pkg/record.go              | 2 +-
 plugin/mp4/recovery.go                | 2 +-
 plugin/onvif/index.go                 | 2 +-
 plugin/room/index.go                  | 2 +-
 plugin/rtmp/index.go                  | 2 +-
 plugin/rtmp/pkg/client.go             | 2 +-
 plugin/rtmp/pkg/net-connection.go     | 2 +-
 plugin/rtp/pkg/transceiver.go         | 2 +-
 plugin/rtsp/index.go                  | 2 +-
 plugin/rtsp/pkg/client.go             | 2 +-
 plugin/rtsp/pkg/connection.go         | 2 +-
 plugin/snap/pkg/transform.go          | 2 +-
 plugin/srt/index.go                   | 2 +-
 plugin/srt/pkg/client.go              | 2 +-
 plugin/srt/pkg/receiver.go            | 2 +-
 plugin/srt/pkg/sender.go              | 2 +-
 plugin/test/accept_push_task.go       | 2 +-
 plugin/test/api.go                    | 2 +-
 plugin/test/index.go                  | 2 +-
 plugin/test/read _task.go             | 2 +-
 plugin/test/snapshot_task.go          | 2 +-
 plugin/test/write_task.go             | 2 +-
 plugin/transcode/pkg/transform.go     | 2 +-
 plugin/webrtc/batchv2.go              | 2 +-
 plugin/webrtc/pkg/connection.go       | 2 +-
 pull_proxy.go                         | 2 +-
 puller.go                             | 2 +-
 push_proxy.go                         | 2 +-
 pusher.go                             | 2 +-
 recoder.go                            | 2 +-
 server.go                             | 2 +-
 server_grpc.go                        | 2 +-
 subscriber.go                         | 2 +-
 test/server_test.go                   | 2 +-
 transformer.go                        | 2 +-
 upload_retry.go                       | 2 +-
 93 files changed, 94 insertions(+), 94 deletions(-)

diff --git a/api.go b/api.go
index 7f3a414c..12d5ef24 100644
--- a/api.go
+++ b/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"
diff --git a/go.mod b/go.mod
index cffc9fa6..bfadeb97 100644
--- a/go.mod
+++ b/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
diff --git a/go.sum b/go.sum
index d2dc25b3..fbb84d60 100644
--- a/go.sum
+++ b/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=
diff --git a/pkg/av_reader.go b/pkg/av_reader.go
index 7f187118..660a34e7 100644
--- a/pkg/av_reader.go
+++ b/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"
 )
diff --git a/pkg/config/quic.go b/pkg/config/quic.go
index fb622f3d..6d535420 100644
--- a/pkg/config/quic.go
+++ b/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"
 )
 
diff --git a/pkg/config/tcp.go b/pkg/config/tcp.go
index 2dacda3a..913742b6 100644
--- a/pkg/config/tcp.go
+++ b/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
diff --git a/pkg/config/udp.go b/pkg/config/udp.go
index 4191e9bd..fe6d6533 100644
--- a/pkg/config/udp.go
+++ b/pkg/config/udp.go
@@ -6,7 +6,7 @@ import (
 	"net"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 type UDP struct {
diff --git a/pkg/http_server_fasthttp.go b/pkg/http_server_fasthttp.go
index 7940fa51..16a46b3b 100644
--- a/pkg/http_server_fasthttp.go
+++ b/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"
diff --git a/pkg/http_server_std.go b/pkg/http_server_std.go
index 51b7d178..d5c25bb0 100644
--- a/pkg/http_server_std.go
+++ b/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"
 )
 
diff --git a/pkg/log.go b/pkg/log.go
index b3e96359..43d26956 100644
--- a/pkg/log.go
+++ b/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)
diff --git a/pkg/ring-writer.go b/pkg/ring-writer.go
index 117e6aaf..023b653c 100644
--- a/pkg/ring-writer.go
+++ b/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"
 )
 
diff --git a/pkg/track.go b/pkg/track.go
index 4424e764..884d96d6 100644
--- a/pkg/track.go
+++ b/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"
 
diff --git a/pkg/util/manager.go b/pkg/util/manager.go
index 38826539..f385490c 100644
--- a/pkg/util/manager.go
+++ b/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 {
diff --git a/plugin.go b/plugin.go
index cc0ccd56..6c9cbffc 100644
--- a/plugin.go
+++ b/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"
diff --git a/plugin/cascade/client.go b/plugin/cascade/client.go
index c75e2ba9..4a0cb7d6 100644
--- a/plugin/cascade/client.go
+++ b/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"
 )
 
diff --git a/plugin/cascade/clientManager.go b/plugin/cascade/clientManager.go
index 90e6ad45..3efcd0af 100644
--- a/plugin/cascade/clientManager.go
+++ b/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 拉流代理数据结构 }}
diff --git a/plugin/cascade/pkg/quic-http.go b/plugin/cascade/pkg/quic-http.go
index 341fb61b..731bc577 100644
--- a/plugin/cascade/pkg/quic-http.go
+++ b/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"
diff --git a/plugin/cascade/server.go b/plugin/cascade/server.go
index fe0e4a06..bf6f6030 100644
--- a/plugin/cascade/server.go
+++ b/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"
diff --git a/plugin/cluster/apiroute_sync.go b/plugin/cluster/apiroute_sync.go
index fd4ac542..1d787898 100644
--- a/plugin/cluster/apiroute_sync.go
+++ b/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"
 )
 
diff --git a/plugin/cluster/cluster_helper_test.go b/plugin/cluster/cluster_helper_test.go
index 5840adde..6f501457 100644
--- a/plugin/cluster/cluster_helper_test.go
+++ b/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"
 )
 
diff --git a/plugin/cluster/membership.go b/plugin/cluster/membership.go
index c4b2d8d6..039275f0 100644
--- a/plugin/cluster/membership.go
+++ b/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 序列化结构。
diff --git a/plugin/cluster/metrics.go b/plugin/cluster/metrics.go
index ee0671d1..467abf34 100644
--- a/plugin/cluster/metrics.go
+++ b/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 字段。
diff --git a/plugin/cluster/metrics_test.go b/plugin/cluster/metrics_test.go
index f90beddc..25bf83d8 100644
--- a/plugin/cluster/metrics_test.go
+++ b/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,
diff --git a/plugin/cluster/streamregistry.go b/plugin/cluster/streamregistry.go
index 3c5737e3..f2324c3a 100644
--- a/plugin/cluster/streamregistry.go
+++ b/plugin/cluster/streamregistry.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"
 )
 
diff --git a/plugin/crontab/crontab.go b/plugin/crontab/crontab.go
index 8740fc0e..16b0b3b9 100644
--- a/plugin/crontab/crontab.go
+++ b/plugin/crontab/crontab.go
@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/plugin/crontab/pkg"
 )
 
diff --git a/plugin/crontab/index.go b/plugin/crontab/index.go
index 373dd10e..6199fa85 100644
--- a/plugin/crontab/index.go
+++ b/plugin/crontab/index.go
@@ -2,7 +2,7 @@ package plugin_crontab
 
 import (
 	"fmt"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 
 	"m7s.live/v5/pkg/util"
 
diff --git a/plugin/crontab/retry_watcher.go b/plugin/crontab/retry_watcher.go
index 8c599c87..3afd9db2 100644
--- a/plugin/crontab/retry_watcher.go
+++ b/plugin/crontab/retry_watcher.go
@@ -8,7 +8,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 // RecordRetryTickTask periodically checks recording status; only one startRecording attempt per slot
diff --git a/plugin/debug/chart.go b/plugin/debug/chart.go
index 9c5062ea..74ee2f9f 100644
--- a/plugin/debug/chart.go
+++ b/plugin/debug/chart.go
@@ -13,7 +13,7 @@ import (
 	"time"
 
 	"github.com/gorilla/websocket"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/shirou/gopsutil/v4/cpu"
 	"github.com/shirou/gopsutil/v4/process"
 )
diff --git a/plugin/debug/index.go b/plugin/debug/index.go
index a6c8a377..70a90f91 100644
--- a/plugin/debug/index.go
+++ b/plugin/debug/index.go
@@ -21,7 +21,7 @@ import (
 	myproc "github.com/cloudwego/goref/pkg/proc"
 	"github.com/go-delve/delve/pkg/config"
 	"github.com/go-delve/delve/service/debugger"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"m7s.live/v5"
diff --git a/plugin/flv/api.go b/plugin/flv/api.go
index a3a3d32b..88bc12e4 100644
--- a/plugin/flv/api.go
+++ b/plugin/flv/api.go
@@ -10,7 +10,7 @@ import (
 	"time"
 	"unsafe"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
 
diff --git a/plugin/flv/pkg/pull-recorder.go b/plugin/flv/pkg/pull-recorder.go
index ac922458..b7e3d77d 100644
--- a/plugin/flv/pkg/pull-recorder.go
+++ b/plugin/flv/pkg/pull-recorder.go
@@ -10,7 +10,7 @@ import (
 	"time"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	m7s "m7s.live/v5"
 	"m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
diff --git a/plugin/flv/pkg/record.go b/plugin/flv/pkg/record.go
index 7211b92b..1dfe7091 100644
--- a/plugin/flv/pkg/record.go
+++ b/plugin/flv/pkg/record.go
@@ -10,7 +10,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
diff --git a/plugin/gb28181/alarmsub.go b/plugin/gb28181/alarmsub.go
index fe118890..eef89ce2 100644
--- a/plugin/gb28181/alarmsub.go
+++ b/plugin/gb28181/alarmsub.go
@@ -3,7 +3,7 @@ package plugin_gb28181pro
 import (
 	"time"
 
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 )
 
 // AlarmSubscribeTask 报警订阅任务
diff --git a/plugin/gb28181/catalogsub.go b/plugin/gb28181/catalogsub.go
index 08ee3a06..9320fa39 100644
--- a/plugin/gb28181/catalogsub.go
+++ b/plugin/gb28181/catalogsub.go
@@ -4,7 +4,7 @@ import (
 	"time"
 
 	"github.com/emiago/sipgo/sip"
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 )
 
 // CatalogSubscribeTask 目录订阅任务
diff --git a/plugin/gb28181/channel.go b/plugin/gb28181/channel.go
index a1421bff..df6d2be2 100644
--- a/plugin/gb28181/channel.go
+++ b/plugin/gb28181/channel.go
@@ -6,7 +6,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/util"
 	gb28181 "m7s.live/v5/plugin/gb28181/pkg"
diff --git a/plugin/gb28181/client.go b/plugin/gb28181/client.go
index 23a95f08..a534932f 100644
--- a/plugin/gb28181/client.go
+++ b/plugin/gb28181/client.go
@@ -12,7 +12,7 @@ import (
 	"github.com/emiago/sipgo/sip"
 	myip "github.com/husanpao/ip"
 	"github.com/icholy/digest"
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 	gb28181 "m7s.live/v5/plugin/gb28181/pkg"
 )
 
diff --git a/plugin/gb28181/device.go b/plugin/gb28181/device.go
index e289320d..c414e2f5 100644
--- a/plugin/gb28181/device.go
+++ b/plugin/gb28181/device.go
@@ -17,7 +17,7 @@ import (
 
 	"github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/util"
 	gb28181 "m7s.live/v5/plugin/gb28181/pkg"
 	mrtp "m7s.live/v5/plugin/rtp/pkg"
diff --git a/plugin/gb28181/dialog.go b/plugin/gb28181/dialog.go
index 14846da6..5e9bfd39 100644
--- a/plugin/gb28181/dialog.go
+++ b/plugin/gb28181/dialog.go
@@ -11,7 +11,7 @@ import (
 
 	sipgo "github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	m7s "m7s.live/v5"
 	pkg "m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/util"
diff --git a/plugin/gb28181/downloaddialog.go b/plugin/gb28181/downloaddialog.go
index b725d73b..4b2b41c0 100644
--- a/plugin/gb28181/downloaddialog.go
+++ b/plugin/gb28181/downloaddialog.go
@@ -14,7 +14,7 @@ import (
 
 	sipgo "github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/config"
 	gb28181 "m7s.live/v5/plugin/gb28181/pkg"
 	mrtp "m7s.live/v5/plugin/rtp/pkg"
diff --git a/plugin/gb28181/forwarddialog.go b/plugin/gb28181/forwarddialog.go
index 6d2ccce5..e488100c 100644
--- a/plugin/gb28181/forwarddialog.go
+++ b/plugin/gb28181/forwarddialog.go
@@ -9,7 +9,7 @@ import (
 
 	sipgo "github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	m7s "m7s.live/v5"
 	"m7s.live/v5/pkg/util"
 	mrtp "m7s.live/v5/plugin/rtp/pkg"
diff --git a/plugin/gb28181/index.go b/plugin/gb28181/index.go
index 545dd2b4..b83ff7f7 100644
--- a/plugin/gb28181/index.go
+++ b/plugin/gb28181/index.go
@@ -23,7 +23,7 @@ import (
 
 	"github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	m7s "m7s.live/v5"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/util"
diff --git a/plugin/gb28181/pkg/forwarder.go b/plugin/gb28181/pkg/forwarder.go
index a65d2233..cca5002b 100644
--- a/plugin/gb28181/pkg/forwarder.go
+++ b/plugin/gb28181/pkg/forwarder.go
@@ -27,7 +27,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/pion/rtp"
 	"m7s.live/v5/pkg/util"
 )
diff --git a/plugin/gb28181/pkg/single_port.go b/plugin/gb28181/pkg/single_port.go
index a460fb2b..1ba4fe29 100644
--- a/plugin/gb28181/pkg/single_port.go
+++ b/plugin/gb28181/pkg/single_port.go
@@ -6,7 +6,7 @@ import (
 	"io"
 	"net"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/pion/rtp"
 	"m7s.live/v5/pkg/util"
 )
diff --git a/plugin/gb28181/platform.go b/plugin/gb28181/platform.go
index 08a6d1f3..c27fb16e 100644
--- a/plugin/gb28181/platform.go
+++ b/plugin/gb28181/platform.go
@@ -18,7 +18,7 @@ import (
 	"github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
 	"github.com/icholy/digest"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	gb28181 "m7s.live/v5/plugin/gb28181/pkg"
 )
 
diff --git a/plugin/gb28181/positionsub.go b/plugin/gb28181/positionsub.go
index 2b008f69..2bc1c127 100644
--- a/plugin/gb28181/positionsub.go
+++ b/plugin/gb28181/positionsub.go
@@ -3,7 +3,7 @@ package plugin_gb28181pro
 import (
 	"time"
 
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 )
 
 // PositionSubscribeTask 位置订阅任务
diff --git a/plugin/gb28181/register.go b/plugin/gb28181/register.go
index 93f81a1a..887718f8 100644
--- a/plugin/gb28181/register.go
+++ b/plugin/gb28181/register.go
@@ -4,7 +4,7 @@ import (
 	"errors"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 type Register struct {
diff --git a/plugin/gb28181/registerhandler.go b/plugin/gb28181/registerhandler.go
index a615a156..2d92030b 100644
--- a/plugin/gb28181/registerhandler.go
+++ b/plugin/gb28181/registerhandler.go
@@ -11,7 +11,7 @@ import (
 	"github.com/emiago/sipgo/sip"
 	myip "github.com/husanpao/ip"
 	"github.com/icholy/digest"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"gorm.io/gorm"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/util"
diff --git a/plugin/gb28181/talk.go b/plugin/gb28181/talk.go
index e309b866..dcc77c22 100644
--- a/plugin/gb28181/talk.go
+++ b/plugin/gb28181/talk.go
@@ -11,7 +11,7 @@ import (
 
 	"github.com/gobwas/ws"
 	"github.com/gobwas/ws/wsutil"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 // TalkWebsocketTask 负责管理一次 WebSocket 对讲/广播会话
diff --git a/plugin/hiksdk/client.go b/plugin/hiksdk/client.go
index 3e0f3ead..76f6898c 100644
--- a/plugin/hiksdk/client.go
+++ b/plugin/hiksdk/client.go
@@ -1,7 +1,7 @@
 package plugin_hiksdk
 
 import (
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/config"
 )
diff --git a/plugin/hiksdk/device.go b/plugin/hiksdk/device.go
index 69de9e74..c28f49cc 100644
--- a/plugin/hiksdk/device.go
+++ b/plugin/hiksdk/device.go
@@ -6,7 +6,7 @@ import (
 
 	"m7s.live/v5/plugin/hiksdk/pkg"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
diff --git a/plugin/hiksdk/pkg/transceiver.go b/plugin/hiksdk/pkg/transceiver.go
index 119f0413..7305f47a 100644
--- a/plugin/hiksdk/pkg/transceiver.go
+++ b/plugin/hiksdk/pkg/transceiver.go
@@ -5,7 +5,7 @@ import (
 	"io"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	mpegps "m7s.live/v5/pkg/format/ps"
 	"m7s.live/v5/pkg/util"
 )
diff --git a/plugin/hls/index.go b/plugin/hls/index.go
index de88bb79..af564d3f 100644
--- a/plugin/hls/index.go
+++ b/plugin/hls/index.go
@@ -14,7 +14,7 @@ import (
 
 	_ "embed"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/util"
diff --git a/plugin/hls/pkg/pull.go b/plugin/hls/pkg/pull.go
index 4c30a51f..7b6c08c9 100644
--- a/plugin/hls/pkg/pull.go
+++ b/plugin/hls/pkg/pull.go
@@ -12,7 +12,7 @@ import (
 	"time"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/quangngotan95/go-m3u8/m3u8"
 	"m7s.live/v5"
 	pkg "m7s.live/v5/pkg"
diff --git a/plugin/hls/pkg/writer.go b/plugin/hls/pkg/writer.go
index 5ecba489..e9405d83 100644
--- a/plugin/hls/pkg/writer.go
+++ b/plugin/hls/pkg/writer.go
@@ -10,7 +10,7 @@ import (
 	"sync"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/codec"
 	"m7s.live/v5/pkg/format"
diff --git a/plugin/mp4/api.go b/plugin/mp4/api.go
index 1b0f0ee4..a4fff788 100644
--- a/plugin/mp4/api.go
+++ b/plugin/mp4/api.go
@@ -14,7 +14,7 @@ import (
 	"time"
 	"unsafe"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/mcuadros/go-defaults"
 	"google.golang.org/protobuf/types/known/emptypb"
 	m7s "m7s.live/v5"
diff --git a/plugin/mp4/exception.go b/plugin/mp4/exception.go
index ce444a55..22cea603 100644
--- a/plugin/mp4/exception.go
+++ b/plugin/mp4/exception.go
@@ -6,7 +6,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/shirou/gopsutil/v4/disk"
 	"gorm.io/gorm"
 	"m7s.live/v5"
diff --git a/plugin/mp4/pkg/pull-recorder.go b/plugin/mp4/pkg/pull-recorder.go
index b2a9ae3d..9da1f82c 100644
--- a/plugin/mp4/pkg/pull-recorder.go
+++ b/plugin/mp4/pkg/pull-recorder.go
@@ -5,7 +5,7 @@ import (
 	"time"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	m7s "m7s.live/v5"
 	"m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/codec"
diff --git a/plugin/mp4/pkg/record.go b/plugin/mp4/pkg/record.go
index adb609df..d2d05334 100644
--- a/plugin/mp4/pkg/record.go
+++ b/plugin/mp4/pkg/record.go
@@ -10,7 +10,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"gorm.io/gorm"
 	m7s "m7s.live/v5"
 	"m7s.live/v5/pkg"
diff --git a/plugin/mp4/recovery.go b/plugin/mp4/recovery.go
index 0af6e775..0c31da7d 100644
--- a/plugin/mp4/recovery.go
+++ b/plugin/mp4/recovery.go
@@ -7,7 +7,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"gorm.io/gorm"
 	"m7s.live/v5"
 	mp4 "m7s.live/v5/plugin/mp4/pkg"
diff --git a/plugin/onvif/index.go b/plugin/onvif/index.go
index bd6382a9..7de66567 100755
--- a/plugin/onvif/index.go
+++ b/plugin/onvif/index.go
@@ -5,7 +5,7 @@ import (
 	"sync"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/util"
 
 	m7s "m7s.live/v5"
diff --git a/plugin/room/index.go b/plugin/room/index.go
index 89fff2f1..5d3bd4ae 100644
--- a/plugin/room/index.go
+++ b/plugin/room/index.go
@@ -13,7 +13,7 @@ import (
 	"github.com/gobwas/ws"
 	"github.com/gobwas/ws/wsutil"
 	"github.com/google/uuid"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	. "m7s.live/v5"
 	"m7s.live/v5/pkg"
diff --git a/plugin/rtmp/index.go b/plugin/rtmp/index.go
index fd2ce32f..5968923f 100644
--- a/plugin/rtmp/index.go
+++ b/plugin/rtmp/index.go
@@ -7,7 +7,7 @@ import (
 	"net"
 	"strings"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/plugin/rtmp/pb"
 	. "m7s.live/v5/plugin/rtmp/pkg"
diff --git a/plugin/rtmp/pkg/client.go b/plugin/rtmp/pkg/client.go
index 228ff78e..9ed424bb 100644
--- a/plugin/rtmp/pkg/client.go
+++ b/plugin/rtmp/pkg/client.go
@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	pkg "m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
 
diff --git a/plugin/rtmp/pkg/net-connection.go b/plugin/rtmp/pkg/net-connection.go
index 6ad33ea4..48a59014 100644
--- a/plugin/rtmp/pkg/net-connection.go
+++ b/plugin/rtmp/pkg/net-connection.go
@@ -9,7 +9,7 @@ import (
 	"time"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/util"
 )
diff --git a/plugin/rtp/pkg/transceiver.go b/plugin/rtp/pkg/transceiver.go
index c6761208..56b83e50 100644
--- a/plugin/rtp/pkg/transceiver.go
+++ b/plugin/rtp/pkg/transceiver.go
@@ -10,7 +10,7 @@ import (
 	"time"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/pion/rtp"
 	mpegps "m7s.live/v5/pkg/format/ps"
 	"m7s.live/v5/pkg/util"
diff --git a/plugin/rtsp/index.go b/plugin/rtsp/index.go
index 0697e47b..f268d2d3 100644
--- a/plugin/rtsp/index.go
+++ b/plugin/rtsp/index.go
@@ -5,7 +5,7 @@ import (
 	"net"
 	"strings"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/util"
 
 	"m7s.live/v5"
diff --git a/plugin/rtsp/pkg/client.go b/plugin/rtsp/pkg/client.go
index 6997ca3f..7d04862e 100644
--- a/plugin/rtsp/pkg/client.go
+++ b/plugin/rtsp/pkg/client.go
@@ -1,7 +1,7 @@
 package rtsp
 
 import (
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/config"
 
 	"m7s.live/v5"
diff --git a/plugin/rtsp/pkg/connection.go b/plugin/rtsp/pkg/connection.go
index 78b8c709..6a17f7b9 100644
--- a/plugin/rtsp/pkg/connection.go
+++ b/plugin/rtsp/pkg/connection.go
@@ -16,7 +16,7 @@ import (
 	"m7s.live/v5/pkg"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/util"
 )
diff --git a/plugin/snap/pkg/transform.go b/plugin/snap/pkg/transform.go
index 0a2f0c18..aa256627 100644
--- a/plugin/snap/pkg/transform.go
+++ b/plugin/snap/pkg/transform.go
@@ -10,7 +10,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/format"
diff --git a/plugin/srt/index.go b/plugin/srt/index.go
index 788a5a08..4dc0e8dd 100644
--- a/plugin/srt/index.go
+++ b/plugin/srt/index.go
@@ -5,7 +5,7 @@ import (
 	"strings"
 
 	srt "github.com/datarhei/gosrt"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	srt_pkg "m7s.live/v5/plugin/srt/pkg"
 )
diff --git a/plugin/srt/pkg/client.go b/plugin/srt/pkg/client.go
index 5b974d10..5e8fa77d 100644
--- a/plugin/srt/pkg/client.go
+++ b/plugin/srt/pkg/client.go
@@ -4,7 +4,7 @@ import (
 	"net/url"
 
 	srt "github.com/datarhei/gosrt"
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 	"m7s.live/v5"
 	pkg "m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
diff --git a/plugin/srt/pkg/receiver.go b/plugin/srt/pkg/receiver.go
index e82a5640..b218fb7f 100644
--- a/plugin/srt/pkg/receiver.go
+++ b/plugin/srt/pkg/receiver.go
@@ -5,7 +5,7 @@ import (
 
 	srt "github.com/datarhei/gosrt"
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	mpegts "m7s.live/v5/pkg/format/ts"
 )
 
diff --git a/plugin/srt/pkg/sender.go b/plugin/srt/pkg/sender.go
index ce5286fa..b164708f 100644
--- a/plugin/srt/pkg/sender.go
+++ b/plugin/srt/pkg/sender.go
@@ -3,7 +3,7 @@ package srt
 import (
 	srt "github.com/datarhei/gosrt"
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/codec"
 	"m7s.live/v5/pkg/format"
diff --git a/plugin/test/accept_push_task.go b/plugin/test/accept_push_task.go
index 37c721ca..b185065f 100644
--- a/plugin/test/accept_push_task.go
+++ b/plugin/test/accept_push_task.go
@@ -7,7 +7,7 @@ import (
 	"os/exec"
 	"strings"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 )
 
 func init() {
diff --git a/plugin/test/api.go b/plugin/test/api.go
index 9e36ad97..d0297690 100644
--- a/plugin/test/api.go
+++ b/plugin/test/api.go
@@ -14,7 +14,7 @@ import (
 	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	pb "m7s.live/v5/pb"
 	"m7s.live/v5/pkg/config"
diff --git a/plugin/test/index.go b/plugin/test/index.go
index d5d771ba..b46ae684 100644
--- a/plugin/test/index.go
+++ b/plugin/test/index.go
@@ -10,7 +10,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/util"
 	"m7s.live/v5/plugin/test/pb"
diff --git a/plugin/test/read _task.go b/plugin/test/read _task.go
index 6afde5bc..58398b4e 100644
--- a/plugin/test/read _task.go	
+++ b/plugin/test/read _task.go	
@@ -3,7 +3,7 @@ package plugin_test
 import (
 	"fmt"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/config"
 	flv "m7s.live/v5/plugin/flv/pkg"
diff --git a/plugin/test/snapshot_task.go b/plugin/test/snapshot_task.go
index 7f26a58e..4e804a06 100644
--- a/plugin/test/snapshot_task.go
+++ b/plugin/test/snapshot_task.go
@@ -7,7 +7,7 @@ import (
 	"strings"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/format"
diff --git a/plugin/test/write_task.go b/plugin/test/write_task.go
index f934c106..ff1a842f 100644
--- a/plugin/test/write_task.go
+++ b/plugin/test/write_task.go
@@ -10,7 +10,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/util"
diff --git a/plugin/transcode/pkg/transform.go b/plugin/transcode/pkg/transform.go
index 9b6f8797..ccf43180 100644
--- a/plugin/transcode/pkg/transform.go
+++ b/plugin/transcode/pkg/transform.go
@@ -11,7 +11,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/filerotate"
 
diff --git a/plugin/webrtc/batchv2.go b/plugin/webrtc/batchv2.go
index ad72dc31..595aed81 100644
--- a/plugin/webrtc/batchv2.go
+++ b/plugin/webrtc/batchv2.go
@@ -9,7 +9,7 @@ import (
 
 	"github.com/gobwas/ws"
 	"github.com/gobwas/ws/wsutil"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	. "github.com/pion/webrtc/v4"
 	"m7s.live/v5/pkg/codec"
 	. "m7s.live/v5/plugin/webrtc/pkg"
diff --git a/plugin/webrtc/pkg/connection.go b/plugin/webrtc/pkg/connection.go
index dd746740..1f3fa366 100644
--- a/plugin/webrtc/pkg/connection.go
+++ b/plugin/webrtc/pkg/connection.go
@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/pion/rtcp"
 	. "github.com/pion/webrtc/v4"
 	"m7s.live/v5"
diff --git a/pull_proxy.go b/pull_proxy.go
index 38033115..1a1fc5ae 100644
--- a/pull_proxy.go
+++ b/pull_proxy.go
@@ -10,7 +10,7 @@ import (
 	"strings"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"github.com/mcuadros/go-defaults"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/emptypb"
diff --git a/puller.go b/puller.go
index 5e224602..b1aeae99 100644
--- a/puller.go
+++ b/puller.go
@@ -13,7 +13,7 @@ import (
 
 	"github.com/gorilla/websocket"
 	"github.com/langhuihui/gomem"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	pkg "m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/format"
diff --git a/push_proxy.go b/push_proxy.go
index e91f6dcd..80115294 100644
--- a/push_proxy.go
+++ b/push_proxy.go
@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/langhuihui/gotask"
+	"github.com/eanfs/gotask"
 	"github.com/mcuadros/go-defaults"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
diff --git a/pusher.go b/pusher.go
index aff716c8..c64a6b95 100644
--- a/pusher.go
+++ b/pusher.go
@@ -1,7 +1,7 @@
 package m7s
 
 import (
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/config"
 )
 
diff --git a/recoder.go b/recoder.go
index aaba24d6..7579cdcd 100644
--- a/recoder.go
+++ b/recoder.go
@@ -8,7 +8,7 @@ import (
 
 	"gorm.io/gorm"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/storage"
 )
diff --git a/server.go b/server.go
index 8e6b61cc..752879a7 100644
--- a/server.go
+++ b/server.go
@@ -22,7 +22,7 @@ import (
 
 	"github.com/shirou/gopsutil/v4/cpu"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/config"
 
 	sysruntime "runtime"
diff --git a/server_grpc.go b/server_grpc.go
index 89bbf863..eb00a8ea 100644
--- a/server_grpc.go
+++ b/server_grpc.go
@@ -8,7 +8,7 @@ import (
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pb"
 	. "m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/auth"
diff --git a/subscriber.go b/subscriber.go
index e7dbee97..a5964d0e 100644
--- a/subscriber.go
+++ b/subscriber.go
@@ -14,7 +14,7 @@ import (
 	"github.com/gobwas/ws"
 	"github.com/gobwas/ws/wsutil"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	. "m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/util"
diff --git a/test/server_test.go b/test/server_test.go
index cff292f0..5512819f 100644
--- a/test/server_test.go
+++ b/test/server_test.go
@@ -5,7 +5,7 @@ import (
 	"testing"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5"
 	"m7s.live/v5/pkg"
 )
diff --git a/transformer.go b/transformer.go
index 61a46ab9..b598358a 100644
--- a/transformer.go
+++ b/transformer.go
@@ -5,7 +5,7 @@ import (
 	"slices"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/util"
diff --git a/upload_retry.go b/upload_retry.go
index 0ec0a190..42ce77a0 100644
--- a/upload_retry.go
+++ b/upload_retry.go
@@ -6,7 +6,7 @@ import (
 	"os"
 	"time"
 
-	task "github.com/langhuihui/gotask"
+	task "github.com/eanfs/gotask"
 	"m7s.live/v5/pkg/config"
 	"m7s.live/v5/pkg/storage"
 )

From 61b0957174d14bdcd949d29cff10269ef8f56096 Mon Sep 17 00:00:00 2001
From: "richard.li" <lirichen32@gmail.com>
Date: Wed, 3 Jun 2026 14:50:46 +0800
Subject: [PATCH 4/4] =?UTF-8?q?fix(cluster-e2e):=20storage.s3=20=E5=B5=8C?=
 =?UTF-8?q?=E5=A5=97=E6=A0=BC=E5=BC=8F=20+=20cluster=20HTTP=20=E5=89=8D?=
 =?UTF-8?q?=E7=BC=80=E8=B7=AF=E5=BE=84(/cluster/api/cluster/*)+=20mp4/down?=
 =?UTF-8?q?load=20=E5=89=8D=E7=BC=80=20+=20Dockerfile=20go1.26?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 example/cluster-e2e/Dockerfile         |  4 +++-
 example/cluster-e2e/config-node-1.yaml | 16 ++++++++--------
 example/cluster-e2e/config-node-2.yaml | 16 ++++++++--------
 example/cluster-e2e/config-node-3.yaml | 16 ++++++++--------
 example/cluster-e2e/smoke.sh           | 14 ++++++++------
 5 files changed, 35 insertions(+), 31 deletions(-)

diff --git a/example/cluster-e2e/Dockerfile b/example/cluster-e2e/Dockerfile
index 100241d5..f95d88f4 100644
--- a/example/cluster-e2e/Dockerfile
+++ b/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 ./
diff --git a/example/cluster-e2e/config-node-1.yaml b/example/cluster-e2e/config-node-1.yaml
index 61a431f7..26f2a789 100644
--- a/example/cluster-e2e/config-node-1.yaml
+++ b/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:
diff --git a/example/cluster-e2e/config-node-2.yaml b/example/cluster-e2e/config-node-2.yaml
index 666dd37f..ef97f76f 100644
--- a/example/cluster-e2e/config-node-2.yaml
+++ b/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:
diff --git a/example/cluster-e2e/config-node-3.yaml b/example/cluster-e2e/config-node-3.yaml
index 88794e6c..b2777541 100644
--- a/example/cluster-e2e/config-node-3.yaml
+++ b/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:
diff --git a/example/cluster-e2e/smoke.sh b/example/cluster-e2e/smoke.sh
index c6a6c9b7..bec581b7 100755
--- a/example/cluster-e2e/smoke.sh
+++ b/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"