add a dumb read-your-writes repro test for local filesystem provider

crates/worker/src/partition/snapshots/repository.rs

@@ -732,10 +732,13 @@ impl object_store::CredentialProvider for AwsSdkCredentialsProvider {
 
 #[cfg(test)]
 mod tests {
-    use bytes::Bytes;
-    use object_store::path::Path as ObjectPath;
-    use object_store::ObjectStore;
+    use std::sync::Arc;
     use std::time::SystemTime;
+
+    use bytes::Bytes;
+    use object_store::local::LocalFileSystem;
+    use object_store::path::{Path as ObjectPath, Path};
+    use object_store::{ObjectStore, PutPayload};
     use tempfile::TempDir;
     use tokio::io::AsyncWriteExt;
     use tracing::info;
@@ -813,6 +816,35 @@ mod tests {
         .await
     }
 
+    #[tokio::test]
+    async fn test_put_and_get_filesystem() -> anyhow::Result<()> {
+        let temporary = TempDir::new()?;
+
+        let store = Arc::new(LocalFileSystem::new_with_prefix(temporary.path())?);
+
+        let mut counter = 1_000_000;
+        loop {
+            let data = format!("snapshot-data-{}", counter);
+
+            let path = Path::from(format!("obj-{}", counter));
+            store.put(&path, PutPayload::from(data.clone())).await?;
+            let read_back = store.get(&path).await;
+
+            assert_eq!(data.as_bytes(), read_back?.bytes().await?.as_ref());
+
+            counter -= 1;
+            if counter == 0 {
+                break;
+            }
+            if counter % 1_000 == 0 {
+                eprint!(".");
+            }
+        }
+        eprintln!();
+
+        Ok(())
+    }
+
     /// For this test to run, set RESTATE_S3_INTEGRATION_TEST_BUCKET_NAME to a writable S3 bucket name
     #[tokio::test]
     async fn test_put_snapshot_s3() -> anyhow::Result<()> {