iroh_db_core/
migration.rs

1use minicbor::{Decoder, Encoder};
2
3use crate::{BlobHash, MigrationId, SchemaDescriptor, SchemaError};
4
5const MAGIC: &str = "iroh-db";
6const FORMAT_MAJOR: u16 = 1;
7const FORMAT_MINOR: u16 = 0;
8const STAGE_KIND: u8 = 4;
9const ACTIVATION_KIND: u8 = 5;
10const MAX_RECORD_ID: usize = 1_024;
11const MAX_TARGET_RECORD: usize = 16 * 1024 * 1024;
12
13/// One bounded, invisible target record in a causal schema migration.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct SchemaStage {
16    migration_id: MigrationId,
17    from: SchemaDescriptor,
18    to: SchemaDescriptor,
19    source_record_id: Vec<u8>,
20    source_digest: BlobHash,
21    target_record: Vec<u8>,
22}
23
24impl SchemaStage {
25    /// Validates and constructs one staged record.
26    pub fn new(
27        migration_id: MigrationId,
28        from: SchemaDescriptor,
29        to: SchemaDescriptor,
30        source_record_id: Vec<u8>,
31        source_digest: BlobHash,
32        target_record: Vec<u8>,
33    ) -> Result<Self, MigrationCodecError> {
34        validate_transition(&from, &to)?;
35        if source_record_id.len() > MAX_RECORD_ID {
36            return Err(MigrationCodecError::RecordIdTooLarge(
37                source_record_id.len(),
38            ));
39        }
40        if target_record.len() > MAX_TARGET_RECORD {
41            return Err(MigrationCodecError::TargetRecordTooLarge(
42                target_record.len(),
43            ));
44        }
45        Ok(Self {
46            migration_id,
47            from,
48            to,
49            source_record_id,
50            source_digest,
51            target_record,
52        })
53    }
54
55    /// Returns the migration identifier.
56    pub const fn migration_id(&self) -> MigrationId {
57        self.migration_id
58    }
59
60    /// Returns the exact predecessor schema.
61    pub const fn from(&self) -> &SchemaDescriptor {
62        &self.from
63    }
64
65    /// Returns the exact target schema.
66    pub const fn to(&self) -> &SchemaDescriptor {
67        &self.to
68    }
69
70    /// Returns the source record's canonical primary key.
71    pub fn source_record_id(&self) -> &[u8] {
72        &self.source_record_id
73    }
74
75    /// Returns the digest of the canonical source materialization.
76    pub const fn source_digest(&self) -> BlobHash {
77        self.source_digest
78    }
79
80    /// Returns the canonical target materialized-record bytes.
81    pub fn target_record(&self) -> &[u8] {
82        &self.target_record
83    }
84
85    /// Encodes the stage into its unique durable representation.
86    pub fn encode_canonical(&self) -> Vec<u8> {
87        let mut encoder = Encoder::new(Vec::new());
88        encode_stage(&mut encoder, self).expect("encoding into Vec cannot fail");
89        encoder.into_writer()
90    }
91
92    /// Strictly decodes one canonical stage.
93    pub fn decode_canonical(bytes: &[u8]) -> Result<Self, MigrationCodecError> {
94        let decoded = decode_stage(bytes)?;
95        if decoded.encode_canonical() != bytes {
96            return Err(MigrationCodecError::NonCanonicalEncoding);
97        }
98        Ok(decoded)
99    }
100}
101
102/// The atomic publication marker for a complete staged migration.
103#[derive(Debug, Clone, PartialEq, Eq)]
104pub struct SchemaActivation {
105    migration_id: MigrationId,
106    from: SchemaDescriptor,
107    to: SchemaDescriptor,
108    staged_records: u64,
109    staging_digest: BlobHash,
110}
111
112impl SchemaActivation {
113    /// Validates and constructs an activation marker.
114    pub fn new(
115        migration_id: MigrationId,
116        from: SchemaDescriptor,
117        to: SchemaDescriptor,
118        staged_records: u64,
119        staging_digest: BlobHash,
120    ) -> Result<Self, MigrationCodecError> {
121        validate_transition(&from, &to)?;
122        Ok(Self {
123            migration_id,
124            from,
125            to,
126            staged_records,
127            staging_digest,
128        })
129    }
130
131    /// Returns the migration identifier.
132    pub const fn migration_id(&self) -> MigrationId {
133        self.migration_id
134    }
135
136    /// Returns the exact predecessor schema.
137    pub const fn from(&self) -> &SchemaDescriptor {
138        &self.from
139    }
140
141    /// Returns the exact target schema.
142    pub const fn to(&self) -> &SchemaDescriptor {
143        &self.to
144    }
145
146    /// Returns the exact number of selected stage records.
147    pub const fn staged_records(&self) -> u64 {
148        self.staged_records
149    }
150
151    /// Returns the digest committing to the complete ordered staging set.
152    pub const fn staging_digest(&self) -> BlobHash {
153        self.staging_digest
154    }
155
156    /// Encodes the activation into its unique durable representation.
157    pub fn encode_canonical(&self) -> Vec<u8> {
158        let mut encoder = Encoder::new(Vec::new());
159        encode_activation(&mut encoder, self).expect("encoding into Vec cannot fail");
160        encoder.into_writer()
161    }
162
163    /// Strictly decodes one canonical activation.
164    pub fn decode_canonical(bytes: &[u8]) -> Result<Self, MigrationCodecError> {
165        let decoded = decode_activation(bytes)?;
166        if decoded.encode_canonical() != bytes {
167            return Err(MigrationCodecError::NonCanonicalEncoding);
168        }
169        Ok(decoded)
170    }
171}
172
173/// Validation or canonical-codec failure for migration protocol objects.
174#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
175pub enum MigrationCodecError {
176    /// The transition crossed collection identities.
177    #[error("migration schemas belong to different collections")]
178    CollectionMismatch,
179    /// The target did not immediately follow its predecessor.
180    #[error("migration versions must be adjacent: {from} -> {to}")]
181    NonAdjacentVersion { from: u32, to: u32 },
182    /// The stable primary-key contract changed.
183    #[error("migration changes the primary-key contract")]
184    PrimaryKeyMismatch,
185    /// Stages from different migrations or schema transitions were combined.
186    #[error("staging set mixes migration identities or schema transitions")]
187    MixedStagingSet,
188    /// Two stages selected the same source primary key.
189    #[error("staging set contains a duplicate source record")]
190    DuplicateSourceRecord,
191    /// A source key exceeded the protocol bound.
192    #[error("migration record id is too large: {0} bytes")]
193    RecordIdTooLarge(usize),
194    /// A target materialization exceeded the operation payload bound.
195    #[error("migration target record is too large: {0} bytes")]
196    TargetRecordTooLarge(usize),
197    /// A nested schema was invalid.
198    #[error(transparent)]
199    Schema(#[from] SchemaError),
200    /// The object format major, minor, or kind is unsupported.
201    #[error("unsupported migration object format")]
202    UnsupportedFormat,
203    /// The binary object was malformed.
204    #[error("invalid migration encoding: {0}")]
205    InvalidEncoding(String),
206    /// The object was valid CBOR but not its unique representation.
207    #[error("migration encoding is not canonical")]
208    NonCanonicalEncoding,
209}
210
211/// Hashes a complete staging set in canonical source-key order.
212pub fn staging_digest(stages: &[SchemaStage]) -> Result<BlobHash, MigrationCodecError> {
213    let mut ordered: Vec<_> = stages.iter().collect();
214    ordered.sort_unstable_by(|left, right| left.source_record_id.cmp(&right.source_record_id));
215    if ordered
216        .windows(2)
217        .any(|pair| pair[0].source_record_id.as_slice() == pair[1].source_record_id.as_slice())
218    {
219        return Err(MigrationCodecError::DuplicateSourceRecord);
220    }
221    if let Some(first) = ordered.first()
222        && ordered.iter().any(|stage| {
223            stage.migration_id != first.migration_id
224                || stage.from.schema_id() != first.from.schema_id()
225                || stage.to.schema_id() != first.to.schema_id()
226        })
227    {
228        return Err(MigrationCodecError::MixedStagingSet);
229    }
230    let mut hasher = blake3::Hasher::new();
231    hasher.update(b"iroh-db/schema-staging/v1");
232    hasher.update(&(ordered.len() as u64).to_be_bytes());
233    for stage in ordered {
234        let bytes = stage.encode_canonical();
235        hasher.update(&(bytes.len() as u64).to_be_bytes());
236        hasher.update(&bytes);
237    }
238    Ok(BlobHash::from_bytes(*hasher.finalize().as_bytes()))
239}
240
241fn validate_transition(
242    from: &SchemaDescriptor,
243    to: &SchemaDescriptor,
244) -> Result<(), MigrationCodecError> {
245    if from.collection_id() != to.collection_id() {
246        return Err(MigrationCodecError::CollectionMismatch);
247    }
248    if from.version().checked_add(1) != Some(to.version()) {
249        return Err(MigrationCodecError::NonAdjacentVersion {
250            from: from.version(),
251            to: to.version(),
252        });
253    }
254    let from_id = from.fields().iter().find(|field| field.is_record_id());
255    let to_id = to.fields().iter().find(|field| field.is_record_id());
256    if from_id.map(|field| (field.id(), field.crdt(), field.value_type()))
257        != to_id.map(|field| (field.id(), field.crdt(), field.value_type()))
258    {
259        return Err(MigrationCodecError::PrimaryKeyMismatch);
260    }
261    Ok(())
262}
263
264fn encode_prefix(
265    encoder: &mut Encoder<Vec<u8>>,
266    map_len: u64,
267    kind: u8,
268    migration_id: MigrationId,
269    from: &SchemaDescriptor,
270    to: &SchemaDescriptor,
271) -> Result<(), minicbor::encode::Error<std::convert::Infallible>> {
272    encoder.map(map_len)?;
273    encoder.u8(0)?.str(MAGIC)?;
274    encoder.u8(1)?.u16(FORMAT_MAJOR)?;
275    encoder.u8(2)?.u16(FORMAT_MINOR)?;
276    encoder.u8(3)?.u8(kind)?;
277    encoder.u8(4)?.bytes(migration_id.as_bytes())?;
278    encoder.u8(5)?.bytes(&from.encode_canonical())?;
279    encoder.u8(6)?.bytes(&to.encode_canonical())?;
280    Ok(())
281}
282
283fn encode_stage(
284    encoder: &mut Encoder<Vec<u8>>,
285    stage: &SchemaStage,
286) -> Result<(), minicbor::encode::Error<std::convert::Infallible>> {
287    encode_prefix(
288        encoder,
289        10,
290        STAGE_KIND,
291        stage.migration_id,
292        &stage.from,
293        &stage.to,
294    )?;
295    encoder.u8(7)?.bytes(&stage.source_record_id)?;
296    encoder.u8(8)?.bytes(stage.source_digest.as_bytes())?;
297    encoder.u8(9)?.bytes(&stage.target_record)?;
298    Ok(())
299}
300
301fn encode_activation(
302    encoder: &mut Encoder<Vec<u8>>,
303    activation: &SchemaActivation,
304) -> Result<(), minicbor::encode::Error<std::convert::Infallible>> {
305    encode_prefix(
306        encoder,
307        9,
308        ACTIVATION_KIND,
309        activation.migration_id,
310        &activation.from,
311        &activation.to,
312    )?;
313    encoder.u8(7)?.u64(activation.staged_records)?;
314    encoder.u8(8)?.bytes(activation.staging_digest.as_bytes())?;
315    Ok(())
316}
317
318fn decode_prefix(
319    decoder: &mut Decoder<'_>,
320    map_len: u64,
321    kind: u8,
322) -> Result<(MigrationId, SchemaDescriptor, SchemaDescriptor), MigrationCodecError> {
323    expect_map_len(decoder, map_len)?;
324    expect_key(decoder, 0)?;
325    if decoder.str().map_err(decode_error)? != MAGIC {
326        return Err(MigrationCodecError::UnsupportedFormat);
327    }
328    expect_key(decoder, 1)?;
329    if decoder.u16().map_err(decode_error)? != FORMAT_MAJOR {
330        return Err(MigrationCodecError::UnsupportedFormat);
331    }
332    expect_key(decoder, 2)?;
333    if decoder.u16().map_err(decode_error)? != FORMAT_MINOR {
334        return Err(MigrationCodecError::UnsupportedFormat);
335    }
336    expect_key(decoder, 3)?;
337    if decoder.u8().map_err(decode_error)? != kind {
338        return Err(MigrationCodecError::UnsupportedFormat);
339    }
340    expect_key(decoder, 4)?;
341    let migration_id = MigrationId::from_bytes(fixed_bytes(decoder, "migration id")?);
342    expect_key(decoder, 5)?;
343    let from = SchemaDescriptor::decode_canonical(decoder.bytes().map_err(decode_error)?)?;
344    expect_key(decoder, 6)?;
345    let to = SchemaDescriptor::decode_canonical(decoder.bytes().map_err(decode_error)?)?;
346    Ok((migration_id, from, to))
347}
348
349fn decode_stage(bytes: &[u8]) -> Result<SchemaStage, MigrationCodecError> {
350    let mut decoder = Decoder::new(bytes);
351    let (migration_id, from, to) = decode_prefix(&mut decoder, 10, STAGE_KIND)?;
352    expect_key(&mut decoder, 7)?;
353    let source_record_id = decoder.bytes().map_err(decode_error)?.to_vec();
354    expect_key(&mut decoder, 8)?;
355    let source_digest = BlobHash::from_bytes(fixed_bytes(&mut decoder, "source digest")?);
356    expect_key(&mut decoder, 9)?;
357    let target_record = decoder.bytes().map_err(decode_error)?.to_vec();
358    if decoder.position() != bytes.len() {
359        return Err(MigrationCodecError::InvalidEncoding("trailing data".into()));
360    }
361    SchemaStage::new(
362        migration_id,
363        from,
364        to,
365        source_record_id,
366        source_digest,
367        target_record,
368    )
369}
370
371fn decode_activation(bytes: &[u8]) -> Result<SchemaActivation, MigrationCodecError> {
372    let mut decoder = Decoder::new(bytes);
373    let (migration_id, from, to) = decode_prefix(&mut decoder, 9, ACTIVATION_KIND)?;
374    expect_key(&mut decoder, 7)?;
375    let staged_records = decoder.u64().map_err(decode_error)?;
376    expect_key(&mut decoder, 8)?;
377    let staging_digest = BlobHash::from_bytes(fixed_bytes(&mut decoder, "staging digest")?);
378    if decoder.position() != bytes.len() {
379        return Err(MigrationCodecError::InvalidEncoding("trailing data".into()));
380    }
381    SchemaActivation::new(migration_id, from, to, staged_records, staging_digest)
382}
383
384fn fixed_bytes(decoder: &mut Decoder<'_>, label: &str) -> Result<[u8; 32], MigrationCodecError> {
385    decoder
386        .bytes()
387        .map_err(decode_error)?
388        .try_into()
389        .map_err(|_| MigrationCodecError::InvalidEncoding(format!("{label} must be 32 bytes")))
390}
391
392fn expect_map_len(decoder: &mut Decoder<'_>, expected: u64) -> Result<(), MigrationCodecError> {
393    let actual = decoder
394        .map()
395        .map_err(decode_error)?
396        .ok_or_else(|| MigrationCodecError::InvalidEncoding("indefinite map".into()))?;
397    if actual != expected {
398        return Err(MigrationCodecError::InvalidEncoding(format!(
399            "expected map length {expected}, got {actual}"
400        )));
401    }
402    Ok(())
403}
404
405fn expect_key(decoder: &mut Decoder<'_>, expected: u8) -> Result<(), MigrationCodecError> {
406    let actual = decoder.u8().map_err(decode_error)?;
407    if actual != expected {
408        return Err(MigrationCodecError::InvalidEncoding(format!(
409            "expected key {expected}, got {actual}"
410        )));
411    }
412    Ok(())
413}
414
415#[allow(clippy::needless_pass_by_value)]
416fn decode_error(error: minicbor::decode::Error) -> MigrationCodecError {
417    MigrationCodecError::InvalidEncoding(error.to_string())
418}
iroh_db_core/migration.rs

iroh_db_core/
migration.rs
