feat(rs): surface rotation drain signal via decrypt_indexed (LAB-1678)

README.md

@@ -227,6 +227,8 @@ let cache = CacheKit::builder()
 
 Rotation is forward-only: a retired key is never re-promoted (re-promoting would resume a used AES-GCM nonce budget), and a config listing the current key among the previous keys is rejected at load.
 
+**Knowing when to drop the old key.** Every read served by a previous key is counted against that key's position; `cache.secure()?.previous_key_hits()` returns the counts (`hits[i]` for `previous_keys[i]`, current-key reads not counted, no key material). The signal confirms a grace window has drained; it does not shorten one. Follow the protocol's [scheduled-rotation runbook](https://github.com/cachekit-io/protocol/blob/main/decisions/key-rotation.md#runbooks-normative-for-docs): audit for non-expiring entries, add the incoming key as decrypt-only fleet-wide, then promote it. The clock starts only when the promotion deploy has completed on every instance — a lagging instance still writes under the retiring key and reads it silently as *its* current key. From then, wait at least the longest TTL in use (including any explicit `set_with_ttl` values), aggregating counts across every instance (they are per process and reset on restart). Once the retiring key's count has stayed flat over that whole window, every live entry has aged out or been re-encrypted on write, and the key can be dropped from `CACHEKIT_PREVIOUS_MASTER_KEYS` without a hard cut-over.
+
 ---
 
 ## Cross-SDK Interop Mode

crates/cachekit/Cargo.toml

@@ -53,7 +53,7 @@ reliability = ["tokio/time"]
 unsync = []
 
 [dependencies]
-cachekit-core = { version = "0.5", features = ["messagepack"] }
+cachekit-core = { version = "0.6", features = ["messagepack"] }
 serde = { version = "1", features = ["derive"] }
 rmp-serde = "1"
 thiserror = "2.0"

crates/cachekit/src/client.rs

@@ -709,6 +709,12 @@ impl std::fmt::Debug for SecureCache<'_> {
 
 #[cfg(feature = "encryption")]
 impl SecureCache<'_> {
+    /// Rotation drain signal; see
+    /// [`EncryptionLayer::previous_key_hits`](crate::EncryptionLayer::previous_key_hits).
+    pub fn previous_key_hits(&self) -> Vec<u64> {
+        self.encryption.previous_key_hits()
+    }
+
     /// Encrypt and store `value` under `key` using the client's default TTL.
     pub async fn set<T: Serialize>(&self, key: &str, value: &T) -> Result<(), CachekitError> {
         self.set_with_ttl(key, value, self.client.default_ttl).await

crates/cachekit/src/encryption.rs

@@ -13,6 +13,8 @@
 //! Each component is length-prefixed with a 4-byte big-endian u32 to prevent
 //! collision attacks from boundary confusion.
 
+use std::sync::atomic::{AtomicU64, Ordering};
+
 use zeroize::Zeroizing;
 
 use cachekit_core::{Keyring, ZeroKnowledgeEncryptor};
@@ -38,12 +40,15 @@ const AAD_VERSION: u8 = 0x03;
 /// previous key in order, rebuilding the identical AAD per attempt
 /// (cachekit-rs entries carry no per-entry key identity — sequential
 /// attempts are the spec-assigned branch, `protocol/spec/encryption.md` →
-/// "Key Rotation (Keyring)").
+/// "Key Rotation (Keyring)"). Reads served by a previous key are counted —
+/// see [`Self::previous_key_hits`] for the rotation drain workflow.
 pub struct EncryptionLayer {
     encryptor: ZeroKnowledgeEncryptor,
     derived_key: Zeroizing<[u8; 32]>,
     keyring: Keyring,
     tenant_id: String,
+    /// `hits[i]` ↔ `previous_keys[i]`; see [`Self::previous_key_hits`].
+    previous_key_hits: Vec<AtomicU64>,
 }
 
 impl EncryptionLayer {
@@ -132,6 +137,7 @@ impl EncryptionLayer {
             derived_key: Zeroizing::new(tenant_keys.encryption_key),
             keyring,
             tenant_id: tenant_id.to_owned(),
+            previous_key_hits: previous_keys.iter().map(|_| AtomicU64::new(0)).collect(),
         })
     }
 
@@ -155,11 +161,15 @@ impl EncryptionLayer {
     /// attempted first, then each decrypt-only previous key in order, with
     /// the identical AAD per attempt. Entries written before a key rotation
     /// stay readable as long as their key remains in the previous list.
+    ///
+    /// A read served by a previous key is counted in
+    /// [`Self::previous_key_hits`] (the rotation drain signal).
     pub fn decrypt(&self, ciphertext: &[u8], cache_key: &str) -> Result<Vec<u8>, CachekitError> {
         // compressed=false is normative, not a stub — see build_aad's invariant note.
         let aad = self.build_aad(cache_key, false);
-        self.keyring
-            .decrypt(&self.encryptor, ciphertext, &self.tenant_id, &aad)
+        let (plaintext, index) = self
+            .keyring
+            .decrypt_indexed(&self.encryptor, ciphertext, &self.tenant_id, &aad)
             .map_err(|e| match e {
                 // Config-class errors stay config-class (LAB-683 decision):
                 // a derivation failure or caller bug must never masquerade as
@@ -169,7 +179,67 @@ impl EncryptionLayer {
                     CachekitError::Config(format!("keyring decrypt misconfiguration: {e}"))
                 }
                 _ => CachekitError::Encryption(format!("decrypt failed: {e}")),
-            })
+            })?;
+        // index 0 = current key (no signal); i >= 1 = previous_keys[i - 1].
+        // The keyring was built from the same slice, so the slot always
+        // exists; `get` keeps a core bug from panicking the read path.
+        debug_assert!(
+            index <= self.previous_key_hits.len(),
+            "keyring index {index} beyond previous-key slots"
+        );
+        if let Some(hits) = index
+            .checked_sub(1)
+            .and_then(|i| self.previous_key_hits.get(i))
+        {
+            hits.fetch_add(1, Ordering::Relaxed);
+        }
+        Ok(plaintext)
+    }
+
+    /// Reads decrypted by each previous key, by position in the previous-key
+    /// list (`hits[i]` ↔ `previous_keys[i]`); empty when there are none.
+    /// Reads served by the current key are not counted.
+    ///
+    /// This is the rotation **drain signal**. It confirms that a grace window
+    /// has drained; it does not shorten one. Follow the protocol's
+    /// [scheduled-rotation runbook](https://github.com/cachekit-io/protocol/blob/main/decisions/key-rotation.md#runbooks-normative-for-docs):
+    /// audit for non-expiring entries first, add the incoming key as
+    /// decrypt-only fleet-wide, then promote it. The window clock starts only
+    /// when that promotion deploy has completed on every instance — until
+    /// then a lagging instance still writes fresh ciphertext under the
+    /// retiring key and reads it silently as *its* current key (index 0).
+    /// From that point, wait at least the longest TTL in use (per-entry TTLs
+    /// passed to `set_with_ttl` count, not just the default), aggregating
+    /// counts across every instance — they are per process and reset on
+    /// restart. Only when the retiring key's count has stayed flat over that
+    /// whole window has every live entry aged out or been re-encrypted on
+    /// write, and the key can be dropped from the previous list without a
+    /// hard cut-over. Positions are comparable across instances only once
+    /// they all run the same keyring configuration. The signal carries no
+    /// key material — positions and counts only.
+    ///
+    /// ```
+    /// use cachekit::EncryptionLayer;
+    ///
+    /// let k1 = [0x11u8; 32]; // retiring master key
+    /// let k2 = [0x22u8; 32]; // current master key after rotation
+    ///
+    /// // Encrypted under k1, before the rotation...
+    /// let ciphertext = EncryptionLayer::new(&k1, "tenant-123")?.encrypt(b"cached value", "user:1")?;
+    ///
+    /// // ...a layer [current=k2, previous=[k1]] serves it from previous[0]:
+    /// // the retiring key is still draining, not yet safe to drop.
+    /// let layer = EncryptionLayer::with_previous_keys(&k2, &[&k1], "tenant-123")?;
+    /// assert_eq!(layer.previous_key_hits(), vec![0]);
+    /// assert_eq!(layer.decrypt(&ciphertext, "user:1")?, b"cached value");
+    /// assert_eq!(layer.previous_key_hits(), vec![1]);
+    /// # Ok::<(), cachekit::CachekitError>(())
+    /// ```
+    pub fn previous_key_hits(&self) -> Vec<u64> {
+        self.previous_key_hits
+            .iter()
+            .map(|h| h.load(Ordering::Relaxed))
+            .collect()
     }
 
     /// Return the tenant ID used for key derivation.
