diff --git a/README.client.md b/README.client.md index 214d835..3534fa6 100644 --- a/README.client.md +++ b/README.client.md @@ -12,6 +12,7 @@ - [前置条件](#前置条件) - [rtr_debug_client](#rtr_debug_client) + - [Docker attach 交互命令](#docker-attach-交互命令) - [TCP](#tcp) - [TLS/mTLS](#tlsmtls) - [SSH](#ssh) @@ -54,7 +55,55 @@ Client 连接地址不要求固定在某个 Docker 网络里。`deploy/client` `rtr_debug_client` 是本仓库提供的调试客户端。它会发起 RTR Reset Query 或 Serial Query，并打印 server 返回的 PDU 摘要。Docker 配置默认执行： ```text - reset --keep-after-error --summary-only + reset --summary-only +``` + +Docker 镜像的 `entrypoint.sh` 会把 stdout/stderr 重定向到 `logs/client/*.stdout.log` 和 `logs/client/*.stderr.log`。因此 `docker attach` 后可以输入交互命令，但命令响应和收到的 PDU 仍然主要从日志文件查看： + +```bash +tail -f logs/client/*.stdout.log +tail -f logs/client/*.stderr.log +``` + +### Docker attach 交互命令 + +如果 debug client 已经通过 compose 启动，可以 attach 到容器后输入命令： + +```bash +docker attach --sig-proxy=false +``` + +脱离 attach 会话时使用 `Ctrl-P` 后接 `Ctrl-Q`，避免 `Ctrl-C` 直接停止客户端进程。 + +可用交互命令如下： + +| 命令 | 作用 | +| --- | --- | +| `help` | 显示可用命令列表。 | +| `state` | 打印当前客户端状态，包括 `session_id`、`serial`、`refresh/retry/expire`、输出模式、自动轮询状态等。 | +| `version` | 查看当前 RTR 协议版本。 | +| `version ` | 修改后续查询使用的 RTR 协议版本，例如 `version 2`。 | +| `reset` | 发送 Reset Query，通常用于让 server 重新下发完整数据集。 | +| `serial` | 使用当前已记录的 `session_id` 和 `serial` 发送 Serial Query。需要客户端已经从 EndOfData 中拿到状态。 | +| `serial ` | 显式指定参数发送 Serial Query，例如 `serial 31320 156`。 | +| `timeout` | 查看当前读取 PDU 的超时时间。 | +| `timeout ` | 修改读取 PDU 的超时时间。超时只打印提示，连接保持打开。 | +| `poll` | 查看当前自动轮询间隔、来源、refresh hint、错误状态和暂停状态。 | +| `poll ` | 修改自动轮询间隔。 | +| `poll pause` | 暂停自动轮询。 | +| `poll resume` | 恢复自动轮询。 | +| `output` | 查看当前输出模式，以及 summary 模式下跳过的 payload PDU 数量。 | +| `output verbose` | 打印所有收到的 PDU，包括 IPv4 Prefix、IPv6 Prefix、Router Key、ASPA 等 payload PDU。 | +| `output summary` | 只打印摘要，跳过 payload PDU 详情。compose 默认使用该模式。 | +| `quit` / `exit` | 退出客户端进程。 | + +常用调试流程： + +```text +output verbose +reset +state +serial ``` ### TCP diff --git a/README.md b/README.md index 0b40f1f..5735262 100644 --- a/README.md +++ b/README.md @@ -32,8 +32,8 @@ ## 协议参考 -- RTR: [RFC 6810](https://www.rfc-editor.org/rfc/rfc6810.html), [RFC 8210](https://www.rfc-editor.org/rfc/rfc8210.html) -- SLURM: [RFC 8416](https://www.rfc-editor.org/rfc/rfc8416.html) +- RTR: [RFC 6810](https://www.rfc-editor.org/rfc/rfc6810.html), [RFC 8210](https://www.rfc-editor.org/rfc/rfc8210.html), [draft-ietf-sidrops-8210bis-26](https://www.ietf.org/archive/id/draft-ietf-sidrops-8210bis-26.html) +- SLURM: [RFC 8416](https://www.rfc-editor.org/rfc/rfc8416.html), [draft-ietf-sidrops-aspa-slurm-04](https://www.ietf.org/archive/id/draft-ietf-sidrops-aspa-slurm-04.html)（已过期，Expired 2026-05-20） - CCR: [draft-ietf-sidrops-rpki-ccr](https://www.ietf.org/archive/id/draft-ietf-sidrops-rpki-ccr-02.html) ## 项目结构 diff --git a/specs/11_rtr.md b/specs/11_rtr.md index 9bb3eff..428ffaf 100644 --- a/specs/11_rtr.md +++ b/specs/11_rtr.md @@ -7,7 +7,7 @@ RTR 用于把 RP/Cache 已完成密码学验证的 RPKI 数据下发给路由器 本文按以下规范整理： - RFC 6810（RTR v0） - RFC 8210（RTR v1，更新 RFC 6810） -- draft-ietf-sidrops-8210bis-25（RTR v2，草案） +- draft-ietf-sidrops-8210bis-26（RTR v2，草案） ## 11.2 协议演进 @@ -35,7 +35,7 @@ RTR 用于把 RP/Cache 已完成密码学验证的 RPKI 数据下发给路由器 PDU 类型（按规范注册表）： -| PDU Type | 名称 | v0 (RFC6810) | v1 (RFC8210) | v2 (8210bis-25) | +| PDU Type | 名称 | v0 (RFC6810) | v1 (RFC8210) | v2 (8210bis-26) | |---|---|---|---|---| | 0 | Serial Notify | 支持 | 支持 | 支持 | | 1 | Serial Query | 支持 | 支持 | 支持 | @@ -105,6 +105,12 @@ PDU 类型（按规范注册表）： 2. Cache 若可增量，返回变化并以 `End of Data` 收尾。 3. 若不可增量，返回 `Cache Reset`；Router 退回 Full Sync。 +### 11.5.3 Cache Restart / Cache Shutdown（v2） + +- `Cache Restart`：Cache 即将对已连接客户端不可用，但预计会在客户端达到 Expire Interval 前恢复；v2 session 应发送 `Error Report` code 12，随后关闭连接。 +- `Cache Shutdown`：Cache 即将有意下线，并希望客户端 flush 从该 cache 学到的数据；v2 session 应发送 `Error Report` code 13，随后关闭连接。 +- 这两个错误码只属于 v2。v0/v1 session 不得发送 code 12/13，只能关闭连接。 + ## 11.6 版本协商与降级 - Router 每次新连接必须由 `Reset Query` 或 `Serial Query` 启动，携带其协议版本。 @@ -138,6 +144,21 @@ PDU 类型（按规范注册表）： - 对 Cache 输出 payload 的顺序提出约束。 - 建议 Router 使用“事务式应用”（例如接收到完整响应后再切换生效）降低中间态影响。 +### 11.8.3 Error Code 版本范围 + +- v0（RFC6810）：0..7。 +- v1（RFC8210）：继承 v0，并新增 8 `Unexpected Protocol Version`。 +- v2（8210bis-26）：继承 v1，并新增 9 `ASPA Provider List Error`、10 `Transport Failure`、11 `Ordering Error`、12 `Cache Restart`、13 `Cache Shutdown`。 +- 实现发送 Error Report 时，只能发送当前 session 协议版本定义过的错误码。 + +### 11.8.4 本实现的生命周期控制 + +- `RPKI_RTR_STOP_REASON=cache-restart|cache-shutdown|close-only` 控制进程收到 Ctrl-C 后的服务停止语义，默认 `cache-restart`。 +- Admin API: + - `POST /admin/rtr/lifecycle/restart`：向 v2 established clients 发送 `Cache Restart`，然后走进程退出路径；实际重新拉起由 systemd/Docker/Kubernetes/Windows Service 等 supervisor 完成。若没有外部机制在客户端 `Expire Interval` 到期前恢复服务，不应使用该接口。 + - `POST /admin/rtr/lifecycle/shutdown`：向 v2 established clients 发送 `Cache Shutdown`，然后退出进程。 + - v0/v1 或尚未完成版本协商的连接不会收到 code 12/13。 + ## 11.9 传输与安全 规范定义可承载于多种传输： @@ -155,4 +176,4 @@ PDU 类型（按规范注册表）： - RFC 6810: https://www.rfc-editor.org/rfc/rfc6810.html - RFC 8210: https://www.rfc-editor.org/rfc/rfc8210.html -- draft-ietf-sidrops-8210bis-25: https://www.ietf.org/archive/id/draft-ietf-sidrops-8210bis-25.html +- draft-ietf-sidrops-8210bis-26: https://www.ietf.org/archive/id/draft-ietf-sidrops-8210bis-26.html diff --git a/src/bin/rtr_debug_client/README.md b/src/bin/rtr_debug_client/README.md index 0bc9926..e4b4637 100644 --- a/src/bin/rtr_debug_client/README.md +++ b/src/bin/rtr_debug_client/README.md @@ -108,7 +108,7 @@ cargo run --bin rtr_debug_client -- \ 通用参数： - `--timeout `：读取 PDU 超时时间（秒） - `--poll `：默认自动轮询间隔（秒） -- `--keep-after-error`：收到 `ErrorReport` 后不暂停自动轮询 +- `--keep-after-error`：兼容旧脚本的保留参数；默认已按 RFC/draft 处理 `ErrorReport` - `--summary-only`：仅打印摘要，抑制 payload PDU 详细内容 TLS 参数： @@ -145,7 +145,6 @@ SSH 参数： - `serial` / `serial ` - `timeout` / `timeout ` - `poll` / `poll ` / `poll pause` / `poll resume` -- `keep-after-error` - `output` / `output verbose` / `output summary` - `quit` diff --git a/src/bin/rtr_debug_client/main.rs b/src/bin/rtr_debug_client/main.rs index 917a9e6..101be80 100644 --- a/src/bin/rtr_debug_client/main.rs +++ b/src/bin/rtr_debug_client/main.rs @@ -27,6 +27,7 @@ mod wire; use crate::pretty::{parse_end_of_data_info, parse_serial_notify_serial, print_pdu, print_raw_pdu}; use crate::protocol::{PduHeader, PduType, QueryMode}; use crate::wire::{read_pdu, send_reset_query, send_serial_query}; +use rpki::rtr::error_type::{ErrorCode, ErrorCodeDisposition}; macro_rules! println { () => { @@ -64,6 +65,12 @@ enum OutputMode { SummaryOnly, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum IncomingAction { + Continue, + Reconnect, +} + #[tokio::main] async fn main() -> io::Result<()> { let config = Config::from_args()?; @@ -77,7 +84,11 @@ async fn main() -> io::Result<()> { "poll : {}s (default before EndOfData refresh is known)", config.default_poll_secs ); - println!("keep-after-error: {}", config.keep_after_error); + if config.keep_after_error { + println!( + "keep-after-error: accepted for compatibility; RFC/draft error handling is default" + ); + } println!("output : {}", config.output_mode.describe()); match &config.mode { QueryMode::Reset => { @@ -98,7 +109,6 @@ async fn main() -> io::Result<()> { config.version, config.read_timeout_secs, config.default_poll_secs, - config.keep_after_error, config.output_mode, ); @@ -108,6 +118,10 @@ async fn main() -> io::Result<()> { loop { let stream = loop { + if state.expire_cache_restart_retention_if_needed() { + println!("Cache Restart retention expired; flushed local session/serial state."); + } + match connect_stream(&config).await { Ok(stream) => { if state.output_mode == OutputMode::Verbose { @@ -124,6 +138,9 @@ async fn main() -> io::Result<()> { }; let (mut reader, mut writer) = tokio_io::split(stream); + if state.expire_cache_restart_retention_if_needed() { + println!("Cache Restart retention expired; send Reset Query on reconnect."); + } send_resume_query(&mut writer, &mut state, &config.mode).await?; state.schedule_next_poll(); println!(); @@ -193,7 +210,8 @@ async fn main() -> io::Result<()> { print_pdu(&pdu.header, &pdu.body); } match handle_incoming_pdu(&mut writer, &mut state, &pdu.header, &pdu.body).await { - Ok(()) => {} + Ok(IncomingAction::Continue) => {} + Ok(IncomingAction::Reconnect) => break true, Err(err) if should_reconnect(&err) => { eprintln!("connection dropped while handling incoming PDU: {}", err); break true; @@ -202,10 +220,11 @@ async fn main() -> io::Result<()> { } } Ok(Err(err)) => { - eprintln!("read PDU failed: {}", err); if should_reconnect(&err) { + eprintln!("RTR transport disconnected while reading PDU: {}", err); break true; } + eprintln!("read PDU failed: {}", err); return Err(err); } Err(_) => { @@ -250,8 +269,6 @@ async fn main() -> io::Result<()> { println!("quit requested, closing client."); return Ok(()); } - } - Ok(false) => { if state.take_reconnect_now() { reconnect_now = true; break; @@ -288,9 +305,8 @@ async fn send_resume_query( ) -> io::Result<()> { if state.force_reset_on_reconnect { state.force_reset_on_reconnect = false; - state.session_id = None; - state.serial = None; state.current_session_id = None; + state.cache_restart_retain_until = None; send_reset_query(writer, state.version).await?; if state.output_mode == OutputMode::Verbose { println!("reconnected, send Reset Query (forced)"); @@ -345,7 +361,7 @@ async fn handle_incoming_pdu( state: &mut ClientState, header: &PduHeader, body: &[u8], -) -> io::Result<()> { +) -> io::Result { match header.pdu_type() { PduType::CacheResponse => { state.current_session_id = Some(header.session_id()); @@ -371,6 +387,7 @@ async fn handle_incoming_pdu( state.retry = eod.retry; state.expire = eod.expire; state.last_error_code = None; + state.cache_restart_retain_until = None; state.schedule_next_poll(); if state.output_mode == OutputMode::Verbose { @@ -468,6 +485,7 @@ async fn handle_incoming_pdu( state.current_session_id = None; state.serial = None; state.last_error_code = None; + state.cache_restart_retain_until = None; state.skipped_payload_pdu_count_in_round = 0; send_reset_query(writer, state.version).await?; state.schedule_next_poll(); @@ -476,22 +494,101 @@ async fn handle_incoming_pdu( PduType::ErrorReport => { println!(); - println!("received Error Report, pause auto polling for debugging."); - state.last_error_code = Some(header.error_code()); - if let Some(retry) = state.retry { - println!("server retry hint currently stored: {}s", retry); - if state.should_prefer_retry_poll() { - println!("when resumed, auto polling will use retry instead of refresh."); + let raw_error_code = header.error_code(); + state.last_error_code = Some(raw_error_code); + let Ok(error_code) = ErrorCode::try_from(raw_error_code) else { + state.handle_fatal_error_report(raw_error_code); + println!( + "received invalid Error Report code {} for RTR version {}.", + raw_error_code, state.version + ); + println!( + "dropped session, flushed local state, and reconnect after {}s.", + state.reconnect_delay_secs() + ); + println!(); + return Ok(IncomingAction::Reconnect); + }; + + let Some(disposition) = error_code.disposition_in_version(state.version) else { + state.handle_fatal_error_report(raw_error_code); + println!( + "received Error Report code {} ({}) which is not defined for RTR version {}.", + raw_error_code, + error_code.description(), + state.version + ); + println!( + "dropped session, flushed local state, and reconnect after {}s.", + state.reconnect_delay_secs() + ); + println!(); + return Ok(IncomingAction::Reconnect); + }; + + match error_code { + ErrorCode::NoDataAvailable => { + state.handle_no_data_available_error(); + println!("received No Data Available Error Report."); + println!( + "retain local state and reconnect after {}s; next request will be Reset Query.", + state.reconnect_delay_secs() + ); + println!(); + return Ok(IncomingAction::Reconnect); + } + ErrorCode::UnsupportedProtocolVersion + if disposition == ErrorCodeDisposition::NonFatal => + { + let old_version = state.version; + state.handle_unsupported_protocol_version_error(); + println!("received Unsupported Protocol Version Error Report."); + println!( + "downgraded RTR version from {} to {} and reconnect immediately.", + old_version, state.version + ); + println!(); + return Ok(IncomingAction::Reconnect); + } + ErrorCode::CacheRestart => { + state.handle_cache_restart_error(); + println!("received Cache Restart Error Report."); + if let Some(remaining) = state.cache_restart_retain_secs_remaining() { + println!( + "retain session/serial for up to {}s before Expire Interval elapses.", + remaining + ); + } + println!( + "retain session/serial and reconnect after {}s using Retry Interval semantics.", + state.reconnect_delay_secs() + ); + println!(); + return Ok(IncomingAction::Reconnect); + } + ErrorCode::CacheShutdown => { + state.handle_cache_shutdown_error(); + println!("received Cache Shutdown Error Report."); + println!( + "flushed local session/serial state and reconnect after {}s using default Retry Interval semantics.", + state.reconnect_delay_secs() + ); + println!(); + return Ok(IncomingAction::Reconnect); + } + _ => { + debug_assert!(disposition.is_fatal()); + state.handle_fatal_error_report(raw_error_code); + println!( + "received fatal Error Report code {} ({}); dropped session and flushed local state.", + raw_error_code, + error_code.description() + ); + println!("reconnect after {}s.", state.reconnect_delay_secs()); + println!(); + return Ok(IncomingAction::Reconnect); } } - if state.keep_after_error { - println!("keep-after-error is enabled, auto polling will continue."); - state.schedule_next_poll(); - } else { - println!("use `reset`, `serial`, or `poll resume` to continue manually."); - state.pause_auto_poll(); - } - println!(); } PduType::SerialQuery | PduType::ResetQuery | PduType::Unknown(_) => { @@ -499,7 +596,7 @@ async fn handle_incoming_pdu( } } - Ok(()) + Ok(IncomingAction::Continue) } async fn handle_poll_tick(writer: &mut ClientWriter, state: &mut ClientState) -> io::Result<()> { @@ -576,6 +673,7 @@ async fn handle_console_command( ["reset"] => { println!("manual command: send Reset Query"); if let Some(writer) = writer.as_mut() { + state.cache_restart_retain_until = None; send_reset_query(writer, state.version).await?; state.schedule_next_poll(); } else { @@ -584,6 +682,7 @@ async fn handle_console_command( state.session_id = None; state.serial = None; state.current_session_id = None; + state.cache_restart_retain_until = None; println!("not connected, queued Reset Query for next reconnect"); } } @@ -750,7 +849,6 @@ fn print_help() { println!(" poll override poll interval seconds"); println!(" poll pause pause auto polling"); println!(" poll resume resume auto polling"); - println!(" keep-after-error show current keep-after-error setting"); println!(" output show current output mode"); println!(" output verbose print all PDUs"); println!(" output summary suppress payload PDU details"); @@ -772,7 +870,10 @@ fn print_state(state: &ClientState) { println!(" effective_poll_secs: {}", state.effective_poll_secs()); println!(" poll_source : {}", state.poll_interval_source()); println!(" last_error_code : {:?}", state.last_error_code); - println!(" keep_after_error : {}", state.keep_after_error); + println!( + " cache_restart_ttl : {:?}", + state.cache_restart_retain_secs_remaining() + ); println!(" output_mode : {}", state.output_mode.describe()); println!(" skipped_payloads : {}", state.skipped_payload_pdu_count); println!(" poll_paused : {}", state.poll_paused); @@ -790,7 +891,6 @@ struct ClientState { retry: Option, expire: Option, last_error_code: Option, - keep_after_error: bool, output_mode: OutputMode, skipped_payload_pdu_count: u64, skipped_payload_pdu_count_in_round: u64, @@ -801,6 +901,7 @@ struct ClientState { poll_paused: bool, force_reset_on_reconnect: bool, reconnect_now: bool, + cache_restart_retain_until: Option, } impl ClientState { @@ -808,7 +909,6 @@ impl ClientState { version: u8, read_timeout_secs: u64, default_poll_secs: u64, - keep_after_error: bool, output_mode: OutputMode, ) -> Self { Self { @@ -820,7 +920,6 @@ impl ClientState { retry: None, expire: None, last_error_code: None, - keep_after_error, output_mode, skipped_payload_pdu_count: 0, skipped_payload_pdu_count_in_round: 0, @@ -830,6 +929,7 @@ impl ClientState { poll_paused: false, force_reset_on_reconnect: false, reconnect_now: false, + cache_restart_retain_until: None, } } @@ -868,8 +968,100 @@ impl ClientState { } } + fn handle_no_data_available_error(&mut self) { + self.last_error_code = Some(ErrorCode::NoDataAvailable.as_u16()); + self.current_session_id = None; + self.force_reset_on_reconnect = true; + self.poll_paused = false; + self.cache_restart_retain_until = None; + self.skipped_payload_pdu_count_in_round = 0; + } + + fn handle_unsupported_protocol_version_error(&mut self) { + self.last_error_code = Some(ErrorCode::UnsupportedProtocolVersion.as_u16()); + self.current_session_id = None; + self.force_reset_on_reconnect = true; + self.poll_paused = false; + self.cache_restart_retain_until = None; + self.skipped_payload_pdu_count_in_round = 0; + if self.version > 0 { + self.version -= 1; + } + } + + fn handle_cache_restart_error(&mut self) { + self.last_error_code = Some(ErrorCode::CacheRestart.as_u16()); + self.current_session_id = None; + self.force_reset_on_reconnect = false; + self.poll_paused = false; + self.skipped_payload_pdu_count_in_round = 0; + self.cache_restart_retain_until = self + .expire + .map(|expire| Instant::now() + Duration::from_secs(expire as u64)); + } + + fn handle_cache_shutdown_error(&mut self) { + self.last_error_code = Some(ErrorCode::CacheShutdown.as_u16()); + self.session_id = None; + self.serial = None; + self.current_session_id = None; + self.refresh = None; + self.retry = None; + self.expire = None; + self.force_reset_on_reconnect = true; + self.poll_paused = false; + self.skipped_payload_pdu_count_in_round = 0; + self.cache_restart_retain_until = None; + } + + fn handle_fatal_error_report(&mut self, error_code: u16) { + self.last_error_code = Some(error_code); + self.session_id = None; + self.serial = None; + self.current_session_id = None; + self.refresh = None; + self.retry = None; + self.expire = None; + self.force_reset_on_reconnect = true; + self.poll_paused = false; + self.skipped_payload_pdu_count_in_round = 0; + self.cache_restart_retain_until = None; + } + + fn expire_cache_restart_retention_if_needed(&mut self) -> bool { + let Some(deadline) = self.cache_restart_retain_until else { + return false; + }; + + if Instant::now() < deadline { + return false; + } + + self.session_id = None; + self.serial = None; + self.current_session_id = None; + self.refresh = None; + self.retry = None; + self.expire = None; + self.force_reset_on_reconnect = true; + self.poll_paused = false; + self.cache_restart_retain_until = None; + true + } + + fn cache_restart_retain_secs_remaining(&self) -> Option { + self.cache_restart_retain_until + .map(|deadline| deadline.saturating_duration_since(Instant::now()).as_secs()) + } + fn should_prefer_retry_poll(&self) -> bool { - matches!(self.last_error_code, Some(2 | 10)) + matches!( + self.last_error_code, + Some(code) + if code == ErrorCode::NoDataAvailable.as_u16() + || code == ErrorCode::TransportFailed.as_u16() + || code == ErrorCode::CacheRestart.as_u16() + ) } fn poll_interval_source(&self) -> &'static str { @@ -883,7 +1075,11 @@ impl ClientState { } fn reconnect_delay_secs(&self) -> u64 { - if self.should_prefer_retry_poll() { + if self.last_error_code == Some(ErrorCode::UnsupportedProtocolVersion.as_u16()) { + 0 + } else if self.last_error_code == Some(ErrorCode::CacheShutdown.as_u16()) { + self.default_poll_secs + } else if self.should_prefer_retry_poll() { self.retry .map(|v| v as u64) .unwrap_or(self.default_poll_secs) diff --git a/src/bin/rtr_debug_client/pretty.rs b/src/bin/rtr_debug_client/pretty.rs index 6610721..8871c42 100644 --- a/src/bin/rtr_debug_client/pretty.rs +++ b/src/bin/rtr_debug_client/pretty.rs @@ -279,6 +279,8 @@ fn error_code_name(code: u16) -> &'static str { 9 => "ASPA Provider List Error", 10 => "Transport Failure", 11 => "Ordering Error", + 12 => "Cache Restart", + 13 => "Cache Shutdown", _ => "Unknown Error Code", } } diff --git a/src/main_rtr.rs b/src/main_rtr.rs index 6855cdf..7b80fc6 100644 --- a/src/main_rtr.rs +++ b/src/main_rtr.rs @@ -10,7 +10,7 @@ use rpki::rtr::bootstrap::{init_shared_cache, open_store, start_servers}; use rpki::rtr::config::{AppConfig, log_startup_config}; use rpki::rtr::report::{ReportConfiguration, ReportContext}; use rpki::rtr::runtime::spawn_refresh_task; -use rpki::rtr::server::RtrService; +use rpki::rtr::server::{RtrService, RtrShutdownReason}; use rpki::slurm::admin::SlurmAdmin; #[tokio::main] @@ -39,14 +39,18 @@ async fn main() -> Result<()> { let service = RtrService::with_config(shared_cache.clone(), config.service_config.clone()); let notifier = service.notifier(); let service_stats = service.stats(); + let shutdown_handle = service.shutdown_handle(); let (source_reload_tx, source_reload_rx) = mpsc::channel(8); let source_reload = SourceReloadHandle::new(source_reload_tx); + let (process_shutdown_tx, process_shutdown_rx) = mpsc::channel(1); let admin_task = config.admin_addr.map(|addr| { let slurm_admin = config.slurm_dir.as_ref().map(SlurmAdmin::new); let admin_state = AdminState::new( runtime_config.clone(), Some(source_reload.clone()), + Some(shutdown_handle.clone()), + Some(process_shutdown_tx.clone()), slurm_admin, LogTailConfig::from_env(), ); @@ -64,9 +68,9 @@ async fn main() -> Result<()> { report_context, ); - wait_for_shutdown().await?; - - running.shutdown(); + let stop_reason = wait_for_shutdown(process_shutdown_rx).await?; + info!("stopping RTR service: reason={}", stop_reason.as_str()); + running.shutdown(stop_reason); running.wait().await; refresh_task.abort(); @@ -80,10 +84,22 @@ async fn main() -> Result<()> { Ok(()) } -async fn wait_for_shutdown() -> Result<()> { - tokio::signal::ctrl_c().await?; - info!("shutdown signal received"); - Ok(()) +async fn wait_for_shutdown( + mut process_shutdown_rx: mpsc::Receiver, +) -> Result { + tokio::select! { + signal = tokio::signal::ctrl_c() => { + signal?; + let reason = RtrShutdownReason::from_env(); + info!("shutdown signal received: reason={}", reason.as_str()); + Ok(reason) + } + reason = process_shutdown_rx.recv() => { + let reason = reason.unwrap_or_else(RtrShutdownReason::from_env); + info!("admin process lifecycle request received: reason={}", reason.as_str()); + Ok(reason) + } + } } fn init_tracing(timezone: chrono_tz::Tz) { diff --git a/src/rtr/admin.rs b/src/rtr/admin.rs index 3bf0419..f15c2e5 100644 --- a/src/rtr/admin.rs +++ b/src/rtr/admin.rs @@ -12,6 +12,7 @@ use tokio::time::sleep; use tracing::{info, warn}; use crate::rtr::config::{RuntimeConfig, RuntimeConfigPatch}; +use crate::rtr::server::{RtrShutdownHandle, RtrShutdownReason}; use crate::slurm::admin::{ SlurmAdmin, SlurmFileActionRequest, SlurmFileOperationResult, SlurmFileWriteRequest, parse_reload_query, @@ -69,6 +70,8 @@ pub struct SourceReloadResult { pub struct AdminState { runtime_config: RuntimeConfigHandle, source_reload: Option, + shutdown: Option, + process_shutdown: Option>, slurm_admin: Option, log_tail: LogTailConfig, } @@ -77,12 +80,16 @@ impl AdminState { pub fn new( runtime_config: RuntimeConfigHandle, source_reload: Option, + shutdown: Option, + process_shutdown: Option>, slurm_admin: Option, log_tail: LogTailConfig, ) -> Self { Self { runtime_config, source_reload, + shutdown, + process_shutdown, slurm_admin, log_tail, } @@ -293,6 +300,8 @@ async fn route_admin_request( status: "ok", config_api: true, source_reload_api: state.source_reload.is_some(), + lifecycle_api: state.shutdown.is_some(), + process_lifecycle_api: state.process_shutdown.is_some(), slurm_api: state.slurm_admin.is_some(), logs_api: true, })?; @@ -300,6 +309,14 @@ async fn route_admin_request( return Ok(()); } + if request.method == "POST" && request.path == "/admin/rtr/lifecycle/restart" { + return trigger_process_lifecycle(stream, &state, RtrShutdownReason::CacheRestart).await; + } + + if request.method == "POST" && request.path == "/admin/rtr/lifecycle/shutdown" { + return trigger_process_lifecycle(stream, &state, RtrShutdownReason::CacheShutdown).await; + } + if request.method == "GET" && request.path == "/admin/rtr/logs/tail" { return tail_log_stream(stream, request.query.as_deref(), &state.log_tail).await; } @@ -497,6 +514,42 @@ async fn reload_source( source_reload.reload(phase, force).await } +async fn trigger_process_lifecycle( + stream: &mut TcpStream, + state: &AdminState, + reason: RtrShutdownReason, +) -> Result<()> { + let Some(process_shutdown) = state.process_shutdown.as_ref() else { + write_response( + stream, + 400, + "process lifecycle admin is disabled", + "text/plain", + ) + .await?; + return Ok(()); + }; + + let action = match reason { + RtrShutdownReason::CacheRestart => "restart", + RtrShutdownReason::CacheShutdown => "shutdown", + RtrShutdownReason::CloseOnly => { + unreachable!("close-only is not a process lifecycle action") + } + }; + let json = serde_json::to_string_pretty(&LifecycleResponse { + status: "ok", + action, + reason: reason.as_str(), + process_exit: true, + })?; + write_response(stream, 200, &json, "application/json").await?; + if process_shutdown.send(reason).await.is_err() { + warn!("process lifecycle request accepted but shutdown receiver is gone"); + } + Ok(()) +} + async fn parse_json Deserialize<'de>>( stream: &mut TcpStream, body: &[u8], @@ -729,10 +782,20 @@ struct AdminHealthResponse { status: &'static str, config_api: bool, source_reload_api: bool, + lifecycle_api: bool, + process_lifecycle_api: bool, slurm_api: bool, logs_api: bool, } +#[derive(Serialize)] +struct LifecycleResponse { + status: &'static str, + action: &'static str, + reason: &'static str, + process_exit: bool, +} + #[derive(Serialize)] struct AdminConfigResponse { status: &'static str, diff --git a/src/rtr/error_type.rs b/src/rtr/error_type.rs index f62a02f..1d6f081 100644 --- a/src/rtr/error_type.rs +++ b/src/rtr/error_type.rs @@ -16,8 +16,130 @@ pub enum ErrorCode { AspaProviderListError = 9, TransportFailed = 10, OrderingError = 11, + CacheRestart = 12, + CacheShutdown = 13, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErrorCodeDisposition { + Fatal, + NonFatal, +} + +impl ErrorCodeDisposition { + #[inline] + pub fn is_fatal(self) -> bool { + self == Self::Fatal + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ErrorCodeVersionSpec { + pub code: ErrorCode, + pub min_version: u8, + pub max_version: u8, + pub disposition: ErrorCodeDisposition, +} + +impl ErrorCodeVersionSpec { + pub fn contains(self, code: ErrorCode, version: u8) -> bool { + self.code == code && (self.min_version..=self.max_version).contains(&version) + } +} + +pub const ERROR_CODE_VERSION_SPECS: &[ErrorCodeVersionSpec] = &[ + ErrorCodeVersionSpec { + code: ErrorCode::CorruptData, + min_version: 0, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::InternalError, + min_version: 0, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::NoDataAvailable, + min_version: 0, + max_version: 2, + disposition: ErrorCodeDisposition::NonFatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::InvalidRequest, + min_version: 0, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::UnsupportedProtocolVersion, + min_version: 0, + max_version: 1, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::UnsupportedProtocolVersion, + min_version: 2, + max_version: 2, + disposition: ErrorCodeDisposition::NonFatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::UnsupportedPduType, + min_version: 0, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::WithdrawalOfUnknownRecord, + min_version: 0, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::DuplicateAnnouncement, + min_version: 0, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::UnexpectedProtocolVersion, + min_version: 1, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::AspaProviderListError, + min_version: 2, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::TransportFailed, + min_version: 2, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::OrderingError, + min_version: 2, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::CacheRestart, + min_version: 2, + max_version: 2, + disposition: ErrorCodeDisposition::NonFatal, + }, + ErrorCodeVersionSpec { + code: ErrorCode::CacheShutdown, + min_version: 2, + max_version: 2, + disposition: ErrorCodeDisposition::Fatal, + }, +]; + impl ErrorCode { #[inline] pub fn as_u16(self) -> u16 { @@ -46,11 +168,28 @@ impl ErrorCode { ErrorCode::AspaProviderListError => "ASPA Provider List Error", - ErrorCode::TransportFailed => "Transport Failed", + ErrorCode::TransportFailed => "Transport Failure", ErrorCode::OrderingError => "Ordering Error", + + ErrorCode::CacheRestart => "Cache Restart", + + ErrorCode::CacheShutdown => "Cache Shutdown", } } + + pub fn is_defined_in_version(self, version: u8) -> bool { + ERROR_CODE_VERSION_SPECS + .iter() + .any(|spec| spec.contains(self, version)) + } + + pub fn disposition_in_version(self, version: u8) -> Option { + ERROR_CODE_VERSION_SPECS + .iter() + .find(|spec| spec.contains(self, version)) + .map(|spec| spec.disposition) + } } impl TryFrom for ErrorCode { @@ -70,6 +209,8 @@ impl TryFrom for ErrorCode { 9 => Ok(ErrorCode::AspaProviderListError), 10 => Ok(ErrorCode::TransportFailed), 11 => Ok(ErrorCode::OrderingError), + 12 => Ok(ErrorCode::CacheRestart), + 13 => Ok(ErrorCode::CacheShutdown), _ => Err(()), } } diff --git a/src/rtr/server/connection.rs b/src/rtr/server/connection.rs index 55effbf..e7cd81a 100644 --- a/src/rtr/server/connection.rs +++ b/src/rtr/server/connection.rs @@ -14,6 +14,7 @@ use x509_parser::prelude::{FromDer, X509Certificate}; use tokio_rustls::TlsAcceptor; use crate::rtr::cache::SharedRtrCache; +use crate::rtr::server::service::RtrServiceControl; use crate::rtr::session::RtrSession; #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -117,7 +118,7 @@ pub async fn handle_tcp_connection( stream: TcpStream, peer_addr: SocketAddr, notify_rx: broadcast::Receiver<()>, - shutdown_rx: watch::Receiver, + shutdown_rx: watch::Receiver, ) -> Result<()> { let session = RtrSession::new(cache, stream, notify_rx, shutdown_rx); session.run().await?; @@ -133,7 +134,7 @@ pub async fn handle_tls_connection( acceptor: TlsAcceptor, enforce_client_san_ip_match: bool, notify_rx: broadcast::Receiver<()>, - shutdown_rx: watch::Receiver, + shutdown_rx: watch::Receiver, handshake_permit: Option, ) -> Result<()> { info!("RTR TLS handshake started for {}", peer_addr); diff --git a/src/rtr/server/listener.rs b/src/rtr/server/listener.rs index b3d4209..f03cc9e 100644 --- a/src/rtr/server/listener.rs +++ b/src/rtr/server/listener.rs @@ -26,6 +26,7 @@ use crate::rtr::server::connection::{ ConnectionGuard, RtrTransportConnectionCounters, RtrTransportKind, handle_tcp_connection, handle_tls_connection, is_expected_disconnect, }; +use crate::rtr::server::service::RtrServiceControl; use crate::rtr::server::ssh::RtrSshRuntimeConfig; use crate::rtr::server::tls::load_rustls_server_config_with_options; use crate::rtr::session::RtrSession; @@ -45,7 +46,7 @@ pub trait TransportAcceptor: Clone + Send + Sync + 'static { stream: TcpStream, peer_addr: SocketAddr, notify_tx: broadcast::Sender<()>, - shutdown_tx: watch::Sender, + shutdown_tx: watch::Sender, handshake_permit: Option, ) -> TransportFuture; } @@ -67,7 +68,7 @@ impl TransportAcceptor for TcpTransport { stream: TcpStream, peer_addr: SocketAddr, notify_tx: broadcast::Sender<()>, - shutdown_tx: watch::Sender, + shutdown_tx: watch::Sender, _handshake_permit: Option, ) -> TransportFuture { Box::pin(async move { @@ -106,7 +107,7 @@ impl TransportAcceptor for TlsTransport { stream: TcpStream, peer_addr: SocketAddr, notify_tx: broadcast::Sender<()>, - shutdown_tx: watch::Sender, + shutdown_tx: watch::Sender, handshake_permit: Option, ) -> TransportFuture { let acceptor = self.acceptor.clone(); @@ -146,7 +147,7 @@ impl TransportAcceptor for SshTransport { stream: TcpStream, peer_addr: SocketAddr, notify_tx: broadcast::Sender<()>, - shutdown_tx: watch::Sender, + shutdown_tx: watch::Sender, _handshake_permit: Option, ) -> TransportFuture { let runtime = self.runtime.clone(); @@ -175,11 +176,12 @@ impl TransportAcceptor for SshTransport { } changed = connection_shutdown_rx.changed() => { match changed { - Ok(()) if *connection_shutdown_rx.borrow() => { + Ok(()) if connection_shutdown_rx.borrow().shutdown_reason().is_some() => { + tokio::time::sleep(Duration::from_secs(2)).await; let _ = handle .disconnect( Disconnect::ByApplication, - "service shutdown".to_string(), + "RTR service shutdown".to_string(), "".to_string(), ) .await; @@ -197,7 +199,7 @@ pub struct RtrServer { bind_addr: SocketAddr, cache: SharedRtrCache, notify_tx: broadcast::Sender<()>, - shutdown_tx: watch::Sender, + shutdown_tx: watch::Sender, connection_limiter: Arc, handshake_limiter: Arc, active_connections: Arc, @@ -212,7 +214,7 @@ impl RtrServer { bind_addr: SocketAddr, cache: SharedRtrCache, notify_tx: broadcast::Sender<()>, - shutdown_tx: watch::Sender, + shutdown_tx: watch::Sender, connection_limiter: Arc, handshake_limiter: Arc, active_connections: Arc, @@ -307,11 +309,12 @@ impl RtrServer { changed = shutdown_rx.changed() => { match changed { Ok(()) => { - if *shutdown_rx.borrow() { + if let Some(reason) = shutdown_rx.borrow().shutdown_reason() { info!( - "RTR {} listener {} shutting down", + "RTR {} listener {} shutting down: reason={}", transport.name(), - self.bind_addr + self.bind_addr, + reason.as_str() ); return Ok(()); } @@ -458,7 +461,7 @@ impl RtrServer { struct RtrSshHandler { cache: SharedRtrCache, notify_rx: broadcast::Receiver<()>, - shutdown_rx: watch::Receiver, + shutdown_rx: watch::Receiver, peer_addr: SocketAddr, authorized_keys: Arc>, username: Arc, @@ -472,7 +475,7 @@ impl RtrSshHandler { fn new( cache: SharedRtrCache, notify_rx: broadcast::Receiver<()>, - shutdown_rx: watch::Receiver, + shutdown_rx: watch::Receiver, peer_addr: SocketAddr, authorized_keys: Arc>, username: Arc, diff --git a/src/rtr/server/mod.rs b/src/rtr/server/mod.rs index d7a0133..fa4e109 100644 --- a/src/rtr/server/mod.rs +++ b/src/rtr/server/mod.rs @@ -10,5 +10,8 @@ pub use config::RtrServiceConfig; pub use connection::RtrTransportConnectionCounts; pub use listener::RtrServer; pub use notifier::RtrNotifier; -pub use service::{RtrService, RtrServiceStats, RunningRtrService}; +pub use service::{ + RtrService, RtrServiceControl, RtrServiceStats, RtrShutdownHandle, RtrShutdownReason, + RunningRtrService, +}; pub use tls::load_rustls_server_config; diff --git a/src/rtr/server/service.rs b/src/rtr/server/service.rs index 734afce..aabfb3e 100644 --- a/src/rtr/server/service.rs +++ b/src/rtr/server/service.rs @@ -1,3 +1,4 @@ +use std::env; use std::net::SocketAddr; use std::path::Path; use std::sync::{ @@ -18,10 +19,69 @@ use crate::rtr::server::listener::RtrServer; use crate::rtr::server::notifier::RtrNotifier; use crate::rtr::server::ssh::{SshAuthMode, load_rtr_ssh_runtime_config}; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RtrShutdownReason { + CloseOnly, + CacheRestart, + CacheShutdown, +} + +impl RtrShutdownReason { + pub fn as_str(self) -> &'static str { + match self { + Self::CloseOnly => "close-only", + Self::CacheRestart => "cache-restart", + Self::CacheShutdown => "cache-shutdown", + } + } + + pub fn from_env() -> Self { + match env::var("RPKI_RTR_STOP_REASON") { + Ok(value) => Self::parse(&value).unwrap_or(Self::CacheRestart), + Err(_) => Self::CacheRestart, + } + } + + pub fn parse(value: &str) -> Option { + match value.trim().to_ascii_lowercase().as_str() { + "close-only" | "close_only" | "close" => Some(Self::CloseOnly), + "restart" | "cache-restart" | "cache_restart" => Some(Self::CacheRestart), + "shutdown" | "cache-shutdown" | "cache_shutdown" => Some(Self::CacheShutdown), + _ => None, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RtrServiceControl { + Running, + Shutdown(RtrShutdownReason), +} + +impl RtrServiceControl { + pub fn shutdown_reason(self) -> Option { + match self { + Self::Running => None, + Self::Shutdown(reason) => Some(reason), + } + } +} + +#[derive(Clone)] +pub struct RtrShutdownHandle { + shutdown_tx: watch::Sender, +} + +impl RtrShutdownHandle { + pub fn shutdown(&self, reason: RtrShutdownReason) { + let _ = self.shutdown_tx.send(RtrServiceControl::Shutdown(reason)); + } +} + pub struct RtrService { cache: SharedRtrCache, notify_tx: broadcast::Sender<()>, - shutdown_tx: watch::Sender, + shutdown_tx: watch::Sender, connection_limiter: Arc, handshake_limiter: Arc, active_connections: Arc, @@ -64,7 +124,7 @@ impl RtrService { pub fn with_config(cache: SharedRtrCache, config: RtrServiceConfig) -> Self { let (notify_tx, _) = broadcast::channel(config.notify_queue_size); - let (shutdown_tx, _) = watch::channel(false); + let (shutdown_tx, _) = watch::channel(RtrServiceControl::Running); let (connection_change_tx, _) = watch::channel(0); Self { @@ -89,6 +149,12 @@ impl RtrService { RtrNotifier::new(self.notify_tx.clone()) } + pub fn shutdown_handle(&self) -> RtrShutdownHandle { + RtrShutdownHandle { + shutdown_tx: self.shutdown_tx.clone(), + } + } + pub fn notify_cache_updated(&self) { let _ = self.notify_tx.send(()); } @@ -165,7 +231,7 @@ impl RtrService { pub fn spawn_tcp(&self, bind_addr: SocketAddr) -> JoinHandle<()> { if self.config.warn_insecure_tcp { warn!( - "starting plain TCP RTR service on {}. Per draft-ietf-sidrops-8210bis-25 Section 9, unsecured TCP must only be used on a trusted and controlled network", + "starting plain TCP RTR service on {}. Per draft-ietf-sidrops-8210bis-26 Section 9, unsecured TCP must only be used on a trusted and controlled network", bind_addr ); } @@ -339,13 +405,13 @@ impl RtrService { } pub struct RunningRtrService { - shutdown_tx: watch::Sender, + shutdown_tx: watch::Sender, handles: Vec>, } impl RunningRtrService { - pub fn shutdown(&self) { - let _ = self.shutdown_tx.send(true); + pub fn shutdown(&self, reason: RtrShutdownReason) { + let _ = self.shutdown_tx.send(RtrServiceControl::Shutdown(reason)); } pub async fn wait(self) { diff --git a/src/rtr/session.rs b/src/rtr/session.rs index 4709924..d61f82d 100644 --- a/src/rtr/session.rs +++ b/src/rtr/session.rs @@ -21,6 +21,7 @@ use crate::rtr::pdu::{ Aspa as AspaPdu, CacheReset, CacheResponse, EndOfData, ErrorReport, Flags, HEADER_LEN, Header, IPv4Prefix, IPv6Prefix, ResetQuery, RouterKey as RouterKeyPdu, SerialNotify, SerialQuery, }; +use crate::rtr::server::service::{RtrServiceControl, RtrShutdownReason}; const SUPPORTED_MAX_VERSION: u8 = 2; const SUPPORTED_MIN_VERSION: u8 = 0; @@ -121,7 +122,7 @@ pub struct RtrSession { stream: S, state: SessionState, notify_rx: broadcast::Receiver<()>, - shutdown_rx: watch::Receiver, + shutdown_rx: watch::Receiver, last_notify_at: Option, transport_timeout_override: Option, } @@ -134,7 +135,7 @@ where cache: SharedRtrCache, stream: S, notify_rx: broadcast::Receiver<()>, - shutdown_rx: watch::Receiver, + shutdown_rx: watch::Receiver, ) -> Self { Self { cache, @@ -167,9 +168,9 @@ where changed = self.shutdown_rx.changed() => { match changed { Ok(()) => { - if *self.shutdown_rx.borrow() { - info!("RTR session closing due to service shutdown: {}", self.session_summary()); - self.state = SessionState::Closed; + let reason = self.shutdown_rx.borrow().shutdown_reason(); + if let Some(reason) = reason { + self.handle_service_shutdown(reason).await?; return Ok(()); } } @@ -182,7 +183,7 @@ where } header_res = async { - // draft-ietf-sidrops-8210bis-25 Section 6 allows routers to wait up to + // draft-ietf-sidrops-8210bis-26 Section 6 allows routers to wait up to // Refresh Interval before polling again (recommended default: 3600s). // In an established session, a long quiet period is therefore expected and // must not be treated as a transport stall. @@ -552,6 +553,24 @@ where .await } + async fn handle_service_shutdown(&mut self, reason: RtrShutdownReason) -> Result<()> { + info!( + "RTR session closing due to service shutdown: reason={}, {}", + reason.as_str(), + self.session_summary() + ); + + if self.state == SessionState::Established { + if let (Some(version), Some(code)) = (self.version, shutdown_error_code(reason)) { + let text = shutdown_error_text(reason); + let _ = self.send_error(version, code, &[], text.as_bytes()).await; + } + } + + self.state = SessionState::Closed; + Ok(()) + } + async fn handle_first_pdu(&mut self, header: Header) -> Result<()> { info!( "RTR session processing first PDU: pdu={}, version={}, length={}", @@ -974,14 +993,14 @@ where } async fn send_payloads(&mut self, payloads: &[Payload], announce: bool) -> Result<()> { - // draft-ietf-sidrops-8210bis-25 Section 11.4 / 12 define Ordering Error + // draft-ietf-sidrops-8210bis-26 Section 11.4 / 12 define Ordering Error // as a receiver-side response to PDUs received in the wrong order. // When we detect an ordering issue before sending, this is a local cache/ // implementation fault rather than a received protocol error, so we fail // the session internally instead of emitting ErrorReport(code=11). // References: - // https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-25#section-11.4 - // https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-25#section-12 + // https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-26#section-11.4 + // https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-26#section-12 validate_payloads_for_rtr(payloads, announce).map_err(|err| anyhow!(err.to_string()))?; let version = self.version()?; let (route_origins, router_keys, aspas) = count_payloads(payloads); @@ -1003,13 +1022,13 @@ where async fn send_delta(&mut self, delta: &Delta) -> Result<()> { let updates = delta.payload_updates_for_rtr(); - // draft-ietf-sidrops-8210bis-25 Section 11.4 / 12 define Ordering Error + // draft-ietf-sidrops-8210bis-26 Section 11.4 / 12 define Ordering Error // for the party receiving out-of-order PDUs. A validator failure here // means we are about to send an invalid sequence, so abort locally // instead of reporting Ordering Error back to the router. // References: - // https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-25#section-11.4 - // https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-25#section-12 + // https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-26#section-11.4 + // https://datatracker.ietf.org/doc/html/draft-ietf-sidrops-8210bis-26#section-12 validate_payload_update_refs_for_rtr(&updates).map_err(|err| anyhow!(err.to_string()))?; let version = self.version()?; let (announced, withdrawn, route_origins, router_keys, aspas) = @@ -1158,6 +1177,19 @@ where offending_pdu: &[u8], text: &[u8], ) -> io::Result<()> { + if !code.is_defined_in_version(version) { + warn_rate_limited("session_error_report_suppressed_for_version", || { + format!( + "RTR session suppressed ErrorReport not defined for protocol version: version={}, error_code={}({}), offending_pdu_len={}", + version, + code.as_u16(), + code.description(), + offending_pdu.len() + ) + }); + return Ok(()); + } + let text_preview = String::from_utf8_lossy(text); warn_rate_limited("session_error_report_sent", || { format!( @@ -1328,6 +1360,22 @@ where } } +fn shutdown_error_code(reason: RtrShutdownReason) -> Option { + match reason { + RtrShutdownReason::CloseOnly => None, + RtrShutdownReason::CacheRestart => Some(ErrorCode::CacheRestart), + RtrShutdownReason::CacheShutdown => Some(ErrorCode::CacheShutdown), + } +} + +fn shutdown_error_text(reason: RtrShutdownReason) -> &'static str { + match reason { + RtrShutdownReason::CloseOnly => "RTR service closing", + RtrShutdownReason::CacheRestart => "RTR cache restarting", + RtrShutdownReason::CacheShutdown => "RTR cache shutting down", + } +} + fn count_payloads(payloads: &[Payload]) -> (usize, usize, usize) { let mut route_origins = 0; let mut router_keys = 0; diff --git a/tests/test_ccr.rs b/tests/test_ccr.rs index 0f09c77..422d981 100644 --- a/tests/test_ccr.rs +++ b/tests/test_ccr.rs @@ -92,7 +92,7 @@ fn snapshot_to_payloads_with_options_skips_invalid_aspa_when_not_strict() { }], vaps: vec![ParsedAspa { customer_asn: 174, - provider_asns: vec![0], + provider_asns: vec![0, 64496], }], }; @@ -102,29 +102,5 @@ fn snapshot_to_payloads_with_options_skips_invalid_aspa_when_not_strict() { assert_eq!(conversion.payloads.len(), 1); assert!(conversion.invalid_vrps.is_empty()); assert_eq!(conversion.invalid_vaps.len(), 1); - assert!(conversion.invalid_vaps[0].contains("provider list must not contain AS0")); -} - -#[test] -fn generated_mini_ccr_files_are_parseable() { - let cases = [ - ("20260403T000001Z-mini-a.ccr", 2usize, 1usize), - ("20260403T000101Z-mini-b.ccr", 3usize, 1usize), - ("20260403T000201Z-mini-c.ccr", 2usize, 2usize), - ]; - - for (name, expect_vrps, expect_vaps) in cases { - let snapshot = load_ccr_snapshot_from_file(fixture_path(name)) - .unwrap_or_else(|e| panic!("failed to parse {}: {:?}", name, e)); - assert_eq!( - snapshot.vrps.len(), - expect_vrps, - "vrp count mismatch for {name}" - ); - assert_eq!( - snapshot.vaps.len(), - expect_vaps, - "vap count mismatch for {name}" - ); - } + assert!(conversion.invalid_vaps[0].contains("provider list containing AS0")); } diff --git a/tests/test_pdu.rs b/tests/test_pdu.rs index 5a138d6..433b63d 100644 --- a/tests/test_pdu.rs +++ b/tests/test_pdu.rs @@ -217,9 +217,12 @@ fn aspa_announcement_rejects_as0() { assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); assert!(err.to_string().contains("AS0")); - let err = PayloadAspa::new(Asn::from(64496u32), vec![Asn::from(0u32)]) - .validate_announcement() - .unwrap_err(); + let err = PayloadAspa::new( + Asn::from(64496u32), + vec![Asn::from(0u32), Asn::from(64497u32)], + ) + .validate_announcement() + .unwrap_err(); assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); assert!(err.to_string().contains("AS0")); } diff --git a/tests/test_rtr_debug_client_ssh_cli.rs b/tests/test_rtr_debug_client_ssh_cli.rs index c68881c..e2a5352 100644 --- a/tests/test_rtr_debug_client_ssh_cli.rs +++ b/tests/test_rtr_debug_client_ssh_cli.rs @@ -32,7 +32,9 @@ fn ssh_requires_key() { ]); assert!(!output.status.success()); let stderr = String::from_utf8_lossy(&output.stderr); - assert!(stderr.contains("SSH mode requires --ssh-key ")); + assert!(stderr.contains("SSH mode requires authentication")); + assert!(stderr.contains("--ssh-key ")); + assert!(stderr.contains("--ssh-password ")); } #[test] diff --git a/tests/test_rtr_error_codes.rs b/tests/test_rtr_error_codes.rs new file mode 100644 index 0000000..d8bfea9 --- /dev/null +++ b/tests/test_rtr_error_codes.rs @@ -0,0 +1,146 @@ +use rpki::rtr::error_type::{ + ERROR_CODE_VERSION_SPECS, ErrorCode, ErrorCodeDisposition, ErrorCodeVersionSpec, +}; + +#[test] +fn error_code_version_specs_match_rfc_and_draft_table() { + use ErrorCode::*; + use ErrorCodeDisposition::*; + + let expected = [ + ErrorCodeVersionSpec { + code: CorruptData, + min_version: 0, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: InternalError, + min_version: 0, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: NoDataAvailable, + min_version: 0, + max_version: 2, + disposition: NonFatal, + }, + ErrorCodeVersionSpec { + code: InvalidRequest, + min_version: 0, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: UnsupportedProtocolVersion, + min_version: 0, + max_version: 1, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: UnsupportedProtocolVersion, + min_version: 2, + max_version: 2, + disposition: NonFatal, + }, + ErrorCodeVersionSpec { + code: UnsupportedPduType, + min_version: 0, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: WithdrawalOfUnknownRecord, + min_version: 0, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: DuplicateAnnouncement, + min_version: 0, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: UnexpectedProtocolVersion, + min_version: 1, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: AspaProviderListError, + min_version: 2, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: TransportFailed, + min_version: 2, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: OrderingError, + min_version: 2, + max_version: 2, + disposition: Fatal, + }, + ErrorCodeVersionSpec { + code: CacheRestart, + min_version: 2, + max_version: 2, + disposition: NonFatal, + }, + ErrorCodeVersionSpec { + code: CacheShutdown, + min_version: 2, + max_version: 2, + disposition: Fatal, + }, + ]; + + assert_eq!(ERROR_CODE_VERSION_SPECS, &expected); +} + +#[test] +fn error_code_definition_scope_is_version_sensitive() { + assert!(ErrorCode::DuplicateAnnouncement.is_defined_in_version(0)); + assert!(!ErrorCode::UnexpectedProtocolVersion.is_defined_in_version(0)); + assert!(ErrorCode::UnexpectedProtocolVersion.is_defined_in_version(1)); + assert!(!ErrorCode::AspaProviderListError.is_defined_in_version(1)); + assert!(ErrorCode::CacheShutdown.is_defined_in_version(2)); +} + +#[test] +fn error_code_disposition_is_version_sensitive() { + assert_eq!( + ErrorCode::NoDataAvailable.disposition_in_version(0), + Some(ErrorCodeDisposition::NonFatal) + ); + assert_eq!( + ErrorCode::UnsupportedProtocolVersion.disposition_in_version(1), + Some(ErrorCodeDisposition::Fatal) + ); + assert_eq!( + ErrorCode::UnsupportedProtocolVersion.disposition_in_version(2), + Some(ErrorCodeDisposition::NonFatal) + ); + assert_eq!( + ErrorCode::CacheRestart.disposition_in_version(2), + Some(ErrorCodeDisposition::NonFatal) + ); + assert_eq!(ErrorCode::CacheRestart.disposition_in_version(1), None); + assert_eq!( + ErrorCode::CacheShutdown.disposition_in_version(2), + Some(ErrorCodeDisposition::Fatal) + ); +} + +#[test] +fn raw_error_code_registry_mapping_accepts_defined_codes_only() { + assert_eq!(ErrorCode::try_from(12), Ok(ErrorCode::CacheRestart)); + assert_eq!(ErrorCode::try_from(13), Ok(ErrorCode::CacheShutdown)); + assert!(ErrorCode::try_from(14).is_err()); + assert!(ErrorCode::try_from(255).is_err()); +} diff --git a/tests/test_server_transports.rs b/tests/test_server_transports.rs index c75017d..c4627c1 100644 --- a/tests/test_server_transports.rs +++ b/tests/test_server_transports.rs @@ -16,8 +16,8 @@ use tokio_rustls::TlsConnector; use rpki::rtr::cache::{RtrCacheBuilder, SessionIds, SharedRtrCache}; use rpki::rtr::payload::Timing; use rpki::rtr::pdu::{CacheResponse, EndOfDataV1, ResetQuery}; -use rpki::rtr::server::RtrService; use rpki::rtr::server::ssh::SshAuthMode; +use rpki::rtr::server::{RtrService, RtrShutdownReason}; use russh::client; use russh::keys; use russh::keys::ssh_key::LineEnding; @@ -135,7 +135,7 @@ async fn unified_server_tcp_handles_reset_query() { assert_eq!(eod.session_id(), 42); assert_eq!(eod.serial_number(), 100); - running.shutdown(); + running.shutdown(RtrShutdownReason::CloseOnly); running.wait().await; } @@ -171,7 +171,7 @@ async fn unified_server_tls_handles_reset_query() { assert_eq!(eod.session_id(), 42); assert_eq!(eod.serial_number(), 100); - running.shutdown(); + running.shutdown(RtrShutdownReason::CloseOnly); running.wait().await; } @@ -220,7 +220,7 @@ async fn unified_server_ssh_opens_listener_and_emits_banner() { banner ); - running.shutdown(); + running.shutdown(RtrShutdownReason::CloseOnly); running.wait().await; } @@ -295,7 +295,7 @@ async fn unified_server_ssh_accepts_password_when_configured() { assert_eq!(eod.session_id(), 42); assert_eq!(eod.serial_number(), 100); - running.shutdown(); + running.shutdown(RtrShutdownReason::CloseOnly); running.wait().await; } @@ -344,6 +344,6 @@ async fn unified_server_ssh_rejects_password_when_not_configured() { .expect("password auth result"); assert!(!auth_result.success(), "password auth should be rejected"); - running.shutdown(); + running.shutdown(RtrShutdownReason::CloseOnly); running.wait().await; } diff --git a/tests/test_session.rs b/tests/test_session.rs index 2a1b881..398ce06 100644 --- a/tests/test_session.rs +++ b/tests/test_session.rs @@ -34,6 +34,7 @@ use rpki::rtr::pdu::{ }; use rpki::rtr::server::connection::handle_tls_connection; use rpki::rtr::server::tls::load_rustls_server_config_with_options; +use rpki::rtr::server::{RtrServiceControl, RtrShutdownReason}; use rpki::rtr::session::RtrSession; use rpki::rtr::store::RtrStore; @@ -46,14 +47,14 @@ async fn start_session_server( ) -> ( SocketAddr, broadcast::Sender<()>, - watch::Sender, + watch::Sender, JoinHandle<()>, ) { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let (notify_tx, notify_rx) = broadcast::channel(16); - let (shutdown_tx, shutdown_rx) = watch::channel(false); + let (shutdown_tx, shutdown_rx) = watch::channel(RtrServiceControl::Running); let handle = tokio::spawn(async move { let Ok((stream, _)) = listener.accept().await else { @@ -73,14 +74,14 @@ async fn start_session_server_with_transport_timeout( ) -> ( SocketAddr, broadcast::Sender<()>, - watch::Sender, + watch::Sender, JoinHandle<()>, ) { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let (notify_tx, notify_rx) = broadcast::channel(16); - let (shutdown_tx, shutdown_rx) = watch::channel(false); + let (shutdown_tx, shutdown_rx) = watch::channel(RtrServiceControl::Running); let handle = tokio::spawn(async move { let Ok((stream, _)) = listener.accept().await else { @@ -99,14 +100,14 @@ async fn start_session_server_returning_result( cache: SharedRtrCache, ) -> ( SocketAddr, - watch::Sender, + watch::Sender, JoinHandle>, ) { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let (_notify_tx, notify_rx) = broadcast::channel(16); - let (shutdown_tx, shutdown_rx) = watch::channel(false); + let (shutdown_tx, shutdown_rx) = watch::channel(RtrServiceControl::Running); let handle = tokio::spawn(async move { let (stream, _) = listener.accept().await.unwrap(); @@ -119,7 +120,7 @@ async fn start_session_server_returning_result( async fn start_tls_session_server( cache: SharedRtrCache, -) -> (SocketAddr, watch::Sender, JoinHandle<()>) { +) -> (SocketAddr, watch::Sender, JoinHandle<()>) { start_tls_session_server_with_cert(cache, "server.crt", "server.key").await } @@ -127,12 +128,12 @@ async fn start_tls_session_server_with_cert( cache: SharedRtrCache, cert_name: &str, key_name: &str, -) -> (SocketAddr, watch::Sender, JoinHandle<()>) { +) -> (SocketAddr, watch::Sender, JoinHandle<()>) { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let (_notify_tx, notify_rx) = broadcast::channel(16); - let (shutdown_tx, shutdown_rx) = watch::channel(false); + let (shutdown_tx, shutdown_rx) = watch::channel(RtrServiceControl::Running); let tls_config = Arc::new( load_rustls_server_config_with_options( fixture_path(cert_name), @@ -167,19 +168,22 @@ async fn start_tls_session_server_with_cert( async fn shutdown_server( mut client: TcpStream, - shutdown_tx: watch::Sender, + shutdown_tx: watch::Sender, server_handle: JoinHandle<()>, ) { shutdown_io(&mut client, shutdown_tx, server_handle).await; } -async fn shutdown_io(io: &mut S, shutdown_tx: watch::Sender, server_handle: JoinHandle<()>) -where +async fn shutdown_io( + io: &mut S, + shutdown_tx: watch::Sender, + server_handle: JoinHandle<()>, +) where S: AsyncWrite + Unpin, { let _ = io.shutdown().await; - let _ = shutdown_tx.send(true); + let _ = shutdown_tx.send(RtrServiceControl::Shutdown(RtrShutdownReason::CloseOnly)); match timeout(Duration::from_secs(1), server_handle).await { Ok(join_res) => { @@ -252,7 +256,7 @@ async fn connect_tls_client_with_server_name( connector.connect(server_name, tcp).await.unwrap() } -/// 用于 dump Serial Notify，保持输出风格一致。 +/// Dump Serial Notify in the same style as the other test helpers. fn dump_serial_notify(notify: &SerialNotify) -> serde_json::Value { json!({ "version": notify.version(), @@ -310,7 +314,7 @@ async fn wait_for_store_serials(store: &RtrStore, expected: [u32; 3]) { ); } -/// 测试：Reset Query 会返回完整 snapshot，并以 End of Data 结束响应。 +/// Reset Query returns a full snapshot and finishes with End of Data. #[tokio::test] async fn reset_query_returns_snapshot_and_end_of_data() { let prefix = IPAddressPrefix { @@ -397,6 +401,70 @@ async fn reset_query_uses_version_specific_session_id() { shutdown_server(client, shutdown_tx, server_handle).await; } +#[tokio::test] +async fn version_two_cache_restart_shutdown_sends_error_code_12() { + let cache = RtrCacheBuilder::new() + .session_ids(SessionIds::from_array([40, 41, 42])) + .serials(serials_all(100)) + .timing(Timing::new(600, 600, 7200)) + .build(); + + let server_cache = shared_cache(cache); + let (addr, _notify_tx, shutdown_tx, server_handle) = start_session_server(server_cache).await; + + let mut client = TcpStream::connect(addr).await.unwrap(); + ResetQuery::new(2).write(&mut client).await.unwrap(); + let _response = CacheResponse::read(&mut client).await.unwrap(); + let _eod = EndOfDataV1::read(&mut client).await.unwrap(); + + shutdown_tx + .send(RtrServiceControl::Shutdown(RtrShutdownReason::CacheRestart)) + .unwrap(); + + let report = ErrorReport::read(&mut client).await.unwrap(); + assert_error_report_matches(&report, 2, ErrorCode::CacheRestart, &[]); + assert!(String::from_utf8_lossy(report.text()).contains("restarting")); + + timeout(Duration::from_secs(1), server_handle) + .await + .expect("server task did not exit within timeout") + .unwrap(); +} + +#[tokio::test] +async fn version_one_cache_restart_shutdown_does_not_send_v2_error_code() { + let cache = RtrCacheBuilder::new() + .session_ids(SessionIds::from_array([40, 41, 42])) + .serials(serials_all(100)) + .timing(Timing::new(600, 600, 7200)) + .build(); + + let server_cache = shared_cache(cache); + let (addr, _notify_tx, shutdown_tx, server_handle) = start_session_server(server_cache).await; + + let mut client = TcpStream::connect(addr).await.unwrap(); + ResetQuery::new(1).write(&mut client).await.unwrap(); + let _response = CacheResponse::read(&mut client).await.unwrap(); + let _eod = EndOfDataV1::read(&mut client).await.unwrap(); + + shutdown_tx + .send(RtrServiceControl::Shutdown(RtrShutdownReason::CacheRestart)) + .unwrap(); + + let read_res = timeout(Duration::from_secs(1), Header::read(&mut client)) + .await + .expect("timed out waiting for session close"); + assert!( + read_res.is_err(), + "v1 session must not receive v2 error code 12" + ); + + timeout(Duration::from_secs(1), server_handle) + .await + .expect("server task did not exit within timeout") + .unwrap(); +} + #[tokio::test] async fn restart_restores_versioned_state_and_serves_queries() { let dir = tempfile::tempdir().unwrap(); @@ -488,7 +556,7 @@ async fn restart_restores_versioned_state_and_serves_queries() { shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：当 Serial Query 的 session_id 和 serial 都与当前 cache 一致时，仅返回 End of Data。 +/// Serial Query returns only End of Data when the client is up to date. #[tokio::test] async fn serial_query_returns_end_of_data_when_up_to_date() { let cache = RtrCacheBuilder::new() @@ -534,7 +602,7 @@ async fn serial_query_returns_end_of_data_when_up_to_date() { shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：当已建立 session 后收到错误的 session_id 时，返回 CorruptData 并关闭连接。 +/// Established sessions reject mismatched Session IDs with Corrupt Data. #[tokio::test] async fn serial_query_returns_corrupt_data_when_session_id_mismatch() { let cache = RtrCacheBuilder::new() @@ -581,7 +649,7 @@ async fn serial_query_returns_corrupt_data_when_session_id_mismatch() { shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：当增量更新可用时，Serial Query 返回 Cache Response + delta payload + End of Data。 +/// Serial Query returns Cache Response, delta payloads, and End of Data when incremental updates exist. #[tokio::test] async fn serial_query_returns_deltas_when_incremental_update_available() { let prefix = IPAddressPrefix { @@ -751,7 +819,7 @@ async fn serial_query_returns_cache_reset_for_future_serial_across_wraparound() shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：Reset Query 返回的 payload 顺序符合当前实现的 RTR 排序规则。 +/// Reset Query returns payloads in RTR ordering. #[tokio::test] async fn reset_query_returns_payloads_in_rtr_order() { let v4_low_prefix = IPAddressPrefix { @@ -850,7 +918,7 @@ async fn reset_query_returns_payloads_in_rtr_order() { shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：Serial Query 返回的增量中，announcement 在前，withdrawal 在后，且各自内部顺序符合当前实现。 +/// Serial Query returns announcements before withdrawals, with each side sorted. #[tokio::test] async fn serial_query_returns_announcements_before_withdrawals() { let announced_low_prefix = IPAddressPrefix { @@ -984,7 +1052,7 @@ async fn serial_query_returns_announcements_before_withdrawals() { shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：session 建立后，收到 notify 广播时会发送 Serial Notify。 +/// Established sessions send Serial Notify after receiving cache update notifications. #[tokio::test] async fn established_session_sends_serial_notify() { let cache = RtrCacheBuilder::new() @@ -1026,7 +1094,7 @@ async fn established_session_sends_serial_notify() { shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：首个 PDU 版本过高时，返回 UnsupportedProtocolVersion 错误并关闭连接。 +/// A too-high first-PDU protocol version returns Unsupported Protocol Version. #[tokio::test] async fn first_pdu_with_too_high_version_returns_unsupported_version_error() { let cache = RtrCacheBuilder::new() @@ -1066,7 +1134,7 @@ async fn first_pdu_with_too_high_version_returns_unsupported_version_error() { shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：版本协商完成后，如果后续请求更换了协议版本，返回 UnexpectedProtocolVersion 并关闭连接。 +/// Established sessions reject later protocol version changes. #[tokio::test] async fn session_rejects_version_change_after_negotiation() { let cache = RtrCacheBuilder::new() @@ -1119,7 +1187,7 @@ async fn session_rejects_version_change_after_negotiation() { shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：在版本协商完成前，即使收到 notify 广播，也不能发送 Serial Notify。 +/// Notifications before version negotiation do not send Serial Notify. #[tokio::test] async fn notify_is_not_sent_before_version_negotiation() { let cache = RtrCacheBuilder::new() @@ -1156,7 +1224,7 @@ async fn notify_is_not_sent_before_version_negotiation() { shutdown_server(client, shutdown_tx, server_handle).await; } -/// 测试：同一 session 在一分钟窗口内连续收到 notify 广播时，只会发送一个 Serial Notify。 +/// Serial Notify is rate-limited to once per minute per session. #[tokio::test] async fn serial_notify_is_rate_limited_to_once_per_minute() { let cache = RtrCacheBuilder::new() @@ -1908,7 +1976,7 @@ async fn invalid_timing_prevents_end_of_data_response() { "server should close instead of sending invalid EndOfData" ); - let _ = shutdown_tx.send(true); + let _ = shutdown_tx.send(RtrServiceControl::Shutdown(RtrShutdownReason::CloseOnly)); let join = timeout(Duration::from_secs(1), server_handle) .await .expect("server task did not exit within timeout") @@ -1947,7 +2015,7 @@ async fn invalid_aspa_prevents_snapshot_response() { "server should close instead of sending invalid ASPA" ); - let _ = shutdown_tx.send(true); + let _ = shutdown_tx.send(RtrServiceControl::Shutdown(RtrShutdownReason::CloseOnly)); let join = timeout(Duration::from_secs(1), server_handle) .await .expect("server task did not exit within timeout") @@ -1989,7 +2057,7 @@ async fn invalid_router_key_prevents_snapshot_response() { "server should close instead of sending invalid RouterKey" ); - let _ = shutdown_tx.send(true); + let _ = shutdown_tx.send(RtrServiceControl::Shutdown(RtrShutdownReason::CloseOnly)); let join = timeout(Duration::from_secs(1), server_handle) .await .expect("server task did not exit within timeout")