reddb_server/storage/unified/store/

impl_file.rs

use super::*;

impl UnifiedStore {
    pub fn new() -> Self {
        Self::with_config(UnifiedStoreConfig::default())
    }

    /// Get the current storage format version
    pub fn format_version(&self) -> u32 {
        self.format_version.load(Ordering::SeqCst)
    }

    pub(crate) fn set_format_version(&self, version: u32) {
        self.format_version.store(version, Ordering::SeqCst);
    }

    /// Allocate a global entity ID
    pub fn next_entity_id(&self) -> EntityId {
        self.resident.next_entity_id()
    }

    /// Reserve `n` contiguous global entity IDs with one fetch_add.
    /// Caller assigns `id = EntityId::new(start + i)` per entity.
    pub fn reserve_entity_ids(&self, n: u64) -> std::ops::Range<u64> {
        self.resident.reserve_entity_ids(n)
    }

    pub(crate) fn register_entity_id(&self, id: EntityId) {
        self.resident.register_entity_id(id)
    }

    /// Load store from binary file
    ///
    /// Binary dump framing is defined by `reddb_file::native_store`.
    ///
    /// Store payload shape:
    /// ```text
    /// [collection_count: varu32]
    /// [collections...]
    /// [cross_ref_count: varu32]
    /// [cross_refs...]
    /// ```
    pub fn load_from_file(path: &Path) -> Result<Self, Box<dyn std::error::Error>> {
        let file = File::open(path)?;
        let mut reader = BufReader::new(file);
        let mut buf = Vec::new();
        reader.read_to_end(&mut buf)?;
        Self::load_from_bytes(&buf)
    }

    pub(crate) fn load_from_bytes(buf: &[u8]) -> Result<Self, Box<dyn std::error::Error>> {
        Self::load_from_bytes_with_config(buf, UnifiedStoreConfig::default())
    }

    pub(crate) fn load_from_bytes_with_config(
        buf: &[u8],
        config: UnifiedStoreConfig,
    ) -> Result<Self, Box<dyn std::error::Error>> {
        let (resident, version) =
            reddb_engine::storage::resident_store::ResidentStore::from_snapshot(
                buf,
                &config.manager_config,
                config.auto_index_refs,
                config.max_cross_refs,
            )?;
        let mut store = Self::with_config(config);
        store.resident = resident;
        store.note_result_cache_snapshot_version(version);
        store.set_format_version(version.max(STORE_VERSION_CURRENT));
        store.load_result_cache_epoch()?;

        Ok(store)
    }

    /// Save store to binary file
    ///
    /// Uses compact binary encoding with varint for efficient storage.
    /// No JSON - pure binary with pages and indices.
    pub fn save_to_file(&self, path: &Path) -> Result<(), Box<dyn std::error::Error>> {
        let buf = self.to_binary_dump_bytes();
        reddb_file::write_native_store_bytes_atomically(path, &buf)?;
        Ok(())
    }

    pub(crate) fn to_binary_dump_bytes(&self) -> Vec<u8> {
        let _vault_guard = self.vault_publication_lock.lock();
        let pending_vault_versions = self.pending_vault_versions.read();
        let (_, excluded) = self.snapshot_manager.get().map_or_else(
            || (0, Default::default()),
            |manager| manager.checkpoint_excluded_state(),
        );
        self.resident
            .snapshot_bytes(&pending_vault_versions, &excluded, || {
                self.set_format_version(STORE_VERSION_CURRENT);
            })
    }

    /// Read entity using the canonical native/browser logical codec.
    pub(crate) fn read_entity_binary(
        buf: &[u8],
        pos: &mut usize,
        format_version: u32,
    ) -> Result<UnifiedEntity, Box<dyn std::error::Error>> {
        Ok(reddb_engine::storage::entity_codec::read_entity_binary(
            buf,
            pos,
            format_version,
        )?)
    }

