client will error if registry rewinds checkpoint log length or change…

…s log root or map root for a previously fetched checkpoint (#239)

crates/client/src/lib.rs

@@ -5,6 +5,7 @@
 use crate::storage::PackageInfo;
 use anyhow::{anyhow, Context, Result};
 use reqwest::{Body, IntoUrl};
+use std::cmp::Ordering;
 use std::{borrow::Cow, collections::HashMap, path::PathBuf, time::Duration};
 use storage::{
     ContentStorage, FileSystemContentStorage, FileSystemRegistryStorage, PublishInfo,
@@ -25,7 +26,7 @@ use warg_crypto::{
 };
 use warg_protocol::{
     operator, package,
-    registry::{LogId, LogLeaf, PackageName, RecordId, TimestampedCheckpoint},
+    registry::{LogId, LogLeaf, PackageName, RecordId, RegistryLen, TimestampedCheckpoint},
     PublishedProtoEnvelope, SerdeEnvelope, Version, VersionReq,
 };
 
@@ -539,17 +540,39 @@ impl<R: RegistryStorage, C: ContentStorage> Client<R, C> {
         }
 
         if let Some(from) = self.registry.load_checkpoint().await? {
-            if from.as_ref().checkpoint.log_length < ts_checkpoint.as_ref().checkpoint.log_length {
-                self.api
-                    .prove_log_consistency(
-                        ConsistencyRequest {
-                            from: from.as_ref().checkpoint.log_length,
-                            to: ts_checkpoint.as_ref().checkpoint.log_length,
-                        },
-                        Cow::Borrowed(&from.as_ref().checkpoint.log_root),
-                        Cow::Borrowed(&ts_checkpoint.as_ref().checkpoint.log_root),
-                    )
-                    .await?;
+            let from_log_length = from.as_ref().checkpoint.log_length;
+            let to_log_length = ts_checkpoint.as_ref().checkpoint.log_length;
+
+            match from_log_length.cmp(&to_log_length) {
+                Ordering::Greater => {
+                    return Err(ClientError::CheckpointLogLengthRewind {
+                        from: from_log_length,
+                        to: to_log_length,
+                    });
+                }
+                Ordering::Less => {
+                    self.api
+                        .prove_log_consistency(
+                            ConsistencyRequest {
+                                from: from_log_length,
+                                to: to_log_length,
+                            },
+                            Cow::Borrowed(&from.as_ref().checkpoint.log_root),
+                            Cow::Borrowed(&ts_checkpoint.as_ref().checkpoint.log_root),
+                        )
+                        .await?
+                }
+                Ordering::Equal => {
+                    if from.as_ref().checkpoint.log_root
+                        != ts_checkpoint.as_ref().checkpoint.log_root
+                        || from.as_ref().checkpoint.map_root
+                            != ts_checkpoint.as_ref().checkpoint.map_root
+                    {
+                        return Err(ClientError::CheckpointChangedLogRootOrMapRoot {
+                            log_length: from_log_length,
+                        });
+                    }
+                }
             }
         }
 
@@ -825,6 +848,24 @@ pub enum ClientError {
     #[error("the package is still missing content after all content was uploaded")]
     PackageMissingContent,
 
+    /// The registry provided a latest checkpoint with a log length less than a previously provided
+    /// checkpoint log length.
+    #[error("registry rewinded checkpoints; latest checkpoint log length `{to}` is less than previously received checkpoint log length `{from}`")]
+    CheckpointLogLengthRewind {
+        /// The previously received checkpoint log length.
+        from: RegistryLen,
+        /// The latest checkpoint log length.
+        to: RegistryLen,
+    },
+
+    /// The registry provided a checkpoint with a different `log_root` and
+    /// `map_root` than a previously provided checkpoint.
+    #[error("registry provided a new checkpoint with the same log length `{log_length}` as previously fetched but different log root or map root")]
+    CheckpointChangedLogRootOrMapRoot {
+        /// The checkpoint log length.
+        log_length: RegistryLen,
+    },
+
     /// An error occurred during an API operation.
     #[error(transparent)]
     Api(#[from] api::ClientError),