libsql: Fix metadata handling (#2149)

Fix metadata handling to ensure that we don't have inconsistent metadata
locally.

Author: penberg

libsql/src/sync.rs

@@ -68,6 +68,8 @@ pub enum SyncError {
     InvalidLocalGeneration(u32, u32),
     #[error("invalid local state: {0}")]
     InvalidLocalState(String),
+    #[error("invalid remote state: {0}")]
+    InvalidRemoteState(String),
     #[error("server returned invalid length of frames: {0}")]
     InvalidPullFrameBytes(usize),
 }
@@ -169,14 +171,10 @@ impl SyncContext {
             initial_server_sync: false,
             remote_encryption,
         };
-
-        if let Err(e) = me.read_metadata().await {
-            tracing::error!(
-                "failed to read sync metadata file, resetting back to defaults: {}",
-                e
-            );
+        me.read_metadata().await?;
+        if me.durable_generation == 0 {
+            return Err(SyncError::InvalidLocalState("generation is 0".to_string()).into());
         }
-
         Ok(me)
     }
 
@@ -529,6 +527,8 @@ impl SyncContext {
     pub(crate) async fn write_metadata(&mut self) -> Result<()> {
         let path = format!("{}-info", self.db_path);
 
+        assert!(self.durable_generation > 0);
+
         let mut metadata = MetadataJson {
             hash: 0,
             version: METADATA_VERSION,
@@ -616,8 +616,10 @@ impl SyncContext {
             .await
             .map_err(SyncError::HttpBody)?;
 
-        let info = serde_json::from_slice(&body).map_err(SyncError::JsonDecode)?;
-
+        let info: InfoResult = serde_json::from_slice(&body).map_err(SyncError::JsonDecode)?;
+        if info.current_generation == 0 {
+            return Err(SyncError::InvalidRemoteState("generation is 0".to_string()).into());
+        }
         Ok(info)
     }
 

libsql/src/sync/test.rs

@@ -14,6 +14,7 @@ async fn test_sync_context_push_frame() {
     let server = MockServer::start();
     let temp_dir = tempdir().unwrap();
     let db_path = temp_dir.path().join("test.db");
+    gen_metadata_file(&db_path, 3278479626, 0, 0, 1);
 
     let sync_ctx = SyncContext::new(
         server.connector(),
@@ -44,6 +45,7 @@ async fn test_sync_context_with_auth() {
     let server = MockServer::start();
     let temp_dir = tempdir().unwrap();
     let db_path = temp_dir.path().join("test.db");
+    gen_metadata_file(&db_path, 3278479626, 0, 0, 1);
 
     let sync_ctx = SyncContext::new(
         server.connector(),
@@ -69,6 +71,7 @@ async fn test_sync_context_multiple_frames() {
     let server = MockServer::start();
     let temp_dir = tempdir().unwrap();
     let db_path = temp_dir.path().join("test.db");
+    gen_metadata_file(&db_path, 3278479626, 0, 0, 1);
 
     let sync_ctx = SyncContext::new(
         server.connector(),
@@ -98,6 +101,7 @@ async fn test_sync_context_corrupted_metadata() {
     let server = MockServer::start();
     let temp_dir = tempdir().unwrap();
     let db_path = temp_dir.path().join("test.db");
+    gen_metadata_file(&db_path, 3278479626, 0, 0, 1);
 
     // Create initial sync context and push a frame
     let sync_ctx = SyncContext::new(
@@ -129,12 +133,9 @@ async fn test_sync_context_corrupted_metadata() {
         None,
         None,
     )
-    .await
-    .unwrap();
+    .await;
 
-    // Verify that the context was reset to default values
-    assert_eq!(sync_ctx.durable_frame_num(), 0);
-    assert_eq!(sync_ctx.durable_generation(), 0);
+    assert!(sync_ctx.is_err());
 }
 
 #[tokio::test]
@@ -144,6 +145,7 @@ async fn test_sync_restarts_with_lower_max_frame_no() {
     let server = MockServer::start();
     let temp_dir = tempdir().unwrap();
     let db_path = temp_dir.path().join("test.db");
+    gen_metadata_file(&db_path, 3278479626, 0, 0, 1);
 
     // Create initial sync context and push a frame
     let sync_ctx = SyncContext::new(
@@ -211,6 +213,7 @@ async fn test_sync_context_retry_on_error() {
     let server = MockServer::start();
     let temp_dir = tempdir().unwrap();
     let db_path = temp_dir.path().join("test.db");
+    gen_metadata_file(&db_path, 3278479626, 0, 0, 1);
 
     let sync_ctx = SyncContext::new(
         server.connector(),
@@ -475,3 +478,15 @@ impl hyper::client::connect::Connection for MockConnection {
         hyper::client::connect::Connected::new()
     }
 }
+
+fn gen_metadata_file(db_path: &Path, hash: u32, version: u32, durable_frame_num: u32, generation: u32) {
+    let metadata_path = format!("{}-info", db_path.to_str().unwrap());
+    std::fs::write(
+        &metadata_path,
+        format!(
+            "{{\"hash\": {hash}, \"version\": {version}, \"durable_frame_num\": {durable_frame_num}, \"generation\": {generation}}}"
+        )
+        .as_bytes(),
+    )
+    .unwrap();
+}