mirror of
https://github.com/rustdesk/hbb_common.git
synced 2026-08-27 12:39:50 +00:00
webrtc: split the send budget, make close_webrtc uncancellable, restore log retention
- send_bytes computed one deadline before acquiring the send gate and reused it for the write, so time queued behind another clone's message was charged to this message. A caller that only just lost the gate race got a few milliseconds to write in and then tore the whole peer connection down for missing them - the narrower the miss, the more certain the teardown. The write gets its own budget; the gate wait keeps the one it had, and still never closes (it never held the permit). - Stream::close_webrtc awaited WebRTCStream::close, which is exactly the form close_detached exists to avoid: most callers sit in a select! arm or a future the UI can abandon, and a cancelled close is unretryable (is_closed is latched before the first await, so later attempts early-return and the handler that evicts the session never runs). It is now a plain fn calling close_detached - with no await point there is nothing to cancel. close_detached is public and carries the runtime-teardown guard the parent repo had written separately for its Drop path, so that copy goes away. - Log retention goes back to 31 files. Raising it to 31*8 defended the flood case badly (a file count cannot outrun a flood; only the rate limits at the log sites can) while silently multiplying steady-state retention and disk for every ordinary install, on every platform. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ExUfAkYbq8UC9pQCiLy8TQ
This commit is contained in:
+7
-7
@@ -464,13 +464,13 @@ pub fn init_log(_is_async: bool, _name: &str) -> Option<flexi_logger::LoggerHand
|
|||||||
// much disk this uses. Bounding it here covers every call site at once.
|
// much disk this uses. Bounding it here covers every call site at once.
|
||||||
Criterion::AgeOrSize(Age::Day, 16 * 1024 * 1024),
|
Criterion::AgeOrSize(Age::Day, 16 * 1024 * 1024),
|
||||||
Naming::Timestamps,
|
Naming::Timestamps,
|
||||||
// Retention is a file count, so adding the size criterion also shortened
|
// Unchanged at 31, which is ~31 days for any machine that stays under the
|
||||||
// the history a flood can leave behind: the same actor that motivates the
|
// size criterion — i.e. every ordinary install, on every platform this
|
||||||
// size cap can now force rotations until every file predating its own
|
// ships to. Raising it to protect the flood case would have bought little
|
||||||
// activity is cleaned up. Keep enough files that ~31 days survives even
|
// (a count cannot outrun a flood; only the rate limits at the log sites
|
||||||
// when every one of them is a full 16 MiB — trading a bounded amount of
|
// can) at the price of multiplying steady-state retention and disk for
|
||||||
// disk for not handing an attacker a log-erasure primitive.
|
// everyone.
|
||||||
Cleanup::KeepLogFiles(31 * 8),
|
Cleanup::KeepLogFiles(31),
|
||||||
)
|
)
|
||||||
.start()
|
.start()
|
||||||
.ok();
|
.ok();
|
||||||
|
|||||||
+8
-2
@@ -90,11 +90,17 @@ impl Stream {
|
|||||||
/// A WebRTC pc is kept alive by the global session cache and its cleanup handler only fires on
|
/// A WebRTC pc is kept alive by the global session cache and its cleanup handler only fires on
|
||||||
/// a terminal ICE state, so it must be closed explicitly at session end. TCP/WebSocket streams
|
/// a terminal ICE state, so it must be closed explicitly at session end. TCP/WebSocket streams
|
||||||
/// release their resources on drop and need nothing here.
|
/// release their resources on drop and need nothing here.
|
||||||
|
///
|
||||||
|
/// Deliberately not `async`: most callers sit in a `select!` arm or a future the UI can
|
||||||
|
/// abandon, and an awaited close that loses that race is unretryable — `close()` latches
|
||||||
|
/// `is_closed` before its first await, so every later attempt early-returns while the state
|
||||||
|
/// handler that would evict the session never runs. With no await point here there is
|
||||||
|
/// nothing to cancel; the teardown finishes on the runtime.
|
||||||
#[inline]
|
#[inline]
|
||||||
pub async fn close_webrtc(&self) {
|
pub fn close_webrtc(&self) {
|
||||||
match self {
|
match self {
|
||||||
#[cfg(feature = "webrtc")]
|
#[cfg(feature = "webrtc")]
|
||||||
Stream::WebRTC(s) => s.close().await,
|
Stream::WebRTC(s) => s.close_detached(),
|
||||||
#[allow(unreachable_patterns)]
|
#[allow(unreachable_patterns)]
|
||||||
_ => {}
|
_ => {}
|
||||||
}
|
}
|
||||||
|
|||||||
+30
-11
@@ -928,18 +928,31 @@ impl WebRTCStream {
|
|||||||
/// `keep` carries anything whose lifetime must span the teardown rather than the caller's:
|
/// `keep` carries anything whose lifetime must span the teardown rather than the caller's:
|
||||||
/// the send path passes its logical-message permit, so no waiting clone can append to a
|
/// the send path passes its logical-message permit, so no waiting clone can append to a
|
||||||
/// partially-written fragment sequence while the close is still in flight.
|
/// partially-written fragment sequence while the close is still in flight.
|
||||||
fn close_detached_with<T: Send + 'static>(&self, keep: T) {
|
pub fn close_detached_with<T: Send + 'static>(&self, keep: T) {
|
||||||
let pc = self.pc.clone();
|
let pc = self.pc.clone();
|
||||||
tokio::spawn(async move {
|
// Take the runtime handle explicitly rather than calling `tokio::spawn`: this also runs
|
||||||
let _keep = keep;
|
// from `Drop` impls, which can execute during runtime teardown where a bare spawn panics.
|
||||||
if let Err(err) = pc.close().await {
|
// `Handle::spawn` can panic while the runtime is shutting down too, so catch it — a brief
|
||||||
log::debug!("WebRTC background close failed: {}", err);
|
// leak until process exit beats aborting the process from a destructor.
|
||||||
}
|
let Ok(handle) = tokio::runtime::Handle::try_current() else {
|
||||||
});
|
log::warn!("no tokio runtime available to close the WebRTC peer connection");
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let spawned = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||||
|
handle.spawn(async move {
|
||||||
|
let _keep = keep;
|
||||||
|
if let Err(err) = pc.close().await {
|
||||||
|
log::debug!("WebRTC background close failed: {}", err);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}));
|
||||||
|
if spawned.is_err() {
|
||||||
|
log::warn!("failed to spawn the WebRTC close (runtime shutting down)");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[inline]
|
#[inline]
|
||||||
fn close_detached(&self) {
|
pub fn close_detached(&self) {
|
||||||
self.close_detached_with(());
|
self.close_detached_with(());
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1041,8 +1054,8 @@ impl WebRTCStream {
|
|||||||
.await
|
.await
|
||||||
.map_err(|_| Error::new(ErrorKind::TimedOut, "WebRTC connect timeout"))??;
|
.map_err(|_| Error::new(ErrorKind::TimedOut, "WebRTC connect timeout"))??;
|
||||||
|
|
||||||
let deadline = Instant::now() + Duration::from_millis(send_timeout);
|
let gate_deadline = Instant::now() + Duration::from_millis(send_timeout);
|
||||||
let send_permit = match timeout_at(deadline, send_gate.acquire_owned()).await {
|
let send_permit = match timeout_at(gate_deadline, send_gate.acquire_owned()).await {
|
||||||
Ok(Ok(permit)) => permit,
|
Ok(Ok(permit)) => permit,
|
||||||
Ok(Err(err)) => {
|
Ok(Err(err)) => {
|
||||||
return Err(Error::new(
|
return Err(Error::new(
|
||||||
@@ -1059,7 +1072,13 @@ impl WebRTCStream {
|
|||||||
return Err(Error::new(ErrorKind::TimedOut, "WebRTC send gate timeout").into());
|
return Err(Error::new(ErrorKind::TimedOut, "WebRTC send gate timeout").into());
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
match timeout_at(deadline, self.send_bytes_inner(bytes)).await {
|
// Fresh budget once the permit is held: queueing behind another clone's message is
|
||||||
|
// not this message's progress, so charging it here meant a caller that only just lost
|
||||||
|
// the gate race got a few milliseconds to write in — and then tore the whole peer
|
||||||
|
// connection down for missing them. The narrower the miss, the more certain the
|
||||||
|
// teardown, which is precisely backwards.
|
||||||
|
let write_deadline = Instant::now() + Duration::from_millis(send_timeout);
|
||||||
|
match timeout_at(write_deadline, self.send_bytes_inner(bytes)).await {
|
||||||
Ok(res) => res,
|
Ok(res) => res,
|
||||||
Err(_) => {
|
Err(_) => {
|
||||||
// Hand the logical-message permit to the teardown so no waiting clone can
|
// Hand the logical-message permit to the teardown so no waiting clone can
|
||||||
|
|||||||
Reference in New Issue
Block a user