From 407b7e01778ff1ec96d6f33d0b88c55079bbf026 Mon Sep 17 00:00:00 2001 From: Chris Mitchell Date: Sat, 8 Aug 2026 15:22:32 -0400 Subject: [PATCH] Track completed clone and pull operations Journal remote read operations until both their graph and asset phases succeed. Use the original operation range for retries, and require one full asset hydration when no successful operation has been recorded before switching to commit-range deltas. --- .../migrations/config/01-initial/up.sql | 20 ++ gen-models/src/operations.rs | 238 ++++++++++++ src/commands/clone.rs | 2 +- src/commands/remote/client.rs | 12 +- src/commands/remote/operations.rs | 340 ++++++++++++++++-- 5 files changed, 583 insertions(+), 29 deletions(-) diff --git a/gen-models/migrations/config/01-initial/up.sql b/gen-models/migrations/config/01-initial/up.sql index e897f316..d8a16837 100644 --- a/gen-models/migrations/config/01-initial/up.sql +++ b/gen-models/migrations/config/01-initial/up.sql @@ -12,6 +12,26 @@ CREATE TABLE remotes ( url TEXT NOT NULL ) STRICT; +CREATE TABLE remote_operations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + remote_name TEXT NOT NULL, + branch_name TEXT NOT NULL, + operation TEXT NOT NULL CHECK(operation IN ('clone', 'pull')), + from_commit TEXT, + asset_from_commit TEXT, + to_commit TEXT, + started_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + completed_at TEXT, + failed_at TEXT, + CHECK(completed_at IS NULL OR failed_at IS NULL), + CHECK(completed_at IS NULL OR to_commit IS NOT NULL), + FOREIGN KEY(remote_name) REFERENCES remotes(name) ON DELETE CASCADE +) STRICT; + +CREATE UNIQUE INDEX remote_operations_pending +ON remote_operations(remote_name, branch_name) +WHERE completed_at IS NULL AND failed_at IS NULL; + CREATE TABLE remote_branch ( remote_name TEXT, name TEXT, diff --git a/gen-models/src/operations.rs b/gen-models/src/operations.rs index 4096645e..20d471fe 100644 --- a/gen-models/src/operations.rs +++ b/gen-models/src/operations.rs @@ -638,6 +638,138 @@ impl RemoteBranch { } } +/// A remote read operation whose graph and asset phases must complete together. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RemoteOperationKind { + /// Initializes a workspace from a remote repository. + Clone, + /// Advances an existing local branch from a remote repository. + Pull, +} + +impl RemoteOperationKind { + const fn as_str(self) -> &'static str { + match self { + Self::Clone => "clone", + Self::Pull => "pull", + } + } +} + +/// Tracks the commit range for a pull or clone until all local assets are verified. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RemoteOperationRecord { + id: i64, + from_commit: Option, + asset_from_commit: Option, +} + +impl RemoteOperationRecord { + /// Resumes an incomplete operation or starts one from the supplied local commit. + /// + /// A pending clone can be resumed by a pull because both operations hydrate the same local + /// branch state. Its original lower bound must survive retries after the graph has advanced. + /// A branch without a successful operation starts without a lower bound so its first transfer + /// verifies the complete reachable asset history. + pub fn begin_or_resume( + conn: &ConfigConnection, + remote_name: &str, + branch_name: &str, + operation: RemoteOperationKind, + from_commit: Option<&DoltHashId>, + ) -> SQLResult { + let pending = conn + .query_row( + "SELECT id, from_commit, asset_from_commit FROM remote_operations \ + WHERE remote_name = ?1 AND branch_name = ?2 \ + AND completed_at IS NULL AND failed_at IS NULL \ + ORDER BY id LIMIT 1", + params![remote_name, branch_name], + |row| { + Ok(Self { + id: row.get(0)?, + from_commit: row.get(1)?, + asset_from_commit: row.get(2)?, + }) + }, + ) + .optional()?; + if let Some(pending) = pending { + return Ok(pending); + } + + let has_completed_operation = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM remote_operations \ + WHERE remote_name = ?1 AND branch_name = ?2 AND completed_at IS NOT NULL)", + params![remote_name, branch_name], + |row| row.get::<_, bool>(0), + )?; + let asset_from_commit = if has_completed_operation { + from_commit + } else { + None + }; + conn.execute( + "INSERT INTO remote_operations \ + (remote_name, branch_name, operation, from_commit, asset_from_commit) \ + VALUES (?1, ?2, ?3, ?4, ?5)", + params![ + remote_name, + branch_name, + operation.as_str(), + from_commit, + asset_from_commit + ], + )?; + Ok(Self { + id: conn.last_insert_rowid(), + from_commit: from_commit.copied(), + asset_from_commit: asset_from_commit.copied(), + }) + } + + /// Returns the local branch commit from before the operation for conflict detection. + pub const fn from_commit(&self) -> Option<&DoltHashId> { + self.from_commit.as_ref() + } + + /// Returns the lower asset boundary known to follow a successful whole operation. + pub const fn asset_from_commit(&self) -> Option<&DoltHashId> { + self.asset_from_commit.as_ref() + } + + /// Records the graph commit whose assets must be hydrated by this operation. + pub fn set_destination( + &self, + conn: &ConfigConnection, + to_commit: &DoltHashId, + ) -> SQLResult<()> { + conn.execute( + "UPDATE remote_operations SET to_commit = ?1 WHERE id = ?2", + params![to_commit, self.id], + )?; + Ok(()) + } + + /// Marks the operation complete only after its graph and asset phases succeed. + pub fn complete(&self, conn: &ConfigConnection) -> SQLResult<()> { + conn.execute( + "UPDATE remote_operations SET completed_at = CURRENT_TIMESTAMP WHERE id = ?1", + [self.id], + )?; + Ok(()) + } + + /// Marks an operation failed when its graph phase did not complete. + pub fn fail(&self, conn: &ConfigConnection) -> SQLResult<()> { + conn.execute( + "UPDATE remote_operations SET failed_at = CURRENT_TIMESTAMP WHERE id = ?1", + [self.id], + )?; + Ok(()) + } +} + #[derive(Clone, Debug, Serialize, Deserialize)] pub struct Defaults { pub id: i64, @@ -936,6 +1068,112 @@ mod tests { } } + #[cfg(test)] + mod remote_operations { + use gen_core::DoltHashId; + + use crate::{ + operations::{Remote, RemoteOperationKind, RemoteOperationRecord}, + test_helpers::setup_gen, + }; + + #[test] + fn test_remote_operation_resumes_until_completed() { + let context = setup_gen(); + let config = context.config().conn(); + Remote::create(config, "origin", "https://example.com/repo") + .expect("should create remote"); + let original_commit = DoltHashId([1_u8; 20]); + let advanced_commit = DoltHashId([2_u8; 20]); + + let baseline = RemoteOperationRecord::begin_or_resume( + config, + "origin", + "main", + RemoteOperationKind::Clone, + Some(&original_commit), + ) + .expect("should begin clone operation"); + assert_eq!(baseline.from_commit(), Some(&original_commit)); + assert_eq!(baseline.asset_from_commit(), None); + baseline + .set_destination(config, &original_commit) + .expect("should record clone destination"); + baseline + .complete(config) + .expect("should complete clone operation"); + + let operation = RemoteOperationRecord::begin_or_resume( + config, + "origin", + "main", + RemoteOperationKind::Pull, + Some(&original_commit), + ) + .expect("should begin pull operation"); + let resumed = RemoteOperationRecord::begin_or_resume( + config, + "origin", + "main", + RemoteOperationKind::Pull, + Some(&advanced_commit), + ) + .expect("should resume pull operation"); + + assert_eq!(resumed, operation); + assert_eq!(resumed.from_commit(), Some(&original_commit)); + assert_eq!(resumed.asset_from_commit(), Some(&original_commit)); + + resumed + .set_destination(config, &advanced_commit) + .expect("should record pull destination"); + resumed + .complete(config) + .expect("should complete pull operation"); + + let next = RemoteOperationRecord::begin_or_resume( + config, + "origin", + "main", + RemoteOperationKind::Pull, + Some(&advanced_commit), + ) + .expect("should begin the next pull operation"); + assert_ne!(next.id, operation.id); + assert_eq!(next.from_commit(), Some(&advanced_commit)); + assert_eq!(next.asset_from_commit(), Some(&advanced_commit)); + } + + #[test] + fn test_failed_graph_operation_does_not_resume() { + let context = setup_gen(); + let config = context.config().conn(); + Remote::create(config, "origin", "https://example.com/repo") + .expect("should create remote"); + let original_commit = DoltHashId([1_u8; 20]); + + let failed = RemoteOperationRecord::begin_or_resume( + config, + "origin", + "main", + RemoteOperationKind::Pull, + Some(&original_commit), + ) + .expect("should begin pull operation"); + failed.fail(config).expect("should fail pull operation"); + let next = RemoteOperationRecord::begin_or_resume( + config, + "origin", + "main", + RemoteOperationKind::Pull, + Some(&original_commit), + ) + .expect("should begin replacement pull operation"); + + assert_ne!(next.id, failed.id); + } + } + #[cfg(test)] mod remote { use super::*; diff --git a/src/commands/clone.rs b/src/commands/clone.rs index 9472b811..1bfb86e8 100644 --- a/src/commands/clone.rs +++ b/src/commands/clone.rs @@ -27,7 +27,7 @@ pub fn execute(url: &str, parent: &Workspace) -> Result<(), Box { +pub struct AssetTransferRequest<'request> { + /// The complete remote operation whose asset phase is being requested. pub operation: RemoteOperation, - pub branch: &'branch str, + /// The branch whose reachable assets should be transferred. + pub branch: &'request str, + /// The lower commit boundary, whose reachable assets are excluded. + pub from_commit: Option<&'request DoltHashId>, + /// The inclusive upper commit boundary for the operation. + pub to_commit: Option<&'request DoltHashId>, } #[derive(Clone, Debug, Deserialize, Eq, PartialEq)] diff --git a/src/commands/remote/operations.rs b/src/commands/remote/operations.rs index c9179379..9a3a1c61 100644 --- a/src/commands/remote/operations.rs +++ b/src/commands/remote/operations.rs @@ -39,6 +39,12 @@ //! safe workspace-relative paths, including `.gen/outside_root` paths used to represent inputs //! that originally came from outside the workspace. //! +//! Clone and pull are journaled as whole operations in the config database. An operation is +//! complete only after both its graph and asset phases succeed. A retry resumes the pending +//! operation's original commit range even when Dolt already advanced the graph. A branch with no +//! successful operation requests its complete reachable asset history once; subsequent successful +//! operations can safely request only the commit-range delta. +//! //! Pull records the branch commit from before the Dolt operation so downloads can //! distinguish a clean old version from a local modification. If the destination still //! matches the previous commit and the remote asset changed, the download replaces it as @@ -69,7 +75,10 @@ use gen_models::{ active_branch, add_remote, branch_hash, checkout, clone_remote, fetch, hash_of, pull, push, push_force, remote_rows, set_remote_url, }, - operations::{Defaults, Remote, RemoteBranch, calculate_file_checksum}, + operations::{ + Defaults, Remote, RemoteBranch, RemoteOperationKind as StoredRemoteOperationKind, + RemoteOperationRecord, calculate_file_checksum, + }, }; use md5::Md5; use reqwest::blocking::{Body, Client}; @@ -462,33 +471,40 @@ fn download_asset( result } -/// Transfers the asset versions needed to move from `previous_hash` to the selected branch state. +/// Transfers the asset versions needed to move from `asset_from_hash` to the selected branch state. /// /// GenHub may advertise the branch's complete asset history, so this function filters those URLs -/// to versions absent from the previous state. The cumulative view supplies that transfer delta -/// and the checksums used for conflict detection, while the materialized view decides which of the -/// selected versions belongs at its logical workspace path instead of under `.gen/assets`. +/// to versions absent from the requested range. `previous_hash` independently describes the local +/// state before the operation so intended updates remain distinguishable from user edits. The +/// materialized view decides which selected version belongs at its logical workspace path instead +/// of under `.gen/assets`. fn transfer_assets( graph: &GraphConnection, workspace: &Workspace, remote: &Remote, operation: RemoteOperation, branch: &str, + asset_from_hash: Option<&DoltHashId>, previous_hash: Option<&DoltHashId>, ) -> Result, Box> { if remote.url.starts_with("file://") { return Ok(Vec::new()); } let repository = RepositoryRemote::parse(&remote.url)?; + let commit_hash = hash_of(graph, branch)?; let response = acquire_asset_transfers( &repository, - &AssetTransferRequest { operation, branch }, + &AssetTransferRequest { + operation, + branch, + from_commit: asset_from_hash, + to_commit: Some(&commit_hash), + }, login_origin, )?; - let commit_hash = hash_of(graph, branch)?; - let current_assets: HashMap<_, _> = - AssetRef::get_cumulative_assets_at(graph, previous_hash, Some(&commit_hash))? + let range_assets: HashMap<_, _> = + AssetRef::get_cumulative_assets_at(graph, asset_from_hash, Some(&commit_hash))? .into_iter() .map(|asset| (asset.id, asset)) .collect(); @@ -497,6 +513,14 @@ fn transfer_assets( .into_iter() .map(|asset| asset.id) .collect(); + let excluded_assets = if let Some(asset_from_hash) = asset_from_hash { + AssetRef::get_cumulative_assets_at(graph, None, Some(asset_from_hash))? + .into_iter() + .map(|asset| (asset.id, asset)) + .collect() + } else { + HashMap::new() + }; let previous_assets = if let Some(previous_hash) = previous_hash { AssetRef::get_cumulative_assets_at(graph, None, Some(previous_hash))? .into_iter() @@ -506,17 +530,16 @@ fn transfer_assets( HashMap::new() }; // These are assets we expect to be in the current batch of transfers - let mut assets: HashMap<_, _> = current_assets + let mut assets: HashMap<_, _> = range_assets .iter() - .filter(|(id, _)| !previous_assets.contains_key(id)) + .filter(|(id, _)| !excluded_assets.contains_key(id)) .map(|(id, asset)| (*id, asset.clone())) .collect(); let client = Client::new(); let mut upload_receipts = Vec::new(); for transfer in response.assets { let Some(asset) = assets.remove(&transfer.id) else { - if current_assets.contains_key(&transfer.id) - || previous_assets.contains_key(&transfer.id) + if range_assets.contains_key(&transfer.id) || excluded_assets.contains_key(&transfer.id) { continue; } @@ -570,6 +593,7 @@ fn transfer_assets( } pub fn clone_into_workspace( + config: &ConfigConnection, remote: &Remote, workspace: &Workspace, ) -> Result> { @@ -612,14 +636,25 @@ pub fn clone_into_workspace( let branch = active_branch(&graph)?; drop(graph); let graph = get_raw_connection(workspace.graph_db_path()?)?; + let operation = RemoteOperationRecord::begin_or_resume( + config, + &remote.name, + &branch, + StoredRemoteOperationKind::Clone, + None, + )?; + let destination_hash = hash_of(&graph, &branch)?; + operation.set_destination(config, &destination_hash)?; transfer_assets( &graph, workspace, remote, RemoteOperation::Clone, &branch, - None, + operation.asset_from_commit(), + operation.from_commit(), )?; + operation.complete(config)?; Ok(branch) } @@ -666,6 +701,7 @@ pub fn execute_push( RemoteOperation::Push, &branch, previous_hash.as_ref(), + previous_hash.as_ref(), )?; if !remote.url.starts_with("file://") { let repository = RepositoryRemote::parse(&remote.url)?; @@ -715,22 +751,40 @@ pub fn execute_pull( .unwrap_or(active_branch(&graph)?); let remote = resolve_remote(&config, explicit_remote, &branch)?; let previous_hash = branch_hash(&graph, &branch)?; - run_graph_transfer( + let operation = RemoteOperationRecord::begin_or_resume( + &config, + &remote.name, + &branch, + StoredRemoteOperationKind::Pull, + previous_hash.as_ref(), + )?; + if let Err(error) = run_graph_transfer( &graph, &remote, RemoteOperation::Pull, &branch, false, || pull(&graph, &remote.name, &branch), - )?; + ) { + if let Err(metadata_error) = operation.fail(&config) { + eprintln!( + "Warning: failed to record unsuccessful pull operation for branch '{branch}': {metadata_error}" + ); + } + return Err(error); + } + let destination_hash = hash_of(&graph, &branch)?; + operation.set_destination(&config, &destination_hash)?; transfer_assets( &graph, workspace, &remote, RemoteOperation::Pull, &branch, - previous_hash.as_ref(), + operation.asset_from_commit(), + operation.from_commit(), )?; + operation.complete(&config)?; RemoteBranch::set_remote_validated(&config, &branch, Some(&remote.name))?; println!("Pulled branch '{branch}' from '{}'.", remote.name); Ok(()) @@ -780,13 +834,16 @@ mod tests { thread, }; - use gen_core::config::Workspace; + use gen_core::{DoltHashId, HashId, config::Workspace}; use gen_models::{ assets::{AssetRef, AssetRole, LocalAssetUri}, collection::Collection, db::GraphConnection, - history::dolt::{commit_all, hash_of, remote_rows, remove_remote}, - operations::{Defaults, Remote, calculate_reader_checksum}, + history::dolt::{clone_remote, commit_all, hash_of, remote_rows, remove_remote}, + operations::{ + Defaults, Remote, RemoteOperationKind as StoredRemoteOperationKind, + RemoteOperationRecord, calculate_reader_checksum, + }, }; use reqwest::blocking::Client; use rusqlite::{Connection, Error as SqlError}; @@ -795,10 +852,10 @@ mod tests { use super::{ DownloadAssetOutcome, RemoteOperation, canonical_remote_url, clone_destination_name, - download_asset, execute_push, file_graph_url, resolve_remote, run_graph_transfer, - transfer_assets, + download_asset, execute_pull, execute_push, file_graph_url, resolve_remote, + run_graph_transfer, transfer_assets, }; - use crate::{get_config_connection, get_connection}; + use crate::{get_config_connection, get_connection, get_raw_connection}; static ENVIRONMENT_LOCK: Mutex<()> = Mutex::new(()); @@ -820,7 +877,7 @@ mod tests { use url::Url; use super::super::clone_into_workspace; - use crate::get_connection; + use crate::{get_config_connection, get_connection}; struct CloneFixture { _temp: TempDir, @@ -970,7 +1027,16 @@ mod tests { workspace, } = CloneFixture::new(); - let result = clone_into_workspace(&remote, &workspace); + workspace.ensure_gen_dir(); + let config = get_config_connection(Some( + workspace + .gen_db_path() + .expect("should resolve clone config path"), + )) + .expect("should create clone config database"); + Remote::create(&config, &remote.name, &remote.url) + .expect("should configure clone remote"); + let result = clone_into_workspace(&config, &remote, &workspace); capability_stop.store(true, Ordering::Release); expired_server .join() @@ -1040,6 +1106,63 @@ mod tests { (format!("http://{address}/asset"), handle) } + fn serve_pull_api( + graph_url: &str, + asset_id: HashId, + asset_url: &str, + pull_count: usize, + fail_first_asset: bool, + ) -> (String, thread::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").expect("should bind pull API server"); + let address = listener + .local_addr() + .expect("should read pull API server address"); + let graph_url = graph_url.to_string(); + let asset_url = asset_url.to_string(); + let handle = thread::spawn(move || { + let mut requests = Vec::new(); + let mut asset_request_count = 0; + for _ in 0..(pull_count * 2) { + let (mut stream, _) = listener.accept().expect("should accept pull API request"); + let mut request = [0_u8; 8192]; + let read = stream + .read(&mut request) + .expect("should read pull API request"); + let request = String::from_utf8_lossy(&request[..read]).into_owned(); + let response_body = if request.contains("/remote-capability ") { + json!({ + "remote_url": graph_url, + "expires_at": "2030-01-01T00:00:00Z", + "default_branch": "main" + }) + .to_string() + } else if request.contains("/asset-transfers ") { + let download_url = if fail_first_asset && asset_request_count == 0 { + "http://127.0.0.1:1/unavailable" + } else { + &asset_url + }; + asset_request_count += 1; + json!({ + "assets": [{ "id": asset_id, "url": download_url }] + }) + .to_string() + } else { + panic!("unexpected pull API request: {request}"); + }; + requests.push(request); + write!( + stream, + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response_body}", + response_body.len() + ) + .expect("should write pull API response"); + } + requests + }); + (format!("http://{address}/api/repos/alice/example"), handle) + } + struct EnvironmentGuard { name: &'static str, previous: Option, @@ -1248,6 +1371,7 @@ mod tests { RemoteOperation::Pull, "main", Some(&previous_hash), + Some(&previous_hash), ) .expect("should transfer only the asset delta"); let transfer_request = transfer_server @@ -1262,6 +1386,172 @@ mod tests { ); } + #[test] + fn test_execute_pull_hydrates_assets_without_completed_operation() { + let temp = tempdir().expect("should create unhydrated pull workspace"); + let remote_graph_path = temp.path().join("remote.db"); + let remote_graph = + get_connection(&remote_graph_path).expect("should create remote graph database"); + Collection::create(&remote_graph, "base").expect("should create remote base state"); + let contents = b"branch-only asset\n"; + let asset = test_asset(contents, "feature.gfa", 1); + AssetRef::create(&remote_graph, &asset).expect("should insert remote asset"); + let current_hash = + commit_all(&remote_graph, "add branch asset").expect("should commit remote asset"); + drop(remote_graph); + + let workspace = Workspace::new(temp.path().join("local")); + workspace.ensure_gen_dir(); + let local_graph = get_raw_connection( + workspace + .graph_db_path() + .expect("should resolve local graph database path"), + ) + .expect("should open local graph database"); + let graph_url = format!("file://{}", remote_graph_path.display()); + clone_remote(&local_graph, &graph_url).expect("should clone remote graph state"); + drop(local_graph); + + let (asset_url, asset_server) = serve_asset(contents); + let (remote_url, api_server) = serve_pull_api(&graph_url, asset.id, &asset_url, 1, false); + let config = get_config_connection(Some( + workspace + .gen_db_path() + .expect("should resolve local config database path"), + )) + .expect("should open local config database"); + let remote = + Remote::create(&config, "origin", &remote_url).expect("should configure origin"); + Defaults::set_default_remote(&config, Some(&remote.name)) + .expect("should set default remote"); + + execute_pull(&workspace, None, None).expect("pull should hydrate the missing asset"); + + let requests = api_server.join().expect("pull API server should finish"); + let asset_request = requests + .iter() + .find(|request| request.contains("/asset-transfers ")) + .expect("should request asset transfers"); + assert!(asset_request.contains("\"from_commit\":null")); + assert!(asset_request.contains(&format!("\"to_commit\":\"{current_hash}\""))); + assert_eq!( + fs::read(temp.path().join("local/feature.gfa")) + .expect("should read hydrated branch asset"), + contents + ); + let completed_operations = config + .query_row( + "SELECT COUNT(*) FROM remote_operations \ + WHERE operation = 'pull' AND completed_at IS NOT NULL", + [], + |row| row.get::<_, i64>(0), + ) + .expect("should count completed pull operations"); + assert_eq!(completed_operations, 1); + asset_server.join().expect("asset server should finish"); + } + + #[test] + fn test_execute_pull_resumes_an_incomplete_asset_operation() { + let temp = tempdir().expect("should create pull retry workspace"); + let remote_graph_path = temp.path().join("remote.db"); + let remote_graph = + get_connection(&remote_graph_path).expect("should create remote graph database"); + Collection::create(&remote_graph, "base").expect("should create remote base state"); + let previous_hash = + commit_all(&remote_graph, "base").expect("should commit remote base state"); + drop(remote_graph); + + let workspace = Workspace::new(temp.path().join("local")); + workspace.ensure_gen_dir(); + let local_graph = get_raw_connection( + workspace + .graph_db_path() + .expect("should resolve local graph database path"), + ) + .expect("should open local graph database"); + let graph_url = format!("file://{}", remote_graph_path.display()); + clone_remote(&local_graph, &graph_url).expect("should clone remote base state"); + drop(local_graph); + + let remote_graph = + get_connection(&remote_graph_path).expect("should reopen remote graph database"); + let contents = b"retry asset\n"; + let asset = test_asset(contents, "retry.gfa", 1); + AssetRef::create(&remote_graph, &asset).expect("should insert remote asset"); + let current_hash = + commit_all(&remote_graph, "add retry asset").expect("should commit remote asset"); + drop(remote_graph); + + let (asset_url, asset_server) = serve_asset(contents); + let (remote_url, api_server) = serve_pull_api(&graph_url, asset.id, &asset_url, 2, true); + let config = get_config_connection(Some( + workspace + .gen_db_path() + .expect("should resolve local config database path"), + )) + .expect("should open local config database"); + let remote = + Remote::create(&config, "origin", &remote_url).expect("should configure origin"); + Defaults::set_default_remote(&config, Some(&remote.name)) + .expect("should set default remote"); + let baseline = RemoteOperationRecord::begin_or_resume( + &config, + &remote.name, + "main", + StoredRemoteOperationKind::Clone, + None, + ) + .expect("should begin baseline clone operation"); + baseline + .set_destination(&config, &previous_hash) + .expect("should record baseline clone destination"); + baseline + .complete(&config) + .expect("should complete baseline clone operation"); + + execute_pull(&workspace, None, None) + .expect_err("first pull should fail during its asset phase"); + assert!(!temp.path().join("local/retry.gfa").exists()); + let pending_commits: (DoltHashId, DoltHashId) = config + .query_row( + "SELECT from_commit, asset_from_commit FROM remote_operations \ + WHERE completed_at IS NULL AND failed_at IS NULL", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .expect("should retain the incomplete operation bounds"); + assert_eq!(pending_commits, (previous_hash, previous_hash)); + execute_pull(&workspace, None, None).expect("pull retry should succeed"); + + let requests = api_server.join().expect("pull API server should finish"); + let asset_requests = requests + .iter() + .filter(|request| request.contains("/asset-transfers ")) + .collect::>(); + let from_commit_json = format!("\"from_commit\":\"{previous_hash}\""); + let to_commit_json = format!("\"to_commit\":\"{current_hash}\""); + assert_eq!(asset_requests.len(), 2); + for request in asset_requests { + assert!(request.contains(&from_commit_json)); + assert!(request.contains(&to_commit_json)); + } + assert_eq!( + fs::read(temp.path().join("local/retry.gfa")).expect("should read retried asset"), + contents + ); + let pending_operations = config + .query_row( + "SELECT COUNT(*) FROM remote_operations \ + WHERE completed_at IS NULL AND failed_at IS NULL", + [], + |row| row.get::<_, i64>(0), + ) + .expect("should count pending remote operations"); + assert_eq!(pending_operations, 0); + asset_server.join().expect("asset server should finish"); + } + #[test] fn test_file_remote_resolves_graph_database() { assert_eq!(