Task iceberg file

app/crowdb-access-server/Cargo.toml

@@ -20,6 +20,7 @@ s3 = [
 ]
 
 [dependencies]
+arc-swap = "1.9"
 hex = "0.4"
 hmac = "0.12"
 async-trait = "0.1"
@@ -59,7 +60,6 @@ bincode = "1"
 crowdb-access-server = { path = ".", default-features = false, features = ["test-util"] }
 futures = "0.3"
 crowdb-console-shared = { path = "../../lib/crowdb-console-shared" }
-arc-swap = "1.9"
 async-trait = "0.1"
 crowdb-chunkdb-client = { path = "../../lib/crowdb-chunkdb-client" }
 crowdb-diskdb-client = { path = "../../lib/crowdb-diskdb-client" }

app/crowdb-access-server/src/config.rs

@@ -206,6 +206,7 @@ pub struct S3Config {
 #[derive(Clone, Debug, Default, Deserialize, Serialize)]
 #[serde(default)]
 pub struct IcebergConfig {
+    pub locator_signing: Option<IcebergLocatorSigningConfig>,
     /// Overrides the legacy shared small-write policy for Iceberg only.
     pub small_write: Option<SmallWriteConfig>,
     pub listen: Option<String>,
@@ -223,6 +224,22 @@ pub struct IcebergConfig {
     pub gc: IcebergGcConfig,
 }
 
+#[derive(Clone, Deserialize, Serialize)]
+pub struct IcebergLocatorSigningConfig {
+    pub active_version: u32,
+    pub keys: std::collections::BTreeMap<u32, [u8; 32]>,
+}
+
+impl std::fmt::Debug for IcebergLocatorSigningConfig {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("IcebergLocatorSigningConfig")
+            .field("active_version", &self.active_version)
+            .field("key_versions", &self.keys.keys().collect::<Vec<_>>())
+            .finish()
+    }
+}
+
 /// Dataset's listener and read-admission limits. Dataset is optional so an
 /// existing S3/Iceberg deployment keeps its current startup behavior.
 #[derive(Clone, Debug, Deserialize, Serialize)]

app/crowdb-access-server/src/iceberg.rs

@@ -5,13 +5,17 @@ mod connection;
 mod file_admission;
 mod file_auth;
 mod file_body;
+mod file_bucket_request;
 mod file_complete;
 mod file_http;
 mod file_list_request;
+pub use file_bucket_request::{parse_file_bucket, FileBucketRequest};
 mod file_list_response;
+mod file_read_conditions;
 mod file_recovery;
 mod file_request;
 pub use file_list_request::parse_file_list;
+pub use file_read_conditions::{file_read_condition, FileReadCondition};
 mod file_response;
 mod file_selection;
 mod file_upload;

app/crowdb-access-server/src/iceberg/file_body.rs

@@ -108,6 +108,13 @@ pub struct FileReadBody {
 }
 
 impl FileReadBody {
+    /// # Errors
+    /// Rejects preflight bytes for a stream or a different file owner/root.
+    pub fn with_root_directory(mut self, bytes: Vec<u8>) -> Result<Self, FileIoError> {
+        let reader = self.reader.take().ok_or(FileIoError::Bounds)?;
+        self.reader = Some(reader.with_root_directory(bytes)?);
+        Ok(self)
+    }
     fn finish(&mut self) {
         self.reader = None;
         self.pending = None;

app/crowdb-access-server/src/iceberg/file_bucket_request.rs

@@ -0,0 +1,43 @@
+use hyper::{Method, Uri};
+
+use super::file_request::{decode, query, FileRequestError};
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum FileBucketRequest {
+    Head(String),
+    Location(String),
+}
+
+/// # Errors
+/// Rejects unsupported bucket selectors and malformed transport encodings.
+pub fn parse_file_bucket(method: &Method, uri: &Uri) -> Result<Option<FileBucketRequest>, FileRequestError> {
+    if uri
+        .path_and_query()
+        .map_or(true, |value| value.as_str().len() > 8192)
+    {
+        return Err(FileRequestError::Invalid);
+    }
+    if uri.path().trim_end_matches('/').split('/').count() != 2 {
+        return Ok(None);
+    }
+    let path = decode(uri.path())?;
+    let bucket = path.strip_prefix('/').ok_or(FileRequestError::Invalid)?;
+    let bucket = bucket.strip_suffix('/').unwrap_or(bucket);
+    if bucket.is_empty() || bucket.contains('/') {
+        return Ok(None);
+    }
+    let mut fields = query(uri.query())?;
+    if method == Method::HEAD {
+        if !fields.is_empty() {
+            return Err(FileRequestError::Unsupported);
+        }
+        return Ok(Some(FileBucketRequest::Head(bucket.to_owned())));
+    }
+    if method == Method::GET && fields.remove("location").as_deref() == Some("") {
+        if !fields.is_empty() {
+            return Err(FileRequestError::Unsupported);
+        }
+        return Ok(Some(FileBucketRequest::Location(bucket.to_owned())));
+    }
+    Ok(None)
+}