coven_database/
forget_records.rs

1//! The slot at which each kept change stopped being current, and forgetting
2//! the records no write still to be judged can name (§16.7).
3//!
4//! A change is current while it supplies its row's generation or sets one of
5//! its present cells. A change that stopped never becomes current again, so a
6//! genuine write names it only if it read less than the slot where it
7//! stopped; once every device that can still write has written past that
8//! slot, and this device's queued writes read past it, its record goes.
9
10use crate::write_encoding::{counter, decoded, encoded};
11use crate::{sqlite::DatabaseConnection, DbError};
12use coven_format::merge_fields;
13use coven_foundation::id_source::DeviceId;
14use coven_merge::{Audience, RowId, RowState, Slot, WriteId};
15use rusqlite::params;
16use std::collections::{BTreeMap, BTreeSet};
17
18/// The other devices that can still write: every device besides this one
19/// not gone for good (§14.1), each with the slot whose entry added it, when
20/// known. Stored files (§17.10) and kept records (§16.7) wait for the same
21/// devices.
22#[derive(Clone, Debug, PartialEq, Eq)]
23pub struct Writers(BTreeMap<DeviceId, Option<u64>>);
24
25impl Writers {
26    /// The devices, each with the slot that added it, when known.
27    pub fn new(devices: BTreeMap<DeviceId, Option<u64>>) -> Self {
28        Self(devices)
29    }
30
31    /// The devices.
32    pub fn devices(&self) -> impl Iterator<Item = DeviceId> + '_ {
33        self.0.keys().copied()
34    }
35
36    /// The greatest slot every one of these devices has seen: each has landed
37    /// a write that read through it, or was added after it, so writes nothing
38    /// that read less. `reads` holds the `read_through` of each device's last
39    /// landed write, for those that landed one. With no other device, every
40    /// slot.
41    pub fn through(&self, reads: &BTreeMap<DeviceId, u64>) -> u64 {
42        self.0
43            .iter()
44            .map(|(device, added)| {
45                // A device added at slot `a` writes nothing that read less
46                // than `a`; one that landed nothing and whose adding slot
47                // isn't known has seen nothing.
48                let seen = added.map(|added| added.saturating_sub(1));
49                reads.get(device).copied().max(seen).unwrap_or(0)
50            })
51            .min()
52            .unwrap_or(u64::MAX)
53    }
54}
55
56/// The writes current in a row's merge state: the one that supplied its
57/// generation, and each one setting a present cell.
58fn current(state: &RowState<coven_format::value::Value>) -> BTreeSet<WriteId> {
59    state
60        .generations()
61        .get(&state.generation())
62        .copied()
63        .into_iter()
64        .chain(state.cells().values().map(|cell| cell.write))
65        .collect()
66}
67
68/// `write`, landing in `slot`, moved `before` to `after`: every other write
69/// current before and not after stopped being current at `slot`, and its
70/// kept record says so.
71pub(crate) fn stopped(
72    db: &DatabaseConnection,
73    write: WriteId,
74    before: &RowState<coven_format::value::Value>,
75    after: &RowState<coven_format::value::Value>,
76    slot: Slot,
77) -> Result<(), DbError> {
78    let after = current(after);
79    for writer in current(before) {
80        if writer != write && !after.contains(&writer) {
81            stop(db, writer, before.row(), slot)?;
82        }
83    }
84    Ok(())
85}
86
87/// Record that `writer`'s change at `row`, named as the row is now, stopped
88/// being current at `slot`.
89fn stop(db: &DatabaseConnection, writer: WriteId, row: &RowId, slot: Slot) -> Result<(), DbError> {
90    let Some((_, table)) = crate::write_sources::kept_part(db, writer, row)? else {
91        // A migration write another raise replaced, or one a loaded
92        // snapshot covers without its part: no history holds its changes,
93        // so there is no record to mark (§9.5).
94        tracing::debug!(?writer, ?row, "a write stopped being current with no record held");
95        return Ok(());
96    };
97    stopped_at(db, writer, &row.audience, &table, &row.key, slot)
98}
99
100/// Whether `write` is current in a row it just changed, as `updates` leave
101/// the rows.
102pub(crate) fn applied<'a>(
103    write: WriteId,
104    mut updates: impl Iterator<Item = &'a RowState<coven_format::value::Value>>,
105) -> bool {
106    updates.any(|state| current(state).contains(&write))
107}
108
109/// `write`'s change at `table` and `key` in `audience`, as its write named
110/// them, stopped being current at `slot`, unless it had already stopped.
111pub(crate) fn stopped_at(
112    db: &DatabaseConnection,
113    write: WriteId,
114    audience: &Audience,
115    table: &str,
116    key: &[u8],
117    slot: Slot,
118) -> Result<(), DbError> {
119    db.internal_execute(
120        "UPDATE _coven_authored_changes SET stop=?5 WHERE write_id=?1 AND audience=?2 AND table_name=?3 AND key=?4 AND stop IS NULL",
121        params![
122            encoded(merge_fields::encode_write_id(&write))?,
123            audience,
124            table,
125            key,
126            slot.get().to_be_bytes().as_slice()
127        ],
128    )?;
129    Ok(())
130}
131
132/// `write` landed in `slot` and was consumed in `audiences` without being
133/// staged, as a write the cut-off refuses: none of its changes there ever
134/// became current, so each stops at its own slot.
135pub(crate) fn none_current(
136    db: &DatabaseConnection,
137    write: WriteId,
138    audiences: &[Audience],
139    slot: Slot,
140) -> Result<(), DbError> {
141    let id = encoded(merge_fields::encode_write_id(&write))?;
142    for audience in audiences {
143        db.internal_execute(
144            "UPDATE _coven_authored_changes SET stop=?3 WHERE write_id=?1 AND audience=?2 AND stop IS NULL",
145            params![id, audience, slot.get().to_be_bytes().as_slice()],
146        )?;
147    }
148    Ok(())
149}
150
151/// Forget every kept record no write still to be judged here can name: those
152/// that stopped being current at or below the slot every other device that
153/// can still write has seen, and this device's queued writes have read
154/// (§16.7). Then forget the applied-write entries no check reads any more
155/// (`forget_entries`). Run in the transaction that applies a slot, after its
156/// writes are judged; it deletes only what nothing can name, so no state is
157/// wrong before it runs.
158pub(crate) fn forget(db: &DatabaseConnection, writers: &Writers) -> Result<(), DbError> {
159    let mut reads = BTreeMap::new();
160    for device in writers.devices() {
161        if let Some(read) = crate::write_metadata::last_landed_read(db, device)? {
162            reads.insert(device, read);
163        }
164    }
165    let own = db
166        .local_device()
167        .ok_or(DbError::Invariant { what: "a writer connection observes this device" })?;
168    // No record stopped after the slot this device has applied, so looking
169    // past it would mark slots forgotten through before they come.
170    let seen = writers.through(&reads).min(crate::sequence::applied(db)?);
171    let through = match crate::write_metadata::queued_read(db, own)? {
172        Some(queued) => seen.min(queued),
173        None => seen,
174    };
175    let (from, newest) = forgotten_through(db)?;
176    if through <= from {
177        return crate::forget_entries::forget(db, through, newest);
178    }
179    let candidates = db.query(
180        "SELECT write_id,audience,table_name,key FROM _coven_authored_changes WHERE stop>?1 AND stop<=?2 ORDER BY write_id,audience",
181        params![from.to_be_bytes().as_slice(), through.to_be_bytes().as_slice()],
182        |r| {
183            Ok((
184                r.get::<_, Vec<u8>>(0)?,
185                crate::write_encoding::audience(r, 1)?,
186                r.get::<_, String>(2)?,
187                r.get::<_, Vec<u8>>(3)?,
188            ))
189        },
190    )?;
191    let mut parts = BTreeMap::<(Vec<u8>, Audience), Vec<(String, Vec<u8>)>>::new();
192    for (write, audience, table, key) in candidates {
193        parts.entry((write, audience)).or_default().push((table, key));
194    }
195    for ((write, audience), records) in parts {
196        forget_in_part(db, &write, &audience, records, through)?;
197    }
198    let newest = crate::forget_entries::passed(db, from, through, newest)?;
199    db.internal_execute(
200        "INSERT INTO _coven_forgotten(singleton,through,newest) VALUES(1,?1,?2)
201         ON CONFLICT(singleton) DO UPDATE SET through=excluded.through,newest=excluded.newest",
202        params![through.to_be_bytes().as_slice(), newest],
203    )?;
204    crate::forget_entries::forget(db, through, newest)
205}
206
207/// Forget the `candidates` of one part that may go once every write still to
208/// be judged has read `through`. Another member's creation that this device
209/// lists proves that member's authorship in its snapshots (format D7.5), so
210/// it stays; a part received whole opens its leaves by their place among all
211/// its records, so its records go together, with its leaf hashes, or stay.
212fn forget_in_part(
213    db: &DatabaseConnection,
214    write: &[u8],
215    audience: &Audience,
216    candidates: Vec<(String, Vec<u8>)>,
217    through: u64,
218) -> Result<(), DbError> {
219    let id = decoded(merge_fields::decode_write_id(write))?;
220    let header = db.authored_part(write, audience)?.ok_or(DbError::DamagedDatabase)?;
221    let version = merge_fields::decode_write_header(&header)?.schema_version;
222    let evidence = crate::authorship::keeps_evidence(db, id)?;
223    let protected = |table: &str, key: &[u8]| -> Result<bool, DbError> {
224        if !evidence {
225            return Ok(false);
226        }
227        let Some(table) = crate::source_names::table(db, version, table, db.schema_version()?)?
228        else {
229            return Ok(false);
230        };
231        let row = RowId { table, key: key.to_vec(), audience: audience.clone() };
232        Ok(crate::ownership::creation(db, &row)?.is_some_and(|creation| creation.write == id))
233    };
234    let leaves = crate::authorship::holds_leaves(db, id, audience)?;
235    let records = if leaves {
236        let mut records = Vec::new();
237        for (table, key, stop) in db.query(
238            "SELECT table_name,key,stop FROM _coven_authored_changes WHERE write_id=?1 AND audience=?2",
239            params![write, audience],
240            |r| {
241                Ok((
242                    r.get::<_, String>(0)?,
243                    r.get::<_, Vec<u8>>(1)?,
244                    r.get::<_, Option<Vec<u8>>>(2)?.map(counter).transpose()?,
245                ))
246            },
247        )? {
248            if stop.is_none_or(|stop| stop > through) || protected(&table, &key)? {
249                return Ok(());
250            }
251            records.push((table, key));
252        }
253        records
254    } else {
255        let mut records = Vec::new();
256        for (table, key) in candidates {
257            if !protected(&table, &key)? {
258                records.push((table, key));
259            }
260        }
261        records
262    };
263    for (table, key) in records {
264        db.internal_execute(
265            "DELETE FROM _coven_authored_changes WHERE write_id=?1 AND audience=?2 AND table_name=?3 AND key=?4",
266            params![write, audience, table, key],
267        )?;
268    }
269    let emptied = db.internal_execute(
270        "DELETE FROM _coven_authored_parts WHERE write_id=?1 AND audience=?2
271         AND NOT EXISTS(SELECT 1 FROM _coven_authored_changes c WHERE c.write_id=?1 AND c.audience=?2)",
272        params![write, audience],
273    )?;
274    if emptied > 0 {
275        db.forget_authored_part(write, audience);
276        if leaves {
277            crate::authorship::forget_part(db, id, audience)?;
278        }
279    }
280    Ok(())
281}
282
283/// The slot through which this device has looked at every stopped record
284/// and every applied-write entry, zero before it looked at any, and the
285/// latest stamped entry at or below it.
286fn forgotten_through(db: &DatabaseConnection) -> Result<(u64, Option<i64>), DbError> {
287    let forgotten = db
288        .query("SELECT through,newest FROM _coven_forgotten WHERE singleton=1", [], |r| {
289            Ok((counter(r.get(0)?)?, r.get::<_, Option<i64>>(1)?))
290        })?
291        .pop();
292    Ok(forgotten.unwrap_or((0, None)))
293}
294
295/// Records and entries at or below the slot looked at through may have
296/// become forgettable, or come back: a loaded snapshot brings the records and
297/// entries its writer still held, a dropped table's creations no longer
298/// protect their records, and merge rows a reload or a migration deletes no
299/// longer name their writes. The next slot looks at every one again.
300pub(crate) fn look_again(db: &DatabaseConnection) -> Result<(), DbError> {
301    db.internal_execute("DELETE FROM _coven_forgotten", [])?;
302    Ok(())
303}
304
305#[cfg(test)]
306#[path = "forget_records_tests.rs"]
307pub(crate) mod tests;
coven_database/forget_records.rs

coven_database/
forget_records.rs
