Add a library to sync conda environments

Summary:
This adds a crate to do rsync-style syncing of a source conda env to a
destination, as per:

```
$ rsync --archive --update --no-dirtimes --delete --delete-excluded ...
```

As `rsync` (and this crate) uses mtimes for syncing, it'll work even with the
differences in prefixes between source and destination conda envs.

The notable differences with rsync are:
1) In current flows, when users "activate" their conda env locally, prefixes in
    the env will be quickly re-written, changing mtimes to be ahead of the
    destination (e.g. on MAST, the conda env is unpacked into it's expected
    prefix location).  This crate uses the `pack-meta/history.jsonl` file to detect
    prefix update windows and use that to filter out these sprurious mtime updates.
2) Files that are copied perform in-place prefix replacement (TODO: not finished
    yet).

Differential Revision: D78424481

Author: andrewjcg

monarch_conda/Cargo.toml

@@ -0,0 +1,35 @@
+# @generated by autocargo from //monarch/monarch_conda:[conda-sync-cli,monarch_conda]
+
+[package]
+name = "monarch_conda"
+version = "0.0.0"
+authors = ["Meta"]
+edition = "2021"
+license = "BSD-3-Clause"
+
+[[bin]]
+name = "conda_sync_cli"
+path = "src/main.rs"
+
+[dependencies]
+anyhow = "1.0.98"
+async-tempfile = "0.7.0"
+bincode = "1.3.3"
+chrono = { version = "0.4.41", features = ["clock", "serde", "std"], default-features = false }
+clap = { version = "4.5.38", features = ["derive", "env", "string", "unicode", "wrap_help"] }
+dashmap = { version = "5.5.3", features = ["rayon", "serde"] }
+digest = "0.10"
+filetime = "0.2.25"
+futures = { version = "0.3.30", features = ["async-await", "compat"] }
+globset = { version = "0.4.13", features = ["serde1"] }
+ignore = "0.4"
+rattler_conda_types = "0.28.3"
+serde = { version = "1.0.185", features = ["derive", "rc"] }
+serde_json = { version = "1.0.140", features = ["alloc", "float_roundtrip", "unbounded_depth"] }
+sha2 = "0.10.6"
+tokio = { version = "1.45.0", features = ["full", "test-util", "tracing"] }
+tokio-util = { version = "0.7.15", features = ["full"] }
+walkdir = "2.3"
+
+[dev-dependencies]
+tempfile = "3.15"

monarch_conda/src/diff.rs

