Replace folder sync with an HTTP sync server, keeping the app offline-first
Folder sync left too much to the user's sync tool: platform file access
(the Android SAF plugin), choosing the document, merging conflict copies —
and its merge base could fall out of step with the document (absorbing
conflict copies, "forget", picking a new Android folder), reading every
difference as a deletion.
The new mtgc-server crate stores one versioned library per user and accepts
a write only when it names the version it replaces. It never merges: devices
still do that with merge3, through a new sync cycle over a `Remote` trait in
mtgc-core. The base advances only on knowledge — to what was read, or to a
write the server confirmed — and is stored with the server and account it
belongs to, so a different account starts from an empty ancestor instead of
deleting. The working copy is only rewritten under the app-state lock, and
sync commands now run off the main thread.
Sign-in happens in the browser on the server's pages, via the owner's OIDC
provider and/or local username/password accounts (Argon2id, managed with
`mtgc-server user ...`), and returns a per-device revocable token to the app
by PKCE-protected code: a loopback port on desktop, the mtgc:// scheme on
Android.
The Info tab gains sign-in, the first-sync combine/replace choice, the
account's devices, and JSON export/import; the top bar shows sync state.
Includes a Dockerfile, Compose/Caddy example, and scripts to build and push
${DOCKER_REGISTRY}/mtgc/sync-server.
This commit is contained in:
@@ -1,11 +1,12 @@
|
||||
//! The library as a portable JSON document, and the merge that reconciles two
|
||||
//! copies of it.
|
||||
//!
|
||||
//! The document is what the user actually keeps: it lives in whatever folder
|
||||
//! they sync, and it is the only file they ever see. The SQLite database is an
|
||||
//! internal working copy that can always be rebuilt by merging a document into
|
||||
//! an empty one — which is why "first run", "cache was cleared", and "the
|
||||
//! document changed elsewhere" are all the same code path.
|
||||
//! The document is the library's shape for exchange: what a sync server stores
|
||||
//! (see [`remote`]), what a backup export writes, and what an import reads. On
|
||||
//! each device the library lives in an internal SQLite working copy, which can
|
||||
//! always be rebuilt by merging a document into an empty one — so "first
|
||||
//! sync", "imported a backup", and "changed on another device" are all the same
|
||||
//! code path.
|
||||
//!
|
||||
//! # Why a three-way merge
|
||||
//!
|
||||
@@ -18,13 +19,14 @@
|
||||
//!
|
||||
//! # The one subtle rule
|
||||
//!
|
||||
//! The stored ancestor must be **the remote document as last read**, never the
|
||||
//! result we last published. If a publish is lost — the sync tool dropped it,
|
||||
//! the process died, the drive was unplugged — the remote is still what we last
|
||||
//! The stored ancestor must be **the remote document as last known**, never
|
||||
//! the result we last *tried* to publish. If a publish is lost — the
|
||||
//! connection dropped, the process died — the remote is still what we last
|
||||
//! read, so the next merge sees our edits as ours and republishes them.
|
||||
//! Recording the merged result instead would make us believe those edits were
|
||||
//! already shared, and the next merge would read their absence from the remote
|
||||
//! as a deletion and destroy them.
|
||||
//! Recording the attempted result instead would make us believe those edits
|
||||
//! were already shared, and the next merge would read their absence from the
|
||||
//! remote as a deletion and destroy them. (A publish the server *confirmed* is
|
||||
//! known, so the ancestor may advance to it — see [`remote::sync_cycle`].)
|
||||
//!
|
||||
//! # Conflicts
|
||||
//!
|
||||
@@ -43,6 +45,9 @@ use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::error::Result;
|
||||
|
||||
pub mod account;
|
||||
pub mod remote;
|
||||
|
||||
/// Bumped only for a breaking change to the document shape. Readers reject
|
||||
/// anything newer rather than silently misreading it.
|
||||
pub const DOC_VERSION: u32 = 1;
|
||||
@@ -480,6 +485,23 @@ pub fn apply_doc(conn: &Connection, doc: &LibraryDoc) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Merge `theirs` into the working copy against `base`, and return the result.
|
||||
///
|
||||
/// The working copy is only rewritten when the merge changed it, so the common
|
||||
/// nothing-happened sync costs one read.
|
||||
pub fn merge_into(
|
||||
conn: &Connection,
|
||||
base: &LibraryDoc,
|
||||
theirs: &LibraryDoc,
|
||||
) -> Result<(LibraryDoc, MergeReport)> {
|
||||
let ours = read_doc(conn)?;
|
||||
let (merged, report) = merge3(base, &ours, theirs);
|
||||
if merged != ours {
|
||||
apply_doc(conn, &merged)?;
|
||||
}
|
||||
Ok((merged, report))
|
||||
}
|
||||
|
||||
/// Render keys as a JSON array, for `json_each` to expand inside a query —
|
||||
/// SQLite has no array parameter, and this avoids building SQL by
|
||||
/// concatenation.
|
||||
|
||||
258
crates/mtgc-core/src/sync/account.rs
Normal file
258
crates/mtgc-core/src/sync/account.rs
Normal file
@@ -0,0 +1,258 @@
|
||||
//! The account side of the sync server: checking an address, finishing a
|
||||
//! sign-in, and managing the account's devices.
|
||||
//!
|
||||
//! The sign-in itself happens in the browser, on the server's own pages (see
|
||||
//! the server's `auth` module); the app only starts it and redeems the
|
||||
//! single-use code it gets back. What the app proves when redeeming is that it
|
||||
//! is the one that started the sign-in — PKCE, RFC 7636 — which is what makes
|
||||
//! it safe for that code to travel through a browser and a URL scheme.
|
||||
|
||||
use base64::Engine;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use super::remote::HttpRemote;
|
||||
use crate::error::{CoreError, Result};
|
||||
use crate::http::{agent, USER_AGENT};
|
||||
|
||||
/// What a server says about itself, before anyone is signed in.
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct ServerInfo {
|
||||
pub server: String,
|
||||
pub api: u32,
|
||||
#[serde(default)]
|
||||
pub local_auth: bool,
|
||||
/// The identity provider's display name, when single sign-on is on.
|
||||
#[serde(default)]
|
||||
pub oidc: Option<String>,
|
||||
}
|
||||
|
||||
/// The newest server API this build speaks.
|
||||
pub const API_VERSION: u32 = 1;
|
||||
|
||||
/// Tidy a server address as typed: trim it, drop trailing slashes, and assume
|
||||
/// `https://` when no scheme is given.
|
||||
pub fn normalize_server_url(input: &str) -> String {
|
||||
let s = input.trim().trim_end_matches('/');
|
||||
if s.contains("://") {
|
||||
s.to_string()
|
||||
} else {
|
||||
format!("https://{s}")
|
||||
}
|
||||
}
|
||||
|
||||
/// Confirm `server_url` is an mtgc sync server this build can talk to, before
|
||||
/// sending the user's browser there.
|
||||
pub fn server_info(server_url: &str) -> Result<ServerInfo> {
|
||||
let url = format!("{server_url}/api/info");
|
||||
let info: ServerInfo = agent()
|
||||
.get(&url)
|
||||
.set("User-Agent", USER_AGENT)
|
||||
.set("Accept", "application/json")
|
||||
.call()
|
||||
.map_err(|e| CoreError::Http(format!("couldn't reach {server_url}: {e}")))?
|
||||
.into_json()
|
||||
.map_err(|_| not_a_server(server_url))?;
|
||||
if info.server != "mtgc" {
|
||||
return Err(not_a_server(server_url));
|
||||
}
|
||||
if info.api > API_VERSION {
|
||||
return Err(CoreError::Other(
|
||||
"that sync server is newer than this app — update the app first".into(),
|
||||
));
|
||||
}
|
||||
Ok(info)
|
||||
}
|
||||
|
||||
fn not_a_server(url: &str) -> CoreError {
|
||||
CoreError::Other(format!("{url} doesn't look like an mtgc sync server"))
|
||||
}
|
||||
|
||||
/// A PKCE verifier and its S256 challenge.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Pkce {
|
||||
pub verifier: String,
|
||||
pub challenge: String,
|
||||
}
|
||||
|
||||
impl Pkce {
|
||||
pub fn new() -> Self {
|
||||
let verifier = random_token();
|
||||
let challenge = Self::challenge_for(&verifier);
|
||||
Self { verifier, challenge }
|
||||
}
|
||||
|
||||
/// base64url(SHA-256(verifier)), unpadded — RFC 7636 §4.2.
|
||||
pub fn challenge_for(verifier: &str) -> String {
|
||||
base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(Sha256::digest(verifier.as_bytes()))
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for Pkce {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
/// 256 random bits, base64url: 43 characters, which is also exactly the
|
||||
/// shortest PKCE verifier RFC 7636 allows.
|
||||
pub fn random_token() -> String {
|
||||
let mut buf = [0u8; 32];
|
||||
getrandom::fill(&mut buf).expect("the OS random number generator failed");
|
||||
base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(buf)
|
||||
}
|
||||
|
||||
/// The browser address that starts a sign-in.
|
||||
pub fn start_url(
|
||||
server_url: &str,
|
||||
redirect_uri: &str,
|
||||
pkce: &Pkce,
|
||||
state: &str,
|
||||
device_name: &str,
|
||||
) -> String {
|
||||
let query: String = url_encode_pairs(&[
|
||||
("redirect_uri", redirect_uri),
|
||||
("code_challenge", &pkce.challenge),
|
||||
("code_challenge_method", "S256"),
|
||||
("state", state),
|
||||
("device_name", device_name),
|
||||
]);
|
||||
format!("{server_url}/auth/start?{query}")
|
||||
}
|
||||
|
||||
fn url_encode_pairs(pairs: &[(&str, &str)]) -> String {
|
||||
pairs
|
||||
.iter()
|
||||
.map(|(k, v)| format!("{k}={}", percent_encode(v)))
|
||||
.collect::<Vec<_>>()
|
||||
.join("&")
|
||||
}
|
||||
|
||||
/// Percent-encode everything but RFC 3986's unreserved characters.
|
||||
fn percent_encode(s: &str) -> String {
|
||||
let mut out = String::with_capacity(s.len());
|
||||
for b in s.bytes() {
|
||||
match b {
|
||||
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => out.push(b as char),
|
||||
_ => out.push_str(&format!("%{b:02X}")),
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// The account a device is signed in to.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct AccountUser {
|
||||
/// Stable id: what a stored merge base is keyed to.
|
||||
pub id: String,
|
||||
pub username: String,
|
||||
#[serde(default)]
|
||||
pub display_name: String,
|
||||
#[serde(default)]
|
||||
pub email: String,
|
||||
/// "local" or "oidc".
|
||||
#[serde(default)]
|
||||
pub method: String,
|
||||
}
|
||||
|
||||
/// The result of a completed sign-in.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct SignedIn {
|
||||
pub token: String,
|
||||
pub device_id: i64,
|
||||
pub user: AccountUser,
|
||||
}
|
||||
|
||||
/// Trade the code from the browser, with the verifier only this app holds, for
|
||||
/// this device's token.
|
||||
pub fn redeem_code(server_url: &str, code: &str, verifier: &str) -> Result<SignedIn> {
|
||||
let resp = agent()
|
||||
.post(&format!("{server_url}/auth/token"))
|
||||
.set("User-Agent", USER_AGENT)
|
||||
.send_json(serde_json::json!({ "code": code, "code_verifier": verifier }));
|
||||
match resp {
|
||||
Ok(r) => Ok(r.into_json()?),
|
||||
Err(ureq::Error::Status(400, _)) => Err(CoreError::Other(
|
||||
"the sign-in expired before it could be completed — try again".into(),
|
||||
)),
|
||||
Err(e) => Err(e.into()),
|
||||
}
|
||||
}
|
||||
|
||||
/// One of an account's signed-in devices.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct DeviceInfo {
|
||||
pub id: i64,
|
||||
pub name: String,
|
||||
pub created_at: i64,
|
||||
pub last_seen_at: i64,
|
||||
/// Whether it is the device asking.
|
||||
pub current: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Me {
|
||||
pub user: AccountUser,
|
||||
pub devices: Vec<DeviceInfo>,
|
||||
}
|
||||
|
||||
impl HttpRemote {
|
||||
pub fn me(&self) -> Result<Me> {
|
||||
Ok(self.request("GET", "/api/me").call().map_err(super::remote::http_error)?.into_json()?)
|
||||
}
|
||||
|
||||
/// Sign another of the account's devices out.
|
||||
pub fn revoke_device(&self, id: i64) -> Result<()> {
|
||||
self.request("DELETE", &format!("/api/devices/{id}"))
|
||||
.call()
|
||||
.map_err(super::remote::http_error)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Sign this device out on the server, so its token stops working even if
|
||||
/// a copy of it survives somewhere.
|
||||
pub fn logout(&self) -> Result<()> {
|
||||
self.request("POST", "/auth/logout").call().map_err(super::remote::http_error)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn server_addresses_are_tidied() {
|
||||
assert_eq!(normalize_server_url(" mtgc.example.com/ "), "https://mtgc.example.com");
|
||||
assert_eq!(normalize_server_url("http://10.0.2.2:8080//"), "http://10.0.2.2:8080");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pkce_matches_the_rfc_7636_example() {
|
||||
// Appendix B of the RFC.
|
||||
assert_eq!(
|
||||
Pkce::challenge_for("dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"),
|
||||
"E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"
|
||||
);
|
||||
let p = Pkce::new();
|
||||
assert_eq!(p.verifier.len(), 43);
|
||||
assert_ne!(p.verifier, Pkce::new().verifier);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_start_url_carries_everything_encoded() {
|
||||
let p = Pkce { verifier: "v".into(), challenge: "c-_".into() };
|
||||
let url = start_url(
|
||||
"https://m.example",
|
||||
"http://127.0.0.1:5000/callback",
|
||||
&p,
|
||||
"s t",
|
||||
"Garrett's phone",
|
||||
);
|
||||
assert_eq!(
|
||||
url,
|
||||
"https://m.example/auth/start?redirect_uri=http%3A%2F%2F127.0.0.1%3A5000%2Fcallback\
|
||||
&code_challenge=c-_&code_challenge_method=S256&state=s%20t&device_name=Garrett%27s%20phone"
|
||||
);
|
||||
}
|
||||
}
|
||||
521
crates/mtgc-core/src/sync/remote.rs
Normal file
521
crates/mtgc-core/src/sync/remote.rs
Normal file
@@ -0,0 +1,521 @@
|
||||
//! Syncing the library with a server that stores it, without ever needing one.
|
||||
//!
|
||||
//! The server is deliberately dumb: it keeps one versioned document per user
|
||||
//! and accepts a write only when the writer names the version it is replacing
|
||||
//! (compare-and-swap). It never merges. All reconciliation happens here, with
|
||||
//! the same [`merge3`](super::merge3) the rest of the app already trusts, so
|
||||
//! the working copy stays fully usable offline and simply catches up the next
|
||||
//! time the server answers.
|
||||
//!
|
||||
//! # The base, and how far it may advance
|
||||
//!
|
||||
//! The ancestor for each merge ([`SyncBase`]) is the server's document as this
|
||||
//! device last *knew* it to be. It advances only on knowledge:
|
||||
//!
|
||||
//! - after a read, to what was read;
|
||||
//! - after a write the server **confirmed**, to what was written.
|
||||
//!
|
||||
//! A write whose outcome is unknown — the connection dropped before the reply
|
||||
//! arrived — leaves the base at what was read. If the write did land, the next
|
||||
//! sync reads it back and finds it equal to our own copy; if it didn't, our
|
||||
//! edits are still recognisably ours and get sent again. Either way nothing is
|
||||
//! read as a deletion. That is the same rule the folder-based sync was built
|
||||
//! on, made enforceable by the server's version numbers.
|
||||
//!
|
||||
//! A base is only meaningful against the library it was read from, so it is
|
||||
//! stored together with the server and account it belongs to ([`AccountBase`])
|
||||
//! and ignored anywhere else. Pointing a device at a different account then
|
||||
//! starts from an empty ancestor — which never deletes anything — instead of
|
||||
//! reading every difference between two unrelated libraries as an edit.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use super::{LibraryDoc, MergeReport};
|
||||
use crate::error::{CoreError, Result};
|
||||
|
||||
/// The server's document at a particular version.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
|
||||
pub struct RemoteDoc {
|
||||
pub version: u64,
|
||||
pub doc: LibraryDoc,
|
||||
}
|
||||
|
||||
/// What became of a write.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum PutOutcome {
|
||||
/// Accepted; the document is now at this version.
|
||||
Stored(u64),
|
||||
/// Refused because someone else wrote first. Carries what they wrote, so
|
||||
/// the caller can merge against it without another round trip.
|
||||
Conflict(RemoteDoc),
|
||||
}
|
||||
|
||||
/// Where the library is stored. [`HttpRemote`] is the real one; tests use an
|
||||
/// in-memory fake, which is what keeps the cycle testable without a network.
|
||||
pub trait Remote {
|
||||
fn get(&self) -> Result<RemoteDoc>;
|
||||
fn put(&self, if_match: u64, doc: &LibraryDoc) -> Result<PutOutcome>;
|
||||
}
|
||||
|
||||
/// The merge ancestor: the server's document as last known.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
|
||||
pub struct SyncBase {
|
||||
pub version: u64,
|
||||
pub doc: LibraryDoc,
|
||||
}
|
||||
|
||||
/// A [`SyncBase`] together with the library it was read from.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct AccountBase {
|
||||
pub server_url: String,
|
||||
pub user_id: String,
|
||||
#[serde(flatten)]
|
||||
pub base: SyncBase,
|
||||
}
|
||||
|
||||
impl AccountBase {
|
||||
/// The base, if it belongs to this server and account. `None` means this
|
||||
/// device has never synced with that library, which callers treat as a
|
||||
/// first connection.
|
||||
pub fn for_account(self, server_url: &str, user_id: &str) -> Option<SyncBase> {
|
||||
(self.server_url == server_url && self.user_id == user_id).then_some(self.base)
|
||||
}
|
||||
}
|
||||
|
||||
/// What one cycle did.
|
||||
#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize)]
|
||||
pub struct CycleReport {
|
||||
/// Whether the server's document was replaced.
|
||||
pub published: bool,
|
||||
/// The server's version once the cycle finished.
|
||||
pub version: u64,
|
||||
/// What changed in the working copy, summed across retries.
|
||||
pub report: MergeReport,
|
||||
}
|
||||
|
||||
/// How many times to re-merge after losing a write race before giving up.
|
||||
/// Losing more than once means another device is syncing in a tight loop;
|
||||
/// the next scheduled sync will get through.
|
||||
const MAX_ATTEMPTS: usize = 4;
|
||||
|
||||
/// Reconcile the working copy with the server.
|
||||
///
|
||||
/// `merge_local(base, theirs)` must merge `theirs` into the working copy
|
||||
/// against `base` and return the merged document — in practice
|
||||
/// [`merge_into`](super::merge_into), run while holding whatever lock keeps
|
||||
/// the user's edits out, so none can land between reading the working copy
|
||||
/// and rewriting it. The network calls happen outside it.
|
||||
///
|
||||
/// `base` is advanced in place as far as is known to be true, and the caller
|
||||
/// should persist it **whether or not this returns an error**: a failure after
|
||||
/// a successful read still leaves a base worth keeping.
|
||||
pub fn sync_cycle<R, F>(remote: &R, base: &mut SyncBase, mut merge_local: F) -> Result<CycleReport>
|
||||
where
|
||||
R: Remote + ?Sized,
|
||||
F: FnMut(&LibraryDoc, &LibraryDoc) -> Result<(LibraryDoc, MergeReport)>,
|
||||
{
|
||||
let mut theirs = remote.get()?;
|
||||
let mut total = MergeReport::default();
|
||||
|
||||
for _ in 0..MAX_ATTEMPTS {
|
||||
let (merged, report) = merge_local(&base.doc, &theirs.doc)?;
|
||||
accumulate(&mut total, report);
|
||||
|
||||
// The working copy now holds theirs merged with ours, so theirs is the
|
||||
// ancestor of everything from here on.
|
||||
*base = SyncBase { version: theirs.version, doc: theirs.doc.clone() };
|
||||
|
||||
if merged == theirs.doc {
|
||||
return Ok(CycleReport { published: false, version: theirs.version, report: total });
|
||||
}
|
||||
|
||||
match remote.put(theirs.version, &merged)? {
|
||||
PutOutcome::Stored(version) => {
|
||||
*base = SyncBase { version, doc: merged };
|
||||
return Ok(CycleReport { published: true, version, report: total });
|
||||
}
|
||||
PutOutcome::Conflict(newer) => theirs = newer,
|
||||
}
|
||||
}
|
||||
|
||||
Err(CoreError::Other(
|
||||
"the library kept changing on the server while syncing — it will retry shortly".into(),
|
||||
))
|
||||
}
|
||||
|
||||
fn accumulate(total: &mut MergeReport, r: MergeReport) {
|
||||
total.cards_changed += r.cards_changed;
|
||||
total.decks_changed += r.decks_changed;
|
||||
total.decks_removed += r.decks_removed;
|
||||
total.conflicts.extend(r.conflicts);
|
||||
}
|
||||
|
||||
/// How to start syncing with a library this device has no base for.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum FirstSync {
|
||||
/// Merge both with an empty ancestor: a union, in which nothing is lost.
|
||||
Combine,
|
||||
/// Discard this device's library and take the server's.
|
||||
UseServer,
|
||||
}
|
||||
|
||||
/// Decide how a first sync proceeds. `None` means both sides hold a library and
|
||||
/// only the user can say which they meant, so the caller must ask.
|
||||
///
|
||||
/// When either side is empty there is nothing to choose between: combining is
|
||||
/// then just uploading or downloading.
|
||||
pub fn first_sync_plan(
|
||||
local: &LibraryDoc,
|
||||
remote: &LibraryDoc,
|
||||
choice: Option<FirstSync>,
|
||||
) -> Option<FirstSync> {
|
||||
if local.is_empty() || remote.is_empty() {
|
||||
return Some(FirstSync::Combine);
|
||||
}
|
||||
choice
|
||||
}
|
||||
|
||||
/// The library server, over HTTP.
|
||||
pub struct HttpRemote {
|
||||
base_url: String,
|
||||
token: String,
|
||||
}
|
||||
|
||||
/// The server no longer accepts this device's token: it was revoked, or the
|
||||
/// account removed. Distinguished so the app can show "signed out" instead of
|
||||
/// a generic failure that retrying will never fix.
|
||||
pub const UNAUTHORIZED: &str = "This device is no longer signed in to the sync server.";
|
||||
|
||||
impl HttpRemote {
|
||||
pub fn new(server_url: &str, token: &str) -> Self {
|
||||
Self { base_url: server_url.trim_end_matches('/').to_string(), token: token.to_string() }
|
||||
}
|
||||
|
||||
pub fn server_url(&self) -> &str {
|
||||
&self.base_url
|
||||
}
|
||||
|
||||
/// An authenticated request to `path` on the server.
|
||||
pub(crate) fn request(&self, method: &str, path: &str) -> ureq::Request {
|
||||
crate::http::agent()
|
||||
.request(method, &format!("{}{path}", self.base_url))
|
||||
.set("User-Agent", crate::http::USER_AGENT)
|
||||
.set("Authorization", &format!("Bearer {}", self.token))
|
||||
.set("Accept", "application/json")
|
||||
}
|
||||
}
|
||||
|
||||
impl Remote for HttpRemote {
|
||||
fn get(&self) -> Result<RemoteDoc> {
|
||||
let resp = self.request("GET", "/api/library").call().map_err(http_error)?;
|
||||
parse_remote(&resp.into_string()?)
|
||||
}
|
||||
|
||||
fn put(&self, if_match: u64, doc: &LibraryDoc) -> Result<PutOutcome> {
|
||||
let body = serde_json::to_string(doc)?;
|
||||
let result = self
|
||||
.request("PUT", "/api/library")
|
||||
.set("Content-Type", "application/json")
|
||||
.set("If-Match", &format!("\"{if_match}\""))
|
||||
.send_string(&body);
|
||||
match result {
|
||||
Ok(resp) => {
|
||||
#[derive(Deserialize)]
|
||||
struct Stored {
|
||||
version: u64,
|
||||
}
|
||||
let stored: Stored = serde_json::from_str(&resp.into_string()?)?;
|
||||
Ok(PutOutcome::Stored(stored.version))
|
||||
}
|
||||
Err(ureq::Error::Status(412, resp)) => {
|
||||
Ok(PutOutcome::Conflict(parse_remote(&resp.into_string()?)?))
|
||||
}
|
||||
Err(e) => Err(http_error(e)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse a `{version, doc}` body, applying the same newer-format check as a
|
||||
/// document read from disk.
|
||||
fn parse_remote(text: &str) -> Result<RemoteDoc> {
|
||||
#[derive(Deserialize)]
|
||||
struct Wire {
|
||||
version: u64,
|
||||
doc: serde_json::Value,
|
||||
}
|
||||
let wire: Wire = serde_json::from_str(text)?;
|
||||
Ok(RemoteDoc { version: wire.version, doc: LibraryDoc::from_json(&wire.doc.to_string())? })
|
||||
}
|
||||
|
||||
pub(crate) fn http_error(e: ureq::Error) -> CoreError {
|
||||
match e {
|
||||
ureq::Error::Status(401, _) => CoreError::Other(UNAUTHORIZED.into()),
|
||||
ureq::Error::Status(code, resp) => {
|
||||
let detail = resp.into_string().unwrap_or_default();
|
||||
let detail = detail.trim();
|
||||
if detail.is_empty() {
|
||||
CoreError::Http(format!("the sync server answered {code}"))
|
||||
} else {
|
||||
CoreError::Http(format!("the sync server answered {code}: {detail}"))
|
||||
}
|
||||
}
|
||||
other => CoreError::Http(other.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::cell::RefCell;
|
||||
|
||||
use super::*;
|
||||
use crate::sync::{merge3, CardEntry};
|
||||
|
||||
/// A server held in memory, with the same compare-and-swap rule as the real
|
||||
/// one, plus knobs for the failures worth testing.
|
||||
#[derive(Default)]
|
||||
struct FakeRemote {
|
||||
state: RefCell<RemoteDoc>,
|
||||
/// The next write is applied, but the reply is lost.
|
||||
drop_next_reply: RefCell<bool>,
|
||||
/// Another device writes this just before our next write lands.
|
||||
sneak_in: RefCell<Option<LibraryDoc>>,
|
||||
puts: RefCell<usize>,
|
||||
}
|
||||
|
||||
impl FakeRemote {
|
||||
fn holding(doc: LibraryDoc) -> Self {
|
||||
let r = Self::default();
|
||||
*r.state.borrow_mut() = RemoteDoc { version: 1, doc };
|
||||
r
|
||||
}
|
||||
|
||||
fn doc(&self) -> LibraryDoc {
|
||||
self.state.borrow().doc.clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl Remote for FakeRemote {
|
||||
fn get(&self) -> Result<RemoteDoc> {
|
||||
Ok(self.state.borrow().clone())
|
||||
}
|
||||
|
||||
fn put(&self, if_match: u64, doc: &LibraryDoc) -> Result<PutOutcome> {
|
||||
*self.puts.borrow_mut() += 1;
|
||||
if let Some(other) = self.sneak_in.borrow_mut().take() {
|
||||
let mut s = self.state.borrow_mut();
|
||||
s.version += 1;
|
||||
s.doc = other;
|
||||
}
|
||||
let mut s = self.state.borrow_mut();
|
||||
if s.version != if_match {
|
||||
return Ok(PutOutcome::Conflict(s.clone()));
|
||||
}
|
||||
s.version += 1;
|
||||
s.doc = doc.clone();
|
||||
if std::mem::take(&mut *self.drop_next_reply.borrow_mut()) {
|
||||
return Err(CoreError::Http("connection reset".into()));
|
||||
}
|
||||
Ok(PutOutcome::Stored(s.version))
|
||||
}
|
||||
}
|
||||
|
||||
/// A device: its working copy, as a document, and its base.
|
||||
struct Device {
|
||||
local: LibraryDoc,
|
||||
base: SyncBase,
|
||||
}
|
||||
|
||||
impl Device {
|
||||
fn new(local: LibraryDoc) -> Self {
|
||||
Self { local, base: SyncBase::default() }
|
||||
}
|
||||
|
||||
fn sync(&mut self, remote: &dyn Remote) -> Result<CycleReport> {
|
||||
let local = &mut self.local;
|
||||
sync_cycle(remote, &mut self.base, |base, theirs| {
|
||||
let (merged, report) = merge3(base, local, theirs);
|
||||
*local = merged.clone();
|
||||
Ok((merged, report))
|
||||
})
|
||||
}
|
||||
|
||||
fn set(&mut self, id: &str, qty: i64) {
|
||||
self.local
|
||||
.cards
|
||||
.insert(id.into(), CardEntry { quantity: qty, location: String::new() });
|
||||
}
|
||||
}
|
||||
|
||||
fn doc(cards: &[(&str, i64)]) -> LibraryDoc {
|
||||
let mut d = LibraryDoc::default();
|
||||
for (id, qty) in cards {
|
||||
d.cards.insert((*id).into(), CardEntry { quantity: *qty, location: String::new() });
|
||||
}
|
||||
d
|
||||
}
|
||||
|
||||
fn qty(d: &LibraryDoc, id: &str) -> Option<i64> {
|
||||
d.cards.get(id).map(|c| c.quantity)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn edits_made_offline_on_two_devices_both_reach_the_server() {
|
||||
let server = FakeRemote::default();
|
||||
let mut phone = Device::new(doc(&[("a", 1), ("b", 1)]));
|
||||
phone.sync(&server).unwrap();
|
||||
let mut desktop = Device::new(LibraryDoc::default());
|
||||
desktop.sync(&server).unwrap();
|
||||
|
||||
// Both go offline and edit different cards.
|
||||
phone.set("a", 3);
|
||||
desktop.set("b", 5);
|
||||
|
||||
phone.sync(&server).unwrap();
|
||||
desktop.sync(&server).unwrap();
|
||||
phone.sync(&server).unwrap();
|
||||
|
||||
for d in [&phone.local, &desktop.local, &server.doc()] {
|
||||
assert_eq!(qty(d, "a"), Some(3));
|
||||
assert_eq!(qty(d, "b"), Some(5));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_sync_with_nothing_to_contribute_does_not_write() {
|
||||
let server = FakeRemote::holding(doc(&[("a", 1)]));
|
||||
let mut d = Device::new(LibraryDoc::default());
|
||||
d.sync(&server).unwrap();
|
||||
let before = *server.puts.borrow();
|
||||
|
||||
let r = d.sync(&server).unwrap();
|
||||
|
||||
assert!(!r.published);
|
||||
assert_eq!(*server.puts.borrow(), before, "an idle sync must not bump the version");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_lost_reply_neither_deletes_nor_strands_the_edit() {
|
||||
let server = FakeRemote::default();
|
||||
let mut d = Device::new(doc(&[("a", 1)]));
|
||||
d.sync(&server).unwrap();
|
||||
|
||||
d.set("a", 4);
|
||||
d.set("b", 2);
|
||||
*server.drop_next_reply.borrow_mut() = true;
|
||||
assert!(d.sync(&server).is_err(), "the caller should hear it failed");
|
||||
|
||||
// The write did land. The base must not have advanced to it on faith,
|
||||
// but rereading it has to settle quietly either way.
|
||||
let r = d.sync(&server).unwrap();
|
||||
assert!(!r.published, "the server already has it");
|
||||
assert_eq!(qty(&d.local, "a"), Some(4));
|
||||
assert_eq!(qty(&d.local, "b"), Some(2));
|
||||
assert_eq!(d.base.doc, server.doc());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_write_that_never_landed_is_sent_again() {
|
||||
// The mirror of the case above: the request died before the server
|
||||
// saw it, so the server is unchanged.
|
||||
struct Unreachable<'a>(&'a FakeRemote);
|
||||
impl Remote for Unreachable<'_> {
|
||||
fn get(&self) -> Result<RemoteDoc> {
|
||||
self.0.get()
|
||||
}
|
||||
fn put(&self, _: u64, _: &LibraryDoc) -> Result<PutOutcome> {
|
||||
Err(CoreError::Http("network unreachable".into()))
|
||||
}
|
||||
}
|
||||
|
||||
let server = FakeRemote::holding(doc(&[("a", 1), ("b", 1)]));
|
||||
let mut d = Device::new(LibraryDoc::default());
|
||||
d.sync(&server).unwrap();
|
||||
|
||||
d.set("a", 7);
|
||||
assert!(d.sync(&Unreachable(&server)).is_err());
|
||||
assert_eq!(qty(&server.doc(), "a"), Some(1));
|
||||
|
||||
let r = d.sync(&server).unwrap();
|
||||
assert!(r.published, "the edit is still ours, so it must go out again");
|
||||
assert_eq!(qty(&server.doc(), "a"), Some(7));
|
||||
assert_eq!(qty(&server.doc(), "b"), Some(1), "nothing may be read as deleted");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn losing_a_write_race_re_merges_instead_of_overwriting() {
|
||||
let server = FakeRemote::holding(doc(&[("a", 1), ("b", 1)]));
|
||||
let mut d = Device::new(LibraryDoc::default());
|
||||
d.sync(&server).unwrap();
|
||||
|
||||
d.set("a", 2);
|
||||
// Another device publishes its own edit between our read and our write.
|
||||
*server.sneak_in.borrow_mut() = Some(doc(&[("a", 1), ("b", 9)]));
|
||||
|
||||
let r = d.sync(&server).unwrap();
|
||||
|
||||
assert!(r.published);
|
||||
assert_eq!(qty(&server.doc(), "a"), Some(2), "our edit");
|
||||
assert_eq!(qty(&server.doc(), "b"), Some(9), "their edit survives too");
|
||||
assert_eq!(d.local, server.doc());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_base_from_another_account_or_server_is_ignored() {
|
||||
let stored = AccountBase {
|
||||
server_url: "https://sync.example".into(),
|
||||
user_id: "u1".into(),
|
||||
base: SyncBase { version: 3, doc: doc(&[("a", 1)]) },
|
||||
};
|
||||
|
||||
assert!(stored.clone().for_account("https://sync.example", "u1").is_some());
|
||||
assert!(stored.clone().for_account("https://sync.example", "u2").is_none());
|
||||
assert!(stored.for_account("https://other.example", "u1").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_account_base_round_trips_through_json() {
|
||||
let stored = AccountBase {
|
||||
server_url: "https://sync.example".into(),
|
||||
user_id: "u1".into(),
|
||||
base: SyncBase { version: 3, doc: doc(&[("a", 1)]) },
|
||||
};
|
||||
let text = serde_json::to_string(&stored).unwrap();
|
||||
assert_eq!(serde_json::from_str::<AccountBase>(&text).unwrap(), stored);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn first_sync_only_asks_when_both_sides_have_a_library() {
|
||||
let some = doc(&[("a", 1)]);
|
||||
let none = LibraryDoc::default();
|
||||
|
||||
assert_eq!(first_sync_plan(&none, &some, None), Some(FirstSync::Combine));
|
||||
assert_eq!(first_sync_plan(&some, &none, None), Some(FirstSync::Combine));
|
||||
assert_eq!(first_sync_plan(&some, &some, None), None, "only the user can choose");
|
||||
assert_eq!(
|
||||
first_sync_plan(&some, &some, Some(FirstSync::UseServer)),
|
||||
Some(FirstSync::UseServer)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn combining_on_first_sync_loses_nothing_from_either_side() {
|
||||
// An empty base is what "combine" means: no difference reads as a
|
||||
// deletion.
|
||||
let server = FakeRemote::holding(doc(&[("a", 1), ("shared", 2)]));
|
||||
let mut d = Device::new(doc(&[("b", 3), ("shared", 2)]));
|
||||
|
||||
d.sync(&server).unwrap();
|
||||
|
||||
for id in ["a", "b", "shared"] {
|
||||
assert!(server.doc().cards.contains_key(id), "{id} should survive");
|
||||
}
|
||||
assert_eq!(d.local, server.doc());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_newer_format_from_the_server_is_refused() {
|
||||
let err = parse_remote(r#"{"version": 4, "doc": {"version": 99}}"#).unwrap_err();
|
||||
assert!(err.to_string().contains("newer version"), "got: {err}");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user