//! The client side: the confirmed server state, this client's unconfirmed batches on top of it, //! the optimistic view the user edits, the user's own undo history, and the queue of batches made //! while the connection is down. use std::collections::{BTreeMap, BTreeSet, VecDeque}; use figcraft_doc::{Document, NodeId}; use crate::change::{Change, Prop}; use crate::history::{BACK, BatchKind, FWD, Notice, Step, Touches}; use crate::message::{Accepted, Access, CatchUp, ClientMsg, Hello, PROTOCOL, Peer, Presence, Processed, Resume, ServerMsg}; use crate::replica::{Mode, Replica}; use crate::{Diff, Error, NoRecompute, Recompute, diff}; /// The most undo steps a client keeps by default. pub const HISTORY_LIMIT: usize = 1000; /// A batch not yet confirmed or rejected: sent, or waiting to be sent. #[derive(Clone, Debug)] struct Pending { batch_id: u64, /// The server state it was made on (kept when it is sent again after a reconnect). base_seq: u64, changes: Vec, kind: BatchKind, /// The undo step this batch applies (its edit, undo or redo). step: Option, /// Someone else's batch was ordered before this one: its derived data may be stale. rebased: bool, /// Sent on the current connection. Batches go out in order; one waits (with every batch /// after it) while the client is offline, and an undo or redo waits until every earlier batch /// is confirmed or refused: it is then made again from the confirmed state (what it shows /// meanwhile is a preview), so it never rests on an edit the server refused or on stale /// values. sent: bool, } /// One participant's sync state. #[derive(Clone, Debug)] pub struct Client { /// The session joined (see [`ServerMsg::Welcome`]). session: u64, id: u32, /// The server's state as of `seq`. confirmed: Replica, seq: u64, pending: VecDeque, /// `confirmed` with the pending batches applied: what the user sees and edits. view: Replica, next_batch: u64, /// Ids below this were handed out; rebuilding the view never hands them out again. next_id: NodeId, peers: BTreeMap, presence: BTreeMap, /// Connected (batches go out); otherwise they wait for [`reconnect`](Self::reconnect). online: bool, undo: Vec, redo: Vec, next_step: u64, history_limit: usize, notices: Vec, /// What the server lets this client do (a viewer's edits are refused). access: Access, } /// The step with id `id` in either stack. fn find_step<'a>(undo: &'a mut [Step], redo: &'a mut [Step], id: u64) -> Option<&'a mut Step> { undo.iter_mut().chain(redo.iter_mut()).find(|s| s.id == id) } impl Client { /// The first message of a connection. pub fn hello(file: &str, token: &str, name: &str, color: &str) -> ClientMsg { ClientMsg::Hello(Hello { protocol: PROTOCOL, file: file.to_string(), token: token.to_string(), name: name.to_string(), color: color.to_string(), client_version: crate::message::client_version(), resume: None, }) } /// Start from the server's `Welcome` (which must carry a snapshot). pub fn join(welcome: ServerMsg) -> Result { let ServerMsg::Welcome { session, client_id, seq, catch_up: CatchUp::Snapshot(snapshot), processed, peers, access } = welcome else { return Err(Error::Protocol("expected a welcome with the document".into())); }; let mut confirmed = Replica::from_snapshot(&snapshot)?; confirmed.document_mut().set_client_id(client_id); let next_id = confirmed.document().next_id(); Ok(Client { session, id: client_id, view: confirmed.clone(), confirmed, seq, pending: VecDeque::new(), // After the batches the server already processed for this id (a client that starts // over under the id it had, e.g. after its app crashed: lower ids would be ignored). next_batch: processed.last_batch.saturating_add(1), next_id, peers: peers.into_iter().map(|p| (p.client_id, p)).collect(), presence: BTreeMap::new(), online: true, undo: Vec::new(), redo: Vec::new(), next_step: 1, history_limit: HISTORY_LIMIT, notices: Vec::new(), access, }) } /// This client's id (its node ids are `id << 32 | counter`). pub fn id(&self) -> u32 { self.id } /// What the server lets this client do (from the latest welcome). pub fn access(&self) -> Access { self.access } /// The document as this user sees it (server state plus their unconfirmed edits). Edit a copy /// (it allocates ids in this client's space) and hand it to [`edit`](Self::edit). pub fn document(&self) -> &Document { self.view.document() } /// The optimistic state (document and positions). pub fn view(&self) -> &Replica { &self.view } /// The last confirmed server state. pub fn confirmed(&self) -> &Replica { &self.confirmed } /// The sequence number of the last server batch applied. pub fn seq(&self) -> u64 { self.seq } /// Batches not yet confirmed or rejected (sent or waiting). pub fn pending(&self) -> usize { self.pending.len() } /// What this client's unconfirmed batch `batch_id` is for. pub fn batch_kind(&self, batch_id: u64) -> Option { self.pending.iter().find(|p| p.batch_id == batch_id).map(|p| p.kind) } /// Connected: batches go out as they are made. pub fn is_online(&self) -> bool { self.online } /// The other people in the session. pub fn peers(&self) -> impl Iterator { self.peers.values() } /// The last presence each peer sent. pub fn presence_of(&self, id: u32) -> Option<&Presence> { self.presence.get(&id) } /// What happened that the user should hear about since the last call (rejected edits…). pub fn take_notices(&mut self) -> Vec { std::mem::take(&mut self.notices) } /// The most undo steps kept (at least one). pub fn set_history_limit(&mut self, limit: usize) { self.history_limit = limit.max(1); self.trim_history(); } /// Undo steps, oldest first, by label. pub fn undo_labels(&self) -> impl DoubleEndedIterator { self.undo.iter().map(|s| s.label.as_str()) } /// Redo steps, the next one last, by label. pub fn redo_labels(&self) -> impl DoubleEndedIterator { self.redo.iter().map(|s| s.label.as_str()) } /// The user edited the document (`doc` is [`document`](Self::document) after a command): show /// it at once and return the batch to send, if anything synced changed and it can go now /// (otherwise it is queued: see [`Pending::sent`]). The edit is one undo step. pub fn edit(&mut self, doc: &Document) -> Result, Error> { self.edit_with(doc, Some("")) } /// [`edit`](Self::edit) with the undo step's label (`None`: not undoable, e.g. a change no /// command made; the redo steps stay). pub fn edit_with(&mut self, doc: &Document, undo_label: Option<&str>) -> Result, Error> { let d = diff(&self.view, doc)?; if d.changes.is_empty() { return Ok(None); } let step = match undo_label { Some(label) => { let id = self.next_step; self.next_step += 1; Some(Step::new(id, label, self.next_batch, &self.view, &d.replica, &d.changes)?) } None => None, }; self.view = d.replica; self.keep_ids(); let step_id = step.as_ref().map(|s| s.id); if let Some(s) = step { self.undo.push(s); self.redo.clear(); self.trim_history(); } self.queue(d.changes, BatchKind::Edit, step_id); // Everything queued before this batch is waiting (or it would have gone already), so at // most this edit goes now, and no undo becomes ready to go. Ok(self.flush(&NoRecompute).pop()) } /// Undo the user's last step that still changes something: write back the values it replaced /// where nobody else wrote since (their later writes win), delete what it created (unless /// others put layers inside) and bring back what it deleted. Returns the batches to send now /// (the undo, then any derived data `recompute` brings up to date as a follow-up); none when /// there is nothing to undo, or when they have to wait (offline, or the step's own edit is /// still unconfirmed). pub fn undo(&mut self, recompute: &dyn Recompute) -> Result, Error> { self.travel(BACK, recompute) } /// Redo the last undone step (the inverse of [`undo`](Self::undo)). pub fn redo(&mut self, recompute: &dyn Recompute) -> Result, Error> { self.travel(FWD, recompute) } /// How many undo and redo steps there are. pub fn history_len(&self) -> (usize, usize) { (self.undo.len(), self.redo.len()) } /// Whether [`undo`](Self::undo) has a step to try. pub fn can_undo(&self) -> bool { !self.undo.is_empty() } /// Whether [`redo`](Self::redo) has a step to try. pub fn can_redo(&self) -> bool { !self.redo.is_empty() } fn travel(&mut self, side: usize, recompute: &dyn Recompute) -> Result, Error> { loop { let popped = if side == BACK { self.undo.pop() } else { self.redo.pop() }; let Some(mut step) = popped else { return Ok(Vec::new()) }; let changes = match step.toward(side, &self.view) { Ok(c) => c, Err(e) => { // Put it back: nothing was done. if side == BACK { self.undo.push(step) } else { self.redo.push(step) } return Err(e); } }; if changes.is_empty() { // Everything it changed was changed again by others (or is gone): skip it. continue; } let before = self.view.clone(); let applied = self.view.apply(&changes, Mode::Lenient)?; self.keep_ids(); step.batch = self.next_batch; let kind = if side == BACK { BatchKind::Undo } else { BatchKind::Redo }; self.queue(changes, kind, Some(step.id)); // The derived data that follows, as its own batch (possibly empty: it is made again // with the undo when that goes out). let follow = self.recomputed(&self.view, before.document(), &applied.touched, recompute)?; if !follow.changes.is_empty() { self.view = follow.replica; self.keep_ids(); } self.queue(follow.changes, BatchKind::FollowUp, Some(step.id)); if side == BACK { self.redo.push(step) } else { self.undo.push(step) } return Ok(self.flush(recompute)); } } fn trim_history(&mut self) { let over = self.undo.len().saturating_sub(self.history_limit); self.undo.drain(..over); } /// Queue `changes` as a new batch (already shown in the view). fn queue(&mut self, changes: Vec, kind: BatchKind, step: Option) { let batch_id = self.next_batch; // `next_batch` starts from the welcome's `last_batch`, the host's word: saturate. self.next_batch = self.next_batch.saturating_add(1); self.pending.push_back(Pending { batch_id, base_seq: self.seq, changes, kind, step, rebased: false, sent: false }); } /// The waiting batches that can go now, in order (see [`Pending::sent`]). fn flush(&mut self, recompute: &dyn Recompute) -> Vec { let mut out = Vec::new(); if !self.online { return out; } let mut i = 0; while let Some(p) = self.pending.get(i) { if p.sent { i += 1; continue; } if matches!(p.kind, BatchKind::Undo | BatchKind::Redo) { if i > 0 { break; } if !self.remake_front(recompute) { // Nothing of it is left to do: it's gone; look at what came after it. continue; } } let Some(p) = self.pending.get_mut(i) else { break }; if p.changes.is_empty() { // A follow-up with nothing to bring up to date. self.pending.remove(i); continue; } p.sent = true; out.push(ClientMsg::Changes { base_seq: p.base_seq, batch_id: p.batch_id, changes: p.changes.clone() }); i += 1; } out } /// The undo or redo at the front of the queue goes out now: make it again from the confirmed /// state (others' writes since it was previewed are left alone, nodes that no longer fit are /// skipped), with its follow-up. False when nothing of it is left (it and its step are /// dropped). fn remake_front(&mut self, recompute: &dyn Recompute) -> bool { let Some((batch, step_id, side)) = self.pending.front().map(|p| (p.batch_id, p.step, if p.kind == BatchKind::Undo { BACK } else { FWD })) else { return true; }; let Some(step) = step_id.and_then(|id| find_step(&mut self.undo, &mut self.redo, id)).filter(|s| s.batch == batch) else { // Its step left the history (the size limit): send it as previewed. return true; }; let Ok(changes) = step.toward(side, &self.confirmed) else { return true }; if changes.is_empty() { let id = step.id; self.pending.pop_front(); self.undo.retain(|s| s.id != id); self.redo.retain(|s| s.id != id); self.pending.retain(|q| q.step != Some(id) || q.sent); self.rebuild(); return false; } step.send(&changes); let mut after = self.confirmed.clone(); let follow = after .apply(&changes, Mode::Lenient) .and_then(|a| self.recomputed(&after, self.confirmed.document(), &a.touched, recompute)) .map(|d| d.changes) .unwrap_or_default(); let seq = self.seq; if let Some(p) = self.pending.front_mut() { p.changes = changes; p.base_seq = seq; } if let Some(f) = self.pending.get_mut(1).filter(|q| q.kind == BatchKind::FollowUp && q.step == step_id && !q.sent) { f.changes = follow; f.base_seq = seq; } self.rebuild(); true } /// Keep allocating in this client's space, after every id already handed out. fn keep_ids(&mut self) { let doc = self.view.document_mut(); if doc.client_id() != self.id { doc.set_client_id(self.id); } if doc.next_id() < self.next_id { doc.reserve_ids(NodeId(self.next_id.0.saturating_sub(1))); } self.next_id = doc.next_id(); } /// The connection dropped: keep editing; batches wait for [`reconnect`](Self::reconnect). pub fn disconnect(&mut self) { self.online = false; } /// The first message of a new connection after [`disconnect`](Self::disconnect): `hello` /// (as for the first join) with this client's session, id and last sequence number. The /// server's `Welcome` then goes to [`receive`](Self::receive), which returns the batches to /// send. pub fn reconnect(&mut self, hello: Hello) -> ClientMsg { self.online = false; ClientMsg::Hello(Hello { resume: Some(self.resume()), ..hello }) } /// What a reconnect resumes with: this client's session, id and last sequence number. pub fn resume(&self) -> Resume { Resume { session: self.session, client_id: self.id, seq: self.seq } } /// Handle a message from the server. Returns messages to send: batches that were waiting for /// this one, a follow-up batch when this client's confirmed batch was ordered after someone /// else's and `recompute` changes derived data over it, or after a reconnect's `Welcome`, the /// queued batches. Errors mean the connection is unusable (rejoin). pub fn receive(&mut self, msg: ServerMsg, recompute: &dyn Recompute) -> Result, Error> { match msg { ServerMsg::Changes(acc) => { self.accepted(acc, recompute)?; Ok(self.flush(recompute)) } ServerMsg::Reject { batch_id, reason, fixes } => { let mine = self.pop_sent(batch_id, "rejection")?; // The confirmed copy already is the server's state (every batch ordered before // this rejection arrived first); the fixes change nothing there unless it drifted. let _ = self.confirmed.apply(&fixes, Mode::Lenient); self.refused(&mine, reason); self.rebuild(); Ok(self.flush(recompute)) } ServerMsg::Presence { client_id, presence } => { self.presence.insert(client_id, presence); Ok(Vec::new()) } ServerMsg::PeerJoined(p) => { self.peers.insert(p.client_id, p); Ok(Vec::new()) } ServerMsg::PeerLeft { client_id } => { self.peers.remove(&client_id); self.presence.remove(&client_id); Ok(Vec::new()) } // Cursor chat, spotlights and what a server says about its hosting are the app's (the // transport hands them over). ServerMsg::Pong | ServerMsg::Signal { .. } | ServerMsg::Hosting(_) | ServerMsg::Reported { .. } | ServerMsg::Resolved { .. } => { Ok(Vec::new()) } ServerMsg::Welcome { .. } if self.online => Err(Error::Protocol("already joined".into())), ServerMsg::Welcome { session, client_id, seq, catch_up, processed, peers, access } => { if session != self.session || client_id != self.id { return self.restart(ServerMsg::Welcome { session, client_id, seq, catch_up, processed, peers, access }); } self.access = access; self.rejoin(seq, catch_up, &processed, peers, recompute) } ServerMsg::Refused { reason } => Err(Error::Protocol(reason)), ServerMsg::Admin(_) | ServerMsg::Library(_) | ServerMsg::Plugins(_) => { Err(Error::Protocol("an answer meant for a managing, library or plugin directory connection".into())) } } } /// The oldest pending batch, which the server just confirmed or rejected as `batch_id`. fn pop_sent(&mut self, batch_id: u64, what: &str) -> Result { match self.pending.front() { Some(p) if p.sent && p.batch_id == batch_id => { self.pending.pop_front().ok_or_else(|| Error::Protocol(format!("{what} of an unknown batch"))) } Some(p) => Err(Error::Protocol(format!("{what} of batch {batch_id}, expected {}", p.batch_id))), None => Err(Error::Protocol(format!("{what} of an unknown batch"))), } } /// A batch the server accepted (in order). Follow-ups it causes are queued. fn accepted(&mut self, acc: Accepted, recompute: &dyn Recompute) -> Result<(), Error> { // The host's sequence number is untrusted: at `u64::MAX` it can't step past it. let Some(next) = self.seq.checked_add(1) else { return Err(Error::Protocol(format!("sequence {} after {}", acc.seq, self.seq))); }; if acc.seq != next { return Err(Error::Protocol(format!("sequence {} after {}", acc.seq, self.seq))); } if acc.author != self.id { self.confirmed.apply(&acc.changes, Mode::Strict)?; self.seq = acc.seq; // Ordered after the steps the server confirmed already: those leave its targets alone. let touches = Touches::of(&acc.changes); for s in self.undo.iter_mut().chain(self.redo.iter_mut()) { s.others_wrote(&touches); } for p in &mut self.pending { p.rebased = true; } self.rebuild(); return Ok(()); } let mine = self.pop_sent(acc.batch_id, "confirmation")?; let before = self.confirmed.clone(); let applied = self.confirmed.apply(&acc.changes, Mode::Strict)?; self.seq = acc.seq; if mine.kind != BatchKind::FollowUp && let Some(s) = mine.step.and_then(|id| find_step(&mut self.undo, &mut self.redo, id)) { // Later batches of the step haven't gone out yet (an undo waits for what came before). s.settle(mine.kind); s.refresh(&before, &self.confirmed, &acc.changes, mine.kind)?; } if mine.rebased { self.repair(&before, &applied.touched, recompute)?; } Ok(()) } /// Batch `p` was refused: its step leaves the undo history, with the batches of that step /// waiting behind it (an undo of the refused edit), and the user hears why. fn refused(&mut self, p: &Pending, reason: String) { if p.kind == BatchKind::FollowUp { return; } if let Some(id) = p.step { self.undo.retain(|s| s.id != id); self.redo.retain(|s| s.id != id); // Those never went out: an undo waits for its edit. self.pending.retain(|q| q.step != Some(id) || q.batch_id < p.batch_id || q.sent); } self.notices.push(Notice::Rejected { batch_id: p.batch_id, kind: p.kind, reason }); } /// Back after a dropped connection: catch up (missed batches, or a snapshot when the server /// no longer has them), settle the batches the server processed meanwhile, and return the /// ones to send. fn rejoin( &mut self, seq: u64, catch_up: CatchUp, processed: &Processed, peers: Vec, recompute: &dyn Recompute, ) -> Result, Error> { let last_batch = processed.last_batch; // An undo or redo lost in flight is made again before it goes out, knowing what others did // meanwhile: its step conflicts with what they wrote while it was in flight, and with what // comes now, as if it had never been sent. for p in self.pending.iter().filter(|p| p.sent && p.batch_id > last_batch && matches!(p.kind, BatchKind::Undo | BatchKind::Redo)) { if let Some(s) = p.step.and_then(|id| find_step(&mut self.undo, &mut self.redo, id)) { s.lost(); } } self.peers = peers.into_iter().map(|p| (p.client_id, p)).collect(); self.presence.clear(); let offline = || "it no longer applied when the connection came back".to_string(); match catch_up { CatchUp::Delta(batches) => { for acc in batches { if acc.author == self.id { // Batches the server processed before this one and didn't log were refused. while let Some(p) = self.pending.front().filter(|p| p.batch_id < acc.batch_id).cloned() { self.pending.pop_front(); self.refused(&p, offline()); } } self.accepted(acc, recompute)?; } } CatchUp::Snapshot(snapshot) => { let mut confirmed = Replica::from_snapshot(&snapshot)?; confirmed.document_mut().set_client_id(self.id); self.confirmed = confirmed; self.seq = seq; // What the server accepted meanwhile is in the snapshot. Which of the batches since // were others' isn't known, so the steps check every value they wrote below. while let Some(p) = self.pending.front().filter(|p| p.batch_id <= last_batch).cloned() { self.pending.pop_front(); if processed.refused.contains(&p.batch_id) { self.refused(&p, offline()); } else if p.kind != BatchKind::FollowUp && let Some(s) = p.step.and_then(|id| find_step(&mut self.undo, &mut self.redo, id)) { s.settle(p.kind); } } for s in self.undo.iter_mut().chain(self.redo.iter_mut()).filter(|s| s.confirmed) { s.verify(&self.confirmed)?; } for p in &mut self.pending { p.rebased = true; } } } if self.seq != seq { return Err(Error::Protocol(format!("caught up to sequence {}, the server is at {seq}", self.seq))); } // Processed by the server but neither logged nor confirmed: refused. while let Some(p) = self.pending.front().filter(|p| p.batch_id <= last_batch).cloned() { self.pending.pop_front(); self.refused(&p, offline()); } // The server never saw the rest: they go again, in order. for p in &mut self.pending { p.sent = false; } self.online = true; self.rebuild(); Ok(self.flush(recompute)) } /// A session that doesn't know this client (it restarted without its saved state) welcomed /// it as new: start over from its document; unconfirmed work and the undo history are gone. fn restart(&mut self, welcome: ServerMsg) -> Result, Error> { let lost = self.pending.iter().filter(|p| p.kind != BatchKind::FollowUp).count(); let notices = std::mem::take(&mut self.notices); let limit = self.history_limit; *self = Client::join(welcome)?; self.notices = notices; self.history_limit = limit; if lost > 0 { self.notices.push(Notice::Lost { batches: lost }); } Ok(Vec::new()) } /// The view is the confirmed state with the pending batches replayed on top (changes that no /// longer apply are skipped; the server will reject their batch). A step whose batch is /// replayed learns the values that batch now replaces (not from an undo or redo that waits /// to go out: that is made again from the step when it goes). fn rebuild(&mut self) { let mut v = self.confirmed.clone(); for p in &self.pending { let learns = p.kind == BatchKind::Edit || (p.sent && p.kind != BatchKind::FollowUp); let step = p.step.filter(|_| learns).and_then(|id| find_step(&mut self.undo, &mut self.redo, id)); let pre = step.as_ref().map(|_| v.clone()); // Lenient application never fails. let _ = v.apply(&p.changes, Mode::Lenient); if let (Some(s), Some(pre)) = (step, pre) { let _ = s.refresh(&pre, &v, &p.changes, p.kind); } } self.view = v; self.keep_ids(); } /// `base`'s document with derived data recomputed over `touched` (`before`: the state before /// the edit that touched them), as a diff from `base`. fn recomputed(&self, base: &Replica, before: &Document, touched: &BTreeSet, recompute: &dyn Recompute) -> Result { let mut work = base.document().clone(); // Ids the recompute makes must not collide with those of pending batches. work.reserve_ids(NodeId(self.next_id.0.saturating_sub(1))); recompute.recompute(&mut work, before, touched); diff(base, &work) } /// This client's batch landed after someone else's: recompute derived data over what it /// touched and queue what changes (minus properties a later pending batch writes anyway). fn repair(&mut self, before: &Replica, touched: &BTreeSet, recompute: &dyn Recompute) -> Result<(), Error> { let fix = self.recomputed(&self.confirmed, before.document(), touched, recompute)?; let later: BTreeSet<(NodeId, Prop)> = self .pending .iter() .flat_map(|p| &p.changes) .filter_map(|c| match c { Change::Set { id, prop, .. } => Some((*id, *prop)), _ => None, }) .collect(); let changes: Vec = fix.changes.into_iter().filter(|c| !matches!(c, Change::Set { id, prop, .. } if later.contains(&(*id, *prop)))).collect(); if changes.is_empty() { return Ok(()); } let _ = self.view.apply(&changes, Mode::Lenient); self.keep_ids(); self.queue(changes, BatchKind::FollowUp, None); Ok(()) } }