Move metadata serialization to repository

Author: pcholakov

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

@@ -18,12 +18,16 @@ use aws_config::BehaviorVersion;
 use aws_credential_types::provider::ProvideCredentials;
 use object_store::aws::AmazonS3Builder;
 use object_store::{MultipartUpload, ObjectStore, PutPayload};
+use serde::{Deserialize, Serialize};
+use serde_with::serde_as;
 use tokio::io::AsyncReadExt;
 use tracing::{debug, trace};
 use url::Url;
 
-use restate_partition_store::snapshots::PartitionSnapshotMetadata;
+use restate_partition_store::snapshots::{PartitionSnapshotMetadata, SnapshotFormatVersion};
 use restate_types::config::SnapshotsOptions;
+use restate_types::identifiers::{PartitionId, SnapshotId};
+use restate_types::logs::Lsn;
 
 /// Provides read and write access to the long-term partition snapshot storage destination.
 ///
@@ -155,12 +159,11 @@ impl SnapshotRepository {
             );
         }
 
-        // todo(pavel): don't write `metadata.json` to disk, serialize it from the struct directly.
-        //  this gives us a chance to include data file checksums into it removes some IO
-        let metadata_json_path = local_snapshot_path.join("metadata.json");
         let metadata_key =
             object_store::path::Path::from(format!("{}/metadata.json", snapshot_prefix.as_str()));
-        let metadata_json_payload = PutPayload::from(tokio::fs::read(metadata_json_path).await?);
+        let metadata_json_payload = PutPayload::from(
+            serde_json::to_string_pretty(snapshot).expect("Can always serialize JSON"),
+        );
         let put_result = self
             .object_store
             .put(&metadata_key, metadata_json_payload)

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

@@ -62,7 +62,7 @@ impl SnapshotPartitionTask {
             )
             .await?;
 
-        let metadata = self.write_snapshot_metadata_header(&snapshot).await?;
+        let metadata = self.metadata(&snapshot, SystemTime::now());
 
         self.snapshot_repository
             .put(&metadata, snapshot.base_dir)
@@ -72,36 +72,22 @@ impl SnapshotPartitionTask {
         Ok(metadata)
     }
 
-    async fn write_snapshot_metadata_header(
+    fn metadata(
         &self,
         snapshot: &LocalPartitionSnapshot,
-    ) -> Result<PartitionSnapshotMetadata, SnapshotError> {
-        let snapshot_meta = PartitionSnapshotMetadata {
+        created_at: SystemTime,
+    ) -> PartitionSnapshotMetadata {
+        PartitionSnapshotMetadata {
             version: SnapshotFormatVersion::V1,
             cluster_name: self.cluster_name.clone(),
             node_name: self.node_name.clone(),
             partition_id: self.partition_id,
-            created_at: humantime::Timestamp::from(SystemTime::now()),
+            created_at: humantime::Timestamp::from(created_at),
             snapshot_id: self.snapshot_id,
             key_range: snapshot.key_range.clone(),
             min_applied_lsn: snapshot.min_applied_lsn,
             db_comparator_name: snapshot.db_comparator_name.clone(),
             files: snapshot.files.clone(),
-        };
-        let metadata_json =
-            serde_json::to_string_pretty(&snapshot_meta).expect("Can always serialize JSON");
-
-        let metadata_path = snapshot.base_dir.join("metadata.json");
-        tokio::fs::write(metadata_path.clone(), metadata_json)
-            .await
-            .map_err(|e| SnapshotError::SnapshotIoError(self.partition_id, e))?;
-
-        debug!(
-            lsn = %snapshot.min_applied_lsn,
-            "Partition snapshot metadata written to {:?}",
-            metadata_path
-        );
-
-        Ok(snapshot_meta)
+        }
     }
 }