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!(