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..b6db0f358 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -311,58 +311,62 @@ 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 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: - **`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..2308de611 --- /dev/null +++ b/gateway/src/world/egress.rs @@ -0,0 +1,352 @@ +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); + 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); + } + 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/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/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/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 2816fc26e..1088066da 100644 --- a/gateway/src/world/mod.rs +++ b/gateway/src/world/mod.rs @@ -1,14 +1,15 @@ 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; 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 +26,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 +90,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,27 +98,22 @@ 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>, } -impl SessionTx { - 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}"); - } - })); - } +#[derive(Clone)] +enum SessionSender { + Async(egress::QueueSender), + #[cfg(test)] + Sync(Sender), +} +impl SessionTx { pub(crate) fn close(&self) { if let Some(close) = &self.close_socket { close(); @@ -124,12 +122,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 +144,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,30 +168,6 @@ impl SessionTx { } } -pub trait DuplexStream: Read + Write + IoDeadline + Send + Sized + 'static { - fn try_clone(&self) -> std::io::Result; - fn shutdown_both(&self) -> std::io::Result<()>; -} - -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, @@ -333,141 +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) -} - -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, - ))) -} - -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 { @@ -508,31 +350,6 @@ impl WriterTrace { }); } - 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(_) => {} - } - } - - 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"), @@ -569,265 +386,121 @@ impl WriterTrace { } } -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()) -} - -struct AdmissionSeat<'a> { - queue: &'a LoginQueue, +struct AddonLimit { + tokens: f32, + refill_at: Instant, + drop_logged_at: Option, } -impl Drop for AdmissionSeat<'_> { - fn drop(&mut self) { - self.queue.depart(); +impl Default for AddonLimit { + fn default() -> Self { + Self { + tokens: 20.0, + refill_at: Instant::now(), + drop_logged_at: None, + } } } -#[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) -} - -fn run_world_session_with_queue_and_deadline( - mut stream: S, - store: Arc, - queue: &LoginQueue, - deadline: &mut PreAuthDeadline, +fn process_frame( + tx: &SessionTx, + conn: &mut WorldConn, + addon: &mut AddonLimit, + hdr: wow_srp::vanilla_header::ClientHeader, + body: Vec, ) -> 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_tokens: f32 = 20.0; - let mut addon_refill_at = std::time::Instant::now(); - let mut addon_drop_logged_at: Option = None; - 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}"))?; - 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 == 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 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 !codec::addon::typed_addon_chat_type(chat_type) { + 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)?; - } - } - 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, + 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)?; } - 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())?; } - })(); - - 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); + return Ok(()); } - drop(tx); - let _ = writer.join(); - // Release the seat exactly once, after the writer and Relays stop. - drop(seat); - result + 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 { @@ -1171,6 +844,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 +868,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 +880,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 +891,21 @@ 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, + let actions = actions.clone(); + let encoders = encoders.clone(); + tokio::spawn(async move { + if let Err(error) = transport::run_at( + sock, Arc::new(coord), - &queue, - &mut deadline, - ) { - log::warn!("world session {peer} ended: {e:#}"); + queue, + actions, + encoders, + connection_permit, + deadline, + ) + .await + { + log::warn!("world session {peer} ended: {error:#}"); } }); } @@ -1248,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 75db18bda..37d81a565 100644 --- a/gateway/src/world/test_support/world_fake/session.rs +++ b/gateway/src/world/test_support/world_fake/session.rs @@ -13,6 +13,11 @@ pub(crate) struct SessionState { pub(crate) username: String, 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>, @@ -54,6 +59,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 @@ -112,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 new file mode 100644 index 000000000..20b0af5d3 --- /dev/null +++ b/gateway/src/world/transport.rs @@ -0,0 +1,779 @@ +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, + connection_permit: tokio::sync::OwnedSemaphorePermit, + 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(async move { + let _connection_permit = connection_permit; + run_session(stream, store, queue, actions, encoders, receiver, deadline).await + }) + .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 (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 = 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(); + } + result + }); + + let mut drain = true; + let result = async { + let mut reader = reader; + loop { + let frame = async { + let mut bytes = [0; 6]; + 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()) + .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 + .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(Some(frame)) => frame, + Ok(None) => break, + Err(error) => return Err(error), + } + }; + let conn = conn.clone(); + let addon = addon.clone(); + let action_tx = tx.clone(); + // Await accepted game work before cleanup so a late login cannot outlive its session. + 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>(()) + } + .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, + 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() { + 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 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 || { + 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()); + 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 || 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)); + } +} 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");