fix(execd): serialize streamHub holder access under h.mu

Author: Pangjiping

components/execd/pkg/web/controller/command_stream.go

@@ -79,21 +79,25 @@ func (r *streamRegistry) registerPrimary(id string, w http.ResponseWriter, ctx c
 		streamID: id,
 		done:     make(chan struct{}),
 	}
-	h.holder = &streamHolder{writer: w, ctx: ctx}
 	r.m[id] = h
+	h.mu.Lock()
+	h.holder = &streamHolder{writer: w, ctx: ctx}
+	h.mu.Unlock()
 	r.mu.Unlock()
 
-	r.watchHolderRelease(id, h, ctx)
+	watchHolderRelease(h, ctx)
 }
 
-func (r *streamRegistry) watchHolderRelease(id string, h *streamHub, ctx context.Context) {
+// watchHolderRelease clears h.holder when ctx is cancelled. All holder mutations use h.mu only
+// (see tryAttachResume, registerPrimary, writeFrame) so r.mu and h.mu are not split across h.holder.
+func watchHolderRelease(h *streamHub, ctx context.Context) {
 	go func() {
 		<-ctx.Done()
-		r.mu.Lock()
-		if cur, ok := r.m[id]; ok && cur == h && h.holder != nil && h.holder.ctx == ctx {
+		h.mu.Lock()
+		defer h.mu.Unlock()
+		if h.holder != nil && h.holder.ctx == ctx {
 			h.holder = nil
 		}
-		r.mu.Unlock()
 	}()
 }
 
@@ -143,14 +147,17 @@ func (r *streamRegistry) tryAttachResume(id string, w http.ResponseWriter, ctx c
 		r.mu.Unlock()
 		return nil, errLiveHubNotFound
 	}
+	h.mu.Lock()
 	if h.holder != nil && h.holder.ctx.Err() == nil {
+		h.mu.Unlock()
 		r.mu.Unlock()
 		return nil, errLiveStreamPrimaryActive
 	}
 	h.holder = &streamHolder{writer: w, ctx: ctx}
+	h.mu.Unlock()
 	r.mu.Unlock()
 
-	r.watchHolderRelease(id, h, ctx)
+	watchHolderRelease(h, ctx)
 	return h, nil
 }
 
@@ -171,11 +178,14 @@ func (r *streamRegistry) writeSSE(id string, data []byte, bufEid int64, handler,
 // Live writeFrame calls block on the same mutex, so chunks appended only to the ring during the initial
 // snapshot replay cannot be missed on this connection (see ResumeCommandStream).
 func (h *streamHub) flushResumeTail(commandID string, afterEid int64) {
-	if h == nil || h.holder == nil {
+	if h == nil {
 		return
 	}
 	h.mu.Lock()
 	defer h.mu.Unlock()
+	if h.holder == nil {
+		return
+	}
 
 	tail, ok := resumeBuffer.EventsAfter(commandID, afterEid)
 	if !ok || len(tail) == 0 {