Support ZFS encryption key rotation for Trust Quorum epoch commits

When Trust Quorum commits a new epoch, all U.2 crypt datasets must have
their encryption keys rotated to use the new epoch's derived key. This
change implements the key rotation flow triggered by epoch commits.

## Trust Quorum Integration

- Add watch channel to `NodeTaskHandle` for epoch change notifications
- Initialize channel with current committed epoch on startup
- Notify subscribers via `send_if_modified()` when epoch changes

## Config Reconciler Integration

- Accept `committed_epoch_rx` watch channel from trust quorum
- Trigger reconciliation when epoch changes
- Track per-disk encryption epoch in `ExternalDisks`
- Add `rekey_for_epoch()` to coordinate key rotation:
  - Filter disks needing rekey (cached epoch < target OR unknown)
  - Derive keys for each disk via `StorageKeyRequester`
  - Send batch request to dataset task
  - Update cached epochs on success
  - Retry on failure via normal reconciliation retry logic

## Dataset Task Changes

- Add `RekeyRequest`/`RekeyResult` types for batch rekey operations
- Add `datasets_rekey()` with idempotency check (skip if already at target)
- Use `Zfs::change_key()` for atomic key + epoch property update

## ZFS Utilities

- Add `Zfs::change_key()` using `zfs_atomic_change_key` crate
- Add `Zfs::load_key()`, `unload_key()`, `dataset_exists()`
- Add `epoch` field to `DatasetProperties`
- Add structured error types for key operations

## Crash Recovery

- Add trial decryption recovery in `sled-storage` for datasets with
  missing epoch property (e.g., crash during initial creation)
- Unload key before each trial attempt to handle crash-after-load-key
- Set epoch property after successful recovery

## Safety Properties

- Atomic: Key and epoch property set together via `zfs_atomic_change_key`
- Idempotent: Skip rekey if dataset already at target epoch
- Crash-safe: Epoch read from ZFS on restart rebuilds cache correctly
- Conservative: Unknown epochs (None) trigger rekey attempt

Author: plaidfinch

Cargo.lock

@@ -829,6 +829,7 @@ wicketd-client = { path = "clients/wicketd-client" }
 xshell = "0.2.7"
 zerocopy = "0.8.26"
 zeroize = { version = "1.8.1", features = ["zeroize_derive", "std"] }
+zfs-atomic-change-key = { git = "https://github.com/oxidecomputer/zfs-atomic-change-key" }
 zfs-test-harness = { path = "sled-storage/zfs-test-harness" }
 zip = { version = "4.2.0", default-features = false, features = ["deflate","bzip2"] }
 zone = { version = "0.3.1", default-features = false, features = ["async"] }

Cargo.toml

@@ -24,6 +24,7 @@ futures.workspace = true
 http.workspace = true
 ipnetwork.workspace = true
 itertools.workspace = true
+key-manager.workspace = true
 libc.workspace = true
 macaddr.workspace = true
 nix.workspace = true
@@ -44,6 +45,7 @@ tokio.workspace = true
 uuid.workspace = true
 whoami.workspace = true
 zone.workspace = true
+zfs-atomic-change-key.workspace = true
 tofino.workspace = true
 rustix.workspace = true
 

illumos-utils/Cargo.toml