    /// Write entity using the canonical native/browser logical codec.
    pub(crate) fn write_entity_binary(
        buf: &mut Vec<u8>,
        entity: &UnifiedEntity,
        format_version: u32,
    ) {
        reddb_engine::storage::entity_codec::write_entity_binary(buf, entity, format_version);
    }
}

#[cfg(test)]
mod aux_metadata_dump_tests {
    use super::*;
    use std::collections::HashMap;

    #[test]
    fn snapshot_load_preserves_existing_wal_and_restores_future_appends() {
        let directory = tempfile::tempdir().expect("directory");
        let path = directory.path().join("snapshot.rdb");
        let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
        let old =
            UnifiedEntity::table_row(EntityId::new(1), "rows", 1, vec![Value::text("snapshot")]);
        store.insert_auto("rows", old).expect("snapshot row");
        let snapshot = store.to_binary_dump_bytes();
        crate::storage::EmbeddedRdbArtifact::create_with_snapshot(&path, &snapshot)
            .expect("artifact");
        let newer = UnifiedEntity::table_row(
            EntityId::new(1),
            "rows",
            1,
            vec![Value::text("durable WAL")],
        );
        let action = StoreWalAction::upsert_entity("rows", &newer, None, STORE_VERSION_CURRENT);
        UnifiedStore::with_config(UnifiedStoreConfig::default().with_embedded_wal_path(&path))
            .finish_paged_write([action])
            .expect("durable update after snapshot");
        let before = std::fs::read(&path).expect("original artifact bytes");
        let loaded = UnifiedStore::load_from_bytes_with_config(
            &snapshot,
            UnifiedStoreConfig::default().with_embedded_wal_path(&path),
        )
        .expect("reconstruct snapshot");
        assert_eq!(std::fs::read(&path).expect("artifact after load"), before);
        assert_eq!(
            loaded.config().embedded_wal_path.as_deref(),
            Some(path.as_path())
        );

        let generation = crate::storage::EmbeddedRdbArtifact::open(&path)
            .expect("boundary before future write")
            .selected_superblock
            .generation;
        loaded
            .insert_auto(
                "rows",
                UnifiedEntity::table_row(
                    EntityId::new(2),
                    "rows",
                    2,
                    vec![Value::text("future write")],
                ),
            )
            .expect("future writes still append WAL");
        assert_eq!(
            crate::storage::EmbeddedRdbArtifact::open(&path)
                .expect("boundary after future write")
                .selected_superblock
                .generation,
            generation + 1
        );
    }

    #[test]
    fn aux_metadata_round_trips_through_binary_dump() {
        let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
        store.create_collection("k").expect("create collection");
        store.set_aux_metadata(b"contract-blob".to_vec());

        let bytes = store.to_binary_dump_bytes();
        let reloaded = UnifiedStore::load_from_bytes(&bytes).expect("reload dump");

        assert_eq!(reloaded.aux_metadata(), b"contract-blob".to_vec());
    }

    #[test]
    fn empty_aux_metadata_round_trips_as_empty() {
        let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
        let bytes = store.to_binary_dump_bytes();
        let reloaded = UnifiedStore::load_from_bytes(&bytes).expect("reload dump");

        assert!(reloaded.aux_metadata().is_empty());
    }

    #[test]
    fn legacy_system_collection_snapshot_loads_and_republishes_canonical_name() {
        let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
        store
            .insert_auto(
                "red_config",
                UnifiedEntity::table_row(
                    EntityId::new(1),
                    "red_config",
                    1,
                    vec![Value::text("preserved")],
                ),
            )
            .expect("config row");
        let mut legacy = store.to_binary_dump_bytes();
        let canonical = b"red.config";
        let old = b"red_config";
        let mut replacements = 0;
        for offset in 0..=legacy.len() - canonical.len() {
            if &legacy[offset..offset + canonical.len()] == canonical {
                legacy[offset..offset + old.len()].copy_from_slice(old);
                replacements += 1;
            }
        }
        assert!(
            replacements >= 2,
            "snapshot carries catalog and entity names"
        );
        legacy.truncate(legacy.len() - 4);
        reddb_file::append_native_store_crc32_footer(&mut legacy);

        let migrated = UnifiedStore::load_from_bytes(&legacy).expect("legacy snapshot");
        assert_eq!(migrated.list_collections(), vec!["red.config".to_string()]);
        assert_eq!(
            migrated
                .get("red.config", EntityId::new(1))
                .expect("preserved row")
                .kind
                .collection(),
            "red.config"
        );
        let republished = migrated.to_binary_dump_bytes();
        assert!(
            republished.windows(old.len()).all(|window| window != old),
            "new snapshots must not republish legacy identities"
        );
        assert!(republished
            .windows(canonical.len())
            .any(|window| window == canonical));
    }

