Merge pull request #1 from bnema/feature/pipewire-loop-teardown

fix: make PipeWire teardown respect loop context

commit: 290220ccbc3b13751b5e399a2b38dc59671ad44f
author: bnema · 5mo ago · 2026-05-15 13:06 UTC
committer: GitHub · 5mo ago · 2026-05-15 13:06 UTC
merge of: dc8c53b + 8edd742 · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown
refs: tag: v0.1.3

6 files · +392 -20

A internal/capi/loop_lock.go +40 -0

@@ -0,0 +1,40 @@
	1	+	package capi
	2	+
	3	+	import (
	4	+	"unsafe"
	5	+
	6	+	"github.com/bnema/purego"
	7	+	)
	8	+
	9	+	// pw_loop_lock/pw_loop_unlock are kept as handwritten internal bindings
	10	+	// because they are only used inside capi to satisfy PipeWire's thread/locking
	11	+	// requirements for stream lifecycle calls. Exposing them through generated
	12	+	// ports would leak a low-level implementation detail upward.
	13	+	type pwLoopControlFunc func(loop unsafe.Pointer) int32
	14	+
	15	+	var (
	16	+	pw_loop_lock pwLoopControlFunc
	17	+	pw_loop_unlock pwLoopControlFunc
	18	+	)
	19	+
	20	+	func registerLoopLock(handle uintptr) {
	21	+	purego.RegisterLibFunc(&pw_loop_lock, handle, "pw_loop_lock")
	22	+	purego.RegisterLibFunc(&pw_loop_unlock, handle, "pw_loop_unlock")
	23	+	}
	24	+
	25	+	var withLoopLock = defaultWithLoopLock
	26	+
	27	+	func defaultWithLoopLock(loop unsafe.Pointer, fn func() int32) int32 {
	28	+	if loop == nil \|\| pw_loop_lock == nil \|\| pw_loop_unlock == nil {
	29	+	return -1
	30	+	}
	31	+	if ret := pw_loop_lock(loop); ret < 0 {
	32	+	return ret
	33	+	}
	34	+	ret := fn()
	35	+	unlockRet := pw_loop_unlock(loop)
	36	+	if ret >= 0 && unlockRet < 0 {
	37	+	return unlockRet
	38	+	}
	39	+	return ret
	40	+	}

@@ -0,0 +1,40 @@
	1	+	package capi
	2	+
	3	+	import (
	4	+	"unsafe"
	5	+
	6	+	"github.com/bnema/purego"
	7	+	)
	8	+
	9	+	// pw_loop_lock/pw_loop_unlock are kept as handwritten internal bindings
	10	+	// because they are only used inside capi to satisfy PipeWire's thread/locking
	11	+	// requirements for stream lifecycle calls. Exposing them through generated
	12	+	// ports would leak a low-level implementation detail upward.
	13	+	type pwLoopControlFunc func(loop unsafe.Pointer) int32
	14	+
	15	+	var (
	16	+	pw_loop_lock pwLoopControlFunc
	17	+	pw_loop_unlock pwLoopControlFunc
	18	+	)
	19	+
	20	+	func registerLoopLock(handle uintptr) {
	21	+	purego.RegisterLibFunc(&pw_loop_lock, handle, "pw_loop_lock")
	22	+	purego.RegisterLibFunc(&pw_loop_unlock, handle, "pw_loop_unlock")
	23	+	}
	24	+
	25	+	var withLoopLock = defaultWithLoopLock
	26	+
	27	+	func defaultWithLoopLock(loop unsafe.Pointer, fn func() int32) int32 {
	28	+	if loop == nil \|\| pw_loop_lock == nil \|\| pw_loop_unlock == nil {
	29	+	return -1
	30	+	}
	31	+	if ret := pw_loop_lock(loop); ret < 0 {
	32	+	return ret
	33	+	}
	34	+	ret := fn()
	35	+	unlockRet := pw_loop_unlock(loop)
	36	+	if ret >= 0 && unlockRet < 0 {
	37	+	return unlockRet
	38	+	}
	39	+	return ret
	40	+	}

M internal/capi/register_all.go +1 -0


@@ -7,6 +7,7 @@ import "github.com/bnema/purego-pipewire/internal/loader"
7	7		func registerAll(h loader.Handles) error {
8	8		registerInit(h.PipeWire)
9	9		registerLoop(h.PipeWire)
	10	+	registerLoopLock(h.PipeWire)
10	11		registerStream_playback(h.PipeWire)
11	12		return nil
12	13		}


@@ -7,6 +7,7 @@ import "github.com/bnema/purego-pipewire/internal/loader"
7	func registerAll(h loader.Handles) error {	7		func registerAll(h loader.Handles) error {
8	registerInit(h.PipeWire)	8		registerInit(h.PipeWire)
9	registerLoop(h.PipeWire)	9		registerLoop(h.PipeWire)
		10	+	registerLoopLock(h.PipeWire)
10	registerStream_playback(h.PipeWire)	11		registerStream_playback(h.PipeWire)
11	return nil	12		return nil
12	}	13		}

M internal/capi/stream_ops.go +62 -5


