handle dropped packets and network delays

Add a per-user jitter buffer with restart resync.
Incoming audio was rendered straight to OpenAL as packets arrived, so
normal network jitter produced gaps and the renderer starved between
frames. Buffer each user's decoded frames for a configurable playout
delay (-jitter-buffer, default 40ms) and let the render thread pull
from that buffer instead.

Drop late packets, and use packet duration to set the expected frame
number.
A packet below the expected frame number is dropped rather than
stalling delivery, and the expected frame is stepped by the packet's
real duration instead of a fixed 10ms.

After detecting a large backward jump in frame numbering, along with a
sustained run of late packets, resync to the new frame numbering.
When a sender switches audio devices, Mumble destroys and recreates
their audio stream. This resets their frame number to zero without
sending a terminator packet, which causes a potentially endless hang
while we wait for a frame number that will not arrive for hours.

Own the OpenAL context on one dedicated render thread.
Contexts are current to an OS thread, so creating sources and buffers
from whichever goroutine happened to be running could operate on no
context at all. Every source and buffer is now created, filled, and
deleted on that thread, and file playback renders through it too.

Report device and capture failures with the device name.
Startup errors said only that a device could not be opened, and a
capture stall was reported as a microphone failure even though it is
normal scheduler timing. Fatal microphone errors now exit instead of
leaving a client that cannot transmit.

Release queued buffers and deactivate the context on shutdown.
Streams were torn down while buffers were still queued on a source,
and the context was destroyed while still current.

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

gumble/gumble/config.go

@@ -25,6 +25,9 @@ type Config struct {
 	AudioInterval time.Duration
 	// AudioDataBytes is the number of bytes that an audio frame can use.
 	AudioDataBytes int
+	// IncomingAudioBuffer is the amount of per-speaker audio retained before
+	// playback starts, absorbing jitter in incoming UDP packet delivery.
+	IncomingAudioBuffer time.Duration
 
 	// DisableUDP forces all audio to use the TCP tunnel instead of UDP.
 	DisableUDP bool
@@ -41,6 +44,7 @@ func NewConfig() *Config {
 		Buffers:             8,
 		AudioInterval:       AudioDefaultInterval,
 		AudioDataBytes:      AudioDefaultDataBytes,
+		IncomingAudioBuffer: 40 * time.Millisecond,
 	}
 }
 
