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 78bb3c4..bae29d0 100644 --- a/server/crates/arbiter-server/src/peers/operator/session/mod.rs +++ b/server/crates/arbiter-server/src/peers/operator/session/mod.rs @@ -1,10 +1,8 @@ use super::{OutOfBand, OperatorConnection}; use crate::{ actors::{ - flow_coordinator::{GetConnectedClientIds, client_connect_approval::ClientApprovalController}, - operator_registry::ConnectOperator, - }, - peers::client::ClientProfile, + flow_coordinator::{GetConnectedClientIds, client_connect_approval::{ClientApprovalAnswer, ClientApprovalController}}, operator_registry::ConnectOperator, + }, peers::client::ClientProfile, }; use arbiter_crypto::authn; use arbiter_proto::transport::Sender; @@ -93,6 +91,7 @@ impl OperatorSession { actor = "operator", event = "failed to announce new client connection" ); + let _ = controller.tell(ClientApprovalAnswer { approved: false }).await; return; }