Offline mutation queue

Deep dive

How it works

Parts:

• local store — SQLite or IndexedDB holding the confirmed base state
• mutation queue — durable ordered rows (client_id, seq, name, args)
• mutators — pure functions apply(state, args), same code on client and server
• push endpoint — applies a batch idempotently, stores last_seq per client in the same transaction as the effects
• pull + rebase — new base from the server, confirmed entries dropped, the rest replayed

Steps:

1. User acts: run the mutator on the view and append (client_id, next seq, mutation) in one local transaction.
2. Coalesce a never-sent tail entry with the new one when they combine (two increments of a counter, two overwrites of a field). Sent entries stay untouched.
3. Online: push the head of the queue as a batch; keep entries until acknowledged.
4. Server, per mutation: seq <= last is a duplicate, skip; seq > last+1 is a gap, stop; else run the mutator and set last = seq in one transaction. A refused mutation still advances last and is recorded as rejected, so it is never retried.
5. Timeout, 5xx or lost response: back off with jitter (backoff) and resend the same entries; step 4 makes that safe.
6. Pull: replace the base, drop seq <= last_seq, replay the rest (they may now compute something else), show the new view and the rejects.

 UI <-- view = base + replay(pending)
 queue [seq 7 | 8 | 9] --push--> server last[client]=6 -> apply 7, 8; 9 rejected; last=9
 pull: base := server state; drop seq<=9; replay rest; show rejects

Conflict rules live in the mutators: server-authoritative checks (Withdraw fails below zero), last-writer-wins for plain fields, increments for counters. New rows get client-made UUIDs, so no temp id needs replacing; if the server must assign ids, keep a temp-to-real map and rewrite later queued mutations on the ack. A mutation refused every time (poison) is set aside and surfaced (dead-letter queue).

Code

Client (base, queue, coalescing), idempotent server keyed by (client, seq) whose rejects survive a lost ack, fault injection. Test (not shown): concurrent writer, rejected withdraw, lost response, lost request; asserts nothing lost or applied twice, view == server state, reject reported once, shorter queue. Compiled and run.

use std::collections::{BTreeMap, HashMap, VecDeque};

#[derive(Clone, Debug, PartialEq)]
pub enum Mutation {
    Add { key: String, delta: i64 },        // commutative: coalescible
    Withdraw { key: String, amount: i64 },  // server-authoritative: may be rejected
}
pub type State = BTreeMap<String, i64>;

/// The same pure function runs on the client (optimistic, rebase) and the server (authoritative).
pub fn apply(s: &mut State, m: &Mutation) -> Result<(), String> {
    match m {
        Mutation::Add { key, delta } => *s.entry(key.clone()).or_insert(0) += delta,
        Mutation::Withdraw { key, amount } => {
            let v = s.entry(key.clone()).or_insert(0);
            if *v < *amount { return Err(format!("insufficient {key}")); }
            *v -= amount;
        }
    }
    Ok(())
}

#[derive(Default)]
pub struct Server {
    pub state: State,
    pub applied: usize,
    last_seq: HashMap<String, u64>,                // per client: highest processed seq
    rejects: HashMap<String, Vec<(u64, String)>>,  // survives a lost ack
}
pub struct Ack { pub last_seq: u64 }

impl Server {
    /// Idempotent and ordered: seq <= last is a duplicate (skipped), a gap stops the batch.
    /// In a real store the state change and last_seq update commit in ONE transaction.
    pub fn push(&mut self, client: &str, batch: &[(u64, Mutation)]) -> Ack {
        let last = self.last_seq.entry(client.to_string()).or_insert(0);
        for (seq, m) in batch {
            if *seq <= *last { continue; }
            if *seq != *last + 1 { break; }
            match apply(&mut self.state, m) {
                Ok(()) => self.applied += 1,
                Err(e) => self.rejects.entry(client.to_string()).or_default().push((*seq, e)),
            }
            *last = *seq;
        }
        Ack { last_seq: *last }
    }
    pub fn pull(&self, client: &str) -> (State, u64, Vec<(u64, String)>) {
        (self.state.clone(), *self.last_seq.get(client).unwrap_or(&0),
         self.rejects.get(client).cloned().unwrap_or_default())
    }
}

#[derive(Clone, Copy)]
pub enum Fault { Ok, DropRequest, DropResponse }

pub struct Client {
    pub id: String,
    base: State,                         // last server-confirmed state
    pub pending: VecDeque<(u64, Mutation)>,
    next_seq: u64,
    sent_max: u64,                       // seqs <= sent_max may already be on the server
    pub rejected: Vec<(u64, String)>,    // surfaced to the user
    pub requests: usize,
}

