Support for read-only Storage (#798)

Why:

* We may implement `Storage` for certain read-only s3-like, such as
  HTTP
* We'll need this for [Tigris support](https://github.com/earth-mover/icechunk/blob/main/design-docs/009-tigris-support.md)

Author: paraseba

icechunk/src/repository.rs

@@ -88,6 +88,8 @@ pub enum RepositoryErrorKind {
     ConcurrencyError(#[from] JoinError),
     #[error("main branch cannot be deleted")]
     CannotDeleteMain,
+    #[error("the storage used by this Icechunk repository is read-only: {0}")]
+    ReadonlyStorage(String),
 }
 
 pub type RepositoryError = ICError<RepositoryErrorKind>;
@@ -685,6 +687,12 @@ impl Repository {
 
     #[instrument(skip(self))]
     pub async fn writable_session(&self, branch: &str) -> RepositoryResult<Session> {
+        if !self.storage.can_write() {
+            return Err(RepositoryErrorKind::ReadonlyStorage(
+                "Cannot create writable_session".to_string(),
+            )
+            .into());
+        }
         let ref_data =
             fetch_branch_tip(self.storage.as_ref(), &self.storage_settings, branch)
                 .await?;

icechunk/src/storage/logging.rs

@@ -53,6 +53,11 @@ impl Storage for LoggingStorage {
     fn default_settings(&self) -> Settings {
         self.backend.default_settings()
     }
+
+    fn can_write(&self) -> bool {
+        self.backend.can_write()
+    }
+
     async fn fetch_config(
         &self,
         settings: &Settings,

icechunk/src/storage/mod.rs

@@ -302,6 +302,9 @@ pub trait Storage: fmt::Debug + fmt::Display + private::Sealed + Sync + Send {
     fn default_settings(&self) -> Settings {
         Default::default()
     }
+
+    fn can_write(&self) -> bool;
+
     async fn fetch_config(&self, settings: &Settings)
         -> StorageResult<FetchConfigResult>;
     async fn update_config(
@@ -624,6 +627,7 @@ pub fn new_s3_storage(
         bucket,
         prefix,
         credentials.unwrap_or(S3Credentials::FromEnv),
+        true,
     )?;
     Ok(Arc::new(st))
 }
@@ -645,6 +649,7 @@ pub fn new_tigris_storage(
         bucket,
         prefix,
         credentials.unwrap_or(S3Credentials::FromEnv),
+        true,
     )?;
     Ok(Arc::new(st))
 }

icechunk/src/storage/object_store.rs

@@ -296,6 +296,10 @@ impl private::Sealed for ObjectStorage {}
 #[async_trait]
 #[typetag::serde]
 impl Storage for ObjectStorage {
+    fn can_write(&self) -> bool {
+        true
+    }
+
     #[instrument(skip(self))]
     fn default_settings(&self) -> Settings {
         self.backend.default_settings()

icechunk/src/storage/s3.rs

@@ -51,6 +51,7 @@ pub struct S3Storage {
     credentials: S3Credentials,
     bucket: String,
     prefix: String,
+    can_write: bool,
 
     #[serde(skip)]
     /// We need to use OnceCell to allow async initialization, because serde
@@ -134,6 +135,7 @@ impl S3Storage {
         bucket: String,
         prefix: Option<String>,
         credentials: S3Credentials,
+        can_write: bool,
     ) -> Result<S3Storage, StorageError> {
         let client = OnceCell::new();
         Ok(S3Storage {
@@ -142,6 +144,7 @@ impl S3Storage {
             bucket,
             prefix: prefix.unwrap_or_default(),
             credentials,
+            can_write,
         })
     }
 
@@ -290,6 +293,10 @@ impl private::Sealed for S3Storage {}
 #[async_trait]
 #[typetag::serde]
 impl Storage for S3Storage {
+    fn can_write(&self) -> bool {
+        self.can_write
+    }
+
     #[instrument(skip(self, _settings))]
     async fn fetch_config(
         &self,
@@ -791,14 +798,15 @@ mod tests {
             "bucket".to_string(),
             Some("prefix".to_string()),
             credentials,
+            true,
         )
         .unwrap();
 
         let serialized = serde_json::to_string(&storage).unwrap();
 
         assert_eq!(
             serialized,
-            r#"{"config":{"region":"us-west-2","endpoint_url":"http://localhost:9000","anonymous":false,"allow_http":true},"credentials":{"s3_credential_type":"static","access_key_id":"access_key_id","secret_access_key":"secret_access_key","session_token":"session_token","expires_after":null},"bucket":"bucket","prefix":"prefix"}"#
+            r#"{"config":{"region":"us-west-2","endpoint_url":"http://localhost:9000","anonymous":false,"allow_http":true},"credentials":{"s3_credential_type":"static","access_key_id":"access_key_id","secret_access_key":"secret_access_key","session_token":"session_token","expires_after":null},"bucket":"bucket","prefix":"prefix","can_write":true}"#
         );
 
         let deserialized: S3Storage = serde_json::from_str(&serialized).unwrap();
@@ -817,6 +825,7 @@ mod tests {
             "bucket".to_string(),
             Some("prefix".to_string()),
             S3Credentials::FromEnv,
+            true,
         )
         .unwrap();