diff options
Diffstat (limited to 'src/document')
| -rw-r--r-- | src/document/file.rs | 94 |
1 files changed, 86 insertions, 8 deletions
diff --git a/src/document/file.rs b/src/document/file.rs index 2d588ee..56114f5 100644 --- a/src/document/file.rs +++ b/src/document/file.rs @@ -2,12 +2,14 @@ //! resumes with it instead of starting empty. //! //! Format: the line `continuum-document/2`, a header line (JSON: the actor IDs this -//! node wrote with, so a later rejoin replays those writes as its own), then records, -//! each a little-endian `u32` length and that many bytes, the first of them its kind -//! (`Record`): first a full save of the document, then increments (the changes after -//! the heads written before), and the signatures of its changes -//! (`document::signatures`), those the file holds none of yet, as -//! `signatures::pack` packs them. A record cut short by a crash is dropped when +//! node wrote with, so a later rejoin replays those writes as its own, and the one it +//! writes with, `current`), then records, each a little-endian `u32` length and that +//! many bytes, the first of them its kind (`Record`): first a full save of the document, +//! then increments (the changes after the heads written before), and the signatures of +//! its changes (`document::signatures`), those the file holds none of yet, as +//! `signatures::pack` packs them; last, at a clean stop, an empty `Closed` record: the +//! file holds everything the node wrote, so it may resume as `current` +//! (`--keep-actor`). A record cut short by a crash is dropped when //! reading. A file of `continuum-document/1` has records of the document alone, //! without their kind: a full save, then increments. //! @@ -44,6 +46,11 @@ enum Record { Save = 0, Increment = 1, Signatures = 2, + /// Empty, last: the node stopped with everything it wrote in the file. + Closed = 3, + /// Empty: the node started on the file, so a `Closed` before it says nothing of + /// what it writes now. + Opened = 4, } #[derive(Debug, Default, Serialize, Deserialize)] @@ -51,6 +58,9 @@ struct Header { /// Hex actor IDs. #[serde(default)] actors: Vec<String>, + /// Hex: the actor the node wrote with when the file was rewritten. + #[serde(default, skip_serializing_if = "Option::is_none")] + current: Option<String>, } /// A document file as read at start. @@ -61,6 +71,9 @@ pub(crate) struct Loaded { pub(crate) actors: Vec<ActorId>, /// The signatures of its changes, in the order they were written. pub(crate) signatures: Vec<(ChangeHash, [u8; 64])>, + /// The actor it wrote with, where the file ends with `Closed`: nothing it wrote is + /// missing, so it may write on as that actor. + pub(crate) resumable: Option<ActorId>, } /// Reads the document file; None if there is none. An error if it can't be read or @@ -109,9 +122,18 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> { }) .collect::<Result<Vec<_>>>()?; + let current = header + .current + .as_deref() + .map(|hex| { + ActorId::try_from(hex).map_err(|_| anyhow!("invalid actor {hex:?} in the header")) + }) + .transpose()?; + let mut records = &rest[line_end + 1..]; let mut doc: Option<Document> = None; let mut signatures = Vec::new(); + let mut closed = false; while !records.is_empty() { let Some((length, after)) = records.split_first_chunk::<4>() else { break; @@ -129,6 +151,8 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> { 0 => Record::Save, 1 => Record::Increment, 2 => Record::Signatures, + 3 => Record::Closed, + 4 => Record::Opened, other => bail!("a record of an unknown kind ({other})"), }; (kind, record) @@ -137,6 +161,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> { } else { (Record::Increment, record) }; + closed = kind == Record::Closed; match (kind, &mut doc) { (Record::Save, None) => doc = Some(super::load(record).context("the full save")?), (Record::Save, Some(_)) => bail!("a second full save"), @@ -147,6 +172,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> { super::signatures::unpack(record) .ok_or_else(|| anyhow!("signatures of {} bytes", record.len()))?, ), + (Record::Closed | Record::Opened, Some(_)) => {} (_, None) => bail!("a record before the full save"), } records = after; @@ -158,6 +184,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> { doc, actors, signatures, + resumable: current.filter(|_| closed && records.is_empty()), }; Ok((loaded, records.len())) } @@ -181,11 +208,13 @@ fn signatures_record(signatures: &[(ChangeHash, [u8; 64])]) -> Vec<u8> { /// A whole file: header, one full save, and the signatures. fn whole( actors: &[ActorId], + current: &ActorId, save: &[u8], signatures: &[(ChangeHash, [u8; 64])], ) -> Result<Vec<u8>> { let header = Header { actors: actors.iter().map(ActorId::to_hex_string).collect(), + current: Some(current.to_hex_string()), }; let mut out = MAGIC.to_vec(); out.extend(serde_json::to_vec(&header)?); @@ -336,7 +365,7 @@ impl DocumentFile { .filter(|actor| *actor == current || present.contains(actor)) .collect(); let save = super::save(©); - let bytes = whole(&actors, &save, &signatures)?; + let bytes = whole(&actors, ¤t, &save, &signatures)?; writer .files .write_atomic(FileKind::Document, &writer.path, &bytes)?; @@ -368,6 +397,44 @@ impl DocumentFile { } } + /// Marks the file as opened (`Record::Opened`), at start, before the node writes: + /// a `Closed` before it no longer says the file holds everything (a crash from here + /// on leaves it without one). Nothing where there is no file yet. + pub(crate) fn open(&self) { + if !self.path.exists() { + return; + } + if let Err(error) = + self.files + .append(FileKind::Document, &self.path, &record(Record::Opened, &[])) + { + error!(path = %self.path.display(), error = %error, "Couldn't mark the document file as opened"); + } + } + + /// Marks the file as holding everything the node wrote (`Record::Closed`), at a + /// clean stop: only where it was written up to `heads`, the document's now (after a + /// last append). Any record written after makes it not so. + pub(crate) fn close(&self, heads: &[ChangeHash]) { + let up_to_date = self + .lock() + .as_ref() + .is_some_and(|written| written.heads == heads); + if !up_to_date { + debug!("The document file isn't up to date at the stop: not marked as closed"); + return; + } + match self + .files + .append(FileKind::Document, &self.path, &record(Record::Closed, &[])) + { + Ok(()) => debug!("Marked the document file as closed"), + Err(error) => { + error!(path = %self.path.display(), error = %error, "Couldn't mark the document file as closed"); + } + } + } + /// Appends `increment`: what changed from `since` (the heads written last) to /// `heads`; and `signatures`, those of the document the file holds none of yet /// (`Due::Append`). @@ -430,7 +497,7 @@ mod tests { let actor = doc.get_actor().clone(); let heads = doc.get_heads(); let first = (heads[0], [1; 64]); - let mut bytes = whole(std::slice::from_ref(&actor), &doc.save(), &[first]).unwrap(); + let mut bytes = whole(std::slice::from_ref(&actor), &actor, &doc.save(), &[first]).unwrap(); change(&mut doc, |doc| doc.put(ROOT, "b", 2_i64).unwrap()); bytes.extend(record(Record::Increment, &doc.save_after(&heads))); @@ -443,6 +510,17 @@ mod tests { assert_ne!(loaded.doc.get_actor(), &actor); assert!(loaded.doc.get(ROOT, "b").unwrap().is_some()); assert_eq!(loaded.signatures, [first, second]); + // Not closed: no actor to resume as. + assert!(loaded.resumable.is_none()); + // Closed last: the actor it wrote with; a record after, not anymore. + let mut closed = bytes.clone(); + closed.extend(record(Record::Closed, &[])); + assert_eq!(parse(&closed).unwrap().0.resumable, Some(actor.clone())); + let mut opened = closed.clone(); + opened.extend(record(Record::Opened, &[])); + assert!(parse(&opened).unwrap().0.resumable.is_none()); + closed.extend(signatures_record(&[(doc.get_heads()[0], [3; 64])])); + assert!(parse(&closed).unwrap().0.resumable.is_none()); // A record cut short by a crash is dropped. let mut cut = bytes.clone(); |