@@ -444,6 +514,60 @@ mod tests {
         assert!(matches!(result, Err(CachekitError::Config(_))));
     }
 
+    // ── Rotation drain signal (LAB-1678) ─────────────────────────────────────
+
+    #[test]
+    fn current_key_hit_is_not_counted() {
+        // No previous keys: the drain signal has nothing to report.
+        let single = EncryptionLayer::new(K2, TEST_TENANT).unwrap();
+        let ct = single.encrypt(b"v", "user:3").unwrap();
+        single.decrypt(&ct, "user:3").unwrap();
+        assert!(single.previous_key_hits().is_empty());
+
+        // With a previous key, a current-key read (index 0) is a zero reading.
+        let rotated = EncryptionLayer::with_previous_keys(K2, &[K1], TEST_TENANT).unwrap();
+        let ct = rotated.encrypt(b"fresh write", "user:3").unwrap();
+        rotated.decrypt(&ct, "user:3").unwrap();
+        assert_eq!(rotated.previous_key_hits(), vec![0]);
+    }
+
+    #[test]
+    fn previous_key_hit_is_counted_at_its_position() {
+        const K3: &[u8] = &[0x33; 32];
+        let k1_ct = EncryptionLayer::new(K1, TEST_TENANT)
+            .unwrap()
+            .encrypt(b"k1 era", "user:4")
+            .unwrap();
+        let k2_ct = EncryptionLayer::new(K2, TEST_TENANT)
+            .unwrap()
+            .encrypt(b"k2 era", "user:5")
+            .unwrap();
+
+        // current=k3, previous=[k2, k1]: counts index by position in the previous list.
+        let rotated = EncryptionLayer::with_previous_keys(K3, &[K2, K1], TEST_TENANT).unwrap();
+        assert_eq!(rotated.previous_key_hits(), vec![0, 0]);
+
+        rotated.decrypt(&k1_ct, "user:4").unwrap();
+        assert_eq!(rotated.previous_key_hits(), vec![0, 1], "k1 is previous[1]");
+
+        rotated.decrypt(&k2_ct, "user:5").unwrap();
+        rotated.decrypt(&k2_ct, "user:5").unwrap();
+        assert_eq!(rotated.previous_key_hits(), vec![2, 1], "k2 is previous[0]");
+    }
+
+    #[test]
+    fn failed_decrypt_is_not_a_previous_key_hit() {
+        let k1_ct = EncryptionLayer::new(K1, TEST_TENANT)
+            .unwrap()
+            .encrypt(b"v", "key:a")
+            .unwrap();
+        let rotated = EncryptionLayer::with_previous_keys(K2, &[K1], TEST_TENANT).unwrap();
+
+        // Wrong cache key: every attempt fails authentication — no key won.
+        assert!(rotated.decrypt(&k1_ct, "key:b").is_err());
+        assert_eq!(rotated.previous_key_hits(), vec![0]);
+    }
+
     #[test]
     fn debug_redacts_key() {
         let layer = EncryptionLayer::new(TEST_MASTER_KEY, TEST_TENANT).unwrap();

crates/cachekit/tests/encryption_tests.rs

@@ -417,3 +417,48 @@ async fn rotation_round_trip_without_reencryption() {
         "dropped-key read must surface as an encryption error, got {result:?}"
     );
 }
