audio: harden Opus playback

Preserve decoder timing and keep blocking speaker writes off codec
callbacks. Playback failures no longer take down video or control.

5 files changed, 184 insertions, 147 deletions

diff --git a/app/src/main/java/invalid/lena/scrcpy/AudioFrames.java b/app/src/main/java/invalid/lena/scrcpy/AudioFrames.java
index da0bb0a..eaf0db1 100644
--- a/app/src/main/java/invalid/lena/scrcpy/AudioFrames.java
+++ b/app/src/main/java/invalid/lena/scrcpy/AudioFrames.java
@@ -4,6 +4,7 @@ package invalid.lena.scrcpy;
 // implementation; tests use a recording stub. Kept android-free.
 //
 // start(fourcc) tells the sink which wire codec it should expect.
+// ptsUs is the wire presentation timestamp with protocol flags removed.
 // For raw payloads are interleaved s16le PCM. For opus the first
 // frame has isConfig=true and payload is the OpusHead (the server
 // pre-strips the AOPUSHDR/AOPUSDLY/AOPUSPRL container); subsequent
@@ -12,7 +13,7 @@ public interface AudioFrames {
 
     void start(int fourcc);
 
-    void feed(byte[] data, int off, int len, boolean isConfig);
+    void feed(byte[] data, int off, int len, long ptsUs, boolean isConfig);
 
     void release();
 }
diff --git a/app/src/main/java/invalid/lena/scrcpy/AudioSink.java b/app/src/main/java/invalid/lena/scrcpy/AudioSink.java
index 5507c5c..f14f85c 100644
--- a/app/src/main/java/invalid/lena/scrcpy/AudioSink.java
+++ b/app/src/main/java/invalid/lena/scrcpy/AudioSink.java
@@ -14,6 +14,8 @@ import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.ArrayDeque;
 import java.util.Deque;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 // Audio output: AudioTrack writing 48 kHz stereo 16-bit PCM. The
