M9: open hosting: fair limits for shared addresses
Third review: - a broadcast counts once per recipient block, welcomes don't count against block caps, and presence, cursor chat and pings have their own per-address budget, so an office behind one address works together - when everyone's reading budget runs out, the biggest readers are cut off until it fits (their bytes freed at once), no cliff - block counters are forgotten once only the table holds them - install.sh: root's downloads, clone, swap file and built binary in a root-only directory; the build user builds its own copy; the binary is copied out with the build user's rights - recommended open-mode settings in install.sh's config and the docs
This commit is contained in:
bflatastic
BFlat
parent
e0435d4eb4
commit
98c4753118
|
|
@ -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),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<T>(raw: &mut Raw, what: &str, mut pick: impl FnMut(ServerMsg) -> Option<T>) -> 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<Raw> = (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<Client> = 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();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<HostInfo> {
|
||||
|
|
|
|||
|
|
@ -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())))
|
||||
|
|
|
|||
|
|
@ -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<AtomicUsize>,
|
||||
counter: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
impl Held {
|
||||
fn new(n: usize, counter: &Arc<AtomicUsize>) -> Arc<Held> {
|
||||
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<Arc<Held>>,
|
||||
blocks: Vec<(usize, Arc<Held>)>,
|
||||
/// 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<AtomicUsize>,
|
||||
block: Arc<AtomicUsize>,
|
||||
_held: Arc<Held>,
|
||||
_all: Arc<Held>,
|
||||
_block: Option<Arc<Held>>,
|
||||
}
|
||||
|
||||
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<Arc<Held>>, 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<Mutex<HashMap<IpAddr, BlockUse>>>);
|
||||
|
||||
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<AtomicUsize>,
|
||||
in_flight: InFlight,
|
||||
blocks: Blocks,
|
||||
reading: Arc<AtomicUsize>,
|
||||
hub_thread: Option<JoinHandle<()>>,
|
||||
accept_thread: Option<JoinHandle<()>>,
|
||||
}
|
||||
|
|
@ -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<AtomicUsize>,
|
||||
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<AtomicUsize>,
|
||||
readers: Readers,
|
||||
}
|
||||
|
||||
fn accept(listener: &TcpListener, hub: &Sender<HubIn>, stop: &AtomicBool, open: &Arc<AtomicUsize>, limits: NetLimits, shared: Shared) {
|
||||
|
|
@ -804,7 +863,15 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, 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::<Out>(limits.queue.max(4));
|
||||
let out = Line {
|
||||
tx,
|
||||
|
|
@ -870,7 +937,8 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, 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<B: Backend> Hub<B> {
|
|||
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<B: Backend> Hub<B> {
|
|||
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<B: Backend> Hub<B> {
|
|||
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<B: Backend> Hub<B> {
|
|||
/// 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<u64, Conn>, conn: u64, b: &Bytes, held: &mut Option<Arc<Held>>, shed: &mut Vec<u64>) -> bool {
|
||||
fn deliver(conns: &HashMap<u64, Conn>, conn: u64, b: &Bytes, held: &mut Charges, shed: &mut Vec<u64>) -> bool {
|
||||
let Some(c) = conns.get(&conn) else { return true };
|
||||
match c.out.push(b, held, false) {
|
||||
Ok(()) => true,
|
||||
|
|
|
|||
|
|
@ -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<Mutex<HashMap<u64, Reader>>>);
|
||||
|
||||
/// One registered reader: what it holds half-read, its block's count, whether it was cut off, and
|
||||
/// its socket.
|
||||
struct Reader {
|
||||
mine: Arc<AtomicUsize>,
|
||||
block: Arc<AtomicUsize>,
|
||||
shed: Arc<AtomicBool>,
|
||||
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<AtomicUsize>,
|
||||
shed: Arc<AtomicBool>,
|
||||
block: Arc<AtomicUsize>,
|
||||
all: Arc<AtomicUsize>,
|
||||
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<AtomicUsize>, all: Arc<AtomicUsize>, 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<AtomicUsize>,
|
||||
all: Arc<AtomicUsize>,
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<AtomicUsize>, all: &Arc<AtomicUsize>, max_block: usize, max_all: usize) -> (Line, Receiver<Out>) {
|
||||
|
|
@ -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");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
Loading…
Reference in New Issue