Merge pull request #239 from Oxen-AI/fix/push-new-remote-branch

Push new remote branch on non-deep cloned repo

Author: gschoeni

src/lib/src/api/local/commits.rs

@@ -278,6 +278,20 @@ pub fn list_all(repo: &LocalRepository) -> Result<Vec<Commit>, OxenError> {
     Ok(commits)
 }
 
+pub fn list_all_paginated(
+    repo: &LocalRepository,
+    page_number: usize,
+    page_size: usize,
+) -> Result<PaginatedCommits, OxenError> {
+    let commits = list_all(repo)?;
+    let (commits, pagination) = util::paginate(commits, page_number, page_size);
+    Ok(PaginatedCommits {
+        status: StatusMessage::resource_found(),
+        commits,
+        pagination,
+    })
+}
+
 /// Get commit history given options
 pub async fn list_with_opts(
     repo: &LocalRepository,

src/lib/src/api/remote/commits.rs

@@ -69,6 +69,44 @@ pub async fn get_by_id(
     }
 }
 
+pub async fn list_all(remote_repo: &RemoteRepository) -> Result<Vec<Commit>, OxenError> {
+    let mut all_commits: Vec<Commit> = Vec::new();
+    let mut page_num = DEFAULT_PAGE_NUM;
+    let page_size = 100;
+
+    let bar = Arc::new(ProgressBar::new_spinner());
+    bar.set_style(ProgressStyle::default_spinner());
+
+    loop {
+        let page_opts = PaginateOpts {
+            page_num,
+            page_size,
+        };
+        match list_all_commits_paginated(remote_repo, &page_opts).await {
+            Ok(paginated_commits) => {
+                if page_num == DEFAULT_PAGE_NUM {
+                    let bar = oxify_bar(bar.clone(), ProgressBarType::Counter);
+                    bar.set_length(paginated_commits.pagination.total_entries as u64);
+                }
+                let n_commits = paginated_commits.commits.len();
+                all_commits.extend(paginated_commits.commits);
+                bar.inc(n_commits as u64);
+                if page_num < paginated_commits.pagination.total_pages {
+                    page_num += 1;
+                } else {
+                    break;
+                }
+            }
+            Err(err) => {
+                return Err(err);
+            }
+        }
+    }
+    bar.finish_and_clear();
+
+    Ok(all_commits)
+}
+
 pub async fn list_commit_history(
     remote_repo: &RemoteRepository,
     revision: &str,
@@ -138,6 +176,33 @@ async fn list_commit_history_paginated(
     }
 }
 
