fix: close stream receiver if lagged (#203)

front/app/src/lib/css/app.css

@@ -9,7 +9,10 @@
 
 /* @import url("./circular/circular.css"); */
 /* @import url("./mada/font.css"); */
+
 @import url("./sofia-sans/font.css");
+/* @import url("./brandon/font.css"); */
+
 @import url("./azeret-mono/azeret-mono.css");
 /* @import url("./chivo-mono/chivo-mono.css"); */
 

front/app/src/lib/css/brandon/font.css

@@ -0,0 +1,83 @@
+@font-face {
+  font-family: Brandon;
+  font-style: normal;
+  font-weight: 200;
+  src: url("./thin.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: italic;
+  font-weight: 200;
+  src: url("./thin_italic.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: normal;
+  font-weight: 300;
+  src: url("./light.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: italic;
+  font-weight: 300;
+  src: url("./light_italic.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: normal;
+  font-weight: 400;
+  src: url("./regular.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: italic;
+  font-weight: 400;
+  src: url("./regular_italic.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: normal;
+  font-weight: 600;
+  src: url("./medium.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: italic;
+  font-weight: 600;
+  src: url("./medium_italic.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: normal;
+  font-weight: 700;
+  src: url("./bold.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: italic;
+  font-weight: 700;
+  src: url("./bold_italic.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: normal;
+  font-weight: 900;
+  src: url("./black.woff2") format('woff2'); 
+}
+
+@font-face {
+  font-family: Brandon;
+  font-style: italic;
+  font-weight: 900;
+  src: url("./black_italic.woff2") format('woff2'); 
+}

rs/packages/db/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2021"
 
 [features]
 default = ["analytics-max-concurrent"]
+# default = []
 analytics-max-concurrent = []
 
 # do not enable this, is for internal benchmarking only

rs/packages/media/src/handle/external_relay.rs

@@ -167,6 +167,8 @@ pub fn run_external_relay_source(
               None => break,
               Some(Err(_e)) => break,
               Some(Ok(bytes)) => {
+                transfer.fetch_add(bytes.len(), Ordering::Relaxed);
+
                 match sender.send(bytes) {
                   Ok(_) => {
                     no_listeners_since = None;

rs/packages/stream/src/lib.rs

@@ -27,6 +27,7 @@ use std::net::SocketAddr;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::{Instant, SystemTime};
+use tokio::time::{sleep, Duration};
 use transfer_map::TransferTracer;
 
 mod error;
@@ -451,23 +452,37 @@ impl StreamHandler {
 
               loop_i += 1;
 
+              let mut last_recv = Instant::now();
+
               'recv: loop {
-                let r = rx.recv().await;
+                match rx.recv().await {
+                  // Here the channel has been dropped
+                  Err(RecvError::Closed) => break 'recv,
 
-                match r {
                   // if lagged we ignore the error and continue with the oldest message buffered in the channel
                   // TODO: maybe we should advance to the newest message with stream.resubscribe()
-                  Err(RecvError::Lagged(_)) => continue 'recv,
+                  Err(RecvError::Lagged(_)) => {
+                    // break this receiver if lagged behind more than 1 minute
+                    if last_recv.elapsed().as_millis() > 60_000 {
+                      break 'root;
+                    }
 
-                  // Here the channel has been dropped
-                  Err(RecvError::Closed) => break 'recv,
+                    continue 'recv;
+                  }
 
                   // Receive bytes and pass it to response body
                   Ok(bytes) => {
                     rx_had_data = true;
+                    last_recv = Instant::now();
 
                     let len = bytes.len();
-                    match body_sender.send_data(bytes).await {
+
+                    let r = tokio::select! {
+                      _ = sleep(Duration::from_millis(60_000)) => break 'root,
+                      r = body_sender.send_data(bytes) => r,
+                    };
+
+                    match r {
                       Err(_) => break 'root,
                       Ok(()) => {
                         transfer_map.increment(&station.account_id, len);