@@ -3,6 +3,7 @@ package capi
3	3		import (
4	4		"errors"
5	5		"fmt"
	6	+	"runtime"
6	7		"sync"
7	8		"unsafe"
8	9

@@ -59,10 +60,17 @@ type streamCallbackStorage struct {
59	60		// PipeWire reads the SPA POD params asynchronously during format negotiation, so the
60	61		// backing byte storage must remain live until DestroyStream releases it.
61	62		// - Both maps are keyed by the stream pointer and must be cleaned up in DestroyStream.
	63	+	type streamLoopContext struct {
	64	+	mainLoop unsafe.Pointer
	65	+	pwLoop unsafe.Pointer
	66	+	}
	67	+
62	68		type streamOpsImpl struct {
63	69		mu sync.Mutex
64	70		pinned map[unsafe.Pointer]*streamCallbackStorage // keyed by stream ptr
65	71		pinnedParams map[unsafe.Pointer]*connectParams // keyed by stream ptr
	72	+	streamLoops map[unsafe.Pointer]streamLoopContext // keyed by stream ptr
	73	+	runningLoops map[unsafe.Pointer]bool // keyed by main loop ptr
66	74		}
67	75
68	76		// Verify interface compliance at compile time.

@@ -104,7 +112,11 @@ func (s *streamOpsImpl) CreatePlaybackStream(loopPtr unsafe.Pointer, name string
104	112		if s.pinned == nil {
105	113		s.pinned = make(map[unsafe.Pointer]*streamCallbackStorage)
106	114		}
	115	+	if s.streamLoops == nil {
	116	+	s.streamLoops = make(map[unsafe.Pointer]streamLoopContext)
	117	+	}
107	118		s.pinned[ptr] = storage
	119	+	s.streamLoops[ptr] = streamLoopContext{mainLoop: loopPtr, pwLoop: pwLoop}
108	120		s.mu.Unlock()
109	121
110	122		return ptr, nil

@@ -138,7 +150,9 @@ func (s *streamOpsImpl) ConnectPlaybackStream(streamPtr unsafe.Pointer, format p
138	150		}
139	151
140	152		func (s *streamOpsImpl) SetStreamActive(streamPtr unsafe.Pointer, active bool) error {
141		-	ret := pw_stream_set_active(streamPtr, active)
	153	+	ret := s.callStreamWithLoopLock(streamPtr, func() int32 {
	154	+	return pw_stream_set_active(streamPtr, active)
	155	+	})
142	156		if ret < 0 {
143	157		return &PWError{Func: "pw_stream_set_active", Code: ret}
144	158		}

@@ -160,27 +174,37 @@ func (s *streamOpsImpl) QueueBuffer(streamPtr unsafe.Pointer, bufPtr unsafe.Poin
160	174		// DisconnectStream disconnects the stream from its port.
161	175		// Returns a PWError if the operation fails.
162	176		func (s *streamOpsImpl) DisconnectStream(streamPtr unsafe.Pointer) error {
163		-	ret := pw_stream_disconnect(streamPtr)
	177	+	ret := s.callStreamWithLoopLock(streamPtr, func() int32 {
	178	+	return pw_stream_disconnect(streamPtr)
	179	+	})
164	180		if ret < 0 {
165	181		return &PWError{Func: "pw_stream_disconnect", Code: ret}
166	182		}
167	183		return nil
168	184		}
169	185
	186	+	// DestroyStream releases the pw_stream and associated pinned Go storage.
	187	+	// Callers must ensure the owning main loop has already stopped, or that they
	188	+	// otherwise satisfy PipeWire's required stream context, before invoking this.
170	189		func (s *streamOpsImpl) DestroyStream(streamPtr unsafe.Pointer) {
171	190		s.mu.Lock()
172		-	if _, alive := s.pinned[streamPtr]; !alive {
	191	+	storage, alive := s.pinned[streamPtr]
	192	+	if !alive {
173	193		// Already destroyed or never tracked — skip.
174	194		s.mu.Unlock()
175	195		return
176	196		}
177		-	// Unpin callback storage and connect params — safe now that the stream is being destroyed.
	197	+	params := s.pinnedParams[streamPtr]
178	198		delete(s.pinned, streamPtr)
179	199		delete(s.pinnedParams, streamPtr)
	200	+	delete(s.streamLoops, streamPtr)
180	201		s.mu.Unlock()
181	202
182		-	// Use the generated pw_stream_destroy binding for proper cleanup.
183	203		pw_stream_destroy(streamPtr)
	204	+
	205	+	// Keep callback and format storage alive until after pw_stream_destroy returns.
	206	+	runtime.KeepAlive(storage)
	207	+	runtime.KeepAlive(params)
184	208		}
185	209
186	210		func (s *streamOpsImpl) CreateMainLoop() (unsafe.Pointer, error) {

@@ -192,6 +216,18 @@ func (s *streamOpsImpl) CreateMainLoop() (unsafe.Pointer, error) {
192	216		}
193	217
194	218		func (s *streamOpsImpl) RunMainLoop(loopPtr unsafe.Pointer) error {
	219	+	s.mu.Lock()
	220	+	if s.runningLoops == nil {
	221	+	s.runningLoops = make(map[unsafe.Pointer]bool)
	222	+	}
	223	+	s.runningLoops[loopPtr] = true
	224	+	s.mu.Unlock()
	225	+	defer func() {
	226	+	s.mu.Lock()
	227	+	delete(s.runningLoops, loopPtr)
	228	+	s.mu.Unlock()
	229	+	}()
	230	+
195	231		ret := pw_main_loop_run(loopPtr)
196	232		if ret < 0 {
197	233		return &PWError{Func: "pw_main_loop_run", Code: ret}

@@ -208,6 +244,27 @@ func (s *streamOpsImpl) DestroyMainLoop(loopPtr unsafe.Pointer) {
208	244		pw_main_loop_destroy(loopPtr)
209	245		}
210	246
	247	+	func (s *streamOpsImpl) callStreamWithLoopLock(streamPtr unsafe.Pointer, fn func() int32) int32 {
	248	+	s.mu.Lock()
	249	+	ctx, running := s.streamLoopContextLocked(streamPtr)
	250	+	s.mu.Unlock()
	251	+	if running && ctx.pwLoop != nil {
	252	+	return withLoopLock(ctx.pwLoop, fn)
	253	+	}
	254	+	return fn()
	255	+	}
	256	+
	257	+	func (s *streamOpsImpl) streamLoopContextLocked(streamPtr unsafe.Pointer) (streamLoopContext, bool) {
	258	+	if s.streamLoops == nil {
	259	+	return streamLoopContext{}, false
	260	+	}
	261	+	ctx, ok := s.streamLoops[streamPtr]
	262	+	if !ok {
	263	+	return streamLoopContext{}, false
	264	+	}
	265	+	return ctx, s.runningLoops != nil && s.runningLoops[ctx.mainLoop]
	266	+	}
	267	+
211	268		// DefaultStreamOps returns a StreamOps implementation backed by the real
212	269		// PipeWire C API. Must be called after Register().
213	270		func DefaultStreamOps() portout.StreamOps {

 import (
 	"errors"
 	"fmt"
+	"runtime"
 	"sync"
 	"unsafe"
 //     PipeWire reads the SPA POD params asynchronously during format negotiation, so the
 //     backing byte storage must remain live until DestroyStream releases it.
 //   - Both maps are keyed by the stream pointer and must be cleaned up in DestroyStream.
+type streamLoopContext struct {
+	mainLoop unsafe.Pointer
+	pwLoop   unsafe.Pointer
+}
 type streamOpsImpl struct {
 	mu           sync.Mutex
 	pinned       map[unsafe.Pointer]*streamCallbackStorage // keyed by stream ptr
 	pinnedParams map[unsafe.Pointer]*connectParams         // keyed by stream ptr
+	streamLoops  map[unsafe.Pointer]streamLoopContext      // keyed by stream ptr
+	runningLoops map[unsafe.Pointer]bool                   // keyed by main loop ptr
 }
 // Verify interface compliance at compile time.
 	if s.pinned == nil {
 		s.pinned = make(map[unsafe.Pointer]*streamCallbackStorage)
 	}
+	if s.streamLoops == nil {
+		s.streamLoops = make(map[unsafe.Pointer]streamLoopContext)
+	}
 	s.pinned[ptr] = storage
+	s.streamLoops[ptr] = streamLoopContext{mainLoop: loopPtr, pwLoop: pwLoop}
 	s.mu.Unlock()
 	return ptr, nil
 }
 func (s *streamOpsImpl) SetStreamActive(streamPtr unsafe.Pointer, active bool) error {
-	ret := pw_stream_set_active(streamPtr, active)
+	ret := s.callStreamWithLoopLock(streamPtr, func() int32 {
+		return pw_stream_set_active(streamPtr, active)
+	})
 	if ret < 0 {
 		return &PWError{Func: "pw_stream_set_active", Code: ret}
 	}
 // DisconnectStream disconnects the stream from its port.
 // Returns a PWError if the operation fails.
 func (s *streamOpsImpl) DisconnectStream(streamPtr unsafe.Pointer) error {
-	ret := pw_stream_disconnect(streamPtr)
+	ret := s.callStreamWithLoopLock(streamPtr, func() int32 {
+		return pw_stream_disconnect(streamPtr)
+	})
 	if ret < 0 {
 		return &PWError{Func: "pw_stream_disconnect", Code: ret}
 	}
 	return nil
 }
+// DestroyStream releases the pw_stream and associated pinned Go storage.
+// Callers must ensure the owning main loop has already stopped, or that they
+// otherwise satisfy PipeWire's required stream context, before invoking this.
 func (s *streamOpsImpl) DestroyStream(streamPtr unsafe.Pointer) {
 	s.mu.Lock()
-	if _, alive := s.pinned[streamPtr]; !alive {
+	storage, alive := s.pinned[streamPtr]
+	if !alive {
 		// Already destroyed or never tracked — skip.
 		s.mu.Unlock()
 		return
 	}
-	// Unpin callback storage and connect params — safe now that the stream is being destroyed.
+	params := s.pinnedParams[streamPtr]
 	delete(s.pinned, streamPtr)
 	delete(s.pinnedParams, streamPtr)
+	delete(s.streamLoops, streamPtr)
 	s.mu.Unlock()
-	// Use the generated pw_stream_destroy binding for proper cleanup.
 	pw_stream_destroy(streamPtr)
+	// Keep callback and format storage alive until after pw_stream_destroy returns.
+	runtime.KeepAlive(storage)
+	runtime.KeepAlive(params)
 }
 func (s *streamOpsImpl) CreateMainLoop() (unsafe.Pointer, error) {
 }
 func (s *streamOpsImpl) RunMainLoop(loopPtr unsafe.Pointer) error {
+	s.mu.Lock()
+	if s.runningLoops == nil {
+		s.runningLoops = make(map[unsafe.Pointer]bool)
+	}
+	s.runningLoops[loopPtr] = true
+	s.mu.Unlock()
+	defer func() {
+		s.mu.Lock()
+		delete(s.runningLoops, loopPtr)
+		s.mu.Unlock()
+	}()
 	ret := pw_main_loop_run(loopPtr)
 	if ret < 0 {
 		return &PWError{Func: "pw_main_loop_run", Code: ret}
 	pw_main_loop_destroy(loopPtr)
 }
+func (s *streamOpsImpl) callStreamWithLoopLock(streamPtr unsafe.Pointer, fn func() int32) int32 {
+	s.mu.Lock()
+	ctx, running := s.streamLoopContextLocked(streamPtr)
+	s.mu.Unlock()
+	if running && ctx.pwLoop != nil {
+		return withLoopLock(ctx.pwLoop, fn)
+	}
+	return fn()
+}
+func (s *streamOpsImpl) streamLoopContextLocked(streamPtr unsafe.Pointer) (streamLoopContext, bool) {
+	if s.streamLoops == nil {
+		return streamLoopContext{}, false
+	}
+	ctx, ok := s.streamLoops[streamPtr]
+	if !ok {
+		return streamLoopContext{}, false
+	}
+	return ctx, s.runningLoops != nil && s.runningLoops[ctx.mainLoop]
+}
 // DefaultStreamOps returns a StreamOps implementation backed by the real
 // PipeWire C API. Must be called after Register().
 func DefaultStreamOps() portout.StreamOps {

M internal/capi/stream_ops_test.go +140 -0


@@ -50,6 +50,121 @@ func TestDestroyStreamCallsPwStreamDestroy(t *testing.T) {
50	50		}
51	51		}
52	52
	53	+	func TestSetStreamActiveUsesOwningLoopLockWhenRunning(t *testing.T) {
	54	+	origWithLoopLock := withLoopLock
	55	+	origSetActive := pw_stream_set_active
	56	+	t.Cleanup(func() {
	57	+	withLoopLock = origWithLoopLock
	58	+	pw_stream_set_active = origSetActive
	59	+	})
	60	+
	61	+	fakeStream := opaquePtr()
	62	+	fakeMainLoop := opaquePtr()
	63	+	fakePWLoop := opaquePtr()
	64	+	ops := &streamOpsImpl{
	65	+	streamLoops: map[unsafe.Pointer]streamLoopContext{fakeStream: {mainLoop: fakeMainLoop, pwLoop: fakePWLoop}},
	66	+	runningLoops: map[unsafe.Pointer]bool{fakeMainLoop: true},
	67	+	}
	68	+
	69	+	var lockedLoop unsafe.Pointer
	70	+	withLoopLock = func(loop unsafe.Pointer, fn func() int32) int32 {
	71	+	lockedLoop = loop
	72	+	return fn()
	73	+	}
	74	+	var setActiveStream unsafe.Pointer
	75	+	var setActiveValue bool
	76	+	pw_stream_set_active = func(stream unsafe.Pointer, active bool) int32 {
	77	+	setActiveStream = stream
	78	+	setActiveValue = active
	79	+	return 0
	80	+	}
	81	+
	82	+	if err := ops.SetStreamActive(fakeStream, false); err != nil {
	83	+	t.Fatalf("SetStreamActive returned error: %v", err)
	84	+	}
	85	+	if lockedLoop != fakePWLoop {
	86	+	t.Fatalf("locked loop = %v, want %v", lockedLoop, fakePWLoop)
	87	+	}
	88	+	if setActiveStream != fakeStream \|\| setActiveValue {
	89	+	t.Fatalf("pw_stream_set_active called with (%v, %v), want (%v, false)", setActiveStream, setActiveValue, fakeStream)
	90	+	}
	91	+	}
	92	+
	93	+	func TestDestroyStreamUnpinsAndDestroys(t *testing.T) {
	94	+	origDestroy := pw_stream_destroy
	95	+	t.Cleanup(func() {
	96	+	pw_stream_destroy = origDestroy
	97	+	})
	98	+
	99	+	fakeStream := opaquePtr()
	100	+	fakeMainLoop := opaquePtr()
	101	+	fakePWLoop := opaquePtr()
	102	+	ops := &streamOpsImpl{
	103	+	pinned: map[unsafe.Pointer]*streamCallbackStorage{fakeStream: {}},
	104	+	pinnedParams: map[unsafe.Pointer]*connectParams{fakeStream: {}},
	105	+	streamLoops: map[unsafe.Pointer]streamLoopContext{fakeStream: {mainLoop: fakeMainLoop, pwLoop: fakePWLoop}},
	106	+	runningLoops: map[unsafe.Pointer]bool{fakeMainLoop: true},
	107	+	}
	108	+
	109	+	var destroyedStream unsafe.Pointer
	110	+	pw_stream_destroy = func(stream unsafe.Pointer) {
	111	+	destroyedStream = stream
	112	+	}
	113	+
	114	+	ops.DestroyStream(fakeStream)
	115	+
	116	+	if destroyedStream != fakeStream {
	117	+	t.Fatalf("destroyed stream = %v, want %v", destroyedStream, fakeStream)
	118	+	}
	119	+	if _, ok := ops.pinned[fakeStream]; ok {
	120	+	t.Fatal("expected pinned callback storage to be removed")
	121	+	}
	122	+	if _, ok := ops.pinnedParams[fakeStream]; ok {
	123	+	t.Fatal("expected pinned params to be removed")
	124	+	}
	125	+	if _, ok := ops.streamLoops[fakeStream]; ok {
	126	+	t.Fatal("expected stream loop context to be removed")
	127	+	}
	128	+	}
	129	+
	130	+	func TestDisconnectStreamUsesOwningLoopLockWhenRunning(t *testing.T) {
	131	+	origWithLoopLock := withLoopLock
	132	+	origDisconnect := pw_stream_disconnect
	133	+	t.Cleanup(func() {
	134	+	withLoopLock = origWithLoopLock
	135	+	pw_stream_disconnect = origDisconnect
	136	+	})
	137	+
	138	+	fakeStream := opaquePtr()
	139	+	fakeMainLoop := opaquePtr()
	140	+	fakePWLoop := opaquePtr()
	141	+	ops := &streamOpsImpl{
	142	+	streamLoops: map[unsafe.Pointer]streamLoopContext{fakeStream: {mainLoop: fakeMainLoop, pwLoop: fakePWLoop}},
	143	+	runningLoops: map[unsafe.Pointer]bool{fakeMainLoop: true},
	144	+	}
	145	+
	146	+	var lockedLoop unsafe.Pointer
	147	+	withLoopLock = func(loop unsafe.Pointer, fn func() int32) int32 {
	148	+	lockedLoop = loop
	149	+	return fn()
	150	+	}
	151	+	var disconnectedStream unsafe.Pointer
	152	+	pw_stream_disconnect = func(stream unsafe.Pointer) int32 {
	153	+	disconnectedStream = stream
	154	+	return 0
	155	+	}
	156	+
	157	+	if err := ops.DisconnectStream(fakeStream); err != nil {
	158	+	t.Fatalf("DisconnectStream returned error: %v", err)
	159	+	}
	160	+	if lockedLoop != fakePWLoop {
	161	+	t.Fatalf("locked loop = %v, want %v", lockedLoop, fakePWLoop)
	162	+	}
	163	+	if disconnectedStream != fakeStream {
	164	+	t.Fatalf("disconnected stream = %v, want %v", disconnectedStream, fakeStream)
	165	+	}
	166	+	}
	167	+
53	168		// TestPWStreamEventsUsesFunctionPointerABI verifies that the generated callback
54	169		// fields use raw function-pointer storage, not Go func pointers.
55	170		func TestPWStreamEventsUsesFunctionPointerABI(t *testing.T) {

@@ -449,6 +564,31 @@ func TestRunMainLoopReturnsNilOnSuccess(t *testing.T) {
449	564		}
450	565		}
451	566
	567	+	func TestRunMainLoopRemovesRunningLoopOnPanic(t *testing.T) {
	568	+	origRun := pw_main_loop_run
	569	+	t.Cleanup(func() { pw_main_loop_run = origRun })
	570	+
	571	+	pw_main_loop_run = func(loop unsafe.Pointer) int32 {
	572	+	panic("boom")
	573	+	}
	574	+
	575	+	ops := &streamOpsImpl{}
	576	+	fakeLoop := opaquePtr()
	577	+
	578	+	defer func() {
	579	+	if r := recover(); r == nil {
	580	+	t.Fatal("expected panic from pw_main_loop_run")
	581	+	}
	582	+	ops.mu.Lock()
	583	+	defer ops.mu.Unlock()
	584	+	if len(ops.runningLoops) != 0 {
	585	+	t.Fatalf("expected runningLoops to be cleaned up, got %d entries", len(ops.runningLoops))
	586	+	}
	587	+	}()
	588	+
	589	+	_ = ops.RunMainLoop(fakeLoop)
	590	+	}
	591	+
452	592		// TestCreatePlaybackStreamUsesPWLoopFromMainLoop verifies that CreatePlaybackStream
453	593		// calls pw_main_loop_get_loop to obtain the pw_loop pointer and passes it to
454	594		// pw_stream_new_simple (not the raw main loop pointer).


@@ -50,6 +50,121 @@ func TestDestroyStreamCallsPwStreamDestroy(t *testing.T) {
50	}	50		}
51	}	51		}
52		52
		53	+	func TestSetStreamActiveUsesOwningLoopLockWhenRunning(t *testing.T) {
		54	+	origWithLoopLock := withLoopLock
		55	+	origSetActive := pw_stream_set_active
		56	+	t.Cleanup(func() {
		57	+	withLoopLock = origWithLoopLock
		58	+	pw_stream_set_active = origSetActive
		59	+	})
		60	+
		61	+	fakeStream := opaquePtr()
		62	+	fakeMainLoop := opaquePtr()
		63	+	fakePWLoop := opaquePtr()
		64	+	ops := &streamOpsImpl{
		65	+	streamLoops: map[unsafe.Pointer]streamLoopContext{fakeStream: {mainLoop: fakeMainLoop, pwLoop: fakePWLoop}},
		66	+	runningLoops: map[unsafe.Pointer]bool{fakeMainLoop: true},
		67	+	}
		68	+
		69	+	var lockedLoop unsafe.Pointer
		70	+	withLoopLock = func(loop unsafe.Pointer, fn func() int32) int32 {
		71	+	lockedLoop = loop
		72	+	return fn()
		73	+	}
		74	+	var setActiveStream unsafe.Pointer
		75	+	var setActiveValue bool
		76	+	pw_stream_set_active = func(stream unsafe.Pointer, active bool) int32 {
		77	+	setActiveStream = stream
		78	+	setActiveValue = active
		79	+	return 0
		80	+	}
		81	+
		82	+	if err := ops.SetStreamActive(fakeStream, false); err != nil {
		83	+	t.Fatalf("SetStreamActive returned error: %v", err)
		84	+	}
		85	+	if lockedLoop != fakePWLoop {
		86	+	t.Fatalf("locked loop = %v, want %v", lockedLoop, fakePWLoop)
		87	+	}
		88	+	if setActiveStream != fakeStream \|\| setActiveValue {
		89	+	t.Fatalf("pw_stream_set_active called with (%v, %v), want (%v, false)", setActiveStream, setActiveValue, fakeStream)
		90	+	}
		91	+	}
		92	+
		93	+	func TestDestroyStreamUnpinsAndDestroys(t *testing.T) {
		94	+	origDestroy := pw_stream_destroy
		95	+	t.Cleanup(func() {
		96	+	pw_stream_destroy = origDestroy
		97	+	})
		98	+
		99	+	fakeStream := opaquePtr()
		100	+	fakeMainLoop := opaquePtr()
		101	+	fakePWLoop := opaquePtr()
		102	+	ops := &streamOpsImpl{
		103	+	pinned: map[unsafe.Pointer]*streamCallbackStorage{fakeStream: {}},
		104	+	pinnedParams: map[unsafe.Pointer]*connectParams{fakeStream: {}},
		105	+	streamLoops: map[unsafe.Pointer]streamLoopContext{fakeStream: {mainLoop: fakeMainLoop, pwLoop: fakePWLoop}},
		106	+	runningLoops: map[unsafe.Pointer]bool{fakeMainLoop: true},
		107	+	}
		108	+
		109	+	var destroyedStream unsafe.Pointer
		110	+	pw_stream_destroy = func(stream unsafe.Pointer) {
		111	+	destroyedStream = stream
		112	+	}
		113	+
		114	+	ops.DestroyStream(fakeStream)
		115	+
		116	+	if destroyedStream != fakeStream {
		117	+	t.Fatalf("destroyed stream = %v, want %v", destroyedStream, fakeStream)
		118	+	}
		119	+	if _, ok := ops.pinned[fakeStream]; ok {
		120	+	t.Fatal("expected pinned callback storage to be removed")
		121	+	}
		122	+	if _, ok := ops.pinnedParams[fakeStream]; ok {
		123	+	t.Fatal("expected pinned params to be removed")
		124	+	}
		125	+	if _, ok := ops.streamLoops[fakeStream]; ok {
		126	+	t.Fatal("expected stream loop context to be removed")
		127	+	}
		128	+	}
		129	+
		130	+	func TestDisconnectStreamUsesOwningLoopLockWhenRunning(t *testing.T) {
		131	+	origWithLoopLock := withLoopLock
		132	+	origDisconnect := pw_stream_disconnect
		133	+	t.Cleanup(func() {
		134	+	withLoopLock = origWithLoopLock
		135	+	pw_stream_disconnect = origDisconnect
		136	+	})
		137	+
		138	+	fakeStream := opaquePtr()
		139	+	fakeMainLoop := opaquePtr()
		140	+	fakePWLoop := opaquePtr()
		141	+	ops := &streamOpsImpl{
		142	+	streamLoops: map[unsafe.Pointer]streamLoopContext{fakeStream: {mainLoop: fakeMainLoop, pwLoop: fakePWLoop}},
		143	+	runningLoops: map[unsafe.Pointer]bool{fakeMainLoop: true},
		144	+	}
		145	+
		146	+	var lockedLoop unsafe.Pointer
		147	+	withLoopLock = func(loop unsafe.Pointer, fn func() int32) int32 {
		148	+	lockedLoop = loop
		149	+	return fn()
		150	+	}
		151	+	var disconnectedStream unsafe.Pointer
		152	+	pw_stream_disconnect = func(stream unsafe.Pointer) int32 {
		153	+	disconnectedStream = stream
		154	+	return 0
		155	+	}
		156	+
		157	+	if err := ops.DisconnectStream(fakeStream); err != nil {
		158	+	t.Fatalf("DisconnectStream returned error: %v", err)
		159	+	}
		160	+	if lockedLoop != fakePWLoop {
		161	+	t.Fatalf("locked loop = %v, want %v", lockedLoop, fakePWLoop)
		162	+	}
		163	+	if disconnectedStream != fakeStream {
		164	+	t.Fatalf("disconnected stream = %v, want %v", disconnectedStream, fakeStream)
		165	+	}
		166	+	}
		167	+
53	// TestPWStreamEventsUsesFunctionPointerABI verifies that the generated callback	168		// TestPWStreamEventsUsesFunctionPointerABI verifies that the generated callback
54	// fields use raw function-pointer storage, not Go func pointers.	169		// fields use raw function-pointer storage, not Go func pointers.
55	func TestPWStreamEventsUsesFunctionPointerABI(t *testing.T) {	170		func TestPWStreamEventsUsesFunctionPointerABI(t *testing.T) {

@@ -449,6 +564,31 @@ func TestRunMainLoopReturnsNilOnSuccess(t *testing.T) {
449	}	564		}
450	}	565		}
451		566
		567	+	func TestRunMainLoopRemovesRunningLoopOnPanic(t *testing.T) {
		568	+	origRun := pw_main_loop_run
		569	+	t.Cleanup(func() { pw_main_loop_run = origRun })
		570	+
		571	+	pw_main_loop_run = func(loop unsafe.Pointer) int32 {
		572	+	panic("boom")
		573	+	}
		574	+
		575	+	ops := &streamOpsImpl{}
		576	+	fakeLoop := opaquePtr()
		577	+
		578	+	defer func() {
		579	+	if r := recover(); r == nil {
		580	+	t.Fatal("expected panic from pw_main_loop_run")
		581	+	}
		582	+	ops.mu.Lock()
		583	+	defer ops.mu.Unlock()
		584	+	if len(ops.runningLoops) != 0 {
		585	+	t.Fatalf("expected runningLoops to be cleaned up, got %d entries", len(ops.runningLoops))
		586	+	}
		587	+	}()
		588	+
		589	+	_ = ops.RunMainLoop(fakeLoop)
		590	+	}
		591	+
452	// TestCreatePlaybackStreamUsesPWLoopFromMainLoop verifies that CreatePlaybackStream	592		// TestCreatePlaybackStreamUsesPWLoopFromMainLoop verifies that CreatePlaybackStream
453	// calls pw_main_loop_get_loop to obtain the pw_loop pointer and passes it to	593		// calls pw_main_loop_get_loop to obtain the pw_loop pointer and passes it to
454	// pw_stream_new_simple (not the raw main loop pointer).	594		// pw_stream_new_simple (not the raw main loop pointer).

M internal/core/player.go +24 -4


@@ -36,6 +36,7 @@ type player struct {
36	36		streamOps portout.StreamOps
37	37		streamPtr unsafe.Pointer
38	38		loopPtr unsafe.Pointer
	39	+	loopDone chan struct{}
39	40		}
40	41
41	42		// newPlayer creates a new player instance

@@ -201,13 +202,16 @@ func (p *player) startCreateResourcesAndActivate() error {
201	202		}
202	203
203	204		// All steps succeeded — store owned resources
	205	+	loopDone := make(chan struct{})
204	206		p.mu.Lock()
205	207		p.loopPtr = loopPtr
	208	+	p.loopDone = loopDone
206	209		atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), streamPtr)
207	210		p.mu.Unlock()
208	211
209	212		// Start main loop in an internal goroutine
210	213		go func() {
	214	+	defer close(loopDone)
211	215		if err := p.streamOps.RunMainLoop(loopPtr); err != nil {
212	216		if state := p.State(); state != PlayerStateClosing && state != PlayerStateClosed {
213	217		p.fail(err)

@@ -336,30 +340,46 @@ func (p *player) deactivateStream() error {
336	340
337	341		// teardown releases player-owned stream and main-loop resources through
338	342		// StreamOps. Safe to call when fields are nil or on repeated invocation.
339		-	// DisconnectStream is called before DestroyStream as best-effort cleanup;
340		-	// if disconnect fails, teardown continues (destroy is the definitive release).
	343	+	// When a main loop exists, DisconnectStream is issued first while the loop is
	344	+	// still running, then the loop is quit and awaited, and only then is the
	345	+	// stream destroyed. This ordering avoids destroying a pw_stream while its
	346	+	// main loop is still active.
341	347		func (p *player) teardown() {
342	348		p.mu.Lock()
343	349		streamOps := p.streamOps
344	350		streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
345	351		loopPtr := p.loopPtr
	352	+	loopDone := p.loopDone
346	353		atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), nil)
347	354		p.loopPtr = nil
	355	+	p.loopDone = nil
348	356		p.mu.Unlock()
349	357
350	358		if streamOps == nil {
351	359		return
352	360		}
353	361
354		-	// Disconnect then destroy stream (disconnect is best-effort, destroy is definitive).
	362	+	// If a stream exists, disconnect it first while the loop is still alive.
355	363		if streamPtr != nil {
356	364		_ = streamOps.DisconnectStream(streamPtr) // Best-effort cleanup
357		-	streamOps.DestroyStream(streamPtr)
358	365		}
359	366
360	367		// Quit then destroy the main loop.
	368	+	// Never destroy the main loop until RunMainLoop has actually returned;
	369	+	// destroying while pw_main_loop_run is still active can crash in PipeWire.
361	370		if loopPtr != nil {
362	371		streamOps.QuitMainLoop(loopPtr)
	372	+	if loopDone != nil {
	373	+	<-loopDone
	374	+	}
	375	+	}
	376	+
	377	+	// Only destroy the stream after the loop has stopped so pw_stream_destroy
	378	+	// can safely flush any queued loop callbacks before freeing stream state.
	379	+	if streamPtr != nil {
	380	+	streamOps.DestroyStream(streamPtr)
	381	+	}
	382	+	if loopPtr != nil {
363	383		streamOps.DestroyMainLoop(loopPtr)
364	384		}
365	385		}

 	streamOps portout.StreamOps
 	streamPtr unsafe.Pointer
 	loopPtr   unsafe.Pointer
+	loopDone  chan struct{}
 }
 // newPlayer creates a new player instance
 	}
 	// All steps succeeded — store owned resources
+	loopDone := make(chan struct{})
 	p.mu.Lock()
 	p.loopPtr = loopPtr
+	p.loopDone = loopDone
 	atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), streamPtr)
 	p.mu.Unlock()
 	// Start main loop in an internal goroutine
 	go func() {
+		defer close(loopDone)
 		if err := p.streamOps.RunMainLoop(loopPtr); err != nil {
 			if state := p.State(); state != PlayerStateClosing && state != PlayerStateClosed {
 				p.fail(err)
 // teardown releases player-owned stream and main-loop resources through
 // StreamOps. Safe to call when fields are nil or on repeated invocation.
-// DisconnectStream is called before DestroyStream as best-effort cleanup;
+// When a main loop exists, DisconnectStream is issued first while the loop is
-// if disconnect fails, teardown continues (destroy is the definitive release).
+// still running, then the loop is quit and awaited, and only then is the
+// stream destroyed. This ordering avoids destroying a pw_stream while its
+// main loop is still active.
 func (p *player) teardown() {
 	p.mu.Lock()
 	streamOps := p.streamOps
 	streamPtr := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)))
 	loopPtr := p.loopPtr
+	loopDone := p.loopDone
 	atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&p.streamPtr)), nil)
 	p.loopPtr = nil
+	p.loopDone = nil
 	p.mu.Unlock()
 	if streamOps == nil {
 		return
 	}
-	// Disconnect then destroy stream (disconnect is best-effort, destroy is definitive).
+	// If a stream exists, disconnect it first while the loop is still alive.
 	if streamPtr != nil {
 		_ = streamOps.DisconnectStream(streamPtr) // Best-effort cleanup
-		streamOps.DestroyStream(streamPtr)
 	}
 	// Quit then destroy the main loop.
+	// Never destroy the main loop until RunMainLoop has actually returned;
+	// destroying while pw_main_loop_run is still active can crash in PipeWire.
 	if loopPtr != nil {
 		streamOps.QuitMainLoop(loopPtr)
+		if loopDone != nil {
+			<-loopDone
+		}
+	}
+	// Only destroy the stream after the loop has stopped so pw_stream_destroy
+	// can safely flush any queued loop callbacks before freeing stream state.
+	if streamPtr != nil {
+		streamOps.DestroyStream(streamPtr)
+	}
+	if loopPtr != nil {
 		streamOps.DestroyMainLoop(loopPtr)
 	}
 }

M internal/core/player_test.go +125 -11


@@ -75,8 +75,8 @@ func TestPlayerStopIsRestartableButCloseIsTerminal(t *testing.T) {
75	75		mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(nil)
76	76		// Close tears down resources
77	77		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
78		-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
79	78		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
	79	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
80	80		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
81	81
82	82		p := newPlayer(cfg, PlayerCallbacks{})

@@ -302,8 +302,8 @@ func TestPlayerCloseIsIdempotent(t *testing.T) {
302	302
303	303		wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
304	304		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
305		-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
306	305		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
	306	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
307	307		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
308	308
309	309		p := newPlayer(cfg, PlayerCallbacks{})

@@ -433,10 +433,10 @@ func TestPlayerCloseTeardownCallsStreamOps(t *testing.T) {
433	433		fakeStream := opaqueTestPtr()
434	434		fakeLoop := opaqueTestPtr()
435	435
436		-	// Expect teardown calls in order: DisconnectStream, DestroyStream, QuitMainLoop, DestroyMainLoop.
	436	+	// Expect teardown calls in order: DisconnectStream, QuitMainLoop, DestroyStream, DestroyMainLoop.
437	437		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
438		-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
439	438		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
	439	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
440	440		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
441	441
442	442		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})

@@ -487,8 +487,8 @@ func TestPlayerRepeatedCloseIsSafe(t *testing.T) {
487	487
488	488		// Only expect one set of teardown calls.
489	489		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
490		-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
491	490		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
	491	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
492	492		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
493	493
494	494		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})

@@ -551,6 +551,120 @@ func TestPlayerTeardownWithOnlyLoopPtr(t *testing.T) {
551	551		}
552	552		}
553	553
	554	+	func TestPlayerCloseWaitsForMainLoopExitBeforeDestroy(t *testing.T) {
	555	+	mockOps := mocks.NewMockStreamOps(t)
	556	+	fakeLoop := opaqueTestPtr()
	557	+	loopDone := make(chan struct{})
	558	+	quitCalled := make(chan struct{})
	559	+	destroyCalled := make(chan struct{})
	560	+	closeReturned := make(chan error, 1)
	561	+
	562	+	mockOps.EXPECT().QuitMainLoop(fakeLoop).Run(func(unsafe.Pointer) {
	563	+	close(quitCalled)
	564	+	}).Return()
	565	+	mockOps.EXPECT().DestroyMainLoop(fakeLoop).Run(func(unsafe.Pointer) {
	566	+	close(destroyCalled)
	567	+	}).Return()
	568	+
	569	+	p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
	570	+	p.streamOps = mockOps
	571	+	p.loopPtr = fakeLoop
	572	+	p.loopDone = loopDone
	573	+	p.setState(PlayerStatePlaying)
	574	+
	575	+	go func() {
	576	+	closeReturned <- p.Close()
	577	+	}()
	578	+
	579	+	select {
	580	+	case <-quitCalled:
	581	+	case <-time.After(2 * time.Second):
	582	+	t.Fatal("timed out waiting for QuitMainLoop")
	583	+	}
	584	+
	585	+	select {
	586	+	case <-destroyCalled:
	587	+	t.Fatal("DestroyMainLoop called before loopDone was closed")
	588	+	case <-time.After(100 * time.Millisecond):
	589	+	}
	590	+
	591	+	close(loopDone)
	592	+
	593	+	select {
	594	+	case err := <-closeReturned:
	595	+	if err != nil {
	596	+	t.Fatalf("Close failed: %v", err)
	597	+	}
	598	+	case <-time.After(2 * time.Second):
	599	+	t.Fatal("timed out waiting for Close to return")
	600	+	}
	601	+
	602	+	select {
	603	+	case <-destroyCalled:
	604	+	case <-time.After(2 * time.Second):
	605	+	t.Fatal("timed out waiting for DestroyMainLoop after loopDone")
	606	+	}
	607	+	}
	608	+
	609	+	func TestPlayerCloseWaitsForMainLoopExitBeforeDestroyingStream(t *testing.T) {
	610	+	mockOps := mocks.NewMockStreamOps(t)
	611	+	fakeLoop := opaqueTestPtr()
	612	+	fakeStream := opaqueTestPtr()
	613	+	loopDone := make(chan struct{})
	614	+	quitCalled := make(chan struct{})
	615	+	streamDestroyed := make(chan struct{})
	616	+	closeReturned := make(chan error, 1)
	617	+
	618	+	mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
	619	+	mockOps.EXPECT().QuitMainLoop(fakeLoop).Run(func(unsafe.Pointer) {
	620	+	close(quitCalled)
	621	+	}).Return()
	622	+	mockOps.EXPECT().DestroyStream(fakeStream).Run(func(unsafe.Pointer) {
	623	+	close(streamDestroyed)
	624	+	}).Return()
	625	+	mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
	626	+
	627	+	p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
	628	+	p.streamOps = mockOps
	629	+	p.streamPtr = fakeStream
	630	+	p.loopPtr = fakeLoop
	631	+	p.loopDone = loopDone
	632	+	p.setState(PlayerStatePlaying)
	633	+
	634	+	go func() {
	635	+	closeReturned <- p.Close()
	636	+	}()
	637	+
	638	+	select {
	639	+	case <-quitCalled:
	640	+	case <-time.After(2 * time.Second):
	641	+	t.Fatal("timed out waiting for QuitMainLoop")
	642	+	}
	643	+
	644	+	select {
	645	+	case <-streamDestroyed:
	646	+	t.Fatal("DestroyStream called before loopDone was closed")
	647	+	case <-time.After(100 * time.Millisecond):
	648	+	}
	649	+
	650	+	close(loopDone)
	651	+
	652	+	select {
	653	+	case err := <-closeReturned:
	654	+	if err != nil {
	655	+	t.Fatalf("Close failed: %v", err)
	656	+	}
	657	+	case <-time.After(2 * time.Second):
	658	+	t.Fatal("timed out waiting for Close to return")
	659	+	}
	660	+
	661	+	select {
	662	+	case <-streamDestroyed:
	663	+	case <-time.After(2 * time.Second):
	664	+	t.Fatal("timed out waiting for DestroyStream after loopDone")
	665	+	}
	666	+	}
	667	+
554	668		// TestPlayerTeardownCallsDisconnectStream verifies that teardown calls
555	669		// DisconnectStream before DestroyStream when a stream exists.
556	670		func TestPlayerTeardownCallsDisconnectStream(t *testing.T) {

@@ -558,10 +672,10 @@ func TestPlayerTeardownCallsDisconnectStream(t *testing.T) {
558	672		fakeStream := opaqueTestPtr()
559	673		fakeLoop := opaqueTestPtr()
560	674
561		-	// Expect disconnect before destroy
	675	+	// Expect disconnect before destroy, and stream destroy only after loop shutdown begins.
562	676		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
563		-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
564	677		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
	678	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
565	679		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
566	680
567	681		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})

@@ -583,10 +697,10 @@ func TestPlayerTeardownContinuesOnDisconnectError(t *testing.T) {
583	697		fakeLoop := opaqueTestPtr()
584	698		disconnectErr := errors.New("disconnect failed")
585	699
586		-	// Disconnect fails, but destroy should still be called
	700	+	// Disconnect fails, but teardown should still quit the loop and destroy the stream.
587	701		mockOps.EXPECT().DisconnectStream(fakeStream).Return(disconnectErr)
588		-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
589	702		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
	703	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
590	704		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
591	705
592	706		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})

