This commit is contained in:
LIm1tZz7 committed 2026-10-12 22:09:22 +08:00
1 parent 328aad2b00
commit 4ee0c0735a
3 files changed
+77 -15

No files matched your search

+22
View File
@@ -74,3 +74,25 @@ cargo fmt --check
## 旧部署升级 ## 旧部署升级
现在只开放 UDP 7001;不再需要 TCP7000、`--udp-listen`、旧 JSON probe/签名确认或 `/chatmask/punch/1.0.0`。请同时升级客户端与服务器。旧密钥仍可复用,旧中继、旧服务和容器需先停止以避免端口冲突;旧数据卷不自动删除。 现在只开放 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` 聊天。
+5 -2
View File
@@ -75,9 +75,11 @@ impl Registry {
pub fn remove(&mut self, peer: PeerId) { pub fn remove(&mut self, peer: PeerId) {
self.clients.remove(&peer); self.clients.remove(&peer);
} }
pub fn expire(&mut self) { pub fn expire(&mut self) -> usize {
let before = self.clients.len();
let ttl = self.ttl; let ttl = self.ttl;
self.clients.retain(|_, r| r.seen.elapsed() < ttl); self.clients.retain(|_, r| r.seen.elapsed() < ttl);
before - self.clients.len()
} }
pub fn register(&mut self, peer: PeerId, endpoint: SocketAddr) -> Response { pub fn register(&mut self, peer: PeerId, endpoint: SocketAddr) -> Response {
self.expire(); self.expire();
@@ -153,9 +155,10 @@ pub fn socket_address(address: &Multiaddr) -> Option<SocketAddr> {
} }
} }
#[derive(NetworkBehaviour)] #[derive(NetworkBehaviour)]
// Check connection limits before notifying protocol behaviours about a new connection.
pub struct Behaviour { pub struct Behaviour {
pub control: request_response::cbor::Behaviour<Request, Response>,
limits: connection_limits::Behaviour, limits: connection_limits::Behaviour,
pub control: request_response::cbor::Behaviour<Request, Response>,
} }
pub fn build(args: &Args, key: identity::Keypair) -> Result<Swarm<Behaviour>> { pub fn build(args: &Args, key: identity::Keypair) -> Result<Swarm<Behaviour>> {
let mut swarm = SwarmBuilder::with_existing_identity(key) let mut swarm = SwarmBuilder::with_existing_identity(key)
+50 -13
View File
@@ -13,32 +13,62 @@ async fn main() -> Result<()> {
let mut registry = Registry::new(args.max_clients, args.ttl_seconds); 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 tick = tokio::time::interval(std::time::Duration::from_secs(1));
let mut endpoints = HashMap::new(); let mut endpoints = HashMap::new();
let mut notices = HashMap::new();
let shutdown = shutdown(); let shutdown = shutdown();
tokio::pin!(shutdown); tokio::pin!(shutdown);
println!("Server PeerId: {peer}"); println!("Server PeerId: {peer}");
log(format_args!(
"server_started peer={peer} max_clients={} ttl_seconds={}",
args.max_clients, args.ttl_seconds
));
loop { loop {
tokio::select! { tokio::select! {
result = &mut shutdown => { result?; println!("Shutting down"); break; }, result = &mut shutdown => { result?; log(format_args!("server_shutdown")); break; },
_ = tick.tick() => registry.expire(), _ = tick.tick() => { let count = registry.expire(); if count > 0 { log(format_args!("registrations_expired count={count}")); } },
event = swarm.select_next_some() => match event { event = swarm.select_next_some() => match event {
SwarmEvent::NewListenAddr { address, .. } => println!("Control address: {}", address.with(Protocol::P2p(peer))), 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::ConnectionEstablished { peer_id, endpoint, .. } => {
SwarmEvent::ConnectionClosed { peer_id, .. } if !swarm.is_connected(&peer_id) => { registry.remove(peer_id); endpoints.remove(&peer_id); }, 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, .. }, .. })) => { SwarmEvent::Behaviour(BehaviourEvent::Control(request_response::Event::Message { peer, message: request_response::Message::Request { request, channel, .. }, .. })) => {
let response = match request { 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::Register => {
Request::Connect { peer: target } => match target.parse().map_err(|_| "Invalid target".into()).and_then(|target| registry.connect(peer, target).map(|pair| (target, pair))) { let response = match endpoints.get(&peer) { Some(endpoint) => registry.register(peer, *endpoint), None => Response::Error { message: "QUIC connection required".into() } };
Ok((target, (a, b, attempt))) => { log(format_args!("registration peer={peer} response={response:?}")); response
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 }); Request::Connect { peer: target } => {
Response::Started { attempt } log(format_args!("node_query source={peer} target={target:?}"));
}, if notices.len() >= 2048 { Response::Error { message: "Coordination busy".into() } }
Err(error) => Response::Error { message: error.to_string() }, 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() }, 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::ListenerError { error, .. } => return Err(error.into()),
SwarmEvent::ListenerClosed { reason, .. } => { reason?; return Err("Listener closed".into()); }, SwarmEvent::ListenerClosed { reason, .. } => { reason?; return Err("Listener closed".into()); },
_ => {}, _ => {},
@@ -47,6 +77,13 @@ async fn main() -> Result<()> {
} }
Ok(()) 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<()> { async fn shutdown() -> std::io::Result<()> {
#[cfg(unix)] #[cfg(unix)]
{ {