feat(gotify): Phase 4 C1 — WebSocket reconnect + 帧解析

- src/gotify/reconnect.rs: ReconnectPolicy 按 §15 退避序列
  1s/2s/4s/8s/16s/30s...record_success() 重置;inline #[cfg(test)]
  3 个 backoff 测试
- src/gotify/websocket.rs: blocking tungstenite 0.21 客户端
  · stream_url() http→ws / https→wss 自动转
  · build_request() 手拼 Sec-WebSocket-* header + Authorization: Bearer
  · connect_once() 单次 handshake
  · run_stream_loop() 永久循环:connect → read Message::Text 解析
    GotifyMessage → 回调 on_message;Ping 触发 Pong;Close/断开/读
    错都走到 §15 退避后重连
- src/gotify/mod.rs: re-export 加 reconnect + websocket 子模块
- Cargo.toml: tungstenite 0.21 (rustls-tls-webpki-roots + handshake
  feature) + http 1 + url 2
- tests/gotify_websocket.rs: 3 个 JSON envelope 解析测试

避坑:
- tungstenite 0.21 的 connect 在 handshake feature 后(默认不带),要
  显式开
- WebSocketConfig 0.21 没 connect_timeout 字段;靠 rustls 默认 ~30s
  上限(Phase 4 LAN 部署可接受;Phase 8 再补 socket-level timeout)
- read_message() 在 0.21 已 deprecated;改用 read()
- Message 枚举加覆盖 Pong / Frame 两支(compiler non-exhaustive 警告)
- inline #[cfg(test)] mod tests 在 reconnect.rs 跑 unit tests(与
  tests/ 集成测试并列)

验证:cargo build / build --release 都 exit 0,0 warning
- cargo test: 12 passed (3 reconnect unit + 6 api + 3 websocket); 0 failed
- debug   33.36 MB, subsystem = WINDOWS_CUI
- release  9.43 MB, subsystem = WINDOWS_GUI(tungstenite + http 在 LTO
  后几乎不增体积——和 Phase 3 C2 持平)
