fix: synchronize compression changes during frame publish

client.go

@@ -125,20 +125,29 @@ func (c *Client) PublishFrame(f Frame) error {
 	pixels := f.Pixels
 	uncompressedSize := saturateUint32(len(pixels))
 
+	// Snapshot the codec while holding the client lock, then release it before
+	// doing the potentially expensive encode. Keeping the snapshot local makes
+	// the compression flags and payload use the same codec even if
+	// SetCompression runs concurrently.
+	c.mu.RLock()
+	frameCodec := c.codec
+	c.mu.RUnlock()
+
 	// Compress if codec is not raw.
 	var flags protocol.Flag
 	compressionID := protocol.CompressionNone
 
-	if c.codec.ID() != codec.IDRaw {
-		maxSize := c.codec.MaxEncodedSize(len(pixels))
+	codecID := frameCodec.ID()
+	if codecID != codec.IDRaw {
+		maxSize := frameCodec.MaxEncodedSize(len(pixels))
 		dst := make([]byte, maxSize)
-		compressed, err := c.codec.Encode(dst, pixels)
+		compressed, err := frameCodec.Encode(dst, pixels)
 		if err != nil {
 			return fmt.Errorf("compose: compress frame: %w", err)
 		}
 		pixels = compressed
 		flags = flags.Set(protocol.FlagCompressed)
-		compressionID = protocol.Compression(c.codec.ID())
+		compressionID = protocol.Compression(codecID)
 	}
 
 	// Set dirty rect flags.

client_codec_test.go

@@ -0,0 +1,89 @@
+package compose
+
+import (
+	"bytes"
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+func TestClientSetCompressionConcurrentPublishFrame(t *testing.T) {
+	addr := tempSocket(t)
+
+	srv, err := Listen(addr)
+	if err != nil {
+		t.Fatalf("Listen: %v", err)
+	}
+	t.Cleanup(func() { _ = srv.Close() })
+
+	const (
+		width  = 32
+		height = 32
+		frames = 128
+	)
+	expectedPixels := makePixels(width, height, 0xA5)
+	var received atomic.Int64
+	var invalid atomic.Bool
+	srv.OnFrame(func(f Frame) {
+		if f.Width != width || f.Height != height || !bytes.Equal(f.Pixels, expectedPixels) {
+			invalid.Store(true)
+		}
+		received.Add(1)
+	})
+
+	client, err := Dial(addr, WithName("codec-race"), WithFrameSize(width, height))
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	t.Cleanup(func() { _ = client.Close() })
+
+	start := make(chan struct{})
+	errs := make(chan error, frames)
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	// Publish and switch codecs from separate goroutines. The scheduler hint
+	// keeps both operations interleaved while retaining deterministic inputs.
+	go func() {
+		defer wg.Done()
+		<-start
+		for i := 0; i < frames; i++ {
+			if err := client.PublishFrame(Frame{
+				Pixels: expectedPixels,
+				Width:  width,
+				Height: height,
+			}); err != nil {
+				errs <- err
+			}
+			runtime.Gosched()
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		<-start
+		for i := 0; i < frames*2; i++ {
+			if i%2 == 0 {
+				client.SetCompression("lz4")
+			} else {
+				client.SetCompression("raw")
+			}
+			runtime.Gosched()
+		}
+	}()
+
+	close(start)
+	wg.Wait()
+	close(errs)
+	for publishErr := range errs {
+		t.Errorf("PublishFrame: %v", publishErr)
+	}
+
+	if !waitFor(t, func() bool { return received.Load() == frames }) {
+		t.Fatalf("received %d/%d frames", received.Load(), frames)
+	}
+	if invalid.Load() {
+		t.Fatal("received frame did not match the published payload")
+	}
+}