libsql: add per-sync-url metadata

This commit changes the filename of the metadata file to contain
`<db_path>-<sync_url>-info`. Where `db_path` is a full file path +
filename and `sync_url` is the host of the authority of the URI. This
also includes upfront uri parsing and two new errors that can be
produced when trying to extract the host.

This approach trades off creating multiple files in exchange for
allowing multiple sync context's to operate at the same time
concurrently. Originally, I had approached updating the metadata to
include a hashmap of endpoints and their metadata but this approach
falls flat if you have multiple `Database` and thus `SyncContext` in the
same process (and beyond).

This commit does NOT include updating from the original v0 metadata
format but instead will force a full re-sync and a new meatadata file
will be produced with verison set to `1`. Not implementing this mean't
simpler code with less space to produce errors and since this only runs
once when a user upgrades that cost is okay.

Closes #1837

libsql/src/sync.rs

@@ -1,18 +1,18 @@
 use crate::{util::ConnectorService, Result};
 
-use std::path::Path;
+use std::{path::Path, str::FromStr};
 
 use bytes::Bytes;
 use chrono::Utc;
-use http::{HeaderValue, StatusCode};
+use http::{uri::InvalidUri, HeaderValue, StatusCode, Uri};
 use hyper::Body;
 use tokio::io::AsyncWriteExt as _;
 use uuid::Uuid;
 
 #[cfg(test)]
 mod test;
 
-const METADATA_VERSION: u32 = 0;
+const METADATA_VERSION: u32 = 1;
 
 const DEFAULT_MAX_RETRIES: usize = 5;
 
@@ -49,6 +49,10 @@ pub enum SyncError {
     InvalidPushFrameNoHigh(u32, u32),
     #[error("failed to pull frame: status={0}, error={1}")]
     PullFrame(StatusCode, String),
+    #[error("invalid sync uri: {0}")]
+    InvalidSyncUri(InvalidUri),
+    #[error("Unable to construct metadata filename: {0}")]
+    UnableToConstructMetadataFilename(String),
 }
 
 impl SyncError {
@@ -60,7 +64,7 @@ impl SyncError {
 pub struct SyncContext {
     db_path: String,
     client: hyper::Client<ConnectorService, Body>,
-    sync_url: String,
+    sync_url: Uri,
     auth_token: Option<HeaderValue>,
     max_retries: usize,
     /// Represents the max_frame_no from the server.
@@ -86,6 +90,8 @@ impl SyncContext {
             None => None,
         };
 
+        let sync_url = Uri::from_str(&sync_url).map_err(SyncError::InvalidSyncUri)?;
+
         let mut me = Self {
             db_path,
             sync_url,
@@ -107,7 +113,11 @@ impl SyncContext {
     }
 
     #[tracing::instrument(skip(self))]
-    pub(crate) async fn pull_one_frame(&mut self, generation: u32, frame_no: u32) -> Result<Option<Bytes>> {
+    pub(crate) async fn pull_one_frame(
+        &mut self,
+        generation: u32,
+        frame_no: u32,
+    ) -> Result<Option<Bytes>> {
         let uri = format!(
             "{}/sync/{}/{}/{}",
             self.sync_url,
@@ -294,7 +304,7 @@ impl SyncContext {
     }
 
     pub(crate) async fn write_metadata(&mut self) -> Result<()> {
-        let path = format!("{}-info", self.db_path);
+        let path = self.sync_metadata_filename()?;
 
         let mut metadata = MetadataJson {
             hash: 0,
@@ -313,9 +323,12 @@ impl SyncContext {
     }
 
     async fn read_metadata(&mut self) -> Result<()> {
-        let path = format!("{}-info", self.db_path);
+        let path = self.sync_metadata_filename()?;
 
-        if !Path::new(&path).try_exists().map_err(SyncError::io("metadata file exists"))? {
+        if !Path::new(&path)
+            .try_exists()
+            .map_err(SyncError::io("metadata file exists"))?
+        {
             tracing::debug!("no metadata info file found");
             return Ok(());
         }
@@ -344,6 +357,16 @@ impl SyncContext {
 
         Ok(())
     }
+
+    fn sync_metadata_filename(&self) -> Result<String> {
+        let authority = self.sync_url.authority().ok_or_else(|| {
+            SyncError::UnableToConstructMetadataFilename("no authority set".into())
+        })?;
+
+        let host = authority.host();
+
+        Ok(format!("{}-{}-info", self.db_path, host))
+    }
 }
 
 #[derive(serde::Serialize, serde::Deserialize, Debug)]

libsql/src/sync/test.rs

@@ -113,8 +113,8 @@ async fn test_sync_context_corrupted_metadata() {
     assert_eq!(durable_frame, 0);
     assert_eq!(server.frame_count(), 1);
 
-    // Update metadata path to use -info instead of .meta
-    let metadata_path = format!("{}-info", db_path.to_str().unwrap());
+    // Inject invalid data into the metadata file to force a recovery
+    let metadata_path = sync_ctx.sync_metadata_filename().unwrap();
     std::fs::write(&metadata_path, b"invalid json data").unwrap();
 
     // Create new sync context with corrupted metadata