hashtree_lmdb/pool/
read_only.rs

1use super::member::{open_member_reader, verify_member_path};
2use super::{
3    decode_manifest, map_heed, validate_controlled_manifest, LocationRecord, PoolCatalogLocation,
4    PoolManifest, PoolMemberConfig, PoolMemberId, PoolMemberState, PoolStoreConfig,
5    CATALOG_DATABASES, CATALOG_MAX_READERS, EXTERNAL_MARKER_NAME, MANIFEST_KEY, MEMBER_MARKER_NAME,
6};
7use crate::{managed_env::ManagedEnv, ExternalBlobOptions, LmdbBlobReader};
8use async_trait::async_trait;
9use hashtree_core::store::{slice_blob_range, Store, StoreError, StoreStats};
10use hashtree_core::{sha256, to_hex, types::Hash};
11use heed::types::Bytes;
12use heed::{Database, EnvFlags};
13use sha2::{Digest, Sha256};
14use std::collections::HashMap;
15use std::path::Path;
16use std::sync::Arc;
17
18struct ReadOnlyPoolStoreInner {
19    // Keep the catalog environment alive for the named database handle.
20    _env: ManagedEnv,
21    manifest: Database<Bytes, Bytes>,
22    opened_manifest_bytes: Vec<u8>,
23    opened_manifest_sha256: String,
24    locations: Database<Bytes, Bytes>,
25    members: HashMap<PoolMemberId, Arc<LmdbBlobReader>>,
26    manifest_snapshot: ReadOnlyPoolManifestSnapshot,
27}
28
29/// Strictly read-only view of an existing PoolStore catalog and its members.
30///
31/// Unlike [`super::PoolStore::open`], this constructor never creates a path,
32/// opens a write transaction, repairs pending records, records temperature
33/// samples, or starts a worker. It is intended for live-data verification.
34/// LMDB can still update reader bookkeeping in `lock.mdb`; `data.mdb` remains
35/// read-only. On Windows, catalog growth requires reopening the reader after
36/// a `MapResized` error because its mapping covers the file size at open.
37#[derive(Clone)]
38pub struct ReadOnlyPoolStore {
39    inner: Arc<ReadOnlyPoolStoreInner>,
40}
41
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct ReadOnlyPoolCatalogAudit {
44    pub stored_locations: u64,
45    pub sha256: String,
46    pub manifest_sha256: String,
47}
48
49/// One exact member record from a read-only Pool manifest snapshot.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct ReadOnlyPoolManifestMember {
52    pub id: PoolMemberId,
53    pub state: PoolMemberState,
54    pub config: PoolMemberConfig,
55}
56
57/// The exact stored Pool manifest observed while opening a read-only reader.
58///
59/// `sha256` covers the stored manifest bytes, including generation, member
60/// ordering, member states, and every member configuration field.
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub struct ReadOnlyPoolManifestSnapshot {
63    pub generation: u64,
64    pub sha256: Hash,
65    pub members: Vec<ReadOnlyPoolManifestMember>,
66}
67
68fn require_opened_manifest(
69    current: &[u8],
70    opened: &[u8],
71    opened_sha256: &str,
72) -> Result<(), StoreError> {
73    if current != opened {
74        return Err(StoreError::Other(format!(
75            "pool manifest changed after read-only members were opened: expected {opened_sha256}, found {}",
76            to_hex(&sha256(current))
77        )));
78    }
79    Ok(())
80}
81
82impl ReadOnlyPoolStore {
83    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, StoreError> {
84        Self::open_inner(path.as_ref(), None)
85    }
86
87    /// Open an exact, read-only Pool authority captured by a controller.
88    ///
89    /// A controlled open requires a pinned catalog LMDB identity, the exact
90    /// manifest SHA-256, one exact runtime binding for every manifest member,
91    /// and disabled temperature tracking. Member readers use only those
92    /// runtime paths and pinned LMDB identities.
93    pub fn open_controlled<P: AsRef<Path>>(
94        path: P,
95        config: PoolStoreConfig,
96    ) -> Result<Self, StoreError> {
97        if config.temperature.enabled {
98            return Err(StoreError::Other(
99                "controlled read-only Pool open requires temperature tracking to be disabled"
100                    .into(),
101            ));
102        }
103        if config.catalog_lmdb_identity.is_none()
104            || config.expected_manifest_sha256.is_none()
105            || config.member_runtime_paths.is_empty()
106        {
107            return Err(StoreError::Other(
108                "controlled read-only Pool open requires catalog identity, exact manifest SHA-256, and every member runtime binding".into(),
109            ));
110        }
111        Self::open_inner(path.as_ref(), Some(config))
112    }
113
114    fn open_inner(path: &Path, config: Option<PoolStoreConfig>) -> Result<Self, StoreError> {
115        if !path.join("data.mdb").is_file() {
116            return Err(StoreError::Other(format!(
117                "read-only pool catalog is missing at {}",
118                path.display()
119            )));
120        }
121
122        let mut options = ManagedEnv::read_only_options(path).map_err(map_heed)?;
123        options
124            .max_dbs(CATALOG_DATABASES)
125            .max_readers(CATALOG_MAX_READERS);
126        unsafe {
127            options.flags(EnvFlags::NO_READ_AHEAD);
128        }
129        let env = unsafe {
130            match config
131                .as_ref()
132                .and_then(|config| config.catalog_lmdb_identity)
133            {
134                Some(identity) => ManagedEnv::open_pinned(&options, path, identity),
135                None => ManagedEnv::open(&options, path),
136            }
137        }
138        .map_err(|error| {
139            StoreError::Other(format!(
140                "open read-only pool catalog {}: {error}",
141                path.display()
142            ))
143        })?;
144        let rtxn = env.read_txn().map_err(map_heed)?;
145        let manifest_db = env
146            .open_database::<Bytes, Bytes>(&rtxn, Some("manifest"))
147            .map_err(map_heed)?
148            .ok_or_else(|| StoreError::Other("pool manifest database is missing".into()))?;
149        let locations = env
150            .open_database::<Bytes, Bytes>(&rtxn, Some("locations"))
151            .map_err(map_heed)?
152            .ok_or_else(|| StoreError::Other("pool locations database is missing".into()))?;
153        let manifest_bytes = manifest_db
154            .get(&rtxn, MANIFEST_KEY)
155            .map_err(map_heed)?
156            .ok_or_else(|| StoreError::Other("pool manifest is missing".into()))?;
157        let opened_manifest_bytes = manifest_bytes.to_vec();
158        let opened_manifest_sha256 = to_hex(&sha256(&opened_manifest_bytes));
159        let mut member_runtime_paths = config
160            .as_ref()
161            .map(|config| {
162                config
163                    .member_runtime_paths
164                    .iter()
165                    .map(|binding| (binding.id, binding.clone()))
166                    .collect::<HashMap<_, _>>()
167            })
168            .unwrap_or_default();
169        if config
170            .as_ref()
171            .is_some_and(|config| member_runtime_paths.len() != config.member_runtime_paths.len())
172        {
173            return Err(StoreError::Other(
174                "duplicate runtime path binding for controlled read-only Pool open".into(),
175            ));
176        }
177        let manifest = match config
178            .as_ref()
179            .and_then(|config| config.expected_manifest_sha256)
180        {
181            Some(expected) => validate_controlled_manifest(
182                &opened_manifest_bytes,
183                expected,
184                &member_runtime_paths,
185            )?,
186            None => decode_manifest(&opened_manifest_bytes)?,
187        };
188        let manifest_snapshot = ReadOnlyPoolManifestSnapshot {
189            generation: manifest.generation,
190            sha256: sha256(&opened_manifest_bytes),
191            members: manifest
192                .members
193                .iter()
194                .map(|member| ReadOnlyPoolManifestMember {
195                    id: member.id,
196                    state: member.state,
197                    config: member.config.clone(),
198                })
199                .collect(),
200        };
201        // Publish DBI handles before later transactions use them.
202        rtxn.commit().map_err(map_heed)?;
203
204        let members = if config.is_some() {
205            open_controlled_members(&manifest, &mut member_runtime_paths)?
206        } else {
207            open_members(&manifest)?
208        };
209        Ok(Self {
210            inner: Arc::new(ReadOnlyPoolStoreInner {
211                _env: env,
212                manifest: manifest_db,
213                opened_manifest_bytes,
214                opened_manifest_sha256,
215                locations,
216                members,
217                manifest_snapshot,
218            }),
219        })
220    }
221
222    /// Return the exact manifest snapshot captured by this reader.
223    ///
224    /// This first proves that the live manifest still matches the opened
225    /// bytes, so callers cannot accidentally pin a stale topology.
226    pub fn manifest_snapshot(&self) -> Result<ReadOnlyPoolManifestSnapshot, StoreError> {
227        let rtxn = self.inner._env.read_txn().map_err(map_heed)?;
228        let manifest = self
229            .inner
230            .manifest
231            .get(&rtxn, MANIFEST_KEY)
232            .map_err(map_heed)?
233            .ok_or_else(|| StoreError::Other("pool manifest is missing".into()))?;
234        require_opened_manifest(
235            manifest,
236            &self.inner.opened_manifest_bytes,
237            &self.inner.opened_manifest_sha256,
238        )?;
239        decode_manifest(manifest)?;
240        Ok(self.inner.manifest_snapshot.clone())
241    }
242
243    /// Return exact catalog state for many hashes in one read transaction.
244    ///
245    /// `Pending` and `Moving` are deliberately preserved so crash recovery
246    /// cannot mistake physical bytes for committed catalog residency.
247    pub fn blob_catalog_locations(
248        &self,
249        hashes: &[Hash],
250    ) -> Result<Vec<PoolCatalogLocation>, StoreError> {
251        let rtxn = self.inner._env.read_txn().map_err(map_heed)?;
252        let manifest = self
253            .inner
254            .manifest
255            .get(&rtxn, MANIFEST_KEY)
256            .map_err(map_heed)?
257            .ok_or_else(|| StoreError::Other("pool manifest is missing".into()))?;
258        require_opened_manifest(
259            manifest,
260            &self.inner.opened_manifest_bytes,
261            &self.inner.opened_manifest_sha256,
262        )?;
263        hashes
264            .iter()
265            .map(|hash| {
266                let record = self
267                    .inner
268                    .locations
269                    .get(&rtxn, hash)
270                    .map_err(map_heed)?
271                    .map(LocationRecord::decode)
272                    .transpose()?;
273                Ok(match record {
274                    None => PoolCatalogLocation::Missing,
275                    Some(LocationRecord::Pending { member, size }) => {
276                        PoolCatalogLocation::Pending { member, size }
277                    }
278                    Some(LocationRecord::Stored { member, size }) => {
279                        PoolCatalogLocation::Stored { member, size }
280                    }
281                    Some(LocationRecord::Moving {
282                        source,
283                        target,
284                        size,
285                    }) => PoolCatalogLocation::Moving {
286                        source,
287                        target,
288                        size,
289                    },
290                })
291            })
292            .collect()
293    }
294
295    fn read_location(&self, hash: &Hash) -> Result<Option<LocationRecord>, StoreError> {
296        let rtxn = self.inner._env.read_txn().map_err(map_heed)?;
297        self.inner
298            .locations
299            .get(&rtxn, hash)
300            .map_err(map_heed)?
301            .map(LocationRecord::decode)
302            .transpose()
303    }
304
305    /// Prove that the entire catalog contains only committed locations.
306    ///
307    /// Terminal verification must not silently read through an interrupted
308    /// pending write or member move, even when one of its copies happens to
309    /// contain valid bytes.
310    pub fn validate_committed_catalog(&self) -> Result<ReadOnlyPoolCatalogAudit, StoreError> {
311        let rtxn = self.inner._env.read_txn().map_err(map_heed)?;
312        let manifest = self
313            .inner
314            .manifest
315            .get(&rtxn, MANIFEST_KEY)
316            .map_err(map_heed)?
317            .ok_or_else(|| StoreError::Other("pool manifest is missing".into()))?;
318        require_opened_manifest(
319            manifest,
320            &self.inner.opened_manifest_bytes,
321            &self.inner.opened_manifest_sha256,
322        )?;
323        // Decode again so corruption cannot be hidden by byte equality alone.
324        decode_manifest(manifest)?;
325        let mut digest = Sha256::new();
326        digest.update(b"hashtree-read-only-pool-catalog-v1\0");
327        digest.update((manifest.len() as u64).to_be_bytes());
328        digest.update(manifest);
329        let mut stored = 0u64;
330        for item in self.inner.locations.iter(&rtxn).map_err(map_heed)? {
331            let (hash, encoded) = item.map_err(map_heed)?;
332            let hash: Hash = hash
333                .try_into()
334                .map_err(|_| StoreError::Other("invalid pool catalog hash key".into()))?;
335            let location = LocationRecord::decode(encoded)?;
336            match location {
337                LocationRecord::Stored { member, .. } => {
338                    if !self.inner.members.contains_key(&member) {
339                        return Err(StoreError::Other(format!(
340                            "pool catalog location {} references unknown member {member}",
341                            to_hex(&hash)
342                        )));
343                    }
344                    stored = stored.checked_add(1).ok_or_else(|| {
345                        StoreError::Other("pool catalog location count overflow".into())
346                    })?;
347                }
348                LocationRecord::Pending { .. } => {
349                    return Err(StoreError::Other(format!(
350                        "pool catalog contains pending location {}",
351                        to_hex(&hash)
352                    )));
353                }
354                LocationRecord::Moving { .. } => {
355                    return Err(StoreError::Other(format!(
356                        "pool catalog contains moving location {}",
357                        to_hex(&hash)
358                    )));
359                }
360            }
361            digest.update(hash);
362            digest.update((encoded.len() as u64).to_be_bytes());
363            digest.update(encoded);
364        }
365        let digest: Hash = digest.finalize().into();
366        Ok(ReadOnlyPoolCatalogAudit {
367            stored_locations: stored,
368            sha256: to_hex(&digest),
369            manifest_sha256: self.inner.opened_manifest_sha256.clone(),
370        })
371    }
372
373    /// Require every member that stores external blobs to fsync those files
374    /// before committing its catalog location. LMDB environment sync alone
375    /// cannot make an external file durable when this manifest flag is false.
376    pub fn require_durable_external_blob_writes(&self) -> Result<(), StoreError> {
377        let manifest = decode_manifest(&self.inner.opened_manifest_bytes)?;
378        for member in manifest.members {
379            if member.config.external_blob_dir.is_some() && !member.config.external_blob_sync {
380                return Err(StoreError::Other(format!(
381                    "pool member {} has external_blob_sync disabled",
382                    member.id
383                )));
384            }
385        }
386        Ok(())
387    }
388
389    pub fn get_sync(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
390        let Some(location) = self.read_location(hash)? else {
391            return Ok(None);
392        };
393        let id = match location {
394            LocationRecord::Stored { member, .. } => member,
395            LocationRecord::Pending { .. } => {
396                return Err(StoreError::Other(format!(
397                    "read-only terminal verification rejected pending pool location {}",
398                    to_hex(hash)
399                )));
400            }
401            LocationRecord::Moving { .. } => {
402                return Err(StoreError::Other(format!(
403                    "read-only terminal verification rejected moving pool location {}",
404                    to_hex(hash)
405                )));
406            }
407        };
408        let member = self.inner.members.get(&id).ok_or_else(|| {
409            StoreError::Other(format!("pool location references unknown member {id}"))
410        })?;
411        match member.get_sync(hash)? {
412            Some(data) if sha256(&data) == *hash && data.len() as u64 == location.size() => {
413                Ok(Some(data))
414            }
415            Some(_) => Err(StoreError::Other(format!(
416                "read-only pool member {id} returned corrupt or size-mismatched bytes"
417            ))),
418            None => Err(StoreError::Other(format!(
419                "pool catalog location {} references missing bytes on member {id}",
420                to_hex(hash)
421            ))),
422        }
423    }
424
425    pub fn blob_size_sync(&self, hash: &Hash) -> Result<Option<u64>, StoreError> {
426        let Some(location) = self.read_location(hash)? else {
427            return Ok(None);
428        };
429        match location {
430            LocationRecord::Stored { member, size } => {
431                let member_store = self.inner.members.get(&member).ok_or_else(|| {
432                    StoreError::Other(format!("pool location references unknown member {member}"))
433                })?;
434                match member_store.blob_size_sync(hash)? {
435                    Some(member_size) if member_size == size => Ok(Some(size)),
436                    Some(member_size) => Err(StoreError::Other(format!(
437                        "pool catalog size {size} for {} differs from member {member} size {member_size}",
438                        to_hex(hash)
439                    ))),
440                    None => Err(StoreError::Other(format!(
441                        "pool catalog location {} references missing bytes on member {member}",
442                        to_hex(hash)
443                    ))),
444                }
445            }
446            LocationRecord::Pending { .. } => Err(StoreError::Other(format!(
447                "read-only terminal verification rejected pending pool location {}",
448                to_hex(hash)
449            ))),
450            LocationRecord::Moving { .. } => Err(StoreError::Other(format!(
451                "read-only terminal verification rejected moving pool location {}",
452                to_hex(hash)
453            ))),
454        }
455    }
456
457    pub fn get_range_sync(
458        &self,
459        hash: &Hash,
460        start: u64,
461        end_inclusive: u64,
462    ) -> Result<Option<Vec<u8>>, StoreError> {
463        self.get_sync(hash)?
464            .map(|data| slice_blob_range(&data, start, end_inclusive))
465            .transpose()
466    }
467}
468
469fn open_members(
470    manifest: &PoolManifest,
471) -> Result<HashMap<PoolMemberId, Arc<LmdbBlobReader>>, StoreError> {
472    let mut stores = HashMap::with_capacity(manifest.members.len());
473    for member in &manifest.members {
474        let config = &member.config;
475        verify_member_path(&config.path, MEMBER_MARKER_NAME, member.id)?;
476        let external = external_options(member.id, config)?;
477        if stores
478            .insert(
479                member.id,
480                Arc::new(LmdbBlobReader::open(&config.path, external)?),
481            )
482            .is_some()
483        {
484            return Err(StoreError::Other(format!(
485                "pool manifest repeats member {}",
486                member.id
487            )));
488        }
489    }
490    Ok(stores)
491}
492
493fn open_controlled_members(
494    manifest: &PoolManifest,
495    bindings: &mut HashMap<PoolMemberId, super::PoolMemberRuntimePaths>,
496) -> Result<HashMap<PoolMemberId, Arc<LmdbBlobReader>>, StoreError> {
497    let mut stores = HashMap::with_capacity(manifest.members.len());
498    for member in &manifest.members {
499        let binding = bindings.remove(&member.id).ok_or_else(|| {
500            StoreError::Other(format!(
501                "controlled read-only Pool open has no runtime binding for member {}",
502                member.id
503            ))
504        })?;
505        let mut runtime_config = member.config.clone();
506        runtime_config.path = binding.runtime_path;
507        runtime_config.external_blob_dir = binding.runtime_external_path;
508        let reader = open_member_reader(
509            member.id,
510            &runtime_config,
511            Some(binding.lmdb_identity),
512            false,
513            1,
514        )?;
515        if stores.insert(member.id, Arc::new(reader)).is_some() {
516            return Err(StoreError::Other(format!(
517                "pool manifest repeats member {}",
518                member.id
519            )));
520        }
521    }
522    if let Some(unexpected) = bindings.keys().next() {
523        return Err(StoreError::Other(format!(
524            "controlled read-only Pool open has runtime binding for unknown member {unexpected}"
525        )));
526    }
527    Ok(stores)
528}
529
530fn external_options(
531    id: PoolMemberId,
532    config: &PoolMemberConfig,
533) -> Result<Option<ExternalBlobOptions>, StoreError> {
534    match (
535        config.external_blob_dir.as_ref(),
536        config.external_blob_min_bytes,
537    ) {
538        (Some(path), Some(min_bytes)) => {
539            verify_member_path(path, EXTERNAL_MARKER_NAME, id)?;
540            Ok(Some(ExternalBlobOptions {
541                base_path: path.clone(),
542                min_bytes: usize::try_from(min_bytes).map_err(|_| {
543                    StoreError::Other("pool external blob threshold exceeds usize".into())
544                })?,
545                sync: config.external_blob_sync,
546                pack_target_bytes: config
547                    .external_pack_target_bytes
548                    .map(usize::try_from)
549                    .transpose()
550                    .map_err(|_| {
551                        StoreError::Other("pool external pack target exceeds usize".into())
552                    })?,
553            }))
554        }
555        (None, None) => Ok(None),
556        _ => Err(StoreError::Other(
557            "invalid pool external blob configuration".into(),
558        )),
559    }
560}
561
562fn read_only_error() -> StoreError {
563    StoreError::Other("read-only PoolStore does not permit mutation".into())
564}
565
566#[async_trait]
567impl Store for ReadOnlyPoolStore {
568    async fn put(&self, _hash: Hash, _data: Vec<u8>) -> Result<bool, StoreError> {
569        Err(read_only_error())
570    }
571
572    async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
573        self.get_sync(hash)
574    }
575
576    async fn get_range(
577        &self,
578        hash: &Hash,
579        start: u64,
580        end_inclusive: u64,
581    ) -> Result<Option<Vec<u8>>, StoreError> {
582        self.get_sync(hash)?
583            .map(|data| slice_blob_range(&data, start, end_inclusive))
584            .transpose()
585    }
586
587    async fn blob_size(&self, hash: &Hash) -> Result<Option<u64>, StoreError> {
588        self.blob_size_sync(hash)
589    }
590
591    async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
592        Ok(self.get_sync(hash)?.is_some())
593    }
594
595    async fn delete(&self, _hash: &Hash) -> Result<bool, StoreError> {
596        Err(read_only_error())
597    }
598
599    async fn stats(&self) -> StoreStats {
600        StoreStats::default()
601    }
602
603    async fn pin(&self, _hash: &Hash) -> Result<(), StoreError> {
604        Err(read_only_error())
605    }
606
607    async fn unpin(&self, _hash: &Hash) -> Result<(), StoreError> {
608        Err(read_only_error())
609    }
610}
611
612#[cfg(test)]
613mod tests {
614    use super::*;
615    use crate::{
616        PinnedLmdbFileIdentity, PinnedLmdbIdentity, PoolMemberRuntimePaths, PoolStore,
617        PoolStoreConfig,
618    };
619    use std::fs;
620    #[cfg(unix)]
621    use std::fs::File;
622    #[cfg(target_os = "linux")]
623    use std::os::fd::AsRawFd;
624    #[cfg(unix)]
625    use std::os::unix::fs::MetadataExt;
626    use std::process::Command;
627    use tempfile::TempDir;
628
629    const MUTATE_CATALOG_ENV: &str = "HASHTREE_READ_ONLY_MUTATE_CATALOG";
630    const MUTATE_MEMBER_ENV: &str = "HASHTREE_READ_ONLY_MUTATE_MEMBER";
631
632    fn file_sha256(path: &Path) -> Hash {
633        sha256(&fs::read(path).expect("read LMDB data file"))
634    }
635
636    fn file_modified(path: &Path) -> std::time::SystemTime {
637        fs::metadata(path)
638            .expect("read LMDB data file metadata")
639            .modified()
640            .expect("read LMDB data file mtime")
641    }
642
643    fn pool_config() -> PoolStoreConfig {
644        let mut config = PoolStoreConfig::default();
645        config.temperature.enabled = false;
646        config
647    }
648
649    #[cfg(unix)]
650    fn lmdb_identity(path: &Path) -> PinnedLmdbIdentity {
651        let data = fs::metadata(path.join("data.mdb")).expect("data.mdb metadata");
652        let lock = fs::metadata(path.join("lock.mdb")).expect("lock.mdb metadata");
653        PinnedLmdbIdentity {
654            data: PinnedLmdbFileIdentity {
655                device: data.dev(),
656                inode: data.ino(),
657            },
658            lock: PinnedLmdbFileIdentity {
659                device: lock.dev(),
660                inode: lock.ino(),
661            },
662        }
663    }
664
665    #[cfg(unix)]
666    struct PinnedDirectory {
667        _directory: File,
668        runtime_path: std::path::PathBuf,
669    }
670
671    #[cfg(unix)]
672    fn pin_directory(path: &Path) -> PinnedDirectory {
673        let directory = File::open(path).expect("open directory");
674        #[cfg(target_os = "linux")]
675        let runtime_path =
676            std::path::PathBuf::from(format!("/proc/self/fd/{}", directory.as_raw_fd()));
677        #[cfg(not(target_os = "linux"))]
678        let runtime_path = path.to_path_buf();
679        PinnedDirectory {
680            _directory: directory,
681            runtime_path,
682        }
683    }
684
685    #[cfg(unix)]
686    #[test]
687    fn controlled_open_exposes_exact_manifest_and_catalog_states() {
688        let temp = TempDir::new().expect("temp dir");
689        let catalog = temp.path().join("catalog");
690        let source_path = temp.path().join("source");
691        let target_path = temp.path().join("target");
692        let source_config = PoolMemberConfig::new(source_path.clone(), 16 * 1024 * 1024);
693        let target_config = PoolMemberConfig::new(target_path.clone(), 16 * 1024 * 1024);
694        let pool = PoolStore::open(&catalog, pool_config()).expect("open pool");
695        let source = pool
696            .add_member(source_config.clone())
697            .expect("add source member");
698        let target = pool
699            .add_member(target_config.clone())
700            .expect("add target member");
701
702        let stored_data = b"controlled read-only stored bytes";
703        let stored_hash = sha256(stored_data);
704        pool.put_sync(stored_hash, stored_data)
705            .expect("put stored blob");
706        let stored_location = pool
707            .read_location(&stored_hash)
708            .expect("read stored location")
709            .expect("stored catalog record");
710        let missing_hash = sha256(b"controlled read-only missing");
711        let pending_hash = sha256(b"controlled read-only pending");
712        let moving_hash = sha256(b"controlled read-only moving");
713        let mut wtxn = pool.env.write_txn().expect("catalog write txn");
714        pool.set_location_txn(
715            &mut wtxn,
716            pending_hash,
717            Some(LocationRecord::Pending {
718                member: source,
719                size: 37,
720            }),
721        )
722        .expect("write pending");
723        pool.set_location_txn(
724            &mut wtxn,
725            moving_hash,
726            Some(LocationRecord::Moving {
727                source,
728                target,
729                size: 41,
730            }),
731        )
732        .expect("write moving");
733        wtxn.commit().expect("commit catalog states");
734        pool.force_sync().expect("sync pool");
735        drop(pool);
736
737        let ordinary =
738            ReadOnlyPoolStore::open(&catalog).expect("open ordinary read-only snapshot reader");
739        let expected_snapshot = ordinary.manifest_snapshot().expect("manifest snapshot");
740        assert_eq!(
741            expected_snapshot.sha256,
742            sha256(&ordinary.inner.opened_manifest_bytes)
743        );
744        assert_eq!(expected_snapshot.members.len(), 2);
745        assert_eq!(
746            expected_snapshot
747                .members
748                .iter()
749                .find(|member| member.id == source)
750                .expect("source snapshot")
751                .config,
752            source_config
753        );
754        assert_eq!(
755            expected_snapshot
756                .members
757                .iter()
758                .find(|member| member.id == target)
759                .expect("target snapshot")
760                .config,
761            target_config
762        );
763        drop(ordinary);
764
765        let catalog_identity = lmdb_identity(&catalog);
766        let source_identity = lmdb_identity(&source_path);
767        let target_identity = lmdb_identity(&target_path);
768        let catalog_pin = pin_directory(&catalog);
769        let source_pin = pin_directory(&source_path);
770        let target_pin = pin_directory(&target_path);
771        let mut controlled = pool_config();
772        controlled.catalog_lmdb_identity = Some(catalog_identity);
773        controlled.expected_manifest_sha256 = Some(expected_snapshot.sha256);
774        controlled.member_runtime_paths = vec![
775            PoolMemberRuntimePaths {
776                id: source,
777                configured_path: source_path,
778                runtime_path: source_pin.runtime_path.clone(),
779                configured_external_path: None,
780                runtime_external_path: None,
781                lmdb_identity: source_identity,
782            },
783            PoolMemberRuntimePaths {
784                id: target,
785                configured_path: target_path,
786                runtime_path: target_pin.runtime_path.clone(),
787                configured_external_path: None,
788                runtime_external_path: None,
789                lmdb_identity: target_identity,
790            },
791        ];
792
793        let mut enabled_temperature = controlled.clone();
794        enabled_temperature.temperature.enabled = true;
795        assert!(
796            ReadOnlyPoolStore::open_controlled(&catalog_pin.runtime_path, enabled_temperature)
797                .err()
798                .expect("temperature-enabled controlled open must fail")
799                .to_string()
800                .contains("temperature tracking")
801        );
802
803        let mut wrong_manifest = controlled.clone();
804        wrong_manifest.expected_manifest_sha256 = Some(sha256(b"wrong manifest authority"));
805        assert!(
806            ReadOnlyPoolStore::open_controlled(&catalog_pin.runtime_path, wrong_manifest)
807                .err()
808                .expect("wrong manifest authority must fail")
809                .to_string()
810                .contains("manifest SHA-256 differs")
811        );
812
813        let mut missing_binding = controlled.clone();
814        missing_binding.member_runtime_paths.pop();
815        assert!(
816            ReadOnlyPoolStore::open_controlled(&catalog_pin.runtime_path, missing_binding)
817                .err()
818                .expect("incomplete member authority must fail")
819                .to_string()
820                .contains("pinned pool topology")
821        );
822
823        let reader = ReadOnlyPoolStore::open_controlled(&catalog_pin.runtime_path, controlled)
824            .expect("open controlled read-only Pool");
825        assert_eq!(
826            reader.manifest_snapshot().expect("controlled snapshot"),
827            expected_snapshot
828        );
829        let expected_stored = match stored_location {
830            LocationRecord::Stored { member, size } => PoolCatalogLocation::Stored { member, size },
831            other => panic!("put committed unexpected location {other:?}"),
832        };
833        let locations = reader
834            .blob_catalog_locations(&[missing_hash, pending_hash, stored_hash, moving_hash])
835            .expect("read exact catalog states");
836        assert_eq!(
837            locations,
838            vec![
839                PoolCatalogLocation::Missing,
840                PoolCatalogLocation::Pending {
841                    member: source,
842                    size: 37,
843                },
844                expected_stored,
845                PoolCatalogLocation::Moving {
846                    source,
847                    target,
848                    size: 41,
849                },
850            ]
851        );
852        assert_eq!(
853            reader.get_sync(&stored_hash).expect("read stored blob"),
854            Some(stored_data.to_vec())
855        );
856    }
857
858    #[tokio::test]
859    async fn reads_real_pool_without_mutating_data_files() {
860        let temp = TempDir::new().expect("temp dir");
861        let catalog = temp.path().join("catalog");
862        let member = temp.path().join("member");
863        let pool = PoolStore::open(&catalog, pool_config()).expect("open pool");
864        pool.add_member(PoolMemberConfig::new(member.clone(), 1024 * 1024))
865            .expect("add member");
866        let data = b"read-only terminal audit bytes";
867        let hash = sha256(data);
868        pool.put_sync(hash, data).expect("put blob");
869        pool.force_sync().expect("sync pool");
870        drop(pool);
871
872        let catalog_data = catalog.join("data.mdb");
873        let member_data = member.join("data.mdb");
874        let catalog_before = file_sha256(&catalog_data);
875        let member_before = file_sha256(&member_data);
876        let catalog_mtime_before = file_modified(&catalog_data);
877        let member_mtime_before = file_modified(&member_data);
878        let reader = ReadOnlyPoolStore::open(&catalog).expect("open read-only pool");
879        reader
880            .require_durable_external_blob_writes()
881            .expect("ordinary pool members use durable external writes");
882        let audit = reader
883            .validate_committed_catalog()
884            .expect("validate catalog");
885        assert_eq!(audit.stored_locations, 1);
886        assert_eq!(audit.manifest_sha256.len(), 64);
887        assert_eq!(
888            reader.get_sync(&hash).expect("get blob"),
889            Some(data.to_vec())
890        );
891        assert_eq!(
892            reader.blob_size_sync(&hash).expect("blob size"),
893            Some(data.len() as u64)
894        );
895        assert_eq!(
896            reader.get_range(&hash, 5, 8).await.expect("blob range"),
897            Some(data[5..=8].to_vec())
898        );
899        assert!(reader.has(&hash).await.expect("has blob"));
900        assert!(reader.put(hash, data.to_vec()).await.is_err());
901        assert!(reader.delete(&hash).await.is_err());
902        assert!(reader.pin(&hash).await.is_err());
903        assert!(reader.unpin(&hash).await.is_err());
904        drop(reader);
905
906        assert_eq!(file_sha256(&catalog_data), catalog_before);
907        assert_eq!(file_sha256(&member_data), member_before);
908        assert_eq!(file_modified(&catalog_data), catalog_mtime_before);
909        assert_eq!(file_modified(&member_data), member_mtime_before);
910    }
911
912    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
913    async fn production_pool_read_allows_nested_reader_on_tokio_worker() {
914        let temp = TempDir::new().expect("generated Pool root");
915        let catalog = temp.path().join("catalog");
916        let member = temp.path().join("member");
917        let pool = PoolStore::open(&catalog, pool_config()).expect("open generated Pool");
918        pool.add_member(PoolMemberConfig::new(member, 16 * 1024 * 1024))
919            .expect("add generated member");
920        let body = b"real nested Pool reader body";
921        let hash = sha256(body);
922        pool.put_sync(hash, body).expect("store generated body");
923        pool.force_sync().expect("sync generated Pool");
924        drop(pool);
925
926        let reader = ReadOnlyPoolStore::open(&catalog).expect("open production Pool reader");
927        tokio::task::block_in_place(|| {
928            let held = reader
929                .inner
930                ._env
931                .read_txn()
932                .expect("hold first catalog read transaction");
933            assert_eq!(
934                reader
935                    .get_sync(&hash)
936                    .expect("open nested production catalog transaction"),
937                Some(body.to_vec())
938            );
939            drop(held);
940        });
941    }
942
943    #[test]
944    fn durable_writer_check_rejects_unsynced_external_pool_members() {
945        let temp = TempDir::new().expect("temp dir");
946        let catalog = temp.path().join("catalog");
947        let member = temp.path().join("member");
948        let external = temp.path().join("external");
949        let pool = PoolStore::open(&catalog, pool_config()).expect("open pool");
950        pool.add_member(
951            PoolMemberConfig::new(member, 1024 * 1024)
952                .with_external_blobs(external, 1, false, None),
953        )
954        .expect("add unsynced external member");
955        pool.force_sync().expect("sync pool manifest");
956        drop(pool);
957
958        let reader = ReadOnlyPoolStore::open(&catalog).expect("open read-only pool");
959        let error = reader
960            .require_durable_external_blob_writes()
961            .expect_err("unsynced external member must fail durable writer validation");
962        assert!(error.to_string().contains("external_blob_sync disabled"));
963    }
964
965    #[test]
966    fn rejects_pending_and_moving_catalog_records() {
967        let temp = TempDir::new().expect("temp dir");
968        let catalog = temp.path().join("catalog");
969        let pool = PoolStore::open(&catalog, pool_config()).expect("open pool");
970        let source = pool
971            .add_member(PoolMemberConfig::new(
972                temp.path().join("source"),
973                1024 * 1024,
974            ))
975            .expect("add source");
976        let target = pool
977            .add_member(PoolMemberConfig::new(
978                temp.path().join("target"),
979                1024 * 1024,
980            ))
981            .expect("add target");
982        let pending_hash = sha256(b"pending audit record");
983        let moving_hash = sha256(b"moving audit record");
984        let mut wtxn = pool.env.write_txn().expect("catalog write txn");
985        pool.set_location_txn(
986            &mut wtxn,
987            pending_hash,
988            Some(LocationRecord::Pending {
989                member: source,
990                size: 20,
991            }),
992        )
993        .expect("write pending location");
994        pool.set_location_txn(
995            &mut wtxn,
996            moving_hash,
997            Some(LocationRecord::Moving {
998                source,
999                target,
1000                size: 19,
1001            }),
1002        )
1003        .expect("write moving location");
1004        wtxn.commit().expect("commit interrupted records");
1005        pool.force_sync().expect("sync catalog");
1006        drop(pool);
1007
1008        let reader = ReadOnlyPoolStore::open(&catalog).expect("open read-only pool");
1009        let error = reader
1010            .validate_committed_catalog()
1011            .expect_err("catalog with interrupted records must fail");
1012        assert!(
1013            error.to_string().contains("pending") || error.to_string().contains("moving"),
1014            "unexpected catalog error: {error}"
1015        );
1016        assert!(reader
1017            .get_sync(&pending_hash)
1018            .expect_err("pending read must fail")
1019            .to_string()
1020            .contains("pending"));
1021        assert!(reader
1022            .get_sync(&moving_hash)
1023            .expect_err("moving read must fail")
1024            .to_string()
1025            .contains("moving"));
1026        assert!(reader
1027            .blob_size_sync(&pending_hash)
1028            .expect_err("pending size must fail")
1029            .to_string()
1030            .contains("pending"));
1031        assert!(reader
1032            .blob_size_sync(&moving_hash)
1033            .expect_err("moving size must fail")
1034            .to_string()
1035            .contains("moving"));
1036    }
1037
1038    #[test]
1039    fn rejects_stored_location_with_missing_member_bytes() {
1040        let temp = TempDir::new().expect("temp dir");
1041        let catalog = temp.path().join("catalog");
1042        let member_path = temp.path().join("member");
1043        let pool = PoolStore::open(&catalog, pool_config()).expect("open pool");
1044        let member = pool
1045            .add_member(PoolMemberConfig::new(member_path, 1024 * 1024))
1046            .expect("add member");
1047        let data = b"catalogued but physically missing";
1048        let hash = sha256(data);
1049        pool.put_sync(hash, data).expect("put blob");
1050        let member_store = pool.get_member(member).expect("open member");
1051        assert!(member_store
1052            .delete_sync(&hash)
1053            .expect("delete only member bytes"));
1054        pool.force_sync().expect("sync corrupt physical state");
1055        drop(member_store);
1056        drop(pool);
1057
1058        let reader = ReadOnlyPoolStore::open(&catalog).expect("open read-only pool");
1059        let get_error = reader
1060            .get_sync(&hash)
1061            .expect_err("stored location without bytes must fail closed");
1062        assert!(get_error.to_string().contains(&to_hex(&hash)));
1063        assert!(get_error.to_string().contains(&member.to_string()));
1064        assert!(get_error.to_string().contains("missing bytes"));
1065        let size_error = reader
1066            .blob_size_sync(&hash)
1067            .expect_err("size must verify physical bytes");
1068        assert!(size_error.to_string().contains("missing bytes"));
1069    }
1070
1071    #[test]
1072    fn rejects_catalog_location_for_unknown_member() {
1073        let temp = TempDir::new().expect("temp dir");
1074        let catalog = temp.path().join("catalog");
1075        let pool = PoolStore::open(&catalog, pool_config()).expect("open pool");
1076        pool.add_member(PoolMemberConfig::new(
1077            temp.path().join("member"),
1078            1024 * 1024,
1079        ))
1080        .expect("add member");
1081        let hash = sha256(b"unknown member audit record");
1082        let unknown = PoolMemberId::new();
1083        let mut wtxn = pool.env.write_txn().expect("catalog write txn");
1084        pool.set_location_txn(
1085            &mut wtxn,
1086            hash,
1087            Some(LocationRecord::Stored {
1088                member: unknown,
1089                size: 27,
1090            }),
1091        )
1092        .expect("write unknown member location");
1093        wtxn.commit().expect("commit unknown member record");
1094        pool.force_sync().expect("sync catalog");
1095        drop(pool);
1096
1097        let reader = ReadOnlyPoolStore::open(&catalog).expect("open read-only pool");
1098        assert!(reader
1099            .validate_committed_catalog()
1100            .expect_err("unknown catalog member must fail")
1101            .to_string()
1102            .contains("unknown member"));
1103        assert!(reader
1104            .get_sync(&hash)
1105            .expect_err("unknown member read must fail")
1106            .to_string()
1107            .contains("unknown member"));
1108        assert!(reader
1109            .blob_size_sync(&hash)
1110            .expect_err("unknown member size must fail")
1111            .to_string()
1112            .contains("unknown member"));
1113    }
1114
1115    #[test]
1116    fn rejects_manifest_snapshot_that_differs_from_opened_members() {
1117        let opened = br#"{"version":1,"members":[]}"#;
1118        let opened_sha256 = to_hex(&sha256(opened));
1119        require_opened_manifest(opened, opened, &opened_sha256).unwrap();
1120
1121        let changed = br#"{"version":1,"members":[{"id":"changed"}]}"#;
1122        let error = require_opened_manifest(changed, opened, &opened_sha256)
1123            .expect_err("changed manifest snapshot must fail");
1124        assert!(error.to_string().contains(&opened_sha256));
1125        assert!(error.to_string().contains(&to_hex(&sha256(changed))));
1126    }
1127
1128    #[test]
1129    #[ignore = "subprocess entry point for read-only PoolStore manifest mutation"]
1130    fn read_only_manifest_mutator_helper() {
1131        let Some(catalog) = std::env::var_os(MUTATE_CATALOG_ENV) else {
1132            return;
1133        };
1134        let member = std::env::var_os(MUTATE_MEMBER_ENV).expect("mutator member path");
1135        let pool = PoolStore::open(catalog, pool_config()).expect("open catalog writer");
1136        pool.add_member(PoolMemberConfig::new(member.into(), 1024 * 1024))
1137            .expect("mutate pool manifest");
1138        pool.force_sync().expect("sync mutated manifest");
1139    }
1140
1141    #[test]
1142    fn live_catalog_audit_rejects_manifest_changed_after_reader_open() {
1143        check_live_catalog_manifest_change(true);
1144    }
1145
1146    #[test]
1147    fn live_catalog_growth_invalidates_reader_without_crashing() {
1148        check_live_catalog_manifest_change(false);
1149    }
1150
1151    fn check_live_catalog_manifest_change(reserve_catalog_space: bool) {
1152        let temp = TempDir::new().expect("temp dir");
1153        let catalog = temp.path().join("catalog");
1154        let initial_member = temp.path().join("initial-member");
1155        let added_member = temp.path().join("added-member");
1156        let pool = PoolStore::open(&catalog, pool_config()).expect("open pool");
1157        pool.add_member(PoolMemberConfig::new(initial_member, 1024 * 1024))
1158            .expect("add initial member");
1159        pool.force_sync().expect("sync initial manifest");
1160        drop(pool);
1161
1162        let catalog_data = catalog.join("data.mdb");
1163        if reserve_catalog_space {
1164            // Keep this mutation within the existing mapping so it still
1165            // exercises the exact manifest comparison on every platform.
1166            let file = fs::OpenOptions::new()
1167                .write(true)
1168                .open(&catalog_data)
1169                .expect("open closed catalog for preallocation");
1170            file.set_len(file.metadata().expect("catalog metadata").len() + 128 * 1024)
1171                .expect("reserve catalog pages");
1172        }
1173        let opened_size = fs::metadata(&catalog_data).expect("catalog size").len();
1174        let reader = ReadOnlyPoolStore::open(&catalog).expect("open pinned read-only pool");
1175        reader
1176            .validate_committed_catalog()
1177            .expect("initial catalog matches opened manifest");
1178
1179        let output = Command::new(std::env::current_exe().expect("test binary"))
1180            .arg("--ignored")
1181            .arg("--exact")
1182            .arg("pool::read_only::tests::read_only_manifest_mutator_helper")
1183            .env(MUTATE_CATALOG_ENV, &catalog)
1184            .env(MUTATE_MEMBER_ENV, &added_member)
1185            .env("RUST_TEST_THREADS", "1")
1186            .output()
1187            .expect("run manifest mutator helper");
1188        assert!(
1189            output.status.success(),
1190            "manifest mutator failed: stdout={} stderr={}",
1191            String::from_utf8_lossy(&output.stdout),
1192            String::from_utf8_lossy(&output.stderr)
1193        );
1194
1195        let changed_size = fs::metadata(&catalog_data)
1196            .expect("changed catalog size")
1197            .len();
1198        if reserve_catalog_space {
1199            assert_eq!(
1200                changed_size, opened_size,
1201                "manifest change must fit the mapping"
1202            );
1203        } else {
1204            assert!(changed_size > opened_size, "writer must grow the catalog");
1205        }
1206        let error = reader
1207            .validate_committed_catalog()
1208            .expect_err("changed live manifest must differ from opened member snapshot");
1209        let expected = if cfg!(windows) && !reserve_catalog_space {
1210            "MDB_MAP_RESIZED"
1211        } else {
1212            "manifest changed after read-only members were opened"
1213        };
1214        assert!(
1215            error.to_string().contains(expected),
1216            "unexpected live manifest mismatch: {error}"
1217        );
1218        drop(reader);
1219        let reopened = ReadOnlyPoolStore::open(&catalog).expect("reopen changed catalog");
1220        assert_eq!(
1221            reopened
1222                .manifest_snapshot()
1223                .expect("new manifest")
1224                .members
1225                .len(),
1226            2
1227        );
1228    }
1229}
hashtree_lmdb/pool/read_only.rs

hashtree_lmdb/pool/
read_only.rs
