From 4ee0c0735a91bd7021b6730e55f4d7de6096f462 Mon Sep 17 00:00:00 2001 From: LIm1tZz7 <36764786+TheBadRoger@users.noreply.github.com> Date: Mon, 12 Oct 2026 22:09:22 +0800 Subject: [PATCH] =?UTF-8?q?=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 22 +++++++++++++++++++ src/lib.rs | 7 ++++-- src/main.rs | 63 ++++++++++++++++++++++++++++++++++++++++++----------- 3 files changed, 77 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index de6172d..03ca34b 100644 --- a/README.md +++ b/README.md @@ -74,3 +74,25 @@ cargo fmt --check ## 旧部署升级 现在只开放 UDP 7001；不再需要 TCP7000、`--udp-listen`、旧 JSON probe/签名确认或 `/chatmask/punch/1.0.0`。请同时升级客户端与服务器。旧密钥仍可复用，旧中继、旧服务和容器需先停止以避免端口冲突；旧数据卷不自动删除。 + +## 活动日志 + +服务端记录带 Unix 时间戳的连接、断开、节点登记、节点查询、失败原因、打洞通知发送与客户端确认、通知失败和登记过期。记录 PeerId、IP 和端口，不记录聊天正文或密钥。日志写入标准输出，由 Docker 保存和轮转。 + +查看最近日志： +```sh +docker compose logs --timestamps --tail=100 server +``` +实时跟随输出（Ctrl+C 只停止查看日志）： +```sh +docker compose logs --timestamps --tail=100 -f server +``` +前台启动显示实时输出： +```sh +docker compose up --build +``` +更新源码后重建：`docker compose up -d --build`。 + +`punch_coordination_dispatched` 表示双方通知已提交，`punch_notice_response ... Ack` 表示客户端确认收到通知。服务器无法观察双方后续直连，不能据此宣布打洞或聊天成功，最终结果以客户端为准。 + +新版客户端使用 `/join 公网IP或域名` 一次添加服务器，之后每次启动自动连接和登记；`/peerid` 只显示本机 PeerId，双方用 `/chat 对方PeerId` 聊天。 diff --git a/src/lib.rs b/src/lib.rs index b9907c8..5ba43d1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -75,9 +75,11 @@ impl Registry { pub fn remove(&mut self, peer: PeerId) { self.clients.remove(&peer); } - pub fn expire(&mut self) { + pub fn expire(&mut self) -> usize { + let before = self.clients.len(); let ttl = self.ttl; self.clients.retain(|_, r| r.seen.elapsed() < ttl); + before - self.clients.len() } pub fn register(&mut self, peer: PeerId, endpoint: SocketAddr) -> Response { self.expire(); @@ -153,9 +155,10 @@ pub fn socket_address(address: &Multiaddr) -> Option { } } #[derive(NetworkBehaviour)] +// Check connection limits before notifying protocol behaviours about a new connection. pub struct Behaviour { - pub control: request_response::cbor::Behaviour, limits: connection_limits::Behaviour, + pub control: request_response::cbor::Behaviour, } pub fn build(args: &Args, key: identity::Keypair) -> Result> { let mut swarm = SwarmBuilder::with_existing_identity(key) diff --git a/src/main.rs b/src/main.rs index 7686e7e..e0251c2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,32 +13,62 @@ async fn main() -> Result<()> { let mut registry = Registry::new(args.max_clients, args.ttl_seconds); let mut tick = tokio::time::interval(std::time::Duration::from_secs(1)); let mut endpoints = HashMap::new(); + let mut notices = HashMap::new(); let shutdown = shutdown(); tokio::pin!(shutdown); println!("Server PeerId: {peer}"); + log(format_args!( + "server_started peer={peer} max_clients={} ttl_seconds={}", + args.max_clients, args.ttl_seconds + )); loop { tokio::select! { - result = &mut shutdown => { result?; println!("Shutting down"); break; }, - _ = tick.tick() => registry.expire(), + result = &mut shutdown => { result?; log(format_args!("server_shutdown")); break; }, + _ = tick.tick() => { let count = registry.expire(); if count > 0 { log(format_args!("registrations_expired count={count}")); } }, event = swarm.select_next_some() => match event { SwarmEvent::NewListenAddr { address, .. } => println!("Control address: {}", address.with(Protocol::P2p(peer))), - SwarmEvent::ConnectionEstablished { peer_id, endpoint, .. } => { if let Some(address) = chatmask_server::socket_address(endpoint.get_remote_address()) { endpoints.insert(peer_id, address); } }, - SwarmEvent::ConnectionClosed { peer_id, .. } if !swarm.is_connected(&peer_id) => { registry.remove(peer_id); endpoints.remove(&peer_id); }, + SwarmEvent::ConnectionEstablished { peer_id, endpoint, .. } => { + if let Some(address) = chatmask_server::socket_address(endpoint.get_remote_address()) { + endpoints.insert(peer_id, address); + log(format_args!("client_connected peer={peer_id} endpoint={address}")); + } + }, + SwarmEvent::ConnectionClosed { peer_id, cause, .. } if !swarm.is_connected(&peer_id) => { + registry.remove(peer_id); endpoints.remove(&peer_id); + log(format_args!("client_disconnected peer={peer_id} cause={cause:?}")); + }, SwarmEvent::Behaviour(BehaviourEvent::Control(request_response::Event::Message { peer, message: request_response::Message::Request { request, channel, .. }, .. })) => { let response = match request { - Request::Register => match endpoints.get(&peer) { Some(endpoint) => registry.register(peer, *endpoint), None => Response::Error { message: "QUIC connection required".into() } }, - Request::Connect { peer: target } => match target.parse().map_err(|_| "Invalid target".into()).and_then(|target| registry.connect(peer, target).map(|pair| (target, pair))) { - Ok((target, (a, b, attempt))) => { - swarm.behaviour_mut().control.send_request(&target, Request::Punch { peer: peer.to_string(), endpoint: a, attempt: attempt.clone(), timeout_ms: 5000 }); - swarm.behaviour_mut().control.send_request(&peer, Request::Punch { peer: target.to_string(), endpoint: b, attempt: attempt.clone(), timeout_ms: 5000 }); - Response::Started { attempt } - }, - Err(error) => Response::Error { message: error.to_string() }, + Request::Register => { + let response = match endpoints.get(&peer) { Some(endpoint) => registry.register(peer, *endpoint), None => Response::Error { message: "QUIC connection required".into() } }; + log(format_args!("registration peer={peer} response={response:?}")); response + }, + Request::Connect { peer: target } => { + log(format_args!("node_query source={peer} target={target:?}")); + if notices.len() >= 2048 { Response::Error { message: "Coordination busy".into() } } + else { match target.parse().map_err(|_| "Invalid target".into()).and_then(|target| registry.connect(peer, target).map(|pair| (target, pair))) { + Ok((target, (a, b, attempt))) => { + let to_target = swarm.behaviour_mut().control.send_request(&target, Request::Punch { peer: peer.to_string(), endpoint: a, attempt: attempt.clone(), timeout_ms: 5000 }); + let to_source = swarm.behaviour_mut().control.send_request(&peer, Request::Punch { peer: target.to_string(), endpoint: b, attempt: attempt.clone(), timeout_ms: 5000 }); + notices.insert(to_target, (attempt.clone(), target)); notices.insert(to_source, (attempt.clone(), peer)); + log(format_args!("punch_coordination_dispatched source={peer} target={target} source_endpoint={a} target_endpoint={b} attempt={attempt}")); + Response::Started { attempt } + }, + Err(error) => { log(format_args!("node_query_failed source={peer} target={target:?} reason={error:?}")); Response::Error { message: error.to_string() } }, + } } }, Request::Punch { .. } => Response::Error { message: "Server-only request".into() }, }; - let _ = swarm.behaviour_mut().control.send_response(channel, response); + if swarm.behaviour_mut().control.send_response(channel, response).is_err() { log(format_args!("response_delivery_failed peer={peer}")); } }, + SwarmEvent::Behaviour(BehaviourEvent::Control(request_response::Event::Message { peer, message: request_response::Message::Response { request_id, response }, .. })) => { + if let Some((attempt, expected)) = notices.remove(&request_id) { log(format_args!("punch_notice_response peer={peer} expected={expected} attempt={attempt} response={response:?}")); } + }, + SwarmEvent::Behaviour(BehaviourEvent::Control(request_response::Event::OutboundFailure { peer, request_id, error, .. })) => { + if let Some((attempt, _)) = notices.remove(&request_id) { log(format_args!("punch_notice_failed peer={peer} attempt={attempt} reason={error:?}")); } + }, + SwarmEvent::Behaviour(BehaviourEvent::Control(request_response::Event::InboundFailure { peer, error, .. })) => log(format_args!("control_request_failed peer={peer} reason={error:?}")), + SwarmEvent::IncomingConnectionError { error, .. } => log(format_args!("incoming_connection_failed reason={error:?}")), SwarmEvent::ListenerError { error, .. } => return Err(error.into()), SwarmEvent::ListenerClosed { reason, .. } => { reason?; return Err("Listener closed".into()); }, _ => {}, @@ -47,6 +77,13 @@ async fn main() -> Result<()> { } Ok(()) } +fn log(message: std::fmt::Arguments<'_>) { + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + println!("[unix={timestamp}] {message}"); +} async fn shutdown() -> std::io::Result<()> { #[cfg(unix)] {