v0.0.8: Server-side message deduplication
Server: - DedupTracker in AppState: bounded HashSet (10,000 IDs, FIFO eviction) - send_message: extracts message ID from bincode, drops duplicates - WS handler: dedup on both binary and JSON message frames - extract_message_id() parses all WireMessage variants Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
10
warzone/Cargo.lock
generated
10
warzone/Cargo.lock
generated
@@ -2647,7 +2647,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "warzone-client"
|
name = "warzone-client"
|
||||||
version = "0.0.7"
|
version = "0.0.8"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"argon2",
|
"argon2",
|
||||||
@@ -2680,7 +2680,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "warzone-mule"
|
name = "warzone-mule"
|
||||||
version = "0.0.7"
|
version = "0.0.8"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"clap",
|
"clap",
|
||||||
@@ -2689,7 +2689,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "warzone-protocol"
|
name = "warzone-protocol"
|
||||||
version = "0.0.7"
|
version = "0.0.8"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"base64",
|
"base64",
|
||||||
"bincode",
|
"bincode",
|
||||||
@@ -2712,7 +2712,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "warzone-server"
|
name = "warzone-server"
|
||||||
version = "0.0.7"
|
version = "0.0.8"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"axum",
|
"axum",
|
||||||
@@ -2739,7 +2739,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "warzone-wasm"
|
name = "warzone-wasm"
|
||||||
version = "0.0.7"
|
version = "0.0.8"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"base64",
|
"base64",
|
||||||
"bincode",
|
"bincode",
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ members = [
|
|||||||
]
|
]
|
||||||
|
|
||||||
[workspace.package]
|
[workspace.package]
|
||||||
version = "0.0.7"
|
version = "0.0.8"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
|
|||||||
@@ -4,10 +4,26 @@ use axum::{
|
|||||||
Json, Router,
|
Json, Router,
|
||||||
};
|
};
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
|
use warzone_protocol::message::WireMessage;
|
||||||
|
|
||||||
use crate::errors::AppResult;
|
use crate::errors::AppResult;
|
||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
|
|
||||||
|
/// Try to extract the message ID from raw bincode-serialized WireMessage bytes.
|
||||||
|
fn extract_message_id(data: &[u8]) -> Option<String> {
|
||||||
|
if let Ok(wire) = bincode::deserialize::<WireMessage>(data) {
|
||||||
|
match wire {
|
||||||
|
WireMessage::KeyExchange { id, .. } => Some(id),
|
||||||
|
WireMessage::Message { id, .. } => Some(id),
|
||||||
|
WireMessage::FileHeader { id, .. } => Some(id),
|
||||||
|
WireMessage::FileChunk { id, .. } => Some(id),
|
||||||
|
WireMessage::Receipt { message_id, .. } => Some(message_id),
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Touch the alias TTL for a fingerprint (renew on authenticated action).
|
/// Touch the alias TTL for a fingerprint (renew on authenticated action).
|
||||||
pub fn renew_alias_ttl(db: &sled::Tree, fp: &str) {
|
pub fn renew_alias_ttl(db: &sled::Tree, fp: &str) {
|
||||||
let alias_key = format!("fp:{}", fp);
|
let alias_key = format!("fp:{}", fp);
|
||||||
@@ -55,6 +71,14 @@ async fn send_message(
|
|||||||
) -> AppResult<Json<serde_json::Value>> {
|
) -> AppResult<Json<serde_json::Value>> {
|
||||||
let to = normalize_fp(&req.to);
|
let to = normalize_fp(&req.to);
|
||||||
|
|
||||||
|
// Dedup: if we have already seen this message ID, silently drop it
|
||||||
|
if let Some(msg_id) = extract_message_id(&req.message) {
|
||||||
|
if state.dedup.check_and_insert(&msg_id) {
|
||||||
|
tracing::debug!("Dedup: dropping duplicate message {}", msg_id);
|
||||||
|
return Ok(Json(serde_json::json!({ "ok": true })));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Try WebSocket push first (instant delivery)
|
// Try WebSocket push first (instant delivery)
|
||||||
if state.push_to_client(&to, &req.message).await {
|
if state.push_to_client(&to, &req.message).await {
|
||||||
tracing::info!("Pushed message to {} via WS ({} bytes)", to, req.message.len());
|
tracing::info!("Pushed message to {} via WS ({} bytes)", to, req.message.len());
|
||||||
|
|||||||
@@ -17,9 +17,25 @@ use axum::{
|
|||||||
Router,
|
Router,
|
||||||
};
|
};
|
||||||
use futures_util::{SinkExt, StreamExt};
|
use futures_util::{SinkExt, StreamExt};
|
||||||
|
use warzone_protocol::message::WireMessage;
|
||||||
|
|
||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
|
|
||||||
|
/// Try to extract the message ID from raw bincode-serialized WireMessage bytes.
|
||||||
|
fn extract_message_id(data: &[u8]) -> Option<String> {
|
||||||
|
if let Ok(wire) = bincode::deserialize::<WireMessage>(data) {
|
||||||
|
match wire {
|
||||||
|
WireMessage::KeyExchange { id, .. } => Some(id),
|
||||||
|
WireMessage::Message { id, .. } => Some(id),
|
||||||
|
WireMessage::FileHeader { id, .. } => Some(id),
|
||||||
|
WireMessage::FileChunk { id, .. } => Some(id),
|
||||||
|
WireMessage::Receipt { message_id, .. } => Some(message_id),
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub fn routes() -> Router<AppState> {
|
pub fn routes() -> Router<AppState> {
|
||||||
Router::new().route("/ws/:fingerprint", get(ws_handler))
|
Router::new().route("/ws/:fingerprint", get(ws_handler))
|
||||||
}
|
}
|
||||||
@@ -90,6 +106,14 @@ async fn handle_socket(socket: WebSocket, state: AppState, fingerprint: String)
|
|||||||
let to_fp = normalize_fp(&header);
|
let to_fp = normalize_fp(&header);
|
||||||
let message = &data[64..];
|
let message = &data[64..];
|
||||||
|
|
||||||
|
// Dedup: skip if we already processed this message ID
|
||||||
|
if let Some(msg_id) = extract_message_id(message) {
|
||||||
|
if state_clone.dedup.check_and_insert(&msg_id) {
|
||||||
|
tracing::debug!("WS dedup: dropping duplicate binary message {}", msg_id);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Try push to connected client first
|
// Try push to connected client first
|
||||||
if !state_clone.push_to_client(&to_fp, message).await {
|
if !state_clone.push_to_client(&to_fp, message).await {
|
||||||
// Queue in DB
|
// Queue in DB
|
||||||
@@ -110,6 +134,14 @@ async fn handle_socket(socket: WebSocket, state: AppState, fingerprint: String)
|
|||||||
.filter_map(|v| v.as_u64().map(|n| n as u8))
|
.filter_map(|v| v.as_u64().map(|n| n as u8))
|
||||||
.collect();
|
.collect();
|
||||||
|
|
||||||
|
// Dedup: skip if we already processed this message ID
|
||||||
|
if let Some(msg_id) = extract_message_id(&message) {
|
||||||
|
if state_clone.dedup.check_and_insert(&msg_id) {
|
||||||
|
tracing::debug!("WS dedup: dropping duplicate JSON message {}", msg_id);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if !state_clone.push_to_client(&to_fp, &message).await {
|
if !state_clone.push_to_client(&to_fp, &message).await {
|
||||||
let key = format!("queue:{}:{}", to_fp, uuid::Uuid::new_v4());
|
let key = format!("queue:{}:{}", to_fp, uuid::Uuid::new_v4());
|
||||||
let _ = state_clone.db.messages.insert(key.as_bytes(), message);
|
let _ = state_clone.db.messages.insert(key.as_bytes(), message);
|
||||||
|
|||||||
@@ -1,19 +1,57 @@
|
|||||||
use std::collections::HashMap;
|
use std::collections::{HashMap, HashSet, VecDeque};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use tokio::sync::{Mutex, mpsc};
|
use tokio::sync::{Mutex, mpsc};
|
||||||
|
|
||||||
use crate::db::Database;
|
use crate::db::Database;
|
||||||
|
|
||||||
|
/// Maximum number of message IDs to track for deduplication.
|
||||||
|
const DEDUP_CAPACITY: usize = 10_000;
|
||||||
|
|
||||||
/// Per-connection sender: messages are pushed here for instant delivery.
|
/// Per-connection sender: messages are pushed here for instant delivery.
|
||||||
pub type WsSender = mpsc::UnboundedSender<Vec<u8>>;
|
pub type WsSender = mpsc::UnboundedSender<Vec<u8>>;
|
||||||
|
|
||||||
/// Connected clients: fingerprint → list of WS senders (multiple devices).
|
/// Connected clients: fingerprint → list of WS senders (multiple devices).
|
||||||
pub type Connections = Arc<Mutex<HashMap<String, Vec<WsSender>>>>;
|
pub type Connections = Arc<Mutex<HashMap<String, Vec<WsSender>>>>;
|
||||||
|
|
||||||
|
/// Bounded dedup tracker: FIFO eviction when capacity is exceeded.
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct DedupTracker {
|
||||||
|
seen: Arc<std::sync::Mutex<HashSet<String>>>,
|
||||||
|
order: Arc<std::sync::Mutex<VecDeque<String>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl DedupTracker {
|
||||||
|
pub fn new() -> Self {
|
||||||
|
DedupTracker {
|
||||||
|
seen: Arc::new(std::sync::Mutex::new(HashSet::with_capacity(DEDUP_CAPACITY))),
|
||||||
|
order: Arc::new(std::sync::Mutex::new(VecDeque::with_capacity(DEDUP_CAPACITY))),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns `true` if this ID was already seen (i.e. it is a duplicate).
|
||||||
|
/// If new, inserts it and evicts the oldest if over capacity.
|
||||||
|
pub fn check_and_insert(&self, id: &str) -> bool {
|
||||||
|
let mut seen = self.seen.lock().unwrap();
|
||||||
|
if seen.contains(id) {
|
||||||
|
return true; // duplicate
|
||||||
|
}
|
||||||
|
let mut order = self.order.lock().unwrap();
|
||||||
|
if seen.len() >= DEDUP_CAPACITY {
|
||||||
|
if let Some(oldest) = order.pop_front() {
|
||||||
|
seen.remove(&oldest);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
seen.insert(id.to_string());
|
||||||
|
order.push_back(id.to_string());
|
||||||
|
false // not a duplicate
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub struct AppState {
|
pub struct AppState {
|
||||||
pub db: Arc<Database>,
|
pub db: Arc<Database>,
|
||||||
pub connections: Connections,
|
pub connections: Connections,
|
||||||
|
pub dedup: DedupTracker,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl AppState {
|
impl AppState {
|
||||||
@@ -22,6 +60,7 @@ impl AppState {
|
|||||||
Ok(AppState {
|
Ok(AppState {
|
||||||
db: Arc::new(db),
|
db: Arc::new(db),
|
||||||
connections: Arc::new(Mutex::new(HashMap::new())),
|
connections: Arc::new(Mutex::new(HashMap::new())),
|
||||||
|
dedup: DedupTracker::new(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user