Repository: earth-mover/icechunk (language: python)

Issue:
Support persistent (default) configuration



Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 6ff09beb..3e0e3a17 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -51,6 +51,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "anyhow"
+version = "1.0.94"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c1fd03a028ef38ba2276dce7e33fcd6369c158a1bca17946c4b1b701891c1ff7"
+
 [[package]]
 name = "async-recursion"
 version = "1.1.1"
@@ -103,9 +109,9 @@ checksum = "0c4b4d0bd25bd0b74681c0ad21497610ce1b7c91b1022cd21c80c6fbdd9476b0"
 
 [[package]]
 name = "aws-config"
-version = "1.5.8"
+version = "1.5.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7198e6f03240fdceba36656d8be440297b6b82270325908c7381f37d826a74f6"
+checksum = "9b49afaa341e8dd8577e1a2200468f98956d6eda50bcf4a53246cc00174ba924"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
@@ -114,7 +120,7 @@ dependencies = [
  "aws-sdk-sts",
  "aws-smithy-async",
  "aws-smithy-http",
- "aws-smithy-json",
+ "aws-smithy-json 0.60.7",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -145,9 +151,9 @@ dependencies = [
 
 [[package]]
 name = "aws-runtime"
-version = "1.4.3"
+version = "1.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a10d5c055aa540164d9561a0e2e74ad30f0dcf7393c3a92f6733ddf9c5762468"
+checksum = "b5ac934720fbb46206292d2c75b57e67acfc56fe7dfd34fb9a02334af08409ea"
 dependencies = [
  "aws-credential-types",
  "aws-sigv4",
@@ -171,11 +177,10 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-s3"
-version = "1.54.0"
+version = "1.65.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2f2a62020f3e06f9b352b2a23547f6e1d110b6bf1e18a6b588ae36114eaf6e2"
+checksum = "d3ba2c5c0f2618937ce3d4a5ad574b86775576fa24006bcb3128c6e2cbf3c34e"
 dependencies = [
- "ahash",
  "aws-credential-types",
  "aws-runtime",
  "aws-sigv4",
@@ -183,7 +188,7 @@ dependencies = [
  "aws-smithy-checksums",
  "aws-smithy-eventstream",
  "aws-smithy-http",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.1",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -206,15 +211,15 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-sso"
-version = "1.45.0"
+version = "1.50.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e33ae899566f3d395cbf42858e433930682cc9c1889fa89318896082fef45efb"
+checksum = "05ca43a4ef210894f93096039ef1d6fa4ad3edfabb3be92b80908b9f2e4b4eab"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-smithy-async",
  "aws-smithy-http",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.1",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -228,15 +233,15 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-ssooidc"
-version = "1.46.0"
+version = "1.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f39c09e199ebd96b9f860b0fce4b6625f211e064ad7c8693b72ecf7ef03881e0"
+checksum = "abaf490c2e48eed0bb8e2da2fb08405647bd7f253996e0f93b981958ea0f73b0"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-smithy-async",
  "aws-smithy-http",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.1",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -250,15 +255,15 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-sts"
-version = "1.45.0"
+version = "1.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3d95f93a98130389eb6233b9d615249e543f6c24a68ca1f109af9ca5164a8765"
+checksum = "b68fde0d69c8bfdc1060ea7da21df3e39f6014da316783336deff0a9ec28f4bf"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-smithy-async",
  "aws-smithy-http",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.1",
  "aws-smithy-query",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
@@ -273,9 +278,9 @@ dependencies = [
 
 [[package]]
 name = "aws-sigv4"
-version = "1.2.4"
+version = "1.2.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cc8db6904450bafe7473c6ca9123f88cc11089e41a025408f992db4e22d3be68"
+checksum = "7d3820e0c08d0737872ff3c7c1f21ebbb6693d832312d6152bf18ef50a5471c2"
 dependencies = [
  "aws-credential-types",
  "aws-smithy-eventstream",
@@ -313,9 +318,9 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-checksums"
-version = "0.60.12"
+version = "0.60.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "598b1689d001c4d4dc3cb386adb07d37786783aee3ac4b324bcadac116bf3d23"
+checksum = "ba1a71073fca26775c8b5189175ea8863afb1c9ea2cceb02a5de5ad9dfbaa795"
 dependencies = [
  "aws-smithy-http",
  "aws-smithy-types",
@@ -373,6 +378,15 @@ dependencies = [
  "aws-smithy-types",
 ]
 
+[[package]]
+name = "aws-smithy-json"
+version = "0.61.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee4e69cc50921eb913c6b662f8d909131bb3e6ad6cb6090d3a39b66fc5c52095"
+dependencies = [
+ "aws-smithy-types",
+]
+
 [[package]]
 name = "aws-smithy-query"
 version = "0.60.7"
@@ -385,9 +399,9 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-runtime"
-version = "1.7.1"
+version = "1.7.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d1ce695746394772e7000b39fe073095db6d45a862d0767dd5ad0ac0d7f8eb87"
+checksum = "9f20685047ca9d6f17b994a07f629c813f08b5bce65523e47124879e60103d45"
 dependencies = [
  "aws-smithy-async",
  "aws-smithy-http",
@@ -412,9 +426,9 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-runtime-api"
-version = "1.7.2"
+version = "1.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e086682a53d3aa241192aa110fa8dfce98f2f5ac2ead0de84d41582c7e8fdb96"
+checksum = "92165296a47a812b267b4f41032ff8069ab7ff783696d217f0994a0d7ab585cd"
 dependencies = [
  "aws-smithy-async",
  "aws-smithy-types",
@@ -429,9 +443,9 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-types"
-version = "1.2.7"
+version = "1.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "147100a7bea70fa20ef224a6bad700358305f5dc0f84649c53769761395b355b"
+checksum = "4fbd94a32b3a7d55d3806fe27d98d3ad393050439dd05eb53ece36ec5e3d3510"
 dependencies = [
  "base64-simd",
  "bytes",
@@ -1234,6 +1248,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_with",
+ "serde_yml",
  "tempfile",
  "test-strategy",
  "thiserror",
@@ -1473,6 +1488,16 @@ version = "0.2.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4ec2a862134d2a7d32d7983ddcdd1c4923530833c9f2ea1a44fc5fa473989058"
 
+[[package]]
+name = "libyml"
+version = "0.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3302702afa434ffa30847a83305f0a69d6abd74293b6554c18ec85c7ef30c980"
+dependencies = [
+ "anyhow",
+ "version_check",
+]
+
 [[package]]
 name = "linux-raw-sys"
 version = "0.4.14"
@@ -2227,6 +2252,21 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "serde_yml"
+version = "0.0.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "59e2dd588bf1597a252c3b920e0143eb99b0f76e4e082f4c92ce34fbc9e71ddd"
+dependencies = [
+ "indexmap 2.2.6",
+ "itoa",
+ "libyml",
+ "memchr",
+ "ryu",
+ "serde",
+ "version_check",
+]
+
 [[package]]
 name = "sha1"
 version = "0.10.6"
@@ -2581,9 +2621,9 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
 
 [[package]]
 name = "typed-path"
-version = "0.9.3"
+version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "82205ffd44a9697e34fc145491aa47310f9871540bb7909eaa9365e0a9a46607"
+checksum = "41713888c5ccfd99979fcd1afd47b71652e331b3d4a0e19d30769e80fec76cce"
 
 [[package]]
 name = "typenum"
diff --git a/icechunk/Cargo.toml b/icechunk/Cargo.toml
index c20554c4..546c630e 100644
--- a/icechunk/Cargo.toml
+++ b/icechunk/Cargo.toml
@@ -35,11 +35,15 @@ rmp-serde = "1.3.0"
 url = "2.5.4"
 async-stream = "0.3.6"
 rmpv = { version = "1.3.0", features = ["serde", "with-serde"] }
-aws-sdk-s3 = "1.54.0"
-aws-config = "1.5.8"
+aws-sdk-s3 = "1.65.0"
+aws-config = "1.5.10"
 aws-credential-types = "1.2.1"
-typed-path = "0.9.3"
-aws-smithy-types-convert = { version = "0.60.8", features = ["convert-chrono", "convert-streams"] }
+typed-path = "0.10.0"
+aws-smithy-types-convert = { version = "0.60.8", features = [
+  "convert-chrono",
+  "convert-streams",
+] }
+serde_yml = "0.0.12"
 
 [dev-dependencies]
 pretty_assertions = "1.4.1"
diff --git a/icechunk/examples/low_level_dataset.rs b/icechunk/examples/low_level_dataset.rs
index a7a980d5..046a9454 100644
--- a/icechunk/examples/low_level_dataset.rs
+++ b/icechunk/examples/low_level_dataset.rs
@@ -209,7 +209,7 @@ ds.flush("a message", Default::default()).await?;
 
     println!("## Creating a new Repository instance @ latest version");
 
-    let mut ds = Repository::update(Arc::clone(&storage), v2_id.clone()).build();
+    let mut ds = Repository::update(Arc::clone(&storage), v2_id.clone()).await?.build();
 
     println!(
         r#"
@@ -261,7 +261,7 @@ ds.flush("commit", Default::default()).await?;
     );
 
     println!("Creating a new Repository instance, on the previous version");
-    let ds = Repository::update(Arc::clone(&storage), v2_id.clone()).build();
+    let ds = Repository::update(Arc::clone(&storage), v2_id.clone()).await?.build();
 
     println!(
         r#"
diff --git a/icechunk/src/repository.rs b/icechunk/src/repository.rs
index f856a3d0..1dc7c075 100644
--- a/icechunk/src/repository.rs
+++ b/icechunk/src/repository.rs
@@ -24,15 +24,19 @@ use crate::{
         manifest::VirtualReferenceError, snapshot::ManifestFileInfo,
         transaction_log::TransactionLog, ManifestId, SnapshotId,
     },
-    storage::virtual_ref::{
-        construct_valid_byte_range, ObjectStoreVirtualChunkResolverConfig,
-        VirtualChunkResolver,
+    storage::{
+        virtual_ref::{
+            construct_valid_byte_range, ObjectStoreVirtualChunkResolverConfig,
+            VirtualChunkResolver,
+        },
+        ETag,
     },
 };
 use bytes::Bytes;
 use chrono::Utc;
 use futures::{future::ready, Future, FutureExt, Stream, StreamExt, TryStreamExt};
 use itertools::Either;
+use serde::{Deserialize, Serialize};
 use thiserror::Error;
 
 use crate::{
@@ -54,7 +58,7 @@ use crate::{
     MemCachingStorage, Storage, StorageError,
 };
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
 pub struct RepositoryConfig {
     // Chunks smaller than this will be stored inline in the manifst
     pub inline_chunk_threshold_bytes: u16,
@@ -74,6 +78,7 @@ impl Default for RepositoryConfig {
 #[derive(Debug)]
 pub struct Repository {
     config: RepositoryConfig,
+    config_etag: Option<ETag>,
     storage: Arc<dyn Storage + Send + Sync>,
     snapshot_id: SnapshotId,
     change_set: ChangeSet,
@@ -83,6 +88,7 @@ pub struct Repository {
 #[derive(Debug, Clone)]
 pub struct RepositoryBuilder {
     config: RepositoryConfig,
+    config_etag: Option<ETag>,
     storage: Arc<dyn Storage + Send + Sync>,
     snapshot_id: SnapshotId,
     change_set: Option<ChangeSet>,
@@ -90,9 +96,15 @@ pub struct RepositoryBuilder {
 }
 
 impl RepositoryBuilder {
-    fn new(storage: Arc<dyn Storage + Send + Sync>, snapshot_id: SnapshotId) -> Self {
+    fn new(
+        storage: Arc<dyn Storage + Send + Sync>,
+        snapshot_id: SnapshotId,
+        config: RepositoryConfig,
+        config_etag: Option<ETag>,
+    ) -> Self {
         Self {
-            config: RepositoryConfig::default(),
+            config,
+            config_etag,
             snapshot_id,
             storage,
             change_set: None,
@@ -110,11 +122,6 @@ impl RepositoryBuilder {
         self
     }
 
-    pub fn with_config(&mut self, config: RepositoryConfig) -> &mut Self {
-        self.config = config;
-        self
-    }
-
     pub fn with_virtual_ref_config(
         &mut self,
         config: ObjectStoreVirtualChunkResolverConfig,
@@ -131,6 +138,7 @@ impl RepositoryBuilder {
     pub fn build(&self) -> Repository {
         Repository::new(
             self.config.clone(),
+            self.config_etag.clone(),
             self.storage.clone(),
             self.snapshot_id.clone(),
             self.change_set.clone(),
@@ -178,18 +186,60 @@ pub enum RepositoryError {
     DeserializationError(#[from] rmp_serde::decode::Error),
     #[error("error finding conflicting path for node `{0}`, this probably indicades a bug in `rebase`")]
     ConflictingPathNotFound(NodeId),
+    #[error("error in config deserialization `{0}`")]
+    ConfigDeserializationError(#[from] serde_yml::Error),
 }
 
 pub type RepositoryResult<T> = Result<T, RepositoryError>;
 
-// FIXME: what do we want to do with implicit groups?
-//
 impl Repository {
-    pub fn update(
+    pub async fn fetch_config(
+        storage: &(dyn Storage + Send + Sync),
+    ) -> RepositoryResult<Option<(RepositoryConfig, ETag)>> {
+        match storage.fetch_config().await? {
+            Some((bytes, etag)) => {
+                let config = serde_yml::from_slice(&bytes)?;
+                Ok(Some((config, etag)))
+            }
+            None => Ok(None),
+        }
+    }
+
+    pub async fn update(
         storage: Arc<dyn Storage + Send + Sync>,
         previous_version_snapshot_id: SnapshotId,
-    ) -> RepositoryBuilder {
-        RepositoryBuilder::new(storage, previous_version_snapshot_id)
+    ) -> RepositoryResult<RepositoryBuilder> {
+        let storage_c1 = Arc::clone(&storage);
+        let storage_c2 = Arc::clone(&storage);
+
+        let snap_id = previous_version_snapshot_id.clone();
+        let h1 = tokio::spawn(async move { storage_c1.fetch_snapshot(&snap_id).await });
+        let h2 =
+            tokio::spawn(async move { Self::fetch_config(storage_c2.as_ref()).await });
+
+        // Fail if the snapshot id cannot be found
+        #[allow(clippy::expect_used)]
+        let _must_exist = h1
+            .await
+            .expect("Error fetching config. Perhaps thread died unexpectedly.")?;
+        #[allow(clippy::expect_used)]
+        match h2
+            .await
+            .expect("Error fetching config. Perhaps thread died unexpectedly.")?
+        {
+            Some((config, etag)) => Ok(RepositoryBuilder::new(
+                storage,
+                previous_version_snapshot_id,
+                config,
+                Some(etag),
+            )),
+            None => Ok(RepositoryBuilder::new(
+                storage,
+                previous_version_snapshot_id,
+                RepositoryConfig::default(),
+                None,
+            )),
+        }
     }
 
     pub async fn from_branch_tip(
@@ -197,7 +247,7 @@ impl Repository {
         branch_name: &str,
     ) -> RepositoryResult<RepositoryBuilder> {
         let snapshot_id = fetch_branch_tip(storage.as_ref(), branch_name).await?.snapshot;
-        Ok(Self::update(storage, snapshot_id))
+        Self::update(storage, snapshot_id).await
     }
 
     pub async fn from_tag(
@@ -205,7 +255,7 @@ impl Repository {
         tag_name: &str,
     ) -> RepositoryResult<RepositoryBuilder> {
         let ref_data = fetch_tag(storage.as_ref(), tag_name).await?;
-        Ok(Self::update(storage, ref_data.snapshot))
+        Self::update(storage, ref_data.snapshot).await
     }
 
     /// Initialize a new repository with a single empty commit to the main branch.
@@ -233,7 +283,12 @@ impl Repository {
 
         debug_assert!(Self::exists(storage.as_ref()).await.unwrap_or(false));
 
-        Ok(RepositoryBuilder::new(storage, new_snapshot_id))
+        Ok(RepositoryBuilder::new(
+            storage,
+            new_snapshot_id,
+            RepositoryConfig::default(),
+            None,
+        ))
     }
 
     pub async fn exists(storage: &(dyn Storage + Send + Sync)) -> RepositoryResult<bool> {
@@ -255,6 +310,7 @@ impl Repository {
 
     fn new(
         config: RepositoryConfig,
+        config_etag: Option<ETag>,
         storage: Arc<dyn Storage + Send + Sync>,
         snapshot_id: SnapshotId,
         change_set: Option<ChangeSet>,
@@ -263,6 +319,7 @@ impl Repository {
         Repository {
             snapshot_id,
             config,
+            config_etag,
             storage,
             change_set: change_set.unwrap_or_default(),
             virtual_resolver: Arc::new(ObjectStoreVirtualChunkResolver::new(
@@ -275,6 +332,12 @@ impl Repository {
         &self.config
     }
 
+    pub async fn update_config(&self) -> RepositoryResult<ETag> {
+        let bytes = Bytes::from(serde_yml::to_string(self.config())?);
+        let res = self.storage.update_config(bytes, self.config_etag.as_deref()).await?;
+        Ok(res)
+    }
+
     pub(crate) fn set_snapshot_id(&mut self, snapshot_id: SnapshotId) {
         self.snapshot_id = snapshot_id;
     }
@@ -842,6 +905,7 @@ impl Repository {
                 let tx_log = self.storage.fetch_transaction_log(&snap_id).await?;
                 let repo = Repository {
                     config: self.config().clone(),
+                    config_etag: self.config_etag.clone(),
                     storage: self.storage.clone(),
                     snapshot_id: snap_id.clone(),
                     change_set: ChangeSet::default(),
@@ -1466,6 +1530,7 @@ mod tests {
         let snapshot_id = ObjectId::random();
         storage.write_snapshot(snapshot_id.clone(), snapshot).await?;
         let mut ds = Repository::update(Arc::new(storage), snapshot_id)
+            .await?
             .with_inline_threshold_bytes(512)
             .build();
 
@@ -1843,7 +1908,7 @@ mod tests {
         .await?;
 
         let snapshot_id = ds.flush("commit", SnapshotProperties::default()).await?;
-        let ds = Repository::update(Arc::clone(&storage), snapshot_id).build();
+        let ds = Repository::update(Arc::clone(&storage), snapshot_id).await?.build();
 
         assert_eq!(
             ds.get_chunk_ref(&new_array_path, &ChunkIndices(vec![0, 0, 0])).await?,
@@ -1869,7 +1934,8 @@ mod tests {
         assert!(logging.fetch_operations().is_empty());
 
         //test the previous version is still alive
-        let ds = Repository::update(Arc::clone(&storage), previous_snapshot_id).build();
+        let ds =
+            Repository::update(Arc::clone(&storage), previous_snapshot_id).await?.build();
         assert_eq!(
             ds.get_chunk_ref(&new_array_path, &ChunkIndices(vec![0, 0, 0])).await?,
             Some(ChunkPayload::Inline("bye".into()))
@@ -2178,7 +2244,7 @@ mod tests {
         )
         .await?;
         let snapshot_id = ds.flush("commit", SnapshotProperties::default()).await?;
-        let ds = Repository::update(Arc::clone(&storage), snapshot_id).build();
+        let ds = Repository::update(Arc::clone(&storage), snapshot_id).await?.build();
         let coords = ds
             .all_chunks()
             .await?
@@ -2761,7 +2827,9 @@ mod tests {
         // let's try to create a new commit, that conflicts with the previous one but writes to
         // different chunks
         let mut repo2 =
-            Repository::update(Arc::clone(&storage), array_created_snap.clone()).build();
+            Repository::update(Arc::clone(&storage), array_created_snap.clone())
+                .await?
+                .build();
         repo2
             .set_chunk_ref(
                 new_array_path.clone(),
@@ -2814,7 +2882,9 @@ mod tests {
         // let's try to create a new commit, that conflicts with the previous one and writes
         // to the same chunk, recovering with "Fail" policy (so it shouldn't recover)
         let mut repo2 =
-            Repository::update(Arc::clone(&storage), array_created_snap.clone()).build();
+            Repository::update(Arc::clone(&storage), array_created_snap.clone())
+                .await?
+                .build();
         repo2
             .set_chunk_ref(
                 new_array_path.clone(),
@@ -2858,7 +2928,9 @@ mod tests {
         // let's try to create a new commit, that conflicts with the previous one and writes
         // to the same chunk, recovering with "UseOurs" policy
         let mut repo2 =
-            Repository::update(Arc::clone(&storage), array_created_snap.clone()).build();
+            Repository::update(Arc::clone(&storage), array_created_snap.clone())
+                .await?
+                .build();
         repo2
             .set_chunk_ref(
                 new_array_path.clone(),
@@ -2900,7 +2972,9 @@ mod tests {
         // let's try to create a new commit, that conflicts with the previous one and writes
         // to the same chunk, recovering with "UseTheirs" policy
         let mut repo2 =
-            Repository::update(Arc::clone(&storage), array_created_snap.clone()).build();
+            Repository::update(Arc::clone(&storage), array_created_snap.clone())
+                .await?
+                .build();
         repo2
             .set_chunk_ref(
                 new_array_path.clone(),
@@ -3175,6 +3249,51 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_repository_persistent_config() -> Result<(), Box<dyn Error>> {
+        let storage: Arc<dyn Storage + Send + Sync> =
+            Arc::new(ObjectStorage::new_in_memory_store(Some("prefix".into())));
+
+        let repo = Repository::init(Arc::clone(&storage), false).await?.build();
+
+        // initializing a repo doesn't create the config file
+        assert!(Repository::fetch_config(storage.as_ref()).await?.is_none());
+        // it inits with the default config
+        assert_eq!(repo.config(), &RepositoryConfig::default());
+        // updating the persistent config create a new file with default values
+        let etag = repo.update_config().await?;
+        assert_ne!(etag, "");
+        assert_eq!(
+            Repository::fetch_config(storage.as_ref()).await?.unwrap().0,
+            RepositoryConfig::default()
+        );
+
+        // reload the repo changing config
+        let repo = Repository::from_branch_tip(storage.clone(), "main")
+            .await?
+            .with_inline_threshold_bytes(42)
+            .build();
+
+        assert_eq!(repo.config().inline_chunk_threshold_bytes, 42);
+
+        // update the persistent config
+        let etag = repo.update_config().await?;
+        assert_ne!(etag, "");
+        assert_eq!(
+            Repository::fetch_config(storage.as_ref())
+                .await?
+                .unwrap()
+                .0
+                .inline_chunk_threshold_bytes,
+            42
+        );
+
+        // verify loading againg gets the value from persistent config
+        let repo = Repository::from_branch_tip(storage.clone(), "main").await?.build();
+        assert_eq!(repo.config().inline_chunk_threshold_bytes, 42);
+        Ok(())
+    }
+
     #[cfg(test)]
     mod state_machine_test {
         use crate::format::snapshot::NodeData;
diff --git a/icechunk/src/storage/caching.rs b/icechunk/src/storage/caching.rs
index 49dba59a..71e3ab90 100644
--- a/icechunk/src/storage/caching.rs
+++ b/icechunk/src/storage/caching.rs
@@ -14,7 +14,7 @@ use crate::{
     private,
 };
 
-use super::{ListInfo, Storage, StorageError, StorageResult};
+use super::{ETag, ListInfo, Storage, StorageError, StorageResult};
 
 #[derive(Debug)]
 pub struct MemCachingStorage {
@@ -50,6 +50,17 @@ impl private::Sealed for MemCachingStorage {}
 
 #[async_trait]
 impl Storage for MemCachingStorage {
+    async fn fetch_config(&self) -> StorageResult<Option<(Bytes, ETag)>> {
+        self.backend.fetch_config().await
+    }
+    async fn update_config(
+        &self,
+        config: Bytes,
+        etag: Option<&str>,
+    ) -> StorageResult<ETag> {
+        self.backend.update_config(config, etag).await
+    }
+
     async fn fetch_snapshot(
         &self,
         id: &SnapshotId,
diff --git a/icechunk/src/storage/logging.rs b/icechunk/src/storage/logging.rs
index 2ef2ffd5..159316b0 100644
--- a/icechunk/src/storage/logging.rs
+++ b/icechunk/src/storage/logging.rs
@@ -4,7 +4,7 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use futures::stream::BoxStream;
 
-use super::{ListInfo, Storage, StorageError, StorageResult};
+use super::{ETag, ListInfo, Storage, StorageError, StorageResult};
 use crate::{
     format::{
         attributes::AttributesTable, manifest::Manifest, snapshot::Snapshot,
@@ -36,6 +36,16 @@ impl private::Sealed for LoggingStorage {}
 #[async_trait]
 #[allow(clippy::expect_used)] // this implementation is intended for tests only
 impl Storage for LoggingStorage {
+    async fn fetch_config(&self) -> StorageResult<Option<(Bytes, ETag)>> {
+        self.backend.fetch_config().await
+    }
+    async fn update_config(
+        &self,
+        config: Bytes,
+        etag: Option<&str>,
+    ) -> StorageResult<ETag> {
+        self.backend.update_config(config, etag).await
+    }
     async fn fetch_snapshot(
         &self,
         id: &SnapshotId,
diff --git a/icechunk/src/storage/mod.rs b/icechunk/src/storage/mod.rs
index 6e86f14a..f32a3e56 100644
--- a/icechunk/src/storage/mod.rs
+++ b/icechunk/src/storage/mod.rs
@@ -61,6 +61,8 @@ pub enum StorageError {
     RefAlreadyExists(String),
     #[error("ref not found: {0}")]
     RefNotFound(String),
+    #[error("the etag does not match")]
+    ConfigUpdateConflict,
     #[error("unknown storage error: {0}")]
     Other(String),
 }
@@ -78,6 +80,9 @@ const MANIFEST_PREFIX: &str = "manifests/";
 const CHUNK_PREFIX: &str = "chunks/";
 const REF_PREFIX: &str = "refs";
 const TRANSACTION_PREFIX: &str = "transactions/";
+const CONFIG_PATH: &str = "config.yaml";
+
+pub type ETag = String;
 
 /// Fetch and write the parquet files that represent the repository in object store
 ///
@@ -85,6 +90,13 @@ const TRANSACTION_PREFIX: &str = "transactions/";
 /// Implementations are free to assume files are never overwritten.
 #[async_trait]
 pub trait Storage: fmt::Debug + private::Sealed {
+    async fn fetch_config(&self) -> StorageResult<Option<(Bytes, ETag)>>;
+    async fn update_config(
+        &self,
+        config: Bytes,
+        etag: Option<&str>,
+    ) -> StorageResult<ETag>;
+
     async fn fetch_snapshot(&self, id: &SnapshotId) -> StorageResult<Arc<Snapshot>>;
     async fn fetch_attributes(
         &self,
diff --git a/icechunk/src/storage/object_store.rs b/icechunk/src/storage/object_store.rs
index 4731c77b..f46d5c73 100644
--- a/icechunk/src/storage/object_store.rs
+++ b/icechunk/src/storage/object_store.rs
@@ -15,7 +15,7 @@ use futures::{
 use object_store::{
     local::LocalFileSystem, memory::InMemory, path::Path as ObjectPath, Attribute,
     AttributeValue, Attributes, GetOptions, GetRange, ObjectMeta, ObjectStore, PutMode,
-    PutOptions, PutPayload,
+    PutOptions, PutPayload, UpdateVersion,
 };
 use std::{
     fs::create_dir_all,
@@ -29,8 +29,8 @@ use std::{
 };
 
 use super::{
-    ListInfo, Storage, StorageError, StorageResult, CHUNK_PREFIX, MANIFEST_PREFIX,
-    REF_PREFIX, SNAPSHOT_PREFIX, TRANSACTION_PREFIX,
+    ETag, ListInfo, Storage, StorageError, StorageResult, CHUNK_PREFIX, CONFIG_PATH,
+    MANIFEST_PREFIX, REF_PREFIX, SNAPSHOT_PREFIX, TRANSACTION_PREFIX,
 };
 
 // Get Range is object_store specific, keep it with this module
@@ -118,6 +118,10 @@ impl ObjectStorage {
         self.get_path_str(file_prefix, id.to_string().as_str())
     }
 
+    fn get_config_path(&self) -> ObjectPath {
+        self.get_path_str("", CONFIG_PATH)
+    }
+
     fn get_snapshot_path(&self, id: &SnapshotId) -> ObjectPath {
         self.get_path(SNAPSHOT_PREFIX, id)
     }
@@ -176,6 +180,59 @@ impl private::Sealed for ObjectStorage {}
 
 #[async_trait]
 impl Storage for ObjectStorage {
+    async fn fetch_config(&self) -> StorageResult<Option<(Bytes, ETag)>> {
+        let path = self.get_config_path();
+        let response = self.store.get(&path).await;
+
+        match response {
+            Ok(result) => match result.meta.e_tag.clone() {
+                Some(etag) => Ok(Some((result.bytes().await?, etag))),
+                None => Err(StorageError::Other("No ETag found for config".to_string())),
+            },
+            Err(object_store::Error::NotFound { .. }) => Ok(None),
+            Err(err) => Err(err.into()),
+        }
+    }
+    async fn update_config(
+        &self,
+        config: Bytes,
+        etag: Option<&str>,
+    ) -> StorageResult<ETag> {
+        let path = self.get_config_path();
+        let attributes = if self.supports_metadata {
+            Attributes::from_iter(vec![(
+                Attribute::ContentType,
+                AttributeValue::from("application/yaml"),
+            )])
+        } else {
+            Attributes::new()
+        };
+
+        let mode = if let Some(etag) = etag {
+            PutMode::Update(UpdateVersion {
+                e_tag: Some(etag.to_string()),
+                version: None,
+            })
+        } else {
+            PutMode::Create
+        };
+
+        let options = PutOptions { mode, attributes, ..PutOptions::default() };
+        let res = self.store.put_opts(&path, config.into(), options).await;
+        match res {
+            Ok(res) => {
+                let etag = res.e_tag.ok_or(StorageError::Other(
+                    "Config object should have an etag".to_string(),
+                ))?;
+                Ok(etag)
+            }
+            Err(object_store::Error::Precondition { .. }) => {
+                Err(StorageError::ConfigUpdateConflict)
+            }
+            Err(err) => Err(err.into()),
+        }
+    }
+
     async fn fetch_snapshot(
         &self,
         id: &SnapshotId,
diff --git a/icechunk/src/storage/s3.rs b/icechunk/src/storage/s3.rs
index 3533bafe..4d658e78 100644
--- a/icechunk/src/storage/s3.rs
+++ b/icechunk/src/storage/s3.rs
@@ -10,11 +10,14 @@ use std::{
 
 use async_stream::try_stream;
 use async_trait::async_trait;
-use aws_config::{meta::region::RegionProviderChain, AppName, BehaviorVersion};
+use aws_config::{
+    meta::region::RegionProviderChain, retry::ProvideErrorKind, AppName, BehaviorVersion,
+};
 use aws_credential_types::Credentials;
 use aws_sdk_s3::{
     config::{Builder, Region},
-    error::ProvideErrorMetadata,
+    error::SdkError,
+    operation::put_object::PutObjectError,
     primitives::ByteStream,
     types::{Delete, Object, ObjectIdentifier},
     Client,
@@ -39,8 +42,8 @@ use crate::{
 };
 
 use super::{
-    ListInfo, StorageResult, CHUNK_PREFIX, MANIFEST_PREFIX, REF_PREFIX, SNAPSHOT_PREFIX,
-    TRANSACTION_PREFIX,
+    ETag, ListInfo, StorageResult, CHUNK_PREFIX, CONFIG_PATH, MANIFEST_PREFIX,
+    REF_PREFIX, SNAPSHOT_PREFIX, TRANSACTION_PREFIX,
 };
 
 #[derive(Debug)]
@@ -146,6 +149,10 @@ impl S3Storage {
         self.get_path_str(file_prefix, id.to_string().as_str())
     }
 
+    fn get_config_path(&self) -> StorageResult<String> {
+        self.get_path_str("", CONFIG_PATH)
+    }
+
     fn get_snapshot_path(&self, id: &SnapshotId) -> StorageResult<String> {
         self.get_path(SNAPSHOT_PREFIX, id)
     }
@@ -265,6 +272,76 @@ impl private::Sealed for S3Storage {}
 
 #[async_trait]
 impl Storage for S3Storage {
+    async fn fetch_config(&self) -> StorageResult<Option<(Bytes, ETag)>> {
+        let key = self.get_config_path()?;
+        let res =
+            self.client.get_object().bucket(self.bucket.clone()).key(key).send().await;
+
+        match res {
+            Ok(output) => match output.e_tag {
+                Some(etag) => Ok(Some((output.body.collect().await?.into_bytes(), etag))),
+                None => Err(StorageError::Other("No ETag found for config".to_string())),
+            },
+            Err(sdk_err) => match sdk_err.as_service_error() {
+                Some(e) if e.is_no_such_key() => Ok(None),
+                _ => Err(sdk_err.into()),
+            },
+        }
+    }
+
+    async fn update_config(
+        &self,
+        config: Bytes,
+        etag: Option<&str>,
+    ) -> StorageResult<ETag> {
+        let key = self.get_config_path()?;
+        let mut req = self
+            .client
+            .put_object()
+            .bucket(self.bucket.clone())
+            .key(key)
+            .content_type("application/yaml")
+            .body(config.into());
+
+        if let Some(etag) = etag {
+            req = req.if_match(etag)
+        } else {
+            req = req.if_none_match("*")
+        }
+
+        let res = req.send().await;
+
+        match res {
+            Ok(out) => {
+                let etag = out.e_tag().ok_or(StorageError::Other(
+                    "Config object should have an etag".to_string(),
+                ))?;
+                Ok(etag.to_string())
+            }
+            // minio returns this
+            Err(SdkError::ServiceError(err)) => {
+                if err.err().meta().code() == Some("PreconditionFailed") {
+                    Err(StorageError::ConfigUpdateConflict)
+                } else {
+                    Err(StorageError::from(SdkError::<PutObjectError>::ServiceError(err)))
+                }
+            }
+            // S3 API documents this
+            Err(SdkError::ResponseError(err)) => {
+                let status = err.raw().status().as_u16();
+                // see https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html#API_PutObject_RequestSyntax
+                if status == 409 || status == 412 {
+                    Err(StorageError::ConfigUpdateConflict)
+                } else {
+                    Err(StorageError::from(SdkError::<PutObjectError>::ResponseError(
+                        err,
+                    )))
+                }
+            }
+            Err(err) => Err(err.into()),
+        }
+    }
+
     async fn fetch_snapshot(&self, id: &SnapshotId) -> StorageResult<Arc<Snapshot>> {
         let key = self.get_snapshot_path(id)?;
         let bytes = self.get_object(key.as_str()).await?;
diff --git a/icechunk/src/zarr.rs b/icechunk/src/zarr.rs
index 94b23a69..b0fc66a5 100644
--- a/icechunk/src/zarr.rs
+++ b/icechunk/src/zarr.rs
@@ -165,7 +165,9 @@ impl RepositoryConfig {
                     (builder, Some(String::from(Ref::DEFAULT_BRANCH)))
                 }
                 Some(VersionInfo::SnapshotId(sid)) => {
-                    let builder = Repository::update(storage, sid.clone());
+                    let builder = Repository::update(storage, sid.clone())
+                        .await
+                        .map_err(|err| format!("Error initializing repository: {err}"))?;
                     (builder, None)
                 }
                 Some(VersionInfo::TagRef(tag)) => {
@@ -1885,7 +1887,7 @@ mod tests {
 
         let oid = store.commit("commit").await?;
 
-        let ds = Repository::update(storage, oid).build();
+        let ds = Repository::update(storage, oid).await?.build();
         let store = Store::from_repository(ds, AccessMode::ReadWrite, None, None);
         assert_eq!(
             store.get("array/c/0/1/0", &ByteRange::ALL).await.unwrap(),
@@ -2373,7 +2375,7 @@ mod tests {
         assert_eq!(store.get("array/c/0/1/0", &ByteRange::ALL).await.unwrap(), new_data);
 
         let new_store_from_snapshot = Store::from_repository(
-            Repository::update(Arc::clone(&storage), snapshot_id).build(),
+            Repository::update(Arc::clone(&storage), snapshot_id).await?.build(),
             AccessMode::ReadWrite,
             None,
             None,
@@ -2456,7 +2458,7 @@ mod tests {
         );
 
         let store = Store::from_repository(
-            Repository::update(Arc::clone(&storage), empty_snap).build(),
+            Repository::update(Arc::clone(&storage), empty_snap).await?.build(),
             AccessMode::ReadWrite,
             None,
             None,
diff --git a/shell.nix b/shell.nix
index b72cb731..732bb493 100644
--- a/shell.nix
+++ b/shell.nix
@@ -1,7 +1,7 @@
 let
-  # Pinned nixpkgs, deterministic. Last updated to nixos-unstable as of: 2024-10-06
+  # Pinned nixpkgs, deterministic. Last updated to nixos-unstable as of: 2024-12-10
   pkgs = import (fetchTarball
-    "https://github.com/NixOS/nixpkgs/archive/7d49afd36b5590f023ec56809c02e05d8164fbc4.tar.gz")
+    "https://github.com/NixOS/nixpkgs/archive/22c3f2cf41a0e70184334a958e6b124fb0ce3e01.tar.gz")
     { };
 
   # Rolling updates, not deterministic.
@@ -29,6 +29,7 @@ in pkgs.mkShell.override {
     just # script launcher with a make flavor
     alejandra # nix code formatter
     markdownlint-cli2
+    python3
   ];
 
   shellHook = ''


Test patch:
diff --git a/icechunk/tests/test_s3_storage.rs b/icechunk/tests/test_s3_storage.rs
deleted file mode 100644
index 0d35291d..00000000
--- a/icechunk/tests/test_s3_storage.rs
+++ /dev/null
@@ -1,190 +0,0 @@
-use std::{collections::HashSet, sync::Arc};
-
-use bytes::Bytes;
-use chrono::Utc;
-use icechunk::{
-    format::{
-        manifest::Manifest, snapshot::Snapshot, ByteRange, ChunkId, ManifestId,
-        SnapshotId,
-    },
-    refs::{
-        create_tag, fetch_branch_tip, fetch_tag, list_refs, update_branch, Ref, RefError,
-    },
-    storage::{
-        s3::{S3Config, S3Credentials, S3Storage, StaticS3Credentials},
-        StorageResult,
-    },
-    Storage,
-};
-use pretty_assertions::assert_eq;
-
-async fn mk_storage() -> StorageResult<S3Storage> {
-    S3Storage::new_s3_store(
-        "testbucket",
-        "test_s3_storage__".to_string() + Utc::now().to_rfc3339().as_str(),
-        Some(&S3Config {
-            region: Some("us-east-1".to_string()),
-            endpoint: Some("http://localhost:9000".to_string()),
-            credentials: S3Credentials::Static(StaticS3Credentials {
-                access_key_id: "minio123".into(),
-                secret_access_key: "minio123".into(),
-                session_token: None,
-            }),
-            allow_http: true,
-        }),
-    )
-    .await
-}
-
-#[tokio::test]
-pub async fn test_snapshot_write_read() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id = SnapshotId::random();
-    let snapshot = Arc::new(Snapshot::empty());
-    storage.write_snapshot(id.clone(), snapshot.clone()).await?;
-    let back = storage.fetch_snapshot(&id).await?;
-    assert_eq!(snapshot, back);
-    Ok(())
-}
-
-#[tokio::test]
-pub async fn test_manifest_write_read() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id = ManifestId::random();
-    let manifest = Arc::new(Manifest::default());
-    storage.write_manifests(id.clone(), manifest.clone()).await?;
-    let back = storage.fetch_manifests(&id).await?;
-    assert_eq!(manifest, back);
-    Ok(())
-}
-
-#[tokio::test]
-pub async fn test_chunk_write_read() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id = ChunkId::random();
-    let bytes = Bytes::from_static(b"hello");
-    storage.write_chunk(id.clone(), bytes.clone()).await?;
-    let back = storage.fetch_chunk(&id, &ByteRange::ALL).await?;
-    assert_eq!(bytes, back);
-
-    let back =
-        storage.fetch_chunk(&id, &ByteRange::from_offset_with_length(1, 2)).await?;
-    assert_eq!(Bytes::from_static(b"el"), back);
-
-    let back = storage.fetch_chunk(&id, &ByteRange::from_offset(1)).await?;
-    assert_eq!(Bytes::from_static(b"ello"), back);
-
-    let back = storage.fetch_chunk(&id, &ByteRange::to_offset(3)).await?;
-    assert_eq!(Bytes::from_static(b"hel"), back); // codespell:ignore
-
-    let back = storage.fetch_chunk(&id, &ByteRange::bounded(1, 4)).await?;
-    assert_eq!(Bytes::from_static(b"ell"), back);
-    Ok(())
-}
-
-#[tokio::test]
-pub async fn test_tag_write_get() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id = SnapshotId::random();
-    create_tag(&storage, "mytag", id.clone(), false).await?;
-    let back = fetch_tag(&storage, "mytag").await?;
-    assert_eq!(id, back.snapshot);
-    Ok(())
-}
-
-#[tokio::test]
-pub async fn test_fetch_non_existing_tag() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id = SnapshotId::random();
-    create_tag(&storage, "mytag", id.clone(), false).await?;
-
-    let back = fetch_tag(&storage, "non-existing-tag").await;
-    assert!(matches!(back, Err(RefError::RefNotFound(r)) if r == "non-existing-tag"));
-    Ok(())
-}
-
-#[tokio::test]
-pub async fn test_create_existing_tag() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id = SnapshotId::random();
-    create_tag(&storage, "mytag", id.clone(), false).await?;
-
-    let res = create_tag(&storage, "mytag", id.clone(), false).await;
-    assert!(matches!(res, Err(RefError::TagAlreadyExists(r)) if r == "mytag"));
-    Ok(())
-}
-
-#[tokio::test]
-pub async fn test_branch_initialization() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id = SnapshotId::random();
-
-    let res = update_branch(&storage, "some-branch", id.clone(), None, false).await?;
-    assert_eq!(res.0, 0);
-
-    let res = fetch_branch_tip(&storage, "some-branch").await?;
-    assert_eq!(res.snapshot, id);
-
-    Ok(())
-}
-
-#[tokio::test]
-pub async fn test_fetch_non_existing_branch() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id = SnapshotId::random();
-    update_branch(&storage, "some-branch", id.clone(), None, false).await?;
-
-    let back = fetch_branch_tip(&storage, "non-existing-branch").await;
-    assert!(matches!(back, Err(RefError::RefNotFound(r)) if r == "non-existing-branch"));
-    Ok(())
-}
-
-#[tokio::test]
-pub async fn test_branch_update() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id1 = SnapshotId::random();
-    let id2 = SnapshotId::random();
-    let id3 = SnapshotId::random();
-
-    let res = update_branch(&storage, "some-branch", id1.clone(), None, false).await?;
-    assert_eq!(res.0, 0);
-
-    let res =
-        update_branch(&storage, "some-branch", id2.clone(), Some(&id1), false).await?;
-    assert_eq!(res.0, 1);
-
-    let res =
-        update_branch(&storage, "some-branch", id3.clone(), Some(&id2), false).await?;
-    assert_eq!(res.0, 2);
-
-    let res = fetch_branch_tip(&storage, "some-branch").await?;
-    assert_eq!(res.snapshot, id3);
-
-    Ok(())
-}
-
-#[tokio::test]
-pub async fn test_ref_names() -> Result<(), Box<dyn std::error::Error>> {
-    let storage = mk_storage().await?;
-    let id1 = SnapshotId::random();
-    let id2 = SnapshotId::random();
-    update_branch(&storage, "main", id1.clone(), None, false).await?;
-    update_branch(&storage, "main", id2.clone(), Some(&id1), false).await?;
-    update_branch(&storage, "foo", id1.clone(), None, false).await?;
-    update_branch(&storage, "bar", id1.clone(), None, false).await?;
-    create_tag(&storage, "my-tag", id1.clone(), false).await?;
-    create_tag(&storage, "my-other-tag", id1.clone(), false).await?;
-
-    let res: HashSet<_> = HashSet::from_iter(list_refs(&storage).await?);
-    assert_eq!(
-        res,
-        HashSet::from_iter([
-            Ref::Tag("my-tag".to_string()),
-            Ref::Tag("my-other-tag".to_string()),
-            Ref::Branch("main".to_string()),
-            Ref::Branch("foo".to_string()),
-            Ref::Branch("bar".to_string()),
-        ])
-    );
-    Ok(())
-}
diff --git a/icechunk/tests/test_storage.rs b/icechunk/tests/test_storage.rs
new file mode 100644
index 00000000..8f37636f
--- /dev/null
+++ b/icechunk/tests/test_storage.rs
@@ -0,0 +1,323 @@
+use std::{collections::HashSet, future::Future, sync::Arc};
+
+use bytes::Bytes;
+use chrono::Utc;
+use icechunk::{
+    format::{
+        manifest::Manifest, snapshot::Snapshot, ByteRange, ChunkId, ManifestId,
+        SnapshotId,
+    },
+    refs::{
+        create_tag, fetch_branch_tip, fetch_tag, list_refs, update_branch, Ref, RefError,
+    },
+    storage::{
+        s3::{S3Config, S3Credentials, S3Storage, StaticS3Credentials},
+        StorageResult,
+    },
+    ObjectStorage, Storage, StorageError,
+};
+use pretty_assertions::{assert_eq, assert_ne};
+
+async fn mk_storage() -> StorageResult<S3Storage> {
+    S3Storage::new_s3_store(
+        "testbucket",
+        "test_s3_storage__".to_string() + Utc::now().to_rfc3339().as_str(),
+        Some(&S3Config {
+            region: Some("us-east-1".to_string()),
+            endpoint: Some("http://localhost:9000".to_string()),
+            credentials: S3Credentials::Static(StaticS3Credentials {
+                access_key_id: "minio123".into(),
+                secret_access_key: "minio123".into(),
+                session_token: None,
+            }),
+            allow_http: true,
+        }),
+    )
+    .await
+}
+
+fn mk_in_memory_storage() -> ObjectStorage {
+    ObjectStorage::new_in_memory_store(Some("prefix".to_string()))
+}
+
+async fn with_storage<F, Fut>(f: F) -> Result<(), Box<dyn std::error::Error>>
+where
+    F: Fn(Arc<dyn Storage + Send + Sync>) -> Fut,
+    Fut: Future<Output = Result<(), Box<dyn std::error::Error>>>,
+{
+    let s1 = Arc::new(mk_storage().await?);
+    let s2 = Arc::new(mk_in_memory_storage());
+    f(s1).await?;
+    f(s2).await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_snapshot_write_read() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id = SnapshotId::random();
+        let snapshot = Arc::new(Snapshot::empty());
+        storage.write_snapshot(id.clone(), snapshot.clone()).await?;
+        let back = storage.fetch_snapshot(&id).await?;
+        assert_eq!(snapshot, back);
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_manifest_write_read() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id = ManifestId::random();
+        let manifest = Arc::new(Manifest::default());
+        storage.write_manifests(id.clone(), manifest.clone()).await?;
+        let back = storage.fetch_manifests(&id).await?;
+        assert_eq!(manifest, back);
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_chunk_write_read() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id = ChunkId::random();
+        let bytes = Bytes::from_static(b"hello");
+        storage.write_chunk(id.clone(), bytes.clone()).await?;
+        let back = storage.fetch_chunk(&id, &ByteRange::ALL).await?;
+        assert_eq!(bytes, back);
+
+        let back =
+            storage.fetch_chunk(&id, &ByteRange::from_offset_with_length(1, 2)).await?;
+        assert_eq!(Bytes::from_static(b"el"), back);
+
+        let back = storage.fetch_chunk(&id, &ByteRange::from_offset(1)).await?;
+        assert_eq!(Bytes::from_static(b"ello"), back);
+
+        let back = storage.fetch_chunk(&id, &ByteRange::to_offset(3)).await?;
+        assert_eq!(Bytes::from_static(b"hel"), back); // codespell:ignore
+
+        let back = storage.fetch_chunk(&id, &ByteRange::bounded(1, 4)).await?;
+        assert_eq!(Bytes::from_static(b"ell"), back);
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_tag_write_get() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id = SnapshotId::random();
+        create_tag(storage.as_ref(), "mytag", id.clone(), false).await?;
+        let back = fetch_tag(storage.as_ref(), "mytag").await?;
+        assert_eq!(id, back.snapshot);
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_fetch_non_existing_tag() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id = SnapshotId::random();
+        create_tag(storage.as_ref(), "mytag", id.clone(), false).await?;
+
+        let back = fetch_tag(storage.as_ref(), "non-existing-tag").await;
+        assert!(matches!(back, Err(RefError::RefNotFound(r)) if r == "non-existing-tag"));
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_create_existing_tag() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id = SnapshotId::random();
+        create_tag(storage.as_ref(), "mytag", id.clone(), false).await?;
+
+        let res = create_tag(storage.as_ref(), "mytag", id.clone(), false).await;
+        assert!(matches!(res, Err(RefError::TagAlreadyExists(r)) if r == "mytag"));
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_branch_initialization() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id = SnapshotId::random();
+
+        let res = update_branch(storage.as_ref(), "some-branch", id.clone(), None, false)
+            .await?;
+        assert_eq!(res.0, 0);
+
+        let res = fetch_branch_tip(storage.as_ref(), "some-branch").await?;
+        assert_eq!(res.snapshot, id);
+
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_fetch_non_existing_branch() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id = SnapshotId::random();
+        update_branch(storage.as_ref(), "some-branch", id.clone(), None, false).await?;
+
+        let back = fetch_branch_tip(storage.as_ref(), "non-existing-branch").await;
+        assert!(
+            matches!(back, Err(RefError::RefNotFound(r)) if r == "non-existing-branch")
+        );
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_branch_update() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id1 = SnapshotId::random();
+        let id2 = SnapshotId::random();
+        let id3 = SnapshotId::random();
+
+        let res =
+            update_branch(storage.as_ref(), "some-branch", id1.clone(), None, false)
+                .await?;
+        assert_eq!(res.0, 0);
+
+        let res = update_branch(
+            storage.as_ref(),
+            "some-branch",
+            id2.clone(),
+            Some(&id1),
+            false,
+        )
+        .await?;
+        assert_eq!(res.0, 1);
+
+        let res = update_branch(
+            storage.as_ref(),
+            "some-branch",
+            id3.clone(),
+            Some(&id2),
+            false,
+        )
+        .await?;
+        assert_eq!(res.0, 2);
+
+        let res = fetch_branch_tip(storage.as_ref(), "some-branch").await?;
+        assert_eq!(res.snapshot, id3);
+
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_ref_names() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let id1 = SnapshotId::random();
+        let id2 = SnapshotId::random();
+        update_branch(storage.as_ref(), "main", id1.clone(), None, false).await?;
+        update_branch(storage.as_ref(), "main", id2.clone(), Some(&id1), false).await?;
+        update_branch(storage.as_ref(), "foo", id1.clone(), None, false).await?;
+        update_branch(storage.as_ref(), "bar", id1.clone(), None, false).await?;
+        create_tag(storage.as_ref(), "my-tag", id1.clone(), false).await?;
+        create_tag(storage.as_ref(), "my-other-tag", id1.clone(), false).await?;
+
+        let res: HashSet<_> = HashSet::from_iter(list_refs(storage.as_ref()).await?);
+        assert_eq!(
+            res,
+            HashSet::from_iter([
+                Ref::Tag("my-tag".to_string()),
+                Ref::Tag("my-other-tag".to_string()),
+                Ref::Branch("main".to_string()),
+                Ref::Branch("foo".to_string()),
+                Ref::Branch("bar".to_string()),
+            ])
+        );
+        Ok(())
+    })
+    .await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_write_config_on_empty() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let config = Bytes::copy_from_slice(b"hello");
+        let etag = storage.update_config(config.clone(), None).await?;
+        assert_ne!(etag, "");
+        let res = storage.fetch_config().await?;
+        assert!(
+            matches!(res, Some((bytes, actual_etag)) if actual_etag == etag && bytes == config )
+        );
+        Ok(())
+    }).await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_write_config_on_existing() -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let first_etag = storage.update_config(Bytes::copy_from_slice(b"hello"), None).await?;
+        let config = Bytes::copy_from_slice(b"bye");
+        let second_etag = storage.update_config(config.clone(), Some(first_etag.as_str())).await?;
+        assert_ne!(second_etag, first_etag);
+        let res = storage.fetch_config().await?;
+        assert!(
+            matches!(res, Some((bytes, actual_etag)) if actual_etag == second_etag && bytes == config )
+        );
+        Ok(())
+    }).await?;
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_write_config_fails_on_bad_etag_when_non_existing(
+) -> Result<(), Box<dyn std::error::Error>> {
+    // FIXME: this test fails in MiniIO but seems to work on S3
+    let storage = mk_in_memory_storage();
+    let etag = storage
+        .update_config(
+            Bytes::copy_from_slice(b"hello"),
+            Some("00000000000000000000000000000000"),
+        )
+        .await;
+
+    assert!(etag.is_err());
+    Ok(())
+}
+
+#[tokio::test]
+pub async fn test_write_config_fails_on_bad_etag_when_existing(
+) -> Result<(), Box<dyn std::error::Error>> {
+    with_storage(|storage| async move {
+        let config = Bytes::copy_from_slice(b"hello");
+        let etag = storage.update_config(config.clone(), None).await?;
+        let res = storage
+            .update_config(
+                Bytes::copy_from_slice(b"bye"),
+                Some("00000000000000000000000000000000"),
+            )
+            .await;
+        assert!(matches!(res, Err(StorageError::ConfigUpdateConflict)));
+
+        let res = storage.fetch_config().await?;
+        assert!(
+            matches!(res, Some((bytes, actual_etag)) if actual_etag == etag && bytes == config )
+        );
+            Ok(())
+    }).await?;
+    Ok(())
+}