    #[test]
    fn interned_timeseries_series_reduces_repeated_tag_storage() {
        let tags = HashMap::from([
            (
                "host".to_string(),
                "srv-very-long-hostname-0001".to_string(),
            ),
            ("region".to_string(), "us-east-1".to_string()),
            ("service".to_string(), "checkout-api".to_string()),
        ]);
        let inline_bytes: usize = (0..100)
            .map(|timestamp_ns| {
                let entity = UnifiedEntity::new(
                    EntityId::new(0),
                    EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
                        series: "cpu_metrics".to_string(),
                        metric: "cpu.idle".to_string(),
                    })),
                    EntityData::TimeSeries(crate::storage::TimeSeriesData {
                        metric: "cpu.idle".to_string(),
                        series_id: None,
                        timestamp_ns,
                        value: 94.8,
                        tags: tags.clone(),
                        fields: HashMap::new(),
                    }),
                );
                UnifiedStore::serialize_entity_record(&entity, None, reddb_file::STORE_VERSION_V10)
                    .len()
            })
            .sum();

        let mut dictionary_fields = HashMap::new();
        dictionary_fields.insert("collection".to_string(), Value::text("cpu_metrics"));
        dictionary_fields.insert("series_id".to_string(), Value::UnsignedInteger(0));
        dictionary_fields.insert("metric".to_string(), Value::text("cpu.idle"));
        dictionary_fields.insert(
            "canonical_tags".to_string(),
            Value::text(
                r#"{"host":"srv-very-long-hostname-0001","region":"us-east-1","service":"checkout-api"}"#,
            ),
        );
        dictionary_fields.insert(
            "tags".to_string(),
            Value::Json(
                br#"{"host":"srv-very-long-hostname-0001","region":"us-east-1","service":"checkout-api"}"#
                    .to_vec(),
            ),
        );
        let dictionary = UnifiedEntity::new(
            EntityId::new(0),
            EntityKind::TableRow {
                table: Arc::from("red_timeseries_series"),
                row_id: 0,
            },
            EntityData::Row(RowData {
                columns: Vec::new(),
                named: Some(dictionary_fields),
                schema: None,
            }),
        );
        let dictionary_bytes =
            UnifiedStore::serialize_entity_record(&dictionary, None, STORE_VERSION_V11).len();
        let interned_bytes = dictionary_bytes
            + (0..100)
                .map(|timestamp_ns| {
                    let entity = UnifiedEntity::new(
                        EntityId::new(0),
                        EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
                            series: "cpu_metrics".to_string(),
                            metric: "cpu.idle".to_string(),
                        })),
                        EntityData::TimeSeries(crate::storage::TimeSeriesData {
                            metric: "cpu.idle".to_string(),
                            series_id: Some(0),
                            timestamp_ns,
                            value: 94.8,
                            tags: HashMap::new(),
                            fields: HashMap::new(),
                        }),
                    );
                    UnifiedStore::serialize_entity_record(&entity, None, STORE_VERSION_V11).len()
                })
                .sum::<usize>();

        assert!(
            interned_bytes < inline_bytes,
            "interned series storage should shrink repeated tags: inline={inline_bytes}, interned={interned_bytes}"
        );
    }
}