feat: block-level encryption at rest

Cargo.toml

@@ -20,6 +20,7 @@ path = "src/lib.rs"
 default = []
 lz4 = ["dep:lz4_flex"]
 zstd = ["dep:zstd"]
+encryption = ["dep:aes-gcm"]
 bytes_1 = ["dep:bytes"]
 metrics = []
 
@@ -40,6 +41,7 @@ sfa = "~1.0.0"
 tempfile = "3.20.0"
 varint-rs = "2.2.0"
 xxhash-rust = { version = "0.8.15", features = ["xxh3"] }
+aes-gcm = { version = "0.10", optional = true, features = ["aes"] }
 
 [dev-dependencies]
 criterion = { version = "0.8.0", features = ["html_reports"] }

src/blob_tree/ingest.rs

@@ -230,6 +230,7 @@ impl<'a> BlobIngestion<'a> {
                     index.config.descriptor_table.clone(),
                     false,
                     false,
+                    index.config.encryption.clone(),
                     index.config.comparator.clone(),
                     #[cfg(feature = "metrics")]
                     index.metrics.clone(),

src/blob_tree/mod.rs

@@ -446,6 +446,7 @@ impl AbstractTree for BlobTree {
 
         table_writer =
             table_writer.use_prefix_extractor(self.index.config.prefix_extractor.clone());
+        table_writer = table_writer.use_encryption(self.index.config.encryption.clone());
 
         #[expect(
             clippy::expect_used,
@@ -532,6 +533,7 @@ impl AbstractTree for BlobTree {
                     self.index.config.descriptor_table.clone(),
                     pin_filter,
                     pin_index,
+                    self.index.config.encryption.clone(),
                     self.index.config.comparator.clone(),
                     #[cfg(feature = "metrics")]
                     self.index.metrics.clone(),

src/compaction/flavour.rs

@@ -105,6 +105,7 @@ pub(super) fn prepare_table_writer(
         // filter writer (preserving the extractor). Only use_partitioned_filter
         // replaces the writer entirely (handled above, lines 85-90).
         .use_prefix_extractor(opts.config.prefix_extractor.clone())
+        .use_encryption(opts.config.encryption.clone())
         .use_bloom_policy({
             use crate::config::FilterPolicyEntry::{Bloom, None};
             use crate::table::filter::BloomConstructionPolicy;
@@ -376,6 +377,7 @@ impl StandardCompaction {
                     opts.config.descriptor_table.clone(),
                     pin_filter,
                     pin_index,
+                    opts.config.encryption.clone(),
                     opts.config.comparator.clone(),
                     #[cfg(feature = "metrics")]
                     opts.metrics.clone(),

src/config/mod.rs

@@ -22,6 +22,7 @@ pub type PartitioningPolicy = PinningPolicy;
 use crate::{
     compaction::filter::Factory,
     comparator::{self, SharedComparator},
+    encryption::EncryptionProvider,
     merge_operator::MergeOperator,
     path::absolute_path,
     prefix::PrefixExtractor,
@@ -260,11 +261,21 @@ pub struct Config {
     #[doc(hidden)]
     pub(crate) comparator: SharedComparator,
 
-    /// The global sequence number generator
+    /// Block-level encryption provider for encryption at rest.
     ///
-    /// Should be shared between multiple trees of a database
+    /// When set, all blocks (data, index, filter, meta) are encrypted
+    /// using this provider after compression and before checksumming.
+    pub(crate) encryption: Option<Arc<dyn EncryptionProvider>>,
+
+    /// The global sequence number generator.
+    ///
+    /// Should be shared between multiple trees of a database.
     pub(crate) seqno: SharedSequenceNumberGenerator,
 
+    /// Sequence number watermark that is visible to readers.
+    ///
+    /// Used for MVCC snapshots and to control which updates are
+    /// observable in a given view of the database.
     pub(crate) visible_seqno: SharedSequenceNumberGenerator,
 }
 
@@ -327,6 +338,7 @@ impl Default for Config {
             kv_separation_opts: None,
 
             comparator: comparator::default_comparator(),
+            encryption: None,
         }
     }
 }
@@ -559,6 +571,29 @@ impl Config {
         self
     }
 
+    /// Sets the block-level encryption provider for encryption at rest.
+    ///
+    /// When set, all blocks written to SST files are encrypted after
+    /// compression and before checksumming, using the provided
+    /// [`EncryptionProvider`].
+    ///
+    /// The caller is responsible for key management and rotation.
+    /// See [`crate::Aes256GcmProvider`] (behind the `encryption` feature)
+    /// for a ready-to-use AES-256-GCM implementation.
+    ///
+    /// **Important constraints:**
+    /// - Encryption state is NOT recorded in SST metadata. Opening an
+    ///   encrypted tree without the correct provider (or vice versa) will
+    ///   cause block validation errors, not silent corruption.
+    /// - Blob files (KV-separated large values) are NOT covered by
+    ///   block-level encryption. Large values stored via KV separation
+    ///   remain in plaintext on disk.
+    #[must_use]
+    pub fn with_encryption(mut self, encryption: Option<Arc<dyn EncryptionProvider>>) -> Self {
+        self.encryption = encryption;
+        self
+    }
+
     /// Opens a tree using the config.
     ///
     /// # Errors