Track completed clone and pull operations

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,

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<DoltHashId>,
+    asset_from_commit: Option<DoltHashId>,
+}
+
+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<Self> {
+        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::*;

src/commands/clone.rs

@@ -27,7 +27,7 @@ pub fn execute(url: &str, parent: &Workspace) -> Result<(), Box<dyn std::error::
         let canonical_url = canonical_remote_url(url)?;
         let remote = Remote::create(&config, "origin", &canonical_url)?;
         Defaults::set_default_remote(&config, Some("origin"))?;
-        let branch = clone_into_workspace(&remote, &workspace)?;
+        let branch = clone_into_workspace(&config, &remote, &workspace)?;
         RemoteBranch::set_remote_validated(&config, &branch, Some("origin"))?;
         Defaults::set_current_branch(&config, Some(&branch))?;
         println!("Cloned {canonical_url} into {}.", destination.display());

src/commands/remote/client.rs

@@ -26,7 +26,7 @@
 
 use std::{env, io};
 
-use gen_core::HashId;
+use gen_core::{DoltHashId, HashId};
 use reqwest::{
     StatusCode, Url,
     blocking::{Client, RequestBuilder},
@@ -148,9 +148,15 @@ pub struct CapabilityResponse {
 }
 
 #[derive(Clone, Debug, Serialize)]
-pub struct AssetTransferRequest<'branch> {
+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)]