From 532d61601f942e2a0f8bd00191ed2653d67906af Mon Sep 17 00:00:00 2001 From: Mathih13 Date: Fri, 16 Oct 2026 07:33:15 +0200 Subject: [PATCH 1/2] perf(gateway): release idle World Sessions from blocking workers --- docs/accept-policy.md | 8 +- docs/architecture.md | 69 +-- gateway/src/accept.rs | 13 + gateway/src/config.rs | 20 +- gateway/src/main.rs | 5 +- gateway/src/world/egress.rs | 339 +++++++++++ gateway/src/world/login_queue.rs | 15 + gateway/src/world/mod.rs | 317 +++++----- .../world/test_support/world_fake/session.rs | 5 + gateway/src/world/transport.rs | 555 ++++++++++++++++++ 10 files changed, 1166 insertions(+), 180 deletions(-) create mode 100644 gateway/src/world/egress.rs create mode 100644 gateway/src/world/transport.rs diff --git a/docs/accept-policy.md b/docs/accept-policy.md index 5034b52e6..a36fc7ded 100644 --- a/docs/accept-policy.md +++ b/docs/accept-policy.md @@ -21,5 +21,9 @@ a non-stream socket, cannot apply to a `TcpListener` created by `bind`. `EPERM` refusal for one connection and also retries. `EMFILE` reports descriptor exhaustion and must leave existing sessions alive while capacity recovers. -A shared non-waiting capacity limit keeps accepted sockets out of Tokio's unbounded blocking-task -queue. Holding the permit for the connection's work makes the limit cover the resource it protects. +Logon handshakes use a non-waiting permit limit before entering Tokio's blocking executor. +World sockets use asynchronous tasks and a separate accepted-connection limit, +`LYRACORE_MAX_WORLD_CONNECTIONS`. Each World Session submits at most one game action at a time +through the bounded worker pool shared with logon. Relay encoding has its own bounded pool. +Permits cover submitted and running work and return on completion or unwinding. Idle World Sessions +and authenticated queue waits hold no worker permit. diff --git a/docs/architecture.md b/docs/architecture.md index 1fa6e4bc3..f15c6506c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -311,58 +311,61 @@ gateway's environment. Omit one and you get a **working-looking single-database | `LYRACORE_AOI` | AOI-scoped subscriptions | **on** (`=0` disables) | — | | `LYRACORE_QUEST_LOG` | quest-log descriptor fields | **on** | — | | `LYRACORE_MAX_SESSIONS` / `LYRACORE_ADMIT_CONCURRENCY` | Established World Sessions / queued admissions per tick, per Gateway process. Divide the intended realm-wide limit by the Gateway count. | Unset or `0` = unlimited | Malformed values stop serving startup before Coordinator connection or listener binding. Surrounding whitespace is allowed; signs, separators, empty values and overflow are refused. Provision mode does not use these limits. | -| `LYRACORE_MAX_BLOCKING_THREADS` | tokio blocking-pool cap — **the real ceiling on concurrent players** (see below) | `512` (tokio's own default) | malformed **or zero** falls back to `512`; the resolved value is logged at startup | +| `LYRACORE_MAX_BLOCKING_THREADS` | Blocking game actions, logon handshakes and Relay encoding | `512` | Malformed or zero falls back to `512`; startup logs the resolved value. | +| `LYRACORE_MAX_WORLD_CONNECTIONS` | All accepted World sockets, including pre-auth and queued clients | `16384` | Malformed or zero falls back to `16384`; excess sockets close at accept. | | `LYRACORE_LOAD_SAMPLE_SECS` | shard load sampling cadence | `30` | — | | `LYRACORE_METRICS_DB_IDS` | `=` map for `/v1/metrics` | `""` | warns loudly at startup; occupancy unmeasured | | `LYRACORE_WRITER_TRACE` | per-session writer black-box ring | off | — | | `LYRACORE_TRANSFER_ABORT_AFTER` | crash-injection harness (aborts a named transfer step) | unset | an unknown step name logs an error and nothing fires | | `LYRACORE_PROFILE_SECS` | `--features dhat-heap` builds only | `120` | — | -Budget three client socket descriptors per World Session: reader, writer and ownership-loss -shutdown handle. Add the database sockets for its subscribed Shards and reserve headroom for -listeners, handshakes and Coordinator connections when setting `LYRACORE_MAX_SESSIONS`. +Each World Session uses one client socket descriptor. Add the shared Coordinator and call-pipe +sockets, listeners and logon handshakes when setting descriptor limits. -Two non-`LYRACORE_` variables matter: `RUST_LOG` (consumed by `env_logger` in `gateway/src/main.rs`) -and `MALLOC_ARENA_MAX` (glibc, not the binary — worth ~4× RSS per connection). +`RUST_LOG` controls logging through `env_logger`. `MALLOC_ARENA_MAX` controls glibc allocation arenas. -### `LYRACORE_MAX_BLOCKING_THREADS`, and the two limits above it (#451) +### World Session concurrency and limits -Both listeners hand admitted sockets to `spawn_blocking` because the `wow_login_messages` and -`wow_world_messages` codecs use blocking `std::io`. A **World Session holds its blocking thread for -the session's entire life**. So the blocking-pool cap is the seat count shared between live World -Sessions and in-flight logon handshakes. It was Tokio's inherited default of 512 until #451, -configured by nothing and reported by nothing. +World socket reads, writes and authenticated login-queue waits use asynchronous tasks. An idle +World Session holds no blocking worker. Each session submits one game action at a time to a bounded +pool shared with logon handshakes. Store interfaces remain synchronous, including Durable Request +completion waits. The pool bounds those active waits independently of connected player count. -That mattered because a full pool does not refuse: `spawn_blocking` **queues**. Before #410, a -socket could be accepted and then wait without a bound for a blocking thread, retaining its file -descriptor and sitting ahead of later logins. Both listeners now share a non-waiting capacity with -the same size as the pool. A connection that cannot take a permit is closed before its task is -submitted, and each task holds its permit through normal return or unwinding. This bounds submitted -plus running connection tasks at the configured pool size. Measured 2026-08-07 on an 8-core/15 GB -container, 600 clients offered: +Relay encoding has a separate permit pool, capped at eight workers and one quarter of +`LYRACORE_MAX_BLOCKING_THREADS`, with a minimum of one. Both pools use Tokio's blocking executor. +This reserves encoding capacity when game actions await database completion. A one-thread executor +still serializes both kinds of work. No capacity measurement for this implementation is recorded yet. -| `max_blocking_threads` | seated | failed | -|---|---|---| -| 512 (tokio default) | 477 | 123 | -| 4096 | 535 | 65 | +The World listener bounds accepted sockets independently through `LYRACORE_MAX_WORLD_CONNECTIONS`. +`LYRACORE_MAX_SESSIONS` controls admitted players and the authenticated queue. Neither value is a +measured player capacity. Logon still refuses a new handshake when it cannot obtain a worker permit. + +Pre-auth has a total deadline of ten seconds for logon and fifteen seconds for World. The World budget +includes socket reads, writes and the session lookup in one asynchronous timeout, then ends before the +authenticated queue. Queue writes have their own timeout. A queue seat returns on cancellation, +including when admission races with cancellation. -477 rather than a clean 512 because logon handshakes draw from the same pool. Raising it is worth -about 58 sessions on that host. (The next ceiling this section used to name — per-player -connection setup — is gone with the per-player connections themselves, #483.) +Each World Session has a non-waiting outbound queue capped at 2,048 entries and a 4 MiB reservation +budget. Raw frames reserve their wire size; each typed frame and deferred Relay job reserves 4 KiB until +encoding. Compression stays on encoding workers. Each encoded projection and combined write batch is +also limited to 4 MiB. +These are queue and encoded-output limits, not an exact bound on arbitrary captured Rust objects. +A queue entry has ten seconds to finish encoding and writing. Exceeding a limit closes the session; +critical packets are not silently discarded. Movement keeps its existing coalescing and shedding. -Each admitted socket also starts one absolute pre-auth I/O deadline at accept: 10 seconds for -logon and 15 seconds for world. Blocking reads and writes receive only the remaining time. An -async watchdog closes the socket at the same instant even when its task has not started. Successful -proof cancels the watchdog and clears both socket timeouts. World does this before the login queue, -so an authenticated queue wait does not spend the pre-auth budget. +One writer owns each session's header cipher and preserves FIFO order and contiguous entry batches. +It combines frames into writes of about 64 KiB. +A supervisor retains cleanup ownership when its caller is cancelled. Accepted game actions finish +before logout cleanup, so a delayed login cannot recreate a departed session after cleanup. A clean +read EOF drains queued replies within the outbound age budget before returning the admission seat. Two related startup behaviours have no environment variable, because there is no case for turning them off: - **`RLIMIT_NOFILE` soft is raised to hard** before the runtime is built (`gateway/src/fd_limit.rs`). A stock Docker container is soft 1024 / hard 524288, and #447 died at ~200 sessions with - `Too many open files (os error 24)` against 512× of unclaimed headroom. A live session costs 3–5 - descriptors. Failure logs and continues — it never blocks startup. + `Too many open files (os error 24)` against 512× of unclaimed headroom. World Sessions now use one + client descriptor each. Failure logs and continues — it never blocks startup. - **A transient `accept(2)` errno costs one connection, not the realm** (`gateway/src/accept.rs`). `EMFILE`, `ENFILE`, `ECONNABORTED`, `EINTR`, `ENOBUFS`, `ENOMEM` and every unenumerated errno are logged and skipped, with a backoff that settles at one attempt per second under sustained diff --git a/gateway/src/accept.rs b/gateway/src/accept.rs index 04157631d..1564e8a0d 100644 --- a/gateway/src/accept.rs +++ b/gateway/src/accept.rs @@ -26,6 +26,19 @@ impl BlockingTaskCapacity { .map(|permit| BlockingTaskPermit { _permit: permit }) } + pub(crate) async fn run(&self, work: F) -> anyhow::Result + where + F: FnOnce() -> T + Send + 'static, + T: Send + 'static, + { + let permit = self.permits.clone().acquire_owned().await?; + Ok(tokio::task::spawn_blocking(move || { + let _permit = permit; + work() + }) + .await?) + } + #[cfg(test)] pub(crate) fn available(&self) -> usize { self.permits.available_permits() diff --git a/gateway/src/config.rs b/gateway/src/config.rs index 301aa1459..546262cf5 100644 --- a/gateway/src/config.rs +++ b/gateway/src/config.rs @@ -48,11 +48,27 @@ impl GatewayConfig { module_name: get("LYRACORE_DATABASE", "lyracore"), coordinator_token: std::env::var("LYRACORE_COORDINATOR_TOKEN").ok(), gateway_id: get("LYRACORE_GATEWAY_ID", &default_gateway_id), - blocking_task_capacity: BlockingTaskCapacity::new(max_blocking_threads()), + blocking_task_capacity: BlockingTaskCapacity::new( + max_blocking_threads() + .saturating_sub(encoding_workers()) + .max(1), + ), } } } +pub(crate) fn encoding_workers() -> usize { + (max_blocking_threads() / 4).clamp(1, 8) +} + +pub(crate) fn world_connection_limit() -> usize { + std::env::var("LYRACORE_MAX_WORLD_CONNECTIONS") + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|limit| *limit > 0 && *limit <= tokio::sync::Semaphore::MAX_PERMITS) + .unwrap_or(16_384) +} + pub const INSTANCE_BUCKETS: u64 = 8; #[derive(Clone, Debug, PartialEq)] @@ -431,7 +447,7 @@ pub fn max_blocking_threads() -> usize { pub fn max_blocking_threads_from_env(raw: Option<&str>) -> usize { raw.map(str::trim) .and_then(|v| v.parse::().ok()) - .filter(|&n| n > 0) + .filter(|&n| n > 0 && n <= tokio::sync::Semaphore::MAX_PERMITS) .unwrap_or(DEFAULT_MAX_BLOCKING_THREADS) } diff --git a/gateway/src/main.rs b/gateway/src/main.rs index 09ff841ce..b298b1741 100644 --- a/gateway/src/main.rs +++ b/gateway/src/main.rs @@ -32,9 +32,8 @@ fn main() -> Result<()> { let max_blocking_threads = config::max_blocking_threads(); log::info!( "tokio blocking pool: max_blocking_threads={max_blocking_threads} \ - (LYRACORE_MAX_BLOCKING_THREADS; default {}). One world session holds one thread for its \ - entire life, shared with in-flight logon handshakes. Excess connection tasks are rejected \ - before the blocking-task queue.", + (LYRACORE_MAX_BLOCKING_THREADS; default {}). World Session socket I/O is asynchronous. \ + Game actions, Relay encoding and logon handshakes use bounded blocking work.", config::DEFAULT_MAX_BLOCKING_THREADS ); diff --git a/gateway/src/world/egress.rs b/gateway/src/world/egress.rs new file mode 100644 index 000000000..87aa777bb --- /dev/null +++ b/gateway/src/world/egress.rs @@ -0,0 +1,339 @@ +use super::{Outbound, SessionSender, SessionTx, WriterTrace}; +use crate::accept::BlockingTaskCapacity; +use anyhow::{anyhow, bail, Result}; +use std::io; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{mpsc::SendError, Arc}; +use std::time::{Duration, Instant}; +use tokio::io::{AsyncWrite, AsyncWriteExt}; +use tokio::sync::{mpsc, watch}; +use wow_srp::vanilla_header::EncrypterHalf; + +const QUEUE_ITEMS: usize = 2048; +const QUEUE_BYTES: usize = 4 * 1024 * 1024; +const JOB_RESERVATION: usize = 4 * 1024; +pub(super) const MAX_AGE: Duration = Duration::from_secs(10); +const WRITE_BYTES: usize = 64 * 1024; +const ENCODE_BATCH: usize = 32; + +#[derive(Clone)] +pub(super) struct QueueSender { + sender: mpsc::Sender, + bytes: Arc, + depth: Arc, + closed: watch::Sender, +} + +pub(super) struct Queued { + out: Option, + since: Instant, + reservation: usize, + bytes: Arc, + depth: Arc, +} + +impl Drop for Queued { + fn drop(&mut self) { + self.bytes.fetch_sub(self.reservation, Ordering::Relaxed); + self.depth.fetch_sub(1, Ordering::Relaxed); + } +} + +impl QueueSender { + pub(super) fn send(&self, out: Outbound) -> std::result::Result<(), SendError> { + let reservation = match queued_bytes(&out) { + Ok(bytes) => bytes.max(1), + Err(error) => { + log::warn!("World Session outbound reservation failed: {error}"); + self.closed.send_replace(true); + return Err(SendError(out)); + } + }; + if *self.closed.borrow() + || self + .bytes + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |used| { + used.checked_add(reservation) + .filter(|next| *next <= QUEUE_BYTES) + }) + .is_err() + { + self.closed.send_replace(true); + return Err(SendError(out)); + } + self.depth.fetch_add(1, Ordering::Relaxed); + let queued = Queued { + out: Some(out), + since: Instant::now(), + reservation, + bytes: self.bytes.clone(), + depth: self.depth.clone(), + }; + self.sender.try_send(queued).map_err(|error| { + self.closed.send_replace(true); + let mut queued = error.into_inner(); + SendError( + queued + .out + .take() + .unwrap_or_else(|| Outbound::Batch(Vec::new())), + ) + }) + } +} + +pub(super) fn channel() -> (SessionTx, mpsc::Receiver, watch::Receiver) { + let (sender, receiver) = mpsc::channel(QUEUE_ITEMS); + let (closed, close_rx) = watch::channel(false); + let depth = Arc::new(AtomicUsize::new(0)); + let queue = QueueSender { + sender, + bytes: Arc::new(AtomicUsize::new(0)), + depth: depth.clone(), + closed: closed.clone(), + }; + let tx = SessionTx { + tx: SessionSender::Async(queue), + depth, + close_socket: Some(Arc::new(move || { + closed.send_replace(true); + })), + }; + (tx, receiver, close_rx) +} + +fn queued_bytes(out: &Outbound) -> io::Result { + match out { + Outbound::Raw { body, .. } => Ok(4 + body.len()), + Outbound::Batch(messages) => messages + .len() + .checked_mul(JOB_RESERVATION) + .ok_or_else(|| io::Error::other("outbound reservation overflow")), + // Typed compression and deferred projections run only on encoding workers. + Outbound::One(_) | Outbound::Job(_) => Ok(JOB_RESERVATION), + } +} + +#[derive(Clone)] +pub(crate) struct Packet { + opcode: u16, + body: Arc<[u8]>, +} + +pub(crate) fn encode(outbound: Vec) -> io::Result> { + let mut packets = Vec::new(); + let mut bytes = 0usize; + for out in outbound { + let mut add = |opcode, body: Arc<[u8]>| -> io::Result<()> { + if body.len() > u16::MAX as usize - 2 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "outbound body exceeds frame size", + )); + } + bytes = bytes.saturating_add(4 + body.len()); + if bytes > QUEUE_BYTES { + return Err(io::Error::other( + "outbound projection exceeds World Session byte limit", + )); + } + packets.push(Packet { opcode, body }); + Ok(()) + }; + let mut typed = + |message: wow_world_messages::vanilla::opcodes::ServerOpcodeMessage| -> io::Result<()> { + let mut frame = Vec::new(); + message.write_unencrypted_server(&mut frame)?; + if frame.len() < 4 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "short outbound frame", + )); + } + let opcode = u16::from_le_bytes([frame[2], frame[3]]); + add(opcode, Arc::from(&frame[4..])) + }; + match out { + Outbound::One(message) => typed(message)?, + Outbound::Batch(messages) => { + for message in messages { + typed(message)?; + } + } + Outbound::Raw { opcode, body } => { + for violation in super::packet_lint::lint_raw(opcode, &body) { + log::error!("packet-lint VIOLATION opcode=0x{opcode:04X}: {violation}"); + } + add(opcode, body.into())?; + } + Outbound::Job(_) => return Err(io::Error::other("nested Relay job")), + } + } + Ok(packets) +} + +pub(super) async fn write( + mut writer: W, + mut encrypt: EncrypterHalf, + mut receiver: mpsc::Receiver, + workers: BlockingTaskCapacity, + account_id: u64, +) -> Result<()> { + let mut trace = crate::config::writer_trace_enabled().then(WriterTrace::new); + let mut buffer = Vec::with_capacity(4096); + while let Some(first) = receiver.recv().await { + let mut oldest = first.since; + let mut batch = vec![first]; + while batch.len() < ENCODE_BATCH { + match receiver.try_recv() { + Ok(next) => { + oldest = oldest.min(next.since); + batch.push(next); + } + Err(_) => break, + } + } + let deadline = tokio::time::Instant::from_std(oldest + MAX_AGE); + if tokio::time::Instant::now() >= deadline { + bail!("World Session outbound queue exceeded age limit"); + } + let encoded = tokio::time::timeout_at( + deadline, + workers.run(move || { + let mut frames = Vec::new(); + let mut total = 0usize; + for queued in &mut batch { + let outbound = match queued.out.take() { + Some(Outbound::Job(job)) => job(), + Some(other) => vec![other], + None => bail!("World Session queue entry consumed twice"), + }; + let packets = encode(outbound)?; + total += packets + .iter() + .map(|packet| 4 + packet.body.len()) + .sum::(); + if total > QUEUE_BYTES { + bail!("World Session encoded batch exceeds byte limit"); + } + frames.extend(packets); + } + Ok::<_, anyhow::Error>((batch, frames)) + }), + ) + .await + .map_err(|_| anyhow!("World Session outbound queue exceeded age limit"))???; + let (_reservations, frames) = encoded; + if tokio::time::Instant::now() >= deadline { + bail!("World Session outbound encoding exceeded age limit"); + } + buffer.clear(); + let result = async { + for frame in frames { + if !buffer.is_empty() && buffer.len() + 4 + frame.body.len() > WRITE_BYTES { + writer.write_all(&buffer).await?; + buffer.clear(); + } + if let Some(trace) = trace.as_mut() { + trace.push( + frame.opcode, + (2 + frame.body.len()) as u16, + super::fnv1a64(&frame.body), + ); + } + encrypt.write_encrypted_server_header( + &mut buffer, + (2 + frame.body.len()) as u16, + frame.opcode, + )?; + buffer.extend_from_slice(&frame.body); + } + if !buffer.is_empty() { + writer.write_all(&buffer).await?; + } + Ok::<_, io::Error>(()) + }; + let result = tokio::time::timeout_at(deadline, result) + .await + .map_err(|_| anyhow!("World Session outbound write exceeded age limit")) + .and_then(|result| result.map_err(Into::into)); + if let Err(error) = result { + if let Some(trace) = trace.as_ref() { + trace.dump(account_id, &error.to_string()); + } + return Err(error); + } + } + writer.shutdown().await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn deferred_jobs_cannot_grow_the_queue_without_bound() { + let (tx, _receiver, closed) = channel(); + for _ in 0..QUEUE_BYTES / JOB_RESERVATION { + tx.send(Outbound::Job(Box::new(Vec::new))) + .unwrap_or_else(|_| panic!("within budget")); + } + assert!(tx.send(Outbound::Job(Box::new(Vec::new))).is_err()); + assert!(*closed.borrow()); + } + + #[tokio::test] + async fn dropping_a_queued_packet_returns_its_budget() { + let (tx, mut receiver, _) = channel(); + for _ in 0..100 { + tx.send(Outbound::Raw { + opcode: 1, + body: vec![0; 60_000], + }) + .unwrap_or_else(|_| panic!("drained queue must retain its budget")); + assert_eq!(tx.depth(), 1); + drop(receiver.recv().await); + assert_eq!(tx.depth(), 0); + } + } + + #[tokio::test] + async fn an_expired_entry_later_in_the_batch_refuses_the_write() { + let (tx, mut queued, _) = channel(); + for _ in 0..2 { + tx.send(Outbound::Raw { + opcode: 0x01dd, + body: vec![0; 4], + }) + .unwrap_or_else(|_| panic!("within budget")); + } + let (sender, receiver) = mpsc::channel(2); + sender + .send(queued.recv().await.unwrap()) + .await + .ok() + .unwrap(); + let mut delayed = queued.recv().await.unwrap(); + delayed.since -= MAX_AGE + Duration::from_secs(1); + sender.send(delayed).await.ok().unwrap(); + drop(sender); + let (_, crypto) = wow_srp::vanilla_header::ProofSeed::new().into_client_header_crypto( + &wow_srp::normalized_string::NormalizedString::new("TESTER").unwrap(), + super::super::test_support::K, + 0, + ); + + let error = write( + tokio::io::sink(), + crypto.split().0, + receiver, + BlockingTaskCapacity::new(1), + 1, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("age limit")); + assert_eq!(tx.depth(), 0); + } +} diff --git a/gateway/src/world/login_queue.rs b/gateway/src/world/login_queue.rs index 8b5cb1b46..ba74335a8 100644 --- a/gateway/src/world/login_queue.rs +++ b/gateway/src/world/login_queue.rs @@ -95,11 +95,26 @@ impl LoginQueue { .map(|i| i + 1) } + #[cfg(test)] pub fn cancel(&self, ticket: Ticket) { let mut s = self.state.lock().unwrap(); s.waiting.retain(|w| w.ticket != ticket); } + pub(super) fn cancel_or_depart(&self, ticket: Ticket) { + let mut state = self.state.lock().unwrap(); + if let Some(index) = state + .waiting + .iter() + .position(|waiter| waiter.ticket == ticket) + { + state.waiting.remove(index); + } else { + state.active = state.active.saturating_sub(1); + self.drain_locked(&mut state); + } + } + pub fn depart(&self) { if self.max_sessions == 0 { return; diff --git a/gateway/src/world/mod.rs b/gateway/src/world/mod.rs index 2816fc26e..df7af41ff 100644 --- a/gateway/src/world/mod.rs +++ b/gateway/src/world/mod.rs @@ -1,14 +1,17 @@ use crate::accept::{classify_accept_error, AcceptBackoff, AcceptOutcome}; use crate::codec::auth_session::{read_auth_session, CMSG_AUTH_SESSION_OPCODE}; -use crate::read_deadline::{ - is_io_deadline, DeadlineClock, IoDeadline, PreAuthDeadline, WORLD_AUTH_DEADLINE, -}; +use crate::read_deadline::WORLD_AUTH_DEADLINE; +#[cfg(test)] +use crate::read_deadline::{is_io_deadline, DeadlineClock, IoDeadline, PreAuthDeadline}; use crate::stdb::PlayerSubscriptions; use crate::{codec, config::GatewayConfig, stdb::Coordinator}; use anyhow::{anyhow, Result}; +#[cfg(test)] use std::io::{Read, Write}; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::mpsc::{channel, Receiver, SendError, Sender}; +use std::sync::mpsc::SendError; +#[cfg(test)] +use std::sync::mpsc::{channel, Receiver, Sender}; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::net::TcpListener; @@ -25,9 +28,11 @@ use wow_world_messages::vanilla::{ mod actor; mod chat_flood; mod coalesce; +pub(crate) mod egress; mod entry; mod family; mod routing; +mod transport; pub(crate) use family::{ProtocolFamily, ProtocolSession}; use family::{ProtocolReply, ProtocolRequest, WorldSessionAction}; pub(crate) mod guild_fee; @@ -87,7 +92,7 @@ pub enum Outbound { opcode: u16, body: Vec, }, - /// SDK advancement is serial per shard; defer encoding to each World Session's writer. + /// SDK advancement is serial per Shard; defer encoding to the bounded worker pool. Job(RelayJob), } @@ -95,15 +100,23 @@ pub type RelayJob = Box Vec + Send>; pub const EGRESS_SHED_DEPTH: usize = 512; -/// Blocking sends from SDK callbacks can deadlock the pump; keep this channel unbounded. +/// SDK callbacks must never wait for a World Session to drain its queue. #[derive(Clone)] pub struct SessionTx { - tx: Sender, + tx: SessionSender, depth: Arc, close_socket: Option>, } +#[derive(Clone)] +enum SessionSender { + Async(egress::QueueSender), + #[cfg(test)] + Sync(Sender), +} + impl SessionTx { + #[cfg(test)] fn bind_socket(&mut self, socket: S) { let socket = std::sync::Mutex::new(socket); self.close_socket = Some(Arc::new(move || { @@ -124,12 +137,14 @@ impl SessionTx { // Count before sending so depth cannot read low; undo the count if the send fails. pub fn send(&self, out: Outbound) -> std::result::Result<(), SendError> { - self.depth.fetch_add(1, Ordering::Relaxed); - match self.tx.send(out) { - Ok(()) => Ok(()), - Err(e) => { - self.depth.fetch_sub(1, Ordering::Relaxed); - Err(e) + match &self.tx { + SessionSender::Async(sender) => sender.send(out), + #[cfg(test)] + SessionSender::Sync(sender) => { + self.depth.fetch_add(1, Ordering::Relaxed); + sender.send(out).inspect_err(|_| { + self.depth.fetch_sub(1, Ordering::Relaxed); + }) } } } @@ -144,12 +159,13 @@ impl SessionTx { } } +#[cfg(test)] pub fn session_channel() -> (SessionTx, Receiver, Arc) { let (tx, rx) = channel::(); let depth = Arc::new(AtomicUsize::new(0)); ( SessionTx { - tx, + tx: SessionSender::Sync(tx), depth: depth.clone(), close_socket: None, }, @@ -167,11 +183,13 @@ impl SessionTx { } } +#[cfg(test)] pub trait DuplexStream: Read + Write + IoDeadline + Send + Sized + 'static { fn try_clone(&self) -> std::io::Result; fn shutdown_both(&self) -> std::io::Result<()>; } +#[cfg(test)] impl DuplexStream for std::net::TcpStream { fn try_clone(&self) -> std::io::Result { std::net::TcpStream::try_clone(self) @@ -351,6 +369,7 @@ pub fn world_handshake_with_queue( world_handshake_with_queue_and_deadline(stream, store, queue, &mut deadline) } +#[cfg(test)] fn world_handshake_with_queue_and_deadline( stream: &mut S, store: Arc, @@ -437,6 +456,7 @@ fn world_handshake_with_queue_and_deadline( stream: &mut S, encrypt: &mut EncrypterHalf, @@ -508,6 +528,7 @@ impl WriterTrace { }); } + #[cfg(test)] fn record(&mut self, out: &Outbound) { match out { Outbound::One(m) => self.record_typed(m), @@ -523,6 +544,7 @@ impl WriterTrace { } } + #[cfg(test)] fn record_typed(&mut self, m: &ServerOpcodeMessage) { let mut buf = Vec::new(); if m.write_unencrypted_server(&mut buf).is_err() || buf.len() < 4 { @@ -569,6 +591,7 @@ impl WriterTrace { } } +#[cfg(test)] fn spawn_writer( mut wsock: S, mut encrypt: EncrypterHalf, @@ -659,10 +682,12 @@ pub fn run_world_session(stream: S, store: Arc) run_world_session_with_queue(stream, store, &LoginQueue::unlimited()) } +#[cfg(test)] struct AdmissionSeat<'a> { queue: &'a LoginQueue, } +#[cfg(test)] impl Drop for AdmissionSeat<'_> { fn drop(&mut self) { self.queue.depart(); @@ -679,6 +704,7 @@ pub fn run_world_session_with_queue( run_world_session_with_queue_and_deadline(stream, store, queue, &mut deadline) } +#[cfg(test)] fn run_world_session_with_queue_and_deadline( mut stream: S, store: Arc, @@ -704,9 +730,7 @@ fn run_world_session_with_queue_and_deadline( let writer = spawn_writer(wsock, encrypt, rx, depth, conn.account_id)?; let result = (|| -> Result<()> { - let mut addon_tokens: f32 = 20.0; - let mut addon_refill_at = std::time::Instant::now(); - let mut addon_drop_logged_at: Option = None; + let mut addon = AddonLimit::default(); loop { let hdr = match conn.decrypt.read_and_decrypt_client_header(&mut stream) { Ok(h) => h, @@ -719,97 +743,7 @@ fn run_world_session_with_queue_and_deadline( stream .read_exact(&mut body) .map_err(|e| anyhow!("world read error (body): {e}"))?; - if hdr.opcode == codec::addon::CMSG_MESSAGECHAT_OPCODE { - if let Some((chat_type, text)) = codec::addon::parse_addon_client_chat(&body) { - if codec::addon::is_bridge_prefixed(&text) { - let now = std::time::Instant::now(); - addon_tokens = (addon_tokens - + now.duration_since(addon_refill_at).as_secs_f32() * 2.0) - .min(20.0); - addon_refill_at = now; - if addon_tokens >= 1.0 { - addon_tokens -= 1.0; - handle_addon_message(&*conn.store.current(), &conn, &text)?; - } else if addon_drop_logged_at - .is_none_or(|t| now.duration_since(t).as_secs() >= 60) - { - addon_drop_logged_at = Some(now); - log::warn!( - "addon bridge: rate limit — dropping frames from account {}", - conn.account_id - ); - } - continue; - } - if !codec::addon::typed_addon_chat_type(chat_type) { - continue; - } - } - } - if hdr.opcode == CMSG_AUCTION_LIST_ITEMS_OPCODE { - let request = - decode_auction_browse(&body).map_err(|e| anyhow!("world read error: {e}"))?; - let store = conn.store.current(); - dispatch( - &tx, - &*store, - &mut conn, - hdr.opcode, - ProtocolRequest::AuctionBrowse(request), - )?; - continue; - } - if hdr.opcode == CMSG_AUTH_SESSION_OPCODE { - return Err(anyhow!("world read error: CMSG_AUTH_SESSION after auth")); - } - if hdr.opcode == 0x0125 { - let store = conn.store.current(); - dispatch( - &tx, - &*store, - &mut conn, - hdr.opcode, - ProtocolRequest::FactionAtWar(body), - )?; - continue; - } - if hdr.opcode == 0x0318 { - let store = conn.store.current(); - dispatch( - &tx, - &*store, - &mut conn, - hdr.opcode, - ProtocolRequest::WatchedFaction(body), - )?; - continue; - } - if handlers::is_silent_movement_request(hdr.opcode) - || handlers::is_ignored_raw_request(hdr.opcode) - { - continue; - } - if let Some(replies) = handlers::raw_unavailable_outbound(hdr.opcode, &body)? { - if matches!(conn.state, WorldState::InWorld(_)) { - for reply in replies { - send(&tx, reply)?; - } - } - continue; - } - let msg = match codec::read_movement_client(hdr.opcode, &body)? { - Some(msg) => msg, - None => { - let mut framed = Vec::with_capacity(6 + body.len()); - framed.extend_from_slice(&hdr.size.to_be_bytes()); - framed.extend_from_slice(&hdr.opcode.to_le_bytes()); - framed.extend_from_slice(&body); - ClientOpcodeMessage::read_unencrypted(&mut std::io::Cursor::new(framed)) - .map_err(|e| anyhow!("world read error: {e}"))? - } - }; - let store = conn.store.current(); - dispatch(&tx, &*store, &mut conn, hdr.opcode, msg.into())?; + process_frame(&tx, &mut conn, &mut addon, hdr, body)?; } })(); @@ -830,6 +764,123 @@ fn run_world_session_with_queue_and_deadline( result } +struct AddonLimit { + tokens: f32, + refill_at: Instant, + drop_logged_at: Option, +} + +impl Default for AddonLimit { + fn default() -> Self { + Self { + tokens: 20.0, + refill_at: Instant::now(), + drop_logged_at: None, + } + } +} + +fn process_frame( + tx: &SessionTx, + conn: &mut WorldConn, + addon: &mut AddonLimit, + hdr: wow_srp::vanilla_header::ClientHeader, + body: Vec, +) -> Result<()> { + if hdr.opcode == codec::addon::CMSG_MESSAGECHAT_OPCODE { + if let Some((chat_type, text)) = codec::addon::parse_addon_client_chat(&body) { + if codec::addon::is_bridge_prefixed(&text) { + let now = std::time::Instant::now(); + addon.tokens = (addon.tokens + + now.duration_since(addon.refill_at).as_secs_f32() * 2.0) + .min(20.0); + addon.refill_at = now; + if addon.tokens >= 1.0 { + addon.tokens -= 1.0; + handle_addon_message(&*conn.store.current(), conn, &text)?; + } else if addon + .drop_logged_at + .is_none_or(|t| now.duration_since(t).as_secs() >= 60) + { + addon.drop_logged_at = Some(now); + log::warn!( + "addon bridge: dropping frames above the rate limit for account {}", + conn.account_id + ); + } + return Ok(()); + } + if !codec::addon::typed_addon_chat_type(chat_type) { + return Ok(()); + } + } + } + if hdr.opcode == CMSG_AUCTION_LIST_ITEMS_OPCODE { + let request = decode_auction_browse(&body).map_err(|e| anyhow!("world read error: {e}"))?; + let store = conn.store.current(); + dispatch( + tx, + &*store, + conn, + hdr.opcode, + ProtocolRequest::AuctionBrowse(request), + )?; + return Ok(()); + } + if hdr.opcode == CMSG_AUTH_SESSION_OPCODE { + return Err(anyhow!("world read error: CMSG_AUTH_SESSION after auth")); + } + if hdr.opcode == 0x0125 { + let store = conn.store.current(); + dispatch( + tx, + &*store, + conn, + hdr.opcode, + ProtocolRequest::FactionAtWar(body), + )?; + return Ok(()); + } + if hdr.opcode == 0x0318 { + let store = conn.store.current(); + dispatch( + tx, + &*store, + conn, + hdr.opcode, + ProtocolRequest::WatchedFaction(body), + )?; + return Ok(()); + } + if handlers::is_silent_movement_request(hdr.opcode) + || handlers::is_ignored_raw_request(hdr.opcode) + { + return Ok(()); + } + if let Some(replies) = handlers::raw_unavailable_outbound(hdr.opcode, &body)? { + if matches!(conn.state, WorldState::InWorld(_)) { + for reply in replies { + send(tx, reply)?; + } + } + return Ok(()); + } + let msg = match codec::read_movement_client(hdr.opcode, &body)? { + Some(msg) => msg, + None => { + let mut framed = Vec::with_capacity(6 + body.len()); + framed.extend_from_slice(&hdr.size.to_be_bytes()); + framed.extend_from_slice(&hdr.opcode.to_le_bytes()); + framed.extend_from_slice(&body); + ClientOpcodeMessage::read_unencrypted(&mut std::io::Cursor::new(framed)) + .map_err(|e| anyhow!("world read error: {e}"))? + } + }; + let store = conn.store.current(); + dispatch(tx, &*store, conn, hdr.opcode, msg.into())?; + Ok(()) +} + fn is_game_master(store: &St, character_guid: u64) -> bool { match store.speaker_gm_level(character_guid) { Ok(gm_level) => gm_level > 0, @@ -1171,6 +1222,11 @@ pub async fn run( }); let mut backoff = AcceptBackoff::new(); let mut capacity_was_full = false; + let connections = Arc::new(tokio::sync::Semaphore::new( + crate::config::world_connection_limit(), + )); + let actions = cfg.blocking_task_capacity.clone(); + let encoders = crate::accept::BlockingTaskCapacity::new(crate::config::encoding_workers()); loop { let (sock, peer) = match listener.accept().await { Ok(pair) => { @@ -1190,9 +1246,8 @@ pub async fn run( log::warn!( "world accept failed ({e}); skipping this connection (consecutive={}, \ backing off {}ms). Players already in world are unaffected; if this is \ - EMFILE, the gateway is out of file descriptors — check for a leaked \ - per-account SDK connection (each account's cached connection releases at \ - session teardown).", + EMFILE, the Gateway is out of file descriptors; inspect open sockets and \ + the configured World connection limit.", backoff.consecutive(), delay.as_millis() ); @@ -1203,13 +1258,10 @@ pub async fn run( } }, }; - let mut deadline = PreAuthDeadline::after(WORLD_AUTH_DEADLINE); - let Some(task_permit) = cfg.blocking_task_capacity.try_admit() else { + let deadline = tokio::time::Instant::now() + WORLD_AUTH_DEADLINE; + let Ok(connection_permit) = connections.clone().try_acquire_owned() else { if !capacity_was_full { - log::warn!( - "world blocking-task capacity is full; rejecting new connections before they \ - enter the blocking-task queue" - ); + log::warn!("world connection capacity is full; refusing new connections"); capacity_was_full = true; } continue; @@ -1217,29 +1269,14 @@ pub async fn run( capacity_was_full = false; let coord = coordinator.clone(); let queue = login_queue.clone(); - let std_sock = match sock.into_std().and_then(|s| { - s.set_nonblocking(false)?; - Ok(s) - }) { - Ok(s) => s, - Err(e) => { - log::warn!("world connection {peer} could not be handed to a blocking task: {e}"); - continue; - } - }; - if let Err(e) = deadline.arm(&std_sock) { - log::warn!("world connection {peer} could not arm its pre-auth watchdog: {e}"); - continue; - } - tokio::task::spawn_blocking(move || { - let _task_permit = task_permit; - if let Err(e) = run_world_session_with_queue_and_deadline( - std_sock, - Arc::new(coord), - &queue, - &mut deadline, - ) { - log::warn!("world session {peer} ended: {e:#}"); + let actions = actions.clone(); + let encoders = encoders.clone(); + tokio::spawn(async move { + let _connection_permit = connection_permit; + if let Err(error) = + transport::run_at(sock, Arc::new(coord), queue, actions, encoders, deadline).await + { + log::warn!("world session {peer} ended: {error:#}"); } }); } diff --git a/gateway/src/world/test_support/world_fake/session.rs b/gateway/src/world/test_support/world_fake/session.rs index 75db18bda..6e0a3c018 100644 --- a/gateway/src/world/test_support/world_fake/session.rs +++ b/gateway/src/world/test_support/world_fake/session.rs @@ -13,6 +13,7 @@ pub(crate) struct SessionState { pub(crate) username: String, pub(crate) session: Option, pub(crate) login_entity: Option, + pub(crate) login_pause: Option>, pub(crate) moves: std::sync::Mutex>, pub(crate) move_bodies: std::sync::Mutex>>, pub(crate) client_commands: std::sync::Mutex>, @@ -54,6 +55,10 @@ impl SessionStore for WorldFake { entry: codec::WorldEntry, ) -> Result { self.rec("player_login"); + if let Some(pause) = &self.session.login_pause { + pause.0.wait(); + pause.1.wait(); + } self.session.login_entries.lock().unwrap().push(entry); let call = self .session diff --git a/gateway/src/world/transport.rs b/gateway/src/world/transport.rs new file mode 100644 index 000000000..7d3eee9be --- /dev/null +++ b/gateway/src/world/transport.rs @@ -0,0 +1,555 @@ +use super::*; +use crate::accept::BlockingTaskCapacity; +use std::sync::Mutex; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + +struct Seat { + queue: Arc, + queued: Option, +} + +impl Drop for Seat { + fn drop(&mut self) { + match self.queued { + Some(ticket) => self.queue.cancel_or_depart(ticket), + None => self.queue.depart(), + } + } +} + +async fn authenticate( + stream: &mut S, + store: Arc, + actions: &BlockingTaskCapacity, +) -> Result> { + let seed = ProofSeed::new(); + let mut frame = Vec::new(); + SMSG_AUTH_CHALLENGE { + server_seed: seed.seed(), + } + .write_unencrypted_server(&mut frame)?; + stream.write_all(&frame).await?; + + let mut header = [0; 6]; + if let Err(error) = stream.read_exact(&mut header).await { + if error.kind() == std::io::ErrorKind::UnexpectedEof { + return Ok(None); + } + return Err(error.into()); + } + let size = u16::from_be_bytes([header[0], header[1]]) as usize; + let body_len = size + .checked_sub(4) + .ok_or_else(|| anyhow!("invalid world auth frame size"))?; + let mut frame = Vec::with_capacity(6 + body_len); + frame.extend_from_slice(&header); + frame.resize(6 + body_len, 0); + stream.read_exact(&mut frame[6..]).await?; + let claim = read_auth_session(&mut frame.as_slice())?; + let username = claim.username.to_uppercase(); + let lookup_store = store.clone(); + let lookup_name = username.clone(); + let session = actions + .run(move || lookup_store.lookup_session(&lookup_name)) + .await??; + let Some(WorldSession { + account_id, + session_key, + }) = session + else { + let mut frame = Vec::new(); + SMSG_AUTH_RESPONSE::AuthUnknownAccount.write_unencrypted_server(&mut frame)?; + stream.write_all(&frame).await?; + return Ok(None); + }; + let name = + NormalizedString::new(&username).map_err(|error| anyhow!("bad username: {error}"))?; + let crypto = match seed.into_server_header_crypto( + &name, + session_key, + claim.client_proof, + claim.client_seed, + ) { + Ok(crypto) => crypto, + Err(_) => { + let mut frame = Vec::new(); + SMSG_AUTH_RESPONSE::AuthFailed.write_unencrypted_server(&mut frame)?; + stream.write_all(&frame).await?; + return Ok(None); + } + }; + let (encrypt, decrypt) = crypto.split(); + Ok(Some(( + WorldConn { + protocol: ProtocolSession::new(account_id, username), + session_claim: None, + decrypt, + move_coalesce: CoalesceState::default(), + unavailable_notices: Default::default(), + store: RoutedStore::new(store), + session_key: Some(session_key), + guild_signed_on: None, + move_desync_drops: 0, + chat_flood: Default::default(), + }, + encrypt, + ))) +} + +async fn admit( + stream: &mut S, + encrypt: &mut EncrypterHalf, + queue: Arc, +) -> Result { + let mut seat = Seat { + queued: None, + queue, + }; + if let Admission::Queued(ticket) = seat.queue.request() { + seat.queued = Some(ticket); + let mut last_sent: Option<(Instant, usize)> = None; + while let Some(position) = seat.queue.position(ticket) { + if last_sent.is_none_or(|(at, previous)| { + previous != position || at.elapsed() >= QUEUE_RESEND_INTERVAL + }) { + let mut frame = Vec::new(); + (SMSG_AUTH_RESPONSE::AuthWaitQueue { + queue_position: position as u32, + }) + .write_encrypted_server(&mut frame, encrypt)?; + tokio::time::timeout(WORLD_AUTH_DEADLINE, stream.write_all(&frame)).await??; + last_sent = Some((Instant::now(), position)); + } + tokio::time::sleep(QUEUE_POLL_INTERVAL).await; + } + seat.queued = None; + } + let mut frame = Vec::new(); + (SMSG_AUTH_RESPONSE::AuthOk { + billing_flags: 0, + billing_rested: 0, + billing_time: 0, + }) + .write_encrypted_server(&mut frame, encrypt)?; + tokio::time::timeout(WORLD_AUTH_DEADLINE, stream.write_all(&frame)).await??; + Ok(seat) +} + +pub(super) async fn run_at( + stream: S, + store: Arc, + queue: Arc, + actions: BlockingTaskCapacity, + encoders: BlockingTaskCapacity, + deadline: tokio::time::Instant, +) -> Result<()> +where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + let (stop, receiver) = tokio::sync::watch::channel(false); + struct StopOnDrop(tokio::sync::watch::Sender); + impl Drop for StopOnDrop { + fn drop(&mut self) { + self.0.send_replace(true); + } + } + let _stop_on_drop = StopOnDrop(stop); + // The supervisor owns cleanup even if the caller drops its session future. + tokio::spawn(run_session( + stream, store, queue, actions, encoders, receiver, deadline, + )) + .await? +} + +async fn run_session( + mut stream: S, + store: Arc, + queue: Arc, + actions: BlockingTaskCapacity, + encoders: BlockingTaskCapacity, + mut stopped: tokio::sync::watch::Receiver, + deadline: tokio::time::Instant, +) -> Result<()> +where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + let authenticated = tokio::select! { + biased; + _ = stopped.wait_for(|stop| *stop) => return Ok(()), + result = tokio::time::timeout_at(deadline, authenticate(&mut stream, store, &actions)) => + result.map_err(|_| anyhow!("no CMSG_AUTH_SESSION before the total pre-auth deadline"))??, + }; + let Some((conn, mut encrypt)) = authenticated else { + return Ok(()); + }; + let _seat = tokio::select! { + biased; + _ = stopped.wait_for(|stop| *stop) => return Ok(()), + result = admit(&mut stream, &mut encrypt, queue) => result?, + }; + let account_id = conn.account_id; + let conn = Arc::new(Mutex::new(conn)); + let addon = Arc::new(Mutex::new(AddonLimit::default())); + let (tx, receiver, mut closed) = egress::channel(); + let (mut reader, writer) = tokio::io::split(stream); + let writer_close = tx.close_socket.clone(); + let mut writer_task = tokio::spawn(async move { + let result = egress::write(writer, encrypt, receiver, encoders, account_id).await; + if let Some(close) = writer_close { + close(); + } + result + }); + + let mut drain = true; + let result = async { + loop { + let frame = async { + let mut bytes = [0; 6]; + reader.read_exact(&mut bytes).await?; + let header = conn + .lock() + .unwrap_or_else(|error| error.into_inner()) + .decrypt + .decrypt_client_header(bytes); + let length = (header.size as usize).checked_sub(4).ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + "invalid client frame size", + ) + })?; + let mut body = vec![0; length]; + reader.read_exact(&mut body).await?; + Ok::<_, std::io::Error>((header, body)) + }; + let (header, body) = tokio::select! { + biased; + _ = stopped.wait_for(|stop| *stop) => { drain = false; break; }, + _ = closed.wait_for(|closed| *closed) => { drain = false; break; }, + result = frame => match result { + Ok(frame) => frame, + Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => break, + Err(error) => return Err(error.into()), + } + }; + let conn = conn.clone(); + let addon = addon.clone(); + let tx = tx.clone(); + // Await accepted game work before cleanup so a late login cannot outlive its session. + actions + .run(move || { + let mut conn = conn.lock().unwrap_or_else(|error| error.into_inner()); + let mut addon = addon.lock().unwrap_or_else(|error| error.into_inner()); + process_frame(&tx, &mut conn, &mut addon, header, body) + }) + .await??; + } + Ok::<_, anyhow::Error>(()) + } + .await; + + drain &= !*closed.borrow() && !*stopped.borrow(); + if !drain { + tx.close(); + writer_task.abort(); + } + let cleanup = actions + .run(move || { + let mut conn = conn.lock().unwrap_or_else(|error| error.into_inner()); + log::debug!( + "world movement account={account_id} received={} forwarded={}", + conn.move_coalesce.received(), + conn.move_coalesce.forwarded() + ); + conn.leave_world() + }) + .await + .and_then(|result| result); + if let Err(error) = cleanup { + log::warn!("logout for account {account_id} failed: {error:#}"); + } + drop(tx); + let writer_result = match tokio::time::timeout(egress::MAX_AGE, &mut writer_task).await { + Ok(result) => result, + Err(_) => { + writer_task.abort(); + writer_task.await + } + }; + result?; + if let Ok(result) = writer_result { + result?; + } + Ok(()) +} + +#[cfg(test)] +async fn run( + stream: S, + store: Arc, + queue: Arc, + actions: BlockingTaskCapacity, + encoders: BlockingTaskCapacity, +) -> Result<()> +where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + run_at( + stream, + store, + queue, + actions, + encoders, + tokio::time::Instant::now() + WORLD_AUTH_DEADLINE, + ) + .await +} + +#[cfg(test)] +mod tests { + use super::*; + + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn idle_world_sessions_leave_the_action_worker_available() { + use crate::world::test_support::{ + client_handshake, tester_store, world_session_socket_pair, K, + }; + use wow_world_messages::vanilla::{ClientMessage, CMSG_PING}; + + let actions = BlockingTaskCapacity::new(1); + let encoders = BlockingTaskCapacity::new(1); + let queue = Arc::new(LoginQueue::new(16, 0)); + let mut sessions = Vec::new(); + let mut clients = Vec::new(); + for account in 1..=8 { + let (mut client, server) = world_session_socket_pair(); + server.set_nonblocking(true).unwrap(); + let server = tokio::net::UnixStream::from_std(server).unwrap(); + sessions.push(tokio::spawn(run( + server, + Arc::new(tester_store(account)), + queue.clone(), + actions.clone(), + encoders.clone(), + ))); + clients.push( + tokio::task::spawn_blocking(move || { + let (encrypt, decrypt) = client_handshake(&mut client, "TESTER", K); + (client, encrypt, decrypt) + }) + .await + .unwrap(), + ); + } + assert_eq!(queue.active(), 8); + assert_eq!(actions.available(), 1); + tokio::task::spawn_blocking(move || { + for (mut client, mut encrypt, mut decrypt) in clients { + CMSG_PING { + sequence_id: 123, + round_time_in_ms: 0, + } + .write_encrypted_client(&mut client, &mut encrypt) + .unwrap(); + match ServerOpcodeMessage::read_encrypted(&mut client, &mut decrypt).unwrap() { + ServerOpcodeMessage::SMSG_PONG(pong) => assert_eq!(pong.sequence_id, 123), + other => panic!("expected SMSG_PONG, got {other}"), + } + } + }) + .await + .unwrap(); + for session in sessions { + tokio::time::timeout(Duration::from_secs(5), session) + .await + .unwrap() + .unwrap() + .unwrap(); + } + assert_eq!(queue.active(), 0); + } + + #[test] + fn cancellation_after_queue_admission_returns_the_promoted_seat() { + let queue = Arc::new(LoginQueue::new(1, 0)); + assert_eq!(queue.request(), Admission::Admitted); + let Admission::Queued(ticket) = queue.request() else { + panic!("queue is full"); + }; + let seat = Seat { + queue: queue.clone(), + queued: Some(ticket), + }; + queue.depart(); + assert_eq!(queue.active(), 1); + drop(seat); + assert_eq!(queue.active(), 0); + } + + #[test] + fn cancellation_before_admission_leaves_the_current_player_seated() { + let queue = Arc::new(LoginQueue::new(1, 0)); + assert_eq!(queue.request(), Admission::Admitted); + let Admission::Queued(ticket) = queue.request() else { + panic!("queue is full"); + }; + drop(Seat { + queue: queue.clone(), + queued: Some(ticket), + }); + assert_eq!(queue.active(), 1); + assert_eq!(queue.depth(), 0); + } + + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn client_half_close_drains_accepted_action_replies() { + use crate::world::test_support::{ + client_handshake, tester_store, world_session_socket_pair, K, + }; + use wow_world_messages::vanilla::{ClientMessage, CMSG_PING}; + + let (mut client, server) = world_session_socket_pair(); + server.set_nonblocking(true).unwrap(); + let server = tokio::net::UnixStream::from_std(server).unwrap(); + let queue = Arc::new(LoginQueue::new(1, 0)); + let session = tokio::spawn(run( + server, + Arc::new(tester_store(1)), + queue.clone(), + BlockingTaskCapacity::new(1), + BlockingTaskCapacity::new(1), + )); + tokio::task::spawn_blocking(move || { + let (mut encrypt, mut decrypt) = client_handshake(&mut client, "TESTER", K); + for sequence_id in 1..=3 { + CMSG_PING { + sequence_id, + round_time_in_ms: 0, + } + .write_encrypted_client(&mut client, &mut encrypt) + .unwrap(); + } + client.shutdown(std::net::Shutdown::Write).unwrap(); + for expected in 1..=3 { + match ServerOpcodeMessage::read_encrypted(&mut client, &mut decrypt).unwrap() { + ServerOpcodeMessage::SMSG_PONG(pong) => assert_eq!(pong.sequence_id, expected), + other => panic!("expected SMSG_PONG, got {other}"), + } + } + }) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), session) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(queue.active(), 0); + } + + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn failed_transfer_delivers_its_refusal_before_session_cleanup() { + use crate::world::test_support::{ + client_handshake, drain_world_entry, quest_store, world_session_socket_pair, K, + }; + use wow_world_messages::vanilla::{ + ClientMessage, CMSG_PLAYER_LOGIN, MSG_MOVE_WORLDPORT_ACK, + }; + + let mut fake = quest_store(); + fake.session.entity_in_world = false; + fake.session.worldport_login_error = Some("destination unavailable".into()); + let store = Arc::new(fake); + let (mut client, server) = world_session_socket_pair(); + server.set_nonblocking(true).unwrap(); + let session = tokio::spawn(run( + tokio::net::UnixStream::from_std(server).unwrap(), + store.clone(), + Arc::new(LoginQueue::unlimited()), + BlockingTaskCapacity::new(1), + BlockingTaskCapacity::new(1), + )); + tokio::task::spawn_blocking(move || { + let (mut encrypt, mut decrypt) = client_handshake(&mut client, "TESTER", K); + CMSG_PLAYER_LOGIN { + guid: wow_world_messages::Guid::new(1), + } + .write_encrypted_client(&mut client, &mut encrypt) + .unwrap(); + drain_world_entry(&mut client, &mut decrypt); + MSG_MOVE_WORLDPORT_ACK {} + .write_encrypted_client(&mut client, &mut encrypt) + .unwrap(); + let header = decrypt.read_and_decrypt_server_header(&mut client).unwrap(); + let mut body = vec![0; usize::from(header.size) - 2]; + std::io::Read::read_exact(&mut client, &mut body).unwrap(); + assert_eq!((header.opcode, body), (0x0040, vec![0x02])); + }) + .await + .unwrap(); + let error = tokio::time::timeout(Duration::from_secs(5), session) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert!(error.to_string().contains("destination unavailable")); + assert!(store.session.logout_called.load(Ordering::Relaxed)); + } + + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelling_during_login_waits_for_the_action_then_releases_the_claim() { + use crate::world::test_support::{ + client_handshake, quest_store, world_session_socket_pair, K, + }; + use wow_world_messages::vanilla::{ClientMessage, CMSG_PLAYER_LOGIN}; + + let pause = Arc::new((std::sync::Barrier::new(2), std::sync::Barrier::new(2))); + let mut fake = quest_store(); + fake.session.login_pause = Some(pause.clone()); + let store = Arc::new(fake); + let queue = Arc::new(LoginQueue::new(1, 0)); + let (mut client, server) = world_session_socket_pair(); + server.set_nonblocking(true).unwrap(); + let server = tokio::net::UnixStream::from_std(server).unwrap(); + let session = tokio::spawn(run( + server, + store.clone(), + queue.clone(), + BlockingTaskCapacity::new(1), + BlockingTaskCapacity::new(1), + )); + let entered = pause.clone(); + let client = tokio::task::spawn_blocking(move || { + let (mut encrypt, _) = client_handshake(&mut client, "TESTER", K); + CMSG_PLAYER_LOGIN { + guid: wow_world_messages::Guid::new(1), + } + .write_encrypted_client(&mut client, &mut encrypt) + .unwrap(); + entered.0.wait(); + client + }) + .await + .unwrap(); + session.abort(); + assert!(session.await.unwrap_err().is_cancelled()); + assert_eq!(queue.active(), 1); + assert!(!store.session.logout_called.load(Ordering::Relaxed)); + tokio::task::spawn_blocking(move || pause.1.wait()) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + while queue.active() != 0 { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .unwrap(); + assert!(store.session.logout_called.load(Ordering::Relaxed)); + drop(client); + } +} From 76dd8df3a951feeefebebd4c0095a1a0af08dcb1 Mon Sep 17 00:00:00 2001 From: Mathih13 Date: Fri, 16 Oct 2026 08:18:43 +0200 Subject: [PATCH 2/2] fix(gateway): preserve World Session shutdown and write ordering --- docs/architecture.md | 3 +- gateway/src/world/egress.rs | 15 +- gateway/src/world/framing_tests.rs | 4 +- gateway/src/world/mail_tests.rs | 2 +- gateway/src/world/mod.rs | 395 +----------------- gateway/src/world/session_tests.rs | 231 +--------- gateway/src/world/test_support/mod.rs | 2 + gateway/src/world/test_support/socket.rs | 91 ++++ .../world/test_support/world_fake/session.rs | 12 + gateway/src/world/tests.rs | 4 - gateway/src/world/transport.rs | 272 ++++++++++-- gateway/src/world/wire_corruption_tests.rs | 107 +---- 12 files changed, 414 insertions(+), 724 deletions(-) create mode 100644 gateway/src/world/test_support/socket.rs diff --git a/docs/architecture.md b/docs/architecture.md index f15c6506c..b6db0f358 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -357,7 +357,8 @@ One writer owns each session's header cipher and preserves FIFO order and contig It combines frames into writes of about 64 KiB. A supervisor retains cleanup ownership when its caller is cancelled. Accepted game actions finish before logout cleanup, so a delayed login cannot recreate a departed session after cleanup. A clean -read EOF drains queued replies within the outbound age budget before returning the admission seat. +read EOF drains direct replies from accepted actions within the outbound age budget before returning +the admission seat. Cleanup invalidates queued viewer Relays, as it does during Transfer. Two related startup behaviours have no environment variable, because there is no case for turning them off: diff --git a/gateway/src/world/egress.rs b/gateway/src/world/egress.rs index 87aa777bb..2308de611 100644 --- a/gateway/src/world/egress.rs +++ b/gateway/src/world/egress.rs @@ -181,12 +181,25 @@ pub(super) async fn write( ) -> Result<()> { let mut trace = crate::config::writer_trace_enabled().then(WriterTrace::new); let mut buffer = Vec::with_capacity(4096); - while let Some(first) = receiver.recv().await { + let mut pending = None; + loop { + let first = match pending.take() { + Some(first) => first, + None => match receiver.recv().await { + Some(first) => first, + None => break, + }, + }; let mut oldest = first.since; let mut batch = vec![first]; while batch.len() < ENCODE_BATCH { match receiver.try_recv() { Ok(next) => { + // A Relay job can close the socket; write preceding packets before invoking it. + if matches!(next.out, Some(Outbound::Job(_))) { + pending = Some(next); + break; + } oldest = oldest.min(next.since); batch.push(next); } diff --git a/gateway/src/world/framing_tests.rs b/gateway/src/world/framing_tests.rs index a5335eb0f..53c6e5dd1 100644 --- a/gateway/src/world/framing_tests.rs +++ b/gateway/src/world/framing_tests.rs @@ -174,7 +174,7 @@ fn a_session_ending_in_a_framing_error_still_gives_its_seat_back() { let (mut client, server_end) = world_session_socket_pair(); let server_queue = queue.clone(); let server = std::thread::spawn(move || { - run_world_session_with_queue(server_end, store.clone(), &server_queue) + run_world_session_with_queue(server_end, store.clone(), server_queue) }); let (mut c_enc, _c_dec) = client_handshake(&mut client, "TESTER", K); assert_eq!( @@ -214,7 +214,7 @@ fn a_session_that_dies_in_world_still_deletes_the_players_entity() { let server_store = store.clone(); let server_queue = queue.clone(); let server = std::thread::spawn(move || { - run_world_session_with_queue(server_end, server_store.clone(), &server_queue) + run_world_session_with_queue(server_end, server_store.clone(), server_queue) }); let (mut c_enc, mut c_dec) = client_handshake(&mut client, "TESTER", K); diff --git a/gateway/src/world/mail_tests.rs b/gateway/src/world/mail_tests.rs index 8ec741251..a92e04f5d 100644 --- a/gateway/src/world/mail_tests.rs +++ b/gateway/src/world/mail_tests.rs @@ -4285,7 +4285,7 @@ fn a_mail_poll_after_realm_core_loss_ends_the_admitted_world_session() { let queue = queue.clone(); std::thread::spawn(move || { finished - .send(run_world_session_with_queue(socket, store, &queue)) + .send(run_world_session_with_queue(socket, store, queue)) .unwrap(); }) }; diff --git a/gateway/src/world/mod.rs b/gateway/src/world/mod.rs index df7af41ff..1088066da 100644 --- a/gateway/src/world/mod.rs +++ b/gateway/src/world/mod.rs @@ -1,8 +1,6 @@ use crate::accept::{classify_accept_error, AcceptBackoff, AcceptOutcome}; use crate::codec::auth_session::{read_auth_session, CMSG_AUTH_SESSION_OPCODE}; use crate::read_deadline::WORLD_AUTH_DEADLINE; -#[cfg(test)] -use crate::read_deadline::{is_io_deadline, DeadlineClock, IoDeadline, PreAuthDeadline}; use crate::stdb::PlayerSubscriptions; use crate::{codec, config::GatewayConfig, stdb::Coordinator}; use anyhow::{anyhow, Result}; @@ -116,19 +114,6 @@ enum SessionSender { } impl SessionTx { - #[cfg(test)] - fn bind_socket(&mut self, socket: S) { - let socket = std::sync::Mutex::new(socket); - self.close_socket = Some(Arc::new(move || { - let socket = socket - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()); - if let Err(error) = socket.shutdown_both() { - log::warn!("World Session socket shutdown failed: {error}"); - } - })); - } - pub(crate) fn close(&self) { if let Some(close) = &self.close_socket { close(); @@ -183,32 +168,6 @@ impl SessionTx { } } -#[cfg(test)] -pub trait DuplexStream: Read + Write + IoDeadline + Send + Sized + 'static { - fn try_clone(&self) -> std::io::Result; - fn shutdown_both(&self) -> std::io::Result<()>; -} - -#[cfg(test)] -impl DuplexStream for std::net::TcpStream { - fn try_clone(&self) -> std::io::Result { - std::net::TcpStream::try_clone(self) - } - fn shutdown_both(&self) -> std::io::Result<()> { - self.shutdown(std::net::Shutdown::Both) - } -} - -#[cfg(all(test, unix))] -impl DuplexStream for std::os::unix::net::UnixStream { - fn try_clone(&self) -> std::io::Result { - std::os::unix::net::UnixStream::try_clone(self) - } - fn shutdown_both(&self) -> std::io::Result<()> { - self.shutdown(std::net::Shutdown::Both) - } -} - #[derive(Clone, Debug)] pub struct WorldSession { pub account_id: u64, @@ -351,143 +310,6 @@ const QUEUE_RESEND_INTERVAL: Duration = Duration::from_secs(5); #[cfg(test)] const QUEUE_RESEND_INTERVAL: Duration = Duration::from_millis(50); -#[cfg(test)] -pub fn world_handshake( - stream: &mut S, - store: Arc, -) -> Result> { - world_handshake_with_queue(stream, store, &LoginQueue::unlimited()) -} - -#[cfg(test)] -pub fn world_handshake_with_queue( - stream: &mut S, - store: Arc, - queue: &LoginQueue, -) -> Result> { - let mut deadline = PreAuthDeadline::after(WORLD_AUTH_DEADLINE); - world_handshake_with_queue_and_deadline(stream, store, queue, &mut deadline) -} - -#[cfg(test)] -fn world_handshake_with_queue_and_deadline( - stream: &mut S, - store: Arc, - queue: &LoginQueue, - deadline: &mut PreAuthDeadline, -) -> Result> { - let seed = ProofSeed::new(); - let server_seed = seed.seed(); - SMSG_AUTH_CHALLENGE { server_seed }.write_unencrypted_server(&mut deadline.io(stream))?; - - let session = match read_auth_session(&mut deadline.io(stream)) { - Ok(claim) => claim, - Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None), - Err(e) if is_io_deadline(&e) => { - return Err(anyhow!( - "no CMSG_AUTH_SESSION before the total pre-auth deadline" - )); - } - Err(e) => return Err(anyhow!("world auth read error: {e}")), - }; - - let username = session.username.to_uppercase(); - let Some(WorldSession { - account_id, - session_key, - }) = store.lookup_session(&username)? - else { - SMSG_AUTH_RESPONSE::AuthUnknownAccount - .write_unencrypted_server(&mut deadline.io(stream))?; - return Ok(None); - }; - - let ns = NormalizedString::new(&username).map_err(|e| anyhow!("bad username: {e}"))?; - let crypto = match seed.into_server_header_crypto( - &ns, - session_key, - session.client_proof, - session.client_seed, - ) { - Ok(c) => c, - Err(_) => { - SMSG_AUTH_RESPONSE::AuthFailed.write_unencrypted_server(&mut deadline.io(stream))?; - return Ok(None); - } - }; - let (mut encrypt, decrypt) = crypto.split(); - - deadline - .finish(stream) - .map_err(|e| anyhow!("clear world pre-auth I/O deadline: {e}"))?; - - if let Admission::Queued(ticket) = queue.request() { - if !wait_for_admission(&mut *stream, &mut encrypt, queue, ticket)? { - queue.cancel(ticket); - return Ok(None); - } - } - - if let Err(e) = (SMSG_AUTH_RESPONSE::AuthOk { - billing_flags: 0, - billing_rested: 0, - billing_time: 0, - }) - .write_encrypted_server(&mut *stream, &mut encrypt) - { - queue.depart(); - return Err(anyhow!("world auth write error (AUTH_OK): {e}")); - } - - Ok(Some(( - WorldConn { - session_claim: None, - protocol: ProtocolSession::new(account_id, username), - decrypt, - move_coalesce: CoalesceState::default(), - unavailable_notices: Default::default(), - store: RoutedStore::new(store), - session_key: Some(session_key), - guild_signed_on: None, - move_desync_drops: 0, - chat_flood: Default::default(), - }, - encrypt, - ))) -} - -#[cfg(test)] -fn wait_for_admission( - stream: &mut S, - encrypt: &mut EncrypterHalf, - queue: &LoginQueue, - ticket: login_queue::Ticket, -) -> Result { - let mut last_sent: Option<(Instant, usize)> = None; - loop { - let Some(position) = queue.position(ticket) else { - return Ok(true); - }; - let due = match last_sent { - None => true, - Some((_, sent_position)) if sent_position != position => true, - Some((at, _)) => at.elapsed() >= QUEUE_RESEND_INTERVAL, - }; - if due { - let sent = (SMSG_AUTH_RESPONSE::AuthWaitQueue { - queue_position: position as u32, - }) - .write_encrypted_server(&mut *stream, encrypt); - if let Err(e) = sent { - log::debug!("world: queued connection hung up at position {position}: {e}"); - return Ok(false); - } - last_sent = Some((Instant::now(), position)); - } - std::thread::sleep(QUEUE_POLL_INTERVAL); - } -} - fn fnv1a64(bytes: &[u8]) -> u64 { let mut hash: u64 = 0xcbf29ce484222325; for &b in bytes { @@ -528,33 +350,6 @@ impl WriterTrace { }); } - #[cfg(test)] - fn record(&mut self, out: &Outbound) { - match out { - Outbound::One(m) => self.record_typed(m), - Outbound::Batch(v) => { - for m in v { - self.record_typed(m); - } - } - Outbound::Raw { opcode, body } => { - self.push(*opcode, (2 + body.len()) as u16, fnv1a64(body)); - } - Outbound::Job(_) => {} - } - } - - #[cfg(test)] - fn record_typed(&mut self, m: &ServerOpcodeMessage) { - let mut buf = Vec::new(); - if m.write_unencrypted_server(&mut buf).is_err() || buf.len() < 4 { - return; - } - let size = u16::from_be_bytes([buf[0], buf[1]]); - let opcode = u16::from_le_bytes([buf[2], buf[3]]); - self.push(opcode, size, fnv1a64(&buf[4..])); - } - fn dump(&self, account_id: u64, reason: &str) { self.dump_to( std::path::Path::new("/tmp/gw-writer-crash"), @@ -591,179 +386,6 @@ impl WriterTrace { } } -#[cfg(test)] -fn spawn_writer( - mut wsock: S, - mut encrypt: EncrypterHalf, - rx: Receiver, - depth: Arc, - account_id: u64, -) -> Result> { - let trace_enabled = crate::config::writer_trace_enabled(); - std::thread::Builder::new() -.name("world-writer".into()) -.spawn(move || { - let mut trace = trace_enabled.then(WriterTrace::new); - while let Ok(out) = rx.recv() { - depth.fetch_sub(1, Ordering::Relaxed); - - let batch: Vec = match out { - Outbound::Job(job) => { - match std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)) { - Ok(items) => items, - Err(_) => { - log::error!( - "world writer: a deferred relay job PANICKED for account \ - {account_id} — contained (this session keeps running), but \ - the packet(s) it would have produced were lost" - ); - Vec::new() - } - } - } - other => vec![other], - }; - for out in batch { - if let Some(t) = trace.as_mut() { - t.record(&out); - } - let res = match out { - Outbound::One(m) => { - log::debug!("OUT {m}"); - m.write_encrypted_server(&mut wsock, &mut encrypt) - } - Outbound::Batch(v) => v.into_iter().try_for_each(|m| { - log::debug!("OUT {m}"); - m.write_encrypted_server(&mut wsock, &mut encrypt) - }), - Outbound::Job(_) => Ok(()), - Outbound::Raw { opcode, body } => { - log::debug!("OUT RAW opcode=0x{opcode:04X} ({} body bytes)", body.len()); - for v in packet_lint::lint_raw(opcode, &body) { - log::error!("packet-lint VIOLATION opcode=0x{opcode:04X}: {v}"); - } - // Size counts the two-byte opcode and body, excluding itself. - if body.len() > u16::MAX as usize - 2 { - log::error!( - "packet-lint VIOLATION opcode=0x{opcode:04X}: raw body is {} bytes — \ - exceeds the u16 frame-size field and would silently wrap (a stream-\ - desync class); refusing to send and ending the session instead of \ - corrupting the stream", - body.len() - ); - Err(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "raw outbound body exceeds u16 frame size", - )) - } else { - let size = (2 + body.len()) as u16; - encrypt -.write_encrypted_server_header(&mut wsock, size, opcode) -.and_then(|()| wsock.write_all(&body)) - } - } - }; - if let Err(e) = res { - log::debug!("world writer stopping: {e}"); - if let Some(t) = trace.as_ref() { - t.dump(account_id, &e.to_string()); - } - let _ = wsock.shutdown_both(); - return; - } - } - } - }) -.map_err(|e| anyhow!("spawn world writer thread for account {account_id}: {e}")) -} - -#[cfg(test)] -pub fn run_world_session(stream: S, store: Arc) -> Result<()> { - run_world_session_with_queue(stream, store, &LoginQueue::unlimited()) -} - -#[cfg(test)] -struct AdmissionSeat<'a> { - queue: &'a LoginQueue, -} - -#[cfg(test)] -impl Drop for AdmissionSeat<'_> { - fn drop(&mut self) { - self.queue.depart(); - } -} - -#[cfg(test)] -pub fn run_world_session_with_queue( - stream: S, - store: Arc, - queue: &LoginQueue, -) -> Result<()> { - let mut deadline = PreAuthDeadline::after(WORLD_AUTH_DEADLINE); - run_world_session_with_queue_and_deadline(stream, store, queue, &mut deadline) -} - -#[cfg(test)] -fn run_world_session_with_queue_and_deadline( - mut stream: S, - store: Arc, - queue: &LoginQueue, - deadline: &mut PreAuthDeadline, -) -> Result<()> { - let Some((mut conn, encrypt)) = - world_handshake_with_queue_and_deadline(&mut stream, store, queue, deadline)? - else { - return Ok(()); - }; - let seat = AdmissionSeat { queue }; - - let wsock = stream - .try_clone() - .map_err(|e| anyhow!("clone world socket for writer: {e}"))?; - let (mut tx, rx, depth) = session_channel(); - tx.bind_socket( - stream - .try_clone() - .map_err(|error| anyhow!("clone World Session shutdown socket: {error}"))?, - ); - let writer = spawn_writer(wsock, encrypt, rx, depth, conn.account_id)?; - - let result = (|| -> Result<()> { - let mut addon = AddonLimit::default(); - loop { - let hdr = match conn.decrypt.read_and_decrypt_client_header(&mut stream) { - Ok(h) => h, - Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(()), - Err(e) => return Err(anyhow!("world read error: {e}")), - }; - // Client size counts the four-byte opcode and body. - let body_len = (hdr.size as usize).saturating_sub(4); - let mut body = vec![0u8; body_len]; - stream - .read_exact(&mut body) - .map_err(|e| anyhow!("world read error (body): {e}"))?; - process_frame(&tx, &mut conn, &mut addon, hdr, body)?; - } - })(); - - log::debug!( - "world: movement coalescing account={} received={} forwarded={}", - conn.account_id, - conn.move_coalesce.received(), - conn.move_coalesce.forwarded(), - ); - - if let Err(e) = conn.leave_world() { - log::warn!("logout for account {} failed: {e:#}", conn.account_id); - } - drop(tx); - let _ = writer.join(); - // Release the seat exactly once, after the writer and Relays stop. - drop(seat); - result -} - struct AddonLimit { tokens: f32, refill_at: Instant, @@ -1272,9 +894,16 @@ pub async fn run( let actions = actions.clone(); let encoders = encoders.clone(); tokio::spawn(async move { - let _connection_permit = connection_permit; - if let Err(error) = - transport::run_at(sock, Arc::new(coord), queue, actions, encoders, deadline).await + if let Err(error) = transport::run_at( + sock, + Arc::new(coord), + queue, + actions, + encoders, + connection_permit, + deadline, + ) + .await { log::warn!("world session {peer} ended: {error:#}"); } @@ -1285,4 +914,8 @@ pub async fn run( #[cfg(all(test, unix))] mod test_support; #[cfg(all(test, unix))] +pub(crate) use test_support::run_world_session; +#[cfg(all(test, unix))] +use transport::world_handshake; +#[cfg(all(test, unix))] pub(crate) mod tests; diff --git a/gateway/src/world/session_tests.rs b/gateway/src/world/session_tests.rs index 78ca78eff..26468798c 100644 --- a/gateway/src/world/session_tests.rs +++ b/gateway/src/world/session_tests.rs @@ -1,56 +1,5 @@ use super::*; -#[derive(Clone)] -struct ManualClock(Rc>); - -impl DeadlineClock for ManualClock { - fn now(&self) -> Instant { - self.0.get() - } -} - -struct AdvancingStream { - input: Cursor>, - output: Vec, - clock: ManualClock, - per_read: Duration, - per_write: Duration, - write_timeout_calls: Rc>, -} - -impl Read for AdvancingStream { - fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - let one_byte = buf.len().min(1); - let read = self.input.read(&mut buf[..one_byte])?; - self.clock.0.set(self.clock.0.get() + self.per_read); - Ok(read) - } -} - -impl Write for AdvancingStream { - fn write(&mut self, buf: &[u8]) -> std::io::Result { - self.output.extend_from_slice(buf); - self.clock.0.set(self.clock.0.get() + self.per_write); - Ok(buf.len()) - } - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } -} - -impl IoDeadline for AdvancingStream { - fn set_read_timeout(&self, _timeout: Option) -> std::io::Result<()> { - Ok(()) - } - - fn set_write_timeout(&self, _timeout: Option) -> std::io::Result<()> { - self.write_timeout_calls - .set(self.write_timeout_calls.get() + 1); - Ok(()) - } -} - #[test] fn world_session_socket_pair_times_out_when_the_server_writes_nothing() { let (mut client, _server) = world_session_socket_pair(); @@ -148,13 +97,13 @@ fn an_auth_session_with_an_absurd_addon_size_still_completes_the_handshake() { fn a_silent_world_connection_is_closed_at_the_pre_auth_deadline() { let store = tester_store(42); let (mut client, server_end) = world_session_socket_pair(); - let mut deadline = PreAuthDeadline::after(Duration::from_millis(200)); + let deadline = Instant::now() + Duration::from_millis(200); let server = std::thread::spawn(move || { - run_world_session_with_queue_and_deadline( + run_world_session_with_deadline( server_end, std::sync::Arc::new(store), - &LoginQueue::unlimited(), - &mut deadline, + Arc::new(LoginQueue::unlimited()), + deadline, ) }); @@ -175,101 +124,18 @@ fn a_silent_world_connection_is_closed_at_the_pre_auth_deadline() { ); } -#[test] -fn slow_auth_session_bytes_cannot_extend_the_pre_auth_deadline() { - let store = tester_store(42); - let mut input = Vec::new(); - auth_session("TESTER", 1, [0; 20]) - .write_unencrypted_client(&mut input) - .unwrap(); - let start = Instant::now(); - let clock = ManualClock(Rc::new(Cell::new(start))); - let mut deadline = PreAuthDeadline::with_clock(start + Duration::from_millis(8), clock.clone()); - let mut stream = AdvancingStream { - input: Cursor::new(input), - output: Vec::new(), - clock, - per_read: Duration::from_millis(1), - per_write: Duration::ZERO, - write_timeout_calls: Rc::new(Cell::new(0)), - }; - - let result = world_handshake_with_queue_and_deadline( - &mut stream, - std::sync::Arc::new(store), - &LoginQueue::unlimited(), - &mut deadline, - ); - let error = match result { - Err(error) => error, - Ok(_) => panic!("a slow auth frame must share one total deadline"), - }; - - assert!( - error.to_string().contains("total pre-auth deadline"), - "{error:#}" - ); - assert!( - !stream.output.is_empty(), - "the server challenge must be written before the slow client frame times out" - ); -} - -#[test] -fn world_challenge_write_uses_the_absolute_pre_auth_deadline() { - let store = tester_store(42); - let start = Instant::now(); - let clock = ManualClock(Rc::new(Cell::new(start))); - let write_timeout_calls = Rc::new(Cell::new(0)); - let mut deadline = - PreAuthDeadline::with_clock(start + Duration::from_millis(10), clock.clone()); - let mut stream = AdvancingStream { - input: Cursor::new(Vec::new()), - output: Vec::new(), - clock, - per_read: Duration::ZERO, - per_write: Duration::from_millis(20), - write_timeout_calls: write_timeout_calls.clone(), - }; - - let result = world_handshake_with_queue_and_deadline( - &mut stream, - std::sync::Arc::new(store), - &LoginQueue::unlimited(), - &mut deadline, - ); - let error = match result { - Err(error) => error, - Ok(_) => panic!("the world challenge must not write past the total deadline"), - }; - - assert_eq!( - write_timeout_calls.get(), - 1, - "the world challenge must write through the pre-auth I/O wrapper" - ); - assert!( - error.to_string().contains("pre-auth I/O deadline"), - "{error:#}" - ); - assert!( - !stream.output.is_empty(), - "the fake advances time after accepting the world challenge bytes" - ); -} - #[test] fn post_auth_world_traffic_has_no_pre_auth_deadline() { let store = std::sync::Arc::new(tester_store(42)); let (mut client, server_end) = world_session_socket_pair(); - let mut deadline = PreAuthDeadline::after(Duration::from_millis(200)); + let deadline = Instant::now() + Duration::from_millis(200); let server_store = store.clone(); let server = std::thread::spawn(move || { - run_world_session_with_queue_and_deadline( + run_world_session_with_deadline( server_end, server_store.clone(), - &LoginQueue::unlimited(), - &mut deadline, + Arc::new(LoginQueue::unlimited()), + deadline, ) }); @@ -296,12 +162,7 @@ fn queued_handshake_sends_wait_queue_then_admits_once_a_seat_frees() { let server_store = store.clone(); let server_queue = queue.clone(); let server = std::thread::spawn(move || { - let mut s = server_end; - let (conn, _encrypt) = - world_handshake_with_queue(&mut s, server_store.clone(), &server_queue) - .unwrap() - .expect("handshake should eventually succeed once admitted"); - assert_eq!(conn.account_id, 42); + run_world_session_with_queue(server_end, server_store, server_queue) }); let (_c_enc, mut c_dec) = drive_auth(&mut client, "TESTER", K); @@ -329,7 +190,7 @@ fn queued_handshake_sends_wait_queue_then_admits_once_a_seat_frees() { } drop(client); - server.join().unwrap(); + server.join().unwrap().unwrap(); } #[test] @@ -342,13 +203,7 @@ fn disconnecting_while_queued_leaves_the_line_without_taking_a_seat() { let server_store = store.clone(); let server_queue = queue.clone(); let server = std::thread::spawn(move || { - let mut s = server_end; - let result = - world_handshake_with_queue(&mut s, server_store.clone(), &server_queue).unwrap(); - assert!( - result.is_none(), - "a hangup while queued must end the session cleanly, not error" - ); + run_world_session_with_queue(server_end, server_store, server_queue) }); let (_c_enc, mut c_dec) = drive_auth(&mut client, "TESTER", K); @@ -361,7 +216,7 @@ fn disconnecting_while_queued_leaves_the_line_without_taking_a_seat() { } drop(client); - server.join().unwrap(); + let _ = server.join().unwrap(); assert_eq!( queue.depth(), @@ -1899,8 +1754,7 @@ fn reducer_transport_loss_ends_an_admitted_session_and_frees_one_queue_seat() { let server_queue = queue.clone(); let (result_tx, result_rx) = std::sync::mpsc::channel(); std::thread::spawn(move || { - let result = - run_world_session_with_queue(server_end, server_store.clone(), server_queue.as_ref()); + let result = run_world_session_with_queue(server_end, server_store.clone(), server_queue); result_tx.send(result).unwrap(); }); @@ -2106,65 +1960,6 @@ fn competing_world_sessions_close_the_old_socket_without_removing_the_winner() { .is_empty()); } -#[cfg(target_os = "linux")] -#[test] -fn closing_a_world_session_interrupts_a_full_socket_without_draining_its_queue() { - use std::os::fd::AsRawFd; - - let (_client, mut socket) = world_session_socket_pair(); - socket.set_nonblocking(true).unwrap(); - loop { - match socket.write(&[0; 65536]) { - Ok(_) => {} - Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => break, - Err(error) => panic!("fill owned socket: {error}"), - } - } - socket.set_nonblocking(false).unwrap(); - let (mut tx, queued, _) = session_channel(); - tx.bind_socket(socket.try_clone().unwrap()); - tx.send(Outbound::Raw { - opcode: 0, - body: vec![], - }) - .unwrap(); - let socket_fd = socket.as_raw_fd(); - let (started, ready) = std::sync::mpsc::channel(); - let (finished, result) = std::sync::mpsc::channel(); - let writer = std::thread::spawn(move || { - let task = std::fs::read_link("/proc/thread-self") - .expect("this Linux socket fixture requires /proc/thread-self"); - started.send(task).unwrap(); - finished.send(socket.write_all(&[1; 65536])).unwrap(); - }); - let task = ready.recv_timeout(Duration::from_secs(2)).unwrap(); - let syscall_path = std::path::Path::new("/proc").join(task).join("syscall"); - assert!( - crate::durable_test_support::poll_until(Duration::from_secs(2), || { - let syscall = std::fs::read_to_string(&syscall_path) - .expect("this Linux socket fixture requires visibility of its writer's syscall"); - let mut fields = syscall.split_whitespace(); - let number = fields - .next() - .and_then(|value| value.parse::().ok()); - let fd = fields - .next() - .and_then(|value| value.strip_prefix("0x")) - .and_then(|value| i32::from_str_radix(value, 16).ok()); - matches!(number, Some(n) if n == libc::SYS_sendto || n == libc::SYS_write) - && fd == Some(socket_fd) - }), - "writer did not enter its blocked socket write" - ); - tx.close(); - assert!(result - .recv_timeout(Duration::from_secs(2)) - .unwrap() - .is_err()); - assert!(matches!(queued.try_recv().unwrap(), Outbound::Raw { .. })); - writer.join().unwrap(); -} - #[test] fn an_addon_command_refusal_is_dropped_and_a_transport_loss_ends_the_session() { let conn = in_world_conn(42, 1); diff --git a/gateway/src/world/test_support/mod.rs b/gateway/src/world/test_support/mod.rs index 6b4b93188..b782323b3 100644 --- a/gateway/src/world/test_support/mod.rs +++ b/gateway/src/world/test_support/mod.rs @@ -1,10 +1,12 @@ mod entry; mod family; +mod socket; mod topology; mod world_fake; pub(crate) use self::entry::*; pub(crate) use self::family::*; +pub(crate) use self::socket::*; pub(crate) use self::topology::*; pub(crate) use self::world_fake::*; diff --git a/gateway/src/world/test_support/socket.rs b/gateway/src/world/test_support/socket.rs new file mode 100644 index 000000000..6d9aba85b --- /dev/null +++ b/gateway/src/world/test_support/socket.rs @@ -0,0 +1,91 @@ +use super::super::*; +use crate::accept::BlockingTaskCapacity; +use std::io; +use tokio::io::{AsyncRead, AsyncWrite}; + +pub(crate) trait TestSocket: Sized { + type Async: AsyncRead + AsyncWrite + Unpin + Send + 'static; + fn into_async(self) -> io::Result; +} + +impl TestSocket for std::net::TcpStream { + type Async = tokio::net::TcpStream; + + fn into_async(self) -> io::Result { + self.set_nonblocking(true)?; + tokio::net::TcpStream::from_std(self) + } +} + +impl TestSocket for std::os::unix::net::UnixStream { + type Async = tokio::net::UnixStream; + + fn into_async(self) -> io::Result { + self.set_nonblocking(true)?; + tokio::net::UnixStream::from_std(self) + } +} + +pub(crate) fn run_world_session( + stream: S, + store: Arc, +) -> Result<()> { + run_world_session_with_queue(stream, store, Arc::new(LoginQueue::unlimited())) +} + +pub(crate) fn run_world_session_with_queue( + stream: S, + store: Arc, + queue: Arc, +) -> Result<()> { + run_world_session_with_deadline(stream, store, queue, Instant::now() + WORLD_AUTH_DEADLINE) +} + +pub(crate) fn run_world_session_with_deadline( + stream: S, + store: Arc, + queue: Arc, + deadline: Instant, +) -> Result<()> { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .max_blocking_threads(4) + .build()? + .block_on(async { + let permit = Arc::new(tokio::sync::Semaphore::new(1)).try_acquire_owned()?; + super::super::transport::run_at( + stream.into_async()?, + store, + queue, + BlockingTaskCapacity::new(2), + BlockingTaskCapacity::new(2), + permit, + tokio::time::Instant::from_std(deadline), + ) + .await + }) +} + +pub(crate) fn start_writer( + stream: std::os::unix::net::UnixStream, + encrypt: EncrypterHalf, +) -> (SessionTx, std::thread::JoinHandle>) { + let (tx, receiver, _) = egress::channel(); + let writer = std::thread::spawn(move || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .max_blocking_threads(1) + .build()? + .block_on(async { + egress::write( + stream.into_async()?, + encrypt, + receiver, + BlockingTaskCapacity::new(1), + 1, + ) + .await + }) + }); + (tx, writer) +} diff --git a/gateway/src/world/test_support/world_fake/session.rs b/gateway/src/world/test_support/world_fake/session.rs index 6e0a3c018..37d81a565 100644 --- a/gateway/src/world/test_support/world_fake/session.rs +++ b/gateway/src/world/test_support/world_fake/session.rs @@ -14,6 +14,10 @@ pub(crate) struct SessionState { pub(crate) session: Option, pub(crate) login_entity: Option, pub(crate) login_pause: Option>, + pub(crate) subscribe_pause: Option<( + std::sync::mpsc::Sender, + std::sync::Mutex>, + )>, pub(crate) moves: std::sync::Mutex>, pub(crate) move_bodies: std::sync::Mutex>>, pub(crate) client_commands: std::sync::Mutex>, @@ -117,6 +121,14 @@ impl SessionStore for WorldFake { arrival.y, )); *self.session.session_depth.lock().unwrap() = Some(tx.depth_handle()); + if let Some((entered, resume)) = &self.session.subscribe_pause { + entered.send(tx.clone()).unwrap(); + resume + .lock() + .unwrap() + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + } if self.session.turn_in_reward_item.is_some() { *self.session.turn_in_tx.lock().unwrap() = Some(tx.clone()); } diff --git a/gateway/src/world/tests.rs b/gateway/src/world/tests.rs index c727ebd04..a9610da5f 100644 --- a/gateway/src/world/tests.rs +++ b/gateway/src/world/tests.rs @@ -4,15 +4,11 @@ use super::handlers::{ use super::party::PartyOutcome; use super::test_support::*; use super::*; -use crate::read_deadline::{DeadlineClock, PreAuthDeadline}; use lyracore_shared::group::{GroupKind, RaidSlot}; use lyracore_shared::item::ItemRefusal; use lyracore_shared::loot::LootRefusal; use lyracore_shared::social::ContactRefusal; -use std::cell::Cell; -use std::io::Cursor; use std::os::unix::net::UnixStream; -use std::rc::Rc; use wow_world_base::shared::friend_result_vanilla_tbc::FriendResult; use wow_world_messages::vanilla::opcodes::ServerOpcodeMessage; use wow_world_messages::vanilla::{ diff --git a/gateway/src/world/transport.rs b/gateway/src/world/transport.rs index 7d3eee9be..20b0af5d3 100644 --- a/gateway/src/world/transport.rs +++ b/gateway/src/world/transport.rs @@ -141,6 +141,7 @@ pub(super) async fn run_at( queue: Arc, actions: BlockingTaskCapacity, encoders: BlockingTaskCapacity, + connection_permit: tokio::sync::OwnedSemaphorePermit, deadline: tokio::time::Instant, ) -> Result<()> where @@ -155,9 +156,10 @@ where } let _stop_on_drop = StopOnDrop(stop); // The supervisor owns cleanup even if the caller drops its session future. - tokio::spawn(run_session( - stream, store, queue, actions, encoders, receiver, deadline, - )) + tokio::spawn(async move { + let _connection_permit = connection_permit; + run_session(stream, store, queue, actions, encoders, receiver, deadline).await + }) .await? } @@ -191,10 +193,17 @@ where let conn = Arc::new(Mutex::new(conn)); let addon = Arc::new(Mutex::new(AddonLimit::default())); let (tx, receiver, mut closed) = egress::channel(); - let (mut reader, writer) = tokio::io::split(stream); + let (reader, writer) = tokio::io::split(stream); let writer_close = tx.close_socket.clone(); + let mut writer_stopped = stopped.clone(); + let mut writer_closed = closed.clone(); let mut writer_task = tokio::spawn(async move { - let result = egress::write(writer, encrypt, receiver, encoders, account_id).await; + let result = tokio::select! { + biased; + _ = writer_stopped.wait_for(|stop| *stop) => Ok(()), + _ = writer_closed.wait_for(|closed| *closed) => Ok(()), + result = egress::write(writer, encrypt, receiver, encoders, account_id) => result, + }; if let Some(close) = writer_close { close(); } @@ -203,10 +212,16 @@ where let mut drain = true; let result = async { + let mut reader = reader; loop { let frame = async { let mut bytes = [0; 6]; - reader.read_exact(&mut bytes).await?; + if let Err(error) = reader.read_exact(&mut bytes).await { + if error.kind() == std::io::ErrorKind::UnexpectedEof { + return Ok(None); + } + return Err(anyhow!(error).context("world read error (header)")); + } let header = conn .lock() .unwrap_or_else(|error| error.into_inner()) @@ -219,30 +234,48 @@ where ) })?; let mut body = vec![0; length]; - reader.read_exact(&mut body).await?; - Ok::<_, std::io::Error>((header, body)) + reader + .read_exact(&mut body) + .await + .map_err(|error| anyhow!(error).context("world read error (body)"))?; + Ok::<_, anyhow::Error>(Some((header, body))) }; let (header, body) = tokio::select! { biased; _ = stopped.wait_for(|stop| *stop) => { drain = false; break; }, _ = closed.wait_for(|closed| *closed) => { drain = false; break; }, result = frame => match result { - Ok(frame) => frame, - Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => break, - Err(error) => return Err(error.into()), + Ok(Some(frame)) => frame, + Ok(None) => break, + Err(error) => return Err(error), } }; let conn = conn.clone(); let addon = addon.clone(); - let tx = tx.clone(); + let action_tx = tx.clone(); // Await accepted game work before cleanup so a late login cannot outlive its session. - actions - .run(move || { - let mut conn = conn.lock().unwrap_or_else(|error| error.into_inner()); - let mut addon = addon.lock().unwrap_or_else(|error| error.into_inner()); - process_frame(&tx, &mut conn, &mut addon, header, body) - }) - .await??; + let action = actions.run(move || { + let mut conn = conn.lock().unwrap_or_else(|error| error.into_inner()); + let mut addon = addon.lock().unwrap_or_else(|error| error.into_inner()); + process_frame(&action_tx, &mut conn, &mut addon, header, body) + }); + tokio::pin!(action); + let finished = tokio::select! { + biased; + _ = stopped.wait_for(|stop| *stop) => None, + _ = closed.wait_for(|closed| *closed) => None, + result = &mut action => Some(result), + }; + match finished { + Some(result) => result??, + None => { + drain = false; + drop(reader); + tx.close(); + writer_task.abort(); + return action.await?; + } + } } Ok::<_, anyhow::Error>(()) } @@ -300,15 +333,113 @@ where queue, actions, encoders, + Arc::new(tokio::sync::Semaphore::new(1)).try_acquire_owned()?, tokio::time::Instant::now() + WORLD_AUTH_DEADLINE, ) .await } +#[cfg(all(test, unix))] +pub(super) fn world_handshake( + stream: &mut std::os::unix::net::UnixStream, + store: Arc, +) -> Result> { + use super::test_support::TestSocket; + let socket = stream.try_clone()?; + let result = tokio::runtime::Builder::new_current_thread() + .enable_all() + .max_blocking_threads(1) + .build()? + .block_on(async { + let mut socket = socket.into_async()?; + let Some((conn, mut encrypt)) = tokio::time::timeout( + WORLD_AUTH_DEADLINE, + authenticate(&mut socket, store, &BlockingTaskCapacity::new(1)), + ) + .await?? + else { + return Ok(None); + }; + let _seat = admit(&mut socket, &mut encrypt, Arc::new(LoginQueue::unlimited())).await?; + Ok(Some((conn, encrypt))) + }); + stream.set_nonblocking(false)?; + result +} + #[cfg(test)] mod tests { use super::*; + #[cfg(unix)] + #[tokio::test] + async fn slow_auth_bytes_cannot_extend_the_total_deadline() { + use crate::world::test_support::{auth_session, tester_store}; + use wow_world_messages::vanilla::ClientMessage; + + let (mut client, server) = tokio::io::duplex(128); + let connections = Arc::new(tokio::sync::Semaphore::new(1)); + let deadline = tokio::time::Instant::now() + Duration::from_millis(80); + let session = tokio::spawn(run_at( + server, + Arc::new(tester_store(1)), + Arc::new(LoginQueue::unlimited()), + BlockingTaskCapacity::new(1), + BlockingTaskCapacity::new(1), + connections.clone().try_acquire_owned().unwrap(), + deadline, + )); + let mut challenge = [0; 8]; + client.read_exact(&mut challenge).await.unwrap(); + let mut frame = Vec::new(); + auth_session("TESTER", 1, [0; 20]) + .write_unencrypted_client(&mut frame) + .unwrap(); + let length = frame.len(); + let sent = Arc::new(AtomicUsize::new(0)); + let progress = sent.clone(); + let sender = tokio::spawn(async move { + for byte in frame { + if client.write_all(&[byte]).await.is_err() { + break; + } + progress.fetch_add(1, Ordering::Relaxed); + tokio::time::sleep(Duration::from_millis(5)).await; + } + }); + let error = tokio::time::timeout(Duration::from_secs(2), session) + .await + .unwrap() + .unwrap() + .unwrap_err(); + sender.await.unwrap(); + assert!(error.to_string().contains("total pre-auth deadline")); + assert!((1..length).contains(&sent.load(Ordering::Relaxed))); + assert_eq!(connections.available_permits(), 1); + } + + #[cfg(unix)] + #[tokio::test] + async fn a_blocked_challenge_write_uses_the_total_deadline() { + use crate::world::test_support::tester_store; + + let (_client, server) = tokio::io::duplex(1); + let connections = Arc::new(tokio::sync::Semaphore::new(1)); + let error = run_at( + server, + Arc::new(tester_store(1)), + Arc::new(LoginQueue::unlimited()), + BlockingTaskCapacity::new(1), + BlockingTaskCapacity::new(1), + connections.clone().try_acquire_owned().unwrap(), + tokio::time::Instant::now() + Duration::from_millis(20), + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("total pre-auth deadline")); + assert_eq!(connections.available_permits(), 1); + } + #[cfg(unix)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn idle_world_sessions_leave_the_action_worker_available() { @@ -515,12 +646,15 @@ mod tests { let (mut client, server) = world_session_socket_pair(); server.set_nonblocking(true).unwrap(); let server = tokio::net::UnixStream::from_std(server).unwrap(); - let session = tokio::spawn(run( + let connections = Arc::new(tokio::sync::Semaphore::new(1)); + let session = tokio::spawn(run_at( server, store.clone(), queue.clone(), BlockingTaskCapacity::new(1), BlockingTaskCapacity::new(1), + connections.clone().try_acquire_owned().unwrap(), + tokio::time::Instant::now() + WORLD_AUTH_DEADLINE, )); let entered = pause.clone(); let client = tokio::task::spawn_blocking(move || { @@ -537,19 +671,109 @@ mod tests { .unwrap(); session.abort(); assert!(session.await.unwrap_err().is_cancelled()); - assert_eq!(queue.active(), 1); - assert!(!store.session.logout_called.load(Ordering::Relaxed)); + let active_before_cleanup = queue.active(); + let permits_before_cleanup = connections.available_permits(); + let logged_out_before_cleanup = store.session.logout_called.load(Ordering::Relaxed); + let closed = tokio::task::spawn_blocking(move || { + let mut client = client; + client + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + std::io::Read::read(&mut client, &mut [0]) + }) + .await + .unwrap(); tokio::task::spawn_blocking(move || pause.1.wait()) .await .unwrap(); tokio::time::timeout(Duration::from_secs(5), async { - while queue.active() != 0 { + while queue.active() != 0 || connections.available_permits() != 1 { tokio::time::sleep(Duration::from_millis(5)).await; } }) .await .unwrap(); + assert_eq!(active_before_cleanup, 1); + assert_eq!(permits_before_cleanup, 0); + assert!(!logged_out_before_cleanup); + assert_eq!( + closed.unwrap(), + 0, + "cancel must close the socket before cleanup" + ); + assert!(store.session.logout_called.load(Ordering::Relaxed)); + } + + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn closing_during_subscription_interrupts_the_socket_before_cleanup() { + use crate::world::test_support::{ + client_handshake, quest_store, world_session_socket_pair, K, + }; + use wow_world_messages::vanilla::{ClientMessage, CMSG_PLAYER_LOGIN}; + + let (entered, paused) = std::sync::mpsc::channel(); + let (resume, release) = std::sync::mpsc::channel(); + let mut fake = quest_store(); + fake.session.subscribe_pause = Some((entered, Mutex::new(release))); + let store = Arc::new(fake); + let queue = Arc::new(LoginQueue::new(1, 0)); + let connections = Arc::new(tokio::sync::Semaphore::new(1)); + let (mut client, server) = world_session_socket_pair(); + server.set_nonblocking(true).unwrap(); + let session = tokio::spawn(run_at( + tokio::net::UnixStream::from_std(server).unwrap(), + store.clone(), + queue.clone(), + BlockingTaskCapacity::new(1), + BlockingTaskCapacity::new(1), + connections.clone().try_acquire_owned().unwrap(), + tokio::time::Instant::now() + WORLD_AUTH_DEADLINE, + )); + let (client, tx) = tokio::task::spawn_blocking(move || { + let (mut encrypt, _) = client_handshake(&mut client, "TESTER", K); + CMSG_PLAYER_LOGIN { + guid: wow_world_messages::Guid::new(1), + } + .write_encrypted_client(&mut client, &mut encrypt) + .unwrap(); + let tx = paused.recv_timeout(Duration::from_secs(2)).unwrap(); + (client, tx) + }) + .await + .unwrap(); + for _ in 0..16 { + tx.send(Outbound::Raw { + opcode: 0, + body: vec![0; 60_000], + }) + .unwrap(); + } + tx.close(); + let closed = tokio::task::spawn_blocking(move || { + let mut client = client; + client + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + std::io::Read::read_to_end(&mut client, &mut Vec::new()) + }) + .await + .unwrap(); + let active_before_cleanup = queue.active(); + let permits_before_cleanup = connections.available_permits(); + let logged_out_before_cleanup = store.session.logout_called.load(Ordering::Relaxed); + resume.send(()).unwrap(); + drop(tx); + let _ = tokio::time::timeout(Duration::from_secs(2), session) + .await + .unwrap() + .unwrap(); + closed.expect("close must interrupt the socket while subscription is paused"); + assert_eq!(active_before_cleanup, 1); + assert_eq!(permits_before_cleanup, 0); + assert!(!logged_out_before_cleanup); + assert_eq!(queue.active(), 0); + assert_eq!(connections.available_permits(), 1); assert!(store.session.logout_called.load(Ordering::Relaxed)); - drop(client); } } diff --git a/gateway/src/world/wire_corruption_tests.rs b/gateway/src/world/wire_corruption_tests.rs index f090e3956..89368be3d 100644 --- a/gateway/src/world/wire_corruption_tests.rs +++ b/gateway/src/world/wire_corruption_tests.rs @@ -26,8 +26,7 @@ fn movement_burst_over_a_real_cipher(n: usize) { .join() .expect("server handshake thread must not panic"); - let (tx, rx, depth) = session_channel(); - let writer = spawn_writer(s, encrypt, rx, depth, 1).expect("spawn the writer thread"); + let (tx, writer) = start_writer(s, encrypt); // Read concurrently with the writer; joining first deadlocks once the socket buffer fills. let reader = std::thread::spawn(move || { @@ -52,6 +51,9 @@ fn movement_burst_over_a_real_cipher(n: usize) { }); for i in 0..n { + while tx.depth() >= 128 { + std::thread::sleep(Duration::from_millis(1)); + } let guid = 0x1000_0000_0000_0000u64 + i as u64; let start = wow_world_messages::vanilla::Vector3d { x: -8938.0 + i as f32, @@ -94,7 +96,8 @@ fn movement_burst_over_a_real_cipher(n: usize) { drop(tx); writer .join() - .expect("writer thread must not panic under a movement burst"); + .expect("writer thread must not panic under a movement burst") + .expect("movement frames must reach the client"); let (monster_moves, heartbeats) = reader.join().expect("reader thread must not panic"); assert_eq!( @@ -114,89 +117,6 @@ fn writer_thread_survives_movement_bursts_from_100_to_500_movers_without_desync( } } -#[test] -fn writer_trace_records_the_typed_path_opcode_and_size() { - let m = codec::build_monster_move( - 0xF130_0000_0000_0001, - wow_world_messages::vanilla::Vector3d { - x: -8938.0, - y: -131.0, - z: 83.5, - }, - wow_world_messages::vanilla::Vector3d { - x: -8900.0, - y: -100.0, - z: 82.4, - }, - 500, - 1, - true, - ); - let msg = ServerOpcodeMessage::SMSG_MONSTER_MOVE(Box::new(m)); - let mut plain = Vec::new(); - msg.write_unencrypted_server(&mut plain).unwrap(); - let expected_size = u16::from_be_bytes([plain[0], plain[1]]); - let expected_opcode = u16::from_le_bytes([plain[2], plain[3]]); - let expected_checksum = fnv1a64(&plain[4..]); - - let mut trace = WriterTrace::new(); - trace.record(&Outbound::One(msg)); - - assert_eq!( - trace.ring.len(), - 1, - "one Outbound::One must record exactly one frame" - ); - let entry = trace.ring[0]; - assert_eq!(entry.opcode, expected_opcode); - assert_eq!(entry.size, expected_size); - assert_eq!(entry.checksum, expected_checksum); -} - -#[test] -fn writer_trace_expands_a_batch_into_one_entry_per_message() { - let make = |i: u32| { - let m = codec::build_monster_move( - 0x2000_0000_0000_0000 + i as u64, - wow_world_messages::vanilla::Vector3d { - x: 0.0, - y: 0.0, - z: 0.0, - }, - wow_world_messages::vanilla::Vector3d { - x: 1.0, - y: 1.0, - z: 1.0, - }, - 100, - i + 1, - false, - ); - ServerOpcodeMessage::SMSG_MONSTER_MOVE(Box::new(m)) - }; - let mut trace = WriterTrace::new(); - trace.record(&Outbound::Batch(vec![make(0), make(1), make(2)])); - assert_eq!( - trace.ring.len(), - 3, - "a 3-message Batch must expand to 3 traced frames" - ); -} - -#[test] -fn writer_trace_records_the_raw_path_verbatim() { - let body = vec![0xAAu8, 0xBB, 0xCC, 0xDD, 0xEE]; - let mut trace = WriterTrace::new(); - trace.record(&Outbound::Raw { - opcode: 0x00EE, - body: body.clone(), - }); - let entry = trace.ring[0]; - assert_eq!(entry.opcode, 0x00EE); - assert_eq!(entry.size, 2 + body.len() as u16); - assert_eq!(entry.checksum, fnv1a64(&body)); -} - #[test] fn writer_trace_ring_caps_at_capacity_and_drops_oldest() { let mut trace = WriterTrace::new(); @@ -273,8 +193,7 @@ fn oversized_raw_body_ends_the_session_instead_of_wrapping_the_size_field() { .join() .expect("server handshake thread must not panic"); - let (tx, rx, depth) = session_channel(); - let writer = spawn_writer(s, encrypt, rx, depth, 1).expect("spawn the writer thread"); + let (tx, writer) = start_writer(s, encrypt); let oversized_body = vec![0xABu8; u16::MAX as usize]; tx.send(Outbound::Raw { @@ -286,7 +205,8 @@ fn oversized_raw_body_ends_the_session_instead_of_wrapping_the_size_field() { writer .join() - .expect("the writer thread must not panic on an oversized raw body"); + .expect("the writer thread must not panic on an oversized raw body") + .expect_err("oversized bodies must end the writer"); match ServerOpcodeMessage::read_encrypted(&mut client, &mut c_dec) { Err(_) => {} @@ -323,8 +243,7 @@ fn combat_cast_burst_over_a_real_cipher(n: usize) { .join() .expect("server handshake thread must not panic"); - let (tx, rx, depth) = session_channel(); - let writer = spawn_writer(s, encrypt, rx, depth, 1).expect("spawn the writer thread"); + let (tx, writer) = start_writer(s, encrypt); const SPELL_ID: u32 = 133; const BOSS_GUID: u64 = 0xF130_0000_0000_0001; @@ -364,6 +283,9 @@ fn combat_cast_burst_over_a_real_cipher(n: usize) { }); for i in 0..n { + while tx.depth() >= 128 { + std::thread::sleep(Duration::from_millis(1)); + } let caster = 0x1000_0000_0000_0000u64 + i as u64; let damage = 40 + (i as u32 % 60); @@ -431,7 +353,8 @@ fn combat_cast_burst_over_a_real_cipher(n: usize) { drop(tx); writer .join() - .expect("writer thread must not panic under a combat+cast burst"); + .expect("writer thread must not panic under a combat+cast burst") + .expect("combat frames must reach the client"); let (starts, goes, damage_logs, cooldowns, swings, heartbeats) = reader.join().expect("reader thread must not panic");