sfu: don't hold bindLock across the blank-frame flush in CloseWithFlush (#4895)

* sfu: don't hold bindLock across the blank-frame flush in CloseWithFlush

CloseWithFlush held bindLock across the up-to-flushTimeout (1s) wait for
the closing blank-frame flush. bindLock is a control-plane lock, also
taken by Bind/SetConnected/ReceiverRestart, so every close serialized
those behind its flush wait. This was the single largest accumulator in
mutex contention profiling under track churn, aggregated across all
closing downtracks.

The flush runs in its own goroutine (writeBlankFrameRTP) and is cancelled
via blankFramesGeneration, so the lock guards nothing during the wait;
release it and re-acquire for the unbind/delete teardown.

Guard the control-plane mutators on isClosed so none can act on a
downtrack through the newly opened window. Bind in particular was
reachable: its bindState != unbound check does not catch a closed track,
since close leaves the state unbound.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* sfu: drop redundant isClosed guard in Bind

Bind already handles a closed downtrack after codec matching (returns the
codec with no error, preserving pion's non-error contract and never
rebuilding state). The early guard returned errDownTrackClosed, which
handleRemoteAnswerReceived treats as a negotiation failure, so a
close/negotiation race could abort the remote answer. Remove it and the
error type.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

pkg/sfu/downtrack.go

@@ -1440,15 +1440,24 @@ func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) {
 		if flush {
 			doneFlushing := d.writeBlankFrameRTP(RTPBlankFramesCloseSeconds, d.blankFramesGeneration.Inc())
 
+			// The flush runs in its own goroutine (writeBlankFrameRTP) and is
+			// cancelled via blankFramesGeneration, so bindLock guards nothing
+			// during the wait. Release it: bindLock is a control-plane lock
+			// (Bind/SetConnected/ReceiverRestart), and holding it across the
+			// up-to-flushTimeout wait serializes all of those behind every
+			// close. isClosed is already set, so no other close can enter.
+			d.bindLock.Unlock()
+
 			// wait a limited time to flush
 			timer := time.NewTimer(flushTimeout)
-			defer timer.Stop()
-
 			select {
 			case <-doneFlushing:
 			case <-timer.C:
 				d.blankFramesGeneration.Inc() // in case flush is still running
 			}
+			timer.Stop()
+
+			d.bindLock.Lock()
 		}
 
 		d.params.Logger.Debugw("closing sender", "kind", d.kind)
@@ -1755,6 +1764,10 @@ func (d *DownTrack) ReceiverRestart(rcvr TrackReceiver) {
 	}
 
 	d.bindLock.Lock()
+	if d.isClosed.Load() {
+		d.bindLock.Unlock()
+		return
+	}
 	codec := d.codec.Load().(webrtc.RTPCodecCapability)
 	d.bindLock.Unlock()
 
@@ -2167,6 +2180,10 @@ func (d *DownTrack) handleRTCPRTX(bytes []byte) {
 
 func (d *DownTrack) SetConnected() {
 	d.bindLock.Lock()
+	if d.isClosed.Load() {
+		d.bindLock.Unlock()
+		return
+	}
 	if !d.connected.Swap(true) {
 		d.onBindAndConnectedChange()
 	}