|
| 1 | +//! Runs the migrations this binary links on a document as it opens: format steps over the stored records, then node migrations |
| 2 | +//! over history. See `node-graph/rfcs/document-format-migrations.md`. |
| 3 | +
|
| 4 | +use std::collections::{BTreeSet, HashMap}; |
| 5 | + |
| 6 | +use document_container::AsyncContainer; |
| 7 | +use document_graph_storage::{Delta, EncodedDeclaration, HistoryMetadata, PeerId, Registry, ResourceEntry, ResourceId, Rev, Session, Subject, TimeStamp}; |
| 8 | +use migration_core::{Declaration, DeclarationResolver, NodeMigration, Payload, PayloadCodec}; |
| 9 | +use migration_runner::{DocumentPayloads, HistoryOutcome, RunnerError, holds, migrate_history, migrate_registry, references, registry_at, run_format_migrations}; |
| 10 | + |
| 11 | +use crate::error::Error; |
| 12 | +use crate::layout::Layout; |
| 13 | +use crate::load::StoredRecords; |
| 14 | +use crate::manifest::{PayloadCodecs, SUPPORTED_FORMAT_VERSION}; |
| 15 | +use crate::{Codec, Gdd}; |
| 16 | + |
| 17 | +/// Bring `records` from `format_version` to [`SUPPORTED_FORMAT_VERSION`]. `None` when they already are. |
| 18 | +pub(crate) fn migrate_format(records: &mut StoredRecords, codecs: PayloadCodecs, format_version: u32) -> Result<Option<HistoryOutcome>, Error> { |
| 19 | + if format_version >= SUPPORTED_FORMAT_VERSION { |
| 20 | + return Ok(None); |
| 21 | + } |
| 22 | + |
| 23 | + let payload = |record: &Vec<u8>, codec: Codec| Payload::new(record.clone(), payload_codec(codec)); |
| 24 | + let mut payloads = DocumentPayloads { |
| 25 | + format_version, |
| 26 | + registry: records.registry.as_ref().map(|record| payload(record, codecs.registry)), |
| 27 | + history: records.history.iter().flatten().map(|record| payload(record, codecs.history)).collect(), |
| 28 | + hot_log: records.hot_log.iter().map(|record| payload(record, codecs.hot_log)).collect(), |
| 29 | + session: records.session.as_ref().map(|record| payload(record, codecs.session)), |
| 30 | + }; |
| 31 | + let outcome = run_format_migrations(&migration_core::registered_format_migrations(), &mut payloads, SUPPORTED_FORMAT_VERSION)?; |
| 32 | + log::info!("Migrated the document format from version {format_version} to {SUPPORTED_FORMAT_VERSION}"); |
| 33 | + |
| 34 | + records.registry = payloads.registry.map(|payload| payload.bytes); |
| 35 | + records.history = match outcome { |
| 36 | + HistoryOutcome::Truncated => None, |
| 37 | + _ => records.history.as_ref().map(|_| payloads.history.into_iter().map(|payload| payload.bytes).collect()), |
| 38 | + }; |
| 39 | + records.hot_log = payloads.hot_log.into_iter().map(|payload| payload.bytes).collect(); |
| 40 | + records.session = payloads.session.map(|payload| payload.bytes); |
| 41 | + |
| 42 | + Ok(Some(outcome)) |
| 43 | +} |
| 44 | + |
| 45 | +/// The facts in `metadata` with their revs mapped by `remap`, which drops the facts of a rev it has no answer for. |
| 46 | +pub(crate) fn remap_metadata(metadata: &HistoryMetadata, remap: impl Fn(Rev) -> Option<Rev>) -> HistoryMetadata { |
| 47 | + let facts = metadata.facts().cloned().filter_map(|mut fact| { |
| 48 | + if let Subject::Rev(rev) = fact.subject { |
| 49 | + fact.subject = Subject::Rev(remap(rev)?); |
| 50 | + } |
| 51 | + Some(fact) |
| 52 | + }); |
| 53 | + HistoryMetadata::fold(facts) |
| 54 | +} |
| 55 | + |
| 56 | +fn payload_codec(codec: Codec) -> PayloadCodec { |
| 57 | + match codec { |
| 58 | + Codec::Json | Codec::JsonLines => PayloadCodec::Json, |
| 59 | + Codec::Postcard | Codec::PostcardFrames => PayloadCodec::Postcard, |
| 60 | + } |
| 61 | +} |
| 62 | + |
| 63 | +impl<L: Layout> Gdd<L> { |
| 64 | + /// Rewrite history through every registered node migration whose source declaration it references. Returns whether any ran. |
| 65 | + pub(crate) async fn migrate_nodes(&mut self) -> Result<bool, Error> { |
| 66 | + let migrations = migration_core::registered_migrations(); |
| 67 | + if migrations.is_empty() { |
| 68 | + return Ok(false); |
| 69 | + } |
| 70 | + |
| 71 | + let resolver = DocumentDeclarations::new(&migrations, self.session.peer()); |
| 72 | + |
| 73 | + let hot_ops = self.session.hot_log().iter().map(|hot_op| &hot_op.op); |
| 74 | + let kinds: Vec<_> = self.session.history().map(|delta| &delta.kind).chain(hot_ops).collect(); |
| 75 | + let registry = self.session.registry(); |
| 76 | + let pending: Vec<Declaration> = migrations |
| 77 | + .iter() |
| 78 | + .map(|migration| migration.source()) |
| 79 | + .filter(|source| holds(registry, *source, &resolver) || kinds.iter().any(|kind| references(kind, *source, &resolver))) |
| 80 | + .collect(); |
| 81 | + if pending.is_empty() { |
| 82 | + return Ok(false); |
| 83 | + } |
| 84 | + |
| 85 | + // Hot ops join history first, so they migrate with it |
| 86 | + self.session.retire_all()?; |
| 87 | + |
| 88 | + let migrated = migrate_registry(&migrations, self.session.retired_registry(), &resolver)?; |
| 89 | + (self.session, self.metadata) = match self.session_with_rewritten_history(&migrations, &migrated, &resolver) { |
| 90 | + Ok(rewritten) => rewritten, |
| 91 | + Err(error) => { |
| 92 | + log::warn!("Migrating this document's history failed, so undo can't reach before this point: {error}"); |
| 93 | + (self.session_with_truncated_history(migrated)?, remap_metadata(&self.metadata, |_| None)) |
| 94 | + } |
| 95 | + }; |
| 96 | + |
| 97 | + // Declarations the migrations introduced need their bytes next to the other resources |
| 98 | + let referenced = self.session.all_declaration_resources(); |
| 99 | + for declaration in resolver.encoded.values().filter(|declaration| referenced.contains_key(&declaration.id)) { |
| 100 | + let path = self.layout.resource_path(&declaration.hash); |
| 101 | + if !self.working.exists(&path).await { |
| 102 | + self.working.write_non_blocking(&path, &declaration.bytes)?; |
| 103 | + } |
| 104 | + } |
| 105 | + |
| 106 | + let names: Vec<String> = pending.iter().map(ToString::to_string).collect(); |
| 107 | + log::info!("Migrated nodes from {}", names.join(", ")); |
| 108 | + Ok(true) |
| 109 | + } |
| 110 | + |
| 111 | + /// A session over the rewritten history, as long as replaying that history reaches `migrated`, which a history missing its beginning wouldn't. |
| 112 | + fn session_with_rewritten_history(&self, migrations: &[&'static dyn NodeMigration], migrated: &Registry, resolver: &DocumentDeclarations) -> Result<(Session, HistoryMetadata), RunnerError> { |
| 113 | + let history: Vec<Delta> = self.session.history().cloned().collect(); |
| 114 | + let rewritten = migrate_history(migrations, history, resolver)?; |
| 115 | + let remap = |rev: Rev| rewritten.revs.get(&rev).copied().flatten(); |
| 116 | + |
| 117 | + let head = self.session.head_rev().and_then(remap); |
| 118 | + let registry = registry_at(&rewritten.deltas, head)?; |
| 119 | + if !registry.value_equal(migrated) { |
| 120 | + return Err(RunnerError::ReplayDiffers); |
| 121 | + } |
| 122 | + |
| 123 | + let previous = &self.session; |
| 124 | + let redo_stack = previous.redo_stack().iter().filter_map(|rev| remap(*rev)).collect(); |
| 125 | + let mut session = Session::load(previous.peer(), previous.user(), registry, rewritten.deltas, head, redo_stack, previous.next_node_counter()); |
| 126 | + if let Some(rev) = previous.last_broadcast_rev().and_then(remap) { |
| 127 | + session.publish_up_to(rev); |
| 128 | + } |
| 129 | + session.restore_hot_sequence(previous.last_hot_sequence()); |
| 130 | + session.restore_clock_counter(previous.clock_counter()); |
| 131 | + session.absorb_retired_marks(previous.retired_marks()); |
| 132 | + Ok((session, remap_metadata(&self.metadata, remap))) |
| 133 | + } |
| 134 | + |
| 135 | + /// A session whose history restarts from `migrated`. |
| 136 | + fn session_with_truncated_history(&self, migrated: Registry) -> Result<Session, Error> { |
| 137 | + let previous = &self.session; |
| 138 | + let restarted = Session::bootstrap_from_registry(previous.peer(), previous.user(), migrated.clone())?; |
| 139 | + |
| 140 | + let history: Vec<Delta> = restarted.history().cloned().collect(); |
| 141 | + let mut session = Session::load(previous.peer(), previous.user(), migrated, history, restarted.head_rev(), Vec::new(), previous.next_node_counter()); |
| 142 | + session.restore_hot_sequence(previous.last_hot_sequence()); |
| 143 | + session.restore_clock_counter(previous.clock_counter().max(restarted.clock_counter())); |
| 144 | + session.absorb_retired_marks(previous.retired_marks()); |
| 145 | + Ok(session) |
| 146 | + } |
| 147 | +} |
| 148 | + |
| 149 | +/// Resolves declaration resources to the declarations migrations name. A declaration's resource derives from its identifier and version |
| 150 | +/// alone, so this needs none of the stored bytes, which the editor keeps in its global cache rather than the document's container. |
| 151 | +struct DocumentDeclarations { |
| 152 | + known: HashMap<ResourceId, Declaration>, |
| 153 | + encoded: HashMap<Declaration, EncodedDeclaration>, |
| 154 | + peer: PeerId, |
| 155 | +} |
| 156 | + |
| 157 | +impl DocumentDeclarations { |
| 158 | + fn new(migrations: &[&'static dyn NodeMigration], peer: PeerId) -> Self { |
| 159 | + let named: BTreeSet<Declaration> = migrations |
| 160 | + .iter() |
| 161 | + .flat_map(|migration| [migration.source(), migration.target()].into_iter().chain(migration.pins().iter().copied())) |
| 162 | + .collect(); |
| 163 | + |
| 164 | + let mut known = HashMap::new(); |
| 165 | + let mut encoded = HashMap::new(); |
| 166 | + for declaration in named { |
| 167 | + match EncodedDeclaration::new(declaration.identifier.to_string(), declaration.version) { |
| 168 | + Ok(declaration_resource) => { |
| 169 | + known.insert(declaration_resource.id, declaration); |
| 170 | + encoded.insert(declaration, declaration_resource); |
| 171 | + } |
| 172 | + Err(error) => log::error!("Encoding the declaration {declaration} failed: {error}"), |
| 173 | + } |
| 174 | + } |
| 175 | + |
| 176 | + Self { known, encoded, peer } |
| 177 | + } |
| 178 | +} |
| 179 | + |
| 180 | +impl DeclarationResolver for DocumentDeclarations { |
| 181 | + fn declaration(&self, resource: ResourceId) -> Option<Declaration> { |
| 182 | + self.known.get(&resource).copied() |
| 183 | + } |
| 184 | + |
| 185 | + fn resource(&self, declaration: Declaration) -> ResourceId { |
| 186 | + match self.encoded.get(&declaration) { |
| 187 | + Some(encoded) => encoded.id, |
| 188 | + None => { |
| 189 | + log::error!("No migration names the declaration {declaration}, so it has no resource"); |
| 190 | + ResourceId::default() |
| 191 | + } |
| 192 | + } |
| 193 | + } |
| 194 | + |
| 195 | + fn resource_entry(&self, declaration: Declaration) -> Option<ResourceEntry> { |
| 196 | + let encoded = self.encoded.get(&declaration)?; |
| 197 | + Some(ResourceEntry::embedded(encoded.hash, self.peer, TimeStamp::ORIGIN)) |
| 198 | + } |
| 199 | +} |
0 commit comments