+async fn list_all_commits_paginated(
+    remote_repo: &RemoteRepository,
+    page_opts: &PaginateOpts,
+) -> Result<PaginatedCommits, OxenError> {
+    let page_num = page_opts.page_num;
+    let page_size = page_opts.page_size;
+    let uri = format!("/commits/all?page={page_num}&page_size={page_size}");
+    let url = api::endpoint::url_from_repo(remote_repo, &uri)?;
+
+    let client = client::new_for_url(&url)?;
+    match client.get(&url).send().await {
+        Ok(res) => {
+            let body = client::parse_json_body(&url, res).await?;
+            let response: Result<PaginatedCommits, serde_json::Error> = serde_json::from_str(&body);
+            match response {
+                Ok(j_res) => Ok(j_res),
+                Err(err) => Err(OxenError::basic_str(format!(
+                    "list_commit_history() Could not deserialize response [{err}]\n{body}"
+                ))),
+            }
+        }
+        Err(err) => Err(OxenError::basic_str(format!(
+            "list_commit_history() Request failed: {err}"
+        ))),
+    }
+}
+
 pub async fn commit_is_synced(
     remote_repo: &RemoteRepository,
     commit_id: &str,

src/lib/src/core/index/pusher.rs

@@ -11,7 +11,7 @@ use flate2::write::GzEncoder;
 use flate2::Compression;
 use futures::prelude::*;
 use indicatif::ProgressBar;
-use std::collections::HashSet;
+use std::collections::{HashSet, VecDeque};
 
 use std::io::{BufReader, Read};
 use std::sync::Arc;
@@ -75,13 +75,6 @@ async fn validate_repo_is_pushable(
     commit_reader: &CommitReader,
     head_commit: &Commit,
 ) -> Result<bool, OxenError> {
-    // Make sure the remote branch is not ahead of the local branch
-    // log::debug!(
-    //     "validating repo is pushable for commit {:#?} and branch {:#?}",
-    //     head_commit,
-    //     branch
-    // );
-
     if remote_is_ahead_of_local(head_commit, remote_repo, commit_reader, branch).await? {
         log::debug!("remote is ahead of local for commit {:#?}", head_commit);
         if api::remote::commits::can_push(remote_repo, &branch.name, local_repo, head_commit)
@@ -94,17 +87,6 @@ async fn validate_repo_is_pushable(
             return Err(OxenError::upstream_merge_conflict());
         }
     }
-    // } else {
-    //     log::debug!("remote is not ahead of local for commit {:#?}", head_commit);
-    // }
-
-    if cannot_push_incomplete_history(local_repo, remote_repo, head_commit, branch).await? {
-        log::debug!(
-            "cannot_push_incomplete_history is true for commit {:#?}",
-            head_commit
-        );
-        return Err(OxenError::incomplete_local_history());
-    }
 
     Ok(false)
 }
@@ -200,18 +182,24 @@ pub async fn try_push_remote_repo(
     mut head_commit: Commit,
     requires_merge: bool,
 ) -> Result<(), OxenError> {
-    let commits_to_sync =
+    let commits_to_push =
         get_commit_objects_to_sync(local_repo, remote_repo, &head_commit, &branch).await?;
 
+    log::debug!("got these commits to push");
+
+    if !commits_to_push_are_synced(local_repo, &commits_to_push)? {
+        return Err(OxenError::incomplete_local_history());
+    }
+
     log::debug!(
         "push_remote_repo commit order after get_commit_objects_to_sync {:?}",
-        commits_to_sync
+        commits_to_push
     );
 
     let maybe_remote_branch = api::remote::branches::get_by_name(remote_repo, &branch.name).await?;
 
     let (unsynced_entries, _total_size) =
-        push_missing_commit_objects(local_repo, remote_repo, &commits_to_sync, &branch).await?;
+        push_missing_commit_objects(local_repo, remote_repo, &commits_to_push, &branch).await?;
 
     log::debug!("🐂 Identifying unsynced commits dbs...");
     let unsynced_db_commits =
@@ -301,18 +289,16 @@ async fn get_commit_objects_to_sync(
     branch: &Branch,
 ) -> Result<Vec<Commit>, OxenError> {
     let remote_branch = api::remote::branches::get_by_name(remote_repo, &branch.name).await?;
-
+    let commit_reader = CommitReader::new(local_repo)?;
     let mut commits_to_sync: Vec<Commit>;
-    // TODO: If remote branch does not yet, recreates all commits regardless of shared history.
-    // Not a huge deal performance-wise right now, but could be for very commit-heavy repos
     if let Some(remote_branch) = remote_branch {
         log::debug!(
             "get_commit_objects_to_sync found remote branch {:?}, calculating missing commits between local and remote heads", remote_branch
         );
         let remote_commit = api::remote::commits::get_by_id(remote_repo, &remote_branch.commit_id)
             .await?
             .unwrap();
-        let commit_reader = CommitReader::new(local_repo)?;
+
         let merger = Merger::new(local_repo)?;
         commits_to_sync =
             merger.list_commits_between_commits(&commit_reader, &remote_commit, local_commit)?;
@@ -333,9 +319,31 @@ async fn get_commit_objects_to_sync(
                 .any(|remote_commit| remote_commit.id == commit.id)
         });
     } else {
-        // Branch does not exist on remote yet - get all commits?
-        log::debug!("get_commit_objects_to_sync remote branch does not exist, getting all commits from local head");
-        commits_to_sync = api::local::commits::list_from(local_repo, &local_commit.id)?;
+        // Remote branch does not exist. Find commits to push with reference to whatever
+        // remote branch head comes first in the local newbranch history, aka what it was branched off of.
+
+        // Early return to avoid checking for remote commits: if full local history and no remote branch,
+        // push full local branch history.
+        if api::local::commits::commit_history_is_complete(local_repo, local_commit) {
+            return api::local::commits::list_from(local_repo, &local_commit.id);
+        }
+
+        // Otherwise, find the remote commit that the local branch was branched off of and push everything since then.
+        let all_commits = api::remote::commits::list_all(remote_repo).await?;
+        log::debug!("got all remote commits as {:#?}", all_commits);
+        let maybe_remote_commit =
+            find_latest_local_commit_synced(local_repo, local_commit, &all_commits)?;
+
+        if let Some(remote_commit) = maybe_remote_commit {
+            let merger = Merger::new(local_repo)?;
+            commits_to_sync = merger.list_commits_between_commits(
+                &commit_reader,
+                &remote_commit,
+                local_commit,
+            )?;
+        } else {
+            commits_to_sync = api::local::commits::list_from(local_repo, &local_commit.id)?;
+        }
     }
 
     // Order from BASE to HEAD
@@ -474,54 +482,47 @@ async fn remote_is_ahead_of_local(
     Ok(!local_head.has_ancestor(&remote_commit.id, reader)?)
 }
 
-async fn cannot_push_incomplete_history(
+fn find_latest_local_commit_synced(
     local_repo: &LocalRepository,
-    remote_repo: &RemoteRepository,
     local_head: &Commit,
-    branch: &Branch,
-) -> Result<bool, OxenError> {
-    log::debug!("Checking if we can push incomplete history.");
-    match api::remote::commits::list_commit_history(remote_repo, &branch.name).await {
-        Err(_) => {
-            return Ok(!api::local::commits::commit_history_is_complete(
-                local_repo, local_head,
-            ));
+    remote_commits: &Vec<Commit>,
+) -> Result<Option<Commit>, OxenError> {
+    let commit_reader = CommitReader::new(local_repo).unwrap();
+    let mut commits_set: HashSet<String> = HashSet::new();
+    for remote_commit in remote_commits {
+        commits_set.insert(remote_commit.id.clone());
+    }
+    // let mut current_commit = local_head.clone();
+    let mut queue: VecDeque<Commit> = VecDeque::new();
+    queue.push_back(local_head.clone());
+
+    while !queue.is_empty() {
+        let current_commit = queue.pop_front().unwrap();
+        if commits_set.contains(&current_commit.id) {
+            return Ok(Some(current_commit));
         }
-        Ok(remote_history) => {
-            let remote_head = remote_history.first().unwrap();
-            log::debug!(
-                "Checking between local head {:?} and remote head {:?} on branch {}",
-                local_head,
-                remote_head,
-                branch.name
-            );
-
-            let commit_reader = CommitReader::new(local_repo)?;
-            let merger = Merger::new(local_repo)?;
-
-            let commits_to_push =
-                merger.list_commits_between_commits(&commit_reader, remote_head, local_head)?;
-
-            let commits_to_push: Vec<Commit> = commits_to_push
-                .into_iter()
-                .filter(|commit| {
-                    !remote_history
-                        .iter()
-                        .any(|remote_commit| remote_commit.id == commit.id)
-                })
-                .collect();
-
-            log::debug!("Found the following commits_to_push: {:?}", commits_to_push);
-            // Ensure all `commits_to_push` are synced
-            for commit in commits_to_push {
-                if !index::commit_sync_status::commit_is_synced(local_repo, &commit) {
-                    return Ok(true);
-                }
-            }
+        for parent_id in current_commit.parent_ids.iter() {
+            let parent_commit = commit_reader.get_commit_by_id(parent_id)?;
+            let Some(parent_commit) = parent_commit else {
+                return Err(OxenError::local_parent_link_broken(&current_commit.id));
+            };
+            queue.push_back(parent_commit);
         }
     }
+    Ok(None)
+}
 
-    Ok(false)
+fn commits_to_push_are_synced(
+    local_repo: &LocalRepository,
+    commits_to_push: &Vec<Commit>,
+) -> Result<bool, OxenError> {
+    for commit in commits_to_push {
+        if !index::commit_sync_status::commit_is_synced(local_repo, commit) {
+            log::debug!("commit is not synced {:?}", commit);
+            return Ok(false);
+        }
+    }
+    Ok(true)
 }
 
 async fn poll_until_synced(

src/lib/src/core/index/schema_reader.rs

@@ -182,42 +182,6 @@ impl SchemaReader {
 
     pub fn list_schema_entries(&self) -> Result<Vec<SchemaEntry>, OxenError> {
         log::debug!("trying to get root hash for commit {:?}", self.commit_id);
-        // Print out all the entries of dir_hashes db
-        // let iter = self.dir_hashes_db.iterator(rocksdb::IteratorMode::Start);
-
-        // log::debug!(
-        //     "iterating over dir_hashes_db for commit {:?}",
-        //     self.commit_id
-        // );
-        // for item in iter {
-        //     match item {
-        //         Ok((key, value)) => {
-        //             let key = match std::str::from_utf8(&key) {
-        //                 Ok(k) => k,
-        //                 Err(e) => {
-        //                     log::error!("Failed to convert key to string: {:?}", e);
-        //                     continue;
-        //                 }
-        //             };
-        //             let value = match std::str::from_utf8(&value) {
-        //                 Ok(v) => v,
-        //                 Err(e) => {
-        //                     log::error!("Failed to convert value to string: {:?}", e);
-        //                     continue;
-        //                 }
-        //             };
-        //             log::debug!("key {:?} value {:?}", key, value);
-        //         }
-        //         Err(e) => {
-        //             log::error!("Iterator error: {:?}", e);
-        //         }
-        //     }
-        // }
-
-        // log::debug!("done iterating");
-
-        // log::debug!("done iterating round 2");
-
         let root_hash: String = path_db::get_entry(&self.dir_hashes_db, "")?.ok_or(
             OxenError::basic_str("Could not find root hash in dir hashes db"),
         )?;

src/server/src/controllers/commits.rs

@@ -117,6 +117,30 @@ pub async fn commit_history(
     }
 }
 
+// List all commits in the rpeo
+pub async fn list_all(
+    req: HttpRequest,
+    query: web::Query<PageNumQuery>,
+) -> actix_web::Result<HttpResponse, OxenHttpError> {
+    let app_data = app_data(&req)?;
+    let namespace: Option<&str> = req.match_info().get("namespace");
+    let repo_name: Option<&str> = req.match_info().get("repo_name");
+
+    let page: usize = query.page.unwrap_or(constants::DEFAULT_PAGE_NUM);
+    let page_size: usize = query.page_size.unwrap_or(constants::DEFAULT_PAGE_SIZE);
+
+    if let (Some(namespace), Some(repo_name)) = (namespace, repo_name) {
+        let repo = get_repo(&app_data.path, namespace, repo_name)?;
+
+        let paginated_commits = api::local::commits::list_all_paginated(&repo, page, page_size)?;
+
+        Ok(HttpResponse::Ok().json(paginated_commits))
+    } else {
+        let msg = "Must supply `namespace`, `repo_name`  params";
+        Err(OxenHttpError::BadRequest(msg.into()))
+    }
+}
+
 pub async fn show(req: HttpRequest) -> actix_web::Result<HttpResponse, OxenHttpError> {
     let app_data = app_data(&req)?;
     let namespace = path_param(&req, "namespace")?;

src/server/src/routes.rs

@@ -56,6 +56,10 @@ pub fn config(cfg: &mut web::ServiceConfig) {
             "/{namespace}/{repo_name}/objects_db",
             web::get().to(controllers::commits::download_objects_db),
         )
+        .route(
+            "/{namespace}/{repo_name}/commits/all",
+            web::get().to(controllers::commits::list_all),
+        )
         .route(
             "/{namespace}/{repo_name}/commits/{commit_id}/latest_synced",
             web::get().to(controllers::commits::latest_synced),