// Copyright 2020 The Jujutsu Authors//// Licensed under the Apache License, Version 2.0 (the "License");// you may not use this file except in compliance with the License.// You may obtain a copy of the License at//// https://www.apache.org/licenses/LICENSE-2.0//// Unless required by applicable law or agreed to in writing, software// distributed under the License is distributed on an "AS IS" BASIS,// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.// See the License for the specific language governing permissions and// limitations under the License. #![expect(missing_docs)] use std::collections::BTreeMap;use std::collections::HashMap;use std::collections::HashSet;use std::fmt::Debug;use std::fs;use std::io;use std::io::ErrorKind;use std::io::Write as _;use std::path::Path;use std::path::PathBuf;use std::time::SystemTime; use async_trait::async_trait;use itertools::Itertools as _;use pollster::FutureExt as _;use prost::Message as _;use smallvec::SmallVec;use tempfile::NamedTempFile;use thiserror::Error; use crate::backend::BackendInitError;use crate::backend::CommitId;use crate::backend::MillisSinceEpoch;use crate::backend::Timestamp;use crate::content_hash::blake2b_hash;use crate::dag_walk_async;use crate::file_util::IoResultExt as _;use crate::file_util::PathError;use crate::file_util::persist_content_addressed_temp_file;use crate::merge::Merge;use crate::object_id::HexPrefix;use crate::object_id::ObjectId;use crate::object_id::PrefixResolution;use crate::op_store;use crate::op_store::OpStore;use crate::op_store::OpStoreError;use crate::op_store::OpStoreResult;use crate::op_store::Operation;use crate::op_store::OperationId;use crate::op_store::OperationMetadata;use crate::op_store::RefTarget;use crate::op_store::RemoteRef;use crate::op_store::RemoteRefState;use crate::op_store::RemoteView;use crate::op_store::RootOperationData;use crate::op_store::TimestampRange;use crate::op_store::View;use crate::op_store::ViewId;use crate::ref_name::GitRefNameBuf;use crate::ref_name::RefNameBuf;use crate::ref_name::RemoteNameBuf;use crate::ref_name::WorkspaceName;use crate::ref_name::WorkspaceNameBuf; // BLAKE2b-512 hash length in bytesconst OPERATION_ID_LENGTH: usize = 64;const VIEW_ID_LENGTH: usize = 64; /// Error that may occur during [`SimpleOpStore`] initialization.#[derive(Debug, Error)]#[error("Failed to initialize simple operation store")]pub struct SimpleOpStoreInitError(#[from] pub PathError); impl From<SimpleOpStoreInitError> for BackendInitError {    fn from(err: SimpleOpStoreInitError) -> Self {        Self(err.into())    }} #[derive(Debug)]pub struct SimpleOpStore {    path: PathBuf,    root_data: RootOperationData,    root_operation_id: OperationId,    root_view_id: ViewId,} impl SimpleOpStore {    pub fn name() -> &'static str {        "simple_op_store"    }     /// Creates an empty OpStore. Returns error if it already exists.    pub fn init(        store_path: &Path,        root_data: RootOperationData,    ) -> Result<Self, SimpleOpStoreInitError> {        let store = Self::new(store_path, root_data);        store.init_base_dirs()?;        Ok(store)    }     /// Load an existing OpStore    pub fn load(store_path: &Path, root_data: RootOperationData) -> Self {        Self::new(store_path, root_data)    }     fn new(store_path: &Path, root_data: RootOperationData) -> Self {        Self {            path: store_path.to_path_buf(),            root_data,            root_operation_id: OperationId::from_bytes(&[0; OPERATION_ID_LENGTH]),            root_view_id: ViewId::from_bytes(&[0; VIEW_ID_LENGTH]),        }    }     fn init_base_dirs(&self) -> Result<(), PathError> {        for dir in [self.views_dir(), self.operations_dir()] {            fs::create_dir(&dir).context(&dir)?;        }        Ok(())    }     fn views_dir(&self) -> PathBuf {        self.path.join("views")    }     fn operations_dir(&self) -> PathBuf {        self.path.join("operations")    }} #[async_trait]impl OpStore for SimpleOpStore {    fn name(&self) -> &str {        Self::name()    }     fn root_operation_id(&self) -> &OperationId {        &self.root_operation_id    }     async fn read_view(&self, id: &ViewId) -> OpStoreResult<View> {        if *id == self.root_view_id {            return Ok(View::make_root(self.root_data.root_commit_id.clone()));        }         let path = self.views_dir().join(id.hex());        let buf = fs::read(&path)            .context(&path)            .map_err(|err| io_to_read_error(err, id))?;         let proto = crate::protos::simple_op_store::View::decode(&*buf)            .map_err(|err| to_read_error(err.into(), id))?;        view_from_proto(proto).map_err(|err| to_read_error(err.into(), id))    }     async fn write_view(&self, view: &View) -> OpStoreResult<ViewId> {        let dir = self.views_dir();        let temp_file = NamedTempFile::new_in(&dir)            .context(&dir)            .map_err(|err| io_to_write_error(err, "view"))?;         let proto = view_to_proto(view);        temp_file            .as_file()            .write_all(&proto.encode_to_vec())            .context(temp_file.path())            .map_err(|err| io_to_write_error(err, "view"))?;         let id = ViewId::new(blake2b_hash(view).to_vec());         let new_path = dir.join(id.hex());        persist_content_addressed_temp_file(temp_file, &new_path)            .context(&new_path)            .map_err(|err| io_to_write_error(err, "view"))?;        Ok(id)    }     async fn read_operation(&self, id: &OperationId) -> OpStoreResult<Operation> {        if *id == self.root_operation_id {            return Ok(Operation::make_root(self.root_view_id.clone()));        }         let path = self.operations_dir().join(id.hex());        let buf = fs::read(&path)            .context(&path)            .map_err(|err| io_to_read_error(err, id))?;         let proto = crate::protos::simple_op_store::Operation::decode(&*buf)            .map_err(|err| to_read_error(err.into(), id))?;        let mut operation =            operation_from_proto(proto).map_err(|err| to_read_error(err.into(), id))?;        if operation.parents.is_empty() {            // Repos created before we had the root operation will have an operation without            // parents.            operation.parents.push(self.root_operation_id.clone());        }        Ok(operation)    }     async fn write_operation(&self, operation: &Operation) -> OpStoreResult<OperationId> {        assert!(!operation.parents.is_empty());        let dir = self.operations_dir();        let temp_file = NamedTempFile::new_in(&dir)            .context(&dir)            .map_err(|err| io_to_write_error(err, "operation"))?;         let proto = operation_to_proto(operation);        temp_file            .as_file()            .write_all(&proto.encode_to_vec())            .context(temp_file.path())            .map_err(|err| io_to_write_error(err, "operation"))?;         let id = OperationId::new(blake2b_hash(operation).to_vec());         let new_path = dir.join(id.hex());        persist_content_addressed_temp_file(temp_file, &new_path)            .context(&new_path)            .map_err(|err| io_to_write_error(err, "operation"))?;        Ok(id)    }     async fn resolve_operation_id_prefix(        &self,        prefix: &HexPrefix,    ) -> OpStoreResult<PrefixResolution<OperationId>> {        let op_dir = self.operations_dir();        let find = || -> io::Result<_> {            let matches_root = prefix.matches(&self.root_operation_id);            let hex_prefix = prefix.hex();            if hex_prefix.len() == OPERATION_ID_LENGTH * 2 {                // Fast path for full-length ID                if matches_root || op_dir.join(hex_prefix).try_exists()? {                    let id = OperationId::from_bytes(prefix.as_full_bytes().unwrap());                    return Ok(PrefixResolution::SingleMatch(id));                } else {                    return Ok(PrefixResolution::NoMatch);                }            }             let mut matched = matches_root.then(|| self.root_operation_id.clone());            for entry in op_dir.read_dir()? {                let Ok(name) = entry?.file_name().into_string() else {                    continue; // Skip invalid UTF-8                };                if !name.starts_with(&hex_prefix) {                    continue;                }                let Some(id) = OperationId::try_from_hex(&name) else {                    continue; // Skip invalid hex                };                if matched.is_some() {                    return Ok(PrefixResolution::AmbiguousMatch);                }                matched = Some(id);            }            if let Some(id) = matched {                Ok(PrefixResolution::SingleMatch(id))            } else {                Ok(PrefixResolution::NoMatch)            }        };        find()            .context(&op_dir)            .map_err(|err| OpStoreError::Other(err.into()))    }     #[tracing::instrument(skip(self))]    async fn gc(&self, head_ids: &[OperationId], keep_newer: SystemTime) -> OpStoreResult<()> {        let to_op_id = |entry: &fs::DirEntry| -> Option<OperationId> {            let name = entry.file_name().into_string().ok()?;            OperationId::try_from_hex(name)        };        let to_view_id = |entry: &fs::DirEntry| -> Option<ViewId> {            let name = entry.file_name().into_string().ok()?;            ViewId::try_from_hex(name)        };        let remove_file_if_not_new = |entry: &fs::DirEntry| -> Result<(), PathError> {            let path = entry.path();            // Check timestamp, but there's still TOCTOU problem if an existing            // file is renewed.            let metadata = entry.metadata().context(&path)?;            let mtime = metadata.modified().expect("unsupported platform?");            if mtime > keep_newer {                tracing::trace!(?path, "not removing");                Ok(())            } else {                tracing::trace!(?path, "removing");                fs::remove_file(&path).context(&path)            }        };         // Reachable objects are resolved without considering the keep_newer        // parameter. We could collect ancestors of the "new" operations here,        // but more files can be added anyway after that.        let read_op = |id: &OperationId| {            self.read_operation(id)                .block_on()                .map(|data| (id.clone(), data))        };        let reachable_ops: HashMap<OperationId, Operation> = dag_walk_async::dfs(            head_ids.iter().map(read_op),            |(id, _)| id.clone(),            |(_, data)| data.parents.iter().map(read_op).collect_vec(),        )        .try_collect()?;        let reachable_views: HashSet<&ViewId> =            reachable_ops.values().map(|data| &data.view_id).collect();        tracing::info!(            reachable_op_count = reachable_ops.len(),            reachable_view_count = reachable_views.len(),            "collected reachable objects"        );         let prune_ops = || -> Result<(), PathError> {            let op_dir = self.operations_dir();            for entry in op_dir.read_dir().context(&op_dir)? {                let entry = entry.context(&op_dir)?;                let Some(id) = to_op_id(&entry) else {                    tracing::trace!(?entry, "skipping invalid file name");                    continue;                };                if reachable_ops.contains_key(&id) {                    continue;                }                // If the operation was added after collecting reachable_views,                // its view mtime would also be renewed. So there's no need to                // update the reachable_views set to preserve the view.                remove_file_if_not_new(&entry)?;            }            Ok(())        };        prune_ops().map_err(|err| OpStoreError::Other(err.into()))?;         let prune_views = || -> Result<(), PathError> {            let view_dir = self.views_dir();            for entry in view_dir.read_dir().context(&view_dir)? {                let entry = entry.context(&view_dir)?;                let Some(id) = to_view_id(&entry) else {                    tracing::trace!(?entry, "skipping invalid file name");                    continue;                };                if reachable_views.contains(&id) {                    continue;                }                remove_file_if_not_new(&entry)?;            }            Ok(())        };        prune_views().map_err(|err| OpStoreError::Other(err.into()))?;         Ok(())    }} fn io_to_read_error(err: PathError, id: &impl ObjectId) -> OpStoreError {    if err.source.kind() == ErrorKind::NotFound {        OpStoreError::ObjectNotFound {            object_type: id.object_type(),            hash: id.hex(),            source: Box::new(err),        }    } else {        to_read_error(err.into(), id)    }} fn to_read_error(    source: Box<dyn std::error::Error + Send + Sync>,    id: &impl ObjectId,) -> OpStoreError {    OpStoreError::ReadObject {        object_type: id.object_type(),        hash: id.hex(),        source,    }} fn io_to_write_error(err: PathError, object_type: &'static str) -> OpStoreError {    OpStoreError::WriteObject {        object_type,        source: Box::new(err),    }} #[derive(Debug, Error)]enum PostDecodeError {    #[error("Invalid hash length (expected {expected} bytes, got {actual} bytes)")]    InvalidHashLength { expected: usize, actual: usize },    #[error("Invalid remote ref state value {0}")]    InvalidRemoteRefStateValue(i32),    #[error("Invalid number of ref target terms {0}")]    EvenNumberOfRefTargetTerms(usize),} fn operation_id_from_proto(bytes: Vec<u8>) -> Result<OperationId, PostDecodeError> {    if bytes.len() != OPERATION_ID_LENGTH {        Err(PostDecodeError::InvalidHashLength {            expected: OPERATION_ID_LENGTH,            actual: bytes.len(),        })    } else {        Ok(OperationId::new(bytes))    }} fn view_id_from_proto(bytes: Vec<u8>) -> Result<ViewId, PostDecodeError> {    if bytes.len() != VIEW_ID_LENGTH {        Err(PostDecodeError::InvalidHashLength {            expected: VIEW_ID_LENGTH,            actual: bytes.len(),        })    } else {        Ok(ViewId::new(bytes))    }} fn timestamp_to_proto(timestamp: &Timestamp) -> crate::protos::simple_op_store::Timestamp {    crate::protos::simple_op_store::Timestamp {        millis_since_epoch: timestamp.timestamp.0,        tz_offset: timestamp.tz_offset,    }} fn timestamp_from_proto(proto: crate::protos::simple_op_store::Timestamp) -> Timestamp {    Timestamp {        timestamp: MillisSinceEpoch(proto.millis_since_epoch),        tz_offset: proto.tz_offset,    }} fn operation_metadata_to_proto(    metadata: &OperationMetadata,) -> crate::protos::simple_op_store::OperationMetadata {    crate::protos::simple_op_store::OperationMetadata {        start_time: Some(timestamp_to_proto(&metadata.time.start)),        end_time: Some(timestamp_to_proto(&metadata.time.end)),        description: metadata.description.clone(),        hostname: metadata.hostname.clone(),        username: metadata.username.clone(),        is_snapshot: metadata.is_snapshot,        workspace_name: metadata.workspace_name.clone().map(Into::into),        attributes: metadata            .attributes            .iter()            .map(|(k, v)| (k.clone(), v.clone()))            .collect(),    }} fn operation_metadata_from_proto(    proto: crate::protos::simple_op_store::OperationMetadata,) -> OperationMetadata {    let time = TimestampRange {        start: timestamp_from_proto(proto.start_time.unwrap_or_default()),        end: timestamp_from_proto(proto.end_time.unwrap_or_default()),    };    let workspace_name = proto.workspace_name.map(Into::into);    OperationMetadata {        time,        description: proto.description,        hostname: proto.hostname,        username: proto.username,        is_snapshot: proto.is_snapshot,        workspace_name,        attributes: proto.attributes.into_iter().collect(),    }} fn commit_predecessors_map_to_proto(    map: &BTreeMap<CommitId, Vec<CommitId>>,) -> Vec<crate::protos::simple_op_store::CommitPredecessors> {    map.iter()        .map(            |(commit_id, predecessor_ids)| crate::protos::simple_op_store::CommitPredecessors {                commit_id: commit_id.to_bytes(),                predecessor_ids: predecessor_ids.iter().map(|id| id.to_bytes()).collect(),            },        )        .collect()} fn commit_predecessors_map_from_proto(    proto: Vec<crate::protos::simple_op_store::CommitPredecessors>,) -> BTreeMap<CommitId, Vec<CommitId>> {    proto        .into_iter()        .map(|entry| {            let commit_id = CommitId::new(entry.commit_id);            let predecessor_ids = entry                .predecessor_ids                .into_iter()

Lines 1–500 of 1297. Load more