@@ -617,10 +731,10 @@ func TestPlayerRestartActivationFailureCleansUpResources(t *testing.T) {
617	731		mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)
618	732		// Restart: activation fails → teardown cleans up
619	733		mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(activateErr)
620		-	// teardown should call DisconnectStream, DestroyStream, QuitMainLoop, DestroyMainLoop
	734	+	// teardown should call DisconnectStream, QuitMainLoop, DestroyStream, DestroyMainLoop
621	735		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
622		-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
623	736		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
	737	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
624	738		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
625	739
626	740		p := newPlayer(cfg, PlayerCallbacks{})


@@ -75,8 +75,8 @@ func TestPlayerStopIsRestartableButCloseIsTerminal(t *testing.T) {
75		mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(nil)	75		mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(nil)
76		// Close tears down resources	76		// Close tears down resources
77		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)	77		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
78	-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
79		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()	78		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
			79	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
80		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()	80		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
81			81
82		p := newPlayer(cfg, PlayerCallbacks{})	82		p := newPlayer(cfg, PlayerCallbacks{})

@@ -302,8 +302,8 @@ func TestPlayerCloseIsIdempotent(t *testing.T) {
302			302
303		wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)	303		wg := expectStartWithSync(mockOps, fakeLoop, fakeStream, cfg)
304		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)	304		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
305	-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
306		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()	305		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
			306	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
