rindle_replica/
progress.rs

1//! **Connection progress + the poke rule** (`OPTIMISTIC-WRITES-DESIGN.md` §8.3–§8.4).
2//!
3//! The server-side bookkeeping that turns a stream of commits into per-connection
4//! [`ProgressFrame`]s. Per connection it tracks the live query set; per query, the
5//! highest commit version (`cv`) through which its result is **known-current**.
6//!
7//! **The poke rule (§8.4):** a connection is poked on a commit only when one of *its*
8//! queries' results actually changed. Mutation confirmation needs no extra rule:
9//! `lmid` is a row in [`crate::CLIENT_MUTATIONS_TABLE`], so a client's own lmid
10//! advance changes its one-row system query — an ordinary data poke, released by the
11//! same `cv_min` as the commit's effects. Irrelevant writes produce no frame — a
12//! read-only/idle client sees zero extra traffic; `cv_min` advancing in the
13//! background is invisible to a client that doesn't care.
14//!
15//! **`cv_min` is server-computed (§8.3)** because per-query silence is ambiguous to the
16//! client (current-and-unchanged vs. not-yet-processed). This tracker is the
17//! single-thread [`Db`](crate::Db) instantiation of the rule: every registered query is
18//! processed synchronously inside each commit, so [`note_commit`](ProgressTracker::note_commit)
19//! advances **every** query's known-current `cv` — the degenerate footprint check
20//! (untouched ⇒ trivially current). The multi-core `Cluster` drives the same rule from
21//! worker positions instead (the drain).
22//!
23//! The *standalone* quiet-window progress frame (§8.4) is a `Cluster`-era concern: on
24//! the single-thread path `cv_min` only moves when a commit happens, and every commit
25//! that matters to a connection already pokes it.
26
27use std::collections::{BTreeSet, HashMap};
28
29use crate::normalize_protocol::ProgressFrame;
30use crate::QueryId;
31
32/// Per-connection state: its live queries.
33struct Conn {
34    queries: BTreeSet<QueryId>,
35}
36
37/// See the module docs. Single-thread (`Db`) semantics; one instance per server.
38pub struct ProgressTracker {
39    /// The latest commit version noted (the replica's committed watermark).
40    frontier: u64,
41    /// Known-current `cv` per live query.
42    queries: HashMap<QueryId, u64>,
43    /// Connections, keyed by the caller's opaque connection id.
44    conns: HashMap<u64, Conn>,
45}
46
47impl ProgressTracker {
48    /// Open at the replica's current committed watermark
49    /// ([`Db::committed_tx_id`](crate::Db::committed_tx_id)).
50    pub fn new(initial_cv: u64) -> ProgressTracker {
51        ProgressTracker {
52            frontier: initial_cv,
53            queries: HashMap::new(),
54            conns: HashMap::new(),
55        }
56    }
57
58    /// Register a connection under the caller's opaque id.
59    pub fn connect(&mut self, conn: u64) {
60        self.conns.insert(
61            conn,
62            Conn {
63                queries: BTreeSet::new(),
64            },
65        );
66    }
67
68    /// Drop a connection and its query registrations.
69    pub fn disconnect(&mut self, conn: u64) {
70        if let Some(c) = self.conns.remove(&conn) {
71            for q in c.queries {
72                self.queries.remove(&q);
73            }
74        }
75    }
76
77    /// Register a live query on `conn`, known-current as of its hydrate watermark.
78    pub fn add_query(&mut self, conn: u64, query: QueryId, hydrated_cv: u64) {
79        if let Some(c) = self.conns.get_mut(&conn) {
80            c.queries.insert(query);
81            self.queries.insert(query, hydrated_cv);
82        }
83    }
84
85    /// Deregister a query (unsubscribed / destroyed) so it no longer pins `cv_min`.
86    pub fn remove_query(&mut self, conn: u64, query: QueryId) {
87        if let Some(c) = self.conns.get_mut(&conn) {
88            c.queries.remove(&query);
89        }
90        self.queries.remove(&query);
91    }
92
93    /// Fold one committed transaction in: `cv` (the commit's `TxId`) and the queries
94    /// whose results changed (the subscriptions that fired — a client's own lmid
95    /// advance fires its system query, so it is in here like any data change).
96    /// Advances every query's known-current `cv` (single-thread semantics — see module
97    /// docs) and returns the poke set: `(conn, frame)` for exactly the connections the
98    /// poke rule selects, in connection-id order.
99    pub fn note_commit(&mut self, cv: u64, changed: &[QueryId]) -> Vec<(u64, ProgressFrame)> {
100        self.frontier = self.frontier.max(cv);
101        for known in self.queries.values_mut() {
102            *known = (*known).max(cv);
103        }
104
105        let changed: BTreeSet<QueryId> = changed.iter().copied().collect();
106        let mut conn_ids: Vec<u64> = self.conns.keys().copied().collect();
107        conn_ids.sort_unstable();
108
109        let mut pokes = Vec::new();
110        for id in conn_ids {
111            let conn = &self.conns[&id];
112            let data_changed = conn.queries.iter().any(|q| changed.contains(q));
113            if !data_changed {
114                continue; // the poke rule: this commit is invisible to this connection
115            }
116            let cv_min = conn_cv_min(&self.queries, self.frontier, conn);
117            pokes.push((id, ProgressFrame { cv_min }));
118        }
119        pokes
120    }
121
122    /// The commit version every one of `conn`'s live queries has processed through
123    /// (the frame's `cv_min`); the frontier for a query-less connection.
124    pub fn cv_min(&self, conn: u64) -> Option<u64> {
125        self.conns
126            .get(&conn)
127            .map(|c| conn_cv_min(&self.queries, self.frontier, c))
128    }
129}
130
131fn conn_cv_min(queries: &HashMap<QueryId, u64>, frontier: u64, conn: &Conn) -> u64 {
132    conn.queries
133        .iter()
134        .map(|q| queries.get(q).copied().unwrap_or(frontier))
135        .min()
136        .unwrap_or(frontier)
137}
rindle_replica/progress.rs

rindle_replica/
progress.rs
