armdb 0.14.4

//! Metadata-only format steps (`FormatMigration::unchanged`).

use armour_core::Typ;
use armour_typ::{SchemaTyp, widens};
use serde::{Deserialize, Deserializer, Serialize, Serializer, ser::SerializeStruct};
use serde_json::value::RawValue;
use std::sync::Arc;

use crate::codec::SchemaEncoding;
use crate::error::{DbError, DbResult};
use crate::format_migration::plan::MigrationPlan;

/// Schema of a collection's stored rows, persisted in `db.info`.
#[derive(Debug, Clone)]
pub struct StoredSchema {
    pub encoding: SchemaEncoding,
    typ: Arc<SchemaJson>,
}

#[derive(Debug)]
struct SchemaJson {
    text: String,
    // db.info stores the Typ JSON as a JSON string. Cache that escaped string
    // once so every metadata update can copy it without scanning it again.
    escaped: Box<RawValue>,
}

impl PartialEq for StoredSchema {
    fn eq(&self, other: &Self) -> bool {
        self.encoding == other.encoding && self.typ.text == other.typ.text
    }
}

impl Eq for StoredSchema {}

impl Serialize for StoredSchema {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let mut state = serializer.serialize_struct("StoredSchema", 2)?;
        state.serialize_field("encoding", &self.encoding)?;
        state.serialize_field("typ", &self.typ.escaped)?;
        state.end()
    }
}

impl<'de> Deserialize<'de> for StoredSchema {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        #[derive(Deserialize)]
        struct Stored {
            encoding: SchemaEncoding,
            typ: String,
        }
        let Stored { encoding, typ } = Stored::deserialize(deserializer)?;
        Self::from_json(encoding, typ).map_err(serde::de::Error::custom)
    }
}

impl StoredSchema {
    /// The stored `Typ` JSON. Immutable so metadata snapshots can share its
    /// cached JSON encoding without stale serialization after a field edit.
    pub fn typ(&self) -> &str {
        &self.typ.text
    }

    fn from_json(encoding: SchemaEncoding, text: String) -> serde_json::Result<Self> {
        let escaped = serde_json::value::to_raw_value(&text)?;
        Ok(Self {
            encoding,
            typ: Arc::new(SchemaJson { text, escaped }),
        })
    }

    pub(crate) fn of(encoding: SchemaEncoding, typ: &Typ) -> DbResult<Self> {
        let typ = serde_json::to_string(typ)
            .map_err(|err| DbError::FormatMismatch(format!("schema serialization: {err}")))?;
        Self::from_json(encoding, typ)
            .map_err(|err| DbError::FormatMismatch(format!("schema serialization: {err}")))
    }

    pub(crate) fn parse(&self) -> DbResult<SchemaTyp> {
        serde_json::from_str(self.typ())
            .map_err(|err| DbError::FormatMismatch(format!("stored schema is unreadable: {err}")))
    }
}

/// A transition applied by `FormatMigration::unchanged`: rows stored at
/// `from_version` with fingerprint `from_typ_hash` are byte-identical at
/// `to_version`. Kept forever: pending txlog records and lagging `format.v`
/// markers refer to it after restarts.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct UnchangedRecord {
    pub from_version: u16,
    pub to_version: u16,
    pub from_typ_hash: u64,
}

/// Applied typed value migrations: rows stored at `from_version` decode at
/// `to_version` without a physical rewrite, so a shard's `format.v` may still
/// read `from_version`. Adjacent steps merge into one record. Kept forever: a
/// replay plan that still declares an earlier checked rewrite crosses them.
/// Unlike [`UnchangedRecord`] it admits no logged fingerprint to replay.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct ValueRecord {
    pub from_version: u16,
    pub to_version: u16,
}

/// Append one applied value step, merging it into an adjacent record.
pub(crate) fn record_value_step(history: &mut Vec<ValueRecord>, from_version: u16) {
    let to_version = from_version + 1;
    match history.last_mut() {
        Some(last) if last.to_version == from_version => last.to_version = to_version,
        _ => history.push(ValueRecord {
            from_version,
            to_version,
        }),
    }
}

/// Oldest version whose bytes equal `source`'s through recorded transitions.
pub(crate) fn physical_floor(
    unchanged: &[UnchangedRecord],
    value: &[ValueRecord],
    source: u16,
) -> u16 {
    let mut floor = source;
    while let Some(from) = unchanged
        .iter()
        .map(|record| (record.from_version, record.to_version))
        .chain(
            value
                .iter()
                .map(|record| (record.from_version, record.to_version)),
        )
        .find_map(|(from, to)| (to == floor && from < floor).then_some(from))
    {
        floor = from;
    }
    floor
}