307		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()	307		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
308			308
309		p := newPlayer(cfg, PlayerCallbacks{})	309		p := newPlayer(cfg, PlayerCallbacks{})

@@ -433,10 +433,10 @@ func TestPlayerCloseTeardownCallsStreamOps(t *testing.T) {
433		fakeStream := opaqueTestPtr()	433		fakeStream := opaqueTestPtr()
434		fakeLoop := opaqueTestPtr()	434		fakeLoop := opaqueTestPtr()
435			435
436	-	// Expect teardown calls in order: DisconnectStream, DestroyStream, QuitMainLoop, DestroyMainLoop.	436	+	// Expect teardown calls in order: DisconnectStream, QuitMainLoop, DestroyStream, DestroyMainLoop.
437		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)	437		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
438	-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
439		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()	438		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
			439	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
440		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()	440		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
441			441
442		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})	442		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})

@@ -487,8 +487,8 @@ func TestPlayerRepeatedCloseIsSafe(t *testing.T) {
487			487
488		// Only expect one set of teardown calls.	488		// Only expect one set of teardown calls.
489		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)	489		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
490	-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
491		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()	490		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
			491	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
492		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()	492		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
493			493
494		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})	494		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})

@@ -551,6 +551,120 @@ func TestPlayerTeardownWithOnlyLoopPtr(t *testing.T) {
551		}	551		}
552		}	552		}
553			553
			554	+	func TestPlayerCloseWaitsForMainLoopExitBeforeDestroy(t *testing.T) {
			555	+	mockOps := mocks.NewMockStreamOps(t)
			556	+	fakeLoop := opaqueTestPtr()
			557	+	loopDone := make(chan struct{})
			558	+	quitCalled := make(chan struct{})
			559	+	destroyCalled := make(chan struct{})
			560	+	closeReturned := make(chan error, 1)
			561	+
			562	+	mockOps.EXPECT().QuitMainLoop(fakeLoop).Run(func(unsafe.Pointer) {
			563	+	close(quitCalled)
			564	+	}).Return()
			565	+	mockOps.EXPECT().DestroyMainLoop(fakeLoop).Run(func(unsafe.Pointer) {
			566	+	close(destroyCalled)
			567	+	}).Return()
			568	+
			569	+	p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
			570	+	p.streamOps = mockOps
			571	+	p.loopPtr = fakeLoop
			572	+	p.loopDone = loopDone
			573	+	p.setState(PlayerStatePlaying)
			574	+
			575	+	go func() {
			576	+	closeReturned <- p.Close()
			577	+	}()
			578	+
			579	+	select {
			580	+	case <-quitCalled:
			581	+	case <-time.After(2 * time.Second):
			582	+	t.Fatal("timed out waiting for QuitMainLoop")
			583	+	}
			584	+
			585	+	select {
			586	+	case <-destroyCalled:
			587	+	t.Fatal("DestroyMainLoop called before loopDone was closed")
			588	+	case <-time.After(100 * time.Millisecond):
			589	+	}
			590	+
			591	+	close(loopDone)
			592	+
			593	+	select {
			594	+	case err := <-closeReturned:
			595	+	if err != nil {
			596	+	t.Fatalf("Close failed: %v", err)
			597	+	}
			598	+	case <-time.After(2 * time.Second):
			599	+	t.Fatal("timed out waiting for Close to return")
			600	+	}
			601	+
			602	+	select {
			603	+	case <-destroyCalled:
			604	+	case <-time.After(2 * time.Second):
			605	+	t.Fatal("timed out waiting for DestroyMainLoop after loopDone")
			606	+	}
			607	+	}
			608	+
			609	+	func TestPlayerCloseWaitsForMainLoopExitBeforeDestroyingStream(t *testing.T) {
			610	+	mockOps := mocks.NewMockStreamOps(t)
			611	+	fakeLoop := opaqueTestPtr()
			612	+	fakeStream := opaqueTestPtr()
			613	+	loopDone := make(chan struct{})
			614	+	quitCalled := make(chan struct{})
			615	+	streamDestroyed := make(chan struct{})
			616	+	closeReturned := make(chan error, 1)
			617	+
			618	+	mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
			619	+	mockOps.EXPECT().QuitMainLoop(fakeLoop).Run(func(unsafe.Pointer) {
			620	+	close(quitCalled)
			621	+	}).Return()
			622	+	mockOps.EXPECT().DestroyStream(fakeStream).Run(func(unsafe.Pointer) {
			623	+	close(streamDestroyed)
			624	+	}).Return()
			625	+	mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
			626	+
			627	+	p := newPlayer(PlayerConfig{}, PlayerCallbacks{})
			628	+	p.streamOps = mockOps
			629	+	p.streamPtr = fakeStream
			630	+	p.loopPtr = fakeLoop
			631	+	p.loopDone = loopDone
			632	+	p.setState(PlayerStatePlaying)
			633	+
			634	+	go func() {
			635	+	closeReturned <- p.Close()
			636	+	}()
			637	+
			638	+	select {
			639	+	case <-quitCalled:
			640	+	case <-time.After(2 * time.Second):
			641	+	t.Fatal("timed out waiting for QuitMainLoop")
			642	+	}
			643	+
			644	+	select {
			645	+	case <-streamDestroyed:
			646	+	t.Fatal("DestroyStream called before loopDone was closed")
			647	+	case <-time.After(100 * time.Millisecond):
			648	+	}
			649	+
			650	+	close(loopDone)
			651	+
			652	+	select {
			653	+	case err := <-closeReturned:
			654	+	if err != nil {
			655	+	t.Fatalf("Close failed: %v", err)
			656	+	}
			657	+	case <-time.After(2 * time.Second):
			658	+	t.Fatal("timed out waiting for Close to return")
			659	+	}
			660	+
			661	+	select {
			662	+	case <-streamDestroyed:
			663	+	case <-time.After(2 * time.Second):
			664	+	t.Fatal("timed out waiting for DestroyStream after loopDone")
			665	+	}
			666	+	}
			667	+