@@ -206,6 +206,42 @@ pub struct GetValueError {
 #[error("Failed to list snapshots: {0}")]
 pub struct ListSnapshotsError(#[from] crate::ExecutionError);
 
+/// Error returned by [`Zfs::change_key`].
+#[derive(thiserror::Error, Debug)]
+#[error("Failed to change encryption key for dataset '{name}'")]
+pub struct ChangeKeyError {
+    pub name: String,
+    #[source]
+    pub err: anyhow::Error,
+}
+
+/// Error returned by [`Zfs::load_key`].
+#[derive(thiserror::Error, Debug)]
+#[error("Failed to load encryption key for dataset '{name}'")]
+pub struct LoadKeyError {
+    pub name: String,
+    #[source]
+    pub err: crate::ExecutionError,
+}
+
+/// Error returned by [`Zfs::dataset_exists`].
+#[derive(thiserror::Error, Debug)]
+#[error("Failed to check if dataset '{name}' exists")]
+pub struct DatasetExistsError {
+    pub name: String,
+    #[source]
+    pub err: crate::ExecutionError,
+}
+
+/// Error returned by [`Zfs::unload_key`].
+#[derive(thiserror::Error, Debug)]
+#[error("Failed to unload encryption key for dataset '{name}'")]
+pub struct UnloadKeyError {
+    pub name: String,
+    #[source]
+    pub err: crate::ExecutionError,
+}
+
 #[derive(Debug, thiserror::Error)]
 #[error(
     "Failed to create snapshot '{snap_name}' from filesystem '{filesystem}': {err}"
@@ -523,11 +559,14 @@ pub struct DatasetProperties {
     /// string so that unexpected compression formats don't prevent inventory
     /// from being collected.
     pub compression: String,
+    /// The encryption key epoch for this dataset.
+    ///
+    /// Only present on encrypted datasets (e.g., crypt datasets on U.2s).
+    pub epoch: Option<u64>,
 }
 
 impl DatasetProperties {
-    const ZFS_GET_PROPS: &'static str =
-        "oxide:uuid,name,mounted,avail,used,quota,reservation,compression";
+    const ZFS_GET_PROPS: &'static str = "oxide:uuid,oxide:epoch,name,mounted,avail,used,quota,reservation,compression";
 }
 
 impl TryFrom<&DatasetProperties> for SharedDatasetConfig {
@@ -648,6 +687,18 @@ impl DatasetProperties {
                     .get("compression")
                     .map(|(prop, _source)| prop.to_string())
                     .ok_or_else(|| anyhow!("Missing 'compression'"))?;
+                // The epoch property is only present on encrypted datasets.
+                // Like oxide:uuid, we ignore inherited values.
+                let epoch = props
+                    .get("oxide:epoch")
+                    .filter(|(prop, source)| {
+                        !source.starts_with("inherited") && *prop != "-"
+                    })
+                    .map(|(prop, _source)| {
+                        prop.parse::<u64>()
+                            .context("Failed to parse 'oxide:epoch'")
+                    })
+                    .transpose()?;
 
                 Ok(DatasetProperties {
                     id,
@@ -658,6 +709,7 @@ impl DatasetProperties {
                     quota,
                     reservation,
                     compression,
+                    epoch,
                 })
             })
             .collect::<Result<Vec<_>, _>>()
@@ -1197,7 +1249,7 @@ impl Zfs {
         name: &str,
         mountpoint: &Mountpoint,
     ) -> Result<(), EnsureDatasetErrorRaw> {
-        let mount_info = Self::dataset_exists(name, mountpoint).await?;
+        let mount_info = Self::dataset_mount_info(name, mountpoint).await?;
         if !mount_info.exists {
             return Err(EnsureDatasetErrorRaw::DoesNotExist);
         }
@@ -1246,7 +1298,7 @@ impl Zfs {
             additional_options,
         }: DatasetEnsureArgs<'_>,
     ) -> Result<(), EnsureDatasetErrorRaw> {
-        let dataset_info = Self::dataset_exists(name, &mountpoint).await?;
+        let dataset_info = Self::dataset_mount_info(name, &mountpoint).await?;
 
         // Non-zoned datasets with an explicit mountpoint and the
         // "canmount=on" property should be mounted within the global zone.
@@ -1365,9 +1417,29 @@ impl Zfs {
         Ok(())
     }
 
-    // Return (true, mounted) if the dataset exists, (false, false) otherwise,
-    // where mounted is if the dataset is mounted.
-    async fn dataset_exists(
+    /// Check if a ZFS dataset exists.
+    pub async fn dataset_exists(
+        name: &str,
+    ) -> Result<bool, DatasetExistsError> {
+        let mut cmd = Command::new(ZFS);
+        cmd.args(&["list", "-H", name]);
+        match execute_async(&mut cmd).await {
+            Ok(_) => Ok(true),
+            Err(crate::ExecutionError::CommandFailure(ref info))
+                if info.stderr.contains("does not exist") =>
+            {
+                Ok(false)
+            }
+            Err(err) => Err(DatasetExistsError { name: name.to_string(), err }),
+        }
+    }
+
+    /// Get mount info for a dataset, validating its mountpoint.
+    ///
+    /// Returns (exists=true, mounted) if the dataset exists with the expected
+    /// mountpoint, (exists=false, mounted=false) if it doesn't exist.
+    /// Returns an error if the dataset exists but has an unexpected mountpoint.
+    async fn dataset_mount_info(
         name: &str,
         mountpoint: &Mountpoint,
     ) -> Result<DatasetMountInfo, EnsureDatasetErrorRaw> {
@@ -1523,6 +1595,62 @@ impl Zfs {
         })
     }
 
+    /// Atomically change the encryption key and set the oxide:epoch property.
+    ///
+    /// This operation is used for ZFS key rotation when a new Trust Quorum
+    /// epoch is committed.
+    pub async fn change_key(
+        dataset: &str,
+        key: &key_manager::VersionedAes256GcmDiskEncryptionKey,
+    ) -> Result<(), ChangeKeyError> {
+        // FIXME: Replace the use of `zfs_atomic_change_key` with a native
+        // invocation of `zfs change-key` using the `-o oxide:epoch` option to
+        // set the epoch. At time of writing, the `zfs change-key` command does
+        // not support setting user properties inline, but a patch is pending to
+        // add this feature.
+
+        let ds = zfs_atomic_change_key::Dataset::new(dataset).map_err(|e| {
+            ChangeKeyError {
+                name: dataset.to_string(),
+                err: anyhow::anyhow!("invalid dataset name: {e}"),
+            }
+        })?;
+
+        ds.change_key(zfs_atomic_change_key::Key::hex(*key.expose_secret()))
+            .property("oxide:epoch", key.epoch().to_string())
+            .await
+            .map_err(|e| ChangeKeyError {
+                name: dataset.to_string(),
+                err: anyhow::anyhow!("{e}"),
+            })
+    }
+
+    /// Load the encryption key for an encrypted ZFS dataset.
+    ///
+    /// This makes the dataset accessible for mounting. The key must have
+    /// previously been written to the dataset's keylocation.
+    pub async fn load_key(name: &str) -> Result<(), LoadKeyError> {
+        let mut cmd = Command::new(PFEXEC);
+        cmd.args(&[ZFS, "load-key", name]);
+        execute_async(&mut cmd)
+            .await
+            .map(|_| ())
+            .map_err(|err| LoadKeyError { name: name.to_string(), err })
+    }
+
+    /// Unload the encryption key for an encrypted ZFS dataset.
+    ///
+    /// This is used for cleanup after failed key operations or during
+    /// trial decryption recovery. The dataset must not be mounted.
+    pub async fn unload_key(name: &str) -> Result<(), UnloadKeyError> {
+        let mut cmd = Command::new(PFEXEC);
+        cmd.args(&[ZFS, "unload-key", name]);
+        execute_async(&mut cmd)
+            .await
+            .map(|_| ())
+            .map_err(|err| UnloadKeyError { name: name.to_string(), err })
+    }
+
     /// Calls "zfs get" to acquire multiple values
     ///
     /// - `names`: The properties being acquired

illumos-utils/src/zfs.rs

@@ -53,11 +53,12 @@ pub enum Error {
 }
 
 /// Derived Disk Encryption key
-#[derive(Default)]
+#[derive(Debug, Default)]
 struct Aes256GcmDiskEncryptionKey(SecretBox<[u8; 32]>);
 
 /// A Disk encryption key for a given epoch to be used with ZFS datasets for
 /// U.2 devices
+#[derive(Debug)]
 pub struct VersionedAes256GcmDiskEncryptionKey {
     epoch: u64,
     key: Aes256GcmDiskEncryptionKey,

key-manager/src/lib.rs

@@ -31,6 +31,7 @@ omicron-common.workspace = true
 omicron-uuid-kinds.workspace = true
 rand.workspace = true
 regex.workspace = true
+secrecy.workspace = true
 serde.workspace = true
 sha2.workspace = true
 sled-agent-api.workspace = true
@@ -42,6 +43,7 @@ slog-error-chain.workspace = true
 strum.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
+trust-quorum-types.workspace = true
 tufaceous-artifact.workspace = true
 zone.workspace = true