diff --git a/rust/rhgitaly/proto/commit.proto b/rust/rhgitaly/proto/commit.proto index aba798503268b844524b9db31003f95ef754a370_cnVzdC9yaGdpdGFseS9wcm90by9jb21taXQucHJvdG8=..63daae2a133c40c11a0abf4f9fe154d3f995ffbb_cnVzdC9yaGdpdGFseS9wcm90by9jb21taXQucHJvdG8= 100644 --- a/rust/rhgitaly/proto/commit.proto +++ b/rust/rhgitaly/proto/commit.proto @@ -20,6 +20,13 @@ } // This comment is left unintentionally blank. + rpc GetTreeEntries(GetTreeEntriesRequest) returns (stream GetTreeEntriesResponse) { + option (op_type) = { + op: ACCESSOR + }; + } + + // This comment is left unintentionally blank. rpc FindCommit(FindCommitRequest) returns (FindCommitResponse) { option (op_type) = { op: ACCESSOR diff --git a/rust/rhgitaly/src/service/commit.rs b/rust/rhgitaly/src/service/commit.rs index aba798503268b844524b9db31003f95ef754a370_cnVzdC9yaGdpdGFseS9zcmMvc2VydmljZS9jb21taXQucnM=..63daae2a133c40c11a0abf4f9fe154d3f995ffbb_cnVzdC9yaGdpdGFseS9zcmMvc2VydmljZS9jb21taXQucnM= 100644 --- a/rust/rhgitaly/src/service/commit.rs +++ b/rust/rhgitaly/src/service/commit.rs @@ -5,6 +5,7 @@ use tracing::{field, info, instrument}; use hg::changelog::Changelog; +use hg::errors::HgError; use hg::repo::Repo; use hg::revlog::{NodePrefix, RevlogError, NULL_REVISION}; use hg::utils::hg_path::HgPath; @@ -14,8 +15,9 @@ use crate::git; use crate::gitaly::commit_service_server::{CommitService, CommitServiceServer}; use crate::gitaly::{ - tree_entry_response::ObjectType, FindCommitRequest, FindCommitResponse, - ListCommitsByOidRequest, ListCommitsByOidResponse, Repository, TreeEntryRequest, - TreeEntryResponse, + get_tree_entries_request::SortBy, tree_entry::EntryType, tree_entry_response::ObjectType, + FindCommitRequest, FindCommitResponse, GetTreeEntriesRequest, GetTreeEntriesResponse, + ListCommitsByOidRequest, ListCommitsByOidResponse, PaginationParameter, Repository, TreeEntry, + TreeEntryRequest, TreeEntryResponse, }; use crate::gitlab::revision::gitlab_revision_node_prefix; @@ -20,6 +22,9 @@ }; use crate::gitlab::revision::gitlab_revision_node_prefix; -use crate::mercurial::{changelog_entry_manifest, ls_path, read_blob, PathContent}; +use crate::mercurial::{ + changelog_entry_manifest, ls_path, read_blob, DirIteratorWithFlatPaths, + DirIteratorWithoutFlatPaths, PathContent, RecursiveDirIterator, +}; use crate::message; use crate::metadata::correlation_id; use crate::oid::tree_oid; @@ -27,7 +32,10 @@ default_repo_spec_error_status, load_repo_and_stream, load_repo_and_then, repo_store_vfs, RequestWithRepo, }; -use crate::streaming::{BlockingResponseSender, ResultResponseStream}; +use crate::streaming::{ + empty_response_stream, stream_with_pagination, BlockingResponseSender, PaginableMessageItem, + ResultResponseStream, +}; #[derive(Debug)] pub struct CommitServiceImpl { @@ -143,9 +151,124 @@ Ok(()) } +fn stream_get_tree_entries( + req: GetTreeEntriesRequest, + repo: &Repo, + node_prefix: NodePrefix, + tx: &BlockingResponseSender<GetTreeEntriesResponse>, +) -> Result<(), Status> { + let changelog = repo + .changelog() + .map_err(|e| Status::internal(format!("Error getting changelog: {:?}", e)))?; + let manifestlog = repo + .manifestlog() + .map_err(|e| Status::internal(format!("Error getting manifestlog: {:?}", e)))?; + let (cs_node, manifest) = changelog_entry_manifest(&changelog, &manifestlog, node_prefix) + .map_err(|e| Status::internal(format!("Error obtaining manifest: {:?}", e)))? + .ok_or_else(|| not_found_for_path(&req.path))?; + + let mut pl = req.path.len(); + while pl > 0 && req.path[pl - 1] == b'/' { + pl -= 1; + } + if pl == 1 && req.path[0] == b'.' { + pl = 0 + }; + let trimmed = &req.path[0..pl]; + + let commit_oid = String::from_utf8_lossy(&req.revision).to_string(); + + if req.recursive { + stream_get_tree_entries_from_iterator( + tx, + &req, + RecursiveDirIterator::new(commit_oid, cs_node, manifest.iter(), trimmed), + ); + } else if req.skip_flat_paths { + stream_get_tree_entries_from_iterator( + tx, + &req, + DirIteratorWithoutFlatPaths::new(commit_oid, cs_node, manifest.iter(), trimmed), + ); + } else { + stream_get_tree_entries_from_iterator( + tx, + &req, + DirIteratorWithFlatPaths::new(commit_oid, cs_node, manifest.iter(), trimmed), + ); + } + Ok(()) +} + +/// Stream `GetTreeEntries` responses, using the inner iterator. +/// +/// Various options (e.g., `recursive`) give rise to different types of iterators on entries. +/// This function applies the sort and pagination parameters, final chunking and streams responses +/// to the channel. +fn stream_get_tree_entries_from_iterator<It>( + tx: &BlockingResponseSender<GetTreeEntriesResponse>, + request: &GetTreeEntriesRequest, + iterator: It, +) where + It: Iterator<Item = Result<TreeEntry, HgError>>, +{ + let pagination = &request.pagination_params; + if request.sort == SortBy::TreesFirst as i32 { + let mut trees = Vec::new(); + let mut blobs = Vec::new(); + for res in iterator { + match res { + Ok(ent) => { + if ent.r#type == EntryType::Tree as i32 { + trees.push(Ok(ent)); + } else if ent.r#type == EntryType::Blob as i32 { + blobs.push(Ok(ent)); + } + } + // arbitrary choice, weakly motivated by errors being more likely with blobs + // than with trees at this point. + err => blobs.push(err), + } + } + stream_get_tree_entries_paginated( + tx, + pagination, + trees.into_iter().chain(blobs.into_iter()), + ); + } else { + stream_get_tree_entries_paginated(tx, pagination, iterator); + } +} + +/// Stream [`GetTreeEntriesResponse`] messages with pagination +/// +/// This function takes care of the offset and limit logic enclosed into the given pagination +/// parameter, regrouping the `TreeEntry` elements yielded by the iterator into suitable chunks, +/// and providing the `next_cursor` value, but does not take care of ordering. +/// Instead it expects the incoming iiterator to be already properly ordered. +fn stream_get_tree_entries_paginated<It>( + tx: &BlockingResponseSender<GetTreeEntriesResponse>, + pagination: &Option<PaginationParameter>, + iterator: It, +) where + It: Iterator<Item = Result<TreeEntry, HgError>>, +{ + stream_with_pagination( + tx, + pagination, + iterator, + |entries, pagination_cursor| GetTreeEntriesResponse { + entries, + pagination_cursor, + }, + |err| Status::internal(format!("error reading manifest: #{:?}", err)), + |token| Status::internal(format!("could not find starting OID: {}", token)), + ) +} + impl RequestWithRepo for TreeEntryRequest { fn repository_ref(&self) -> Option<&Repository> { self.repository.as_ref() } } @@ -146,9 +269,15 @@ impl RequestWithRepo for TreeEntryRequest { fn repository_ref(&self) -> Option<&Repository> { self.repository.as_ref() } } +impl RequestWithRepo for GetTreeEntriesRequest { + fn repository_ref(&self) -> Option<&Repository> { + self.repository.as_ref() + } +} + impl RequestWithRepo for ListCommitsByOidRequest { fn repository_ref(&self) -> Option<&Repository> { self.repository.as_ref() @@ -161,7 +290,17 @@ } } +impl PaginableMessageItem for TreeEntry { + fn match_token(&self, token: &str) -> bool { + self.oid == token + } + + fn next_cursor(&self) -> String { + self.oid.clone() + } +} + #[tonic::async_trait] impl CommitService for CommitServiceImpl { type ListCommitsByOidStream = ReceiverStream<Result<ListCommitsByOidResponse, Status>>; type TreeEntryStream = ReceiverStream<Result<TreeEntryResponse, Status>>; @@ -164,7 +303,8 @@ #[tonic::async_trait] impl CommitService for CommitServiceImpl { type ListCommitsByOidStream = ReceiverStream<Result<ListCommitsByOidResponse, Status>>; type TreeEntryStream = ReceiverStream<Result<TreeEntryResponse, Status>>; + type GetTreeEntriesStream = ReceiverStream<Result<GetTreeEntriesResponse, Status>>; async fn list_commits_by_oid( &self, @@ -195,6 +335,16 @@ self.inner_tree_entry(inner, correlation_id(&metadata)) .await } + + async fn get_tree_entries( + &self, + request: Request<GetTreeEntriesRequest>, + ) -> Result<Response<Self::GetTreeEntriesStream>, Status> { + let (metadata, _ext, inner) = request.into_parts(); + + self.inner_get_tree_entries(inner, correlation_id(&metadata)) + .await + } } struct TreeEntryTracingRequest<'a>(&'a TreeEntryRequest); @@ -211,6 +361,22 @@ } } +struct GetTreeEntriesTracingRequest<'a>(&'a GetTreeEntriesRequest); + +impl Debug for GetTreeEntriesTracingRequest<'_> { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("GetTreeEntriesRequest") + .field("repository", &self.0.repository) + .field("revision", &String::from_utf8_lossy(&self.0.revision)) + .field("path", &String::from_utf8_lossy(&self.0.path)) + .field("recursive", &self.0.recursive) + .field("sort", &self.0.sort) + .field("pagination_params", &self.0.pagination_params) + .field("skip_flat_paths", &self.0.skip_flat_paths) + .finish() + } +} + struct FindCommitTracingRequest<'a>(&'a FindCommitRequest); impl Debug for FindCommitTracingRequest<'_> { @@ -267,6 +433,56 @@ } } + #[instrument(name = "get_tree_entries", + skip(self, request), + fields(request=field::debug(GetTreeEntriesTracingRequest(&request))))] + async fn inner_get_tree_entries( + &self, + request: GetTreeEntriesRequest, + correlation_id: Option<&MetadataValue<Ascii>>, + ) -> ResultResponseStream<GetTreeEntriesResponse> { + info!("Processing"); + + if request.revision.is_empty() { + return Err(Status::invalid_argument("empty revision")); + } + + let config = self.config.clone(); + let store_vfs = repo_store_vfs(&config, &request.repository) + .await + .map_err(default_repo_spec_error_status)?; + + match gitlab_revision_node_prefix(&store_vfs, &request.revision) + .await + .map_err(|e| Status::internal(format!("Error resolving revision: {:?}", e)))? + { + None => { + info!("Revision not resolved"); + empty_response_stream() + } + Some(node_prefix) => { + info!("Revision resolved as {:x}", &node_prefix); + + if request.pagination_params.as_ref().map(|p| p.limit) == Some(0) { + // let's not even load the repo + // TODO comparison test for unknown revision with limit 0 + return empty_response_stream(); + } + + load_repo_and_stream( + self.config.clone(), + request, + default_repo_spec_error_status, + move |req, repo, tx| { + if let Err(e) = stream_get_tree_entries(req, &repo, node_prefix, &tx) { + tx.send(Err(e)); + } + }, + ) + } + } + } + #[instrument(name = "list_commits_by_oid", skip(self))] async fn inner_list_commits_by_oid( &self, diff --git a/tests_with_gitaly/test_blob_tree.py b/tests_with_gitaly/test_blob_tree.py index aba798503268b844524b9db31003f95ef754a370_dGVzdHNfd2l0aF9naXRhbHkvdGVzdF9ibG9iX3RyZWUucHk=..63daae2a133c40c11a0abf4f9fe154d3f995ffbb_dGVzdHNfd2l0aF9naXRhbHkvdGVzdF9ibG9iX3RyZWUucHk= 100644 --- a/tests_with_gitaly/test_blob_tree.py +++ b/tests_with_gitaly/test_blob_tree.py @@ -177,9 +177,6 @@ # reusing content to test GetTreeEntries, hence with a new rpc_helper # - if hg_server == 'rhgitaly': # TODO split with a module fixture - return - def gte_normalizer(rpc_helper, responses, vcs='hg'): for resp in responses: for entry in resp.entries: @@ -188,6 +185,7 @@ entry.root_oid = b'' rpc_helper = fixture.rpc_helper(stub_cls=CommitServiceStub, + hg_server=hg_server, method_name='GetTreeEntries', request_cls=GetTreeEntriesRequest, request_defaults=dict( @@ -237,8 +235,9 @@ sort=SortBy.TREES_FIRST) -def test_compare_get_tree_entries_pagination(gitaly_comparison): - fixture = gitaly_comparison +@parametrize('hg_server', ('hgitaly', 'rhgitaly')) +def test_compare_get_tree_entries_pagination(gitaly_rhgitaly_comparison, hg_server): + fixture = gitaly_rhgitaly_comparison wrapper = fixture.hg_repo_wrapper wrapper.write_commit('foo', message="Some foo") @@ -271,6 +270,7 @@ cursor2git = {} # hg cursor -> git cursor rpc_helper = fixture.rpc_helper(stub_cls=CommitServiceStub, + hg_server=hg_server, method_name='GetTreeEntries', request_cls=GetTreeEntriesRequest, request_defaults=dict(