+
+/// Rotation drain signal (LAB-1678): the builder wires the counters into the
+/// user-held secure handle, so a read served by the retiring key is visible
+/// there. Index-0 silence is owned and tested at the layer (`encryption.rs`).
+#[tokio::test]
+async fn rotation_drain_signal_is_visible_on_secure_cache() {
+    const K1: &[u8] = &[0x11; 32];
+    const K2: &[u8] = &[0x22; 32];
+
+    let backend = common::MockBackend::shared();
+
+    let writer = CacheKit::builder()
+        .backend(backend.clone())
+        .default_ttl(Duration::from_secs(60))
+        .no_l1()
+        .encryption_from_bytes(K1, "test-tenant")
+        .expect("encryption setup")
+        .build()
+        .expect("client builds");
+    writer
+        .secure()
+        .expect("secure()")
+        .set("drain:old", &"written under k1")
+        .await
+        .expect("secure set under k1");
+
+    let rotated = CacheKit::builder()
+        .backend(backend)
+        .default_ttl(Duration::from_secs(60))
+        .no_l1()
+        .encryption_from_bytes_with_previous(K2, &[K1], "test-tenant")
+        .expect("keyring setup")
+        .build()
+        .expect("client builds");
+    let secure = rotated.secure().expect("secure()");
+    assert_eq!(secure.previous_key_hits(), vec![0], "nothing read yet");
+
+    // The k1-era entry is served by previous[0]: the grace window is still live.
+    let _: Option<String> = secure.get("drain:old").await.expect("secure get");
+    assert_eq!(
+        secure.previous_key_hits(),
+        vec![1],
+        "previous-key hit is counted"
+    );
+}