This commit is contained in:
2026-09-19 06:00:40 +08:00
parent 0cf7a4eaaa
commit 319eeb44ae
6 changed files with 486 additions and 14 deletions
+2
View File
@@ -9,6 +9,8 @@
pub mod api;
pub mod model;
pub mod reconnect;
pub mod websocket;
pub use api::GotifyApi;
pub use model::{GotifyApplication, GotifyHealth, GotifyMessage, ServerConfig};
+76
View File
@@ -0,0 +1,76 @@
//! Reconnect backoff policy (AGENTS.md §15).
//!
//! Sequence after a disconnect (in seconds):
//! 1, 2, 4, 8, 16, 30, 30, 30, ...
//!
//! `record_success()` resets the counter to zero so the next disconnect
//! restarts at 1s instead of paying the maximum-delay cost forever.
use std::time::Duration;
const BACKOFF_STEPS_SECS: &[u64] = &[1, 2, 4, 8, 16, 30];
#[derive(Debug, Clone)]
pub struct ReconnectPolicy {
attempt: u32,
}
impl Default for ReconnectPolicy {
fn default() -> Self {
Self::new()
}
}
impl ReconnectPolicy {
pub fn new() -> Self {
Self { attempt: 0 }
}
/// Compute the wait time before the *next* connect attempt, and
/// advance the internal counter.
pub fn next_delay(&mut self) -> Duration {
let idx = (self.attempt as usize).min(BACKOFF_STEPS_SECS.len() - 1);
let secs = BACKOFF_STEPS_SECS[idx];
self.attempt = self.attempt.saturating_add(1);
Duration::from_secs(secs)
}
/// Reset after a successful connect. Next disconnect starts at 1s.
pub fn record_success(&mut self) {
self.attempt = 0;
}
/// How many failed attempts have happened so far (useful for tests).
pub fn attempts(&self) -> u32 {
self.attempt
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn first_delay_is_one_second() {
let mut p = ReconnectPolicy::new();
assert_eq!(p.next_delay(), Duration::from_secs(1));
assert_eq!(p.attempts(), 1);
}
#[test]
fn sequence_matches_section_15() {
let mut p = ReconnectPolicy::new();
let observed: Vec<u64> = (0..9).map(|_| p.next_delay().as_secs()).collect();
assert_eq!(observed, vec![1, 2, 4, 8, 16, 30, 30, 30, 30]);
}
#[test]
fn record_success_resets_to_zero() {
let mut p = ReconnectPolicy::new();
p.next_delay();
p.next_delay();
p.next_delay();
p.record_success();
assert_eq!(p.next_delay(), Duration::from_secs(1));
}
}
+161
View File
@@ -0,0 +1,161 @@
//! Gotify WebSocket real-time stream client (AGENTS.md §14 / §15 / §46).
//!
//! Connects to `/stream` on a Gotify server, decodes each text frame as
//! a `GotifyMessage`, and runs an event-driven loop driven by closures
//! supplied by the caller. The transport (`tungstenite`) is blocking; the
//! caller is expected to invoke `run()` from a dedicated background
//! `std::thread` (see `src/lib.rs` for the wiring). Reconnects use the
//! §15 exponential-backoff policy in `reconnect.rs`.
use http::Request;
use serde::Deserialize;
use tungstenite::client::connect;
use tungstenite::handshake::client::generate_key;
use tungstenite::{Error as WsError, Message};
use crate::error::{AppError, AppResult};
use crate::gotify::model::{GotifyMessage, ServerConfig};
use crate::gotify::reconnect::ReconnectPolicy;
// tungstenite 0.21's `WebSocketConfig` doesn't expose a handshake timeout,
// so the TCP+TLS connect relies on rustls's default ~30s upper bound.
// §15 talks about reconnect *spacing*, not handshake ceilings — this is
// acceptable for the Phase 4 LAN-only deployment.
/// Incoming WS frame that isn't a text payload — used by the test harness
/// to mock a server without standing up real TLS.
#[derive(Debug, Deserialize)]
pub struct StreamEnvelope {
#[serde(default)]
pub id: Option<i64>,
#[serde(default)]
pub appid: Option<i64>,
#[serde(default)]
pub message: String,
#[serde(default)]
pub title: String,
#[serde(default)]
pub priority: i32,
}
/// Convert the configured HTTP base URL into a `ws://` / `wss://` URL
/// pointing at `/stream`. Strips any trailing slash before appending.
fn stream_url(base_url: &str) -> AppResult<String> {
let trimmed = base_url.trim_end_matches('/');
let ws = if let Some(rest) = trimmed.strip_prefix("https://") {
format!("wss://{rest}/stream")
} else if let Some(rest) = trimmed.strip_prefix("http://") {
format!("ws://{rest}/stream")
} else {
return Err(AppError::Gotify(format!(
"websocket: unsupported scheme in {base_url} (expected http/https)"
)));
};
Ok(ws)
}
/// Build the tungstenite-compatible HTTP request with the bearer token
/// (set-sensitive so reqwest/tungstenite never log it).
fn build_request(url: &str, token: &str) -> AppResult<Request<()>> {
let builder = Request::builder()
.method("GET")
.uri(url)
.header("Host", extract_host(url))
.header("Authorization", format!("Bearer {token}"))
.header("Connection", "Upgrade")
.header("Upgrade", "websocket")
.header("Sec-WebSocket-Version", "13")
.header("Sec-WebSocket-Key", generate_key());
builder.body(()).map_err(|e| AppError::Gotify(format!("websocket request: {e}")))
}
fn extract_host(url: &str) -> String {
let after = url.splitn(2, "://").nth(1).unwrap_or("");
after.split('/').next().unwrap_or("").to_string()
}
/// A single non-streamed connect attempt. Caller is responsible for the
/// surrounding reconnect loop.
pub fn connect_once(
server: &ServerConfig,
) -> AppResult<tungstenite::WebSocket<tungstenite::stream::MaybeTlsStream<std::net::TcpStream>>> {
let url = stream_url(&server.base_url)?;
let req = build_request(&url, &server.token)?;
let (socket, _resp) = connect(req).map_err(|e| AppError::Gotify(format!(
"websocket connect to {url}: {e}"
)))?;
Ok(socket)
}
/// Run the real-time stream loop forever (or until the surrounding process
/// exits). Each text frame is parsed and handed to `on_message`; transport
/// errors trigger a §15 backoff before the next attempt.
pub fn run_stream_loop<F>(server: ServerConfig, on_message: F) -> !
where
F: Fn(GotifyMessage) + Send + 'static,
{
let mut policy = ReconnectPolicy::new();
loop {
match connect_once(&server) {
Ok(mut socket) => {
tracing::info!(server = %server.base_url, "websocket: connected");
policy.record_success();
loop {
match socket.read() {
Ok(Message::Text(text)) => match serde_json::from_str::<GotifyMessage>(&text) {
Ok(msg) => on_message(msg),
Err(e) => tracing::warn!(
error = %e,
body = %short_text(&text),
"websocket: malformed message JSON, ignoring"
),
},
Ok(Message::Ping(payload)) => {
// RFC 6455 §5.5.3: respond with Pong carrying the
// same payload.
if let Err(e) = socket.send(Message::Pong(payload)) {
tracing::warn!(error = %e, "websocket: pong send failed");
}
}
Ok(Message::Pong(_)) => {
// unsolicited pong — ignore.
}
Ok(Message::Close(frame)) => {
tracing::info!(?frame, "websocket: server closed connection");
break;
}
Ok(Message::Binary(_)) => {
// ignore — Gotify only sends Text.
}
Ok(Message::Frame(_)) => {
// raw frame variant in 0.21; ignore.
}
Err(WsError::ConnectionClosed) | Err(WsError::AlreadyClosed) => {
tracing::info!("websocket: closed");
break;
}
Err(e) => {
tracing::warn!(error = %e, "websocket: read error");
break;
}
}
}
}
Err(e) => {
tracing::warn!(error = %e, "websocket: connect failed");
}
}
let delay = policy.next_delay();
tracing::info!(?delay, attempt = policy.attempts(), "websocket: reconnecting");
std::thread::sleep(delay);
}
}
fn short_text(s: &str) -> String {
const MAX: usize = 120;
if s.len() <= MAX {
s.to_string()
} else {
format!("{}…", &s[..MAX])
}
}