diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/config/mod.rs | 12 | ||||
| -rw-r--r-- | src/document/file.rs | 94 | ||||
| -rw-r--r-- | src/main.rs | 47 | ||||
| -rw-r--r-- | src/swarm/bootstrap.rs | 10 | ||||
| -rw-r--r-- | src/swarm/document_file.rs | 14 |
5 files changed, 156 insertions, 21 deletions
diff --git a/src/config/mod.rs b/src/config/mod.rs index 7df68e7..c0abc09 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -264,6 +264,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "This node's document on disk. Read at start: the node resumes with it, ready at once. Kept up to date while ready: changes are appended every --document-append, and the file is rewritten after a switch, a rejoin or a restart, and when the appended changes outgrew it.", ), option( + "keep-actor", + Kind::Flag, + "BOOL", + Some("false"), + "Resume writing as the actor this node wrote with before it stopped, if it stopped cleanly (its document file holds everything it wrote then), instead of a fresh one: an epoch then gathers a writer per node and per crash or drop rather than per start, which every transfer of its history carries a signature for. Needs document-file.", + ), + option( "document-append", Kind::Duration, "DURATION", @@ -654,6 +661,8 @@ pub(crate) struct Settings { pub(crate) genesis: bool, pub(crate) state_file: Option<PathBuf>, pub(crate) document_file: Option<PathBuf>, + #[serde(default)] + pub(crate) keep_actor: bool, #[serde( default = "default_document_append", with = "crate::units::duration_option" @@ -982,6 +991,9 @@ impl Settings { if self.document_append.is_zero() || self.document_rewrite.is_zero() { bail!("document-append and document-rewrite must be longer than 0"); } + if self.keep_actor && self.document_file.is_none() { + bail!("keep-actor needs document-file: it tells whether this node stopped cleanly"); + } if self.keepalive < Duration::from_millis(100) { bail!("keepalive must be at least 100ms"); } 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(); diff --git a/src/main.rs b/src/main.rs index 880d793..7c819e5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -165,18 +165,37 @@ async fn main() -> Result<()> { &files, &force, )?; - let (initial_doc, starting_epoch, starting_swarm, earlier_actors, initial_signatures) = - match starting { - Some(starting) => ( - starting.doc, - Some(starting.epoch), - Some((starting.swarm_id, starting.origin)), - starting.earlier_actors, - starting.signatures, - ), - None => (Document::new(), None, None, Vec::new(), Vec::new()), - }; + let ( + initial_doc, + starting_epoch, + starting_swarm, + earlier_actors, + initial_signatures, + resumable, + ) = match starting { + Some(starting) => ( + starting.doc, + Some(starting.epoch), + Some((starting.swarm_id, starting.origin)), + starting.earlier_actors, + starting.signatures, + starting.resumable, + ), + None => (Document::new(), None, None, Vec::new(), Vec::new(), None), + }; let mut initial_doc = initial_doc; + // Writing on as the actor of before, where this node stopped cleanly: everything it + // wrote as that one is in the document, so no change of it comes twice. + match resumable.filter(|_| args.keep_actor) { + Some(actor) => { + info!(actor = %actor, "Writing on as the actor of before (--keep-actor)"); + initial_doc.set_actor(actor); + } + None if args.keep_actor && starting_epoch.is_some() => { + info!("Writing as a fresh actor: this node didn't stop cleanly (--keep-actor)"); + } + None => {} + } if let Some(spec) = &args.remake { remake::remake_genesis(spec, &files, &mut initial_doc) .with_context(|| format!("Making the genesis document (--remake {spec})"))?; @@ -228,6 +247,11 @@ async fn main() -> Result<()> { args.document_rewrite, )) }); + // A clean stop's mark undone before anything is written: a crash from here on + // leaves a file that doesn't say so. + if let Some(file) = &document_file { + file.open(); + } let node = node::Node::new(node::NodeConfig { transport, identity: Some(transport::identity::IdentityConfig { @@ -367,6 +391,7 @@ async fn main() -> Result<()> { swarm::peer::farewell(&state).await; swarm::state_file::update(&state, true).await; swarm::document_file::flush(&state).await; + swarm::document_file::close(&state).await; Ok(()) } diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index d2a593b..5c3b5ef 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -266,6 +266,9 @@ pub(crate) struct StartingDocument { pub(crate) earlier_actors: Vec<ActorId>, /// The signatures of its changes (from the document file). pub(crate) signatures: Vec<(crate::document::ChangeHash, [u8; 64])>, + /// The actor this node wrote with, where it stopped cleanly (the document file's + /// `resumable`): it may write on as that one (`--keep-actor`). + pub(crate) resumable: Option<ActorId>, } /// What `--recover-from` and `--document-file` say about where to start. @@ -314,7 +317,7 @@ pub(crate) fn starting_document( Some(path) => Some(load_backup(path, files)?), }; - let (doc, origin, earlier_actors, signatures) = match (backup, from_file) { + let (doc, origin, earlier_actors, signatures, resumable) = match (backup, from_file) { (Some(_), Some(_)) if !force.allows("replaceDocumentFile") => { bail!( "recover-from would replace the document in the document file; add --force replaceDocumentFile to recover the backup anyway" @@ -326,13 +329,14 @@ pub(crate) fn starting_document( "Recovering the backup instead of the document file's document (--force replaceDocumentFile)" ); } - (backup, Origin::Backup, Vec::new(), Vec::new()) + (backup, Origin::Backup, Vec::new(), Vec::new(), None) } (None, Some(loaded)) => ( loaded.doc, Origin::DocumentFile, loaded.actors, loaded.signatures, + loaded.resumable, ), (None, None) => return Ok(None), }; @@ -343,6 +347,7 @@ pub(crate) fn starting_document( } let mut starting = identify(doc, origin, earlier_actors)?; starting.signatures = signatures; + starting.resumable = resumable; if let Some(remembered) = options.remembered.filter(|id| *id != starting.swarm_id) { let what = origin.describe(); @@ -379,6 +384,7 @@ fn identify( origin, earlier_actors, signatures: Vec::new(), + resumable: None, }) } diff --git a/src/swarm/document_file.rs b/src/swarm/document_file.rs index e066e27..4c7843d 100644 --- a/src/swarm/document_file.rs +++ b/src/swarm/document_file.rs @@ -25,6 +25,20 @@ pub(crate) async fn run(state: AppState) { } } +/// At a clean stop, after the last `flush`: the file marked as holding everything this +/// node wrote (`DocumentFile::close`), where it does (ready, not rejoining, up to the +/// document's heads), so a restart may write on as the same actor (`--keep-actor`). +pub(crate) async fn close(state: &AppState) { + let Some(file) = state.document_file.clone() else { + return; + }; + if state.remake.is_rejoining() || !state.is_ready().await { + return; + } + let heads = state.doc.read().await.doc.get_heads(); + file.close(&heads); +} + /// Writes what is due now (`DocumentFile::due`). Also called at shutdown. pub(crate) async fn flush(state: &AppState) { let Some(file) = state.document_file.clone() else { |