554		// TestPlayerTeardownCallsDisconnectStream verifies that teardown calls	668		// TestPlayerTeardownCallsDisconnectStream verifies that teardown calls
555		// DisconnectStream before DestroyStream when a stream exists.	669		// DisconnectStream before DestroyStream when a stream exists.
556		func TestPlayerTeardownCallsDisconnectStream(t *testing.T) {	670		func TestPlayerTeardownCallsDisconnectStream(t *testing.T) {

@@ -558,10 +672,10 @@ func TestPlayerTeardownCallsDisconnectStream(t *testing.T) {
558		fakeStream := opaqueTestPtr()	672		fakeStream := opaqueTestPtr()
559		fakeLoop := opaqueTestPtr()	673		fakeLoop := opaqueTestPtr()
560			674
561	-	// Expect disconnect before destroy	675	+	// Expect disconnect before destroy, and stream destroy only after loop shutdown begins.
562		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)	676		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
563	-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
564		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()	677		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
			678	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
565		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()	679		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
566			680
567		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})	681		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})

@@ -583,10 +697,10 @@ func TestPlayerTeardownContinuesOnDisconnectError(t *testing.T) {
583		fakeLoop := opaqueTestPtr()	697		fakeLoop := opaqueTestPtr()
584		disconnectErr := errors.New("disconnect failed")	698		disconnectErr := errors.New("disconnect failed")
585			699
586	-	// Disconnect fails, but destroy should still be called	700	+	// Disconnect fails, but teardown should still quit the loop and destroy the stream.
587		mockOps.EXPECT().DisconnectStream(fakeStream).Return(disconnectErr)	701		mockOps.EXPECT().DisconnectStream(fakeStream).Return(disconnectErr)
588	-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
589		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()	702		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
			703	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
590		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()	704		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
591			705
592		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})	706		p := newPlayer(PlayerConfig{}, PlayerCallbacks{})

@@ -617,10 +731,10 @@ func TestPlayerRestartActivationFailureCleansUpResources(t *testing.T) {
617		mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)	731		mockOps.EXPECT().SetStreamActive(fakeStream, false).Return(nil)
618		// Restart: activation fails → teardown cleans up	732		// Restart: activation fails → teardown cleans up
619		mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(activateErr)	733		mockOps.EXPECT().SetStreamActive(fakeStream, true).Return(activateErr)
620	-	// teardown should call DisconnectStream, DestroyStream, QuitMainLoop, DestroyMainLoop	734	+	// teardown should call DisconnectStream, QuitMainLoop, DestroyStream, DestroyMainLoop
621		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)	735		mockOps.EXPECT().DisconnectStream(fakeStream).Return(nil)
622	-	mockOps.EXPECT().DestroyStream(fakeStream).Return()
623		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()	736		mockOps.EXPECT().QuitMainLoop(fakeLoop).Return()
			737	+	mockOps.EXPECT().DestroyStream(fakeStream).Return()
624		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()	738		mockOps.EXPECT().DestroyMainLoop(fakeLoop).Return()
625			739
626		p := newPlayer(cfg, PlayerCallbacks{})	740		p := newPlayer(cfg, PlayerCallbacks{})