/// Prove every unchanged run of `plan` before any durable write.
pub(crate) fn verify_unchanged(
    plan: &MigrationPlan,
    stored: Option<&StoredSchema>,
    encoding: Option<SchemaEncoding>,
    target: &Typ,
) -> DbResult<()> {
    for run in &plan.unchanged_runs {
        if encoding != Some(SchemaEncoding::Rapira) {
            return Err(DbError::FormatUnsupportedCodec {
                collection: plan.collection,
                from: run.from_version,
            });
        }
        let before = match (run.before_typ, run.starts_at_stored) {
            (Some(typ), _) => SchemaTyp::from(typ),
            (None, true) => {
                let stored = stored.ok_or(DbError::FormatSchemaUnknown {
                    collection: plan.collection,
                    version: run.from_version,
                })?;
                if stored.encoding != SchemaEncoding::Rapira {
                    return Err(DbError::FormatUnsupportedCodec {
                        collection: plan.collection,
                        from: run.from_version,
                    });
                }
                stored.parse()?
            }
            (None, false) => {
                return Err(DbError::FormatSchemaUnknown {
                    collection: plan.collection,
                    version: run.from_version,
                });
            }
        };
        let after = SchemaTyp::from(run.after_typ.unwrap_or(target));
        widens(&before, &after).map_err(|err| DbError::FormatIncompatible {
            collection: plan.collection,
            from: run.from_version,
            path: err.path,
            reason: err.reason,
        })?;
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use armour_core::GetType;

    fn probe_schema() -> StoredSchema {
        StoredSchema::of(
            SchemaEncoding::Rapira,
            &Typ::Struct(armour_typ::StructType {
                name: "MetadataProbe",
                fields: armour_typ::Fields::Unnamed(&[String::TYPE; 64]),
            }),
        )
        .unwrap()
    }

    #[test]
    fn stored_schema_clone_shares_immutable_json_and_preserves_serialization() {
        let schema = probe_schema();
        let cloned = schema.clone();
        assert!(
            std::ptr::eq(schema.typ(), cloned.typ()),
            "metadata snapshots must share immutable schema text"
        );
        #[derive(Serialize)]
        struct Previous<'a> {
            encoding: SchemaEncoding,
            typ: &'a str,
        }
        let previous = Previous {
            encoding: schema.encoding,
            typ: schema.typ(),
        };
        assert_eq!(
            serde_json::to_vec(&schema).unwrap(),
            serde_json::to_vec(&previous).unwrap()
        );
        assert_eq!(
            serde_json::to_vec_pretty(&schema).unwrap(),
            serde_json::to_vec_pretty(&previous).unwrap()
        );
        let decoded: StoredSchema =
            serde_json::from_slice(&serde_json::to_vec_pretty(&schema).unwrap()).unwrap();
        assert_eq!(decoded, schema);
        assert_eq!(decoded.parse().unwrap(), schema.parse().unwrap());
        assert!(Arc::ptr_eq(&schema.typ, &cloned.typ));
        assert!(Arc::ptr_eq(&decoded.typ, &decoded.clone().typ));
    }

    #[test]
    fn physical_floor_walks_unchanged_and_value_transitions() {
        // Rewrite to 2, then value 2..3, unchanged 3..5, value 5..6.
        let unchanged = [UnchangedRecord {
            from_version: 3,
            to_version: 5,
            from_typ_hash: 11,
        }];
        let value = [
            ValueRecord {
                from_version: 2,
                to_version: 3,
            },
            ValueRecord {
                from_version: 5,
                to_version: 6,
            },
        ];
        assert_eq!(physical_floor(&unchanged, &value, 6), 2);
        assert_eq!(physical_floor(&unchanged, &value, 5), 2);
        assert_eq!(physical_floor(&[], &value, 3), 2);
        // The unchanged run alone stops at the value step it cannot see past.
        assert_eq!(physical_floor(&unchanged, &[], 6), 6);
        assert_eq!(physical_floor(&unchanged, &[], 5), 3);
        // No recorded transition ends at 7.
        assert_eq!(physical_floor(&unchanged, &value, 7), 7);
    }
}