Hear the station here: stop cutting out behind a proxy or CDN

src/engine.rs

@@ -869,15 +869,27 @@ pub struct NetSource {
     cur: Vec<f32>,
     pos: usize,
     prebuffer: usize,
+    /// the most `prebuffer` may grow to after the sound runs dry (see `net_feed_adaptive`);
+    /// always at least `prebuffer`
+    grow_to: usize,
     buffering: bool,
 }
 
 /// A network feed: push blocks (with `try_send`, so a slow consumer never holds the network up)
 /// into the sender; the `NetSource` plays them.
+#[allow(dead_code)] // RadioMaster-Server (the mic)
 pub fn net_feed(prebuffer: usize) -> (Sender<Vec<f32>>, NetSource) {
+    net_feed_adaptive(prebuffer, prebuffer)
+}
+
+/// Like `net_feed`, but each time the sound runs dry after playing, the prebuffer doubles (up to
+/// `grow_to` blocks). Over a LAN it stays short; behind a reverse proxy or CDN that passes the sound
+/// on in bursts, it grows until a burst's gap fits, instead of cutting in and out for good.
+pub fn net_feed_adaptive(prebuffer: usize, grow_to: usize) -> (Sender<Vec<f32>>, NetSource) {
     let prebuffer = prebuffer.max(1);
-    let (tx, rx) = crossbeam_channel::bounded(prebuffer * 4);
-    (tx, NetSource { rx, cur: Vec::new(), pos: 0, prebuffer, buffering: true })
+    let grow_to = grow_to.max(prebuffer);
+    let (tx, rx) = crossbeam_channel::bounded(grow_to * 4);
+    (tx, NetSource { rx, cur: Vec::new(), pos: 0, prebuffer, grow_to, buffering: true })
 }
 
 impl NetSource {
@@ -898,6 +910,8 @@ impl NetSource {
                     self.pos = 0;
                 }
                 Err(_) => {
+                    // ran dry while playing: wait for more next time
+                    self.prebuffer = (self.prebuffer * 2).min(self.grow_to);
                     self.buffering = true;
                     self.cur.clear();
                     self.pos = 0;
@@ -1017,6 +1031,39 @@ mod tests {
         std::fs::write(path, h).unwrap();
     }
 
+    #[test]
+    fn network_feed_waits_longer_after_running_dry() {
+        let (tx, mut src) = net_feed_adaptive(2, 8);
+        let play = |src: &mut NetSource, frames: usize| (0..frames).filter(|_| src.frame() != (0.0, 0.0)).count();
+        // two blocks of one frame each start it
+        tx.send(vec![1.0, 1.0]).unwrap();
+        assert_eq!(play(&mut src, 1), 0, "still buffering");
+        tx.send(vec![1.0, 1.0]).unwrap();
+        assert_eq!(play(&mut src, 2), 2);
+        // it ran dry: now it waits for four blocks
+        assert_eq!(play(&mut src, 1), 0);
+        for _ in 0..3 {
+            tx.send(vec![1.0, 1.0]).unwrap();
+        }
+        assert_eq!(play(&mut src, 1), 0, "three aren't enough any more");
+        tx.send(vec![1.0, 1.0]).unwrap();
+        assert_eq!(play(&mut src, 4), 4);
+        // never past grow_to; a burst that big still fits
+        for _ in 0..3 {
+            assert_eq!(play(&mut src, 1), 0);
+        }
+        assert_eq!(src.prebuffer, 8);
+        for _ in 0..24 {
+            tx.try_send(vec![1.0, 1.0]).unwrap();
+        }
+        assert_eq!(play(&mut src, 24), 24);
+        // a fixed feed (the mic) keeps its prebuffer
+        let (_tx, mut fixed) = net_feed(3);
+        fixed.frame();
+        fixed.frame();
+        assert_eq!(fixed.prebuffer, 3);
+    }
+
     #[test]
     fn speech_with_pauses_plays_to_the_end_even_with_silence_trimming() {
         if std::process::Command::new("espeak-ng").arg("--version").output().is_err() {

src/link.rs

@@ -455,6 +455,9 @@ impl Fs {
 
 /// How much of the server's mix is buffered before it plays (decoded MP3 frames of about 26 ms).
 const LISTEN_PREBUFFER: usize = 12;
+/// How far that may grow when the sound keeps running dry (about 6 s): a reverse proxy or CDN in
+/// front of the server (Cloudflare, nginx, IIS) can pass the stream on in bursts a few seconds apart.
+const LISTEN_PREBUFFER_MAX: usize = 230;
 /// MP3 bitrate of the mic sent to the server.
 const MIC_KBPS: u32 = 128;
 /// The mic goes up in slices this long.
@@ -496,7 +499,7 @@ impl Listener {
         token: Arc<Mutex<String>>,
         say: Box<dyn Fn(String) + Send>,
     ) -> Listener {
-        let (feed, src) = crate::engine::net_feed(LISTEN_PREBUFFER);
+        let (feed, src) = crate::engine::net_feed_adaptive(LISTEN_PREBUFFER, LISTEN_PREBUFFER_MAX);
         engine.set_feed(Some(src));
         let stop = Arc::new(AtomicBool::new(false));
         let state = Arc::new(Mutex::new(Hearing::Connecting));

src/server.rs

@@ -339,6 +339,9 @@ const TOKEN_IDLE: Duration = Duration::from_secs(24 * 3600);
 
 /// MP3 bitrate of the mix sent to the apps.
 const LISTEN_KBPS: u32 = 192;
+/// How many mix blocks (about 23 ms each) may pile up while an app (or a proxy in front of the
+/// server) reads slowly before the oldest are skipped: about 3 s. The app trims its own delay.
+const LISTEN_BACKLOG: usize = 130;
 /// An app's mic goes off when nothing arrives for this long (the app quit or lost the network).
 const MIC_IDLE: Duration = Duration::from_secs(3);
 /// Mic audio waits for this many decoded MP3 frames (about 26 ms each) before it plays.
@@ -1065,19 +1068,20 @@ impl Api {
             };
             let engine = api.core.lock().engine.clone();
             let mut w = req.into_writer();
-            let head = "HTTP/1.1 200 OK\r\nContent-Type: audio/mpeg\r\nCache-Control: no-cache, no-store\r\nX-Accel-Buffering: no\r\nConnection: close\r\n\r\n";
+            let head = "HTTP/1.1 200 OK\r\nContent-Type: audio/mpeg\r\nCache-Control: no-cache, no-store, no-transform\r\nX-Accel-Buffering: no\r\nConnection: close\r\n\r\n";
             if w.write_all(head.as_bytes()).and_then(|_| w.flush()).is_err() {
                 return;
             }
-            let (tx, rx) = crossbeam_channel::bounded::<Arc<Vec<i16>>>(64);
+            // room for a few seconds: a proxy in front of the server can stop reading for a moment
+            let (tx, rx) = crossbeam_channel::bounded::<Arc<Vec<i16>>>(LISTEN_BACKLOG * 2);
             engine.add_tap(tx);
             let mut out = Vec::new();
             let mut checked = Instant::now();
             loop {
                 match rx.recv_timeout(Duration::from_secs(1)) {
                     Ok(pcm) => {
                         // keep the delay short: skip what piled up while the network was slow
-                        while rx.len() > 12 {
+                        while rx.len() > LISTEN_BACKLOG {
                             let _ = rx.try_recv();
                         }
                         out.clear();