@@ -25,6 +27,7 @@ public final class AudioSink implements AudioFrames {
     private static final int CHANNEL_OUT = AudioFormat.CHANNEL_OUT_STEREO;
     private static final int ENCODING    = AudioFormat.ENCODING_PCM_16BIT;
     private static final int MAX_PENDING_OPUS = 16;
+    private static final int MAX_PENDING_PCM = 16;
     private static final int MAX_PACKET_BYTES = 256 * 1024;
 
     // Defaults documented at <https://developer.android.com/reference/android/media/MediaCodec#CSD>.
@@ -34,24 +37,29 @@ public final class AudioSink implements AudioFrames {
     private volatile MediaCodec opusCodec;
     private HandlerThread opusThread;
     private Handler       opusHandler;
+    private Thread        opusOutputThread;
     private volatile boolean opusConfigured;
     private int           fourcc;
 
     private final Object lock = new Object();
-    private final Runnable onFatalError;
-    private final AtomicBoolean fatalReported = new AtomicBoolean();
+    private final AtomicBoolean failed = new AtomicBoolean();
     private final AtomicBoolean playbackReported = new AtomicBoolean();
     private final Deque<Integer> freeOpusInputs = new ArrayDeque<>();
-    private final Deque<byte[]> pendingOpus = new ArrayDeque<>(MAX_PENDING_OPUS);
+    private final Deque<Packet> pendingOpus = new ArrayDeque<>(MAX_PENDING_OPUS);
+    private final ArrayBlockingQueue<byte[]> pendingPcm =
+            new ArrayBlockingQueue<>(MAX_PENDING_PCM);
 
-    private volatile boolean released;
-    private long droppedBytes;
-    private long lastDropLogMs;
+    private static final class Packet {
+        final byte[] data;
+        final long ptsUs;
 
-    public AudioSink(Runnable onFatalError) {
-        this.onFatalError = onFatalError;
+        Packet(byte[] data, long ptsUs) {
+            this.data = data;
+            this.ptsUs = ptsUs;
+        }
     }
 
+    private volatile boolean released;
     @Override
     public void start(int fourcc) {
         this.fourcc = fourcc;
@@ -96,25 +104,31 @@ public final class AudioSink implements AudioFrames {
             } catch (IOException e) {
                 throw new IllegalStateException("audio sink: no opus decoder", e);
             }
-             opusCodec.setCallback(new MediaCodec.Callback() {
-                 @Override public void onInputBufferAvailable(MediaCodec mc, int idx) {
+            opusOutputThread = new Thread(this::playOpus, "audio-out");
+            opusOutputThread.start();
+            opusCodec.setCallback(new MediaCodec.Callback() {
+                @Override public void onInputBufferAvailable(MediaCodec mc, int idx) {
                     synchronized (lock) {
-                        if (released || mc != opusCodec) return;
-                        byte[] packet = pendingOpus.pollFirst();
-                        if (packet == null) freeOpusInputs.offerLast(idx);
-                        else submitOpus(mc, idx, packet);
+                        if (released || failed.get() || mc != opusCodec) return;
+                        Packet packet = pendingOpus.pollFirst();
+                        if (packet == null) {
+                            freeOpusInputs.offerLast(idx);
+                        } else {
+                            lock.notifyAll();
+                            submitOpus(mc, idx, packet);
+                        }
                     }
                 }
                 @Override public void onOutputBufferAvailable(MediaCodec mc, int idx,
                                                               MediaCodec.BufferInfo info) {
+                    byte[] pcm = null;
                     try {
                         ByteBuffer out = mc.getOutputBuffer(idx);
-                        if (out != null && info.size > 0 && !released) {
-                            byte[] pcm = new byte[info.size];
+                        if (out != null && info.size > 0 && !released && !failed.get()) {
+                            pcm = new byte[info.size];
                             out.position(info.offset);
                             out.limit(info.offset + info.size);
                             out.get(pcm);
-                            writePcm(pcm, 0, pcm.length);
                         }
                     } catch (IllegalStateException e) {
                         reportFatal(e, "audio sink: opus output");
@@ -122,6 +136,7 @@ public final class AudioSink implements AudioFrames {
                         try { mc.releaseOutputBuffer(idx, false); }
                         catch (IllegalStateException ignored) {}
                     }
+                    if (pcm != null) queuePcm(pcm);
                 }
                 @Override public void onError(MediaCodec mc, MediaCodec.CodecException e) {
                     reportFatal(e, "audio sink: opus codec error");
@@ -134,7 +149,7 @@ public final class AudioSink implements AudioFrames {
     }
 
     @Override
-    public void feed(byte[] data, int off, int len, boolean isConfig) {
+    public void feed(byte[] data, int off, int len, long ptsUs, boolean isConfig) {
         if (data == null || off < 0 || len < 0 || off > data.length - len) {
             throw new IndexOutOfBoundsException("invalid audio frame range");
         }
@@ -142,16 +157,16 @@ public final class AudioSink implements AudioFrames {
             reportFatal(null, "audio sink: packet too large (" + len + " bytes)");
             return;
         }
-        if (released || len == 0) return;
+        if (released || failed.get() || len == 0) return;
         if (fourcc == Wire.CODEC_OPUS) {
-            feedOpus(data, off, len, isConfig);
+            feedOpus(data, off, len, ptsUs, isConfig);
         } else {
             // Raw PCM: passthrough.
             writePcm(data, off, len);
         }
     }
 
-    private void feedOpus(byte[] data, int off, int len, boolean isConfig) {
+    private void feedOpus(byte[] data, int off, int len, long ptsUs, boolean isConfig) {
         if (isConfig) {
             if (opusConfigured) return; // already configured
             // OpusHead is 19 bytes; pre_skip lives at bytes [10..11]. Reject
@@ -183,31 +198,45 @@ public final class AudioSink implements AudioFrames {
         if (!opusConfigured) return;
         byte[] packet = new byte[len];
         System.arraycopy(data, off, packet, 0, len);
+        Packet p = new Packet(packet, ptsUs);
         synchronized (lock) {
             if (released || opusCodec == null) return;
+            // Apply socket backpressure during decoder bursts. Dropping a
+            // compressed packet corrupts playback; growing the queue only
+            // postpones the same failure.
+            while (!released && !failed.get() && freeOpusInputs.isEmpty()
+                    && pendingOpus.size() == MAX_PENDING_OPUS) {
+                try {
+                    lock.wait();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    return;
+                }
+            }
+            if (released || failed.get() || opusCodec == null) return;
             Integer idx = freeOpusInputs.pollFirst();
             if (idx != null) {
-                submitOpus(opusCodec, idx, packet);
+                submitOpus(opusCodec, idx, p);
                 return;
             }
-            if (pendingOpus.size() == MAX_PENDING_OPUS) pendingOpus.pollFirst();
-            pendingOpus.offerLast(packet);
+            pendingOpus.offerLast(p);
         }
     }
 
     // Must be called with lock held. Async MediaCodec input indices belong
     // to the codec instance that delivered them.
-    private void submitOpus(MediaCodec codec, int idx, byte[] packet) {
+    private void submitOpus(MediaCodec codec, int idx, Packet packet) {
         try {
             ByteBuffer in = codec.getInputBuffer(idx);
-            if (in == null || packet.length > in.capacity()) {
+            if (in == null || packet.data.length > in.capacity()) {
                 reportFatal(null,
-                        "audio sink: opus packet exceeds codec input (" + packet.length + " bytes)");
+                        "audio sink: opus packet exceeds codec input ("
+                                + packet.data.length + " bytes)");
                 return;
             }
             in.clear();
-            in.put(packet);
-            codec.queueInputBuffer(idx, 0, packet.length, 0, 0);
+            in.put(packet.data);
+            codec.queueInputBuffer(idx, 0, packet.data.length, packet.ptsUs, 0);
         } catch (IllegalStateException e) {
             reportFatal(e, "audio sink: opus queueInputBuffer");
         }
@@ -220,32 +249,60 @@ public final class AudioSink implements AudioFrames {
         return b;
     }
 
-    private void writePcm(byte[] data, int off, int len) {
-        AudioTrack t = track;
-        if (released || t == null || len <= 0) return;
-        int written = t.write(data, off, len, AudioTrack.WRITE_NON_BLOCKING);
-        if (written < 0) {
-            reportFatal(null, "audio sink: AudioTrack write rc=" + written);
-            return;
+    private void queuePcm(byte[] pcm) {
+        try {
+            while (!released && !failed.get()
+                    && !pendingPcm.offer(pcm, 100, TimeUnit.MILLISECONDS)) {
+                // Wait for the blocking AudioTrack writer to make room.
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
         }
-        if (written > 0 && playbackReported.compareAndSet(false, true)) {
-            Log.i("audio sink: playback started");
+    }
+
+    private void playOpus() {
+        try {
+            while (!released && !failed.get()) {
+                byte[] pcm = pendingPcm.take();
+                writePcm(pcm, 0, pcm.length);
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        } catch (RuntimeException e) {
+            reportFatal(e, "audio sink: AudioTrack write");
         }
-        if (written < len) {
-            droppedBytes += (len - written);
-            long now = System.currentTimeMillis();
-            if (now - lastDropLogMs > 1000L) {
-                lastDropLogMs = now;
-                Log.w("audio sink: ring full, dropped %d bytes so far", droppedBytes);
+    }
+
+    private void writePcm(byte[] data, int off, int len) {
+        AudioTrack t = track;
+        if (released || failed.get() || t == null || len <= 0) return;
+        int end = off + len;
+        while (!released && !failed.get() && off < end) {
+            int written = t.write(data, off, end - off, AudioTrack.WRITE_BLOCKING);
+            if (written <= 0) {
+                if (!released) {
+                    reportFatal(null, "audio sink: AudioTrack write rc=" + written);
+                }
+                return;
+            }
+            off += written;
+            if (playbackReported.compareAndSet(false, true)) {
+                Log.i("audio sink: playback started");
             }
         }
     }
 
     private void reportFatal(Exception error, String message) {
-        if (!fatalReported.compareAndSet(false, true)) return;
+        if (!failed.compareAndSet(false, true)) return;
         if (error == null) Log.e("%s", message);
         else Log.e(error, "%s", message);
-        if (onFatalError != null) onFatalError.run();
+        synchronized (lock) {
+            pendingOpus.clear();
+            lock.notifyAll();
+        }
+        pendingPcm.clear();
+        Thread t = opusOutputThread;
+        if (t != null) t.interrupt();
     }
 
     @Override
@@ -257,20 +314,35 @@ public final class AudioSink implements AudioFrames {
             opusCodec = null;
             freeOpusInputs.clear();
             pendingOpus.clear();
+            lock.notifyAll();
         }
+        AudioTrack t = track;
+        track = null;
+        // Stop playback first so the blocking writer can return before
+        // AudioTrack is released.
+        if (t != null) {
+            try { t.pause(); t.flush(); t.stop(); } catch (Exception ignored) {}
+        }
+
         HandlerThread ht = opusThread;
         opusThread = null;
         opusHandler = null;
+        Thread out = opusOutputThread;
+        opusOutputThread = null;
+        if (out != null) out.interrupt();
         if (c != null) {
             try { c.stop(); } catch (Exception ignored) {}
             try { c.release(); } catch (Exception ignored) {}
         }
         if (ht != null) ht.quitSafely();
-
-        AudioTrack t = track;
-        track = null;
-        if (t == null) return;
-        try { t.pause(); t.flush(); t.stop(); } catch (Exception ignored) {}
-        try { t.release(); } catch (Exception ignored) {}
+        if (out != null && out != Thread.currentThread()) {
+            try { out.join(1_000); }
+            catch (InterruptedException e) { Thread.currentThread().interrupt(); }
+            if (out.isAlive()) Log.e("audio sink: output thread did not stop");
+        }
+        pendingPcm.clear();
+        if (t != null) {
+            try { t.release(); } catch (Exception ignored) {}
+        }
     }
 }
diff --git a/app/src/main/java/invalid/lena/scrcpy/AudioStream.java b/app/src/main/java/invalid/lena/scrcpy/AudioStream.java
index 13b39f5..45ea69c 100644
--- a/app/src/main/java/invalid/lena/scrcpy/AudioStream.java
+++ b/app/src/main/java/invalid/lena/scrcpy/AudioStream.java
@@ -2,7 +2,6 @@ package invalid.lena.scrcpy;
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.concurrent.TimeUnit;
 
 // Reads the scrcpy audio socket and drives an AudioFrames sink.
 //
@@ -19,7 +18,10 @@ import java.util.concurrent.TimeUnit;
 // Takes a plain InputStream; caller owns stream lifecycle.
 public final class AudioStream {
 
-    private static final long FLAG_CONFIG = 1L << 62;
+    private static final long FLAG_SESSION  = 1L << 63;
+    private static final long FLAG_CONFIG   = 1L << 62;
+    private static final long FLAG_KEYFRAME = 1L << 61;
+    private static final long PTS_MASK      = ~(FLAG_SESSION | FLAG_CONFIG | FLAG_KEYFRAME);
 
     // Generous upper bound for one audio packet (a raw PCM block or an
     // opus packet is a few KB). A corrupt or hostile length field must
@@ -28,30 +30,12 @@ public final class AudioStream {
 
     private final InputStream  source;
     private final AudioFrames  sink;
-    private final Runnable     onFatalError;
     private Thread             thread;
     private volatile boolean   stop;
 
-    // Liveness signal for Session's stall watchdog. The audio socket is
-    // the only one that is reliably continuous: the server captures PCM
-    // at 48 kHz whether or not anything on the target's screen moves, so
-    // silence here means the wire is gone. Video is useless for this - a
-    // static screen queues nothing to the encoder and legitimately
-    // produces no frames for minutes.
-    private volatile boolean started;
-    private volatile long    lastPacketAtMs;
-
-    public boolean isStarted()      { return started; }
-    public long    lastPacketAtMs() { return lastPacketAtMs; }
-
     public AudioStream(InputStream source, AudioFrames sink) {
-        this(source, sink, null);
-    }
-
-    public AudioStream(InputStream source, AudioFrames sink, Runnable onFatalError) {
         this.source = source;
         this.sink = sink;
-        this.onFatalError = onFatalError;
     }
 
     public void start() {
@@ -85,9 +69,13 @@ public final class AudioStream {
                 throw new IOException("audio: unexpected codec " + Wire.fourccName(fourcc));
             }
             Log.i("audio meta codec=%s", Wire.fourccName(fourcc));
-            sink.start(fourcc);
-            lastPacketAtMs = monotonicMs();
-            started = true;
+            boolean playback = true;
+            try {
+                sink.start(fourcc);
+            } catch (RuntimeException e) {
+                Log.e(e, "audio output disabled");
+                playback = false;
+            }
 
             byte[] hdr = new byte[12];
             byte[] payload = new byte[16 * 1024];
@@ -97,33 +85,28 @@ public final class AudioStream {
                 long ptsAndFlags = Wire.readBe64(hdr, 0);
                 int  size = Wire.readBe32(hdr, 8);
                 boolean cfg = (ptsAndFlags & FLAG_CONFIG) != 0;
+                long ptsUs = ptsAndFlags & PTS_MASK;
                 if (size <= 0 || size > MAX_FRAME_SIZE) {
                     throw new IOException("audio frame size out of range: " + size);
                 }
                 if (size > payload.length) payload = new byte[size];
                 Wire.readFully(source, payload, 0, size);
-                lastPacketAtMs = monotonicMs();
-                sink.feed(payload, 0, size, cfg);
+                if (playback) {
+                    try {
+                        sink.feed(payload, 0, size, ptsUs, cfg);
+                    } catch (RuntimeException e) {
+                        Log.e(e, "audio output disabled");
+                        playback = false;
+                    }
+                }
                 if (++frames == 1) Log.i("audio frame n=1 size=%d cfg=%s", size, cfg);
             }
         } catch (IOException e) {
-            if (!stop) {
-                Log.e(e, "audio reader");
-                reportFatal();
-            }
+            if (!stop) Log.e(e, "audio reader");
         } catch (Exception e) {
             Log.e(e, "audio reader unexpected");
-            if (!stop) reportFatal();
         } finally {
             Log.i("audio reader: end");
         }
     }
-
-    private void reportFatal() {
-        if (onFatalError != null) onFatalError.run();
-    }
-
-    private static long monotonicMs() {
-        return TimeUnit.NANOSECONDS.toMillis(System.nanoTime());
-    }
 }
diff --git a/app/src/main/java/invalid/lena/scrcpy/Session.java b/app/src/main/java/invalid/lena/scrcpy/Session.java
index 294b27d..39ef38a 100644
--- a/app/src/main/java/invalid/lena/scrcpy/Session.java
+++ b/app/src/main/java/invalid/lena/scrcpy/Session.java
@@ -67,22 +67,8 @@ public final class Session {
     // deterministic failure walks the ladder and then gives up.
     private static final long[] RECONNECT_BACKOFF_MS =
             {1_000L, 2_000L, 5_000L, 10_000L, 15_000L};
-    // Comfortably longer than STALL_MS + STALL_POLL_MS. If it were not,
-    // a stall-detected drop would always look "healthy" (a stall is only
-    // declared after STALL_MS of silence, so the connection is at least
-    // that old by then), the budget would reset on every stall and the
-    // ladder would never be reached for the one failure it exists for.
     private static final long HEALTHY_MS = 120_000L;
 
-    // Stall watchdog. adb streams have no read timeout - AdbStream.read()
-    // waits on its queue until data arrives or the stream is closed - so a
-    // target that disappears without closing the socket (Wi-Fi dropping
-    // mid-frame, NAT rebinding, a carrier idle timeout on a VPN link)
-    // leaves the readers parked and the session frozen forever with no
-    // reconnect. Poll for silence on the audio socket instead; see
-    // AudioStream.isStarted() for why audio and not video.
-    private static final long STALL_MS      = 30_000L;
-    private static final long STALL_POLL_MS = 5_000L;
     private static final long BRING_UP_DEADLINE_MS = 90_000L;
     private static final long STOP_JOIN_MS = 10_000L;
 
@@ -118,9 +104,8 @@ public final class Session {
         }
     }
 
-    // Counted down by the video reader when its loop exits; the session
-    // thread parks on it for the live duration of a connection. Swapped
-    // for a fresh latch on each reconnect cycle.
+    // Counted down when video or control ends; audio is optional and never
+    // ends a session. Swapped for a fresh latch on each reconnect cycle.
     private volatile CountDownLatch endSignal = new CountDownLatch(1);
 
     public Session(Context ctx, Adb adb, Devices.Device target, Surface surface, Listener listener) {
@@ -242,14 +227,12 @@ public final class Session {
             if (!connect()) return;        // gave up: onError + onStopped fired
             long upAt = monotonicMs();
             CountDownLatch latch;
-            AudioStream    as;
             synchronized (this) {
                 if (stopped) return;
                 latch = endSignal;
-                as    = audioStream;
             }
             try {
-                awaitEndOrStall(latch, as);
+                latch.await();
             } catch (InterruptedException ie) {
                 return;
             }
@@ -286,29 +269,6 @@ public final class Session {
         }
     }
 
-    // Park until the pipeline dies, stop() fires, or the audio socket has
-    // been silent long enough that the link must be gone. Returning
-    // without the latch firing leaves the readers parked; the caller's
-    // tearDownInstalled() closes their streams, which unblocks them.
-    //
-    // The watchdog only arms once audio is actually flowing. If the
-    // target cannot capture audio the server reports the stream disabled,
-    // AudioStream returns immediately, and there is no reliable idle
-    // signal left - in that case fall back to waiting indefinitely rather
-    // than inventing one from the video socket, which is legitimately
-    // silent whenever the target's screen is static.
-    private void awaitEndOrStall(CountDownLatch latch, AudioStream as)
-            throws InterruptedException {
-        while (!latch.await(STALL_POLL_MS, TimeUnit.MILLISECONDS)) {
-            if (stopped || as == null || !as.isStarted()) continue;
-            long idle = monotonicMs() - as.lastPacketAtMs();
-            if (idle >= STALL_MS) {
-                Log.w("session: no audio for %d ms - link presumed dead", idle);
-                return;
-            }
-        }
-    }
-
     // Run the bring-up retry ladder once. Returns true when a session is
     // live (read threads started); false if the budget was exhausted, in
     // which case onError() + onStopped() have already fired.
@@ -429,8 +389,10 @@ public final class Session {
             ctrl = new Controller(ctx, cs::send);
             cs.setInboundSink(ctrl);
 
-            ak = new AudioSink(ended);
-            as = new AudioStream(s.audioIn, ak, ended);
+            // Audio is optional. Failure to capture or play it must not tear
+            // down video and control.
+            ak = new AudioSink();
+            as = new AudioStream(s.audioIn, ak);
 
             vk = new VideoSink(surface, ended);
             Controller ctrlRef = ctrl;
diff --git a/app/src/test/java/invalid/lena/scrcpy/AudioStreamTest.java b/app/src/test/java/invalid/lena/scrcpy/AudioStreamTest.java
index 005a2e3..1fca17c 100644
--- a/app/src/test/java/invalid/lena/scrcpy/AudioStreamTest.java
+++ b/app/src/test/java/invalid/lena/scrcpy/AudioStreamTest.java
@@ -11,21 +11,24 @@ import java.io.DataOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
 
 public class AudioStreamTest {
 
     private static final long FLAG_CONFIG = 1L << 62;
+    private static final long FLAG_KEYFRAME = 1L << 61;
 
     private static final class RecordingFrames implements AudioFrames {
         int starts = 0, releases = 0, startFourcc = 0;
         final List<byte[]>  feeds = new ArrayList<>();
+        final List<Long>     pts   = new ArrayList<>();
         final List<Boolean> cfgs  = new ArrayList<>();
         @Override public void start(int fourcc) { starts++; startFourcc = fourcc; }
-        @Override public void feed(byte[] data, int off, int len, boolean isConfig) {
+        @Override public void feed(byte[] data, int off, int len,
+                                   long ptsUs, boolean isConfig) {
             byte[] cp = new byte[len];
             System.arraycopy(data, off, cp, 0, len);
             feeds.add(cp);
+            pts.add(ptsUs);
             cfgs.add(isConfig);
         }
         @Override public void release() { releases++; }
@@ -73,6 +76,8 @@ public class AudioStreamTest {
         for (int i = 0; i < pcm2.length; i++) assertEquals(pcm2[i], sink.feeds.get(1)[i]);
         assertEquals(false, sink.cfgs.get(0));
         assertEquals(false, sink.cfgs.get(1));
+        assertEquals(Long.valueOf(1_000_000L), sink.pts.get(0));
+        assertEquals(Long.valueOf(2_000_000L), sink.pts.get(1));
     }
 
     @Test
@@ -90,7 +95,7 @@ public class AudioStreamTest {
         byte[] bytes = cat(
                 fourcc(Wire.CODEC_OPUS),
                 frame(FLAG_CONFIG, head),
-                frame(0L, pkt));
+                frame(FLAG_KEYFRAME | 3_000_000L, pkt));
 
         RecordingFrames sink = new RecordingFrames();
         new AudioStream(new ByteArrayInputStream(bytes), sink).run();
@@ -102,6 +107,8 @@ public class AudioStreamTest {
         assertTrue("second frame must not carry FLAG_CONFIG", !sink.cfgs.get(1));
         assertEquals(head.length, sink.feeds.get(0).length);
         assertEquals(pkt.length, sink.feeds.get(1).length);
+        assertEquals(Long.valueOf(0L), sink.pts.get(0));
+        assertEquals(Long.valueOf(3_000_000L), sink.pts.get(1));
     }
 
     @Test
@@ -117,18 +124,30 @@ public class AudioStreamTest {
     public void errorCodecDoesNotStart() throws Exception {
         byte[] bytes = fourcc(1);
         RecordingFrames sink = new RecordingFrames();
-        AtomicInteger fatal = new AtomicInteger();
-        new AudioStream(new ByteArrayInputStream(bytes), sink, fatal::incrementAndGet).run();
+        new AudioStream(new ByteArrayInputStream(bytes), sink).run();
         assertEquals(0, sink.starts);
         assertEquals(0, sink.feeds.size());
-        assertEquals(1, fatal.get());
     }
 
     @Test
-    public void disabledCodecIsNotFatal() throws Exception {
-        AtomicInteger fatal = new AtomicInteger();
-        new AudioStream(new ByteArrayInputStream(fourcc(0)),
-                new RecordingFrames(), fatal::incrementAndGet).run();
-        assertEquals(0, fatal.get());
+    public void outputFailureStillDrainsWire() throws Exception {
+        byte[] bytes = cat(
+                fourcc(Wire.CODEC_RAW),
+                frame(1_000_000L, new byte[]{1, 2, 3, 4}));
+        ByteArrayInputStream in = new ByteArrayInputStream(bytes);
+        AudioFrames sink = new AudioFrames() {
+            @Override public void start(int fourcc) {
+                throw new IllegalStateException("no audio output");
+            }
+            @Override public void feed(byte[] data, int off, int len,
+                                       long ptsUs, boolean isConfig) {
+                throw new AssertionError("disabled output must not receive frames");
+            }
+            @Override public void release() {}
+        };
+
+        new AudioStream(in, sink).run();
+
+        assertEquals(0, in.available());
     }
 }