1use std::collections::BTreeMap;
41use std::fs;
42use std::path::{Path, PathBuf};
43use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
44
45use meerkat_core::REALM_MANIFEST_FILE_NAME;
46use rusqlite::OptionalExtension;
47use serde::{Deserialize, Serialize};
48use sha2::{Digest, Sha256};
49
50use crate::error::StoreError;
51
52pub const REALM_SQLITE_FILES: &[&str] = &[
58 "sessions.sqlite3",
59 "runtime.sqlite3",
60 "workgraph.sqlite3",
61 "jobs.sqlite3",
62 "memory/memory.sqlite3",
63 "tasks.db",
64 "sessions_jsonl/session_index.sqlite3",
65];
66
67pub const FINDING_LEGACY_HOME_SESSIONS_DIR: &str = "legacy-home-sessions-dir";
70
71pub fn enumerate_realm_sqlite_files(realm_dir: &Path) -> Vec<PathBuf> {
76 enumerate_realm_sqlite_inventory(realm_dir).files
77}
78
79#[non_exhaustive]
82#[derive(Debug, Default)]
83pub struct RealmSqliteInventory {
84 pub files: Vec<PathBuf>,
86 pub symlinks: Vec<PathBuf>,
90}
91
92pub fn enumerate_realm_sqlite_inventory(realm_dir: &Path) -> RealmSqliteInventory {
95 let mut inventory = RealmSqliteInventory::default();
96 for path in REALM_SQLITE_FILES
97 .iter()
98 .map(|relative| realm_dir.join(relative))
99 {
100 classify_inventory_path(path, &mut inventory);
101 }
102 if let Ok(entries) = fs::read_dir(realm_dir.join("mobs")) {
103 for entry in entries.filter_map(Result::ok) {
104 let path = entry.path();
105 if path.extension().and_then(|ext| ext.to_str()) == Some("db") {
106 classify_inventory_path(path, &mut inventory);
107 }
108 }
109 }
110 inventory.files.sort();
111 inventory.symlinks.sort();
112 inventory
113}
114
115fn classify_inventory_path(path: PathBuf, inventory: &mut RealmSqliteInventory) {
116 match fs::symlink_metadata(&path) {
117 Ok(metadata) if metadata.file_type().is_symlink() => inventory.symlinks.push(path),
118 Ok(metadata) if metadata.is_file() => inventory.files.push(path),
119 _ => {}
121 }
122}
123
124fn mob_database_files(realm_dir: &Path) -> Vec<PathBuf> {
127 let mut files = Vec::new();
128 if let Ok(entries) = fs::read_dir(realm_dir.join("mobs")) {
129 for entry in entries.filter_map(Result::ok) {
130 let path = entry.path();
131 if path.extension().and_then(|ext| ext.to_str()) == Some("db") && path.is_file() {
132 files.push(path);
133 }
134 }
135 }
136 files.sort();
137 files
138}
139
140pub const REALM_WRITE_ADMISSION_STEM: &str = "realm";
152
153pub fn realm_write_admission_target(realm_dir: &Path) -> PathBuf {
156 realm_dir.join(REALM_WRITE_ADMISSION_STEM)
157}
158
159pub fn store_realm_admission_target(store_dir: &Path) -> Option<PathBuf> {
167 let parent = store_dir.parent()?;
168 parent
169 .join(REALM_MANIFEST_FILE_NAME)
170 .is_file()
171 .then(|| realm_write_admission_target(parent))
172}
173
174#[derive(Debug)]
194pub struct RealmMaintenanceFence {
195 admission: meerkat_sqlite::ExclusiveFence,
196 fences: Vec<meerkat_sqlite::ExclusiveFence>,
197 databases: Vec<PathBuf>,
198}
199
200impl RealmMaintenanceFence {
201 pub fn acquire(realm_dir: &Path, deadline: Duration) -> Result<Self, StoreError> {
214 let started = Instant::now();
215 let remaining = |started: &Instant| deadline.saturating_sub(started.elapsed());
216
217 let admission = meerkat_sqlite::ExclusiveFence::acquire(
218 &realm_write_admission_target(realm_dir),
219 remaining(&started),
220 )
221 .map_err(StoreError::from)?;
222
223 let mut databases: Vec<PathBuf> = REALM_SQLITE_FILES
226 .iter()
227 .map(|relative| realm_dir.join(relative))
228 .collect();
229 databases.sort();
230 let mut fences = Vec::with_capacity(databases.len());
231 for database in &databases {
232 if let Some(parent) = database.parent() {
233 fs::create_dir_all(parent)?;
234 }
235 let fence = meerkat_sqlite::ExclusiveFence::acquire(database, remaining(&started))
236 .map_err(|error| {
237 StoreError::from(error)
239 })?;
240 fences.push(fence);
241 }
242
243 let mut first_pass = true;
248 loop {
249 let new: Vec<PathBuf> = mob_database_files(realm_dir)
250 .into_iter()
251 .filter(|database| !databases.contains(database))
252 .collect();
253 if new.is_empty() {
254 break;
255 }
256 if !first_pass && started.elapsed() >= deadline {
257 return Err(StoreError::Internal(format!(
258 "mob databases kept appearing under '{}' during maintenance-fence \
259 acquisition; realm is not quiescent",
260 realm_dir.join("mobs").display()
261 )));
262 }
263 first_pass = false;
264 for database in new {
265 let fence = meerkat_sqlite::ExclusiveFence::acquire(&database, remaining(&started))
266 .map_err(StoreError::from)?;
267 fences.push(fence);
268 databases.push(database);
269 }
270 }
271 databases.sort();
272
273 Ok(Self {
274 admission,
275 fences,
276 databases,
277 })
278 }
279
280 pub fn fenced_databases(&self) -> &[PathBuf] {
284 &self.databases
285 }
286
287 pub fn admission_lock_path(&self) -> &Path {
289 self.admission.lock_path()
290 }
291
292 pub fn len(&self) -> usize {
294 self.fences.len()
295 }
296
297 pub fn is_empty(&self) -> bool {
300 self.fences.is_empty()
301 }
302}
303
304pub fn backup_artifact_name(original: &str, purpose: &str) -> String {
314 let timestamp = SystemTime::now()
315 .duration_since(UNIX_EPOCH)
316 .map(|elapsed| elapsed.as_secs())
317 .unwrap_or(0);
318 let version = env!("CARGO_PKG_VERSION");
319 if purpose.is_empty() {
320 format!("{original}.pre-{version}-{timestamp}")
321 } else {
322 format!("{original}.pre-{version}-{timestamp}.{purpose}")
323 }
324}
325
326pub fn is_backup_artifact_name(name: &str) -> bool {
334 let Some(idx) = name.rfind(".pre-") else {
335 return false;
336 };
337 if idx == 0 {
338 return false;
340 }
341 let suffix = &name[idx + ".pre-".len()..];
344 let Some((version, rest)) = suffix.split_once('-') else {
345 return false;
346 };
347 let version_ok = !version.is_empty()
348 && version.split('.').count() >= 2
349 && version
350 .split('.')
351 .all(|part| !part.is_empty() && part.chars().all(|c| c.is_ascii_digit()));
352 if !version_ok {
353 return false;
354 }
355 let (timestamp, purpose) = match rest.split_once('.') {
356 Some((timestamp, purpose)) => (timestamp, Some(purpose)),
357 None => (rest, None),
358 };
359 let timestamp_ok = !timestamp.is_empty() && timestamp.chars().all(|c| c.is_ascii_digit());
360 let purpose_ok = purpose.is_none_or(|p| {
361 !p.is_empty()
362 && p.chars()
363 .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
364 });
365 timestamp_ok && purpose_ok
366}
367
368pub fn is_quarantine_artifact_name(name: &str) -> bool {
372 let Some(idx) = name.rfind(".corrupt-") else {
373 return false;
374 };
375 if idx == 0 {
376 return false;
377 }
378 let timestamp = &name[idx + ".corrupt-".len()..];
379 !timestamp.is_empty() && timestamp.chars().all(|c| c.is_ascii_digit())
380}
381
382#[non_exhaustive]
388#[derive(Debug)]
389pub struct ArchivedPath {
390 pub archive: PathBuf,
392 pub warnings: Vec<String>,
395}
396
397pub fn archive_path_read_only_reported(
403 path: &Path,
404 purpose: &str,
405) -> Result<ArchivedPath, StoreError> {
406 let name = path
407 .file_name()
408 .and_then(|name| name.to_str())
409 .ok_or_else(|| {
410 StoreError::Internal(format!(
411 "cannot archive '{}': path has no UTF-8 file name",
412 path.display()
413 ))
414 })?;
415 let archive = path.with_file_name(backup_artifact_name(name, purpose));
416 fs::rename(path, &archive)?;
417 let mut warnings = Vec::new();
418 strip_write_permissions(&archive, &mut warnings);
419 sync_parent_dir_reported(&archive, &mut warnings);
420 Ok(ArchivedPath { archive, warnings })
421}
422
423pub fn archive_path_read_only(path: &Path, purpose: &str) -> Result<PathBuf, StoreError> {
427 archive_path_read_only_reported(path, purpose).map(|archived| archived.archive)
428}
429
430fn sync_parent_dir_reported(path: &Path, warnings: &mut Vec<String>) {
434 #[cfg(unix)]
435 {
436 let Some(parent) = path
437 .parent()
438 .filter(|parent| !parent.as_os_str().is_empty())
439 else {
440 return;
441 };
442 let synced = fs::File::open(parent).and_then(|dir| dir.sync_all());
443 if let Err(error) = synced {
444 warnings.push(format!(
445 "archive rename to '{}' is not crash-durable: fsync of parent directory '{}' \
446 failed: {error}",
447 path.display(),
448 parent.display()
449 ));
450 }
451 }
452 #[cfg(not(unix))]
453 {
454 let _ = (path, &mut *warnings);
455 }
456}
457
458fn strip_write_permissions(path: &Path, warnings: &mut Vec<String>) {
462 let metadata = match fs::symlink_metadata(path) {
463 Ok(metadata) => metadata,
464 Err(error) => {
465 warnings.push(format!(
466 "archive hardening skipped for '{}': {error}",
467 path.display()
468 ));
469 return;
470 }
471 };
472 if metadata.file_type().is_symlink() {
473 return;
474 }
475 if metadata.is_dir() {
476 match fs::read_dir(path) {
477 Ok(entries) => {
478 for entry in entries.filter_map(Result::ok) {
479 strip_write_permissions(&entry.path(), warnings);
480 }
481 }
482 Err(error) => warnings.push(format!(
483 "archive hardening could not list '{}': {error}",
484 path.display()
485 )),
486 }
487 }
488 let mut permissions = metadata.permissions();
489 permissions.set_readonly(true);
490 if let Err(error) = fs::set_permissions(path, permissions) {
491 warnings.push(format!(
492 "archive '{}' may remain writable: {error}",
493 path.display()
494 ));
495 }
496}
497
498fn restore_write_permissions_best_effort(path: &Path) {
499 let Ok(metadata) = fs::symlink_metadata(path) else {
500 return;
501 };
502 if metadata.file_type().is_symlink() {
503 return;
505 }
506 #[allow(clippy::permissions_set_readonly_false)] {
509 let mut permissions = metadata.permissions();
510 permissions.set_readonly(false);
511 let _ = fs::set_permissions(path, permissions);
512 }
513 if metadata.is_dir()
514 && let Ok(entries) = fs::read_dir(path)
515 {
516 for entry in entries.filter_map(Result::ok) {
517 restore_write_permissions_best_effort(&entry.path());
518 }
519 }
520}
521
522#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
529#[serde(rename_all = "snake_case")]
530pub enum MigrateMode {
531 #[default]
533 DryRun,
534 Apply,
536}
537
538#[non_exhaustive]
540#[derive(Debug, Clone, Default, Serialize, Deserialize)]
541pub struct MigrateReport {
542 #[serde(default)]
544 pub mode: MigrateMode,
545 #[serde(default)]
547 pub swept_roots: Vec<PathBuf>,
548 #[serde(default)]
550 pub realms: Vec<RealmMigrateReport>,
551 #[serde(default)]
553 pub split_brain: Vec<SplitBrainReport>,
554 #[serde(default)]
557 pub findings: Vec<meerkat_core::StorageFinding>,
558 #[serde(default)]
560 pub errors: Vec<String>,
561}
562
563impl MigrateReport {
564 pub fn new(mode: MigrateMode, swept_roots: Vec<PathBuf>) -> Self {
566 Self {
567 mode,
568 swept_roots,
569 ..Self::default()
570 }
571 }
572
573 pub fn has_errors(&self) -> bool {
576 !self.errors.is_empty() || self.realms.iter().any(|realm| !realm.errors.is_empty())
577 }
578}
579
580#[non_exhaustive]
582#[derive(Debug, Clone, Serialize, Deserialize)]
583pub struct RealmMigrateReport {
584 pub realm: String,
586 pub root: PathBuf,
589 #[serde(default, skip_serializing_if = "Option::is_none")]
591 pub backend: Option<String>,
592 #[serde(default)]
594 pub ledger: Vec<LedgerBaselineEntry>,
595 #[serde(default)]
597 pub notes: Vec<String>,
598 #[serde(default)]
600 pub errors: Vec<String>,
601}
602
603impl RealmMigrateReport {
604 pub fn new(realm: impl Into<String>, root: PathBuf) -> Self {
606 Self {
607 realm: realm.into(),
608 root,
609 backend: None,
610 ledger: Vec::new(),
611 notes: Vec::new(),
612 errors: Vec::new(),
613 }
614 }
615}
616
617#[non_exhaustive]
619#[derive(Debug, Clone, Serialize, Deserialize)]
620pub struct LedgerBaselineEntry {
621 pub database: PathBuf,
623 pub domain: String,
625 #[serde(default, skip_serializing_if = "Option::is_none")]
627 pub before: Option<i64>,
628 #[serde(default, skip_serializing_if = "Option::is_none")]
630 pub after: Option<i64>,
631 pub action: LedgerBaselineAction,
633}
634
635impl LedgerBaselineEntry {
636 pub fn new(database: PathBuf, domain: impl Into<String>, action: LedgerBaselineAction) -> Self {
638 Self {
639 database,
640 domain: domain.into(),
641 before: None,
642 after: None,
643 action,
644 }
645 }
646}
647
648#[non_exhaustive]
650#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
651#[serde(rename_all = "kebab-case")]
652pub enum LedgerBaselineAction {
653 MissingRow,
657 Recorded,
660 Stamped,
662 AlreadyCurrent,
664 ReportOnly,
667}
668
669#[non_exhaustive]
671#[derive(Debug, Clone, Serialize, Deserialize)]
672pub struct SplitBrainReport {
673 pub realm: String,
675 pub locations: Vec<PathBuf>,
677 #[serde(default)]
679 pub sessions_equal: usize,
680 #[serde(default)]
683 pub sessions: Vec<SessionDivergenceEntry>,
684 #[serde(default)]
690 pub files: Vec<FileDivergenceEntry>,
691 pub resolution: SplitBrainResolution,
693 #[serde(default)]
697 pub errors: Vec<String>,
698}
699
700impl SplitBrainReport {
701 pub fn new(realm: impl Into<String>, locations: Vec<PathBuf>) -> Self {
703 Self {
704 realm: realm.into(),
705 locations,
706 sessions_equal: 0,
707 sessions: Vec::new(),
708 files: Vec::new(),
709 resolution: SplitBrainResolution::Refused {
710 reason: "split-brain unresolved: rerun with `--apply --adopt-root <path>` to \
711 adopt one root and archive the other copies read-only"
712 .to_string(),
713 },
714 errors: Vec::new(),
715 }
716 }
717
718 pub fn comparison_is_conclusive(&self) -> bool {
727 self.errors.is_empty()
728 && self
729 .sessions
730 .iter()
731 .all(|entry| entry.status != DivergenceStatus::Unknown)
732 && self
733 .files
734 .iter()
735 .all(|entry| entry.status != DivergenceStatus::Unknown)
736 }
737}
738
739#[non_exhaustive]
741#[derive(Debug, Clone, Serialize, Deserialize)]
742pub struct SessionDivergenceEntry {
743 pub session_id: String,
745 pub status: DivergenceStatus,
747}
748
749#[non_exhaustive]
751#[derive(Debug, Clone, Serialize, Deserialize)]
752pub struct FileDivergenceEntry {
753 pub file: String,
755 pub status: DivergenceStatus,
757}
758
759#[non_exhaustive]
761#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
762#[serde(rename_all = "snake_case", tag = "kind")]
763pub enum DivergenceStatus {
764 Equal,
766 Divergent,
769 OnlyIn {
771 location: PathBuf,
773 },
774 Unknown,
778}
779
780#[non_exhaustive]
782#[derive(Debug, Clone, Serialize, Deserialize)]
783#[serde(rename_all = "snake_case", tag = "kind")]
784pub enum SplitBrainResolution {
785 Refused {
788 reason: String,
790 },
791 Archived {
795 adopted: PathBuf,
797 archived: Vec<PathBuf>,
799 },
800 ArchiveFailed {
805 adopted: PathBuf,
807 archived: Vec<PathBuf>,
809 reason: String,
811 },
812}
813
814#[non_exhaustive]
816#[derive(Debug, Clone, Default, Serialize, Deserialize)]
817pub struct PruneReport {
818 #[serde(default)]
820 pub mode: MigrateMode,
821 #[serde(default)]
823 pub swept_roots: Vec<PathBuf>,
824 #[serde(default)]
827 pub older_than_days: u64,
828 #[serde(default)]
830 pub artifacts: Vec<PruneArtifact>,
831 #[serde(default)]
833 pub errors: Vec<String>,
834}
835
836impl PruneReport {
837 pub fn new(mode: MigrateMode, swept_roots: Vec<PathBuf>, older_than_days: u64) -> Self {
839 Self {
840 mode,
841 swept_roots,
842 older_than_days,
843 ..Self::default()
844 }
845 }
846}
847
848#[non_exhaustive]
850#[derive(Debug, Clone, Serialize, Deserialize)]
851pub struct PruneArtifact {
852 pub path: PathBuf,
854 pub kind: PruneArtifactKind,
856 #[serde(default)]
858 pub bytes: u64,
859 #[serde(default)]
861 pub age_days: u64,
862 pub action: PruneAction,
864}
865
866#[non_exhaustive]
869#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
870#[serde(rename_all = "kebab-case")]
871pub enum PruneArtifactKind {
872 BackupArtifact,
874 QuarantinedIndex,
876}
877
878#[non_exhaustive]
880#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
881#[serde(rename_all = "kebab-case")]
882pub enum PruneAction {
883 WouldDelete,
885 Deleted,
887 Kept,
889 DeleteFailed,
891}
892
893#[non_exhaustive]
899#[derive(Debug, Clone)]
900pub struct RealmDirEntry {
901 pub realm_id: String,
904 pub backend: Option<String>,
906 pub state_root: PathBuf,
908 pub dir: PathBuf,
910 pub manifest_readable: bool,
912}
913
914pub fn list_realm_dirs(state_root: &Path) -> Vec<RealmDirEntry> {
918 let Ok(entries) = fs::read_dir(state_root) else {
919 return Vec::new();
920 };
921 let mut dirs: Vec<PathBuf> = entries
922 .filter_map(Result::ok)
923 .map(|entry| entry.path())
924 .filter(|path| path.is_dir())
925 .collect();
926 dirs.sort();
927
928 let mut realms = Vec::new();
929 for dir in dirs {
930 let name = dir
931 .file_name()
932 .map(|name| name.to_string_lossy().into_owned())
933 .unwrap_or_default();
934 if is_backup_artifact_name(&name) {
935 continue; }
937 let manifest_path = dir.join(REALM_MANIFEST_FILE_NAME);
938 if !manifest_path.is_file() {
939 continue;
940 }
941 let parsed = fs::read(&manifest_path)
942 .ok()
943 .and_then(|bytes| serde_json::from_slice::<serde_json::Value>(&bytes).ok())
944 .and_then(|value| {
945 let realm_id = value.get("realm_id")?.as_str()?.to_string();
946 let backend = value
947 .get("backend")
948 .and_then(|backend| backend.as_str())
949 .map(ToString::to_string);
950 Some((realm_id, backend))
951 });
952 let (realm_id, backend, manifest_readable) = match parsed {
953 Some((realm_id, backend)) => (realm_id, backend, true),
954 None => (name, None, false),
955 };
956 realms.push(RealmDirEntry {
957 realm_id,
958 backend,
959 state_root: state_root.to_path_buf(),
960 dir,
961 manifest_readable,
962 });
963 }
964 realms
965}
966
967pub fn read_domain_versions(db_path: &Path) -> Result<Option<Vec<(String, i64)>>, StoreError> {
970 let conn = meerkat_sqlite::open(db_path, meerkat_sqlite::ConnectionProfile::ReadOnly)
973 .map_err(StoreError::from)?;
974 let table_exists = conn
975 .query_row(
976 "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'meerkat_schema'",
977 [],
978 |_| Ok(()),
979 )
980 .optional()?;
981 if table_exists.is_none() {
982 return Ok(None);
983 }
984 let mut statement =
985 conn.prepare("SELECT domain, version FROM meerkat_schema ORDER BY domain")?;
986 let rows = statement
987 .query_map([], |row| {
988 Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
989 })?
990 .collect::<Result<Vec<_>, _>>()?;
991 Ok(Some(rows))
992}
993
994pub fn supported_domain_version(domain: &str) -> Option<i64> {
1000 match domain {
1001 #[cfg(feature = "sqlite")]
1002 "session-store" => Some(crate::sqlite_store::SESSION_STORE_DOMAIN.supported_version()),
1003 #[cfg(feature = "sqlite")]
1004 "schedule-store" => {
1005 Some(crate::schedule_sqlite_store::SCHEDULE_STORE_DOMAIN.supported_version())
1006 }
1007 "jsonl-index" => Some(crate::index::JSONL_INDEX_DOMAIN.supported_version()),
1008 _ => None,
1009 }
1010}
1011
1012#[non_exhaustive]
1015#[derive(Debug, Clone)]
1016pub struct LedgerDomainReading {
1017 pub database: PathBuf,
1019 pub domain: String,
1021 pub version: Option<i64>,
1023}
1024
1025#[non_exhaustive]
1027#[derive(Debug, Clone)]
1028pub struct FutureDomainVersion {
1029 pub database: PathBuf,
1031 pub domain: String,
1033 pub found: i64,
1035 pub supported: i64,
1037}
1038
1039#[non_exhaustive]
1042#[derive(Debug, Default)]
1043pub struct RealmLedgerBaseline {
1044 pub rows: Vec<LedgerDomainReading>,
1046 pub future: Vec<FutureDomainVersion>,
1050 pub errors: Vec<String>,
1053}
1054
1055pub fn read_realm_ledger_baseline(realm_dir: &Path) -> RealmLedgerBaseline {
1063 let mut baseline = RealmLedgerBaseline::default();
1064 for (relative, expected_domains) in crate::doctor::REALM_DATABASE_FILES {
1065 let db_path = realm_dir.join(relative);
1066 if !db_path.is_file() {
1067 continue;
1068 }
1069 let rows = match read_domain_versions(&db_path) {
1070 Ok(rows) => rows.unwrap_or_default(),
1071 Err(error) => {
1072 baseline.errors.push(format!(
1073 "ledger unreadable for {}: {error}",
1074 db_path.display()
1075 ));
1076 continue;
1077 }
1078 };
1079 for (domain, version) in &rows {
1080 if let Some(supported) = supported_domain_version(domain)
1081 && *version > supported
1082 {
1083 baseline.future.push(FutureDomainVersion {
1084 database: db_path.clone(),
1085 domain: domain.clone(),
1086 found: *version,
1087 supported,
1088 });
1089 }
1090 baseline.rows.push(LedgerDomainReading {
1091 database: db_path.clone(),
1092 domain: domain.clone(),
1093 version: Some(*version),
1094 });
1095 }
1096 for expected in *expected_domains {
1097 if !rows.iter().any(|(domain, _)| domain == expected) {
1098 baseline.rows.push(LedgerDomainReading {
1099 database: db_path.clone(),
1100 domain: (*expected).to_string(),
1101 version: None,
1102 });
1103 }
1104 }
1105 }
1106 baseline
1107}
1108
1109fn table_exists(conn: &rusqlite::Connection, table: &str) -> Result<bool, rusqlite::Error> {
1110 Ok(conn
1111 .query_row(
1112 "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1",
1113 [table],
1114 |_| Ok(()),
1115 )
1116 .optional()?
1117 .is_some())
1118}
1119
1120type SessionDigests = BTreeMap<String, [u8; 32]>;
1125
1126fn session_digests(db_path: &Path) -> Result<SessionDigests, StoreError> {
1134 let conn = meerkat_sqlite::open(db_path, meerkat_sqlite::ConnectionProfile::ReadOnly)
1137 .map_err(StoreError::from)?;
1138 let mut hashers: BTreeMap<String, Sha256> = BTreeMap::new();
1139 let mut feed = |session_id: String, tag: &str, chunks: &[&[u8]]| {
1140 let hasher = hashers.entry(session_id).or_default();
1141 hasher.update(tag.as_bytes());
1142 hasher.update([0u8]);
1143 for chunk in chunks {
1144 hasher.update(chunk);
1145 hasher.update([0u8]);
1146 }
1147 };
1148
1149 if table_exists(&conn, "sessions")? {
1151 let mut statement = conn.prepare(
1152 "SELECT session_id, metadata_json, session_json FROM sessions ORDER BY session_id",
1153 )?;
1154 let mut rows = statement.query([])?;
1155 while let Some(row) = rows.next()? {
1156 let session_id: String = row.get(0)?;
1157 let metadata = row
1158 .get::<_, meerkat_sqlite::JsonColumnBytes>(1)?
1159 .into_bytes();
1160 let document = row
1161 .get::<_, meerkat_sqlite::JsonColumnBytes>(2)?
1162 .into_bytes();
1163 feed(session_id, "sessions", &[&metadata, &document]);
1164 }
1165 }
1166 if table_exists(&conn, "session_heads")? {
1167 let mut statement = conn.prepare(
1168 "SELECT session_id, metadata_json, head_json FROM session_heads ORDER BY session_id",
1169 )?;
1170 let mut rows = statement.query([])?;
1171 while let Some(row) = rows.next()? {
1172 let session_id: String = row.get(0)?;
1173 let metadata = row
1174 .get::<_, meerkat_sqlite::JsonColumnBytes>(1)?
1175 .into_bytes();
1176 let head = row
1177 .get::<_, meerkat_sqlite::JsonColumnBytes>(2)?
1178 .into_bytes();
1179 feed(session_id, "head", &[&metadata, &head]);
1180 }
1181 }
1182 if table_exists(&conn, "session_strand_messages")? {
1183 let mut statement = conn.prepare(
1184 "SELECT session_id, strand, seq, message_json FROM session_strand_messages \
1185 ORDER BY session_id, strand, seq",
1186 )?;
1187 let mut rows = statement.query([])?;
1188 while let Some(row) = rows.next()? {
1189 let session_id: String = row.get(0)?;
1190 let strand: String = row.get(1)?;
1191 let seq: i64 = row.get(2)?;
1192 let message = row
1193 .get::<_, meerkat_sqlite::JsonColumnBytes>(3)?
1194 .into_bytes();
1195 feed(
1196 session_id,
1197 "message",
1198 &[strand.as_bytes(), seq.to_be_bytes().as_slice(), &message],
1199 );
1200 }
1201 }
1202 if table_exists(&conn, "session_strand_links")? {
1203 let mut statement = conn.prepare(
1204 "SELECT session_id, strand, successor, strand_len, splice_start, splice_end, \
1205 successor_end FROM session_strand_links ORDER BY session_id, strand",
1206 )?;
1207 let mut rows = statement.query([])?;
1208 while let Some(row) = rows.next()? {
1209 let session_id: String = row.get(0)?;
1210 let strand: String = row.get(1)?;
1211 let successor: String = row.get(2)?;
1212 let strand_len: i64 = row.get(3)?;
1213 let splice_start: i64 = row.get(4)?;
1214 let splice_end: i64 = row.get(5)?;
1215 let successor_end: i64 = row.get(6)?;
1216 feed(
1217 session_id,
1218 "strand-link",
1219 &[
1220 strand.as_bytes(),
1221 successor.as_bytes(),
1222 strand_len.to_be_bytes().as_slice(),
1223 splice_start.to_be_bytes().as_slice(),
1224 splice_end.to_be_bytes().as_slice(),
1225 successor_end.to_be_bytes().as_slice(),
1226 ],
1227 );
1228 }
1229 }
1230 if table_exists(&conn, "session_rewrites")? {
1231 let mut statement = conn.prepare(
1232 "SELECT session_id, rewrite_idx, commit_json FROM session_rewrites \
1233 ORDER BY session_id, rewrite_idx",
1234 )?;
1235 let mut rows = statement.query([])?;
1236 while let Some(row) = rows.next()? {
1237 let session_id: String = row.get(0)?;
1238 let index: i64 = row.get(1)?;
1239 let commit = row
1240 .get::<_, meerkat_sqlite::JsonColumnBytes>(2)?
1241 .into_bytes();
1242 feed(
1243 session_id,
1244 "rewrite",
1245 &[index.to_be_bytes().as_slice(), &commit],
1246 );
1247 }
1248 }
1249 if table_exists(&conn, "session_component_events")? {
1250 let mut statement = conn.prepare(
1251 "SELECT session_id, component, seq, event_json, event_digest \
1252 FROM session_component_events ORDER BY session_id, component, seq",
1253 )?;
1254 let mut rows = statement.query([])?;
1255 while let Some(row) = rows.next()? {
1256 let session_id: String = row.get(0)?;
1257 let component: String = row.get(1)?;
1258 let seq: i64 = row.get(2)?;
1259 let event = row
1260 .get::<_, meerkat_sqlite::JsonColumnBytes>(3)?
1261 .into_bytes();
1262 let event_digest: String = row.get(4)?;
1263 feed(
1264 session_id,
1265 "component-event",
1266 &[
1267 component.as_bytes(),
1268 seq.to_be_bytes().as_slice(),
1269 &event,
1270 event_digest.as_bytes(),
1271 ],
1272 );
1273 }
1274 }
1275 if table_exists(&conn, "session_head_metadata_cells")? {
1276 let mut statement = conn.prepare(
1277 "SELECT session_id, metadata_key, key_route, exact_value_digest, \
1278 metadata_json, created_at_ms \
1279 FROM session_head_metadata_cells \
1280 ORDER BY session_id, metadata_key, exact_value_digest",
1281 )?;
1282 let mut rows = statement.query([])?;
1283 while let Some(row) = rows.next()? {
1284 let session_id: String = row.get(0)?;
1285 let metadata_key: String = row.get(1)?;
1286 let key_route: Vec<u8> = row.get(2)?;
1287 let exact_value_digest: String = row.get(3)?;
1288 let metadata = row
1289 .get::<_, meerkat_sqlite::JsonColumnBytes>(4)?
1290 .into_bytes();
1291 let created_at_ms: i64 = row.get(5)?;
1292 feed(
1293 session_id,
1294 "head-metadata-cell",
1295 &[
1296 metadata_key.as_bytes(),
1297 &key_route,
1298 exact_value_digest.as_bytes(),
1299 &metadata,
1300 created_at_ms.to_be_bytes().as_slice(),
1301 ],
1302 );
1303 }
1304 }
1305 if table_exists(&conn, "session_head_metadata_current")? {
1306 let mut statement = conn.prepare(
1307 "SELECT session_id, metadata_key, key_route, exact_value_digest \
1308 FROM session_head_metadata_current ORDER BY session_id, metadata_key",
1309 )?;
1310 let mut rows = statement.query([])?;
1311 while let Some(row) = rows.next()? {
1312 let session_id: String = row.get(0)?;
1313 let metadata_key: String = row.get(1)?;
1314 let key_route: Vec<u8> = row.get(2)?;
1315 let exact_value_digest: String = row.get(3)?;
1316 feed(
1317 session_id,
1318 "head-metadata-current",
1319 &[
1320 metadata_key.as_bytes(),
1321 &key_route,
1322 exact_value_digest.as_bytes(),
1323 ],
1324 );
1325 }
1326 }
1327 if table_exists(&conn, "session_head_metadata_states")? {
1328 let mut statement = conn.prepare(
1329 "SELECT session_id, state_id, predecessor_state_id, identity_json, \
1330 transition_id, created_at_ms FROM session_head_metadata_states \
1331 ORDER BY session_id, state_id",
1332 )?;
1333 let mut rows = statement.query([])?;
1334 while let Some(row) = rows.next()? {
1335 let session_id: String = row.get(0)?;
1336 let state_id: String = row.get(1)?;
1337 let predecessor_state_id: Option<String> = row.get(2)?;
1338 let identity = row
1339 .get::<_, meerkat_sqlite::JsonColumnBytes>(3)?
1340 .into_bytes();
1341 let transition_id: String = row.get(4)?;
1342 let created_at_ms: i64 = row.get(5)?;
1343 let predecessor_present = [u8::from(predecessor_state_id.is_some())];
1344 feed(
1345 session_id,
1346 "head-metadata-state",
1347 &[
1348 state_id.as_bytes(),
1349 &predecessor_present,
1350 predecessor_state_id
1351 .as_deref()
1352 .unwrap_or_default()
1353 .as_bytes(),
1354 &identity,
1355 transition_id.as_bytes(),
1356 created_at_ms.to_be_bytes().as_slice(),
1357 ],
1358 );
1359 }
1360 }
1361 if table_exists(&conn, "session_head_metadata_state_deltas")? {
1362 let mut statement = conn.prepare(
1363 "SELECT session_id, state_id, ordinal, metadata_key, key_route, \
1364 predecessor_exact_value_digest, successor_exact_value_digest \
1365 FROM session_head_metadata_state_deltas \
1366 ORDER BY session_id, state_id, ordinal",
1367 )?;
1368 let mut rows = statement.query([])?;
1369 while let Some(row) = rows.next()? {
1370 let session_id: String = row.get(0)?;
1371 let state_id: String = row.get(1)?;
1372 let ordinal: i64 = row.get(2)?;
1373 let metadata_key: String = row.get(3)?;
1374 let key_route: Vec<u8> = row.get(4)?;
1375 let predecessor_exact_value_digest: Option<String> = row.get(5)?;
1376 let successor_exact_value_digest: Option<String> = row.get(6)?;
1377 let predecessor_present = [u8::from(predecessor_exact_value_digest.is_some())];
1378 let successor_present = [u8::from(successor_exact_value_digest.is_some())];
1379 feed(
1380 session_id,
1381 "head-metadata-state-delta",
1382 &[
1383 state_id.as_bytes(),
1384 ordinal.to_be_bytes().as_slice(),
1385 metadata_key.as_bytes(),
1386 &key_route,
1387 &predecessor_present,
1388 predecessor_exact_value_digest
1389 .as_deref()
1390 .unwrap_or_default()
1391 .as_bytes(),
1392 &successor_present,
1393 successor_exact_value_digest
1394 .as_deref()
1395 .unwrap_or_default()
1396 .as_bytes(),
1397 ],
1398 );
1399 }
1400 }
1401 if table_exists(&conn, "session_head_metadata_refs")? {
1402 let mut statement = conn.prepare(
1403 "SELECT session_id, owner, head_cas_token, state_id \
1404 FROM session_head_metadata_refs ORDER BY session_id, owner",
1405 )?;
1406 let mut rows = statement.query([])?;
1407 while let Some(row) = rows.next()? {
1408 let session_id: String = row.get(0)?;
1409 let owner: String = row.get(1)?;
1410 let head_cas_token: String = row.get(2)?;
1411 let state_id: String = row.get(3)?;
1412 feed(
1413 session_id,
1414 "head-metadata-ref",
1415 &[
1416 owner.as_bytes(),
1417 head_cas_token.as_bytes(),
1418 state_id.as_bytes(),
1419 ],
1420 );
1421 }
1422 }
1423 if table_exists(&conn, "session_head_metadata_head_lineage")? {
1424 let mut statement = conn.prepare(
1425 "SELECT session_id, transition_id, predecessor_head_cas_token, \
1426 successor_head_cas_token, predecessor_state_id, successor_state_id, \
1427 created_at_ms FROM session_head_metadata_head_lineage \
1428 ORDER BY session_id, transition_id",
1429 )?;
1430 let mut rows = statement.query([])?;
1431 while let Some(row) = rows.next()? {
1432 let session_id: String = row.get(0)?;
1433 let transition_id: String = row.get(1)?;
1434 let predecessor_head_cas_token: Option<String> = row.get(2)?;
1435 let successor_head_cas_token: String = row.get(3)?;
1436 let predecessor_state_id: Option<String> = row.get(4)?;
1437 let successor_state_id: String = row.get(5)?;
1438 let created_at_ms: i64 = row.get(6)?;
1439 let predecessor_head_present = [u8::from(predecessor_head_cas_token.is_some())];
1440 let predecessor_state_present = [u8::from(predecessor_state_id.is_some())];
1441 feed(
1442 session_id,
1443 "head-metadata-head-lineage",
1444 &[
1445 transition_id.as_bytes(),
1446 &predecessor_head_present,
1447 predecessor_head_cas_token
1448 .as_deref()
1449 .unwrap_or_default()
1450 .as_bytes(),
1451 successor_head_cas_token.as_bytes(),
1452 &predecessor_state_present,
1453 predecessor_state_id
1454 .as_deref()
1455 .unwrap_or_default()
1456 .as_bytes(),
1457 successor_state_id.as_bytes(),
1458 created_at_ms.to_be_bytes().as_slice(),
1459 ],
1460 );
1461 }
1462 }
1463
1464 Ok(hashers
1465 .into_iter()
1466 .map(|(session_id, hasher)| (session_id, hasher.finalize().into()))
1467 .collect())
1468}
1469
1470fn stream_file_into(hasher: &mut Sha256, path: &Path) -> Result<(), StoreError> {
1474 let mut file = fs::File::open(path)?;
1475 std::io::copy(&mut file, hasher)?;
1476 Ok(())
1477}
1478
1479fn file_digest(path: &Path) -> Result<[u8; 32], StoreError> {
1484 let mut hasher = Sha256::new();
1485 stream_file_into(&mut hasher, path)?;
1486 let mut wal = path.as_os_str().to_os_string();
1487 wal.push("-wal");
1488 let wal = PathBuf::from(wal);
1489 match fs::symlink_metadata(&wal) {
1490 Ok(metadata) if metadata.file_type().is_symlink() => {
1493 return Err(StoreError::Internal(format!(
1494 "refusing to follow symlink {} while digesting {}",
1495 wal.display(),
1496 path.display()
1497 )));
1498 }
1499 Ok(metadata) if metadata.is_file() => stream_file_into(&mut hasher, &wal)?,
1500 _ => {}
1501 }
1502 Ok(hasher.finalize().into())
1503}
1504
1505#[derive(Debug)]
1508struct RealmContentSnapshot {
1509 sessions: Option<SessionDigests>,
1513 files: BTreeMap<String, [u8; 32]>,
1515 unreadable: Vec<String>,
1519 errors: Vec<String>,
1521}
1522
1523impl RealmContentSnapshot {
1524 fn poisons(&self, relative: &str) -> bool {
1525 self.unreadable.iter().any(|entry| {
1526 entry == relative || (entry.ends_with('/') && relative.starts_with(entry.as_str()))
1527 })
1528 }
1529
1530 fn record_unreadable(&mut self, relative: String, error: String) {
1531 self.errors.push(error);
1532 self.unreadable.push(relative);
1533 }
1534}
1535
1536fn snapshot_relative(location: &Path, path: &Path) -> Option<String> {
1537 path.strip_prefix(location)
1538 .ok()
1539 .map(|relative| relative.to_string_lossy().replace('\\', "/"))
1540}
1541
1542fn digest_entry_into(path: &Path, relative: String, snapshot: &mut RealmContentSnapshot) {
1547 match fs::symlink_metadata(path) {
1548 Ok(metadata) if metadata.file_type().is_symlink() => snapshot.record_unreadable(
1549 relative,
1550 format!(
1551 "refusing to follow symlink {} during divergence comparison",
1552 path.display()
1553 ),
1554 ),
1555 Ok(metadata) if metadata.is_file() => match file_digest(path) {
1556 Ok(digest) => {
1557 snapshot.files.insert(relative, digest);
1558 }
1559 Err(error) => snapshot.record_unreadable(
1560 relative,
1561 format!("file digest unavailable for {}: {error}", path.display()),
1562 ),
1563 },
1564 Ok(_) => snapshot.record_unreadable(
1565 relative,
1566 format!(
1567 "{} is not a regular file; divergence cannot account for it",
1568 path.display()
1569 ),
1570 ),
1571 Err(error) => {
1572 snapshot
1573 .record_unreadable(relative, format!("cannot stat {}: {error}", path.display()));
1574 }
1575 }
1576}
1577
1578fn snapshot_jsonl_sessions(location: &Path, snapshot: &mut RealmContentSnapshot) {
1582 let dir = location.join("sessions_jsonl");
1583 let entries = match fs::read_dir(&dir) {
1584 Ok(entries) => entries,
1585 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return,
1586 Err(error) => {
1587 snapshot.record_unreadable(
1588 "sessions_jsonl/".to_string(),
1589 format!("cannot list {}: {error}", dir.display()),
1590 );
1591 return;
1592 }
1593 };
1594 let mut paths: Vec<PathBuf> = entries
1595 .filter_map(Result::ok)
1596 .map(|entry| entry.path())
1597 .collect();
1598 paths.sort();
1599 for path in paths {
1600 if path.extension().and_then(|ext| ext.to_str()) != Some("jsonl") {
1601 continue;
1602 }
1603 let Some(relative) = snapshot_relative(location, &path) else {
1604 continue;
1605 };
1606 digest_entry_into(&path, relative, snapshot);
1607 }
1608}
1609
1610fn snapshot_tree(location: &Path, tree: &str, snapshot: &mut RealmContentSnapshot) {
1613 let root = location.join(tree);
1614 match fs::symlink_metadata(&root) {
1615 Ok(metadata) if metadata.file_type().is_symlink() => {
1616 snapshot.record_unreadable(
1617 format!("{tree}/"),
1618 format!(
1619 "refusing to follow symlink {} during divergence comparison",
1620 root.display()
1621 ),
1622 );
1623 return;
1624 }
1625 Ok(metadata) if metadata.is_dir() => {}
1626 Ok(_) | Err(_) => return, }
1628 walk_tree_into(location, &root, snapshot);
1629}
1630
1631fn walk_tree_into(location: &Path, dir: &Path, snapshot: &mut RealmContentSnapshot) {
1632 let entries = match fs::read_dir(dir) {
1633 Ok(entries) => entries,
1634 Err(error) => {
1635 let relative = snapshot_relative(location, dir).unwrap_or_default();
1636 snapshot.record_unreadable(
1637 format!("{relative}/"),
1638 format!("cannot list {}: {error}", dir.display()),
1639 );
1640 return;
1641 }
1642 };
1643 let mut paths: Vec<PathBuf> = entries
1644 .filter_map(Result::ok)
1645 .map(|entry| entry.path())
1646 .collect();
1647 paths.sort();
1648 for path in paths {
1649 let Some(relative) = snapshot_relative(location, &path) else {
1650 continue;
1651 };
1652 match fs::symlink_metadata(&path) {
1653 Ok(metadata) if metadata.is_dir() && !metadata.file_type().is_symlink() => {
1654 walk_tree_into(location, &path, snapshot);
1655 }
1656 _ => digest_entry_into(&path, relative, snapshot),
1657 }
1658 }
1659}
1660
1661fn snapshot_realm_content(location: &Path) -> RealmContentSnapshot {
1666 let mut snapshot = RealmContentSnapshot {
1667 sessions: Some(SessionDigests::new()),
1668 files: BTreeMap::new(),
1669 unreadable: Vec::new(),
1670 errors: Vec::new(),
1671 };
1672
1673 let sessions_db = location.join("sessions.sqlite3");
1674 match fs::symlink_metadata(&sessions_db) {
1675 Ok(metadata) if metadata.file_type().is_symlink() => snapshot.sessions = None,
1679 Ok(metadata) if metadata.is_file() => match session_digests(&sessions_db) {
1680 Ok(digests) => snapshot.sessions = Some(digests),
1681 Err(error) => {
1682 snapshot.sessions = None;
1683 snapshot.errors.push(format!(
1684 "session divergence unavailable for {}: {error}",
1685 sessions_db.display()
1686 ));
1687 }
1688 },
1689 _ => {}
1690 }
1691
1692 let inventory = enumerate_realm_sqlite_inventory(location);
1693 for link in &inventory.symlinks {
1694 let Some(relative) = snapshot_relative(location, link) else {
1695 continue;
1696 };
1697 snapshot.record_unreadable(
1698 relative,
1699 format!(
1700 "refusing to follow symlink {} during divergence comparison",
1701 link.display()
1702 ),
1703 );
1704 }
1705 for db in inventory.files {
1706 let Some(relative) = snapshot_relative(location, &db) else {
1707 continue;
1708 };
1709 match file_digest(&db) {
1710 Ok(digest) => {
1711 snapshot.files.insert(relative, digest);
1712 }
1713 Err(error) => snapshot.record_unreadable(
1714 relative,
1715 format!("file digest unavailable for {}: {error}", db.display()),
1716 ),
1717 }
1718 }
1719
1720 snapshot_jsonl_sessions(location, &mut snapshot);
1721 snapshot_tree(location, "blobs", &mut snapshot);
1722 snapshot_tree(location, "artifacts", &mut snapshot);
1723 snapshot
1724}
1725
1726pub fn compute_split_brain_report(realm: &str, locations: &[PathBuf]) -> SplitBrainReport {
1745 let mut report = SplitBrainReport::new(realm, locations.to_vec());
1746
1747 let snapshots: Vec<(PathBuf, RealmContentSnapshot)> = locations
1748 .iter()
1749 .map(|location| (location.clone(), snapshot_realm_content(location)))
1750 .collect();
1751 for (_, snapshot) in &snapshots {
1752 report.errors.extend(snapshot.errors.iter().cloned());
1753 }
1754
1755 let sessions_poisoned = snapshots
1758 .iter()
1759 .any(|(_, snapshot)| snapshot.sessions.is_none());
1760 let mut all_ids: Vec<String> = snapshots
1761 .iter()
1762 .filter_map(|(_, snapshot)| snapshot.sessions.as_ref())
1763 .flat_map(|digests| digests.keys().cloned())
1764 .collect();
1765 all_ids.sort();
1766 all_ids.dedup();
1767 for session_id in all_ids {
1768 if sessions_poisoned {
1769 report.sessions.push(SessionDivergenceEntry {
1770 session_id,
1771 status: DivergenceStatus::Unknown,
1772 });
1773 continue;
1774 }
1775 let holders: Vec<(&PathBuf, &[u8; 32])> = snapshots
1776 .iter()
1777 .filter_map(|(location, snapshot)| {
1778 snapshot
1779 .sessions
1780 .as_ref()?
1781 .get(&session_id)
1782 .map(|digest| (location, digest))
1783 })
1784 .collect();
1785 if holders.len() == 1 {
1786 report.sessions.push(SessionDivergenceEntry {
1787 session_id,
1788 status: DivergenceStatus::OnlyIn {
1789 location: holders[0].0.clone(),
1790 },
1791 });
1792 } else if holders.len() == snapshots.len()
1793 && holders.iter().all(|(_, digest)| *digest == holders[0].1)
1794 {
1795 report.sessions_equal += 1;
1796 } else {
1797 report.sessions.push(SessionDivergenceEntry {
1798 session_id,
1799 status: DivergenceStatus::Divergent,
1800 });
1801 }
1802 }
1803
1804 let mut relative_files: Vec<String> = snapshots
1806 .iter()
1807 .flat_map(|(_, snapshot)| {
1808 snapshot.files.keys().cloned().chain(
1809 snapshot
1810 .unreadable
1811 .iter()
1812 .filter(|entry| !entry.ends_with('/'))
1813 .cloned(),
1814 )
1815 })
1816 .collect();
1817 relative_files.sort();
1818 relative_files.dedup();
1819 for relative in relative_files {
1820 if snapshots
1821 .iter()
1822 .any(|(_, snapshot)| snapshot.poisons(&relative))
1823 {
1824 report.files.push(FileDivergenceEntry {
1825 file: relative,
1826 status: DivergenceStatus::Unknown,
1827 });
1828 continue;
1829 }
1830 let holders: Vec<(&PathBuf, &[u8; 32])> = snapshots
1831 .iter()
1832 .filter_map(|(location, snapshot)| {
1833 snapshot
1834 .files
1835 .get(&relative)
1836 .map(|digest| (location, digest))
1837 })
1838 .collect();
1839 let status = if holders.len() == 1 {
1840 DivergenceStatus::OnlyIn {
1841 location: holders[0].0.clone(),
1842 }
1843 } else if holders.len() == snapshots.len()
1844 && holders.iter().all(|(_, digest)| *digest == holders[0].1)
1845 {
1846 DivergenceStatus::Equal
1847 } else {
1848 DivergenceStatus::Divergent
1849 };
1850 report.files.push(FileDivergenceEntry {
1851 file: relative,
1852 status,
1853 });
1854 }
1855
1856 report
1857}
1858
1859fn recursive_size(path: &Path) -> u64 {
1864 let Ok(metadata) = fs::symlink_metadata(path) else {
1865 return 0;
1866 };
1867 if metadata.is_dir() {
1868 let Ok(entries) = fs::read_dir(path) else {
1869 return 0;
1870 };
1871 entries
1872 .filter_map(Result::ok)
1873 .map(|entry| recursive_size(&entry.path()))
1874 .sum()
1875 } else {
1876 metadata.len()
1877 }
1878}
1879
1880pub fn registered_artifact_timestamp(name: &str) -> Option<u64> {
1884 if is_backup_artifact_name(name) {
1885 let idx = name.rfind(".pre-")?;
1886 let suffix = &name[idx + ".pre-".len()..];
1887 let (_, rest) = suffix.split_once('-')?;
1888 let timestamp = rest
1889 .split_once('.')
1890 .map_or(rest, |(timestamp, _)| timestamp);
1891 return timestamp.parse().ok();
1892 }
1893 if is_quarantine_artifact_name(name) {
1894 let idx = name.rfind(".corrupt-")?;
1895 return name[idx + ".corrupt-".len()..].parse().ok();
1896 }
1897 None
1898}
1899
1900fn age_days(path: &Path, name: &str) -> u64 {
1907 let now = SystemTime::now()
1908 .duration_since(UNIX_EPOCH)
1909 .map(|elapsed| elapsed.as_secs())
1910 .unwrap_or(0);
1911 if let Some(registered) = registered_artifact_timestamp(name) {
1912 return now.saturating_sub(registered) / 86_400;
1913 }
1914 fs::symlink_metadata(path)
1915 .and_then(|metadata| metadata.modified())
1916 .ok()
1917 .and_then(|modified| SystemTime::now().duration_since(modified).ok())
1918 .map(|age| age.as_secs() / 86_400)
1919 .unwrap_or(0)
1920}
1921
1922fn artifact_kind(name: &str) -> Option<PruneArtifactKind> {
1923 if is_backup_artifact_name(name) {
1924 Some(PruneArtifactKind::BackupArtifact)
1925 } else if is_quarantine_artifact_name(name) {
1926 Some(PruneArtifactKind::QuarantinedIndex)
1927 } else {
1928 None
1929 }
1930}
1931
1932pub fn registered_artifact_original(name: &str) -> Option<&str> {
1936 if is_backup_artifact_name(name) {
1937 return Some(&name[..name.rfind(".pre-")?]);
1938 }
1939 if is_quarantine_artifact_name(name) {
1940 return Some(&name[..name.rfind(".corrupt-")?]);
1941 }
1942 None
1943}
1944
1945fn push_artifacts_in(
1946 dir: &Path,
1947 dirs_too: bool,
1948 original_filter: Option<&str>,
1949 artifacts: &mut Vec<PruneArtifact>,
1950) {
1951 let Ok(entries) = fs::read_dir(dir) else {
1952 return;
1953 };
1954 let mut paths: Vec<PathBuf> = entries
1955 .filter_map(Result::ok)
1956 .map(|entry| entry.path())
1957 .collect();
1958 paths.sort();
1959 for path in paths {
1960 if path.is_dir() && !dirs_too {
1961 continue;
1962 }
1963 let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
1964 continue;
1965 };
1966 let Some(kind) = artifact_kind(name) else {
1967 continue;
1968 };
1969 if kind == PruneArtifactKind::QuarantinedIndex
1973 && !fs::symlink_metadata(&path)
1974 .map(|metadata| metadata.is_file())
1975 .unwrap_or(false)
1976 {
1977 continue;
1978 }
1979 if let Some(filter) = original_filter
1980 && registered_artifact_original(name) != Some(filter)
1981 {
1982 continue;
1983 }
1984 artifacts.push(PruneArtifact {
1985 bytes: recursive_size(&path),
1986 age_days: age_days(&path, name),
1987 path,
1988 kind,
1989 action: PruneAction::Kept,
1990 });
1991 }
1992}
1993
1994pub fn enumerate_maintenance_artifacts(state_roots: &[PathBuf]) -> Vec<PruneArtifact> {
2001 enumerate_maintenance_artifacts_filtered(state_roots, None)
2002}
2003
2004pub fn enumerate_maintenance_artifacts_filtered(
2011 state_roots: &[PathBuf],
2012 realm_filter: Option<&str>,
2013) -> Vec<PruneArtifact> {
2014 let realm_dir_name = realm_filter.map(meerkat_core::sanitize_realm_id);
2015 let mut artifacts = Vec::new();
2016 let mut seen_roots: Vec<PathBuf> = Vec::new();
2017 for root in state_roots {
2018 let canonical = fs::canonicalize(root).unwrap_or_else(|_| root.clone());
2019 if seen_roots.contains(&canonical) {
2020 continue;
2021 }
2022 seen_roots.push(canonical);
2023
2024 push_artifacts_in(root, true, realm_dir_name.as_deref(), &mut artifacts);
2026
2027 for realm in list_realm_dirs(root) {
2030 if let Some(filter) = realm_filter
2031 && realm.realm_id != filter
2032 {
2033 continue;
2034 }
2035 for scan_dir in [
2036 realm.dir.clone(),
2037 realm.dir.join("memory"),
2038 realm.dir.join("sessions_jsonl"),
2039 realm.dir.join("mobs"),
2040 ] {
2041 push_artifacts_in(&scan_dir, false, None, &mut artifacts);
2042 }
2043 }
2044 }
2045 artifacts
2046}
2047
2048pub fn remove_maintenance_artifact(path: &Path) -> Result<(), StoreError> {
2053 let name = path
2054 .file_name()
2055 .and_then(|name| name.to_str())
2056 .unwrap_or_default();
2057 if artifact_kind(name).is_none() {
2058 return Err(StoreError::Internal(format!(
2059 "refusing to remove '{}': not a registered maintenance artifact \
2060 (*.pre-* / *.corrupt-*)",
2061 path.display()
2062 )));
2063 }
2064 restore_write_permissions_best_effort(path);
2065 let metadata = fs::symlink_metadata(path)?;
2066 if metadata.is_dir() {
2067 fs::remove_dir_all(path)?;
2068 } else {
2069 fs::remove_file(path)?;
2070 }
2071 Ok(())
2072}
2073
2074#[cfg(test)]
2075#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
2076mod tests {
2077
2078 #[test]
2079 fn artifact_name_validation_rejects_lookalikes() {
2080 for lookalike in [
2082 "notes.pre-release",
2083 "config.pre-prod",
2084 ".pre-0.8.3-1700000000",
2085 "db.pre-0.8.3-notadigit",
2086 "db.pre-083-1700000000",
2087 "db.pre-0.8.3-1700000000.",
2088 "db.pre-0.8.3-1700000000.bad purpose",
2089 ] {
2090 assert!(!is_backup_artifact_name(lookalike), "{lookalike}");
2091 }
2092 for valid in [
2093 "sessions.sqlite3.pre-0.8.3-1700000000",
2094 "sessions.sqlite3.pre-0.8.3-1700000000.split-brain",
2095 "realm-dir.pre-10.20.30-1.adopt_other",
2096 ] {
2097 assert!(is_backup_artifact_name(valid), "{valid}");
2098 }
2099 for lookalike in ["notes.corrupt-ish", ".corrupt-123", "x.corrupt-12a"] {
2100 assert!(!is_quarantine_artifact_name(lookalike), "{lookalike}");
2101 }
2102 assert!(is_quarantine_artifact_name(
2103 "session_index.sqlite3.corrupt-1700000000"
2104 ));
2105 }
2106 use super::*;
2107 use rusqlite::Connection;
2108
2109 fn create_db(path: &Path) {
2110 if let Some(parent) = path.parent() {
2111 fs::create_dir_all(parent).unwrap();
2112 }
2113 let conn = Connection::open(path).unwrap();
2114 conn.execute_batch("CREATE TABLE t (x INTEGER)").unwrap();
2115 }
2116
2117 fn write_manifest(state_root: &Path, realm_id: &str, backend: &str) -> PathBuf {
2118 let dir = state_root.join(meerkat_core::sanitize_realm_id(realm_id));
2119 fs::create_dir_all(&dir).unwrap();
2120 fs::write(
2121 dir.join(REALM_MANIFEST_FILE_NAME),
2122 serde_json::to_vec(&serde_json::json!({
2123 "realm_id": realm_id,
2124 "backend": backend,
2125 "origin": "explicit",
2126 "created_at": "0",
2127 }))
2128 .unwrap(),
2129 )
2130 .unwrap();
2131 dir
2132 }
2133
2134 #[test]
2135 fn fence_acquires_full_inventory_and_mobs_in_sorted_order() {
2136 let temp = tempfile::tempdir().unwrap();
2137 let realm = temp.path().join("realm");
2138 create_db(&realm.join("sessions.sqlite3"));
2139 create_db(&realm.join("workgraph.sqlite3"));
2140 create_db(&realm.join("memory/memory.sqlite3"));
2141 create_db(&realm.join("mobs/alpha.db"));
2142 create_db(&realm.join("mobs/realm_profiles.db"));
2143 fs::write(realm.join("notes.txt"), b"x").unwrap();
2145
2146 let fence = RealmMaintenanceFence::acquire(&realm, Duration::from_secs(1)).unwrap();
2147 assert_eq!(fence.len(), REALM_SQLITE_FILES.len() + 2);
2149 assert!(!fence.is_empty());
2150 let mut expected: Vec<PathBuf> = REALM_SQLITE_FILES
2151 .iter()
2152 .map(|relative| realm.join(relative))
2153 .collect();
2154 expected.push(realm.join("mobs/alpha.db"));
2155 expected.push(realm.join("mobs/realm_profiles.db"));
2156 expected.sort();
2157 assert_eq!(fence.fenced_databases(), expected.as_slice());
2158 for database in fence.fenced_databases() {
2161 assert!(meerkat_sqlite::fence_lock_path(database).is_file());
2162 }
2163 assert!(!realm.join("tasks.db").exists());
2164 assert!(fence.admission_lock_path().ends_with("realm.mfence"));
2166 assert!(fence.admission_lock_path().is_file());
2167 }
2168
2169 #[test]
2170 fn foreign_admission_holder_blocks_fence_acquisition() {
2171 let temp = tempfile::tempdir().unwrap();
2172 let realm = temp.path().join("realm");
2173 create_db(&realm.join("sessions.sqlite3"));
2174
2175 let admission_lock = meerkat_sqlite::fence_lock_path(&realm_write_admission_target(&realm));
2178 let foreign = fs::OpenOptions::new()
2179 .read(true)
2180 .write(true)
2181 .create(true)
2182 .truncate(false)
2183 .open(&admission_lock)
2184 .unwrap();
2185 foreign.try_lock().unwrap();
2186
2187 let error = RealmMaintenanceFence::acquire(&realm, Duration::from_millis(100))
2188 .expect_err("foreign admission holder must fail acquisition");
2189 assert!(
2190 matches!(error, StoreError::MaintenanceFenceHeld { ref path }
2191 if path.ends_with(REALM_WRITE_ADMISSION_STEM)),
2192 "{error:?}"
2193 );
2194 drop(foreign);
2195 RealmMaintenanceFence::acquire(&realm, Duration::from_secs(1)).unwrap();
2196 }
2197
2198 #[test]
2199 fn foreign_holder_fails_typed_and_releases_partial_acquisition() {
2200 let temp = tempfile::tempdir().unwrap();
2201 let realm = temp.path().join("realm");
2202 create_db(&realm.join("sessions.sqlite3"));
2203 create_db(&realm.join("workgraph.sqlite3"));
2204
2205 let foreign_lock = meerkat_sqlite::fence_lock_path(&realm.join("workgraph.sqlite3"));
2208 let foreign = fs::OpenOptions::new()
2209 .read(true)
2210 .write(true)
2211 .create(true)
2212 .truncate(false)
2213 .open(&foreign_lock)
2214 .unwrap();
2215 foreign.try_lock().unwrap();
2216
2217 let error = RealmMaintenanceFence::acquire(&realm, Duration::from_millis(100))
2218 .expect_err("foreign fence holder must fail acquisition");
2219 assert!(
2220 matches!(error, StoreError::MaintenanceFenceHeld { ref path }
2221 if path.ends_with("workgraph.sqlite3")),
2222 "{error:?}"
2223 );
2224
2225 let reacquired =
2228 meerkat_sqlite::ExclusiveFence::try_acquire(&realm.join("sessions.sqlite3")).unwrap();
2229 assert!(reacquired.is_some(), "partial acquisition must be released");
2230 drop(reacquired);
2231 let admission =
2232 meerkat_sqlite::ExclusiveFence::try_acquire(&realm_write_admission_target(&realm))
2233 .unwrap();
2234 assert!(admission.is_some(), "admission fence must be released");
2235 drop(admission);
2236 drop(foreign);
2237
2238 let fence = RealmMaintenanceFence::acquire(&realm, Duration::from_secs(1)).unwrap();
2240 assert_eq!(fence.len(), REALM_SQLITE_FILES.len());
2241 }
2242
2243 #[test]
2244 fn empty_realm_still_fences_the_full_fixed_inventory() {
2245 let temp = tempfile::tempdir().unwrap();
2246 let fence = RealmMaintenanceFence::acquire(temp.path(), Duration::ZERO).unwrap();
2247 assert_eq!(fence.len(), REALM_SQLITE_FILES.len());
2250 assert!(!fence.is_empty());
2251 for relative in REALM_SQLITE_FILES {
2252 let lock = meerkat_sqlite::fence_lock_path(&temp.path().join(relative));
2253 assert!(lock.is_file(), "{}", lock.display());
2254 }
2255 }
2256
2257 #[test]
2258 fn backup_names_follow_the_registered_discipline() {
2259 let name = backup_artifact_name("sessions.sqlite3", "split-brain");
2260 assert!(
2261 name.starts_with(&format!(
2262 "sessions.sqlite3.pre-{}-",
2263 env!("CARGO_PKG_VERSION")
2264 )),
2265 "{name}"
2266 );
2267 assert!(name.ends_with(".split-brain"), "{name}");
2268 assert!(is_backup_artifact_name(&name));
2269 let bare = backup_artifact_name("team", "");
2270 assert!(is_backup_artifact_name(&bare));
2271 assert!(!bare.ends_with('.'), "{bare}");
2272 assert!(is_quarantine_artifact_name(
2273 "session_index.sqlite3.corrupt-1"
2274 ));
2275 assert!(!is_backup_artifact_name("sessions.sqlite3"));
2276 }
2277
2278 #[test]
2279 fn archive_renames_and_strips_write_permission() {
2280 let temp = tempfile::tempdir().unwrap();
2281 let dir = temp.path().join("team");
2282 fs::create_dir_all(&dir).unwrap();
2283 fs::write(dir.join("sessions.sqlite3"), b"data").unwrap();
2284
2285 let archive = archive_path_read_only(&dir, "split-brain").unwrap();
2286 assert!(!dir.exists(), "original must be gone");
2287 assert!(archive.is_dir());
2288 assert!(archive.join("sessions.sqlite3").is_file());
2289 let name = archive.file_name().unwrap().to_str().unwrap();
2290 assert!(is_backup_artifact_name(name), "{name}");
2291 let inner = fs::metadata(archive.join("sessions.sqlite3")).unwrap();
2292 assert!(inner.permissions().readonly(), "archive must be read-only");
2293
2294 remove_maintenance_artifact(&archive).unwrap();
2296 assert!(!archive.exists());
2297 }
2298
2299 #[test]
2300 fn archive_reported_returns_archive_path_and_no_warnings_on_success() {
2301 let temp = tempfile::tempdir().unwrap();
2302 let file = temp.path().join("sessions.sqlite3");
2303 fs::write(&file, b"data").unwrap();
2304
2305 let archived = archive_path_read_only_reported(&file, "split-brain").unwrap();
2306 assert!(archived.archive.is_file());
2307 assert!(
2308 archived.warnings.is_empty(),
2309 "successful hardening + parent fsync must not warn: {:?}",
2310 archived.warnings
2311 );
2312 let name = archived.archive.file_name().unwrap().to_str().unwrap();
2313 assert!(is_backup_artifact_name(name), "{name}");
2314 remove_maintenance_artifact(&archived.archive).unwrap();
2315 }
2316
2317 #[cfg(unix)]
2318 #[test]
2319 fn archive_hardening_never_chmods_through_symlinks() {
2320 use std::os::unix::fs::PermissionsExt;
2321
2322 let temp = tempfile::tempdir().unwrap();
2323 let external = temp.path().join("external.txt");
2324 fs::write(&external, b"do not chmod").unwrap();
2325 let dir = temp.path().join("team");
2326 fs::create_dir_all(&dir).unwrap();
2327 std::os::unix::fs::symlink(&external, dir.join("link")).unwrap();
2328
2329 let archived = archive_path_read_only_reported(&dir, "split-brain").unwrap();
2330 let mode = fs::metadata(&external).unwrap().permissions().mode();
2331 assert!(
2332 mode & 0o200 != 0,
2333 "external symlink target must stay writable (mode {mode:o})"
2334 );
2335 remove_maintenance_artifact(&archived.archive).unwrap();
2337 let mode = fs::metadata(&external).unwrap().permissions().mode();
2338 assert!(mode & 0o200 != 0, "mode {mode:o}");
2339 assert!(external.is_file());
2340 }
2341
2342 #[test]
2343 fn remove_refuses_unregistered_paths() {
2344 let temp = tempfile::tempdir().unwrap();
2345 let victim = temp.path().join("precious.txt");
2346 fs::write(&victim, b"do not touch").unwrap();
2347 let error = remove_maintenance_artifact(&victim).expect_err("must refuse");
2348 assert!(matches!(error, StoreError::Internal(_)));
2349 assert!(victim.is_file(), "unregistered path must survive");
2350 }
2351
2352 #[test]
2353 fn split_brain_divergence_classifies_sessions_and_files() {
2354 let temp = tempfile::tempdir().unwrap();
2355 let dir_a = temp.path().join("a/team");
2356 let dir_b = temp.path().join("b/team");
2357 for dir in [&dir_a, &dir_b] {
2358 fs::create_dir_all(dir).unwrap();
2359 }
2360 let ddl = "CREATE TABLE sessions (
2361 session_id TEXT PRIMARY KEY,
2362 created_at_ms INTEGER NOT NULL,
2363 updated_at_ms INTEGER NOT NULL,
2364 message_count INTEGER NOT NULL,
2365 total_tokens INTEGER NOT NULL,
2366 metadata_json TEXT NOT NULL,
2367 session_json BLOB NOT NULL
2368 )";
2369 let insert = |conn: &Connection, id: &str, body: &str| {
2370 conn.execute(
2371 "INSERT INTO sessions VALUES (?1, 0, 0, 0, 0, '{}', ?2)",
2372 rusqlite::params![id, body.as_bytes()],
2373 )
2374 .unwrap();
2375 };
2376 let shared = "00000000-0000-4000-8000-000000000001";
2377 let divergent = "00000000-0000-4000-8000-000000000002";
2378 let only_a = "00000000-0000-4000-8000-000000000003";
2379 {
2380 let conn = Connection::open(dir_a.join("sessions.sqlite3")).unwrap();
2381 conn.execute_batch(ddl).unwrap();
2382 insert(&conn, shared, "same");
2383 insert(&conn, divergent, "version-a");
2384 insert(&conn, only_a, "solo");
2385 }
2386 {
2387 let conn = Connection::open(dir_b.join("sessions.sqlite3")).unwrap();
2388 conn.execute_batch(ddl).unwrap();
2389 insert(&conn, shared, "same");
2390 insert(&conn, divergent, "version-b");
2391 }
2392 create_db(&dir_b.join("workgraph.sqlite3"));
2394
2395 let report = compute_split_brain_report("team", &[dir_a.clone(), dir_b.clone()]);
2396 assert!(report.errors.is_empty(), "{:?}", report.errors);
2397 assert_eq!(report.sessions_equal, 1);
2398 let status_of = |id: &str| {
2399 report
2400 .sessions
2401 .iter()
2402 .find(|entry| entry.session_id == id)
2403 .map(|entry| entry.status.clone())
2404 };
2405 assert_eq!(status_of(divergent), Some(DivergenceStatus::Divergent));
2406 assert_eq!(
2407 status_of(only_a),
2408 Some(DivergenceStatus::OnlyIn { location: dir_a })
2409 );
2410 assert_eq!(
2411 status_of(shared),
2412 None,
2413 "equal sessions are counted, not listed"
2414 );
2415 let workgraph = report
2416 .files
2417 .iter()
2418 .find(|entry| entry.file == "workgraph.sqlite3")
2419 .expect("workgraph file entry");
2420 assert_eq!(
2421 workgraph.status,
2422 DivergenceStatus::OnlyIn { location: dir_b }
2423 );
2424 let sessions_file = report
2425 .files
2426 .iter()
2427 .find(|entry| entry.file == "sessions.sqlite3")
2428 .expect("sessions file entry");
2429 assert_eq!(sessions_file.status, DivergenceStatus::Divergent);
2430 assert!(report.comparison_is_conclusive());
2431 }
2432
2433 #[test]
2434 fn split_brain_covers_jsonl_blobs_and_artifacts() {
2435 let temp = tempfile::tempdir().unwrap();
2436 let dir_a = temp.path().join("a/team");
2437 let dir_b = temp.path().join("b/team");
2438 for dir in [&dir_a, &dir_b] {
2439 fs::create_dir_all(dir.join("sessions_jsonl")).unwrap();
2440 fs::create_dir_all(dir.join("artifacts")).unwrap();
2441 }
2442 for dir in [&dir_a, &dir_b] {
2444 fs::write(dir.join("sessions_jsonl/equal.jsonl"), b"{\"same\":1}").unwrap();
2445 }
2446 fs::write(dir_a.join("sessions_jsonl/split.jsonl"), b"version-a").unwrap();
2447 fs::write(dir_b.join("sessions_jsonl/split.jsonl"), b"version-b").unwrap();
2448 fs::create_dir_all(dir_a.join("blobs/ab")).unwrap();
2450 fs::write(dir_a.join("blobs/ab/abcd.json"), b"blob-bytes").unwrap();
2451 fs::write(dir_a.join("artifacts/r1.json"), b"{\"v\":1}").unwrap();
2453 fs::write(dir_b.join("artifacts/r1.json"), b"{\"v\":2}").unwrap();
2454
2455 let report = compute_split_brain_report("team", &[dir_a.clone(), dir_b]);
2456 assert!(report.errors.is_empty(), "{:?}", report.errors);
2457 let status_of = |file: &str| {
2458 report
2459 .files
2460 .iter()
2461 .find(|entry| entry.file == file)
2462 .map(|entry| entry.status.clone())
2463 };
2464 assert_eq!(
2465 status_of("sessions_jsonl/equal.jsonl"),
2466 Some(DivergenceStatus::Equal)
2467 );
2468 assert_eq!(
2469 status_of("sessions_jsonl/split.jsonl"),
2470 Some(DivergenceStatus::Divergent)
2471 );
2472 assert_eq!(
2473 status_of("blobs/ab/abcd.json"),
2474 Some(DivergenceStatus::OnlyIn { location: dir_a })
2475 );
2476 assert_eq!(
2477 status_of("artifacts/r1.json"),
2478 Some(DivergenceStatus::Divergent)
2479 );
2480 assert!(report.comparison_is_conclusive());
2481 }
2482
2483 #[test]
2484 fn split_brain_read_failure_poisons_the_session_comparison() {
2485 let temp = tempfile::tempdir().unwrap();
2486 let dir_a = temp.path().join("a/team");
2487 let dir_b = temp.path().join("b/team");
2488 for dir in [&dir_a, &dir_b] {
2489 fs::create_dir_all(dir).unwrap();
2490 }
2491 fs::write(dir_a.join("sessions.sqlite3"), b"this is not sqlite").unwrap();
2495 {
2496 let conn = Connection::open(dir_b.join("sessions.sqlite3")).unwrap();
2497 conn.execute_batch(
2498 "CREATE TABLE sessions (
2499 session_id TEXT PRIMARY KEY,
2500 created_at_ms INTEGER NOT NULL,
2501 updated_at_ms INTEGER NOT NULL,
2502 message_count INTEGER NOT NULL,
2503 total_tokens INTEGER NOT NULL,
2504 metadata_json TEXT NOT NULL,
2505 session_json BLOB NOT NULL
2506 );
2507 INSERT INTO sessions VALUES
2508 ('00000000-0000-4000-8000-000000000001', 0, 0, 0, 0, '{}', X'AA');",
2509 )
2510 .unwrap();
2511 }
2512
2513 let report = compute_split_brain_report("team", &[dir_a, dir_b]);
2514 assert!(!report.errors.is_empty());
2515 assert_eq!(report.sessions_equal, 0);
2516 assert_eq!(report.sessions.len(), 1);
2517 assert_eq!(report.sessions[0].status, DivergenceStatus::Unknown);
2518 assert!(!report.comparison_is_conclusive());
2519 }
2520
2521 #[cfg(unix)]
2522 #[test]
2523 fn split_brain_never_follows_symlinks_and_marks_them_unknown() {
2524 let temp = tempfile::tempdir().unwrap();
2525 let outside = temp.path().join("outside.json");
2526 fs::write(&outside, b"external bytes").unwrap();
2527 let dir_a = temp.path().join("a/team");
2528 let dir_b = temp.path().join("b/team");
2529 for dir in [&dir_a, &dir_b] {
2530 fs::create_dir_all(dir.join("blobs")).unwrap();
2531 }
2532 std::os::unix::fs::symlink(&outside, dir_a.join("blobs/link.json")).unwrap();
2533 fs::write(dir_b.join("blobs/link.json"), b"external bytes").unwrap();
2534
2535 let report = compute_split_brain_report("team", &[dir_a, dir_b]);
2536 assert!(!report.errors.is_empty());
2537 let entry = report
2538 .files
2539 .iter()
2540 .find(|entry| entry.file == "blobs/link.json")
2541 .expect("symlinked entry");
2542 assert_eq!(entry.status, DivergenceStatus::Unknown);
2543 assert!(!report.comparison_is_conclusive());
2544 }
2545
2546 #[cfg(unix)]
2551 #[test]
2552 fn split_brain_never_digests_through_symlinked_databases() {
2553 let temp = tempfile::tempdir().unwrap();
2554 let outside = temp.path().join("outside.sqlite3");
2555 {
2556 let conn = Connection::open(&outside).unwrap();
2557 conn.execute_batch(
2558 "CREATE TABLE sessions (
2559 session_id TEXT PRIMARY KEY,
2560 created_at_ms INTEGER NOT NULL,
2561 updated_at_ms INTEGER NOT NULL,
2562 message_count INTEGER NOT NULL,
2563 total_tokens INTEGER NOT NULL,
2564 metadata_json TEXT NOT NULL,
2565 session_json BLOB NOT NULL
2566 );
2567 INSERT INTO sessions VALUES
2568 ('00000000-0000-4000-8000-000000000001', 0, 0, 0, 0, '{}', X'AA');",
2569 )
2570 .unwrap();
2571 }
2572 let dir_a = temp.path().join("a/team");
2573 let dir_b = temp.path().join("b/team");
2574 fs::create_dir_all(&dir_a).unwrap();
2575 fs::create_dir_all(&dir_b).unwrap();
2576 std::os::unix::fs::symlink(&outside, dir_a.join("sessions.sqlite3")).unwrap();
2577 fs::copy(&outside, dir_b.join("sessions.sqlite3")).unwrap();
2578
2579 let inventory = enumerate_realm_sqlite_inventory(&dir_a);
2580 assert!(inventory.files.is_empty());
2581 assert_eq!(inventory.symlinks, vec![dir_a.join("sessions.sqlite3")]);
2582 assert!(enumerate_realm_sqlite_files(&dir_a).is_empty());
2583
2584 let report = compute_split_brain_report("team", &[dir_a, dir_b]);
2585 assert!(
2586 report
2587 .errors
2588 .iter()
2589 .any(|error| error.contains("refusing to follow symlink")),
2590 "{:?}",
2591 report.errors
2592 );
2593 let entry = report
2594 .files
2595 .iter()
2596 .find(|entry| entry.file == "sessions.sqlite3")
2597 .expect("sessions database entry");
2598 assert_eq!(entry.status, DivergenceStatus::Unknown);
2599 assert_eq!(report.sessions.len(), 1, "{:?}", report.sessions);
2601 assert_eq!(report.sessions[0].status, DivergenceStatus::Unknown);
2602 assert!(!report.comparison_is_conclusive());
2603 }
2604
2605 #[test]
2606 fn ledger_baseline_surfaces_read_failures_and_future_versions() {
2607 let temp = tempfile::tempdir().unwrap();
2608 let realm = temp.path().join("team");
2609 fs::create_dir_all(realm.join("sessions_jsonl")).unwrap();
2610 fs::write(realm.join("tasks.db"), b"this is not sqlite").unwrap();
2613 {
2616 let conn =
2617 Connection::open(realm.join("sessions_jsonl/session_index.sqlite3")).unwrap();
2618 conn.execute_batch(
2619 "CREATE TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL);",
2620 )
2621 .unwrap();
2622 conn.execute(
2623 "INSERT INTO meerkat_schema VALUES ('jsonl-index', ?1)",
2624 [i64::MAX],
2625 )
2626 .unwrap();
2627 }
2628 create_db(&realm.join("runtime.sqlite3"));
2630
2631 let baseline = read_realm_ledger_baseline(&realm);
2632 assert_eq!(baseline.errors.len(), 1, "{:?}", baseline.errors);
2633 assert!(
2634 baseline.errors[0].contains("tasks.db"),
2635 "{:?}",
2636 baseline.errors
2637 );
2638 assert!(
2639 !baseline
2640 .rows
2641 .iter()
2642 .any(|row| row.database.ends_with("tasks.db")),
2643 "an unreadable database must contribute no ledger rows"
2644 );
2645 assert_eq!(baseline.future.len(), 1);
2646 assert_eq!(baseline.future[0].domain, "jsonl-index");
2647 assert_eq!(baseline.future[0].found, i64::MAX);
2648 assert_eq!(
2649 baseline.future[0].supported,
2650 crate::index::JSONL_INDEX_DOMAIN.supported_version()
2651 );
2652 let runtime_row = baseline
2653 .rows
2654 .iter()
2655 .find(|row| row.database.ends_with("runtime.sqlite3"))
2656 .expect("runtime row");
2657 assert_eq!(runtime_row.domain, "runtime-store");
2658 assert_eq!(runtime_row.version, None);
2659 }
2660
2661 #[test]
2662 fn prune_enumeration_sees_only_registered_patterns() {
2663 let temp = tempfile::tempdir().unwrap();
2664 let root = temp.path().join("realms");
2665 let realm_dir = write_manifest(&root, "artifacts", "sqlite");
2666 fs::write(
2667 realm_dir.join("sessions.sqlite3.pre-0.0.1-1700000000"),
2668 b"backup",
2669 )
2670 .unwrap();
2671 let jsonl = realm_dir.join("sessions_jsonl");
2672 fs::create_dir_all(&jsonl).unwrap();
2673 fs::write(jsonl.join("session_index.sqlite3.corrupt-42"), b"q").unwrap();
2674 let archived = root.join("team.pre-0.0.1-1700000000.split-brain");
2676 fs::create_dir_all(&archived).unwrap();
2677 fs::write(archived.join("sessions.sqlite3"), b"old").unwrap();
2678 fs::write(realm_dir.join("sessions.sqlite3"), b"live").unwrap();
2680 fs::write(realm_dir.join("notes.txt"), b"keep me").unwrap();
2681
2682 let artifacts = enumerate_maintenance_artifacts(&[root.clone(), root]);
2683 let mut names: Vec<String> = artifacts
2684 .iter()
2685 .map(|artifact| {
2686 artifact
2687 .path
2688 .file_name()
2689 .unwrap()
2690 .to_string_lossy()
2691 .into_owned()
2692 })
2693 .collect();
2694 names.sort();
2695 assert_eq!(
2696 names,
2697 vec![
2698 "session_index.sqlite3.corrupt-42".to_string(),
2699 "sessions.sqlite3.pre-0.0.1-1700000000".to_string(),
2700 "team.pre-0.0.1-1700000000.split-brain".to_string(),
2701 ],
2702 "duplicate roots must not double-count"
2703 );
2704 let archived_entry = artifacts
2705 .iter()
2706 .find(|artifact| artifact.path == archived)
2707 .expect("archived dir entry");
2708 assert_eq!(archived_entry.kind, PruneArtifactKind::BackupArtifact);
2709 assert_eq!(archived_entry.bytes, 3);
2710 }
2711
2712 #[test]
2713 fn quarantine_artifacts_must_be_files_even_at_root_level() {
2714 let temp = tempfile::tempdir().unwrap();
2715 let root = temp.path().join("realms");
2716 fs::create_dir_all(&root).unwrap();
2717 fs::create_dir_all(root.join("data.corrupt-1700000000")).unwrap();
2722 fs::write(root.join("index.sqlite3.corrupt-1700000000"), b"q").unwrap();
2723
2724 let artifacts = enumerate_maintenance_artifacts(&[root]);
2725 let names: Vec<&str> = artifacts
2726 .iter()
2727 .filter_map(|artifact| artifact.path.file_name().and_then(|name| name.to_str()))
2728 .collect();
2729 assert_eq!(names, vec!["index.sqlite3.corrupt-1700000000"]);
2730 }
2731
2732 #[test]
2733 fn report_shapes_round_trip_through_json() {
2734 let mut report = MigrateReport {
2735 mode: MigrateMode::Apply,
2736 ..MigrateReport::default()
2737 };
2738 let mut realm = RealmMigrateReport::new("team", PathBuf::from("/roots/a/team"));
2739 realm.backend = Some("sqlite".to_string());
2740 realm.ledger.push(LedgerBaselineEntry {
2741 database: PathBuf::from("/roots/a/team/sessions.sqlite3"),
2742 domain: "session-store".to_string(),
2743 before: None,
2744 after: Some(1),
2745 action: LedgerBaselineAction::Stamped,
2746 });
2747 report.realms.push(realm);
2748 report.split_brain.push(SplitBrainReport {
2749 realm: "team".to_string(),
2750 locations: vec![
2751 PathBuf::from("/roots/a/team"),
2752 PathBuf::from("/roots/b/team"),
2753 ],
2754 sessions_equal: 4,
2755 sessions: vec![SessionDivergenceEntry {
2756 session_id: "s".to_string(),
2757 status: DivergenceStatus::Divergent,
2758 }],
2759 files: vec![],
2760 resolution: SplitBrainResolution::Archived {
2761 adopted: PathBuf::from("/roots/a/team"),
2762 archived: vec![PathBuf::from("/roots/b/team.pre-0.8.3-1-split-brain")],
2763 },
2764 errors: vec![],
2765 });
2766 report.split_brain.push(SplitBrainReport {
2767 realm: "solo".to_string(),
2768 locations: vec![PathBuf::from("/roots/a/solo")],
2769 sessions_equal: 0,
2770 sessions: vec![SessionDivergenceEntry {
2771 session_id: "s2".to_string(),
2772 status: DivergenceStatus::Unknown,
2773 }],
2774 files: vec![],
2775 resolution: SplitBrainResolution::ArchiveFailed {
2776 adopted: PathBuf::from("/roots/a/solo"),
2777 archived: vec![PathBuf::from("/roots/b/solo.pre-0.8.3-1.split-brain")],
2778 reason: "rename failed".to_string(),
2779 },
2780 errors: vec!["sessions unreadable".to_string()],
2781 });
2782 let json = serde_json::to_string(&report).expect("serialize");
2783 let parsed: MigrateReport = serde_json::from_str(&json).expect("deserialize");
2784 assert!(matches!(parsed.mode, MigrateMode::Apply));
2785 assert_eq!(parsed.realms.len(), 1);
2786 assert_eq!(
2787 parsed.realms[0].ledger[0].action,
2788 LedgerBaselineAction::Stamped
2789 );
2790 assert!(matches!(
2791 parsed.split_brain[0].resolution,
2792 SplitBrainResolution::Archived { .. }
2793 ));
2794 assert!(matches!(
2798 &parsed.split_brain[1].resolution,
2799 SplitBrainResolution::ArchiveFailed { archived, .. } if archived.len() == 1
2800 ));
2801 assert!(!parsed.split_brain[1].comparison_is_conclusive());
2802 assert!(parsed.split_brain[0].comparison_is_conclusive());
2803 assert!(!parsed.has_errors());
2804
2805 let sparse: PruneReport = serde_json::from_str(r#"{"future_field":1}"#).expect("sparse");
2807 assert!(matches!(sparse.mode, MigrateMode::DryRun));
2808 assert!(sparse.artifacts.is_empty());
2809 }
2810
2811 #[test]
2812 fn read_domain_versions_reads_ledgers_read_only() {
2813 let temp = tempfile::tempdir().unwrap();
2814 let db = temp.path().join("db.sqlite3");
2815 create_db(&db);
2816 assert!(read_domain_versions(&db).unwrap().is_none());
2817 {
2818 let conn = Connection::open(&db).unwrap();
2819 conn.execute_batch(
2820 "CREATE TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL);
2821 INSERT INTO meerkat_schema VALUES ('session-store', 1);",
2822 )
2823 .unwrap();
2824 }
2825 assert_eq!(
2826 read_domain_versions(&db).unwrap(),
2827 Some(vec![("session-store".to_string(), 1)])
2828 );
2829 }
2830
2831 #[test]
2832 fn list_realm_dirs_skips_archives_and_reads_manifests_leniently() {
2833 let temp = tempfile::tempdir().unwrap();
2834 let root = temp.path().join("realms");
2835 write_manifest(&root, "alpha", "sqlite");
2836 let corrupt = root.join("corrupt");
2837 fs::create_dir_all(&corrupt).unwrap();
2838 fs::write(corrupt.join(REALM_MANIFEST_FILE_NAME), b"not-json").unwrap();
2839 let archived = root.join("beta.pre-0.0.1-1700000000");
2840 fs::create_dir_all(&archived).unwrap();
2841 fs::write(
2842 archived.join(REALM_MANIFEST_FILE_NAME),
2843 serde_json::to_vec(&serde_json::json!({
2844 "realm_id": "beta", "backend": "sqlite",
2845 "origin": "explicit", "created_at": "0",
2846 }))
2847 .unwrap(),
2848 )
2849 .unwrap();
2850 fs::create_dir_all(root.join("not-a-realm")).unwrap();
2852
2853 let realms = list_realm_dirs(&root);
2854 let ids: Vec<&str> = realms.iter().map(|realm| realm.realm_id.as_str()).collect();
2855 assert_eq!(ids, vec!["alpha", "corrupt"]);
2856 assert!(realms[0].manifest_readable);
2857 assert_eq!(realms[0].backend.as_deref(), Some("sqlite"));
2858 assert!(!realms[1].manifest_readable);
2859 }
2860}