Compare commits
4 Commits
022003ac5e
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 357726bc5d | |||
|
|
23827c613e | ||
|
|
8381b68a52 | ||
|
|
b843105533 |
@@ -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<Self, ()>) {
|
||||
if self.pending > 0 {
|
||||
self.send_reply(Ok(false));
|
||||
ctx.stop();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -266,7 +266,7 @@ pub async fn metadata_unchanged_does_not_append_history() {
|
||||
|
||||
#[tokio::test]
|
||||
#[test_log::test]
|
||||
pub async fn metadata_change_appends_history_and_repoints_binding() {
|
||||
pub async fn metadata_frozen_after_approval_ignores_reconnect_changes() {
|
||||
let db = db::create_test_pool().await;
|
||||
let actors = spawn_test_actors(&db).await;
|
||||
let new_key = MlDsa87::key_gen(&mut rand::rng());
|
||||
@@ -287,6 +287,7 @@ pub async fn metadata_change_appends_history_and_repoints_binding() {
|
||||
connect_client(props, &mut server_transport).await;
|
||||
});
|
||||
|
||||
// Reconnect presenting different metadata — must be silently ignored.
|
||||
test_transport
|
||||
.send(auth::Inbound::AuthChallengeRequest {
|
||||
pubkey: verifying_key(&new_key).into(),
|
||||
@@ -313,6 +314,7 @@ pub async fn metadata_change_appends_history_and_repoints_binding() {
|
||||
client_metadata, client_metadata_history, program_client,
|
||||
};
|
||||
let mut conn = db.get().await.unwrap();
|
||||
// Metadata is frozen: no new row, no history entry.
|
||||
let metadata_count: i64 = client_metadata::table
|
||||
.count()
|
||||
.get_result(&mut conn)
|
||||
@@ -338,15 +340,16 @@ pub async fn metadata_change_appends_history_and_repoints_binding() {
|
||||
.first::<(String, Option<String>, Option<String>)>(&mut conn)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(metadata_count, 2);
|
||||
assert_eq!(history_count, 1);
|
||||
assert_eq!(metadata_count, 1, "frozen: no new metadata row on reconnect");
|
||||
assert_eq!(history_count, 0, "frozen: no history entry on reconnect");
|
||||
assert_eq!(
|
||||
current,
|
||||
(
|
||||
"client".to_owned(),
|
||||
Some("new".to_owned()),
|
||||
Some("2.0.0".to_owned())
|
||||
)
|
||||
Some("old".to_owned()),
|
||||
Some("1.0.0".to_owned())
|
||||
),
|
||||
"frozen: original metadata must be preserved"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user