//! A repository on disk: the `.oxy` directory, one workspace, the operation //! log, and the working-copy change (spec §12, §13.1). //! //! Not yet implemented from the spec: packs, checkpoints, multiple //! workspaces, multi-signer policy updates, and the `objects`/`checkpoint` //! endpoints. A repository without a policy is unverified and accepts //! unsigned pushes; with one, everything received is checked (spec §14.5). Changes are signed individually when `core.sign` is //! `change`; the default `seal` leaves them unsigned until push exists. use crate::attest::{Attest, AttestBody}; use crate::diff::{ranges, split_lines, text_edits}; use crate::hash::{hash_object, ChangeHash, KeyId, ObjectRef, ObjectType}; use crate::identity::{ verify as verify_signature, Delegate, Domain, Identity, Scope, SecretKey, Signature, }; use crate::ids::{ChangeId, NodeId, WorkspaceId}; use crate::objects::{ Blob, Change, ChangeKind, Edit, Hlc, Mode, NodeKind, OpRef, StateObject, TreeOp, }; use crate::oplog::{Format1Op, Op, View, SNAPSHOT_SUFFIX}; use crate::protocol::{ open_remote, Bundle, RefResult, RefUpdate, RefsResponse, RemoteOptions, RemoteRef, SignedAttest, SignedPolicy, }; use crate::state::{FileContent, Snapshot, State}; use crate::store::{Store, StoreError}; #[allow(unused_imports)] use crate::subrepo::Pin; use crate::tree::OpKey; use rand::RngCore; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::fs; use std::io::Write; use std::path::{Path, PathBuf}; pub const OXY_DIR: &str = ".oxy"; pub const FORMAT_LINE: &str = "oxy repository format 2\n"; const FORMAT1_LINE: &str = "oxy repository format 1\n"; const BINARY_SNIFF: usize = 8 * 1024; /// Loose objects tolerated before a commit packs them (spec §13.4). pub const AUTOPACK_THRESHOLD: usize = 4096; #[derive(Debug, thiserror::Error)] pub enum RepoError { #[error("io: {0}")] Io(#[from] std::io::Error), #[error(transparent)] Store(#[from] StoreError), #[error(transparent)] Object(#[from] crate::objects::ObjectError), #[error(transparent)] State(#[from] crate::state::StateError), #[error("{0}")] Msg(String), /// The operation would leave a conflict where none may be (a rebased /// change cannot hold one); the caller may offer another way. #[error("{0}")] Conflicted(String), } pub type Result = std::result::Result; /// The operation head's reference record on a store that is not this /// disk's directory (oxygen#68). pub const HEAD_RECORD: &str = "ops/head"; /// How often recording an operation reads the head again when other /// processes keep moving it. const HEAD_TRIES: usize = 8; /// The object store of the repository directory `oxy`: `.oxy/objects`, /// or the store `storage.url` names behind a cache of `storage.cache` /// bytes in `.oxy/cache` (oxygen#68). `.oxy/objects` is then only /// scratch space: packs being written, and derived caches. fn open_objects(oxy: &Path, config: &Config) -> Result { let Some(url) = &config.storage_url else { return Ok(Store::open(oxy.join("objects"))?); }; let options = crate::protocol::RemoteOptions { cache_dir: Some(oxy.join("cache")), ..Default::default() }; let remote = crate::protocol::open_store(url, &options) .ok_or_else(|| { RepoError::Msg(format!( "storage.url {url}: this oxy cannot open that kind of store (dir://, or s3://, gs:// and az:// in builds with cloud stores)" )) })? .map_err(|e| RepoError::Msg(format!("open the repository's store {url}: {e}")))?; let cached = crate::strata::CachingStore::open( std::sync::Arc::from(remote), oxy.join("cache"), config.storage_cache, ) .map_err(|e| RepoError::Msg(format!("open the cache for {url}: {e}")))?; Ok(Store::open_on( std::sync::Arc::new(cached), oxy.join("objects"), )?) } fn fail(s: impl Into) -> Result { Err(RepoError::Msg(s.into())) } mod bookmarks; mod checkout; mod checkpoint; mod config; mod doctor; mod fsck; mod gc; mod git; mod history; mod hooks; mod hosted; mod identity; mod inspect; mod ops; mod policy; mod prompt; mod receipts; mod revset; mod snapshot; pub mod strata; mod subrepos; mod sync; mod tags; #[cfg(test)] mod tests; mod touched; mod walk; mod workspace; pub use bookmarks::{remote_bookmark, BookmarkInfo, MergeReport, RemoteOnlyBookmark, StateGap}; pub use checkout::CheckoutReport; pub use config::{parse_size, read_config_file, Config}; pub use doctor::{Check, DoctorReport, Health}; pub use fsck::{FsckReport, Problem, Severity}; pub use gc::GcReport; pub use git::{GitExportReport, GitImportReport, GitVerifyReport}; pub use hooks::{json_array, json_str, HookOutcome}; pub use hosted::FileEdit; pub use identity::{Rotation, SignatureStatus}; pub use inspect::{BisectStep, BlameLine, Description}; pub use ops::{RestoreReport, RevertReport, ViewChange}; pub use policy::PolicyUpdate; pub use prompt::{PromptInfo, DEFAULT_PROMPT_FORMAT, GLYPH_PROMPT_FORMAT}; pub use receipts::ReceiptInfo; pub use revset::{Revset, REVSET_BUDGET}; pub use snapshot::{ added_node_paths, diff_files_of, has_conflict_markers, is_binary, referenced_changes, render_change, render_hunks, similarity, ChangeStatus, DiffHunk, FileDiff, Hunk, PathStatus, Selection, SnapshotResult, }; pub use subrepos::{SubrepoInfo, SubrepoSync}; pub use sync::{FetchReport, PullReport, PushReport, RefUpdated}; pub use tags::TagInfo; pub use workspace::{FileLock, SnapshotLock, WorkingCopyLock, WorkspaceInfo}; /// The file name under which per-bookmark or per-tag data is kept: `/` /// (and `%`) escaped, so `feature/x` is one file beside `feature` rather /// than a directory that would clash with it. pub(crate) fn name_file(name: &str) -> String { name.replace('%', "%25").replace('/', "%2F") } /// The name a [`name_file`] file name stands for. pub(crate) fn file_name_to_name(file: &str) -> String { file.replace("%2F", "/").replace("%25", "%") } fn write_atomic(path: &Path, data: &[u8]) -> Result<()> { Ok(crate::durable::write_atomic(path, data)?) } /// Replace a cache file: whole or not at all, but without waiting for the /// disk. A cache checks its own bytes when read, so one a crash leaves /// short or stale is rebuilt rather than trusted. /// How many changes a state index may fall behind its working copy before /// it is written again; until then each load replays them. const INDEX_REPLAY_LIMIT: usize = 16; /// The changes whose encodings are kept for building graphs: a batch of /// files often takes inline text from a few changes in turn. const CHANGE_CACHE_COUNT: usize = 4; const CHANGE_CACHE_BYTES: usize = 512 << 20; /// How many state indexes a workspace keeps. const STATE_INDEXES_KEPT: usize = 2; const INDEX_REPLAY_TIME: std::time::Duration = std::time::Duration::from_millis(50); pub(crate) fn write_cache(path: &Path, data: &[u8]) -> Result<()> { let tmp = path.with_extension(format!("tmp-{}", std::process::id())); fs::write(&tmp, data)?; fs::rename(&tmp, path)?; Ok(()) } pub struct LogEntry { pub hash: ChangeHash, /// Metadata from the state; the full change may be absent in a shallow repository. pub change: crate::state::ChangeMeta, pub is_working_copy: bool, pub is_head: bool, } /// A text file's base graph key, by path, as a walk asks for it. pub(crate) type KeyOf<'a> = dyn Fn(&str) -> Option<[u8; 32]> + Sync + 'a; /// A file as found on disk or in an imported tree. #[derive(Clone)] pub(crate) struct FsEntry { pub kind: NodeKind, pub exec: bool, /// File content, or a symlink's target. Empty when `blob` is set. /// Shared, so a new file's insert holds it without a copy. pub content: std::sync::Arc>, /// The stored blob of an opaque file whose size and mtime are unchanged /// since the last manifest (spec §12.3 stat cache); the content was not /// read. pub blob: Option, /// BLAKE3 of the content, carried over from the manifest when the stat /// is unchanged, so the manifest can be rewritten without rehashing. pub content_hash: Option<[u8; 32]>, /// A text file known equal to the base's rendering without reading it: /// its stat matches the manifest and the manifest recorded it equal to /// a base file graph with the same key (spec §12.3). `content` is empty. pub unchanged: bool, /// The base graph key this file equals, for the next manifest. pub graph_key: Option<[u8; 32]>, /// Size and mtime already known to be current (a checkout leaving the /// file as it was), so the manifest need not stat it again. pub stat: Option<(u64, u128)>, } impl FsEntry { pub(crate) fn new(kind: NodeKind, exec: bool, content: Vec) -> FsEntry { FsEntry { kind, exec, content: std::sync::Arc::new(content), blob: None, content_hash: None, unchanged: false, graph_key: None, stat: None, } } } pub struct Repo { /// Oxygen-pinned subrepos skipped by the running Git export. pub(crate) git_export_skipped: std::cell::Cell, /// The manifest as last read, with the file identity it was read from, /// so one command parses it once (see `read_manifest`). pub(crate) manifest_cache: std::cell::RefCell>, /// A listing of the working copy left by a failed clean check, for the /// snapshot that follows it to reuse; taken by the next walk and /// dropped whenever the manifest is written. pub(crate) listing_cache: std::cell::RefCell>, /// The nested repositories the last walk left out, being neither /// registered nor tracked subrepos (oxygen#43). pub(crate) untracked_nested: std::cell::RefCell>, root: PathBuf, oxy: PathBuf, store: Store, workspace: WorkspaceId, /// A hosted repository with no working copy (spec §12.2, hub spec §4.1). bare: bool, keys_dir: PathBuf, user_config: PathBuf, pub config: Config, /// What the working copy's file system can represent (spec §7.7). platform: crate::portable::Platform, /// The encodings of the changes last read for their inline inserts, /// oldest first. last_change: std::cell::RefCell>)>>, /// `--at-op`: read the repository as it was after this operation; /// recording an operation is refused. at_op: Option, /// The directories (with their times) and nested repositories of the /// last walk, for the manifest written from it. pub(crate) walked_dirs: std::cell::RefCell>, /// The directories a manifest check listed, for its restamp only. pub(crate) listed_dirs: std::cell::RefCell>, } /// A walk's directories with their times, and its nested repositories. pub(crate) type WalkedDirs = (Vec<(String, u128)>, Vec); /// Marker file of a bare repository, inside `.oxy/`. const BARE_MARKER: &str = "bare"; /// Marker of a hosted repository that has no id yet: the first push binds /// it to the pusher's id (spec §16.1). const UNBOUND_MARKER: &str = "unbound"; /// The user's oxy directory: `$OXY_HOME`, else `$XDG_CONFIG_HOME/oxy`, else /// `~/.config/oxy`. It holds `config.toml` (spec §19) and `keys/`. pub fn default_home() -> PathBuf { if let Ok(h) = std::env::var("OXY_HOME") { return PathBuf::from(h); } if let Ok(x) = std::env::var("XDG_CONFIG_HOME") { return PathBuf::from(x).join("oxy"); } let home = std::env::var("HOME").unwrap_or_else(|_| ".".into()); PathBuf::from(home).join(".config").join("oxy") } pub fn default_keys_dir() -> PathBuf { default_home().join("keys") } pub fn default_user_config() -> PathBuf { default_home().join("config.toml") } /// Where `create_identity` records the author. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum IdentityScope { /// The user file: every repository of this user signs with it. User, /// This repository only. Repo, } impl Repo { // ------------------------------------------------------------ lifecycle pub fn init(root: &Path) -> Result { let mut id = [0u8; 16]; rand::rngs::OsRng.fill_bytes(&mut id); Self::init_with_id(root, id) } /// Initialize a repository with a given repository ID (clones share the /// ID of their origin, spec §14.4). pub fn init_with_id(root: &Path, id: [u8; 16]) -> Result { Self::init_on(root, id, None) } /// A new repository with a fresh id, bare or not, on `storage` or in /// `.oxy/objects` (`oxy init [--bare] [--store URL]`). pub fn init_new(root: &Path, bare: bool, storage: Option<&str>) -> Result { let mut id = [0u8; 16]; rand::rngs::OsRng.fill_bytes(&mut id); if bare { Self::init_bare_on(root, id, storage) } else { Self::init_on(root, id, storage) } } /// Initialize a repository whose objects and operation head are kept /// on the store at `storage` (`storage.url`), or in `.oxy/objects`. pub fn init_on(root: &Path, id: [u8; 16], storage: Option<&str>) -> Result { fs::create_dir_all(root)?; let root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf()); let oxy = root.join(OXY_DIR); if oxy.exists() { return fail(format!("{} is already an oxy repository", root.display())); } fs::create_dir_all(oxy.join("ops"))?; write_atomic(&oxy.join("format"), FORMAT_LINE.as_bytes())?; write_atomic(&oxy.join("id"), hex::encode(id).as_bytes())?; let mut config = Config::load(Some(&default_user_config()), &oxy.join("config"))?; if let Some(url) = storage { config.set("storage.url", url)?; } config.save(&oxy.join("config"))?; let workspace = WorkspaceId::random(); write_atomic(&oxy.join("workspace"), workspace.to_hex().as_bytes())?; let store = open_objects(&oxy, &config)?; let repo = Repo { last_change: Default::default(), at_op: None, walked_dirs: Default::default(), listed_dirs: Default::default(), git_export_skipped: Default::default(), manifest_cache: Default::default(), listing_cache: Default::default(), untracked_nested: Default::default(), root, oxy, store, workspace, bare: false, keys_dir: default_keys_dir(), platform: crate::portable::Platform::current(), user_config: default_user_config(), config, }; repo.register_workspace(&workspace, &repo.root)?; let wc = repo.new_working_copy(vec![], ""); let wc_hash = repo.store.put_change(&wc)?; let mut view = View::default(); view.bookmarks.insert( repo.config.bookmark.clone(), repo.store.put_state(&StateObject::new(vec![]))?, ); view.workspaces.insert(workspace, wc_hash); repo.record_op(&[], "oxy init", view)?; Ok(repo) } /// Create a bare repository: an object store, operation log and /// references with no workspace and no working copy, for hosting /// (hub spec §4.1). `root` holds `.oxy/` and nothing else. pub fn init_bare(root: &Path) -> Result { let mut id = [0u8; 16]; rand::rngs::OsRng.fill_bytes(&mut id); Self::init_bare_with_id(root, id) } pub fn init_bare_with_id(root: &Path, id: [u8; 16]) -> Result { Self::init_bare_on(root, id, None) } /// [`Repo::init_on`] for a bare repository. pub fn init_bare_on(root: &Path, id: [u8; 16], storage: Option<&str>) -> Result { fs::create_dir_all(root)?; let root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf()); let oxy = root.join(OXY_DIR); if oxy.exists() { return fail(format!("{} is already an oxy repository", root.display())); } fs::create_dir_all(oxy.join("ops"))?; write_atomic(&oxy.join("format"), FORMAT_LINE.as_bytes())?; write_atomic(&oxy.join("id"), hex::encode(id).as_bytes())?; write_atomic(&oxy.join(BARE_MARKER), b"")?; let mut config = Config::load(Some(&default_user_config()), &oxy.join("config"))?; if let Some(url) = storage { config.set("storage.url", url)?; } config.save(&oxy.join("config"))?; let store = open_objects(&oxy, &config)?; let repo = Repo { last_change: Default::default(), at_op: None, walked_dirs: Default::default(), listed_dirs: Default::default(), git_export_skipped: Default::default(), manifest_cache: Default::default(), listing_cache: Default::default(), untracked_nested: Default::default(), root, oxy, store, workspace: WorkspaceId::ZERO, bare: true, keys_dir: default_keys_dir(), platform: crate::portable::Platform::current(), user_config: default_user_config(), config, }; let mut view = View::default(); view.bookmarks.insert( repo.config.bookmark.clone(), repo.store.put_state(&StateObject::new(vec![]))?, ); repo.record_op(&[], "oxy init --bare", view)?; Ok(repo) } /// A bare repository with no id yet. Its `id()` is all zero until the /// first push binds it (`bind_id`), so an existing repository can be /// pushed to a freshly created hosted one and keep its identity. pub fn init_bare_unbound(root: &Path) -> Result { let repo = Self::init_bare_with_id(root, [0; 16])?; fs::remove_file(repo.oxy.join("id"))?; write_atomic(&repo.oxy.join(UNBOUND_MARKER), b"")?; Ok(repo) } pub fn is_unbound(&self) -> bool { self.oxy.join(UNBOUND_MARKER).is_file() } /// Give an unbound repository its id. pub fn bind_id(&self, id: [u8; 16]) -> Result<()> { if !self.is_unbound() { return fail("this repository already has an id"); } if id == [0; 16] { return fail("a repository id cannot be all zero"); } write_atomic(&self.oxy.join("id"), hex::encode(id).as_bytes())?; fs::remove_file(self.oxy.join(UNBOUND_MARKER))?; Ok(()) } /// Whether this repository has no working copy (`init_bare`). pub fn is_bare(&self) -> bool { self.bare } /// Open the repository containing `start` (searching upwards). pub fn open(start: &Path) -> Result { let start = start.canonicalize()?; let mut dir: Option<&Path> = Some(&start); while let Some(d) = dir { let oxy = d.join(OXY_DIR); if oxy.is_file() { // A secondary workspace (spec §12.2): the file points at the repository. let text = fs::read_to_string(&oxy)?; let Some((shared, workspace)) = workspace::parse_workspace_file(&text) else { return fail(format!("{} is not a valid workspace file", oxy.display())); }; let format = fs::read_to_string(shared.join("format"))?; if format != FORMAT_LINE { return fail(format!("unsupported repository format: {}", format.trim())); } let config = Config::load(Some(&default_user_config()), &shared.join("config"))?; let store = open_objects(&shared, &config)?; let mut repo = Repo { last_change: Default::default(), at_op: None, walked_dirs: Default::default(), listed_dirs: Default::default(), git_export_skipped: Default::default(), manifest_cache: Default::default(), listing_cache: Default::default(), untracked_nested: Default::default(), root: d.to_path_buf(), oxy: shared, store, workspace, bare: false, keys_dir: default_keys_dir(), platform: crate::portable::Platform::current(), user_config: default_user_config(), config, }; repo.apply_workspace_bookmark(); return Ok(repo); } if oxy.is_dir() { let format = fs::read_to_string(oxy.join("format"))?; let config = Config::load(Some(&default_user_config()), &oxy.join("config"))?; let bare = oxy.join(BARE_MARKER).is_file(); let workspace = if bare { WorkspaceId::ZERO } else { WorkspaceId::from_hex(fs::read_to_string(oxy.join("workspace"))?.trim()) .ok_or_else(|| RepoError::Msg("corrupt .oxy/workspace".into()))? }; let store = open_objects(&oxy, &config)?; let mut repo = Repo { last_change: Default::default(), at_op: None, walked_dirs: Default::default(), listed_dirs: Default::default(), git_export_skipped: Default::default(), manifest_cache: Default::default(), listing_cache: Default::default(), untracked_nested: Default::default(), root: d.to_path_buf(), oxy, store, workspace, bare, keys_dir: default_keys_dir(), platform: crate::portable::Platform::current(), user_config: default_user_config(), config, }; if !bare { repo.apply_workspace_bookmark(); } if format == FORMAT1_LINE { repo.migrate_format1()?; } else if format != FORMAT_LINE { return fail(format!("unsupported repository format: {}", format.trim())); } return Ok(repo); } dir = d.parent(); } fail(format!( "no oxy repository found at or above {}; start one with `oxy init` or `oxy clone URL`", start.display() )) } /// Rewrite a format-1 operation log, whose bookmarks were head lists, /// into format 2, where each bookmark names a `state` object. fn migrate_format1(&self) -> Result<()> { let (head, version) = self.read_head()?; // Collect the chain root-first. let mut chain: Vec = Vec::new(); let mut cur = Some(head); while let Some(r) = cur { let op = Op::decode_format1(&self.store.get_typed(&r, ObjectType::Op)?)?; cur = op.parents.first().copied(); chain.push(op); } chain.reverse(); let mut new_parent: Option = None; for old in chain { let mut view = View::default(); for (name, heads) in &old.view.bookmarks { view.bookmarks.insert( name.clone(), self.store.put_state(&StateObject::new(heads.clone()))?, ); } view.workspaces = old.view.workspaces.clone(); let op = Op { parents: new_parent.into_iter().collect(), hlc: old.hlc, command: old.command, view, }; new_parent = Some(self.store.put(ObjectType::Op, &op.encode())?); } if !self.swap_head(&new_parent.unwrap(), version.as_ref())? { return fail("the operation log changed while it was rewritten; run the command again"); } write_atomic(&self.oxy.join("format"), FORMAT_LINE.as_bytes())?; Ok(()) } /// Pack automatically once many loose objects have accumulated. pub(crate) fn maybe_autopack(&self) -> Result<()> { if self.store.loose_count()? > AUTOPACK_THRESHOLD { self.store.pack_loose()?; self.store.merge_packs()?; } Ok(()) } /// The repository ID (spec §14.4). Repositories made before IDs existed get one on first use. pub fn id(&self) -> Result<[u8; 16]> { if self.is_unbound() { return Ok([0; 16]); } let path = self.oxy.join("id"); match fs::read_to_string(&path) { Ok(h) => hex::decode(h.trim()) .ok() .and_then(|v| v.try_into().ok()) .ok_or_else(|| RepoError::Msg("corrupt .oxy/id".into())), Err(e) if e.kind() == std::io::ErrorKind::NotFound => { let mut id = [0u8; 16]; rand::rngs::OsRng.fill_bytes(&mut id); write_atomic(&path, hex::encode(id).as_bytes())?; Ok(id) } Err(e) => Err(e.into()), } } pub fn root(&self) -> &Path { &self.root } pub fn store(&self) -> &Store { &self.store } pub fn workspace(&self) -> WorkspaceId { self.workspace } /// Use a different user directory for keys and the user config file /// (tests, or `$OXY_HOME`). Reloads the effective configuration. /// Act as if the working copy were on another platform's file system /// (tests of Windows and macOS behaviour on any host). pub fn with_platform(mut self, platform: crate::portable::Platform) -> Repo { self.platform = platform; self } pub fn platform(&self) -> crate::portable::Platform { self.platform } pub fn with_home(mut self, dir: impl Into) -> Repo { let dir = dir.into(); self.keys_dir = dir.join("keys"); self.user_config = dir.join("config.toml"); if let Ok(c) = Config::load(Some(&self.user_config), &self.oxy.join("config")) { self.config = c; } self.apply_workspace_bookmark(); self } /// The workspace's current bookmark (spec §10.1) overrides the /// configured default when `oxy bookmark switch` has recorded one. fn apply_workspace_bookmark(&mut self) { if let Ok(b) = fs::read_to_string( self.oxy .join("workspaces") .join(self.workspace.to_hex()) .join("bookmark"), ) { let b = b.trim(); if !b.is_empty() { self.config.bookmark = b.to_string(); } } } pub fn user_config_path(&self) -> &Path { &self.user_config } /// Set a key in the user file and refresh the effective configuration. pub fn set_user_config(&mut self, key: &str, value: &str) -> Result<()> { Config::set_user(&self.user_config, key, value)?; self.config = Config::load(Some(&self.user_config), &self.oxy.join("config"))?; Ok(()) } pub fn save_config(&self) -> Result<()> { self.config.save(&self.oxy.join("config")) } fn author(&self) -> KeyId { self.config.author_key.unwrap_or(ObjectRef::ZERO) } } impl Repo { fn new_working_copy(&self, deps: Vec, message: &str) -> Change { let mut c = Change { version: crate::objects::CHANGE_VERSION, change_id: ChangeId::random(), deps, author: self.author(), hlc: Hlc::now(), message: message.into(), tree_ops: vec![], edits: vec![], kind: ChangeKind::WorkingCopy, }; c.normalize(); c } // ------------------------------------------------------------ operation log pub fn head_op(&self) -> Result<(ObjectRef, Op)> { if let Some(r) = self.at_op { return Ok((r, Op::decode(&self.store.get_typed(&r, ObjectType::Op)?)?)); } let r = self.head_ref()?; let op = Op::decode(&self.store.get_typed(&r, ObjectType::Op)?)?; Ok((r, op)) } pub fn view(&self) -> Result { Ok(self.head_op()?.1.view) } fn record_op(&self, parents: &[ObjectRef], command: &str, view: View) -> Result { if self.at_op.is_some() { return fail( "--at-op shows the repository as it was; this command would change it (run it without --at-op)", ); } let op = Op { parents: parents.to_vec(), hlc: Hlc::now(), command: command.into(), view, }; let r = self.store.put(ObjectType::Op, &op.encode())?; // The head is the commit point: everything it names must be on disk first. let flushing = crate::timing::phase("op: flush the store"); self.store.flush()?; drop(flushing); // Compare and swap (spec §12.1): when another process recorded an // operation since this one read the head, the two are merged // instead of one overwriting the other. On this disk the head moves // under a short lock; on any other store by its record's version, // and a record that moved meanwhile is read again and merged again. let _lock = if self.store.is_local() { Some(self.lock_ops()?) } else { None }; let Some(parent) = parents.first() else { // The first operation (init): there is no head yet. if !self.swap_head(&r, None)? { return fail(format!( "{} already holds a repository's operation log", self.store.backend().describe() )); } return Ok(r); }; let base = Op::decode(&self.store.get_typed(parent, ObjectType::Op)?)?.view; for _ in 0..HEAD_TRIES { let (current, version) = self.read_head()?; let to = if current == *parent { r } else { let theirs = Op::decode(&self.store.get_typed(¤t, ObjectType::Op)?)?.view; let merged = Op { parents: vec![current, r], hlc: Hlc::now(), command: format!("{command} (merged with a concurrent operation)"), view: self.merge_views(&base, &op.view, &theirs)?, }; let m = self.store.put(ObjectType::Op, &merged.encode())?; self.store.flush()?; m }; if self.swap_head(&to, version.as_ref())? { return Ok(to); } } fail("other oxy processes kept changing the operation log while this one recorded; run the command again") } /// The operation head's lock, held only while the head pointer moves. fn lock_ops(&self) -> Result { fs::create_dir_all(self.oxy.join("ops"))?; FileLock::acquire( self.oxy.join("ops").join("lock"), "another oxy process holds the operation log", ) } fn head_ref(&self) -> Result { Ok(self.read_head()?.0) } /// The operation head, and on a store its record's version for /// [`Repo::swap_head`]. On this disk the head is `.oxy/ops/head`; on /// any other store (a bucket, or memory in tests) it is the store's /// [`HEAD_RECORD`] reference record (oxygen#68). fn read_head(&self) -> Result<(ObjectRef, Option)> { if self.store.is_local() { let hex = fs::read_to_string(self.oxy.join("ops").join("head"))?; let r = ObjectRef::from_hex(hex.trim()) .ok_or_else(|| RepoError::Msg("corrupt .oxy/ops/head".into()))?; return Ok((r, None)); } let backend = self.store.backend(); match backend.read_ref(HEAD_RECORD) { Ok(Some((bytes, version))) => { let r = ObjectRef::from_hex(String::from_utf8_lossy(&bytes).trim()).ok_or_else( || { RepoError::Msg(format!( "the operation head in {} is corrupt", backend.describe() )) }, )?; Ok((r, Some(version))) } Ok(None) => fail(format!( "{} holds no operation log (no {HEAD_RECORD} record)", backend.describe() )), Err(e) => fail(format!("read the operation head: {e}")), } } /// Move the operation head to `to` from the `expected` version /// [`Repo::read_head`] gave (`None`: there is no head yet). On this /// disk the caller holds the operations lock and the head always /// moves; on a store it moves only if the record is unchanged, and /// `false` says another process moved it first. fn swap_head(&self, to: &ObjectRef, expected: Option<&crate::strata::Version>) -> Result { if self.store.is_local() { write_atomic(&self.oxy.join("ops").join("head"), to.to_hex().as_bytes())?; return Ok(true); } match self .store .backend() .write_ref(HEAD_RECORD, to.to_hex().as_bytes(), expected) { Ok(_) => Ok(true), Err(crate::strata::Error::Conflict(_)) => Ok(false), Err(e) => fail(format!("move the operation head: {e}")), } } /// Three-way merge of two views that both started from `base` (spec /// §12.1). A reference only one side changed takes that side's value. /// When both changed it differently: a bookmark keeps both sides' /// heads (it may then have several), a change beats a deletion, a /// workspace changed by both is an error (only one of them matches the /// files on disk), and otherwise this process's value wins for remote /// references and the policy, the first value for a tag (tags never /// move). fn merge_views(&self, base: &View, ours: &View, theirs: &View) -> Result { fn pick( b: &BTreeMap, o: &BTreeMap, t: &BTreeMap, mut both: impl FnMut(&V, &V) -> Result, ) -> Result> { let keys: BTreeSet<&K> = b.keys().chain(o.keys()).chain(t.keys()).collect(); let mut out = BTreeMap::new(); for k in keys { let (bv, ov, tv) = (b.get(k), o.get(k), t.get(k)); let v = if ov == tv || tv == bv { ov.cloned() } else if ov == bv { tv.cloned() } else { match (ov, tv) { (Some(x), Some(y)) => Some(both(x, y)?), (Some(x), None) | (None, Some(x)) => Some(x.clone()), (None, None) => None, } }; if let Some(v) = v { out.insert(k.clone(), v); } } Ok(out) } let bookmarks = pick( &base.bookmarks, &ours.bookmarks, &theirs.bookmarks, |x, y| { let mut heads = self.store.get_state(x)?.heads; heads.extend(self.store.get_state(y)?.heads); let heads = self.load_state(&heads)?.maximal(&heads); Ok(self.store.put_state(&StateObject::new(heads))?) }, )?; // Two operations that both replaced one workspace's working copy // cannot both be right: the files on disk match only one of them. // The working-copy lock keeps this from happening; should it happen // anyway, this operation is not recorded. let workspaces = pick( &base.workspaces, &ours.workspaces, &theirs.workspaces, |_, _| { fail( "another oxy process changed this working copy at the same time; \ this command was not recorded: check `oxy status` and run it again", ) }, )?; let tags = pick(&base.tags, &ours.tags, &theirs.tags, |_, y| Ok(*y))?; let empty = BTreeMap::new(); let names: BTreeSet<&String> = base .remotes .keys() .chain(ours.remotes.keys()) .chain(theirs.remotes.keys()) .collect(); let mut remotes = BTreeMap::new(); for n in names { let m = pick( base.remotes.get(n).unwrap_or(&empty), ours.remotes.get(n).unwrap_or(&empty), theirs.remotes.get(n).unwrap_or(&empty), |x, _| Ok(*x), )?; if !m.is_empty() { remotes.insert(n.clone(), m); } } let policy = if ours.policy == theirs.policy || theirs.policy == base.policy { ours.policy } else if ours.policy == base.policy { theirs.policy } else { ours.policy.or(theirs.policy) }; self.store.flush()?; Ok(View { bookmarks, workspaces, remotes, policy, tags, }) } pub fn op_log(&self) -> Result> { let mut out = Vec::new(); let (mut r, mut op) = self.head_op()?; loop { let next = op.parents.first().copied(); out.push((r, op)); match next { Some(p) => { r = p; op = Op::decode(&self.store.get_typed(&r, ObjectType::Op)?)?; } None => break, } } Ok(out) } /// Undo the last user command (spec §12.1). Operations that only recorded /// a working-copy snapshot are skipped, since the files are still on disk /// and the next snapshot recreates them, and a command's own checkout /// (`… (checkout)`) goes with the command. When the working copy is /// clean and was clean after the operation undone to, its files follow /// too, since nothing on disk can be lost; otherwise they stay as they /// are (an undone commit's edits come back as uncommitted work). pub fn undo(&self) -> Result { let _wc = self.lock_working_copy()?; let clean_now = self.clean_by_manifest()? || self.snapshot_status("oxy undo")?.statuses.is_empty(); let (read_head, mut op) = self.head_op()?; let wc_now = op.view.workspaces.get(&self.workspace).copied(); let up = |op: Op| -> Result> { match op.parents.first() { Some(p) => Ok(Some(Op::decode(&self.store.get_typed(p, ObjectType::Op)?)?)), None => Ok(None), } }; loop { while op.is_snapshot() { match up(op.clone())? { Some(p) => op = p, None => break, } } if !op.command.ends_with(" (checkout)") { break; } match up(op.clone())? { Some(p) => op = p, None => break, } } let Some(parent) = op.parents.first() else { return fail("nothing to undo"); }; let parent_op = Op::decode(&self.store.get_typed(parent, ObjectType::Op)?)?; { let _lock = if self.store.is_local() { Some(self.lock_ops()?) } else { None }; let (now, version) = self.read_head()?; if now != read_head || !self.swap_head(parent, version.as_ref())? { return fail("another oxy command ran while undoing; run undo again"); } } self.drop_manifest(); if let Some(then) = parent_op.view.workspaces.get(&self.workspace) { if clean_now && Some(*then) != wc_now { let wc = self.store.get_change(then)?; if wc.edits.is_empty() && wc.tree_ops.is_empty() { let base = self.load_state_history(&wc.deps)?.maximal(&wc.deps); // The files differ from the restored working copy by // exactly what the undone command did: nothing else is // on disk, as the working copy was clean. self.checkout(&base, true, "oxy undo (checkout)")?; } } } Ok(parent_op) } /// The last operation that was not a snapshot. pub fn last_command(&self) -> Result { let (_, mut op) = self.head_op()?; while op.is_snapshot() { let Some(parent) = op.parents.first() else { break; }; op = Op::decode(&self.store.get_typed(parent, ObjectType::Op)?)?; } Ok(op) } // ------------------------------------------------------------ states /// Load the dependency closure of `heads`, through a checkpoint when one applies. pub fn load_state(&self, heads: &[ChangeHash]) -> Result { if let Some(st) = self.load_state_checkpointed(heads)? { return Ok(st); } self.load_state_full(heads) } /// The state of `heads` for its history (changes, tree, metadata) with /// no file graph built: much cheaper than [`Repo::load_state`] for log /// and friends. Build the graphs a command needs with /// [`Repo::load_graphs_of`]. pub fn load_state_history(&self, heads: &[ChangeHash]) -> Result { if let Some(st) = self.load_state_checkpointed_as(heads, true)? { return Ok(st); } self.load_state_replayed(heads, State::new_history_only()) } /// Build these file graphs in a history-only state. pub fn load_graphs_of(&self, state: &mut State, nodes: &[crate::ids::NodeId]) -> Result<()> { Ok(state.load_graphs( nodes, |refs| { self.read_blobs(refs).map_err(|e| { crate::objects::ObjectError::Schema(format!("graph and text blobs: {e}")) }) }, |h| self.change_bytes(h), )?) } /// The graphs of `nodes` of a history-only state, built beside it (see /// [`State::build_graphs_detached`]). pub fn build_graphs_detached( &self, state: &State, nodes: &[crate::ids::NodeId], ) -> Result> { Ok(state .build_graphs_detached( nodes, |refs| { self.read_blobs(refs).map_err(|e| { crate::objects::ObjectError::Schema(format!("graph and text blobs: {e}")) }) }, |h| self.change_bytes(h), )? .into_iter() .collect()) } /// Build every graph a history-only state still lacks, making it full. pub fn load_all_graphs(&self, state: &mut State) -> Result<()> { Ok(state.load_all_graphs( |refs| { self.read_blobs(refs).map_err(|e| { crate::objects::ObjectError::Schema(format!("graph and text blobs: {e}")) }) }, |h| self.change_bytes(h), )?) } /// A change's encoding, for building graphs of its inline inserts. The /// last few read are kept: a checkout building graphs a batch at a /// time reads the same large changes for every batch, often in turn. fn change_bytes( &self, h: &ChangeHash, ) -> std::result::Result>, crate::objects::ObjectError> { let mut cache = self.last_change.borrow_mut(); if let Some(i) = cache.iter().position(|(k, _)| k == h) { let hit = cache.remove(i); let b = hit.1.clone(); cache.push(hit); return Ok(b); } let b = std::sync::Arc::new(self.store.get_typed(h, ObjectType::Change).map_err(|e| { crate::objects::ObjectError::Schema(format!("change {}: {e}", h.short(12))) })?); cache.push((*h, b.clone())); // The last few, within a memory budget; the newest always stays. while cache.len() > CHANGE_CACHE_COUNT || (cache.len() > 1 && cache.iter().map(|(_, b)| b.len()).sum::() > CHANGE_CACHE_BYTES) { cache.remove(0); } Ok(b) } /// Load the dependency closure of `heads` change by change. pub fn load_state_full(&self, heads: &[ChangeHash]) -> Result { self.load_state_replayed(heads, State::new()) } fn load_state_replayed(&self, heads: &[ChangeHash], mut state: State) -> Result { let history = state.is_history_only(); let mut stack: Vec = heads.to_vec(); let mut seen: BTreeSet = BTreeSet::new(); { let _t = crate::timing::phase("load: read and add changes"); let (mut read, mut add, mut fetch) = ( std::time::Duration::ZERO, std::time::Duration::ZERO, std::time::Duration::ZERO, ); while let Some(h) = stack.pop() { // A change the state holds brings its history with it. if !seen.insert(h) || state.contains(&h) { continue; } let t0 = std::time::Instant::now(); // A history-only state reads text only for the graphs it // builds; either way the hash is the one asked for. let (c, texts) = if history { let bytes = self.store.get_typed(&h, ObjectType::Change)?; fetch += t0.elapsed(); Change::decode_skeleton(&bytes, h) .map_err(|e| crate::store::StoreError::Corrupt(h, e.to_string()))? } else { (self.store.get_change(&h)?, Vec::new()) }; read += t0.elapsed(); stack.extend(c.deps.iter().copied()); let t1 = std::time::Instant::now(); state.add_known(h, c, texts); add += t1.elapsed(); } crate::timing::record("load: of which reading", read); crate::timing::record("load: of which fetching", fetch); crate::timing::record("load: of which applying", add); } let _t = crate::timing::phase("load: validate"); state.validate()?; Ok(state) } pub fn working_copy(&self) -> Result<(ChangeHash, Change)> { if self.bare { return fail("this is a bare repository: it has no working copy"); } let view = self.view()?; let h = *view .workspaces .get(&self.workspace) .ok_or_else(|| RepoError::Msg("this workspace is not in the current view".into()))?; Ok((h, self.store.get_change(&h)?)) } /// Heads of a bookmark's state (spec §10.1). pub fn heads_of(&self, view: &View, bookmark: &str) -> Result> { match view.bookmarks.get(bookmark) { Some(st) => Ok(self.store.get_state(st)?.heads), None => Ok(vec![]), } } pub fn bookmark_heads(&self) -> Result> { self.heads_of(&self.view()?, &self.config.bookmark) } // ------------------------------------------------------------ blobs fn blob_for(content: &[u8]) -> Blob { Blob::of(content) } pub fn blob_ref(content: &[u8]) -> ObjectRef { Self::blob_for(content).hash() } pub fn store_blob(&self, content: &[u8]) -> Result { for c in crate::chunk::chunks(content) { self.store.put_chunk(c)?; } Ok(self .store .put(ObjectType::Blob, &Self::blob_for(content).encode())?) } /// Read many blobs at once (checkpoint graphs): the blob objects, then /// all their chunks, are read in parallel; a blob the store lacks goes /// through [`Repo::read_blob`], which may fetch it lazily. pub fn read_blobs(&self, refs: &[ObjectRef]) -> Result>> { let mut blobs: Vec> = Vec::with_capacity(refs.len()); for (r, got) in refs.iter().zip(self.store.get_many(refs)) { blobs.push(match got { Ok((ObjectType::Blob, body)) => Some(Blob::decode(&body)?), Ok((ty, _)) => return fail(format!("{} is a {ty:?}, not a blob", r.short(12))), Err(crate::store::StoreError::NotFound(_)) => None, Err(e) => return Err(e.into()), }); } let chunk_refs: Vec = blobs .iter() .flatten() .flat_map(|b| b.chunks.iter().copied()) .collect(); let mut chunks = self.store.get_many(&chunk_refs).into_iter(); let mut out = Vec::with_capacity(refs.len()); for (r, blob) in refs.iter().zip(blobs) { match blob { None => out.push(self.read_blob(r)?), Some(b) => { let mut content = Vec::with_capacity(b.length as usize); for c in &b.chunks { match chunks.next() { Some(Ok((ObjectType::Chunk, body))) => content.extend(body), Some(Ok(_)) => return fail(format!("{} is not a chunk", c.short(12))), Some(Err(e)) => return Err(e.into()), None => return fail("chunk list ended early"), } } out.push(content); } } } Ok(out) } /// Read a blob's content, fetching it from the remote that withheld it /// when it is lazily absent (spec §13.5). pub fn read_blob(&self, r: &ObjectRef) -> Result> { if !self.store.has(r) && self.lazy_remote(r).is_some() { self.fetch_lazy_blob(r)?; } let blob = Blob::decode(&self.store.get_typed(r, ObjectType::Blob)?)?; let mut out = Vec::with_capacity(blob.length as usize); for c in &blob.chunks { out.extend(self.store.get_typed(c, ObjectType::Chunk)?); } Ok(out) } // ------------------------------------------------------------ working copy /// Attributes from the `.oxyattributes` files in the working directory. pub fn disk_attributes(&self) -> crate::attributes::Attributes { // Nested repositories and ignored paths are left out, as in a walk. let listing = walk::list(&self.root, &|p: &str| is_attributes(p)).unwrap_or_default(); let files: Vec<(String, Vec)> = listing .files .into_iter() .filter_map(|(rel, path, _)| Some((rel, fs::read(path).ok()?))) .collect(); crate::attributes::Attributes::from_files( files.iter().map(|(p, c)| (p.as_str(), c.as_slice())), ) } /// Folders holding a repository of their own that the last look at the /// working copy left out, not being subrepos (oxygen#43). pub fn untracked_nested(&self) -> Vec { self.untracked_nested.borrow().clone() } /// Attributes from a materialized state's `.oxyattributes` files. pub fn state_attributes(snapshot: &Snapshot) -> crate::attributes::Attributes { crate::attributes::Attributes::from_files(snapshot.files.iter().filter_map(|(p, f)| { match &f.content { FileContent::Text { bytes, .. } if p == ".oxyattributes" || p.ends_with("/.oxyattributes") => { Some((p.as_str(), bytes.as_slice())) } _ => None, } })) } /// Store a blob, raw when the path is `nocompress`. pub fn store_blob_for(&self, content: &[u8], nocompress: bool) -> Result { if !nocompress { return self.store_blob(content); } for c in crate::chunk::chunks(content) { self.store.put_chunk_nocompress(c)?; } Ok(self .store .put(ObjectType::Blob, &Self::blob_for(content).encode())?) } /// This workspace's state index (oxygen#41): its working copy's base, /// saved so a snapshot reads it rather than replaying history. /// The state index of the state with these heads: this workspace keeps /// the last few it used, named by their heads, so moving between two /// states finds both. fn state_index_path(&self, heads: &[ChangeHash]) -> PathBuf { let mut heads = heads.to_vec(); heads.sort(); let mut h = blake3::Hasher::new(); for c in &heads { h.update(&c.0); } self.oxy.join("index").join(format!( "{}{}", self.state_index_prefix(), &h.finalize().to_hex()[..16] )) } fn state_index_prefix(&self) -> String { format!("state-{}-", self.workspace.to_hex()) } /// This workspace's state indexes, newest first. fn state_indexes(&self) -> Vec { let prefix = self.state_index_prefix(); let mut found: Vec<(std::time::SystemTime, PathBuf)> = fs::read_dir(self.oxy.join("index")) .into_iter() .flatten() .flatten() .filter(|e| e.file_name().to_string_lossy().starts_with(&prefix)) .filter_map(|e| Some((e.metadata().ok()?.modified().ok()?, e.path()))) .collect(); found.sort_by_key(|f| std::cmp::Reverse(f.0)); found.into_iter().map(|(_, p)| p).collect() } /// The history-only state of `deps` (as [`Repo::load_state_history`]), /// from this workspace's state index when it holds that history: every /// one of `deps` is in it and their heads are its heads. Otherwise it /// is loaded from history and indexed for next time. pub fn load_state_indexed(&self, deps: &[ChangeHash]) -> Result { if let Some((st, heads)) = self.read_state_index(deps) { let mut have = st.maximal(deps); have.sort(); if deps.iter().all(|d| st.contains(d)) && have == heads { #[cfg(test)] assert_eq!( st.write_index(), self.load_state_history(deps)?.write_index(), "a state index holds the state history gives" ); return Ok(st); } // An index a few changes behind (a commit, a pull since): add // what it lacks, as long as all it holds is history of `deps`. // Anything amiss (a change beyond a shallow boundary) takes the // long way instead. let (indexed, started) = (st.len(), std::time::Instant::now()); if let Ok(st) = self.load_state_replayed(deps, st) { let mut all = deps.to_vec(); all.extend(heads.iter().copied()); let (mut within, mut want) = (st.maximal(&all), st.maximal(deps)); within.sort(); want.sort(); if within == want { #[cfg(test)] self.assert_same_state(&st, &self.load_state_history(deps)?); // Writing the index costs far more than replaying a // few small changes: it is rewritten once it falls // behind, or when replaying took long (large changes). if st.len() - indexed > INDEX_REPLAY_LIMIT || started.elapsed() > INDEX_REPLAY_TIME { self.write_state_index(deps, &st); } return Ok(st); } } } let st = self.load_state_history(deps)?; self.write_state_index(deps, &st); Ok(st) } /// Two history-only states hold the same changes and render the same /// files. Their indexes may differ: one loaded from a checkpoint keeps /// the checkpoint's graphs as the bases of its files' histories. #[cfg(test)] fn assert_same_state(&self, a: &State, b: &State) { let ids = |s: &State| s.changes().map(|(h, _)| *h).collect::>(); assert_eq!(ids(a), ids(b), "the same changes"); let files = |s: &State| { let mut s = s.clone(); self.load_all_graphs(&mut s).unwrap(); s.materialize().files }; assert_eq!(files(a), files(b), "the same files"); } /// The state index file: its heads, a hash of its changes block and one /// of the rest, then the body [`State::write_index`] gives, whose first /// block is the changes. Hashed in two parts so the log can read and /// check the changes alone. /// The index of `deps` if there is one, else the newest, which the /// caller may bring up to date. fn read_state_index(&self, deps: &[ChangeHash]) -> Option<(State, Vec)> { let _t = crate::timing::phase("load: state index"); let exact = self.state_index_path(deps); let bytes = match fs::read(&exact) { Ok(b) => { // Used now: the newest, so the last to be dropped. let _ = fs::File::options() .write(true) .open(&exact) .and_then(|f| f.set_modified(std::time::SystemTime::now())); b } Err(_) => fs::read(self.state_indexes().first()?).ok()?, }; let mut r = crate::flat::Reader::new(&bytes); let mut heads = Vec::new(); for _ in 0..r.count().ok()? { heads.push(r.object().ok()?); } let (changes_sum, rest_sum) = (r.object().ok()?, r.object().ok()?); let at = 4 + 32 * heads.len() + 64; let body = bytes.get(at..)?; let split = 4 + r.count().ok()?; // Checked whole, so a part decoded later (the tree's maps, a // file's history) can trust its bytes. let (changes, rest) = (body.get(..split)?, body.get(split..)?); let sum = |b: &[u8]| *blake3::Hasher::new().update_rayon(b).finalize().as_bytes(); if sum(changes) != changes_sum.0 || sum(rest) != rest_sum.0 { return None; } // The histories point into this buffer rather than copying it. Some(( State::read_index_in(&std::sync::Arc::new(bytes), at).ok()?, heads, )) } /// The changes of this workspace's state index (the closure of its /// heads), read without the rest of the index. pub(crate) fn read_state_index_changes( &self, ) -> Option>> { use std::io::Read; let _t = crate::timing::phase("load: state index changes"); let mut f = fs::File::open(self.state_indexes().first()?).ok()?; let mut word = [0u8; 4]; f.read_exact(&mut word).ok()?; let n = u32::from_le_bytes(word) as usize; let mut head = vec![0u8; 32 * n + 64 + 4]; f.read_exact(&mut head).ok()?; let sum = &head[32 * n..32 * n + 32]; let len = u32::from_le_bytes(head[32 * n + 64..].try_into().unwrap()) as usize; let mut changes = vec![0u8; 4 + len]; changes[..4].copy_from_slice(&head[32 * n + 64..]); f.read_exact(&mut changes[4..]).ok()?; let checking = crate::timing::phase("load: of which checking"); let mut h = blake3::Hasher::new(); h.update_rayon(&changes); if h.finalize().as_bytes() != sum { return None; } drop(checking); let _decoding = crate::timing::phase("load: of which decoding"); State::read_index_changes(&changes[4..]).ok() } /// Save `st`, the history-only state of `deps`, as this workspace's /// state index; quietly nothing when it cannot be (it is only a cache). pub(crate) fn write_state_index(&self, deps: &[ChangeHash], st: &State) { let _t = crate::timing::phase("write: state index"); let encoding = crate::timing::phase("write: of which encoding"); let Some(body) = st.write_index() else { return; }; drop(encoding); #[cfg(test)] assert_eq!( State::read_index(&body).ok().and_then(|s| s.write_index()), Some(body.clone()), "a state index reads back as written" ); let mut heads = st.maximal(deps); heads.sort(); let mut w = crate::flat::Writer::default(); w.count(heads.len()); for h in &heads { w.object(h); } let split = 4 + u32::from_le_bytes(body[..4].try_into().unwrap()) as usize; w.fixed( blake3::Hasher::new() .update_rayon(&body[..split]) .finalize() .as_bytes(), ); let mut sum = blake3::Hasher::new(); sum.update_rayon(&body[split..]); w.fixed(sum.finalize().as_bytes()); w.fixed(&body); let path = self.state_index_path(&heads); let _ = fs::create_dir_all(self.oxy.join("index")); let _ = write_cache(&path, &w.0); // The last two used: the index of a state moved away from stays for // a move back. Older ones (and the unnamed one of before) go. for old in self.state_indexes().iter().skip(STATE_INDEXES_KEPT) { let _ = fs::remove_file(old); } let _ = fs::remove_file( self.oxy .join("index") .join(format!("state-{}", self.workspace.to_hex())), ); } fn locks_dir(&self) -> PathBuf { self.oxy.join("locks") } /// Whether `path` is locked locally for editing (spec §18, advisory). pub fn is_locked(&self, path: &str) -> bool { self.locks_dir().join(path.replace('/', "%2F")).exists() } /// Take the advisory local lock on a `lockable` file and make it writable. pub fn lock_path(&self, path: &str) -> Result<()> { let attrs = self.disk_attributes(); if !attrs.lookup(path).lockable { return fail(format!("{path} is not marked lockable in .oxyattributes")); } fs::create_dir_all(self.locks_dir())?; write_atomic( &self.locks_dir().join(path.replace('/', "%2F")), std::process::id().to_string().as_bytes(), )?; set_readonly(&self.root.join(path), false)?; Ok(()) } /// Release the local lock and make the file read-only again. pub fn unlock_path(&self, path: &str) -> Result<()> { let _ = fs::remove_file(self.locks_dir().join(path.replace('/', "%2F"))); if self.root.join(path).exists() { set_readonly(&self.root.join(path), true)?; } Ok(()) } /// Whether a repository path is inside the sparse filter (spec §17). pub fn in_scope(&self, path: &str) -> bool { self.config.sparse_paths.is_empty() || self .config .sparse_paths .iter() .any(|p| path == p || path.starts_with(&format!("{p}/"))) } fn walk_working_copy(&self) -> Result> { self.walk_working_copy_with(None, &BTreeMap::new(), None) } /// Walk the working copy; with a manifest, a file whose size and mtime /// match it is not read: an opaque file yields its known blob, and a /// text file whose recorded graph key equals its base graph's key in /// `keys` (path → key) is marked unchanged; other text files are read /// but keep their known hash. Files modified within two seconds of the /// manifest are never trusted (a racy stat). /// /// `.oxyattributes` files are always read, since their content is /// needed; and when any of them is not known unchanged (added, removed, /// edited, or changed in the base), no text file is taken as unchanged, /// because attributes decide how a file compares with the base. /// /// `.oxyignore` applies to untracked paths only: every path in `tracked` /// (the base state's files) that exists on disk is included even when /// an ignore rule matches it, as in Git. fn walk_working_copy_with( &self, cache: Option<&prompt::Manifest>, tracked: &BTreeMap)>, keys: Option<&KeyOf>, ) -> Result> { let mut out: BTreeMap; let sparse = &self.config.sparse_paths; let in_scope = |p: &str| { sparse.is_empty() || sparse .iter() .any(|s| p == s || p.starts_with(&format!("{s}/"))) }; let walk_phase = crate::timing::phase("walk: of which listing"); let walk::Listing { files, symlinks, nested, dirs, } = match self.listing_cache.take() { Some(l) => l, // Ignore rules apply to untracked paths only, so the base's // files are not matched against them (an added file is: a rule // written after it was snapshotted leaves it out again). None => walk::list_knowing(&self.root, &in_scope, &|p: &str| tracked.contains_key(p))?, }; let keys = keys.filter(|k| { let on_disk: HashSet<&str> = files.iter().map(|(r, _, _)| r.as_str()).collect(); let attrs_unchanged = files .iter() .filter(|(r, _, _)| is_attributes(r)) .all(|(rel, _, meta)| { stat_cached(rel, meta, cache) .is_some_and(|e| e.graph.is_some() && e.graph == k(rel)) }); let none_removed = tracked .keys() .filter(|r| is_attributes(r)) .all(|r| on_disk.contains(r.as_str())); attrs_unchanged && none_removed }); *self.walked_dirs.borrow_mut() = Some((dirs, nested.iter().map(|d| self.rel_path(d)).collect())); drop(walk_phase); let _read_phase = crate::timing::phase("walk: of which stat and compare"); // Reading files is most of a walk's cost; do it in parallel. { use rayon::prelude::*; let mut read: Vec<(String, FsEntry)> = files .par_iter() .map(|(rel, path, meta)| { // A key is worked out only for a file the manifest // knows with the same stat and a key of its own. let key = keys .filter(|_| !is_attributes(rel)) .filter(|_| { stat_cached(rel, meta, cache).is_some_and(|e| e.graph.is_some()) }) .and_then(|k| k(rel)); Ok((rel.clone(), read_file_entry(rel, path, meta, cache, key)?)) }) .collect::>>()?; for (rel, e) in read.iter_mut() { // A cached blob the store lost (gc, a partial clone): read the file after all. if e.content.is_empty() && e.blob.is_some_and(|b| !self.store.has(&b)) { e.content = std::sync::Arc::new(fs::read(self.root.join(rel.as_str()))?); e.blob = None; } } // The listing is in path order, so the map is built without // searching. out = read.into_iter().collect(); for (rel, target) in symlinks { out.insert(rel, FsEntry::new(NodeKind::Symlink, false, target)); } } drop(_read_phase); let _rest = crate::timing::phase("walk: of which subrepos and hidden"); // Nested repositories (spec §7.8): registered in `.oxymodules` or // already tracked as subrepo nodes, they contribute their pin; // unregistered ones are left alone. let modules = self.modules_on_disk(); let removed = self.removed_subrepos(); for rel in &removed { if tracked.get(rel).map(|(k, _)| *k) != Some(NodeKind::Subrepo) { self.clear_removed_subrepo(rel); } } let mut untracked_nested = Vec::new(); for dir in nested { let rel = self.rel_path(&dir); if !self.in_scope(&rel) || removed.contains(&rel) { continue; } let registered = modules.entries.get(&rel); if registered.is_none() && tracked.get(&rel).map(|(k, _)| *k) != Some(NodeKind::Subrepo) { untracked_nested.push(rel); continue; } let pin = subrepos::read_pin(&dir, registered.and_then(|m| m.bookmark.as_deref()))?; out.insert(rel, FsEntry::new(NodeKind::Subrepo, false, pin.encode())); } untracked_nested.sort(); *self.untracked_nested.borrow_mut() = untracked_nested; // Tracked paths the ignore rules hid. for (rel, (kind, blob)) in tracked { if out.contains_key(rel) || !self.in_scope(rel) { continue; } let path = self.root.join(rel); if *kind == NodeKind::Subrepo { if removed.contains(rel) { continue; } let present = path.is_dir() && (path.join(OXY_DIR).exists() || path.join(".git").exists()); if present { let pin = subrepos::read_pin( &path, modules.entries.get(rel).and_then(|m| m.bookmark.as_deref()), )?; out.insert( rel.clone(), FsEntry::new(NodeKind::Subrepo, false, pin.encode()), ); } else if let Some(b) = blob { // Not cloned (or emptied): the pin stays as recorded, as a // Git gitlink does when its directory is absent. out.insert( rel.clone(), FsEntry { blob: Some(*b), ..FsEntry::new(NodeKind::Subrepo, false, vec![]) }, ); } continue; } let Ok(meta) = fs::symlink_metadata(&path) else { continue; }; let ft = meta.file_type(); if ft.is_symlink() { let target = fs::read_link(&path)?; out.insert( rel.clone(), FsEntry::new( NodeKind::Symlink, false, target.to_string_lossy().as_bytes().to_vec(), ), ); } else if ft.is_file() { let e = self.file_entry(rel, &path, &walk::Stat::of(&meta), cache)?; out.insert(rel.clone(), e); } } Ok(out) } /// The entry for one regular file, read or served from the stat cache. fn file_entry( &self, rel: &str, path: &Path, meta: &walk::Stat, cache: Option<&prompt::Manifest>, ) -> Result { let mut e = read_file_entry(rel, path, meta, cache, None)?; if e.content.is_empty() && e.blob.is_some_and(|b| !self.store.has(&b)) { e.content = std::sync::Arc::new(fs::read(path)?); e.blob = None; } Ok(e) } /// The non-directory paths of a snapshot with their kinds, for the /// walker's `tracked` set. pub(crate) fn tracked_kinds( snapshot: &Snapshot, ) -> BTreeMap)> { snapshot .files .iter() .filter(|(_, f)| f.kind != NodeKind::Dir) .map(|(p, f)| { let blob = match &f.content { FileContent::Opaque { blobs } => blobs.first().copied(), _ => None, }; (p.clone(), (f.kind, blob)) }) .collect() } /// A path under the root as a `/`-separated relative string. pub(crate) fn rel_path(&self, path: &Path) -> String { path.strip_prefix(&self.root) .unwrap_or(path) .components() .map(|c| c.as_os_str().to_string_lossy().to_string()) .collect::>() .join("/") } } impl Repo { /// Changes reachable from the bookmark and the working copy, newest first. pub fn log(&self) -> Result> { let heads = self.bookmark_heads()?; if self.bare { return self.log_state(&heads); } let (wc_hash, wc) = self.working_copy()?; let mut all = heads.clone(); all.push(wc_hash); let head_set: BTreeSet = heads.into_iter().collect(); let mut entries: Vec = Vec::new(); // A shallow clone knows changes beyond its boundary from a // checkpoint's metadata only: its log comes from the state. let metas = if self.is_shallow() { self.load_state_history(&all)? .changes() .map(|(h, m)| (*h, m.clone())) .collect() } else { // The working copy's change is in hand, and is never in the // state index: its metadata is not looked up again. let mut below: Vec = head_set.iter().copied().collect(); below.extend(wc.deps.iter().copied()); let _t = crate::timing::phase("log: metadata"); let mut metas = self.metas_of(&below)?; if !metas.iter().any(|(h, _)| *h == wc_hash) { metas.push((wc_hash, crate::state::ChangeMeta::of(&wc))); } metas }; for (h, meta) in metas { entries.push(LogEntry { hash: h, change: meta, is_working_copy: h == wc_hash, is_head: head_set.contains(&h), }); } let _sorting = crate::timing::phase("log: sort"); // The hash makes the order total, so an unstable sort gives the same. entries.sort_unstable_by(|a, b| { b.is_working_copy .cmp(&a.is_working_copy) .then(b.change.hlc.cmp(&a.change.hlc)) .then(b.hash.cmp(&a.hash)) }); Ok(entries) } /// A window of `log_state`: entries `offset..offset+limit` and the total /// count, for paginated views. pub fn log_window( &self, heads: &[ChangeHash], offset: usize, limit: usize, ) -> Result<(Vec, usize)> { let all = self.log_state(heads)?; let total = all.len(); Ok((all.into_iter().skip(offset).take(limit).collect(), total)) } /// The log of the state with these heads, newest first, with no working /// copy involved (for hosted repositories and any explicit state). pub fn log_state(&self, heads: &[ChangeHash]) -> Result> { // The log needs change metadata only: neither graphs nor the tree. if self.is_shallow() { return Ok(self.log_of(&self.load_state_history(heads)?)); } let metas = self.metas_of(heads)?; let depended: BTreeSet = metas .iter() .flat_map(|(_, m)| m.deps.iter().copied()) .collect(); let mut entries: Vec = metas .into_iter() .map(|(h, meta)| LogEntry { hash: h, is_head: !depended.contains(&h), change: meta, is_working_copy: false, }) .collect(); entries.sort_unstable_by(|a, b| b.change.hlc.cmp(&a.change.hlc).then(b.hash.cmp(&a.hash))); Ok(entries) } /// Every change of the state with these heads and its metadata, read /// without its operations ([`Store::get_change_meta`]). A change beyond /// a shallow clone's boundary is left out. pub fn metas_of( &self, heads: &[ChangeHash], ) -> Result> { let boundary = self.boundary(); // The state index holds the metadata of its heads' closure: a // change found there is not read from the store (a change object // can be as large as the tree it adds). let mut indexed = self.read_state_index_changes(); let _walk = crate::timing::phase("metadata: walk"); let (mut stored, mut from_store) = (0usize, std::time::Duration::ZERO); let mut seen: crate::fasthash::FastSet = Default::default(); let mut stack: Vec = heads.to_vec(); let mut out = Vec::with_capacity(indexed.as_ref().map_or(0, |m| m.len())); while let Some(h) = stack.pop() { if !seen.insert(h) { continue; } // Taken out of the index's map, each once: not copied. if let Some(meta) = indexed.as_mut().and_then(|m| m.remove(&h)) { #[cfg(test)] assert_eq!( *meta, self.store.get_change_meta(&h)?, "the state index holds a change's metadata" ); let meta = std::sync::Arc::unwrap_or_clone(meta); stack.extend(meta.deps.iter().copied()); out.push((h, meta)); continue; } let started = std::time::Instant::now(); let meta = match self.store.get_change_meta(&h) { Ok(m) => m, Err(crate::store::StoreError::NotFound(_)) if boundary.contains(&h) => continue, Err(e) => return Err(e.into()), }; stored += 1; from_store += started.elapsed(); stack.extend(meta.deps.iter().copied()); out.push((h, meta)); } if stored > 0 && crate::timing::enabled() { eprintln!("timing metadata: {stored} changes read whole, {from_store:.2?}"); } Ok(out) } /// The log of a state already loaded, newest first. /// The history of lines `first..=last` (1-based) of `path` in the /// state `heads` make (spec §27): every change that inserted or /// deleted one of their atoms, or an atom deleted from between them, /// newest first. Exact, as atoms keep their identity across edits. pub fn line_history( &self, heads: &[ChangeHash], path: &str, first: usize, last: usize, ) -> Result> { if first == 0 || last < first { return fail(format!( "lines {first},{last}: expected FIRST,LAST from 1 with FIRST ≤ LAST" )); } let mut state = self.load_state_history(heads)?; let Some(node) = state.tree().view().entries.get(path).map(|e| e.node) else { return fail(format!("{path} is not in that state")); }; self.load_graphs_of(&mut state, &[node])?; let live = state.live_lines(&node); if last > live.len() { return fail(format!( "{path} has {} lines; asked for {first},{last}", live.len() )); } let Some(g) = state.file(&node) else { return fail(format!("{path} is not a text file")); }; let atoms = g.between(&live[first - 1].0, &live[last - 1].0); let touched = g.touching(&atoms); Ok(self .log_of(&state) .into_iter() .filter(|e| touched.contains(&e.hash)) .collect()) } pub fn log_of(&self, state: &crate::state::State) -> Vec { let head_set: BTreeSet = state.heads().into_iter().collect(); let mut entries: Vec = state .changes() .map(|(h, meta)| LogEntry { hash: *h, change: meta.clone(), is_working_copy: false, is_head: head_set.contains(h), }) .collect(); entries.sort_unstable_by(|a, b| b.change.hlc.cmp(&a.change.hlc).then(b.hash.cmp(&a.hash))); entries } /// Resolve a spec to a set of heads: a bookmark's name, a remote's /// bookmark as `REMOTE/NAME`, a tag name, or a single change /// (`resolve_change`). pub fn resolve_heads(&self, spec: &str) -> Result> { let view = self.view()?; if let Some(st) = view.bookmarks.get(spec) { return Ok(self.store.get_state(st)?.heads); } if let Some((_, _, st)) = remote_bookmark(&view, spec) { return Ok(self.store.get_state(&st)?.heads); } if let Some(heads) = self.tag_heads(spec)? { return Ok(heads); } Ok(vec![self.resolve_change(spec)?]) } /// Resolve a change by ChangeId prefix, `#hash` prefix, `@` (working /// copy), bookmark, `REMOTE/NAME` or tag (the one head of its state). pub fn resolve_change(&self, spec: &str) -> Result { if spec == "@" { return Ok(self.working_copy()?.0); } let view = self.view()?; let named = match view.bookmarks.get(spec) { Some(st) => Some(("bookmark", self.store.get_state(st)?.heads)), None => match remote_bookmark(&view, spec) { Some((_, _, st)) => Some(("bookmark", self.store.get_state(&st)?.heads)), None => self.tag_heads(spec)?.map(|h| ("tag", h)), }, }; if let Some((kind, heads)) = named { return match heads.len() { 1 => Ok(heads[0]), n => fail(format!("{kind} {spec} has {n} heads; name a change, or use it where a state is expected")), }; } if spec.is_empty() || spec == "#" { return fail("empty change specifier"); } let entries = self.log()?; let matches: Vec<&LogEntry> = if let Some(hp) = spec.strip_prefix('#') { entries .iter() .filter(|e| e.hash.starts_with_hex(hp)) .collect() } else { entries .iter() .filter(|e| e.change.change_id.starts_with_hex(spec)) .collect() }; match matches.len() { 1 => Ok(matches[0].hash), 0 => match self.resolve_in_remotes(spec)? { Some(h) => Ok(h), None => fail(format!("no change matches {spec:?}")), }, _ => { let ids: BTreeSet = matches.iter().map(|e| e.change.change_id).collect(); if ids.len() == 1 { fail(format!( "change {spec} is divergent: {} variants", matches.len() )) } else { fail(format!("{spec:?} is ambiguous")) } } } } /// A change no local bookmark holds, found by id or `#hash` prefix in /// the remote bookmarks' states: fetched but not yet pulled (oxygen#56). /// `None` when none matches, or more than one. fn resolve_in_remotes(&self, spec: &str) -> Result> { let view = self.view()?; let mut found: BTreeSet = BTreeSet::new(); for bookmarks in view.remotes.values() { for st in bookmarks.values() { let heads = self.store.get_state(st)?.heads; for e in self.log_state(&heads)? { let hit = match spec.strip_prefix('#') { Some(hp) => e.hash.starts_with_hex(hp), None => e.change.change_id.starts_with_hex(spec), }; if hit { found.insert(e.hash); } } } } Ok((found.len() == 1).then(|| *found.first().unwrap())) } /// Where a change is when no local bookmark or the working copy holds /// it: the first remote bookmark that does (`origin/main`). `None` when /// it is held locally, or nowhere. pub fn remote_only_location(&self, hash: &ChangeHash) -> Result> { if self.log()?.iter().any(|e| e.hash == *hash) { return Ok(None); } let view = self.view()?; for (remote, bookmarks) in &view.remotes { for (name, st) in bookmarks { let heads = self.store.get_state(st)?.heads; if self.log_state(&heads)?.iter().any(|e| e.hash == *hash) { return Ok(Some(format!("{remote}/{name}"))); } } } Ok(None) } /// Human-readable description of what a change does, with line content. pub fn describe_change(&self, hash: &ChangeHash) -> Result { let change = self.store.get_change(hash)?; let (base, base_snapshot) = self.change_base(&change)?; Ok(render_change(&change, &base, &base_snapshot)) } /// The base state of `change` (its dependencies) and its rendering, as /// much as showing the change needs: the whole tree, and the graphs and /// contents of the files the change edits. Other text files are left /// unrendered ([`Snapshot::unrendered`]). pub fn change_base(&self, change: &Change) -> Result<(State, Snapshot)> { let base = self.load_state_history(&change.deps)?; self.render_base_for(change, base) } /// [`Repo::change_base`] with the base already loaded (the working /// copy's comes from the state index). pub(crate) fn render_base_for( &self, change: &Change, mut base: State, ) -> Result<(State, Snapshot)> { let _t = crate::timing::phase("diff: render the base"); let edited: std::collections::HashSet = change.edits.iter().map(|e| e.node()).collect(); let nodes: Vec = edited.iter().copied().collect(); self.load_graphs_of(&mut base, &nodes)?; let snapshot = { let _t = crate::timing::phase("diff: of which materialize"); base.materialize_nodes(&edited) }; Ok((base, snapshot)) } } fn remove_any(path: &Path) -> std::io::Result<()> { let m = fs::symlink_metadata(path)?; if m.is_dir() { fs::remove_dir_all(path) } else { fs::remove_file(path) } } fn set_readonly(path: &Path, readonly: bool) -> std::io::Result<()> { let mut perms = fs::metadata(path)?.permissions(); perms.set_readonly(readonly); fs::set_permissions(path, perms) } #[cfg(unix)] fn set_executable(path: &Path, exec: bool) -> std::io::Result<()> { use std::os::unix::fs::PermissionsExt; let mut perms = fs::metadata(path)?.permissions(); let mode = perms.mode(); let want = if exec { mode | 0o111 } else { mode & !0o111 }; if want != mode { perms.set_mode(want); fs::set_permissions(path, perms)?; } Ok(()) } #[cfg(not(unix))] fn set_executable(_path: &Path, _exec: bool) -> std::io::Result<()> { Ok(()) } #[cfg(unix)] pub(crate) fn is_executable(m: &fs::Metadata) -> bool { use std::os::unix::fs::PermissionsExt; m.permissions().mode() & 0o111 != 0 } #[cfg(not(unix))] pub(crate) fn is_executable(_m: &fs::Metadata) -> bool { false } fn op_ref(k: OpKey) -> OpRef { OpRef { change: k.0, index: k.1, } } fn split_path(path: &str) -> (&str, &str) { match path.rfind('/') { Some(i) => (&path[..i], &path[i + 1..]), None => ("", path), } } /// Whether `rel` is an attributes file (spec §7.6). fn is_attributes(rel: &str) -> bool { rel == ".oxyattributes" || rel.ends_with("/.oxyattributes") } /// The manifest entry for `rel` when its size and mtime match it outside /// the racy window. fn stat_cached<'m>( rel: &str, meta: &walk::Stat, cache: Option<&'m prompt::Manifest>, ) -> Option<&'m prompt::ManifestEntry> { let m = cache?; let e = m.entries.get(rel)?; let racy = meta.mtime_ns + 2_000_000_000 >= m.written_ns; (e.size == meta.size && e.mtime_ns == meta.mtime_ns && !racy).then_some(e) } /// Read one working-copy file, or serve it from the stat cache (spec /// §12.3). With a manifest whose stat for it is unchanged (and not racy), /// an opaque file is not read at all (its stored blob stands in), and a /// text file the manifest recorded equal to a base graph whose key is /// `key`, the key the base has now, is not read either: equal keys mean /// equal renderings, so it still equals the base. Touches no repository /// state, so the walk runs it in parallel; the caller checks that a cached /// blob is still in the store. fn read_file_entry( rel: &str, path: &Path, meta: &walk::Stat, cache: Option<&prompt::Manifest>, key: Option<[u8; 32]>, ) -> Result { let exec = meta.exec; let stat = Some((meta.size, meta.mtime_ns)); let mut e = match stat_cached(rel, meta, cache) { Some(e) if e.blob.is_some() => FsEntry { blob: e.blob, content_hash: Some(e.hash), ..FsEntry::new(NodeKind::File, exec, vec![]) }, Some(e) if key.is_some() && e.graph == key => FsEntry { content_hash: Some(e.hash), unchanged: true, graph_key: e.graph, ..FsEntry::new(NodeKind::File, exec, vec![]) }, Some(e) => FsEntry { content_hash: Some(e.hash), ..FsEntry::new(NodeKind::File, exec, fs::read(path)?) }, None => FsEntry::new(NodeKind::File, exec, fs::read(path)?), }; // The stat the manifest records: taken before the read, so a file that // changes meanwhile is read again next time. e.stat = stat; Ok(e) }