From b843105533b8253b0401cef53088e8ea4f87fd46 Mon Sep 17 00:00:00 2001 From: CleverWild Date: Thu, 18 Jun 2026 16:29:43 +0200 Subject: [PATCH] fix(user-agent): zombie sessions #74 --- .../client_connect_approval.rs | 22 ++++++++++++++++++- .../src/peers/operator/session/mod.rs | 3 ++- 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/server/crates/arbiter-server/src/actors/flow_coordinator/client_connect_approval.rs b/server/crates/arbiter-server/src/actors/flow_coordinator/client_connect_approval.rs index e33cff2..31eecc3 100644 --- a/server/crates/arbiter-server/src/actors/flow_coordinator/client_connect_approval.rs +++ b/server/crates/arbiter-server/src/actors/flow_coordinator/client_connect_approval.rs @@ -11,7 +11,9 @@ use kameo::{ prelude::{ActorId, ActorRef, ActorStopReason, Context, WeakActorRef}, reply::ReplySender, }; -use std::ops::ControlFlow; +use std::{ops::ControlFlow, time::Duration}; + +const APPROVAL_TIMEOUT: Duration = Duration::from_secs(30); pub struct Args { pub client: ClientProfile, @@ -64,6 +66,14 @@ impl Actor for ClientApprovalController { .await; } + let weak = actor_ref.downgrade(); + tokio::spawn(async move { + tokio::time::sleep(APPROVAL_TIMEOUT).await; + if let Some(r) = weak.upgrade() { + let _ = r.tell(OnApprovalTimeout {}).await; + } + }); + Ok(this) } @@ -104,4 +114,14 @@ impl ClientApprovalController { ctx.stop(); } } + + /// Fired after `APPROVAL_TIMEOUT` elapses. Any operator that hasn't responded + /// by then is treated as a denial to prevent zombie sessions from blocking the flow. + #[message(ctx)] + pub fn on_approval_timeout(&mut self, ctx: &mut Context) { + if self.pending > 0 { + self.send_reply(Ok(false)); + ctx.stop(); + } + } } diff --git a/server/crates/arbiter-server/src/peers/operator/session/mod.rs b/server/crates/arbiter-server/src/peers/operator/session/mod.rs index 0fe2c84..d7566ac 100644 --- a/server/crates/arbiter-server/src/peers/operator/session/mod.rs +++ b/server/crates/arbiter-server/src/peers/operator/session/mod.rs @@ -1,7 +1,7 @@ use super::{OutOfBand, OperatorConnection}; use crate::{ actors::{ - flow_coordinator::client_connect_approval::ClientApprovalController, + flow_coordinator::client_connect_approval::{ClientApprovalAnswer, ClientApprovalController}, operator_registry::ConnectOperator, }, peers::client::ClientProfile, @@ -88,6 +88,7 @@ impl OperatorSession { actor = "operator", event = "failed to announce new client connection" ); + let _ = controller.tell(ClientApprovalAnswer { approved: false }).await; return; } -- 2.49.1