use std::time::Duration;
use gate4agent_node_protocol::{
read_json_frame_limited_body_timeout, write_json_frame_limited, FrameError,
NodeCallHomeAnnounce, NodeId, BUILD_STAMP, MAX_NODE_HELLO_FRAME_BYTES,
};
use tokio::io::{AsyncRead, AsyncWrite};
const ANNOUNCE_TIMEOUT_MS: u64 = 2_000;
#[derive(Debug)]
pub enum CallHomeAnnounceError {
TimedOut,
Frame(FrameError),
BuildStamp { announced: String },
InvalidNodeId,
}
impl std::fmt::Display for CallHomeAnnounceError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::TimedOut => write!(formatter, "call-home peer did not announce itself in time"),
Self::Frame(error) => write!(formatter, "call-home announce frame is invalid: {error}"),
Self::BuildStamp { announced } => write!(
formatter,
"build stamp mismatch: local={BUILD_STAMP} remote={announced}",
),
Self::InvalidNodeId => write!(formatter, "call-home announce carried an invalid node id"),
}
}
}
impl std::error::Error for CallHomeAnnounceError {}
pub async fn write_call_home_announce<W>(
writer: &mut W,
node_id: &NodeId,
) -> Result<(), FrameError>
where
W: AsyncWrite + Unpin,
{
write_json_frame_limited(
writer,
&NodeCallHomeAnnounce::new(node_id.as_str()),
MAX_NODE_HELLO_FRAME_BYTES,
)
.await
}
pub async fn read_call_home_announce<R>(reader: &mut R) -> Result<NodeId, CallHomeAnnounceError>
where
R: AsyncRead + Unpin,
{
let announce: NodeCallHomeAnnounce = tokio::time::timeout(
Duration::from_millis(ANNOUNCE_TIMEOUT_MS),
read_json_frame_limited_body_timeout(
reader,
MAX_NODE_HELLO_FRAME_BYTES,
Duration::from_millis(ANNOUNCE_TIMEOUT_MS),
),
)
.await
.map_err(|_| CallHomeAnnounceError::TimedOut)?
.map_err(CallHomeAnnounceError::Frame)?;
if announce.build_stamp != BUILD_STAMP {
return Err(CallHomeAnnounceError::BuildStamp {
announced: announce.build_stamp,
});
}
NodeId::new(announce.node_id).map_err(|_| CallHomeAnnounceError::InvalidNodeId)
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn a_node_names_itself_and_the_relay_reads_the_name() {
let node_id = NodeId::new("fixture-node").unwrap();
let mut wire = Vec::new();
write_call_home_announce(&mut wire, &node_id).await.unwrap();
let read = read_call_home_announce(&mut wire.as_slice()).await.unwrap();
assert_eq!(read, node_id);
}
#[tokio::test]
async fn a_build_stamp_mismatch_names_itself_rather_than_looking_like_garbage() {
let foreign_stamp = "f".repeat(BUILD_STAMP.len());
let mut wire = Vec::new();
write_json_frame_limited(
&mut wire,
&NodeCallHomeAnnounce {
build_stamp: foreign_stamp.clone(),
node_id: "fixture-node".to_owned(),
},
MAX_NODE_HELLO_FRAME_BYTES,
)
.await
.unwrap();
let error = read_call_home_announce(&mut wire.as_slice()).await.unwrap_err();
assert!(
matches!(&error, CallHomeAnnounceError::BuildStamp { announced } if *announced == foreign_stamp),
"expected a named build stamp mismatch, got {error:?}",
);
assert_eq!(
error.to_string(),
format!("build stamp mismatch: local={BUILD_STAMP} remote={foreign_stamp}"),
);
}
#[tokio::test]
async fn a_socket_that_says_nothing_is_dropped_rather_than_held() {
let (client, mut server) = tokio::io::duplex(64);
let error = read_call_home_announce(&mut server).await.unwrap_err();
drop(client);
assert!(
matches!(error, CallHomeAnnounceError::TimedOut),
"expected the announce deadline to fire, got {error:?}",
);
}
}