@@ -54,6 +58,9 @@ func (c *Config) Validate() error {
 	if c.AudioDataBytes <= 0 {
 		return fmt.Errorf("gumble: AudioDataBytes must be positive")
 	}
+	if c.IncomingAudioBuffer < 0 {
+		return fmt.Errorf("gumble: IncomingAudioBuffer must not be negative")
+	}
 	if c.Buffers <= 0 {
 		return fmt.Errorf("gumble: Buffers must be positive")
 	}

gumble/gumbleopenal/stream.go

@@ -3,16 +3,35 @@ package gumbleopenal
 import (
 	"encoding/binary"
 	"errors"
-	"os/exec"
+	"fmt"
+	"math"
+	"runtime"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"git.stormux.org/storm/barnard/audio"
 	"git.stormux.org/storm/barnard/gumble/go-openal/openal"
 	"git.stormux.org/storm/barnard/gumble/gumble"
+	"git.stormux.org/storm/barnard/log"
 	"git.stormux.org/storm/barnard/noise"
 )
 
+func deviceName(name string) string {
+	if name == "" {
+		return "default"
+	}
+	return name
+}
+
+func openInputDeviceError(name string, format openal.Format) error {
+	return fmt.Errorf("%w: could not open capture device %q (format=%v, rate=%d)", ErrInputDevice, deviceName(name), format, gumble.AudioSampleRate)
+}
+
+func openOutputDeviceError(name string) error {
+	return fmt.Errorf("%w: could not open playback device %q", ErrOutputDevice, deviceName(name))
+}
+
 // NoiseProcessor interface for noise suppression
 type NoiseProcessor interface {
 	ProcessSamples(samples []int16)
@@ -33,8 +52,48 @@ const recorderOutgoingSource uint32 = ^uint32(0)
 
 const (
 	maxBufferSize    = 11520 // Max frame size (2880) * bytes per stereo sample (4)
+	jitterMaxPackets = 50
+	// Mumble destroys and recreates AudioInput when the sender switches audio
+	// devices, which restarts its frame numbering at zero. The destructor
+	// sends no terminator, so a sender that never unkeys leaves us expecting a
+	// frame number the new stream will not reach for hours: every packet looks
+	// permanently late and gets discarded. Detect that and resync.
+	//
+	// Two conditions must hold together. A sustained run of late packets
+	// distinguishes a restarted stream from a clump of reordered packets,
+	// which is bounded and then recovers on its own. The backwards jump must
+	// also be too large to be network reordering; a smaller jump needs no
+	// intervention because the restarted stream climbs back past the stale
+	// expectation within jitterResyncJump frames anyway.
+	jitterLateResync = 5
+	// Frame numbers are Mumble timestamps in 10 ms units, so this is 1 second
+	// — far beyond any real reordering window.
+	jitterResyncJump = 100
 )
 
+// jitterShouldResync reports whether the sender restarted its frame numbering
+// rather than merely delivering a few packets out of order. lateRun is the
+// number of consecutive late packets and backJump is how far the current
+// packet sits below the expected sequence.
+func jitterShouldResync(lateRun int, backJump int64) bool {
+	return lateRun >= jitterLateResync && backJump >= jitterResyncJump
+}
+
+// jitterPlaybackReady holds the requested initial playout delay only once.
+// Requiring the delay on every packet drains and refills the renderer in bursts.
+func jitterPlaybackReady(started bool, buffered, target time.Duration) bool {
+	return started || buffered >= target
+}
+
+func audioPacketDuration(packet *gumble.AudioPacket) time.Duration {
+	if packet == nil || len(packet.AudioBuffer) == 0 {
+		return 0
+	}
+	// Opus decoders deliver interleaved stereo PCM to this renderer.
+	frames := len(packet.AudioBuffer) / gumble.AudioChannels
+	return time.Duration(frames) * time.Second / gumble.AudioSampleRate
+}
+
 var (
 	ErrState        = errors.New("gumbleopenal: invalid state")
 	ErrMic          = errors.New("gumbleopenal: microphone disconnected or misconfigured")
@@ -42,14 +101,9 @@ var (
 	ErrOutputDevice = errors.New("gumbleopenal: invalid output device or parameters")
 )
 
-func beep() {
-	cmd := exec.Command("beep")
-	cmdout, err := cmd.Output()
-	if err != nil {
-		panic(err)
-	}
-	if cmdout != nil {
-	}
+type renderCommand struct {
+	fn   func()
+	done chan struct{}
 }
 
 type Stream struct {
@@ -57,21 +111,32 @@ type Stream struct {
 	link   gumble.Detacher
 
 	deviceSource     *openal.CaptureDevice
+	inputDeviceName  string
+	outputDeviceName string
 	sourceFormat     openal.Format
 	sourceChannels   int
 	sourceFrameSize  int
-	micVolume       float32
+	micVolume        atomic.Uint32 // float32 stored as bits
+	sourceMu         sync.Mutex
 	sourceStop       chan bool
+	sourceDone       chan struct{}
 
 	deviceSink   *openal.Device
 	contextSink  *openal.Context
+	renderMu     sync.RWMutex
+	renderCh     chan renderCommand
+	renderDone   chan struct{}
+	renderClosed bool
 
 	noiseProcessor      NoiseProcessor
 	noiseProcessorRight NoiseProcessor
 	micAGC              *audio.AGC
 	micAGCRight         *audio.AGC
 	filePlayer          FilePlayer
+	localSource         *openal.Source
+	localBuffers        openal.Buffers
 	recorderMu          sync.RWMutex
+	errorFunc           func(error) // called on capture errors
 	recorder            Recorder
 }
 
@@ -81,22 +146,47 @@ func New(client *gumble.Client, inputDevice *string, outputDevice *string, test
 		frmsz = client.Config.AudioFrameSize()
 	}
 
+	devName := ""
+	if inputDevice != nil {
+		devName = *inputDevice
+	}
+	log.Info("OpenAL capture: requested device=%q rate=%d frameSize=%d", devName, gumble.AudioSampleRate, frmsz)
+
 	inputFormat := openal.FormatStereo16
 	sourceChannels := 2
-	idev := openal.CaptureOpenDevice(*inputDevice, gumble.AudioSampleRate, inputFormat, uint32(frmsz))
+	// Keep several frames in the capture ring so normal scheduler jitter does
+	// not overflow a PipeWire/Pulse capture stream.
+	captureBufferSize := uint32(frmsz * 4)
+	idev := openal.CaptureOpenDevice(devName, gumble.AudioSampleRate, inputFormat, captureBufferSize)
 	if idev == nil {
+		log.Info("OpenAL capture: stereo failed, trying mono")
 		inputFormat = openal.FormatMono16
 		sourceChannels = 1
-		idev = openal.CaptureOpenDevice(*inputDevice, gumble.AudioSampleRate, inputFormat, uint32(frmsz))
+		idev = openal.CaptureOpenDevice(devName, gumble.AudioSampleRate, inputFormat, captureBufferSize)
 	}
 	if idev == nil {
-		return nil, ErrInputDevice
+		log.Error("OpenAL capture: failed to open device %q", devName)
+		return nil, openInputDeviceError(devName, inputFormat)
 	}
+	if err := idev.Err(); err != nil {
+		idev.CaptureCloseDevice()
+		return nil, fmt.Errorf("%w: capture device %q: %v", ErrInputDevice, deviceName(devName), err)
+	}
+	log.Info("OpenAL capture: opened device %q format=%v channels=%d", devName, inputFormat, sourceChannels)
 
-	odev := openal.OpenDevice(*outputDevice)
+	outName := ""
+	if outputDevice != nil {
+		outName = *outputDevice
+	}
+	odev := openal.OpenDevice(outName)
 	if odev == nil {
 		idev.CaptureCloseDevice()
-		return nil, ErrOutputDevice
+		return nil, openOutputDeviceError(outName)
+	}
+	if err := odev.Err(); err != nil {
+		idev.CaptureCloseDevice()
+		odev.CloseDevice()
+		return nil, fmt.Errorf("%w: playback device %q: %v", ErrOutputDevice, deviceName(outName), err)
 	}
 
 	if test {
@@ -107,35 +197,88 @@ func New(client *gumble.Client, inputDevice *string, outputDevice *string, test
 
 	s := &Stream{
 		client:           client,
+		inputDeviceName:  devName,
+		outputDeviceName: outName,
 		sourceFormat:     inputFormat,
 		sourceChannels:   sourceChannels,
 		sourceFrameSize:  frmsz,
-		micVolume:       1.0,
 		micAGC:           audio.NewAGC(), // Always enable AGC for outgoing mic
 	}
+	s.micVolume.Store(math.Float32bits(1.0))
 	if sourceChannels == 2 {
 		s.micAGCRight = audio.NewAGC()
 	}
 
 	s.deviceSource = idev
 	if s.deviceSource == nil {
-		return nil, ErrInputDevice
+		return nil, fmt.Errorf("%w: capture device %q is unavailable", ErrInputDevice, deviceName(devName))
 	}
 
 	s.deviceSink = odev
 	if s.deviceSink == nil {
-		return nil, ErrOutputDevice
+		return nil, fmt.Errorf("%w: playback device %q is unavailable", ErrOutputDevice, deviceName(outName))
 	}
 	s.contextSink = s.deviceSink.CreateContext()
 	if s.contextSink == nil {
+		err := s.deviceSink.Err()
 		s.Destroy()
-		return nil, ErrOutputDevice
+		if err != nil {
+			return nil, fmt.Errorf("%w: creating context for playback device %q: %v", ErrOutputDevice, deviceName(outName), err)
 		}
-	s.contextSink.Activate()
+		return nil, fmt.Errorf("%w: could not create context for playback device %q", ErrOutputDevice, deviceName(outName))
+	}
+	// OpenAL contexts are current to an OS thread. Move ownership to one
+	// dedicated render thread before any source or buffer is created.
+	openal.NullContext.Activate()
+	s.startRenderer()
+
+	// Log OpenAL device info on the render thread
+	s.render(func() {
+		log.Info("OpenAL playback: vendor=%q version=%q renderer=%q",
+			openal.GetString(0xB001),
+			openal.GetString(0xB002),
+			openal.GetString(0xB003))
+	})
 
 	return s, nil
 }
 
+func (s *Stream) startRenderer() {
+	s.renderMu.Lock()
+	s.renderClosed = false
+	s.renderCh = make(chan renderCommand)
+	s.renderDone = make(chan struct{})
+	s.renderMu.Unlock()
+	ready := make(chan struct{})
+	go func() {
+		runtime.LockOSThread()
+		defer runtime.UnlockOSThread()
+		s.contextSink.Activate()
+		close(ready)
+		defer close(s.renderDone)
+		for command := range s.renderCh {
+			command.fn()
+			close(command.done)
+		}
+		openal.NullContext.Activate()
+	}()
+	<-ready
+}
+
+// render executes fn on the sole OS thread that owns the OpenAL context. It
+// returns false after shutdown instead of sending to a closed renderer channel.
+func (s *Stream) render(fn func()) bool {
+	s.renderMu.RLock()
+	defer s.renderMu.RUnlock()
+	if s.renderClosed || s.renderCh == nil {
+		return false
+	}
+	command := renderCommand{fn: fn, done: make(chan struct{})}
+	s.renderCh <- command
+	<-command.done
+	return true
+}
+
 func (s *Stream) AttachStream(client *gumble.Client) {
 	s.link = client.Config.AttachAudio(s)
 }
@@ -145,14 +288,99 @@ func (s *Stream) SetNoiseProcessor(np NoiseProcessor) {
 	s.noiseProcessorRight = cloneNoiseProcessor(np)
 }
 
+// SetAGCEnabled turns microphone automatic gain control on or off. The AGC
+// objects themselves are created up front, so this only flips their flag and is
+// safe to call while capture is running.
+func (s *Stream) SetAGCEnabled(enabled bool) {
+	if s.micAGC != nil {
+		s.micAGC.SetEnabled(enabled)
+	}
+	if s.micAGCRight != nil {
+		s.micAGCRight.SetEnabled(enabled)
+	}
+}
+
+// IsAGCEnabled reports whether microphone automatic gain control is active.
+func (s *Stream) IsAGCEnabled() bool {
+	return s.micAGC != nil && s.micAGC.IsEnabled()
+}
+
 func (s *Stream) SetFilePlayer(fp FilePlayer) {
 	s.filePlayer = fp
+	if player, ok := fp.(interface{ SetLocalPlayback(func([]byte)) }); ok {
+		player.SetLocalPlayback(s.playLocalAudio)
+	}
+}
+
+func (s *Stream) playLocalAudio(data []byte) {
+	s.render(func() {
+		if data == nil {
+			if s.localSource != nil {
+				s.localSource.Stop()
+				queued := s.localSource.BuffersQueued()
+				if queued > 0 {
+					buffers := make(openal.Buffers, queued)
+					s.localSource.UnqueueBuffers(buffers)
+					s.localBuffers = append(s.localBuffers, buffers...)
+				}
+				s.localSource.Delete()
+				s.localSource = nil
+			}
+			if len(s.localBuffers) > 0 {
+				s.localBuffers.Delete()
+				s.localBuffers = nil
+			}
+			return
+		}
+		if s.localSource == nil {
+			source := openal.NewSource()
+			source.SetGain(1)
+			s.localSource = &source
+			s.localBuffers = openal.NewBuffers(64)
+		}
+		if n := s.localSource.BuffersProcessed(); n > 0 {
+			buffers := make(openal.Buffers, n)
+			s.localSource.UnqueueBuffers(buffers)
+			s.localBuffers = append(s.localBuffers, buffers...)
+		}
+		if len(s.localBuffers) == 0 {
+			return
+		}
+		last := len(s.localBuffers) - 1
+		buffer := s.localBuffers[last]
+		s.localBuffers = s.localBuffers[:last]
+		buffer.SetData(openal.FormatStereo16, data, gumble.AudioSampleRate)
+		s.localSource.QueueBuffer(buffer)
+		if s.localSource.State() != openal.Playing {
+			s.localSource.Play()
+		}
+	})
 }
 
 func (s *Stream) GetFilePlayer() FilePlayer {
 	return s.filePlayer
 }
 
+// UpdateUserGain applies a user's current mute and volume state on the
+// renderer thread.
+func (s *Stream) UpdateUserGain(user *gumble.User) {
+	s.render(func() {
+		if source := user.AudioSource(); source != nil {
+			if user.LocallyMuted() {
+				source.SetGain(0)
+			} else {
+				source.SetGain(user.Volume())
+			}
+		}
+	})
+}
+
+// SetErrorFunc sets a callback that is invoked when the microphone
+// capture device fails to provide audio data.
+func (s *Stream) SetErrorFunc(f func(error)) {
+	s.errorFunc = f
+}
+
 func (s *Stream) SetRecorder(recorder Recorder) {
 	s.recorderMu.Lock()
 	defer s.recorderMu.Unlock()
@@ -175,41 +403,69 @@ func (s *Stream) Destroy() {
 		s.deviceSource = nil
 	}
 	if s.deviceSink != nil {
+		if s.contextSink != nil {
+			s.render(func() {
+				openal.NullContext.Activate()
 				s.contextSink.Destroy()
-		s.deviceSink.CloseDevice()
+			})
+			s.renderMu.Lock()
+			if !s.renderClosed {
+				close(s.renderCh)
+				s.renderClosed = true
+			}
+			s.renderMu.Unlock()
+			<-s.renderDone
 			s.contextSink = nil
+		}
+		s.deviceSink.CloseDevice()
 		s.deviceSink = nil
 	}
 }
 
 func (s *Stream) StartSource(inputDevice *string) error {
+	s.sourceMu.Lock()
+	defer s.sourceMu.Unlock()
 	if s.sourceStop != nil {
 		return ErrState
 	}
 	if s.deviceSource == nil {
-		return ErrMic
+		return fmt.Errorf("%w: capture device %q is unavailable", ErrMic, deviceName(s.inputDeviceName))
 	}
 	s.deviceSource.CaptureStart()
-	s.sourceStop = make(chan bool)
-	go s.sourceRoutine(inputDevice)
+	if err := s.deviceSource.Err(); err != nil {
+		return fmt.Errorf("%w: starting capture device %q: %v", ErrMic, deviceName(s.inputDeviceName), err)
+	}
+	stop := make(chan bool)
+	done := make(chan struct{})
+	s.sourceStop, s.sourceDone = stop, done
+	go s.sourceRoutine(inputDevice, stop, done)
 	return nil
 }
 
 func (s *Stream) StopSource() error {
-	if s.deviceSource == nil {
-		return ErrMic
-	}
-	s.deviceSource.CaptureStop()
+	s.sourceMu.Lock()
 	if s.sourceStop == nil {
+		s.sourceMu.Unlock()
 		return ErrState
 	}
-	close(s.sourceStop)
-	s.sourceStop = nil
+	stop, done := s.sourceStop, s.sourceDone
+	s.sourceStop, s.sourceDone = nil, nil
+	close(stop)
+	s.sourceMu.Unlock()
+	// The routine owns capture access; wait for it before closing/reusing it.
+	<-done
+	if s.deviceSource == nil {
+		return fmt.Errorf("%w: capture device %q is unavailable", ErrMic, deviceName(s.inputDeviceName))
+	}
+	s.deviceSource.CaptureStop()
+	if err := s.deviceSource.Err(); err != nil {
+		return fmt.Errorf("%w: stopping capture device %q: %v", ErrMic, deviceName(s.inputDeviceName), err)
+	}
 	return nil
 }
 
 func (s *Stream) GetMicVolume() float32 {
-	return s.micVolume
+	return math.Float32frombits(s.micVolume.Load())
 }
 
 func (s *Stream) SetMicVolume(change float32, relative bool) {
@@ -225,50 +481,273 @@ func (s *Stream) SetMicVolume(change float32, relative bool) {
 	if val <= 0 {
 		val = 0
 	}
-	s.micVolume = val
+	s.micVolume.Store(math.Float32bits(val))
 }
 
 func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 	go func(e *gumble.AudioStreamEvent) {
-		var source = openal.NewSource()
+		log.Info("audio stream started for user %s", e.User.Name)
+		var source openal.Source
+		var emptyBufs openal.Buffers
+		var raw [maxBufferSize]byte
+		s.render(func() {
+			source = openal.NewSource()
 			e.User.SetAudioSource(&source)
-
-		// Set initial gain based on volume and mute state
 			if e.User.LocallyMuted() {
 				source.SetGain(0)
 			} else {
 				source.SetGain(e.User.Volume())
 			}
-
 			bufferCount := e.Client.Config.Buffers
 			if bufferCount < 64 {
 				bufferCount = 64
 			}
-		emptyBufs := openal.NewBuffers(bufferCount)
+			log.Info("OnAudioStream: creating %d buffers for %s (volume=%.2f gain=%.2f)",
+				bufferCount, e.User.Name, e.User.Volume(), source.GetGain())
+			emptyBufs = openal.NewBuffers(bufferCount)
+		})
 
+		var reclaimLogCounter int
 		reclaim := func() {
-			if n := source.BuffersProcessed(); n > 0 {
-				reclaimedBufs := make(openal.Buffers, n)
+			s.render(func() {
+				processed := source.BuffersProcessed()
+				queued := source.BuffersQueued()
+				srcState := source.State()
+				if processed > 0 {
+					reclaimedBufs := make(openal.Buffers, processed)
 					source.UnqueueBuffers(reclaimedBufs)
 					emptyBufs = append(emptyBufs, reclaimedBufs...)
 				}
+				reclaimLogCounter++
+				// Log every 50th reclaim, or if state is not Playing
+				if reclaimLogCounter%50 == 1 || srcState != openal.Playing {
+					log.Debug("reclaim #%d: state=%s processed=%d queued=%d empty=%d",
+						reclaimLogCounter, srcState, processed, queued, len(emptyBufs))
+				}
+				if oe := openal.Err(); oe != nil {
+					log.Error("reclaim: OpenAL error: %v", oe)
+				}
+			})
 		}
 
-		var raw [maxBufferSize]byte
+		// Jitter buffer: collects incoming packets, reorders by
+		// sequence number, and releases them after a small initial delay.
+		var jitterBuf []*gumble.AudioPacket
+		var jitterDuration time.Duration
+		var jitterNextSeq int64
+		var jitterInit, jitterStarted bool
+		var jitterLateRun int
+		var jitterDrainLogCounter, jitterAnomalyLogCounter int
+		resetJitter := func() {
+			jitterBuf = nil
+			jitterDuration = 0
+			jitterNextSeq = 0
+			jitterInit = false
+			jitterStarted = false
+			jitterLateRun = 0
+		}
+
+		// insertSorted inserts a packet into the jitter buffer sorted
+		// by sequence number.
+		insertSorted := func(p *gumble.AudioPacket) {
+			// Drop if we already have too many (protect against memory bloat)
+			if len(jitterBuf) >= jitterMaxPackets {
+				return
+			}
+			// Find insertion point (ascending sequence order)
+			i := 0
+			for i < len(jitterBuf) && jitterBuf[i].Sequence < p.Sequence {
+				i++
+			}
+			// Don't insert duplicates
+			if i < len(jitterBuf) && jitterBuf[i].Sequence == p.Sequence {
+				return
+			}
+			jitterBuf = append(jitterBuf, nil)
+			copy(jitterBuf[i+1:], jitterBuf[i:])
+			jitterBuf[i] = p
+			jitterDuration += audioPacketDuration(p)
+		}
+
+		// popNext removes and returns the packet with the expected next
+		// sequence number, or nil if not yet available.
+		popNext := func() *gumble.AudioPacket {
+			if len(jitterBuf) == 0 || jitterBuf[0].Sequence != jitterNextSeq {
+				return nil
+			}
+			p := jitterBuf[0]
+			jitterBuf = jitterBuf[1:]
+			jitterDuration -= audioPacketDuration(p)
+			// Frame numbers are Mumble timestamps in 10 ms units.
+			// Compute the actual step from the PCM sample count so we
+			// never skip a legitimate gap.
+			samples := len(p.AudioBuffer)
+			if samples > gumble.AudioDefaultFrameSize && samples%2 == 0 {
+				// Stereo: step = stereo frames / base frame size
+				step := int64((samples / 2) / gumble.AudioDefaultFrameSize)
+				if step >= 1 {
+					jitterNextSeq = p.Sequence + step
+				} else {
+					jitterNextSeq = p.Sequence + 1
+				}
+			} else {
+				step := int64(samples / gumble.AudioDefaultFrameSize)
+				if step >= 1 {
+					jitterNextSeq = p.Sequence + step
+				} else {
+					jitterNextSeq = p.Sequence + 1
+				}
+			}
+			return p
+		}
 
 		for packet := range e.C {
+			// A talk burst may restart its frame numbers from zero. Reset before
+			// testing local mute so an unmute cannot retain the previous burst's
+			// timestamp and discard the new burst as permanently late.
+			if packet.Terminator {
+				resetJitter()
+				continue
+			}
+
 			// Skip processing if user is locally muted
 			if e.User.LocallyMuted() {
 				continue
 			}
 
-			var boost uint16 = uint16(1)
-			samples := len(packet.AudioBuffer)
-			if samples > cap(raw)/2 {
-				continue
+			// Insert into jitter buffer
+			insertSorted(packet)
+
+			// Initialize the expected sequence on first packet
+			if !jitterInit {
+				jitterNextSeq = jitterBuf[0].Sequence
+				jitterInit = true
 			}
 
-			boost = e.User.Boost()
+			// Hold only the initial packets. Once playback starts, drain every
+			// ready packet so the renderer is fed continuously rather than in
+			// bursts of packets.
+			if !jitterPlaybackReady(jitterStarted, jitterDuration, e.Client.Config.IncomingAudioBuffer) {
+				continue
+			}
+			jitterStarted = true
+
+			// Drain all packets that are ready (in sequence order)
+			for {
+				pkt := popNext()
+				if pkt == nil {
+					if len(jitterBuf) > 0 {
+						if jitterBuf[0].Sequence < jitterNextSeq {
+							jitterLateRun++
+							if jitterShouldResync(jitterLateRun, jitterNextSeq-jitterBuf[0].Sequence) {
+								// The sender restarted its frame numbering
+								// mid-burst. Follow it instead of discarding
+								// every remaining packet until it unkeys.
+								log.Debug("jitter: sequence restart for %s, resyncing from %d to %d",
+									e.User.Name, jitterNextSeq, jitterBuf[0].Sequence)
+								jitterNextSeq = jitterBuf[0].Sequence
+								jitterLateRun = 0
+								continue
+							}
+							// Late or duplicate: discard so it doesn't
+							// permanently block the drain loop.
+							jitterAnomalyLogCounter++
+							if jitterAnomalyLogCounter <= 3 || jitterAnomalyLogCounter%1000 == 0 {
+								log.Debug("jitter: discarding late seq=%d for %s (next=%d buf=%d)",
+									jitterBuf[0].Sequence, e.User.Name, jitterNextSeq, len(jitterBuf))
+							}
+							jitterDuration -= audioPacketDuration(jitterBuf[0])
+							jitterBuf = jitterBuf[1:]
+							continue
+						}
+						if jitterBuf[0].Sequence > jitterNextSeq {
+							// Gap in sequence: skip ahead so we don't
+							// wait forever for a lost packet.
+							jitterAnomalyLogCounter++
+							if jitterAnomalyLogCounter <= 3 || jitterAnomalyLogCounter%1000 == 0 {
+								log.Debug("jitter: seq gap for %s, skipping from %d to %d (buf=%d)",
+									e.User.Name, jitterNextSeq, jitterBuf[0].Sequence, len(jitterBuf))
+							}
+							jitterNextSeq = jitterBuf[0].Sequence
+							continue
+						}
+						// Sequence == jitterNextSeq but popNext returned nil?
+						// Shouldn't happen; break to avoid infinite loop.
+					}
+					break
+				}
+				jitterLateRun = 0
+				jitterDrainLogCounter++
+				if jitterDrainLogCounter <= 3 || jitterDrainLogCounter%1000 == 0 {
+					log.Debug("jitter: draining seq=%d for %s (buf=%d emptyBufs=%d)",
+						pkt.Sequence, e.User.Name, len(jitterBuf), len(emptyBufs))
+				}
+				reclaim()
+				s.render(func() {
+					emptyBufs = s.processAudioPacket(pkt, e.User, &source, emptyBufs, &raw)
+				})
+			}
+		}
+
+		// Drain remaining buffered packets on stream close
+		for len(jitterBuf) > 0 {
+			pkt := popNext()
+			if pkt == nil {
+				// Gap in sequence at end; skip
+				jitterNextSeq = jitterBuf[0].Sequence
+				pkt = popNext()
+			}
+			if pkt != nil {
+				reclaim()
+				s.render(func() {
+					emptyBufs = s.processAudioPacket(pkt, e.User, &source, emptyBufs, &raw)
+				})
+			}
+		}
+		reclaim()
+		s.render(func() {
+			// OpenAL does not delete buffers when a source is deleted. Reclaim
+			// queued buffers after stopping so every generated buffer is freed.
+			source.Stop()
+			if n := source.BuffersQueued(); n > 0 {
+				queuedBufs := make(openal.Buffers, n)
+				source.UnqueueBuffers(queuedBufs)
+				emptyBufs = append(emptyBufs, queuedBufs...)
+			}
+			source.Delete()
+			emptyBufs.Delete()
+			e.User.SetAudioSource(nil)
+		})
+		log.Debug("audio stream ended for user %s", e.User.Name)
+	}(e)
+}
+
+func applyVolumeAdjustment(sample int16, adjustment float32) int16 {
+	if adjustment == 0 || adjustment == 1 {
+		return sample
+	}
+	adjusted := float32(sample) * adjustment
+	if adjusted > 32767 {
+		return 32767
+	}
+	if adjusted < -32768 {
+		return -32768
+	}
+	return int16(adjusted)
+}
+
+// processAudioPacket decodes and queues a single audio packet for playback.
+// Returns the updated emptyBufs slice after consuming a buffer.
+// The caller must call reclaim() before invoking this to ensure buffers
+// are available.
+func (s *Stream) processAudioPacket(packet *gumble.AudioPacket, user *gumble.User, source *openal.Source, emptyBufs openal.Buffers, raw *[maxBufferSize]byte) openal.Buffers {
+	samples := len(packet.AudioBuffer)
+	if samples > cap(*raw)/2 {
+		return emptyBufs
+	}
+
+	boost := user.Boost()
+	userVolume := user.Volume()
 	recorder := s.getRecorder()
 	var recordBuffer []int16
 	recordPtr := 0
@@ -289,7 +768,7 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 		// Process stereo samples as pairs
 		for i := 0; i < samples*2; i += 2 {
 			// Process left channel with saturation protection
-					sample := packet.AudioBuffer[i]
+			sample := applyVolumeAdjustment(packet.AudioBuffer[i], packet.VolumeAdjustment)
 			if boost > 1 {
 				boosted := int32(sample) * int32(boost)
 				if boosted > 32767 {
@@ -301,14 +780,14 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 				}
 			}
 			if recorder != nil {
-						recordBuffer[recordPtr] = scaleForRecording(sample, e.User.Volume())
+				recordBuffer[recordPtr] = scaleForRecording(sample, userVolume)
 				recordPtr++
 			}
-					binary.LittleEndian.PutUint16(raw[rawPtr:], uint16(sample))
+			binary.LittleEndian.PutUint16((*raw)[rawPtr:], uint16(sample))
 			rawPtr += 2
 
 			// Process right channel with saturation protection
-					sample = packet.AudioBuffer[i+1]
+			sample = applyVolumeAdjustment(packet.AudioBuffer[i+1], packet.VolumeAdjustment)
 			if boost > 1 {
 				boosted := int32(sample) * int32(boost)
 				if boosted > 32767 {
@@ -320,16 +799,16 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 				}
 			}
 			if recorder != nil {
-						recordBuffer[recordPtr] = scaleForRecording(sample, e.User.Volume())
+				recordBuffer[recordPtr] = scaleForRecording(sample, userVolume)
 				recordPtr++
 			}
-					binary.LittleEndian.PutUint16(raw[rawPtr:], uint16(sample))
+			binary.LittleEndian.PutUint16((*raw)[rawPtr:], uint16(sample))
 			rawPtr += 2
 		}
 	} else {
 		// Process mono samples with saturation protection
 		for i := 0; i < samples; i++ {
-					sample := packet.AudioBuffer[i]
+			sample := applyVolumeAdjustment(packet.AudioBuffer[i], packet.VolumeAdjustment)
 			if boost > 1 {
 				boosted := int32(sample) * int32(boost)
 				if boosted > 32767 {
@@ -341,67 +820,90 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 				}
 			}
 			if recorder != nil {
-						recordSample := scaleForRecording(sample, e.User.Volume())
+				recordSample := scaleForRecording(sample, userVolume)
 				recordBuffer[recordPtr] = recordSample
 				recordBuffer[recordPtr+1] = recordSample
 				recordPtr += 2
 			}
-					binary.LittleEndian.PutUint16(raw[rawPtr:], uint16(sample))
+			binary.LittleEndian.PutUint16((*raw)[rawPtr:], uint16(sample))
 			rawPtr += 2
 		}
 	}
 	if recorder != nil && recordPtr > 0 {
-				recorder.RecordAudioFrame(e.User.Session, recordBuffer[:recordPtr], true)
+		recorder.RecordAudioFrame(user.Session, recordBuffer[:recordPtr], true)
 	}
 
-			reclaim()
 	if len(emptyBufs) == 0 {
-				continue
+		log.Warn("processAudioPacket: NO EMPTY BUFFERS for %s seq=%d — audio packet dropped!", user.Name, packet.Sequence)
+		return emptyBufs
 	}
 
 	last := len(emptyBufs) - 1
 	buffer := emptyBufs[last]
+	emptyBufs[last] = 0
 	emptyBufs = emptyBufs[:last]
 
-			buffer.SetData(format, raw[:rawPtr], gumble.AudioSampleRate)
+	buffer.SetData(format, (*raw)[:rawPtr], gumble.AudioSampleRate)
+	if oe := openal.Err(); oe != nil {
+		log.Error("processAudioPacket: Buffer.SetData error for %s seq=%d: %v", user.Name, packet.Sequence, oe)
+	}
 	source.QueueBuffer(buffer)
+	if oe := openal.Err(); oe != nil {
+		log.Error("processAudioPacket: QueueBuffer error for %s seq=%d: %v", user.Name, packet.Sequence, oe)
+	}
 
-			if source.State() != openal.Playing {
+	srcState := source.State()
+	if srcState != openal.Playing {
+		log.Debug("processAudioPacket: source state=%s (not playing), calling Play() for %s seq=%d bufs=%d", srcState, user.Name, packet.Sequence, len(emptyBufs))
 		source.Play()
+		if oe := openal.Err(); oe != nil {
+			log.Error("processAudioPacket: Source.Play error for %s seq=%d: %v", user.Name, packet.Sequence, oe)
 		}
+		log.Debug("processAudioPacket: after Play(), state=%s for %s seq=%d", source.State(), user.Name, packet.Sequence)
 	}
-		reclaim()
-		emptyBufs.Delete()
-		source.Delete()
-	}(e)
+	return emptyBufs
 }
 
-func (s *Stream) sourceRoutine(inputDevice *string) {
+func (s *Stream) sourceRoutine(inputDevice *string, stop chan bool, done chan struct{}) {
+	defer close(done)
+	log.Info("source routine started: interval=%v frameSize=%d channels=%d",
+		s.client.Config.AudioInterval, s.client.Config.AudioFrameSize(), s.sourceChannels)
 	interval := s.client.Config.AudioInterval
 	frameSize := s.client.Config.AudioFrameSize()
 
+	devName := ""
+	if inputDevice != nil {
+		devName = *inputDevice
+	}
+
+	reopened := false
 	if frameSize != s.sourceFrameSize {
 		s.deviceSource.CaptureCloseDevice()
+		reopened = true
 		s.sourceFrameSize = frameSize
-		s.deviceSource = openal.CaptureOpenDevice(*inputDevice, gumble.AudioSampleRate, s.sourceFormat, uint32(s.sourceFrameSize))
+		captureBufferSize := uint32(s.sourceFrameSize * 4)
+		s.deviceSource = openal.CaptureOpenDevice(devName, gumble.AudioSampleRate, s.sourceFormat, captureBufferSize)
 		if s.deviceSource == nil && s.sourceFormat == openal.FormatStereo16 {
 			s.sourceFormat = openal.FormatMono16
 			s.sourceChannels = 1
-			s.deviceSource = openal.CaptureOpenDevice(*inputDevice, gumble.AudioSampleRate, s.sourceFormat, uint32(s.sourceFrameSize))
+			s.deviceSource = openal.CaptureOpenDevice(devName, gumble.AudioSampleRate, s.sourceFormat, captureBufferSize)
 		}
 	}
 	if s.deviceSource == nil {
 		return
 	}
+	// Reopening after an interval change creates a stopped capture device.
+	if reopened {
+		s.deviceSource.CaptureStart()
+	}
 
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 
-	stop := s.sourceStop
-
 	outgoing := s.client.AudioOutgoing()
 	defer close(outgoing)
 
+	var micFailed bool
 	for {
 		select {
 		case <-stop:
@@ -410,15 +912,28 @@ func (s *Stream) sourceRoutine(inputDevice *string) {
 			sampleCount := frameSize * s.sourceChannels
 			int16Buffer := make([]int16, sampleCount)
 
-			// Capture microphone if available
+			// alcCaptureSamples requires the requested frames to already be
+			// available. PipeWire and PulseAudio do not guarantee that a Go
+			// ticker fires precisely on a capture-frame boundary.
 			hasMicInput := false
-			buff := s.deviceSource.CaptureSamples(uint32(frameSize))
+			available := s.deviceSource.CapturedSamples()
+			var buff []byte
+			if available >= uint32(frameSize) {
+				buff = s.deviceSource.CaptureSamples(uint32(frameSize))
+			}
 			if len(buff) == sampleCount*2 {
 				hasMicInput = true
+				if micFailed {
+					micFailed = false
+					if s.errorFunc != nil {
+						s.errorFunc(nil) // nil signals recovery
+					}
+				}
 				for i := 0; i < sampleCount; i++ {
 					sample := int16(binary.LittleEndian.Uint16(buff[i*2:]))
-					if s.micVolume != 1.0 {
-						sample = int16(float32(sample) * s.micVolume)
+					vol := s.GetMicVolume()
+					if vol != 1.0 {
+						sample = int16(float32(sample) * vol)
 					}
 					int16Buffer[i] = sample
 				}
@@ -428,6 +943,11 @@ func (s *Stream) sourceRoutine(inputDevice *string) {
 				} else {
 					s.processStereoSamples(int16Buffer, frameSize)
 				}
+			} else if available >= uint32(frameSize) && !micFailed {
+				micFailed = true
+				if s.errorFunc != nil {
+					s.errorFunc(ErrMic)
+				}
 			}
 
 			// Mix with or use file audio if playing
@@ -503,10 +1023,21 @@ func (s *Stream) sourceRoutine(inputDevice *string) {
 					recorder.RecordAudioFrame(recorderOutgoingSource, outputBuffer, true)
 				}
 			} else if hasMicInput {
-				// Send mic when no file is playing
-				outgoing <- gumble.AudioBuffer(int16Buffer)
+				// Send mic when no file is playing. If the microphone is
+				// stereo, downmix to mono since Mumble voice transmission
+				// uses mono Opus encoding.
+				outBuf := int16Buffer
+				if s.sourceChannels == 2 {
+					monoBuf := make([]int16, frameSize)
+					for i := 0; i < frameSize; i++ {
+						// Average left and right channels
+						monoBuf[i] = int16((int32(int16Buffer[i*2]) + int32(int16Buffer[i*2+1])) / 2)
+					}
+					outBuf = monoBuf
+				}
+				outgoing <- gumble.AudioBuffer(outBuf)
 				if recorder := s.getRecorder(); recorder != nil {
-					recorder.RecordAudioFrame(recorderOutgoingSource, int16Buffer, false)
+					recorder.RecordAudioFrame(recorderOutgoingSource, outBuf, false)
 				}
 			}
 		}
@@ -559,7 +1090,7 @@ func (s *Stream) processChannel(samples []int16, noiseProcessor NoiseProcessor,
 	if noiseProcessor != nil && noiseProcessor.IsEnabled() {
 		noiseProcessor.ProcessSamples(samples)
 	}
-	if micAGC != nil {
+	if micAGC != nil && micAGC.IsEnabled() {
 		micAGC.ProcessSamples(samples)
 	}
 }
@@ -567,6 +1098,9 @@ func (s *Stream) processChannel(samples []int16, noiseProcessor NoiseProcessor,
 func (s *Stream) ensureStereoProcessors() {
 	if s.micAGCRight == nil {
 		s.micAGCRight = audio.NewAGC()
+		if s.micAGC != nil {
+			s.micAGCRight.SetEnabled(s.micAGC.IsEnabled())
+		}
 	}
 	if s.noiseProcessorRight == nil {
 		s.noiseProcessorRight = cloneNoiseProcessor(s.noiseProcessor)

gumble/gumbleopenal/stream_regression_test.go

@@ -0,0 +1,100 @@
+package gumbleopenal
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"git.stormux.org/storm/barnard/gumble/go-openal/openal"
+	"git.stormux.org/storm/barnard/gumble/gumble"
+)
+
+// Regression: audio cleanup could send a final render command after Destroy
+// had closed renderCh, panicking instead of safely discarding that work.
+// Regression: StopSource returned before the capture worker ended, allowing
+// Destroy to close the device while that worker still used it.
+// Regression: OpenAL returned only a generic input/output error, hiding the
+// actual configured device that a user must correct.
+func TestDeviceOpenErrorsIncludeConfiguredDevice(t *testing.T) {
+	input := openInputDeviceError("virtual_mic.monitor", openal.FormatMono16)
+	if !errors.Is(input, ErrInputDevice) || !strings.Contains(input.Error(), "virtual_mic.monitor") {
+		t.Fatalf("input error %q", input)
+	}
+	output := openOutputDeviceError("")
+	if !errors.Is(output, ErrOutputDevice) || !strings.Contains(output.Error(), "default") {
+		t.Fatalf("output error %q", output)
+	}
+}
+
+// Regression: later capture start failures also omitted the configured device.
+func TestStartSourceUnavailableDeviceIncludesName(t *testing.T) {
+	s := &Stream{inputDeviceName: "virtual_mic.monitor"}
+	err := s.StartSource(nil)
+	if !errors.Is(err, ErrMic) || !strings.Contains(err.Error(), "virtual_mic.monitor") {
+		t.Fatalf("start error %q", err)
+	}
+}
+
+func TestStopSourceWaitsForWorker(t *testing.T) {
+	stop, done := make(chan bool), make(chan struct{})
+	s := &Stream{sourceStop: stop, sourceDone: done}
+	returned := make(chan struct{})
+	go func() { _ = s.StopSource(); close(returned) }()
+	select {
+	case <-returned:
+		t.Fatal("StopSource returned before worker")
+	default:
+	}
+	close(done)
+	<-returned
+}
+
+func TestJitterPlaybackDelayAppliesOnlyAtStartup(t *testing.T) {
+	if jitterPlaybackReady(false, 20*time.Millisecond, 40*time.Millisecond) {
+		t.Fatal("jitter playback started before initial buffer filled")
+	}
+	if !jitterPlaybackReady(false, 40*time.Millisecond, 40*time.Millisecond) {
+		t.Fatal("jitter playback did not start after initial buffer filled")
+	}
+	if !jitterPlaybackReady(true, 0, 40*time.Millisecond) {
+		t.Fatal("jitter playback paused while refilling after startup")
+	}
+}
+
+func TestJitterResyncsAfterSenderRestartsSequence(t *testing.T) {
+	// Mumble restarts frame numbering at zero when the sender switches audio
+	// devices mid-burst, and sends no terminator to announce it.
+	if !jitterShouldResync(jitterLateResync, 52724) {
+		t.Fatal("jitter did not resync after the sender restarted its frame numbering")
+	}
+	if jitterShouldResync(jitterLateResync-1, 52724) {
+		t.Fatal("jitter resynced before the late run was conclusive")
+	}
+	// A clump of reordered packets is bounded and recovers on its own; it must
+	// not drag the expected sequence backwards.
+	if jitterShouldResync(jitterLateResync, jitterResyncJump-1) {
+		t.Fatal("jitter resynced on a backwards jump small enough to be reordering")
+	}
+	if jitterShouldResync(1, 52724) {
+		t.Fatal("jitter resynced on a single late packet")
+	}
+}
+
+func TestAudioPacketDurationUsesStereoFrameCount(t *testing.T) {
+	packet := &gumble.AudioPacket{AudioBuffer: make(gumble.AudioBuffer, 2*gumble.AudioDefaultFrameSize)}
+	if got := audioPacketDuration(packet); got != 10*time.Millisecond {
+		t.Fatalf("audioPacketDuration = %v, want 10ms", got)
+	}
+}
+
+func TestRenderRejectsWorkAfterShutdown(t *testing.T) {
+	s := &Stream{renderClosed: true}
+	called := false
+	if s.render(func() { called = true }) {
+		t.Fatal("closed renderer accepted work")
+	}
+	if called {
+		t.Fatal("closed renderer executed work")
+	}
+}

main.go

@@ -14,6 +14,7 @@ import (
 	"os/exec"
 	"strings"
 	"syscall"
+	"time"
 
 	barnlog "git.stormux.org/storm/barnard/log"
 
@@ -114,6 +115,7 @@ func main() {
 	serverSet := false
 	usernameSet := false
 	buffers := flag.Int("buffers", 16, "number of audio buffers to use")
+	jitterBuffer := flag.Int("jitter-buffer", 40, "incoming per-user audio buffer in ms (0, 20, 40, or 60)")
 	profile := flag.Bool("profile", false, "add http server to serve profiles")
 	noiseSuppressionEnabled := flag.Bool("noise-suppression", false, "enable noise suppression for microphone input")
 	tcpOnly := flag.Bool("tcp", false, "disable UDP, force audio through TCP tunnel")
@@ -121,6 +123,10 @@ func main() {
 	logFile := flag.String("logfile", "", "write logs to this file (logging is disabled when omitted)")
 
 	flag.Parse()
+	selectedJitterBuffer, err := jitterBufferDuration(*jitterBuffer)
+	if err != nil {
+		handle_raw_error(err)
+	}
 
 	// Set up logging
 	var level barnlog.Level
@@ -214,6 +220,7 @@ func main() {
 	}
 	b.Config.Buffers = *buffers
 	b.Config.DisableUDP = *tcpOnly
+	b.Config.IncomingAudioBuffer = selectedJitterBuffer
 
 	b.Hotkeys = b.UserConfig.GetHotkeys()
 	b.UserConfig.SaveConfig()
@@ -253,6 +260,18 @@ func main() {
 	handle_error(&b)
 }
 
+// jitterBufferDuration converts the requested incoming playout delay to a
+// supported duration. Zero starts playback without an initial safety buffer.
+func jitterBufferDuration(milliseconds int) (time.Duration, error) {
+	interval := time.Duration(milliseconds) * time.Millisecond
+	switch interval {
+	case 0, 20 * time.Millisecond, 40 * time.Millisecond, 60 * time.Millisecond:
+		return interval, nil
+	default:
+		return 0, fmt.Errorf("jitter buffer must be 0, 20, 40, or 60 ms, got %d", milliseconds)
+	}
+}
+
 func handle_raw_error(e error) {
 	fmt.Fprintf(os.Stderr, "%s\n", e.Error())
 	os.Exit(1)