ekkolon · ekkolon · Feb 27, 2026 · Feb 27, 2026
diff --git a/minikv-core/src/error.rs b/minikv-core/src/error.rs
@@ -4,6 +4,14 @@ pub enum Error {
     /// Errors from the rusty-leveldb storage engine.
     #[error("LevelDB error: {0}")]
     LevelDb(String),
+
+    /// Malformed record bytes that cannot be decoded.
+    #[error("Record decode error: {0}")]
+    RecordDecode(String),
+
+    /// A HARD-deleted record was passed to `Record::encode`, which is forbidden.
+    #[error("Cannot encode a HARD-deleted record into the database")]
+    HardDeleteEncode,
 }
 
 impl From<rusty_leveldb::Status> for Error {

diff --git a/minikv-core/src/lib.rs b/minikv-core/src/lib.rs
@@ -1,6 +1,7 @@
 pub mod error;
 pub mod hashing;
 pub mod locking;
+pub mod record;
 pub mod storage;
 pub mod volumes;
 

diff --git a/minikv-core/src/record.rs b/minikv-core/src/record.rs
@@ -0,0 +1,373 @@
+//! Metadata record stored in LevelDB.
+//!
+//! Each key in the database maps to a `Record`, which encodes the
+//! deletion state, content hash, content type, and the list of volume
+//! addresses holding the object. The wire format is byte-exact and must
+//! never change once data is written:
+//!
+//! ```text
+//! [DELETED][HASH<64 hex chars>][TYPE<mime>|]<vol1>,<vol2>,...
+//! ```
+//!
+//! The `DELETED` prefix indicates a soft-deleted object. `HASH` stores
+//! a BLAKE3-256 checksum of the object. `TYPE` is optional and terminated
+//! by a pipe (`|`), representing the MIME type. The remaining bytes are
+//! comma-separated volume addresses.
+//!
+//! The encode/decode functions are strict inverses for `No` and `Soft`
+//! records, ensuring that `Record::encode(Record::decode(bytes)) == bytes`.
+//!
+//! `Hard` deletion is represented by `Record::not_found()` and is never
+//! written to the database.
+
+use std::fmt;
+
+use crate::error::Error;
+
+/// Whether a key has been deleted.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Deleted {
+    /// Key exists and is accessible.
+    No,
+    /// Key has been soft-deleted (UNLINK). Still in DB, not on volumes.
+    Soft,
+    /// Sentinel: key was never written or has been hard-deleted from DB.
+    /// Never stored in LevelDB. Only returned by `AppState::get_record`
+    /// when a key is absent.
+    Hard,
+}
+
+impl fmt::Display for Deleted {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Deleted::No => write!(f, "no"),
+            Deleted::Soft => write!(f, "soft"),
+            Deleted::Hard => write!(f, "hard"),
+        }
+    }
+}
+
+/// Metadata record for a stored object.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Record {
+    /// Volume server addresses that hold replicas of this object.
+    /// Format: `hostname:port` or `hostname:port/svXX` for subvolumes.
+    pub volumes: Vec<String>,
+
+    /// Deletion state.
+    pub deleted: Deleted,
+
+    /// BLAKE3-256 hex digest of the object body (64 hex chars), or `None`
+    /// if checksum was disabled at write time or not yet computed.
+    pub hash: Option<String>,
+
+    /// MIME type of the stored object (e.g. `"image/jpeg"`), or `None`
+    /// if the client did not supply a `Content-Type` header on PUT.
+    ///
+    /// Stored in the wire format as `TYPE<mimetype>|`.
+    /// Used to set the `Content-Type` response header on GET/HEAD so that
+    /// nginx X-Accel-Redirect responses carry the correct type.
+    pub content_type: Option<String>,
+}
+
+impl Record {
+    /// The hash field length we expect: BLAKE3-256 = 32 bytes = 64 hex chars.
+    pub const HASH_HEX_LEN: usize = 64;
+
+    /// Construct the "hard-deleted / not found" sentinel record.
+    /// This is never written to the database.
+    pub fn not_found() -> Self {
+        Record {
+            volumes: Vec::new(),
+            deleted: Deleted::Hard,
+            hash: None,
+            content_type: None,
+        }
+    }
+
+    /// Decode a raw LevelDB value into a `Record`.
+    ///
+    /// Returns `Err(Error::RecordDecode)` on malformed input.
+    pub fn decode(bytes: &[u8]) -> Result<Self, Error> {
+        let mut s = std::str::from_utf8(bytes)
+            .map_err(|e| Error::RecordDecode(format!("non-UTF-8 record: {e}")))?;
+
+        let mut deleted = Deleted::No;
+        if let Some(rest) = s.strip_prefix("DELETED") {
+            deleted = Deleted::Soft;
+            s = rest;
+        }
+
+        let mut hash: Option<String> = None;
+        if let Some(rest) = s.strip_prefix("HASH") {
+            if rest.len() < Self::HASH_HEX_LEN {
+                return Err(Error::RecordDecode(format!(
+                    "HASH prefix present but only {} hex chars follow (expected {})",
+                    rest.len(),
+                    Self::HASH_HEX_LEN
+                )));
+            }
+            let (hex_part, remainder) = rest.split_at(Self::HASH_HEX_LEN);
+            // Validate it really is lowercase hex.
+            if !hex_part.chars().all(|c| c.is_ascii_hexdigit()) {
+                return Err(Error::RecordDecode(format!(
+                    "HASH field contains non-hex characters: {hex_part:?}"
+                )));
+            }
+            hash = Some(hex_part.to_string());
+            s = remainder;
+        }
+
+        // Optional TYPE field: TYPE<mimetype>|
+        // The pipe terminator makes parsing unambiguous without a length prefix.
+        // MIME types (e.g. "image/jpeg", "application/json") never contain '|'.
+        let mut content_type: Option<String> = None;
+        if let Some(rest) = s.strip_prefix("TYPE") {
+            match rest.find('|') {
+                Some(pipe_pos) => {
+                    let mime = &rest[..pipe_pos];
+                    if mime.is_empty() {
+                        return Err(Error::RecordDecode("TYPE field is empty".to_string()));
+                    }
+                    content_type = Some(mime.to_string());
+                    s = &rest[pipe_pos + 1..];
+                }
+                None => {
+                    return Err(Error::RecordDecode(
+                        "TYPE field missing \'|\' terminator".to_string(),
+                    ));
+                }
+            }
+        }
+
+        // Remaining bytes are comma-separated volume addresses.
+        let volumes: Vec<String> = s.split(',').map(str::to_string).collect();
+        if volumes.is_empty() || (volumes.len() == 1 && volumes[0].is_empty()) {
+            return Err(Error::RecordDecode(
+                "record has no volume entries".to_string(),
+            ));
+        }
+
+        Ok(Record {
+            volumes,
+            deleted,
+            hash,
+            content_type,
+        })
+    }
+
+    /// Encode a `Record` into its raw LevelDB byte representation.
+    ///
+    /// Returns `Err(Error::HardDeleteEncode)` if `deleted == Hard`.
+    /// Such records must never be written to the database.
+    pub fn encode(&self) -> Result<Vec<u8>, Error> {
+        if self.deleted == Deleted::Hard {
+            return Err(Error::HardDeleteEncode);
+        }
+
+        let mut s = String::new();
+
+        if self.deleted == Deleted::Soft {
+            s.push_str("DELETED");
+        }
+
+        if let Some(ref h) = self.hash {
+            // Defensive check: only write well-formed hashes.
+            if h.len() == Self::HASH_HEX_LEN && h.chars().all(|c| c.is_ascii_hexdigit()) {
+                s.push_str("HASH");
+                s.push_str(h);
+            }
+        }
+
+        s.push_str(&self.volumes.join(","));
+        Ok(s.into_bytes())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Helper: encode then decode must be a perfect round-trip.
+    fn round_trip(rec: &Record) {
+        let encoded = rec.encode().expect("encode should succeed");
+        let decoded = Record::decode(&encoded).expect("decode should succeed");
+        assert_eq!(rec, &decoded, "round-trip failed for {rec:?}");
+    }
+
+    #[test]
+    fn encode_decode_soft_deleted_multi_volume() {
+        let rec = Record {
+            volumes: vec!["hello".into(), "world".into()],
+            deleted: Deleted::Soft,
+            hash: None,
+            content_type: None,
+        };
+        let encoded = rec.encode().unwrap();
+        assert_eq!(encoded, b"DELETEDhello,world");
+        round_trip(&rec);
+    }
+
+    #[test]
+    fn encode_decode_not_deleted_multi_volume() {
+        let rec = Record {
+            volumes: vec!["hello".into(), "world".into()],
+            deleted: Deleted::No,
+            hash: None,
+            content_type: None,
+        };
+        let encoded = rec.encode().unwrap();
+        assert_eq!(encoded, b"hello,world");
+        round_trip(&rec);
+    }
+
+    #[test]
+    fn encode_decode_single_volume_no_delete() {
+        let rec = Record {
+            volumes: vec!["hello".into()],
+            deleted: Deleted::No,
+            hash: None,
+            content_type: None,
+        };
+        let encoded = rec.encode().unwrap();
+        assert_eq!(encoded, b"hello");
+        round_trip(&rec);
+    }
+
+    #[test]
+    fn encode_decode_single_volume_soft_deleted() {
+        let rec = Record {
+            volumes: vec!["hello".into()],
+            deleted: Deleted::Soft,
+            hash: None,
+            content_type: None,
+        };
+        let encoded = rec.encode().unwrap();
+        assert_eq!(encoded, b"DELETEDhello");
+        round_trip(&rec);
+    }
+
+    #[test]
+    fn encode_decode_with_blake3_hash_soft_deleted() {
+        // 64-char lowercase hex BLAKE3 digest
+        let hash = "ea8f163db38682925e4491c5e58d4bb3506ef8c14eb78a86e908c5624a67200f";
+        let rec = Record {
+            volumes: vec!["hello".into()],
+            deleted: Deleted::Soft,
+            hash: Some(hash.to_string()),
+            content_type: None,
+        };
+        let encoded = rec.encode().unwrap();
+        let expected = format!("DELETEDHASH{hash}hello");
+        assert_eq!(String::from_utf8(encoded).unwrap(), expected);
+        round_trip(&rec);
+    }
+
+    #[test]
+    fn encode_decode_with_blake3_hash_not_deleted() {
+        let hash = "ea8f163db38682925e4491c5e58d4bb3506ef8c14eb78a86e908c5624a67200f";
+        let rec = Record {
+            volumes: vec!["hello".into()],
+            deleted: Deleted::No,
+            hash: Some(hash.to_string()),
+            content_type: None,
+        };
+        let encoded = rec.encode().unwrap();
+        let expected = format!("HASH{hash}hello");
+        assert_eq!(String::from_utf8(encoded).unwrap(), expected);
+        round_trip(&rec);
+    }
+
+    #[test]
+    fn hard_delete_encode_is_error() {
+        let rec = Record::not_found();
+        assert!(rec.encode().is_err());
+    }
+
+    #[test]
+    fn decode_rejects_short_hash() {
+        // HASH prefix followed by only 10 hex chars — must fail
+        let bad = b"HASH0123456789hello";
+        assert!(Record::decode(bad).is_err());
+    }
+
+    #[test]
+    fn decode_rejects_non_hex_hash() {
+        let bad = format!(
+            "HASH{}hello",
+            "ZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ"
+        );
+        assert!(Record::decode(bad.as_bytes()).is_err());
+    }
+    #[test]
+    fn encode_decode_with_content_type() {
+        let rec = Record {
+            volumes: vec!["hello".into()],
+            deleted: Deleted::No,
+            hash: None,
+            content_type: Some("image/jpeg".to_string()),
+        };
+        let encoded = rec.encode().unwrap();
+        assert_eq!(String::from_utf8(encoded).unwrap(), "TYPEimage/jpeg|hello");
+        round_trip(&rec);
+    }
+
+    #[test]
+    fn encode_decode_with_hash_and_content_type() {
+        let hash = "ea8f163db38682925e4491c5e58d4bb3506ef8c14eb78a86e908c5624a67200f";
+        let rec = Record {
+            volumes: vec!["vol1".into(), "vol2".into()],
+            deleted: Deleted::No,
+            hash: Some(hash.to_string()),
+            content_type: Some("application/json".to_string()),
+        };
+        let encoded = rec.encode().unwrap();
+        let expected = format!("HASH{hash}TYPEapplication/json|vol1,vol2");
+        assert_eq!(String::from_utf8(encoded).unwrap(), expected);
+        round_trip(&rec);
+    }
+
+    #[test]
+    fn encode_decode_deleted_with_content_type() {
+        let rec = Record {
+            volumes: vec!["vol1".into()],
+            deleted: Deleted::Soft,
+            hash: None,
+            content_type: Some("video/mp4".to_string()),
+        };
+        let encoded = rec.encode().unwrap();
+        assert_eq!(
+            String::from_utf8(encoded).unwrap(),
+            "DELETEDTYPEvideo/mp4|vol1"
+        );
+        round_trip(&rec);
+    }
+
+    #[test]
+    fn decode_rejects_type_without_pipe_terminator() {
+        // TYPE field with no | terminator must be rejected
+        let bad = b"TYPEimage/jpegvolume1:8080";
+        // This will be interpreted as no TYPE prefix found (no | found) -> error
+        assert!(Record::decode(bad).is_err());
+    }
+
+    #[test]
+    fn decode_rejects_empty_type_field() {
+        let bad = b"TYPE|volume1:8080";
+        assert!(Record::decode(bad).is_err());
+    }
+
+    #[test]
+    fn content_type_none_round_trips_without_type_field() {
+        let rec = Record {
+            volumes: vec!["vol1".into()],
+            deleted: Deleted::No,
+            hash: None,
+            content_type: None,
+        };
+        let encoded = rec.encode().unwrap();
+        // Must not contain TYPE at all
+        assert!(!String::from_utf8_lossy(&encoded).contains("TYPE"));
+        round_trip(&rec);
+    }
+}