impl Client {
    pub fn new(id: &str) -> Self {
        Client { id: id.into(), base: State::new(), pending: VecDeque::new(), next_seq: 1,
                 sent_max: 0, rejected: vec![], requests: 0 }
    }
    /// Optimistic view = base + pending replayed (rejects are skipped, as the server will).
    pub fn view(&self) -> State {
        let mut s = self.base.clone();
        for (_, m) in &self.pending { let _ = apply(&mut s, m); }
        s
    }
    pub fn enqueue(&mut self, m: Mutation) {
        // Coalesce into a tail entry that was never sent (its seq is not known to the server).
        if let (Some((seq, Mutation::Add { key: k, delta: d })), Mutation::Add { key, delta }) =
            (self.pending.back_mut(), &m)
        {
            if *seq > self.sent_max && k == key { *d += delta; return; }
        }
        self.pending.push_back((self.next_seq, m));
        self.next_seq += 1;
    }
    /// One push of up to `max` mutations; the injected fault decides what the network does.
    pub fn flush(&mut self, srv: &mut Server, f: Fault, max: usize) -> Result<(), ()> {
        if self.pending.is_empty() { return Ok(()); }
        let batch: Vec<_> = self.pending.iter().take(max).cloned().collect();
        self.sent_max = self.sent_max.max(batch.last().unwrap().0);
        self.requests += 1;
        if let Fault::DropRequest = f { return Err(()); }
        let ack = srv.push(&self.id, &batch);
        if let Fault::DropResponse = f { return Err(()); } // server applied, client does not know
        self.pending.retain(|(s, _)| *s > ack.last_seq);
        Ok(())
    }
    /// Rebase: new base from the server, drop confirmed mutations, keep the rest pending.
    pub fn pull(&mut self, srv: &Server) {
        let (state, last, rejects) = srv.pull(&self.id);
        self.base = state;
        self.pending.retain(|(s, _)| *s > last);
        for r in rejects { if !self.rejected.contains(&r) { self.rejected.push(r); } }
    }
}

Complexity and constants

• Enqueue: one local transaction, dominated by the durable write (fsync or IndexedDB commit), not CPU.
• Push: one round trip per batch; server work O(batch) plus one last_seq row per client.

Efficiency

• Cost: one action = one local durable write plus a share of one HTTPS request; the bound is round-trip latency and radio wake-ups, not compute.
• Levers:
  • batch push — my sketch, 1,000 edits: 1,000 requests become 2 at batch 50 (measured, model) — always
  • coalescing — same run: 1,000 queued edits shrink to 100 (measured, model) — commutative or overwrite mutations
  • one local and one server transaction per batch — Replicache documents applying a batch in one transaction (push) — bulk replays
  • delta pulls — cursor returns only changes since the last pull — datasets larger than a screen
  • HTTP compression of the batch body — gain depends on the payload, order of several times for repetitive JSON — batches above a few KB
• Naive → standard → state of the art:
  • one request per edit, full re-download · O(edits) requests, O(dataset) bytes
  • batched push, full pull of changed rows · O(1) requests per batch, O(changes) bytes
  • coalesced batches, cursor delta pull, background sync · fewest requests and wake-ups, queue bounded by targets

Implementation notes

• Persist the seq counter with the queue; never reuse a seq, it breaks the duplicate test.
• Mutators deterministic: pass clocks and randomness in as arguments.
• Bound retries (Workbox maxRetentionTime).

Pitfalls

• Coalescing an already-sent entry: the server holds the old one, so the sum applies twice.
• Showing server state after a pull without replaying pending: the user's edits vanish. Retrying a poison mutation forever starves the ordered queue.
• Rebase costs O(pending) mutator runs per pull.

Variants and alternatives

In production

• Replicache: lastMutationID per client group, named mutators, pull, rebase (how it works).
• Cloud Firestore: offline persistence queues writes; for several changes to one document the last write wins.
• Workbox Background Sync: failed requests kept in IndexedDB, replayed on the sync event.
• AWS Amplify DataStore: local store synced to the cloud with configurable conflict resolution.

Implementations

Sources

• Replicache docs, How Replicache Works. docs
• Replicache docs, Push Endpoint. docs
• PowerSync docs, Writing Data. docs
• Firebase docs, Access data offline. docs
• Chrome for Developers, workbox-background-sync. docs
• WatermelonDB docs, Sync Intro. docs