From 6e22834c5c1fd3afb508fbba1dbacb2370372365 Mon Sep 17 00:00:00 2001
From: Stanislav Fifik <standa.fifik@gmail.com>
Date: Sat, 16 May 2026 14:50:30 +0200
Subject: [PATCH] Fix ESP32 audio dropouts and improve streaming reliability

FLAC encoder now tracks frame numbers and reuses per-channel sample
buffers to reduce allocations. Client dialer always frees the active
slot on release so mDNS re-emissions can reconnect cleanly. Server
send buffer uses drop-oldest instead of hard failure, with lateness
tracking. Added engine stats logging (chunks/sec, kbps) and writer
diagnostics. UI simplified to a single play/pause toggle that
auto-starts playback on preset selection. Added --preset CLI flag
for headless startup. Removed stale systemd service files.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 Dockerfile.cross                              |  1 +
 cmd/live-server/main.go                       |  9 +++
 dist/systemd/sendspin-client.service          | 16 -----
 dist/systemd/sendspin-server.service          | 17 ------
 internal/live/manager.go                      |  6 +-
 internal/live/web/index.html                  | 20 ++++---
 .../internal/server/flac_encoder.go           | 20 +++++--
 .../sendspin-go/pkg/sendspin/client_dialer.go | 18 +++---
 .../pkg/sendspin/client_dialer_test.go        | 37 ++++++------
 .../sendspin-go/pkg/sendspin/server.go        | 14 +++++
 .../sendspin-go/pkg/sendspin/server_client.go | 60 ++++++++++++++++++-
 .../sendspin-go/pkg/sendspin/server_stream.go | 22 +++++++
 12 files changed, 158 insertions(+), 82 deletions(-)
 delete mode 100644 dist/systemd/sendspin-client.service
 delete mode 100644 dist/systemd/sendspin-server.service

diff --git a/Dockerfile.cross b/Dockerfile.cross
index 251e000..5199d53 100644
--- a/Dockerfile.cross
+++ b/Dockerfile.cross
@@ -20,6 +20,7 @@ RUN dpkg --add-architecture armhf \
 
 WORKDIR /src
 COPY go.mod go.sum ./
+COPY third_party/sendspin-go/go.mod third_party/sendspin-go/go.sum ./third_party/sendspin-go/
 RUN go mod download
 
 COPY . .
diff --git a/cmd/live-server/main.go b/cmd/live-server/main.go
index 761db57..5fdaa5d 100644
--- a/cmd/live-server/main.go
+++ b/cmd/live-server/main.go
@@ -60,6 +60,7 @@ func main() {
 		mqttDiscovery string
 		rate          int
 		channels      int
+		preset        string
 	)
 
 	addInt(&port, 8927, "port", "p", "Sendspin WebSocket port")