@@ -0,0 +1,289 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+use std::collections::HashMap;
+use std::path::Path;
+use std::path::PathBuf;
+use std::time::Duration;
+use std::time::SystemTime;
+use std::time::UNIX_EPOCH;
+
+use anyhow::Context;
+use anyhow::Result;
+use anyhow::ensure;
+use chrono::DateTime;
+use chrono::Utc;
+use digest::Digest;
+use digest::Output;
+use rattler_conda_types::PrefixRecord;
+use rattler_conda_types::prefix_record::PathsEntry;
+use serde::Deserialize;
+use serde::Serialize;
+use serde_json;
+use sha2::Sha256;
+use tokio::fs;
+use walkdir::WalkDir;
+
+use crate::hash_utils;
+use crate::pack_meta::History;
+use crate::pack_meta::Offsets;
+
+/// Fingerprint of the conda environment, used to detect if two envs are using the same
+/// "base" env.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct CondaMetaFingerprint {
+    hash: Output<Sha256>,
+}
+
+impl CondaMetaFingerprint {
+    async fn from_env(path: &Path) -> Result<Self> {
+        let mut hasher = Sha256::new();
+        hash_utils::hash_directory_tree(&path.join("conda-meta"), &mut hasher).await?;
+        Ok(Self {
+            hash: hasher.finalize(),
+        })
+    }
+
+    fn ensure_can_sync_to(&self, other: &CondaMetaFingerprint) -> Result<()> {
+        ensure!(self.hash == other.hash);
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct PackMetaFingerprint {
+    offsets: Output<Sha256>,
+    pub history: History,
+}
+
+impl PackMetaFingerprint {
+    async fn from_env(path: &Path) -> Result<Self> {
+        let pack_meta = path.join("pack-meta");
+
+        // Read first line of history.jsonl
+        let contents = fs::read_to_string(pack_meta.join("history.jsonl")).await?;
+        let history = History::from_contents(&contents)?;
+
+        // Read entire offsets.jsonl file
+        let mut hasher = Sha256::new();
+        let contents = fs::read_to_string(pack_meta.join("offsets.jsonl")).await?;
+        let offsets = Offsets::from_contents(&contents)?;
+        for ent in offsets.entries {
+            let contents = bincode::serialize(&(ent.path, ent.mode, ent.offsets.len()))?;
+            hasher.update(contents.len().to_le_bytes());
+            hasher.update(&contents);
+        }
+        let offsets = hasher.finalize();
+
+        Ok(Self { history, offsets })
+    }
+
+    fn ensure_can_sync_to(&self, other: &PackMetaFingerprint) -> Result<()> {
+        ensure!(
+            self.history.entries.first().context("empty history")?
+                == other.history.entries.first().context("empty history")?
+        );
+        ensure!(
+            self.history
+                .entries
+                .last()
+                .context("empty history")?
+                .timestamp
+                <= other
+                    .history
+                    .entries
+                    .last()
+                    .context("empty history")?
+                    .timestamp
+        );
+
+        ensure!(self.offsets == other.offsets);
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct CondaFingerprint {
+    pub conda_meta: CondaMetaFingerprint,
+    pub pack_meta: PackMetaFingerprint,
+}
+
+impl CondaFingerprint {
+    pub async fn from_env(path: &Path) -> Result<Self> {
+        Ok(Self {
+            conda_meta: CondaMetaFingerprint::from_env(path).await?,
+            pack_meta: PackMetaFingerprint::from_env(path).await?,
+        })
+    }
+
+    pub fn ensure_can_sync_to(&self, other: &CondaFingerprint) -> Result<()> {
+        self.conda_meta
+            .ensure_can_sync_to(&other.conda_meta)
+            .context("conda-meta")?;
+        self.pack_meta
+            .ensure_can_sync_to(&other.pack_meta)
+            .context("pack-meta")?;
+        Ok(())
+    }
+
+    pub fn mtime_comparator(
+        a: &Self,
+        b: &Self,
+    ) -> Result<Box<dyn Fn(&SystemTime, &SystemTime) -> std::cmp::Ordering + Send + Sync>> {
+        let (a_prefix, a_base) = a.pack_meta.history.first()?;
+        let (b_prefix, b_base) = b.pack_meta.history.first()?;
+        ensure!(a_prefix == b_prefix);
+        let a_base = UNIX_EPOCH + Duration::from_secs(a_base + 1);
+        let b_base = UNIX_EPOCH + Duration::from_secs(b_base + 1);
+
+        let a_window = a
+            .pack_meta
+            .history
+            .prefix_and_last_update_window()?
+            .1
+            .map(|(s, e)| {
+                (
+                    UNIX_EPOCH + Duration::from_secs(s),
+                    UNIX_EPOCH + Duration::from_secs(e + 1),
+                )
+            });
+        let b_window = b
+            .pack_meta
+            .history
+            .prefix_and_last_update_window()?
+            .1
+            .map(|(s, e)| {
+                (
+                    UNIX_EPOCH + Duration::from_secs(s),
+                    UNIX_EPOCH + Duration::from_secs(e + 1),
+                )
+            });
+        Ok(Box::new(move |a: &SystemTime, b: &SystemTime| {
+            match (
+                *a > a_base && a_window.is_none_or(|(s, e)| *a < s || *a > e),
+                *b > b_base && b_window.is_none_or(|(s, e)| *b < s || *b > e),
+            ) {
+                (true, false) => std::cmp::Ordering::Greater,
+                (false, true) => std::cmp::Ordering::Less,
+                (false, false) => std::cmp::Ordering::Equal,
+                (true, true) => a.cmp(b),
+            }
+        }))
+    }
+}
+
+async fn load_metadata(env: &Path) -> Result<Vec<PrefixRecord>> {
+    let conda_meta_path = env.join("conda-meta");
+    let mut records = Vec::new();
+
+    for entry in WalkDir::new(&conda_meta_path)
+        .min_depth(1)
+        .into_iter()
+        .filter_map(|e| e.ok())
+        .filter(|e| e.file_name() != "history")
+    {
+        let content = fs::read_to_string(entry.path())
+            .await
+            .with_context(|| format!("reading {:?}", entry.path()))?;
+        let record: PrefixRecord = serde_json::from_str(&content)?;
+        records.push(record);
+    }
+
+    Ok(records)
+}
+
+fn index_metadata(records: &[PrefixRecord]) -> HashMap<PathBuf, (&PathsEntry, &PrefixRecord)> {
+    let mut path_map = HashMap::new();
+
+    for record in records {
+        // Add all paths from paths_data to the map
+        for paths_entry in &record.paths_data.paths {
+            path_map.insert(paths_entry.relative_path.clone(), (paths_entry, record));
+        }
+    }
+
+    path_map
+}
+
+pub enum Change {
+    Missing,
+    Added,
+    Modified,
+}
+
+pub async fn diff(env: &Path) -> Result<Vec<(PathBuf, Change)>> {
+    let metadata = load_metadata(env).await?;
+    let mut index = index_metadata(&metadata);
+    let mut changes = Vec::new();
+
+    for entry in WalkDir::new(env)
+        .into_iter()
+        .filter_map(|e| e.ok())
+        .filter(|e| e.file_type().is_file())
+        .filter(|e| !e.path().starts_with(env.join("conda-meta")))
+    {
+        let file_path = entry.path();
+        let relative_path = file_path.strip_prefix(env)?;
+
+        if let Some((relative_path, (paths_entry, prefix_record))) =
+            index.remove_entry(relative_path)
+        {
+            // File is tracked in metadata, check if it's modified
+            let file_metadata = fs::symlink_metadata(file_path).await?;
+            let file_size = file_metadata.len();
+            let file_mtime = file_metadata.modified()?;
+
+            let mut is_modified = false;
+
+            // Compare size if available (now we have direct access to PathsEntry)
+            if !is_modified {
+                if let Some(size_in_bytes) = paths_entry.size_in_bytes {
+                    is_modified = file_size != size_in_bytes;
+                }
+            }
+
+            // Compare mtime with package record timestamp
+            // If package timestamp > file mtime, then file is not modified
+            if !is_modified {
+                let package_timestamp = prefix_record
+                    .repodata_record
+                    .package_record
+                    .timestamp
+                    .context("no timestamp")?;
+
+                // Convert file mtime to chrono DateTime for comparison
+                let duration_since_epoch = file_mtime.duration_since(UNIX_EPOCH)?;
+                let file_datetime = DateTime::<Utc>::from_timestamp(
+                    duration_since_epoch.as_secs() as i64,
+                    duration_since_epoch.subsec_nanos(),
+                )
+                .context("failed to convert mtime to DateTime")?;
+
+                // If file was modified after package installation, it's been modified
+                if file_datetime > package_timestamp {
+                    is_modified = true;
+                }
+            }
+
+            if is_modified {
+                changes.push((relative_path, Change::Modified));
+            }
+        } else {
+            // File exists but is not tracked in metadata
+            changes.push((relative_path.to_path_buf(), Change::Added));
+        }
+    }
+
+    // Remaining paths in index are missing files
+    for path in index.into_keys() {
+        changes.push((path, Change::Missing));
+    }
+
+    Ok(changes)
+}