bstein/lesavka
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

harden stream instrumentation and input/write resilience

Browse Source
This commit is contained in:
Brad Stein 2026-04-08 23:01:31 -03:00
parent b001aa8dd3
commit 6823c8caf6
4 changed files with 81 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
client/src/app.rs
View File

@ -341,7 +341,7 @@ impl LesavkaClientApp {
let max_bitrate = std::env::var("LESAVKA_VIDEO_MAX_KBIT")
.ok()
.and_then(|v| v.parse::<u32>().ok())
.unwrap_or(6_000);
.unwrap_or(4_000);
for monitor_id in 0..=1 {
let ep = ep.clone();
let tx = tx.clone();

1
scripts/install/client.sh
View File

@ -109,4 +109,5 @@ echo " Service: systemctl status lesavka-client --no-pager"
echo
echo "Fish quick start:"
echo " set -gx LESAVKA_SERVER_ADDR http://<server-ip>:50051"
echo " set -gx LESAVKA_VIDEO_MAX_KBIT 4000"
echo " /usr/local/bin/lesavka-client"

1
scripts/install/server.sh
View File

@ -210,6 +210,7 @@ Environment=LESAVKA_UVC_CODEC=mjpeg
Environment=LESAVKA_UVC_EXTERNAL=1
Environment=LESAVKA_EYE_ADAPTIVE=1
Environment=LESAVKA_EYE_MIN_FPS=12
Environment=LESAVKA_EYE_FPS=20
Environment=LESAVKA_MIC_INIT_ATTEMPTS=5
Environment=LESAVKA_MIC_INIT_DELAY_MS=250
Restart=always

101
server/src/main.rs
View File

@ -28,6 +28,7 @@ use lesavka_server::{audio, camera, gadget::UsbGadget, handshake::HandshakeSvc,
/*──────────────── constants ────────────────*/
const VERSION: &str = env!("CARGO_PKG_VERSION");
const PKG_NAME: &str = env!("CARGO_PKG_NAME");
static STREAM_SEQ: AtomicU64 = AtomicU64::new(1);
/*──────────────── logging ───────────────────*/
fn init_tracing() -> anyhow::Result<WorkerGuard> {
@ -168,6 +169,30 @@ async fn open_voice_with_retry(uac_dev: &str) -> anyhow::Result<audio::Voice> {
Err(last_err.unwrap_or_else(|| anyhow::anyhow!("microphone sink init failed")))
}
fn next_stream_id() -> u64 {
STREAM_SEQ.fetch_add(1, Ordering::Relaxed)
}
async fn write_hid_report(dev: &Arc<Mutex<tokio::fs::File>>, data: &[u8]) -> std::io::Result<()> {
let mut last: Option<std::io::Error> = None;
for attempt in 0..5 {
let mut f = dev.lock().await;
match f.write_all(data).await {
Ok(()) => return Ok(()),
Err(e)
if e.kind() == std::io::ErrorKind::WouldBlock
|| e.raw_os_error() == Some(libc::EAGAIN) =>
{
last = Some(e);
}
Err(e) => return Err(e),
}
drop(f);
tokio::time::sleep(Duration::from_millis((attempt as u64 + 1) * 2)).await;
}
Err(last.unwrap_or_else(|| std::io::Error::from_raw_os_error(libc::EAGAIN)))
}
/// Pick the UVC gadget video node.
/// Priority: 1) `LESAVKA_UVC_DEV` override; 2) first `video_output` node.
/// Returns an error when nothing matches instead of guessing a capture card.
@ -438,6 +463,8 @@ impl Relay for Handler {
&self,
req: Request<tonic::Streaming<KeyboardReport>>,
) -> Result<Response<Self::StreamKeyboardStream>, Status> {
let rpc_id = next_stream_id();
info!(rpc_id, "⌨️ stream_keyboard opened");
let (tx, rx) = tokio::sync::mpsc::channel(32);
let kb = self.kb.clone();
let ms = self.ms.clone();
@ -447,19 +474,24 @@ impl Relay for Handler {
tokio::spawn(async move {
let mut s = req.into_inner();
while let Some(pkt) = s.next().await.transpose()? {
if let Err(e) = kb.lock().await.write_all(&pkt.data).await {
warn!("⌨️ write failed: {e} (dropped)");
recover_hid_if_needed(
&e,
gadget.clone(),
kb.clone(),
ms.clone(),
did_cycle.clone(),
)
.await;
if let Err(e) = write_hid_report(&kb, &pkt.data).await {
if e.raw_os_error() == Some(libc::EAGAIN) {
debug!(rpc_id, "⌨️ write would block (dropped)");
} else {
warn!(rpc_id, "⌨️ write failed: {e} (dropped)");
recover_hid_if_needed(
&e,
gadget.clone(),
kb.clone(),
ms.clone(),
did_cycle.clone(),
)
.await;
}
}
tx.send(Ok(pkt)).await.ok();
}
info!(rpc_id, "⌨️ stream_keyboard closed");
Ok::<(), Status>(())
});
@ -470,6 +502,8 @@ impl Relay for Handler {
&self,
req: Request<tonic::Streaming<MouseReport>>,
) -> Result<Response<Self::StreamMouseStream>, Status> {
let rpc_id = next_stream_id();
info!(rpc_id, "🖱️ stream_mouse opened");
let (tx, rx) = tokio::sync::mpsc::channel(1024);
let ms = self.ms.clone();
let kb = self.kb.clone();
@ -479,19 +513,24 @@ impl Relay for Handler {
tokio::spawn(async move {
let mut s = req.into_inner();
while let Some(pkt) = s.next().await.transpose()? {
if let Err(e) = ms.lock().await.write_all(&pkt.data).await {
warn!("🖱️ write failed: {e} (dropped)");
recover_hid_if_needed(
&e,
gadget.clone(),
kb.clone(),
ms.clone(),
did_cycle.clone(),
)
.await;
if let Err(e) = write_hid_report(&ms, &pkt.data).await {
if e.raw_os_error() == Some(libc::EAGAIN) {
debug!(rpc_id, "🖱️ write would block (dropped)");
} else {
warn!(rpc_id, "🖱️ write failed: {e} (dropped)");
recover_hid_if_needed(
&e,
gadget.clone(),
kb.clone(),
ms.clone(),
did_cycle.clone(),
)
.await;
}
}
tx.send(Ok(pkt)).await.ok();
}
info!(rpc_id, "🖱️ stream_mouse closed");
Ok::<(), Status>(())
});
@ -502,6 +541,8 @@ impl Relay for Handler {
&self,
req: Request<tonic::Streaming<AudioPacket>>,
) -> Result<Response<Self::StreamMicrophoneStream>, Status> {
let rpc_id = next_stream_id();
info!(rpc_id, "🎤 stream_microphone opened");
// 1 ─ build once, early
let uac_dev = std::env::var("LESAVKA_UAC_DEV").unwrap_or_else(|_| "hw:UAC2Gadget,0".into());
info!(%uac_dev, "🎤 stream_microphone using UAC sink");
@ -520,12 +561,13 @@ impl Relay for Handler {
while let Some(pkt) = inbound.next().await.transpose()? {
let n = CNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if n < 10 || n % 300 == 0 {
tracing::trace!("🎤⬇ srv pkt#{n} {} bytes", pkt.data.len());
tracing::trace!(rpc_id, "🎤⬇ srv pkt#{n} {} bytes", pkt.data.len());
}
sink.push(&pkt);
}
sink.finish(); // flush on EOS
let _ = tx.send(Ok(Empty {})).await;
info!(rpc_id, "🎤 stream_microphone closed");
Ok::<(), Status>(())
});
@ -536,8 +578,10 @@ impl Relay for Handler {
&self,
req: Request<tonic::Streaming<VideoPacket>>,
) -> Result<Response<Self::StreamCameraStream>, Status> {
let rpc_id = next_stream_id();
let cfg = camera::current_camera_config();
info!(
rpc_id,
output = cfg.output.as_str(),
codec = cfg.codec.as_str(),
width = cfg.width,
@ -549,6 +593,7 @@ impl Relay for Handler {
let (session_id, relay) = self.camera_rt.activate(&cfg).await?;
let camera_rt = self.camera_rt.clone();
info!(rpc_id, session_id, "🎥 stream_camera opened");
// dummy outbound (same pattern as other streams)
let (tx, rx) = tokio::sync::mpsc::channel(1);
@ -557,12 +602,13 @@ impl Relay for Handler {
let mut s = req.into_inner();
while let Some(pkt) = s.next().await.transpose()? {
if !camera_rt.is_active(session_id) {
info!(session_id, "🎥 stream_camera session superseded");
info!(rpc_id, session_id, "🎥 stream_camera session superseded");
break;
}
relay.feed(pkt); // ← all logging inside video.rs
}
tx.send(Ok(Empty {})).await.ok();
info!(rpc_id, session_id, "🎥 stream_camera closed");
Ok::<(), Status>(())
});
@ -573,6 +619,7 @@ impl Relay for Handler {
&self,
req: Request<MonitorRequest>,
) -> Result<Response<Self::CaptureVideoStream>, Status> {
let rpc_id = next_stream_id();
let req = req.into_inner();
let id = req.id;
let dev = match id {
@ -580,7 +627,13 @@ impl Relay for Handler {
1 => "/dev/lesavka_r_eye",
_ => return Err(Status::invalid_argument("monitor id must be 0 or 1")),
};
debug!("🎥 streaming {dev}");
info!(
rpc_id,
id,
max_bitrate = req.max_bitrate,
"🎥 capture_video opened"
);
debug!(rpc_id, "🎥 streaming {dev}");
let s = video::eye_ball(dev, id, req.max_bitrate)
.await
.map_err(|e| Status::internal(format!("{e:#}")))?;
@ -591,10 +644,12 @@ impl Relay for Handler {
&self,
req: Request<MonitorRequest>,
) -> Result<Response<Self::CaptureAudioStream>, Status> {
let rpc_id = next_stream_id();
// Only one speaker stream for now; both 0/1 → same ALSA dev.
let _id = req.into_inner().id;
// Allow override (`LESAVKA_ALSA_DEV=hw:2,0` for debugging).
let dev = std::env::var("LESAVKA_ALSA_DEV").unwrap_or_else(|_| "hw:UAC2Gadget,0".into());
info!(rpc_id, %dev, "🔊 capture_audio opened");
let s = audio::ear(&dev, 0)
.await