diff --git a/apps/figcraft-server/src/abuse.rs b/apps/figcraft-server/src/abuse.rs index 05cd848..8d3a892 100644 --- a/apps/figcraft-server/src/abuse.rs +++ b/apps/figcraft-server/src/abuse.rs @@ -44,6 +44,8 @@ pub enum Action { Connect, Publish, Message, + /// Presence (cursors, selections, viewports), cursor chat and pings: twice `messages_per_second`. + Presence, Report, /// Downloading this many bytes. Download(usize), @@ -53,6 +55,7 @@ struct Address { connects: Bucket, publishes: Bucket, messages: Bucket, + presence: Bucket, reports: Bucket, downloads: Bucket, used: Instant, @@ -79,7 +82,7 @@ impl Abuse { /// Whether `ip` may do `what` now (counting it, in every block it is in, when it may). pub fn allow(&mut self, ip: IpAddr, what: Action, now: Instant) -> bool { - if !self.cfg.open && matches!(what, Action::Connect | Action::Publish | Action::Message) { + if !self.cfg.open && matches!(what, Action::Connect | Action::Publish | Action::Message | Action::Presence) { return true; } let keys: Vec<(Block, f64)> = ip_blocks(ip).into_iter().map(|(k, times)| (k, f64::from(times))).collect(); @@ -108,6 +111,7 @@ impl Abuse { publishes: bucket(f64::from(c.publishes_per_hour) * times, 3600.0), // A burst of four seconds' worth (a big paste, a quick drag). messages: Bucket::new(f64::from(c.messages_per_second) * times, f64::from(c.messages_per_second) * times * 4.0), + presence: Bucket::new(f64::from(c.messages_per_second) * times * 2.0, f64::from(c.messages_per_second) * times * 8.0), reports: bucket(REPORTS_PER_HOUR * times, 3600.0), downloads: Bucket::new(DOWNLOAD_RATE * times, DOWNLOAD_BURST * times), used: now, @@ -121,6 +125,7 @@ impl Abuse { Action::Connect => (&mut a.connects, 1.0), Action::Publish => (&mut a.publishes, 1.0), Action::Message => (&mut a.messages, 1.0), + Action::Presence => (&mut a.presence, 1.0), Action::Report => (&mut a.reports, 1.0), Action::Download(n) => (&mut a.downloads, n as f64), } diff --git a/apps/figcraft-server/src/hardening_tests.rs b/apps/figcraft-server/src/hardening_tests.rs index 3aaab2b..641ec4a 100644 --- a/apps/figcraft-server/src/hardening_tests.rs +++ b/apps/figcraft-server/src/hardening_tests.rs @@ -660,3 +660,131 @@ fn problems_stay_bounded_and_a_full_disk_refuses_writes() { assert!(e.contains("disk is nearly full"), "{e}"); } } + +// Third review: R1, N3, R2 ------------------------------------------------------------------- + +/// Read messages until `pick` takes one (others in between are skipped). +fn wait_for(raw: &mut Raw, what: &str, mut pick: impl FnMut(ServerMsg) -> Option) -> T { + loop { + match raw.next() { + ServerMsg::Refused { reason } => panic!("refused while waiting for {what}: {reason}"), + m => { + if let Some(t) = pick(m) { + return t; + } + } + } + } +} + +#[test] +fn an_office_behind_one_address_works_together() { + let dir = TempDir::new("office"); + let cfg = Config { max_file_mb: Some(4), max_plugin_mb: 1, ..open(dir.path()) }; + let mut server = crate::start(&cfg).unwrap(); + let addr = server.addr(); + // A file near the size limit. + let mut d = Document::new("Office"); + let page = d.first_page().unwrap(); + d.create(page, NodeKind::Rectangle, &"o".repeat(3_000_000), 0.0, 0.0, 10.0, 10.0).unwrap(); + let (_, replies) = manage::hosting( + &Endpoint::plain(addr.to_string()), + "", + "", + vec![Admin::Publish { name: "Office".into(), snapshot: Replica::new(d).snapshot().unwrap() }], + &NetLimits::default(), + Duration::from_secs(10), + ) + .unwrap(); + let Some(AdminReply::Hosted { file, edit, .. }) = replies.into_iter().next() else { panic!("not hosted") }; + // Twenty colleagues, one address, all joining at once: twenty welcomes of 3 MB. + let mut people: Vec = (0..20).map(|_| Raw::connect(addr)).collect(); + for (i, p) in people.iter_mut().enumerate() { + p.send(&Client::hello(&file.id, &edit.token, &format!("Person {i}"), "")); + } + let mut clients: Vec = people + .iter_mut() + .map(|p| Client::join(wait_for(p, "the welcome", |m| matches!(m, ServerMsg::Welcome { .. }).then_some(m))).unwrap()) + .collect(); + // Everyone moves their cursor (twenty people at ~30 a second share the address's budget). + for p in people.iter_mut() { + for k in 0..60 { + p.send(&ClientMsg::Presence(figcraft_collab::Presence { cursor: Some((page, f64::from(k), 0.0)), ..Default::default() })); + } + } + // One pastes 1 MB: the broadcast reaches all twenty. + let mut doc = clients[0].document().clone(); + doc.create(page, NodeKind::Rectangle, &"p".repeat(1_000_000), 0.0, 0.0, 5.0, 5.0).unwrap(); + let batch = clients[0].edit(&doc).unwrap().unwrap(); + people[0].send(&batch); + for (i, p) in people.iter_mut().enumerate() { + let seq = wait_for(p, "the paste", |m| { + if let ServerMsg::Changes(a) = m { + a.changes.iter().any(|c| matches!(c, figcraft_collab::Change::Create { .. })).then_some(a.seq) + } else { + None + } + }); + assert!(seq > 0, "person {i}"); + } + // Still all connected. + for p in people.iter_mut() { + p.send(&ClientMsg::Ping); + wait_for(p, "the pong", |m| (m == ServerMsg::Pong).then_some(())); + } + assert_eq!(server.connections(), 20); + drop(clients.drain(..)); + server.stop(); +} + +#[test] +fn the_biggest_reader_is_cut_off_when_everyones_reading_budget_runs_out() { + count_connection_panics(); + let dir = TempDir::new("readers"); + let cfg = open(dir.path()); + let store = Store::open(dir.path(), cfg.store()).unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let limits = NetLimits { max_reading_block: 10 << 20, max_reading_total: 300_000, ..cfg.net() }; + let mut busy = Hosting::serve(listener, store, limits).unwrap(); + let addr = busy.addr(); + // A slow sender: it starts a 300 KB message, sends 250 KB of it and stops. + let mut hog = Raw::connect(addr); + hog.send(&hosting()); + assert!(matches!(hog.next(), ServerMsg::Hosting(_))); + let mut frame = vec![0x82, 0x80 | 127]; + frame.extend_from_slice(&300_000u64.to_be_bytes()); + frame.extend_from_slice(&[1, 2, 3, 4]); + frame.extend(std::iter::repeat_n(0x55, 250_000)); + hog.raw_write(&frame); + until(10, "the server to read the hog's bytes", || busy.reading_bytes() >= 250_000); + // Someone else sends a 100 KB message: the hog is cut off, not them. + let mut other = Raw::connect(addr); + other.send(&hosting()); + assert!(matches!(other.next(), ServerMsg::Hosting(_))); + let mut d = Document::new("Other"); + let page = d.first_page().unwrap(); + d.create(page, NodeKind::Rectangle, &"q".repeat(100_000), 0.0, 0.0, 1.0, 1.0).unwrap(); + other.send(&ClientMsg::Admin(Admin::Publish { name: "Other".into(), snapshot: Replica::new(d).snapshot().unwrap() })); + assert!(matches!(other.next(), ServerMsg::Admin(_)), "the message was read and answered"); + assert!(hog.closed_within(Duration::from_secs(10)), "the hog's connection was cut"); + until(10, "the hog's bytes to be given back", || busy.reading_bytes() == 0); + assert_eq!(CONNECTION_PANICS.load(std::sync::atomic::Ordering::SeqCst), 0); + busy.stop(); +} + +#[test] +fn the_server_forgets_addresses_whose_connections_are_gone() { + let dir = TempDir::new("forget"); + let cfg = Config { trusted_proxy: Some("127.0.0.1".into()), ..open(dir.path()) }; + let mut server = crate::start(&cfg).unwrap(); + let addr = server.addr(); + for n in 1..=10u8 { + let (ws, r) = via_proxy(addr, &format!("203.0.113.{n}")); + assert!(matches!(r, ServerMsg::Hosting(_))); + drop(ws); + } + until(10, "the connections to close", || server.connections() == 0); + let (_last, _) = via_proxy(addr, "203.0.113.200"); + assert!(server.blocks_remembered() <= 2, "{} blocks remembered", server.blocks_remembered()); + server.stop(); +} diff --git a/apps/figcraft-server/src/lib.rs b/apps/figcraft-server/src/lib.rs index 5a21ac5..a07f047 100644 --- a/apps/figcraft-server/src/lib.rs +++ b/apps/figcraft-server/src/lib.rs @@ -44,6 +44,11 @@ impl Running { self.hosting.queued_bytes() } + /// Blocks of addresses the server keeps counters for (see [`Hosting::blocks_remembered`]). + pub fn blocks_remembered(&self) -> usize { + self.hosting.blocks_remembered() + } + /// Stop (see the type's docs). pub fn stop(&mut self) { self.hosting.stop(); diff --git a/apps/figcraft-server/src/store.rs b/apps/figcraft-server/src/store.rs index e0d5172..b7f9a7c 100644 --- a/apps/figcraft-server/src/store.rs +++ b/apps/figcraft-server/src/store.rs @@ -36,8 +36,8 @@ use std::path::{Path, PathBuf}; use std::time::{Instant, SystemTime, UNIX_EPOCH}; use figcraft_collab::message::{ - Access, Admin, AdminReply, FileInfo, FileRow, FileUsage, HostInfo, LibraryReply, LibraryReq, LinkInfo, NewLink, PluginReply, PluginReq, - ReportInfo, Usage, + Access, Admin, AdminReply, ClientMsg, FileInfo, FileRow, FileUsage, HostInfo, LibraryReply, LibraryReq, LinkInfo, NewLink, PluginReply, + PluginReq, ReportInfo, Usage, }; use figcraft_collab::{Change, Replica, Server, same_secret}; use figcraft_collab_net::hub::{Admission, Backend, Closing, Handled, OpenFile, ip_key}; @@ -1219,8 +1219,11 @@ impl Backend for Store { Ok(()) } - fn message(&mut self, ip: IpAddr) -> bool { - self.abuse.allow(ip, Action::Message, Instant::now()) + fn message(&mut self, ip: IpAddr, msg: &ClientMsg) -> bool { + // Cursors and cursor chat (many people behind one address each send ~30 a second while + // they move) have a budget of their own, beside everything else. + let what = if matches!(msg, ClientMsg::Presence(_) | ClientMsg::Signal(_) | ClientMsg::Ping) { Action::Presence } else { Action::Message }; + self.abuse.allow(ip, what, Instant::now()) } fn hosting(&mut self, file: &str) -> Option { diff --git a/apps/figcraft-server/src/test_util.rs b/apps/figcraft-server/src/test_util.rs index a1b75c4..e0423e0 100644 --- a/apps/figcraft-server/src/test_util.rs +++ b/apps/figcraft-server/src/test_util.rs @@ -123,6 +123,30 @@ impl Raw { self.0.send(Message::Binary(Bytes::from(message::encode(m).unwrap()))).unwrap(); } + /// Write raw bytes on the socket (a hand-made, partial frame). + pub fn raw_write(&mut self, bytes: &[u8]) { + use std::io::Write as _; + self.0.get_mut().write_all(bytes).unwrap(); + } + + /// The server closed the connection within `limit` (anything it sends meanwhile is skipped). + pub fn closed_within(&mut self, limit: Duration) -> bool { + use std::io::Read as _; + let s = self.0.get_mut(); + let end = Instant::now() + limit; + let mut buf = [0u8; 4096]; + let _ = s.set_read_timeout(Some(Duration::from_millis(200))); + while Instant::now() < end { + match s.read(&mut buf) { + Ok(0) => return true, + Ok(_) => {} + Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {} + Err(_) => return true, + } + } + false + } + /// Send without waiting to be read (`Err` once the server has hung up). pub fn try_send(&mut self, m: &ClientMsg) -> tungstenite::Result<()> { self.0.send(Message::Binary(Bytes::from(message::encode(m).unwrap()))) diff --git a/crates/collab-net/src/hub.rs b/crates/collab-net/src/hub.rs index 74e4f42..45a0ac3 100644 --- a/crates/collab-net/src/hub.rs +++ b/crates/collab-net/src/hub.rs @@ -43,7 +43,7 @@ use tungstenite::{Bytes, Message}; use crate::inbox::Inbox; use crate::join::{ENDED, RESTARTING}; -use crate::net::{Bucket, Duplex, Meter, NetLimits, config, shutdown}; +use crate::net::{Bucket, Duplex, Meter, NetLimits, Readers, config, shutdown}; use crate::throttle::{PRESENCE_INTERVAL, Throttle, earliest}; /// How long a closing connection waits for the peer to hang up. @@ -178,7 +178,7 @@ pub trait Backend: Send + 'static { } /// A message from a connection from `ip`: false disconnects it (that address sends too much). - fn message(&mut self, _ip: IpAddr) -> bool { + fn message(&mut self, _ip: IpAddr, _msg: &ClientMsg) -> bool { true } @@ -216,32 +216,51 @@ enum Out { Close(CloseCode, String), } -/// One message's bytes held for one or more connections (a broadcast shares one buffer): counted -/// once against the server's budget, until the last copy is sent or dropped. +/// One message's bytes held for one or more connections (a broadcast shares one buffer), counted +/// once against a shared counter (the server's, or a block's) until the last copy is sent or +/// dropped. struct Held { n: usize, - all: Arc, + counter: Arc, +} + +impl Held { + fn new(n: usize, counter: &Arc) -> Arc { + counter.fetch_add(n, Ordering::Relaxed); + Arc::new(Held { n, counter: counter.clone() }) + } } impl Drop for Held { fn drop(&mut self) { - self.all.fetch_sub(self.n, Ordering::Relaxed); + self.counter.fetch_sub(self.n, Ordering::Relaxed); } } -/// Bytes waiting in a connection's queue, counted against its cap and its block's until dropped -/// (and, through [`Held`], once against the server's). +/// What one message counts against, made by its first recipient and shared by the others: the +/// server's budget once, and each recipient block's once (twenty people behind one office's +/// address cost their block one copy of a broadcast, not twenty). +#[derive(Default)] +struct Charges { + all: Option>, + blocks: Vec<(usize, Arc)>, + /// Count it against blocks at all: a welcome isn't (it is bounded by the file's size and each + /// connection's cap, and a whole office joining at once must fit). + skip_blocks: bool, +} + +/// Bytes waiting in a connection's queue, counted against its cap until dropped (and, through +/// [`Held`], once against the server's budget and once against its block's). struct Charge { n: usize, mine: Arc, - block: Arc, - _held: Arc, + _all: Arc, + _block: Option>, } impl Drop for Charge { fn drop(&mut self) { self.mine.fetch_sub(self.n, Ordering::Relaxed); - self.block.fetch_sub(self.n, Ordering::Relaxed); } } @@ -272,29 +291,40 @@ struct Line { impl Line { /// Queue `b`. One message always fits an empty queue (a welcome as large as the largest file - /// gets out). `held` is the message's count against the server's budget, made by its first - /// recipient and shared by the others; `over_budget` lets it past that budget (the hub just - /// dropped the biggest consumer to make room). - fn push(&self, b: &Bytes, held: &mut Option>, over_budget: bool) -> Result<(), Push> { + /// gets out). `charges` is what the message counts against, shared by its recipients; + /// `over_budget` lets it past the server's budget (the hub just dropped the biggest consumer + /// to make room). + fn push(&self, b: &Bytes, charges: &mut Charges, over_budget: bool) -> Result<(), Push> { let n = b.len(); let mine = self.queued.load(Ordering::Relaxed); - let block = self.block.load(Ordering::Relaxed); - if (mine > 0 && mine.saturating_add(n) > self.max) || (block > 0 && block.saturating_add(n) > self.max_block) { + if mine > 0 && mine.saturating_add(n) > self.max { return Err(Push::Full); } - let held = match held { + let key = Arc::as_ptr(&self.block) as usize; + let block = if charges.skip_blocks { + None + } else if let Some((_, h)) = charges.blocks.iter().find(|(k, _)| *k == key) { + Some(h.clone()) + } else { + let used = self.block.load(Ordering::Relaxed); + if used > 0 && used.saturating_add(n) > self.max_block { + return Err(Push::Full); + } + let h = Held::new(n, &self.block); + charges.blocks.push((key, h.clone())); + Some(h) + }; + let all = match &charges.all { Some(h) => h.clone(), None => { if !over_budget && self.all.load(Ordering::Relaxed).saturating_add(n) > self.max_all { return Err(Push::Everyone); } - self.all.fetch_add(n, Ordering::Relaxed); - held.insert(Arc::new(Held { n, all: self.all.clone() })).clone() + charges.all.insert(Held::new(n, &self.all)).clone() } }; self.queued.fetch_add(n, Ordering::Relaxed); - self.block.fetch_add(n, Ordering::Relaxed); - let charge = Charge { n, mine: self.queued.clone(), block: self.block.clone(), _held: held }; + let charge = Charge { n, mine: self.queued.clone(), _all: all, _block: block }; self.tx.try_send(Out::Bytes(b.clone(), Some(charge))).map_err(|e| match e { TrySendError::Full(_) => Push::Full, TrySendError::Disconnected(_) => Push::Gone, @@ -315,17 +345,18 @@ struct BlockUse { struct Blocks(Arc>>); impl Blocks { + /// Block `key`'s counters; blocks nobody uses any more (only the table holds their counters, + /// whose connections' queues and readers are all gone) are forgotten first, so the table + /// never holds more than the blocks of open connections. fn get(&self, key: IpAddr) -> BlockUse { - self.0.lock().unwrap_or_else(PoisonError::into_inner).entry(key).or_default().clone() + let mut t = self.0.lock().unwrap_or_else(PoisonError::into_inner); + t.retain(|_, u| Arc::strong_count(&u.queued) > 1 || Arc::strong_count(&u.reading) > 1); + t.entry(key).or_default().clone() } - /// A connection of block `key` ended (its `held` handles are dropped by now or soon). - fn leave(&self, key: IpAddr, held: BlockUse) { - let mut t = self.0.lock().unwrap_or_else(PoisonError::into_inner); - // The table's copy and this one: nobody else uses the block. - if Arc::strong_count(&held.queued) <= 2 && Arc::strong_count(&held.reading) <= 2 { - t.remove(&key); - } + /// Blocks remembered. + fn len(&self) -> usize { + self.0.lock().unwrap_or_else(PoisonError::into_inner).len() } } @@ -409,6 +440,8 @@ pub struct Hosting { /// Bytes queued for every connection together, and in flight to the hub. queued: Arc, in_flight: InFlight, + blocks: Blocks, + reading: Arc, hub_thread: Option>, accept_thread: Option>, } @@ -439,8 +472,11 @@ impl Hosting { in_flight: in_flight.clone(), blocks: Blocks::default(), reading: Arc::default(), + readers: Readers::default(), }; let queued = shared.queued.clone(); + let blocks = shared.blocks.clone(); + let reading = shared.reading.clone(); let accept_thread = { let (hub_tx, stop, open) = (hub_tx.clone(), stop.clone(), open.clone()); spawn("live-accept", move || accept(&listener, &hub_tx, &stop, &open, limits, shared)) @@ -454,7 +490,18 @@ impl Hosting { } }; log::info!("live session: listening on {addr}"); - Ok(Hosting { addr, hub: hub_tx, stop, open, queued, in_flight, hub_thread: Some(hub_thread), accept_thread: Some(accept_thread) }) + Ok(Hosting { + addr, + hub: hub_tx, + stop, + open, + queued, + in_flight, + blocks, + reading, + hub_thread: Some(hub_thread), + accept_thread: Some(accept_thread), + }) } /// Bytes waiting to be sent to clients, all connections together. @@ -462,6 +509,17 @@ impl Hosting { self.queued.load(Ordering::Relaxed) } + /// Blocks of addresses the server keeps counters for (those of open connections, and any + /// whose last connection just ended, until the next connection prunes them). + pub fn blocks_remembered(&self) -> usize { + self.blocks.len() + } + + /// Bytes being read into messages not whole yet, all connections together. + pub fn reading_bytes(&self) -> usize { + self.reading.load(Ordering::Relaxed) + } + /// Bytes of messages read and not handled by the hub yet. pub fn in_flight_bytes(&self) -> usize { self.in_flight.0.load(Ordering::Relaxed) @@ -676,8 +734,9 @@ struct Shared { queued: Arc, in_flight: InFlight, blocks: Blocks, - /// Bytes being read into messages, all connections together. + /// Bytes being read into messages, all connections together, and who is reading them. reading: Arc, + readers: Readers, } fn accept(listener: &TcpListener, hub: &Sender, stop: &AtomicBool, open: &Arc, limits: NetLimits, shared: Shared) { @@ -804,7 +863,15 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender, limits: NetLimit let counted = shared.per_address.enter_all(ip, limits.max_per_address, limits.max_connections); let wide = wide_block(ip); let block = shared.blocks.get(wide); - ws.get_mut().meter(Meter::new(block.reading.clone(), shared.reading.clone(), limits.max_reading_block, limits.max_reading_total)); + ws.get_mut().meter(Meter::new( + conn, + &sock, + &shared.readers, + block.reading.clone(), + shared.reading.clone(), + limits.max_reading_block, + limits.max_reading_total, + )); let (tx, rx) = sync_channel::(limits.queue.max(4)); let out = Line { tx, @@ -870,7 +937,8 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender, limits: NetLimit } } drop(counted); - shared.blocks.leave(wide, block); + // The block's counters go with the last handle on them (the table forgets it then). + drop(block); linger(&mut ws); drop(out); // The writer closes the socket once it has sent what's queued (a refusal, a close). @@ -1111,7 +1179,7 @@ impl Hub { let Some(c) = self.conns.get(&conn) else { return }; let ip = c.ip; // The operator (with the server key) isn't held to the limits on addresses. - if !matches!(c.kind, Kind::Managing) && !self.backend.message(ip) { + if !matches!(c.kind, Kind::Managing) && !self.backend.message(ip, &msg) { log::info!("event=refused ip={ip} reason=messages_per_address"); return self.kick(&[conn], "too many messages from your address: wait a minute and try again"); } @@ -1384,7 +1452,7 @@ impl Hub { return; } let mut drop = Vec::new(); - if !encode(msg).is_some_and(|b| deliver(&self.conns, conn, &b, &mut None, &mut drop)) { + if !encode(msg).is_some_and(|b| deliver(&self.conns, conn, &b, &mut Charges::default(), &mut drop)) { drop.push(conn); } self.drop_conns(drop); @@ -1487,8 +1555,9 @@ impl Hub { Outgoing::Others(x, m) => (room.server.peers().map(|p| p.client_id).filter(|id| *id != x).collect(), m), }; let mut bytes = None; - // One count against the server's budget for every copy of this message. - let mut held = None; + // One count against the server's budget, and each block's, for every copy of this + // message; welcomes aren't counted against blocks. + let mut held = Charges { skip_blocks: matches!(msg, ServerMsg::Welcome { .. }), ..Charges::default() }; for id in to { if let Some((local, inbox)) = room.local.as_ref().filter(|(l, _)| *l == id) { to_local(inbox, *local, &msg); @@ -1510,7 +1579,7 @@ impl Hub { /// gone). When the server's budget for queued bytes is used up, the connection holding the most /// queued bytes is shed (added to `shed`) to make room, unless that is `conn` itself: whoever /// hogs the budget goes, not the next client that happens to need a little of it. -fn deliver(conns: &HashMap, conn: u64, b: &Bytes, held: &mut Option>, shed: &mut Vec) -> bool { +fn deliver(conns: &HashMap, conn: u64, b: &Bytes, held: &mut Charges, shed: &mut Vec) -> bool { let Some(c) = conns.get(&conn) else { return true }; match c.out.push(b, held, false) { Ok(()) => true, diff --git a/crates/collab-net/src/net.rs b/crates/collab-net/src/net.rs index 7dd1fab..8c841e9 100644 --- a/crates/collab-net/src/net.rs +++ b/crates/collab-net/src/net.rs @@ -6,9 +6,10 @@ //! machines over the same socket; every write goes through one lock and writes whole frames, so //! frames from the two never interleave. +use std::collections::HashMap; use std::io::{self, Read, Write}; use std::net::{Shutdown, TcpStream}; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, PoisonError}; use std::time::{Duration, Instant}; @@ -56,45 +57,123 @@ impl Duplex { } } +/// Every reader that counts what it reads, with what it holds half-read and its socket: when +/// everyone's reading budget runs out, the reader holding the most is cut off. +#[derive(Clone, Default)] +pub struct Readers(Arc>>); + +/// One registered reader: what it holds half-read, its block's count, whether it was cut off, and +/// its socket. +struct Reader { + mine: Arc, + block: Arc, + shed: Arc, + sock: TcpStream, +} + +impl Readers { + /// Cut off the reader holding the most half-read bytes, unless that is `me` (then `false`: the + /// caller is the biggest). Its bytes are given back to the budget at once, it can't read any + /// more, and its socket is shut (where that wakes a blocked read, it ends now; elsewhere at its + /// read timeout). + fn shed_biggest(&self, me: u64, all: &AtomicUsize) -> bool { + let t = self.0.lock().unwrap_or_else(PoisonError::into_inner); + let Some((&id, r)) = t.iter().filter(|(_, r)| !r.shed.load(Ordering::Relaxed)).max_by_key(|(_, r)| r.mine.load(Ordering::Relaxed)) else { + return false; + }; + if id == me || r.mine.load(Ordering::Relaxed) == 0 { + return false; + } + log::info!("event=shed conn={id} reason=reading_bytes"); + r.shed.store(true, Ordering::Relaxed); + let held = r.mine.swap(0, Ordering::Relaxed); + r.block.fetch_sub(held, Ordering::Relaxed); + all.fetch_sub(held, Ordering::Relaxed); + let _ = r.sock.shutdown(Shutdown::Both); + true + } + + /// Readers registered. + pub fn len(&self) -> usize { + self.0.lock().unwrap_or_else(PoisonError::into_inner).len() + } + + /// None registered. + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + /// Bytes being read into messages (not whole yet), per connection, per block of addresses and -/// for every connection together: a read that takes a block over its share, or takes everyone -/// over the budget while its block holds a large part of it, fails, so the biggest readers are -/// dropped and memory for half-read messages stays bounded. +/// for every connection together. A read that takes its block over its share fails. One that takes +/// everyone over the budget cuts off the readers holding the most until it fits (their bytes are +/// given back at once, they can't read any more and their sockets are shut) or, when this reader is +/// the biggest, fails: the biggest readers go, not the next one. pub struct Meter { - mine: usize, + id: u64, + mine: Arc, + shed: Arc, block: Arc, all: Arc, max_block: usize, max_all: usize, + readers: Readers, } +/// What a reader cut off for using too much of the budget is told. +const BUSY_READING: &str = "the server is busy reading from others: try again in a moment"; + impl Meter { - pub fn new(block: Arc, all: Arc, max_block: usize, max_all: usize) -> Self { - Self { mine: 0, block, all, max_block, max_all } + /// Connection `id`'s meter (registered with `readers`, cut off through `sock`). + pub fn new( + id: u64, + sock: &TcpStream, + readers: &Readers, + block: Arc, + all: Arc, + max_block: usize, + max_all: usize, + ) -> Self { + let (mine, shed) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicBool::new(false))); + if let Ok(s) = sock.try_clone() { + let r = Reader { mine: mine.clone(), block: block.clone(), shed: shed.clone(), sock: s }; + readers.0.lock().unwrap_or_else(PoisonError::into_inner).insert(id, r); + } + Self { id, mine, shed, block, all, max_block, max_all, readers: readers.clone() } } fn add(&mut self, n: usize) -> io::Result<()> { - self.mine += n; + if self.shed.load(Ordering::Relaxed) { + return Err(io::Error::other(BUSY_READING)); + } + self.mine.fetch_add(n, Ordering::Relaxed); let block = self.block.fetch_add(n, Ordering::Relaxed).saturating_add(n); - let all = self.all.fetch_add(n, Ordering::Relaxed).saturating_add(n); - // Over its block's share; or over everyone's budget while its block holds a quarter of it - // (the big readers go first); or far over everyone's budget. - if block > self.max_block || (all > self.max_all && block.saturating_mul(4) >= self.max_all) || all > self.max_all.saturating_mul(2) { - return Err(io::Error::other("the server is busy reading from others: try again in a moment")); + let mut all = self.all.fetch_add(n, Ordering::Relaxed).saturating_add(n); + if block > self.max_block { + return Err(io::Error::other(BUSY_READING)); + } + // Over everyone's budget: the biggest readers are cut off until it fits, or this one goes + // when it is the biggest. + while all > self.max_all { + if !self.readers.shed_biggest(self.id, &self.all) { + return Err(io::Error::other(BUSY_READING)); + } + all = self.all.load(Ordering::Relaxed); } Ok(()) } fn release(&mut self) { - self.block.fetch_sub(self.mine, Ordering::Relaxed); - self.all.fetch_sub(self.mine, Ordering::Relaxed); - self.mine = 0; + let mine = self.mine.swap(0, Ordering::Relaxed); + self.block.fetch_sub(mine, Ordering::Relaxed); + self.all.fetch_sub(mine, Ordering::Relaxed); } } impl Drop for Meter { fn drop(&mut self) { self.release(); + self.readers.0.lock().unwrap_or_else(PoisonError::into_inner).remove(&self.id); } } diff --git a/crates/collab-net/src/queue_tests.rs b/crates/collab-net/src/queue_tests.rs index a52ae20..0d8468a 100644 --- a/crates/collab-net/src/queue_tests.rs +++ b/crates/collab-net/src/queue_tests.rs @@ -11,7 +11,7 @@ use std::sync::mpsc::{Receiver, sync_channel}; use tungstenite::Bytes; -use super::{Conn, Kind, Line, Out, Push, deliver}; +use super::{BlockUse, Blocks, Charges, Conn, Kind, Line, Out, Push, deliver}; /// A line with its own queue counter, the given block counter and the server's. fn line(block: &Arc, all: &Arc, max_block: usize, max_all: usize) -> (Line, Receiver) { @@ -29,7 +29,7 @@ fn a_broadcast_counts_once_against_the_server() { let all = Arc::new(AtomicUsize::new(0)); let lines: Vec<_> = (0..4).map(|_| line(&Arc::default(), &all, usize::MAX, 10_000)).collect(); let b = Bytes::from(vec![0u8; 3_000]); - let mut held = None; + let mut held = Charges::default(); for (l, _) in &lines { l.push(&b, &mut held, false).unwrap(); } @@ -51,9 +51,9 @@ fn one_block_gets_only_its_share() { let (b, _rb) = line(&block, &all, 1_000, usize::MAX); let (c, _rc) = line(&Arc::default(), &all, 1_000, usize::MAX); let msg = Bytes::from(vec![0u8; 800]); - a.push(&msg, &mut None, false).unwrap(); - assert_eq!(b.push(&msg, &mut None, false), Err(Push::Full), "the block's second connection is over its share"); - c.push(&msg, &mut None, false).unwrap(); + a.push(&msg, &mut Charges::default(), false).unwrap(); + assert_eq!(b.push(&msg, &mut Charges::default(), false), Err(Push::Full), "the block's second connection is over its share"); + c.push(&msg, &mut Charges::default(), false).unwrap(); } #[test] @@ -69,15 +69,54 @@ fn the_biggest_consumer_is_shed_when_the_budget_runs_out() { } // Connection 1 doesn't read: it holds most of the budget. let big = Bytes::from(vec![0u8; 9_000]); - assert!(deliver(&conns, 1, &big, &mut None, &mut Vec::new())); + assert!(deliver(&conns, 1, &big, &mut Charges::default(), &mut Vec::new())); // Connection 2 needs a little: the hog is shed and 2 still gets its message. let small = Bytes::from(vec![0u8; 2_000]); let mut shed = Vec::new(); - assert!(deliver(&conns, 2, &small, &mut None, &mut shed)); + assert!(deliver(&conns, 2, &small, &mut Charges::default(), &mut shed)); assert_eq!(shed, vec![1]); assert_eq!(conns[&2].out.queued.load(Ordering::Relaxed), 2_000); // When the receiver is itself the biggest, it is the one to go. let mut shed = Vec::new(); - assert!(!deliver(&conns, 1, &small, &mut None, &mut shed), "the hog asking for more is dropped"); + assert!(!deliver(&conns, 1, &small, &mut Charges::default(), &mut shed), "the hog asking for more is dropped"); assert!(shed.is_empty()); } + +#[test] +fn a_broadcast_counts_once_per_block_and_welcomes_not_at_all() { + let (office, all) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0))); + // Twenty people behind one office's address, a block share of 3 MB. + let lines: Vec<_> = (0..20).map(|_| line(&office, &all, 3_000_000, usize::MAX)).collect(); + // A 2 MB paste to all of them: one copy against the block, so everyone gets it. + let paste = Bytes::from(vec![0u8; 2_000_000]); + let mut charges = Charges::default(); + for (l, _) in &lines { + l.push(&paste, &mut charges, false).unwrap(); + } + assert_eq!(office.load(Ordering::Relaxed), 2_000_000); + drop(charges); + for (_, rx) in &lines { + drop(rx.try_recv().unwrap()); + } + assert_eq!(office.load(Ordering::Relaxed), 0); + // Twenty welcomes of a large file at once (each its own message): not counted against the block. + let welcome = Bytes::from(vec![0u8; 1_000_000]); + for (l, _) in &lines { + l.push(&welcome, &mut Charges { skip_blocks: true, ..Charges::default() }, false).unwrap(); + } + assert_eq!(office.load(Ordering::Relaxed), 0); + assert_eq!(all.load(Ordering::Relaxed), 20_000_000, "they still count against the server's budget"); +} + +#[test] +fn blocks_nobody_uses_are_forgotten() { + let blocks = Blocks::default(); + let a: IpAddr = "203.0.113.1".parse().unwrap(); + let held: BlockUse = blocks.get(a); + let _b = blocks.get("203.0.113.2".parse().unwrap()); + assert_eq!(blocks.len(), 2); + drop(held); + // The next connection prunes the block whose connections are all gone. + let _c = blocks.get("203.0.113.3".parse().unwrap()); + assert_eq!(blocks.len(), 2, "203.0.113.1 went"); +} diff --git a/docs/server.md b/docs/server.md index 2093a42..7ff2f41 100644 --- a/docs/server.md +++ b/docs/server.md @@ -388,11 +388,11 @@ are larger. | Message size | A file's size and 1 MiB (`max_file_mb`, 16 by default when open, or `max_plugin_mb` if larger): checked as the message is read, before anything is decoded; 64 KiB on a connection that only browses plugins (a client sending more is disconnected with the limit as the reason; the app keeps its own copy). The core's 64 MiB otherwise | | A file as it grows | Every batch is checked first (*always*): past `max_file_mb`, past its publisher's `max_mb_per_ip` or past `max_total_mb`, it is refused with the reason and the app rolls it back. A long log is compacted to make room at most once a minute per file (a compaction rewrites the whole file). A file's snapshot also holds the recent batches kept for reconnecting clients, at most a quarter of `max_file_mb` | | Recent batches in memory | A quarter of `max_file_mb` (256 KiB at least) per open file; older ones are dropped (a client that missed them catches up with the whole file) | -| Bytes queued for slow clients | The largest message and 1 MiB per connection, 32 MiB per block of addresses, 256 MiB for everyone (a message sent to many people counts once): a client that doesn't read its answers is dropped; when everyone's budget runs out, the connection holding the most is dropped | -| Bytes being read | Messages not yet whole: the largest message and 1 MiB per block, 192 MiB for everyone; past them the biggest readers are dropped as busy | +| Bytes queued for slow clients | The largest message and 1 MiB per connection, 32 MiB per block of addresses, 256 MiB for everyone. A message sent to many people counts once for everyone and once per block (twenty colleagues behind one office address cost their block one copy of a broadcast), and welcomes (bounded by the file's size) don't count against blocks, so a whole office can join a file at once. A client that doesn't read its answers is dropped; when everyone's budget runs out, the connection holding the most is dropped | +| Bytes being read | Messages not yet whole: the largest message and 1 MiB per block, 192 MiB for everyone; past everyone's budget the readers holding the most are cut off until it fits (their bytes count as free at once), and a reader is refused only when it is the biggest | | Bytes waiting for the server | 128 MiB of read messages not handled yet: past it a connection is dropped as busy | | Connections | `max_connections` (512) in all (*always*); `max_connections_per_ip` (32) per address, ×2 per IPv6 /56 (at most a sixteenth of all), ×4 per /48 (at most an eighth of all), joins included, so no block can take every connection; one-shot requests (asking about hosting, libraries, browsing plugins) can't take the last eighth, which stays for people joining files, and are dropped after 20 idle seconds | -| Rates | Per address and block: `connects_per_minute`, `publishes_per_hour` (and 32 addresses' worth for everyone), `messages_per_second`; reports, ten an hour (*always*); downloads of libraries and plugins, 1 MiB a second with bursts of 16 MiB per address (*always*), checked before anything is read, within 32 MiB a second (512 MiB at once) for libraries and for plugins, everyone together | +| Rates | Per address and block: `connects_per_minute`, `publishes_per_hour` (and 32 addresses' worth for everyone), `messages_per_second` (cursors, selections, cursor chat and pings have their own budget beside it, twice that, so an office moving its cursors together doesn't trip it); reports, ten an hour (*always*); downloads of libraries and plugins, 1 MiB a second with bursts of 16 MiB per address (*always*), checked before anything is read, within 32 MiB a second (512 MiB at once) for libraries and for plugins, everyone together | | Team libraries | Only files published with the key hold one on a public server; an archive is at most `max_file_mb` and everything it unpacks to counts against the same cap (*always*), so a zip bomb is refused before it expands | | Usage reports | Kept in memory, written at most once a minute per library and when the server stops; at most 1,000 components per file and 200 files per library (*always*) | | Disk writes | A batch is confirmed only once it is on disk; when its log can't be written, nobody hears it was accepted: the file closes, everyone in it reconnects and their apps send again what wasn't confirmed. Below 1 GiB (or two of the largest files) free, changes and new files are refused with "the server's disk is nearly full" instead (*always*, where the free space can be told: Unix). The server keeps the last 200 problems for the operator | @@ -410,12 +410,20 @@ open_publish = true operator_contact = "abuse@example.com" terms_url = "https://example.com/figcraft-terms" idle_days = 30 -max_file_mb = 16 -max_total_mb = 10240 +max_file_mb = 8 +max_plugin_mb = 8 +max_total_mb = 6144 max_files_per_ip = 20 max_mb_per_ip = 256 +messages_per_second = 600 +connects_per_minute = 60 ``` +These are the recommended values for a small public server (2 GB of memory, a 25-50 GB disk with +the nightly backups on it): files up to 8 MiB keep a whole office's welcomes within the queues, +6 GiB of files leaves room for a week of backups, and the per-address rates leave room for an +office behind one address. + ### On a fresh Ubuntu server with install.sh [`packaging/server/install.sh`](../packaging/server/install.sh) sets up everything below on a new @@ -429,7 +437,9 @@ installs the systemd units and its Caddy site (`/etc/caddy/figcraft-server.caddy the main Caddyfile, whose other sites it leaves alone), opens the SSH ports `sshd` listens on, 80 and 443 in `ufw`, and turns `ufw` on unless other services listen on the machine (then it says so: allow their ports and enable it yourself, or pass `--firewall-anyway`). Builds run as the -unprivileged `figcraft-build` user, and the binary is only ever run as `figcraft` (never as root); +unprivileged `figcraft-build` user on a copy of the source (root installs nothing from that copy; +the swap file and the built binary, copied out with the build user's rights, live in the root-only +`/var/cache/figcraft-install`), and the binary is only ever run as `figcraft` (never as root); a fresh clone builds the latest release tag. A Caddyfile with other sites is never replaced: it gains one `import figcraft-server.caddy` line, once (`packaging/server/install-test.sh` checks this). It checks new settings before they replace anything, starts it all and prints what to do next. Running it again diff --git a/packaging/server/install.sh b/packaging/server/install.sh index 1dfa5ed..11c489a 100755 --- a/packaging/server/install.sh +++ b/packaging/server/install.sh @@ -34,6 +34,9 @@ DATA=/var/lib/figcraft-server BIN=/usr/local/bin/figcraft-server BUILD=/var/cache/figcraft-build BUILDER=figcraft-build +# Root's own work (downloads, its clone, the swap file, the built binary once copied out): never +# a place the build user can touch. +WORK=/var/cache/figcraft-install SERVICE=figcraft-server HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" # The source tree this script sits in (packaging/server/ of a copy or checkout), if any. @@ -129,6 +132,7 @@ export DEBIAN_FRONTEND=noninteractive say "Packages" apt-get update -qq apt-get install -y -qq curl ca-certificates tar caddy ufw >/dev/null +install -d -o root -g root -m 700 "$WORK" ARCH="$(uname -m)" case "$ARCH" in @@ -147,7 +151,7 @@ from_release() { # download and check the release archive; sets NEW_BIN and FILE [ -n "$version" ] || return 1 fi name="figcraft-server-$version-linux-$ARCH" - dir="$BUILD/release" + dir="$WORK/release" rm -rf "$dir" mkdir -p "$dir" say "Downloading $name" @@ -176,7 +180,7 @@ as_builder() { # run a command as the unprivileged build user, with its own tool } from_source() { # build the source tree around this script, else a fresh clone; sets NEW_BIN and FILES - local mem_kb src ref="$REF" + local mem_kb src ref="$REF" copy="$BUILD/src" apt-get install -y -qq build-essential git pkg-config >/dev/null id -u "$BUILDER" >/dev/null 2>&1 || useradd --system --home-dir "$BUILD" --shell /usr/sbin/nologin "$BUILDER" install -d -o "$BUILDER" -g "$BUILDER" -m 755 "$BUILD" @@ -185,34 +189,35 @@ from_source() { # build the source tree around this script, else a fresh clone; else as_builder "$BUILD/cargo/bin/rustup" update stable >/dev/null fi + # The source root installs files from is root's own (the copy beside this script, or root's + # clone); the build user only ever builds a copy of it. if [ -n "$LOCAL_SRC" ]; then - # A copy the build user owns (the original may sit in root's home). - src="$BUILD/src-local" + src="$LOCAL_SRC" say "Building the source in $LOCAL_SRC: this takes a while" - rm -rf "$src" - install -d -o "$BUILDER" -g "$BUILDER" -m 755 "$src" - tar -C "$LOCAL_SRC" --exclude=./target --exclude=./plugins/target -cf - . | as_builder tar -C "$src" -xf - else - src="$BUILD/src" + src="$WORK/src" if [ -d "$src/.git" ]; then - as_builder git -C "$src" fetch --quiet --tags origin + git -C "$src" fetch --quiet --tags origin else - as_builder git clone --quiet "https://github.com/$REPO" "$src" || + git clone --quiet "https://github.com/$REPO" "$src" || die "can't clone $REPO (a private repository: copy the source to this machine and run its packaging/server/install.sh)" fi if [ -z "$ref" ]; then - if [ -n "$VERSION" ]; then ref="v$VERSION"; else ref="$(as_builder git -C "$src" tag --list 'v*' --sort=-v:refname | head -n 1)"; fi + if [ -n "$VERSION" ]; then ref="v$VERSION"; else ref="$(git -C "$src" tag --list 'v*' --sort=-v:refname | head -n 1)"; fi [ -n "$ref" ] || ref=main fi say "Building $ref from a clone: this takes a while" # A branch as it is upstream now, or a tag. - as_builder git -C "$src" checkout --quiet --force --detach "origin/$ref" 2>/dev/null || - as_builder git -C "$src" checkout --quiet --force --detach "$ref" + git -C "$src" checkout --quiet --force --detach "origin/$ref" 2>/dev/null || git -C "$src" checkout --quiet --force --detach "$ref" fi + # The build user clears and fills its own copy (root never touches paths it controls). + as_builder rm -rf "$copy" + as_builder mkdir -p "$copy" + tar -C "$src" --exclude=./target --exclude=./plugins/target --exclude=./.git -cf - . | as_builder tar -C "$copy" -xf - # The compiler needs about 2 GB: add a temporary swap file on smaller machines. mem_kb="$(awk '/^MemTotal:/ {print $2}' /proc/meminfo)" if [ "$mem_kb" -lt 2000000 ] && [ -z "$(swapon --show --noheadings)" ]; then - SWAP="$BUILD/swapfile" + SWAP="$WORK/swapfile" say "Adding a temporary 2 GB swap file for the build" fallocate -l 2G "$SWAP" 2>/dev/null || dd if=/dev/zero of="$SWAP" bs=1M count=2048 status=none chmod 600 "$SWAP" @@ -220,10 +225,15 @@ from_source() { # build the source tree around this script, else a fresh clone; swapon "$SWAP" export CARGO_BUILD_JOBS=1 fi - (cd "$src" && as_builder "$BUILD/cargo/bin/cargo" build --release --locked -p figcraft-server) + (cd "$copy" && as_builder "$BUILD/cargo/bin/cargo" build --release --locked -p figcraft-server) cleanup SWAP="" - NEW_BIN="$BUILD/target/release/figcraft-server" + # Copied out with the build user's rights into a file root made: a link the build user planted + # can't make root read (or install) anything else. + NEW_BIN="$WORK/figcraft-server.built" + rm -f "$NEW_BIN" + as_builder cat "$BUILD/target/release/figcraft-server" >"$NEW_BIN" || die "the build left no binary" + chmod 755 "$NEW_BIN" FILES="$src/packaging/server" } @@ -273,11 +283,15 @@ if [ ! -f "$CONFIG" ]; then printf 'terms_url = "%s"\n' "$TERMS" printf '# Files published without the key are deleted after this many days unopened.\n' printf 'idle_days = 30\n' - printf 'max_file_mb = 16\n' - printf '# All files together, and per address (see docs/server.md for every limit).\n' - printf 'max_total_mb = 10240\n' + printf 'max_file_mb = 8\n' + printf 'max_plugin_mb = 8\n' + printf '# All files together (leave room for the nightly backups on the same disk), and per address.\n' + printf 'max_total_mb = 6144\n' printf 'max_files_per_ip = 20\n' printf 'max_mb_per_ip = 256\n' + printf '# Per address: an office behind one address shares these (see docs/server.md for every limit).\n' + printf 'messages_per_second = 600\n' + printf 'connects_per_minute = 60\n' } >"$CONFIG.new" chmod 644 "$CONFIG.new" # Checked before it replaces anything.