@@ -74,6 +75,7 @@ func main() {
 	addStr(&mqttDiscovery, "homeassistant", "mqtt-discovery", "", "Home Assistant discovery prefix")
 	addInt(&rate, 48000, "rate", "r", "capture sample rate")
 	addInt(&channels, 2, "channels", "", "capture channels")
+	addStr(&preset, "", "preset", "", "activate this preset id at startup (begins playback immediately)")
 
 	flag.Usage = usage
 	flag.Parse()
@@ -147,6 +149,13 @@ func main() {
 	go func() { serverDone <- server.Start() }()
 	log.Printf("Sendspin server %q on :%d", name, port)
 
+	if preset != "" {
+		if err := mgr.SetActivePreset(preset); err != nil {
+			log.Fatalf("activate preset %q: %v", preset, err)
+		}
+		log.Printf("preset %q activated at startup", preset)
+	}
+
 	sig := make(chan os.Signal, 1)
 	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
 	select {
diff --git a/dist/systemd/sendspin-client.service b/dist/systemd/sendspin-client.service
deleted file mode 100644
index aad7841..0000000
--- a/dist/systemd/sendspin-client.service
+++ /dev/null
@@ -1,16 +0,0 @@
-[Unit]
-Description=Sendspin audio player
-After=pipewire.service network-online.target
-Wants=pipewire.service
-
-[Service]
-ExecStart=%h/.local/bin/sendspin-client \
-    --server 192.168.1.100:8927 \
-    --name %H \
-    --device alsa_output.usb-Device-00.analog-stereo \
-    --volume 80
-Restart=on-failure
-RestartSec=5
-
-[Install]
-WantedBy=default.target
diff --git a/dist/systemd/sendspin-server.service b/dist/systemd/sendspin-server.service
deleted file mode 100644
index b894cb9..0000000
--- a/dist/systemd/sendspin-server.service
+++ /dev/null
@@ -1,17 +0,0 @@
-[Unit]
-Description=Sendspin audio streaming server
-After=pipewire.service network-online.target
-Wants=pipewire.service
-
-[Service]
-ExecStart=%h/.local/bin/sendspin-server \
-    --source alsa_input.usb-Device-00.analog-stereo \
-    --name %H \
-    --port 8927 \
-    --rate 48000 \
-    --channels 2
-Restart=on-failure
-RestartSec=5
-
-[Install]
-WantedBy=default.target
diff --git a/internal/live/manager.go b/internal/live/manager.go
index 65620ba..de79e06 100644
--- a/internal/live/manager.go
+++ b/internal/live/manager.go
@@ -186,9 +186,11 @@ func (m *Manager) SetActivePreset(id string) error {
 		return nil
 	}
 
-	// On entry from off, default to paused; otherwise carry over.
+	// On entry from off, auto-start playback so picking a preset is a
+	// one-click action. Switching between presets carries over the
+	// current playback state.
 	if prev == "" {
-		m.playback = Paused
+		m.playback = Playing
 	}
 
 	m.src.SetTarget(nextSource)
diff --git a/internal/live/web/index.html b/internal/live/web/index.html
index 7e13570..21c3084 100644
--- a/internal/live/web/index.html
+++ b/internal/live/web/index.html
@@ -205,9 +205,7 @@
       <div class="row">
         <span class="grow">Playback</span>
         <div class="btn-group">
-          <button id="btn-play">Play</button>
           <button id="btn-pause">Pause</button>
-          <button id="btn-stop">Stop</button>
         </div>
       </div>
       <div class="row">
@@ -331,10 +329,12 @@ function render() {
   }
   sel.value = state.active_preset || "";
 
-  // playback buttons
-  $("btn-play").disabled  = !state.active_preset || state.playback === "playing";
-  $("btn-pause").disabled = state.playback !== "playing";
-  $("btn-stop").disabled  = !state.active_preset;
+  // playback toggle button: pauses when playing, resumes when paused.
+  // Disabled when no preset is active (selector controls play/stop).
+  const pauseBtn = $("btn-pause");
+  pauseBtn.disabled = !state.active_preset;
+  pauseBtn.textContent = state.playback === "paused" ? "Play" : "Pause";
+  pauseBtn.dataset.cmd = state.playback === "paused" ? "play" : "pause";
 
   // volume slider
   $("volume-label").textContent = state.volume + "%";
@@ -429,9 +429,11 @@ $("active-preset").addEventListener("change", async (e) => {
   refreshAll();
 });
 
-$("btn-play").onclick  = async () => { await api("POST", "/api/playback", { command: "play" }).catch(e => alert(e.message)); refreshAll(); };
-$("btn-pause").onclick = async () => { await api("POST", "/api/playback", { command: "pause" }).catch(e => alert(e.message)); refreshAll(); };
-$("btn-stop").onclick  = async () => { await api("POST", "/api/playback", { command: "stop"  }).catch(e => alert(e.message)); refreshAll(); };
+$("btn-pause").onclick = async () => {
+  const cmd = $("btn-pause").dataset.cmd || "pause";
+  await api("POST", "/api/playback", { command: cmd }).catch(e => alert(e.message));
+  refreshAll();
+};
 
 const vol = $("volume");
 vol.addEventListener("input", () => { volumeDirty = true; $("volume-label").textContent = vol.value + "%"; });
diff --git a/third_party/sendspin-go/internal/server/flac_encoder.go b/third_party/sendspin-go/internal/server/flac_encoder.go
index 18ea2a2..2e91a5e 100644
--- a/third_party/sendspin-go/internal/server/flac_encoder.go
+++ b/third_party/sendspin-go/internal/server/flac_encoder.go
@@ -24,6 +24,8 @@ type FLACEncoder struct {
 	channels    int
 	bitDepth    int
 	blockSize   int
+	frameNum    uint64
+	chanBufs    [][]int32 // reused per-channel sample buffers
 }
 
 // NewFLACEncoder creates a FLAC encoder with prediction analysis enabled.
@@ -54,6 +56,11 @@ func NewFLACEncoder(sampleRate, channels, bitDepth, blockSize int) (*FLACEncoder
 	copy(codecHeader, buf.Bytes())
 	buf.Reset()
 
+	chanBufs := make([][]int32, channels)
+	for ch := range chanBufs {
+		chanBufs[ch] = make([]int32, blockSize)
+	}
+
 	return &FLACEncoder{
 		encoder:     enc,
 		buf:         buf,
@@ -62,6 +69,7 @@ func NewFLACEncoder(sampleRate, channels, bitDepth, blockSize int) (*FLACEncoder
 		channels:    channels,
 		bitDepth:    bitDepth,
 		blockSize:   blockSize,
+		chanBufs:    chanBufs,
 	}, nil
 }
 
@@ -92,20 +100,18 @@ func (e *FLACEncoder) Encode(samples []int32) ([]byte, error) {
 		shift = uint(24 - e.bitDepth)
 	}
 
-	// De-interleave into per-channel slices.
+	// De-interleave into pre-allocated per-channel buffers.
 	subframes := make([]*frame.Subframe, e.channels)
 	for ch := 0; ch < e.channels; ch++ {
-		channelSamples := make([]int32, e.blockSize)
+		buf := e.chanBufs[ch]
 		for i := 0; i < e.blockSize; i++ {
-			channelSamples[i] = samples[i*e.channels+ch] >> shift
+			buf[i] = samples[i*e.channels+ch] >> shift
 		}
 		subframes[ch] = &frame.Subframe{
 			SubHeader: frame.SubHeader{
-				// PredVerbatim signals the encoder to run prediction analysis
-				// and pick the optimal method (Constant, Fixed, or Verbatim).
 				Pred: frame.PredVerbatim,
 			},
-			Samples:  channelSamples,
+			Samples:  buf,
 			NSamples: e.blockSize,
 		}
 	}
@@ -127,9 +133,11 @@ func (e *FLACEncoder) Encode(samples []int32) ([]byte, error) {
 			SampleRate:        uint32(e.sampleRate),
 			Channels:          channelAssign,
 			BitsPerSample:     uint8(e.bitDepth),
+			Num:               e.frameNum,
 		},
 		Subframes: subframes,
 	}
+	e.frameNum++
 
 	e.buf.Reset()
 	if err := e.encoder.WriteFrame(f); err != nil {
diff --git a/third_party/sendspin-go/pkg/sendspin/client_dialer.go b/third_party/sendspin-go/pkg/sendspin/client_dialer.go
index 4df9664..23dc06e 100644
--- a/third_party/sendspin-go/pkg/sendspin/client_dialer.go
+++ b/third_party/sendspin-go/pkg/sendspin/client_dialer.go
@@ -128,27 +128,23 @@ func (d *clientDialer) claim(instance string) bool {
 }
 
 // release updates the instance slot after a dial attempt completes.
-// On success (dialErr == nil) the instance stays latched in the active
-// set — we never re-dial a successfully-connected instance. On error
-// it frees the slot and schedules an exponentially-backed-off cooldown
-// before the next retry.
+// The slot is always freed — dial() only returns after handleConnection
+// has returned, so the connection is already dead by the time we get
+// here, and a future mDNS re-emission should be free to redial. On
+// error it also schedules an exponentially-backed-off cooldown before
+// the next retry.
 func (d *clientDialer) release(instance string, dialErr error) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 
+	delete(d.active, instance)
+
 	if dialErr == nil {
-		// Latch: keep active[instance] = true so future discovery
-		// events for this instance are silently ignored. The server
-		// already handled the connection; re-dialing would create a
-		// duplicate session.
 		delete(d.failures, instance)
 		delete(d.cooldown, instance)
 		return
 	}
 
-	// Dial failed — release the slot so the instance can be retried
-	// after the backoff period.
-	delete(d.active, instance)
 	d.failures[instance]++
 	backoff := d.baseBackoff * time.Duration(1<<(d.failures[instance]-1))
 	if backoff > d.maxBackoff || backoff <= 0 {
diff --git a/third_party/sendspin-go/pkg/sendspin/client_dialer_test.go b/third_party/sendspin-go/pkg/sendspin/client_dialer_test.go
index 7bbc641..f68f42c 100644
--- a/third_party/sendspin-go/pkg/sendspin/client_dialer_test.go
+++ b/third_party/sendspin-go/pkg/sendspin/client_dialer_test.go
@@ -54,11 +54,17 @@ func TestClientDialerDedupesByInstance(t *testing.T) {
 	var mu sync.Mutex
 	var seen []string
 
+	// Block each dial until the test releases it, so concurrent
+	// duplicate emissions for the same instance arrive while the
+	// dial is in flight — which is what the dedupe protects against.
+	gate := make(chan struct{})
+
 	dial := func(ctx context.Context, info *discovery.ClientInfo) error {
 		atomic.AddInt32(&dialCalls, 1)
 		mu.Lock()
 		seen = append(seen, info.Instance)
 		mu.Unlock()
+		<-gate
 		return nil
 	}
 
@@ -72,39 +78,32 @@ func TestClientDialerDedupesByInstance(t *testing.T) {
 		close(done)
 	}()
 
-	// Emit the same instance 3 times and a different instance once
+	// Emit the same instance 3 times and a different instance once,
+	// all while dials are blocked in `gate`.
 	in <- &discovery.ClientInfo{Instance: "a._sendspin._tcp.local.", Host: "1.1.1.1", Port: 8928, Path: "/sendspin"}
 	in <- &discovery.ClientInfo{Instance: "a._sendspin._tcp.local.", Host: "1.1.1.1", Port: 8928, Path: "/sendspin"}
 	in <- &discovery.ClientInfo{Instance: "b._sendspin._tcp.local.", Host: "1.1.1.2", Port: 8928, Path: "/sendspin"}
 	in <- &discovery.ClientInfo{Instance: "a._sendspin._tcp.local.", Host: "1.1.1.1", Port: 8928, Path: "/sendspin"}
 
-	// Wait until at least 2 unique dials have happened
-	deadline := time.After(500 * time.Millisecond)
-	for {
-		if atomic.LoadInt32(&dialCalls) >= 2 {
-			break
-		}
-		select {
-		case <-deadline:
-			t.Fatalf("timed out waiting for dial calls, got %d", atomic.LoadInt32(&dialCalls))
-		case <-time.After(10 * time.Millisecond):
-		}
-	}
+	waitForCalls(t, &dialCalls, 2, 500*time.Millisecond)
 
-	// Give any extra (incorrect) calls a chance to fire
+	// Give any extra (incorrect) calls a chance to fire while dials
+	// are still blocked.
 	time.Sleep(50 * time.Millisecond)
 
-	cancel()
-	<-done
-
 	if got := atomic.LoadInt32(&dialCalls); got != 2 {
-		t.Errorf("dialCalls = %d, want 2 (one per unique instance)", got)
+		t.Errorf("dialCalls = %d, want 2 (one per unique instance while in flight)", got)
 	}
 	mu.Lock()
-	defer mu.Unlock()
 	if len(seen) != 2 {
+		mu.Unlock()
 		t.Fatalf("seen = %v, want 2 entries", seen)
 	}
+	mu.Unlock()
+
+	close(gate)
+	cancel()
+	<-done
 }
 
 func waitForCalls(t *testing.T, counter *int32, want int32, timeout time.Duration) {
diff --git a/third_party/sendspin-go/pkg/sendspin/server.go b/third_party/sendspin-go/pkg/sendspin/server.go
index 06fe810..088f601 100644
--- a/third_party/sendspin-go/pkg/sendspin/server.go
+++ b/third_party/sendspin-go/pkg/sendspin/server.go
@@ -96,6 +96,12 @@ type Server struct {
 	audioSource         AudioSource
 	consecutiveReadErrs int
 
+	// Accumulates bytes-on-wire across one streamAudio rate window
+	// (encoded chunk size, not raw PCM) so the periodic stats line can
+	// report both chunks/sec and kbps. Touched only from the streamAudio
+	// goroutine so no locking needed.
+	lastWindowBytes int64
+
 	mdnsManager *discovery.Manager
 
 	// server-initiated discovery dialer cancel
@@ -554,12 +560,19 @@ func (s *Server) clientWriter(c *ServerClient) {
 	for {
 		select {
 		case msg := <-c.sendChan:
+			queueDepth := len(c.sendChan)
 			switch v := msg.(type) {
 			case []byte:
 				c.conn.SetWriteDeadline(time.Now().Add(writeDeadline))
+				wStart := time.Now()
 				if err := c.conn.WriteMessage(websocket.BinaryMessage, v); err != nil {
+					log.Printf("clientWriter %s: binary write error: %v (queue %d)", c.name, err, queueDepth)
 					return
 				}
+				if wDur := time.Since(wStart); wDur > 5*time.Millisecond || queueDepth > 5 {
+					log.Printf("clientWriter %s: %s for %d bytes (queue %d)",
+						c.name, wDur.Round(time.Microsecond), len(v), queueDepth)
+				}
 			default:
 				data, err := json.Marshal(v)
 				if err != nil {
@@ -567,6 +580,7 @@ func (s *Server) clientWriter(c *ServerClient) {
 				}
 				c.conn.SetWriteDeadline(time.Now().Add(writeDeadline))
 				if err := c.conn.WriteMessage(websocket.TextMessage, data); err != nil {
+					log.Printf("clientWriter %s: text write error: %v", c.name, err)
 					return
 				}
 			}
diff --git a/third_party/sendspin-go/pkg/sendspin/server_client.go b/third_party/sendspin-go/pkg/sendspin/server_client.go
index 536416d..0577e2a 100644
--- a/third_party/sendspin-go/pkg/sendspin/server_client.go
+++ b/third_party/sendspin-go/pkg/sendspin/server_client.go
@@ -3,6 +3,7 @@
 package sendspin
 
 import (
+	"encoding/binary"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -102,13 +103,68 @@ func (c *ServerClient) Send(msgType string, payload interface{}) error {
 // enqueued frames are dropped silently. Callers should treat this as
 // best-effort once they've observed a client leaving.
 func (c *ServerClient) SendBinary(data []byte) error {
+	// Fast path: queue has room.
 	select {
 	case c.sendChan <- data:
 		return nil
 	default:
-		return fmt.Errorf("client send buffer full (depth=%d, cap=%d)",
-			len(c.sendChan), cap(c.sendChan))
 	}
+
+	// Queue full. Drop-oldest is timestamp-aware in effect: the engine
+	// pushes chunks in monotonically increasing playbackTime order, so
+	// the head of the queue has the earliest scheduled time = the chunk
+	// most likely to be stale by the time the writer catches up.
+	//
+	// We use the *new* chunk's playbackTime as our "now + BufferAheadMs"
+	// reference (the engine just stamped it that way), so we don't need
+	// an external clock to decide what's stale. The actual log includes
+	// how far past the scheduling headroom the dropped chunks were.
+	newTS, newOK := extractAudioPlaybackTime(data)
+	const bufferAheadUs = int64(BufferAheadMs) * 1000
+
+	dropped := 0
+	var maxLatenessUs int64
+	for dropped < cap(c.sendChan) {
+		select {
+		case head := <-c.sendChan:
+			if newOK {
+				if headBytes, ok := head.([]byte); ok {
+					if headTS, headOK := extractAudioPlaybackTime(headBytes); headOK {
+						// Lateness = how long past "now" this chunk's playback
+						// time was. "Now" ≈ newTS - bufferAhead.
+						late := (newTS - bufferAheadUs) - headTS
+						if late > maxLatenessUs {
+							maxLatenessUs = late
+						}
+					}
+				}
+			}
+			dropped++
+		default:
+		}
+		select {
+		case c.sendChan <- data:
+			if dropped > 0 {
+				log.Printf("client %s send pressure: dropped %d stale chunks (oldest was %s past schedule)",
+					c.name, dropped,
+					time.Duration(maxLatenessUs)*time.Microsecond)
+			}
+			return nil
+		default:
+		}
+	}
+	return fmt.Errorf("client send buffer full (depth=%d, cap=%d)",
+		len(c.sendChan), cap(c.sendChan))
+}
+
+// extractAudioPlaybackTime returns the playbackTime (in microseconds) embedded
+// in a sendspin audio chunk's 9-byte header. Returns false for non-audio frames.
+func extractAudioPlaybackTime(data []byte) (int64, bool) {
+	const audioChunkHeaderSize = 9
+	if len(data) < audioChunkHeaderSize || data[0] != AudioChunkMessageType {
+		return 0, false
+	}
+	return int64(binary.BigEndian.Uint64(data[1:audioChunkHeaderSize])), true
 }
 
 // State returns the client's current playback state ("synchronized",
diff --git a/third_party/sendspin-go/pkg/sendspin/server_stream.go b/third_party/sendspin-go/pkg/sendspin/server_stream.go
index 45d97dc..1fe1224 100644
--- a/third_party/sendspin-go/pkg/sendspin/server_stream.go
+++ b/third_party/sendspin-go/pkg/sendspin/server_stream.go
@@ -18,6 +18,8 @@ func (s *Server) streamAudio() {
 
 	tickBudget := time.Duration(ChunkDurationMs) * time.Millisecond
 	var lastTick time.Time
+	var chunkCount int
+	rateWindowStart := time.Now()
 	for {
 		select {
 		case t := <-ticker.C:
@@ -35,6 +37,21 @@ func (s *Server) streamAudio() {
 			lastTick = t
 
 			s.generateAndSendChunk()
+			chunkCount++
+
+			if elapsed := time.Since(rateWindowStart); elapsed >= 10*time.Second {
+				rate := float64(chunkCount) / elapsed.Seconds()
+				avgBytes := 0
+				if chunkCount > 0 {
+					avgBytes = int(s.lastWindowBytes) / chunkCount
+				}
+				kbps := float64(s.lastWindowBytes) * 8 / 1000.0 / elapsed.Seconds()
+				log.Printf("ENGINE STATS: %.3f chunks/sec, avg %d bytes/chunk, %.1f kbps over %s (%d chunks)",
+					rate, avgBytes, kbps, elapsed.Round(time.Millisecond), chunkCount)
+				chunkCount = 0
+				s.lastWindowBytes = 0
+				rateWindowStart = time.Now()
+			}
 		case <-s.stopChan:
 			log.Printf("Audio streaming stopping")
 			return
@@ -100,6 +117,10 @@ func (s *Server) generateAndSendChunk() {
 		tracker := c.bufferTracker
 		c.mu.RUnlock()
 
+		if codec == "" {
+			continue
+		}
+
 		switch codec {
 		case "opus":
 			if opusEncoder != nil {
@@ -139,6 +160,7 @@ func (s *Server) generateAndSendChunk() {
 		}
 
 		chunk := CreateAudioChunk(playbackTime, audioData)
+		s.lastWindowBytes += int64(len(chunk))
 
 		if tracker != nil {
 			chunkDurationUs := int64(ChunkDurationMs) * 1000