diff --git a/Cargo.lock b/Cargo.lock
index ad4c743..fdccdc6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -166,6 +166,7 @@ dependencies = [
  "control",
  "crypto",
  "index",
+ "keys",
  "rt",
  "spool",
  "store",
diff --git a/crates/control/src/error.rs b/crates/control/src/error.rs
index 56e58a5..68ddd25 100644
--- a/crates/control/src/error.rs
+++ b/crates/control/src/error.rs
@@ -27,19 +27,15 @@ impl fmt::Display for Error {
             Self::Keys {
                 tenant,
                 generation,
-                error: KeyError::Unavailable,
-            } => write!(
-                f,
-                "keys unavailable for tenant {tenant} generation {generation}"
-            ),
-            Self::Keys {
-                tenant,
-                generation,
-                error: KeyError::Unknown,
-            } => write!(
-                f,
-                "keys do not know tenant {tenant} generation {generation}"
-            ),
+                error,
+            } => {
+                let what = match error {
+                    KeyError::Unavailable => "unavailable",
+                    KeyError::Unknown => "unknown",
+                    KeyError::Failed => "failed",
+                };
+                write!(f, "keys {what} for tenant {tenant} generation {generation}")
+            }
         }
     }
 }
diff --git a/crates/control/src/lib.rs b/crates/control/src/lib.rs
index 51aa4f8..df8b925 100644
--- a/crates/control/src/lib.rs
+++ b/crates/control/src/lib.rs
@@ -11,7 +11,7 @@ mod writer;
 pub use cache::{Cache, CacheStats};
 pub use error::Error;
 pub use leases::{Ask, Gcra, IDLE, LIFETIME, LeaseStats, Leases, Taken, Tokens, WAIT, limit_core};
-pub use names::{Call, Loaded as LoadedQueues, Naming, NamingStats, Queues};
+pub use names::{Call, Loaded as LoadedQueues, Naming, NamingStats, Queues, Unnamed};
 pub use queue::{
     KeyError, KeySource, LABEL_SIZE, Label, MAX_DOMAIN, Name, Queue, Ready, Tenancy, domain_ok,
     mix, naming_shard, queue_shard, tenant_keys,
diff --git a/crates/control/src/names.rs b/crates/control/src/names.rs
index 9024e18..e96b1b7 100644
--- a/crates/control/src/names.rs
+++ b/crates/control/src/names.rs
@@ -22,7 +22,7 @@ pub enum Call {
     Name(Name),
     Named {
         name: Name,
-        id: Option<u32>,
+        id: Result<u32, Unnamed>,
     },
     Forget {
         tenant: u32,
@@ -41,16 +41,25 @@ pub struct Naming {
     pub cache: usize,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Unnamed {
+    Refused,
+    Failed,
+    Keys(KeyError),
+}
+
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub struct NamingStats {
     pub cache: CacheStats,
     pub allocated: u64,
     pub failed: u64,
+    pub keys: u64,
     pub refused: u64,
     pub skipped: u64,
 }
 
 type Lock = (u32, Box<str>);
+type Named = Result<u32, Unnamed>;
 
 struct Ready {
     name: Name,
@@ -204,7 +213,7 @@ struct Inner {
     changes: Changes,
     owned: RefCell<Owned>,
     cache: RefCell<Cache>,
-    waiting: RefCell<BTreeMap<Name, Vec<Sender<Option<u32>>>>>,
+    waiting: RefCell<BTreeMap<Name, Vec<Sender<Named>>>>,
     forgetting: RefCell<BTreeMap<u64, (u16, Sender<()>)>>,
     tickets: Cell<u64>,
     peers: OnceCell<Peers<Call>>,
@@ -266,8 +275,13 @@ impl Queues {
                         hidden: parts.sealed.to_vec(),
                     })
                 });
+                let fits = |head: &Head| {
+                    head.id == id
+                        && head.tenant == item.tenant
+                        && queue_shard(id, naming.shards) == shard
+                };
                 match stored {
-                    Some(stored) if stored.head.id == id => found
+                    Some(stored) if fits(&stored.head) => found
                         .entry((stored.head.tenant, stored.head.generation))
                         .or_default()
                         .push(stored),
@@ -324,7 +338,15 @@ impl Queues {
             let core = self.naming_core(name.tenant, &name.domain);
             self.send(core, Call::Name(name.clone()));
         }
-        answer.recv().await.flatten().ok_or(Error::Refused)
+        match answer.recv().await {
+            Some(Ok(id)) => Ok(id),
+            Some(Err(Unnamed::Keys(error))) => Err(Error::Keys {
+                tenant: name.tenant,
+                generation: name.generation,
+                error,
+            }),
+            _ => Err(Error::Refused),
+        }
     }
 
     pub async fn rebind(&self, name: &Name, pool: u32) -> Result<u32, Error> {
@@ -443,7 +465,7 @@ impl Queues {
         match call {
             Call::Name(name) => self.asked(name, from),
             Call::Named { name, id } => {
-                if let Some(id) = id {
+                if let Ok(id) = id {
                     self.inner.cache.borrow_mut().put(name.clone(), id);
                 }
                 let waiting = self.inner.waiting.borrow_mut().remove(&name);
@@ -495,7 +517,7 @@ impl Queues {
             let mut owned = self.inner.owned.borrow_mut();
             if let Some(&id) = owned.names.get(&name) {
                 drop(owned);
-                self.send(from, Call::Named { name, id: Some(id) });
+                self.send(from, Call::Named { name, id: Ok(id) });
                 return false;
             }
             let shard = naming_shard(&naming.key, name.tenant, &name.domain, naming.shards);
@@ -507,7 +529,13 @@ impl Queues {
         };
         let Some((shard, id)) = id else {
             self.count(|s| s.refused += 1);
-            self.send(from, Call::Named { name, id: None });
+            self.send(
+                from,
+                Call::Named {
+                    name,
+                    id: Err(Unnamed::Refused),
+                },
+            );
             return false;
         };
         self.inner.ready.borrow_mut().push_back(Ready {
@@ -574,13 +602,20 @@ impl Queues {
             name: name.clone(),
             draining: false,
         };
-        let id = if self.write(shard, &queue).await.is_ok() {
-            self.inner.owned.borrow_mut().insert(queue);
-            self.count(|s| s.allocated += 1);
-            Some(id)
-        } else {
-            self.count(|s| s.failed += 1);
-            None
+        let id = match self.write(shard, &queue).await {
+            Ok(()) => {
+                self.inner.owned.borrow_mut().insert(queue);
+                self.count(|s| s.allocated += 1);
+                Ok(id)
+            }
+            Err(Error::Keys { error, .. }) => {
+                self.count(|s| s.keys += 1);
+                Err(Unnamed::Keys(error))
+            }
+            Err(_) => {
+                self.count(|s| s.failed += 1);
+                Err(Unnamed::Failed)
+            }
         };
         self.send(from, Call::Named { name, id });
         drop(held);
diff --git a/crates/control/src/queue.rs b/crates/control/src/queue.rs
index 32c2743..89efef3 100644
--- a/crates/control/src/queue.rs
+++ b/crates/control/src/queue.rs
@@ -144,6 +144,7 @@ impl Label {
 pub enum KeyError {
     Unavailable,
     Unknown,
+    Failed,
 }
 
 pub type Ready<'a> = Pin<Box<dyn Future<Output = Result<TenantKeys, KeyError>> + 'a>>;
diff --git a/crates/control/tests/queues.rs b/crates/control/tests/queues.rs
index f0da7e2..8e3c5a6 100644
--- a/crates/control/tests/queues.rs
+++ b/crates/control/tests/queues.rs
@@ -54,6 +54,7 @@ const DOMAINS: [&str; 12] = [
 struct Toy {
     wrong: BTreeSet<u32>,
     unknown: BTreeSet<u32>,
+    failed: BTreeSet<u32>,
     unavailable: Cell<u32>,
     asked: Cell<u32>,
 }
@@ -68,6 +69,9 @@ impl Toy {
         if self.unknown.contains(&tenant) {
             return Err(KeyError::Unknown);
         }
+        if self.failed.contains(&tenant) {
+            return Err(KeyError::Failed);
+        }
         let seed = if self.wrong.contains(&tenant) { 9 } else { 1 };
         let root = Tenant::new(&Secret::new([seed; 32]));
         Ok(TenantKeys::derive(tenant, generation, &root, 1))
@@ -502,7 +506,7 @@ fn records_that_do_not_open_or_decode_are_skipped_at_start() {
         let shard = naming_shard(&KEY, 1, "b.example", SHARDS);
         let ids: Vec<u32> = (900..)
             .filter(|&id| queue_shard(id, SHARDS) == shard)
-            .take(4)
+            .take(5)
             .collect();
         let real = |id| Queue {
             id,
@@ -515,6 +519,17 @@ fn records_that_do_not_open_or_decode_are_skipped_at_start() {
             (ids[1], 1, real(ids[1]).encode(), b"not sealed".to_vec()),
             (ids[2], 3, real(ids[2]).encode(), seal_for(ids[2])),
             (ids[3], 1, real(ids[2]).encode(), seal_for(ids[2])),
+            (
+                ids[4],
+                1,
+                Queue {
+                    id: ids[4],
+                    name: name(9, 1, 1, "b.example"),
+                    draining: false,
+                }
+                .encode(),
+                seal_for(ids[4]),
+            ),
         ] {
             let batch = Batch {
                 shard,
@@ -539,12 +554,13 @@ fn records_that_do_not_open_or_decode_are_skipped_at_start() {
     machine.block_on(0, node.stop());
     let refusing = Rc::new(Toy {
         wrong: BTreeSet::from([2]),
+        unknown: BTreeSet::from([9]),
         ..Toy::default()
     });
     let driver = Rc::new(SimDriver::with_disk(2, Config::default(), driver.crash()));
     let machine = self::machine(&driver, cores);
     let node = Node::start(&machine, cores, &refusing).expect("restart");
-    assert_eq!(node.skipped, 5);
+    assert_eq!(node.skipped, 6);
     assert_eq!(node.loaded, 1);
     assert!(node.queues[0].entry(kept).is_some());
     assert!(node.queues[0].entry(gone).is_none());
@@ -645,6 +661,23 @@ fn start_waits_for_keys_asks_once_per_tenant_and_fails_on_unknown_ones() {
         ),
         "{failed:?}"
     );
+    let broken = Rc::new(Toy {
+        failed: BTreeSet::from([1]),
+        ..Toy::default()
+    });
+    let driver = Rc::new(SimDriver::with_disk(6, Config::default(), disk.clone()));
+    let failed = Node::try_start(&self::machine(&driver, cores), cores, &broken).err();
+    assert!(
+        matches!(
+            failed,
+            Some(Some(Error::Keys {
+                tenant: 1,
+                generation: 1,
+                error: KeyError::Failed,
+            }))
+        ),
+        "{failed:?}"
+    );
     let slow = Rc::new(Toy {
         unavailable: Cell::new(3),
         ..Toy::default()
@@ -657,6 +690,28 @@ fn start_waits_for_keys_asks_once_per_tenant_and_fails_on_unknown_ones() {
     assert_eq!(node.skipped, 0);
     let groups = 2 * SHARDS;
     assert!(slow.asked.get() <= groups + 3, "{} asks", slow.asked.get());
+    slow.unavailable.set(1_000);
+    let queues = node.queues[0].clone();
+    let refused = machine.block_on(1, async move {
+        let named = queues.name(&name(1, 1, 1, "new.example")).await;
+        (named, queues.stats().keys)
+    });
+    assert!(
+        matches!(
+            refused,
+            (
+                Err(Error::Keys {
+                    tenant: 1,
+                    generation: 1,
+                    error: KeyError::Unavailable,
+                }),
+                _
+            )
+        ),
+        "{refused:?}"
+    );
+    let keys: u64 = node.queues.iter().map(|q| q.stats().keys).sum();
+    assert_eq!(keys, 1);
     machine.block_on(0, node.stop());
 }
 
diff --git a/crates/crypto/src/keys.rs b/crates/crypto/src/keys.rs
index dfcea06..8f70d99 100644
--- a/crates/crypto/src/keys.rs
+++ b/crates/crypto/src/keys.rs
@@ -1,9 +1,9 @@
-use std::rc::Rc;
+use std::sync::Arc;
 
 use crate::{DataKey, Error, Secret, Tenant};
 
 #[derive(Clone)]
-pub struct TenantKeys(Rc<Held>);
+pub struct TenantKeys(Arc<Held>);
 
 struct Held {
     tenant: u32,
@@ -20,7 +20,7 @@ impl TenantKeys {
         version: u8,
         hash: Secret<16>,
     ) -> Self {
-        Self(Rc::new(Held {
+        Self(Arc::new(Held {
             tenant,
             generation,
             data: DataKey::new(data, version),
diff --git a/crates/keys/src/cache.rs b/crates/keys/src/cache.rs
new file mode 100644
index 0000000..a6d7529
--- /dev/null
+++ b/crates/keys/src/cache.rs
@@ -0,0 +1,157 @@
+use std::collections::HashMap;
+use std::hash::Hash;
+
+struct Slot<K, V> {
+    key: K,
+    value: V,
+    used: bool,
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct CacheStats {
+    pub hits: u64,
+    pub misses: u64,
+    pub evicted: u64,
+}
+
+pub struct Cache<K, V> {
+    capacity: usize,
+    slots: Vec<Option<Slot<K, V>>>,
+    index: HashMap<K, usize>,
+    free: Vec<usize>,
+    hand: usize,
+    stats: CacheStats,
+}
+
+impl<K: Copy + Eq + Hash, V: Clone> Cache<K, V> {
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            capacity: capacity.max(1),
+            slots: Vec::new(),
+            index: HashMap::new(),
+            free: Vec::new(),
+            hand: 0,
+            stats: CacheStats::default(),
+        }
+    }
+
+    pub fn len(&self) -> usize {
+        self.index.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.index.is_empty()
+    }
+
+    pub fn stats(&self) -> CacheStats {
+        self.stats
+    }
+
+    pub fn get(&mut self, key: &K) -> Option<V> {
+        let found = self.index.get(key).and_then(|&at| self.slots[at].as_mut());
+        if let Some(slot) = found {
+            slot.used = true;
+            self.stats.hits += 1;
+            Some(slot.value.clone())
+        } else {
+            self.stats.misses += 1;
+            None
+        }
+    }
+
+    pub fn put(&mut self, key: K, value: V) {
+        if let Some(&at) = self.index.get(&key) {
+            if let Some(slot) = self.slots[at].as_mut() {
+                slot.value = value;
+            }
+            return;
+        }
+        let at = if let Some(at) = self.free.pop() {
+            at
+        } else if self.slots.len() < self.capacity {
+            self.slots.push(None);
+            self.slots.len() - 1
+        } else {
+            self.evict()
+        };
+        self.index.insert(key, at);
+        self.slots[at] = Some(Slot {
+            key,
+            value,
+            used: false,
+        });
+    }
+
+    pub fn remove(&mut self, key: &K) -> Option<V> {
+        let at = self.index.remove(key)?;
+        let slot = self.slots[at].take()?;
+        self.free.push(at);
+        Some(slot.value)
+    }
+
+    pub fn retain(&mut self, mut keep: impl FnMut(&K, &V) -> bool) {
+        for at in 0..self.slots.len() {
+            let drop = self.slots[at]
+                .as_ref()
+                .is_some_and(|slot| !keep(&slot.key, &slot.value));
+            if drop {
+                let slot = self.slots[at].take().expect("slot present");
+                self.index.remove(&slot.key);
+                self.free.push(at);
+            }
+        }
+    }
+
+    pub fn clear(&mut self) {
+        self.slots.clear();
+        self.index.clear();
+        self.free.clear();
+        self.hand = 0;
+    }
+
+    fn evict(&mut self) -> usize {
+        loop {
+            let at = self.hand;
+            self.hand = (self.hand + 1) % self.slots.len();
+            match self.slots[at].as_mut() {
+                Some(slot) if slot.used => slot.used = false,
+                Some(_) => {
+                    let slot = self.slots[at].take().expect("slot present");
+                    self.index.remove(&slot.key);
+                    self.stats.evicted += 1;
+                    return at;
+                }
+                None => return at,
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Cache;
+
+    #[test]
+    fn keeps_used_entries_and_evicts_the_others() {
+        let mut cache = Cache::new(3);
+        for n in 0..3u32 {
+            cache.put(n, n * 10);
+        }
+        assert_eq!(cache.get(&0), Some(0));
+        assert_eq!(cache.get(&2), Some(20));
+        cache.put(3, 30);
+        assert_eq!(cache.get(&1), None);
+        assert_eq!(cache.get(&0), Some(0));
+        assert_eq!(cache.get(&3), Some(30));
+        assert_eq!(cache.len(), 3);
+        assert_eq!(cache.stats().evicted, 1);
+        assert_eq!(cache.remove(&0), Some(0));
+        cache.put(4, 40);
+        assert_eq!(cache.len(), 3);
+        cache.retain(|&key, _| key != 4);
+        assert_eq!(cache.get(&4), None);
+        assert_eq!(cache.len(), 2);
+        cache.clear();
+        assert!(cache.is_empty());
+    }
+}
diff --git a/crates/keys/src/client.rs b/crates/keys/src/client.rs
index f868a75..fde4a20 100644
--- a/crates/keys/src/client.rs
+++ b/crates/keys/src/client.rs
@@ -16,6 +16,7 @@ const CONNECT_TRIES: u32 = 8;
 const CONNECT_PAUSE: Duration = Duration::from_millis(10);
 
 pub struct Client {
+    close: Box<dyn Fn()>,
     next: Cell<u32>,
     waiting: Rc<RefCell<HashMap<u32, Sender<Answer>>>>,
     frames: Sender<Plain>,
@@ -29,8 +30,24 @@ impl Client {
         let closed = Rc::new(Cell::new(false));
         let (frames, outgoing) = rt::channel();
         rt::spawn(writer(Rc::clone(&stream), outgoing, Rc::clone(&closed)));
-        rt::spawn(reader(stream, Rc::clone(&waiting), Rc::clone(&closed)));
+        rt::spawn(reader(
+            Rc::clone(&stream),
+            Rc::clone(&waiting),
+            Rc::clone(&closed),
+        ));
+        let close = {
+            let closed = Rc::clone(&closed);
+            let shut = Cell::new(false);
+            Box::new(move || {
+                closed.set(true);
+                if !shut.replace(true) {
+                    let stream = Rc::clone(&stream);
+                    rt::spawn(async move { drop(stream.shutdown().await) });
+                }
+            })
+        };
         Self {
+            close,
             next: Cell::new(0),
             waiting,
             frames,
@@ -59,6 +76,10 @@ impl Client {
         self.closed.get()
     }
 
+    pub fn close(&self) {
+        (self.close)();
+    }
+
     pub async fn call(&self, request: &Request) -> Result<Answer, Error> {
         let (_, mut answer) = self.send(request)?;
         answer.recv().await.ok_or(Error::Closed)
@@ -92,6 +113,14 @@ impl Client {
     }
 }
 
+impl Drop for Client {
+    fn drop(&mut self) {
+        if rt::in_runtime() {
+            self.close();
+        }
+    }
+}
+
 async fn writer<S: Stream>(stream: Rc<S>, mut outgoing: Receiver<Plain>, closed: Rc<Cell<bool>>) {
     while let Some(first) = outgoing.recv().await {
         let mut batch = vec![first];
diff --git a/crates/keys/src/error.rs b/crates/keys/src/error.rs
index 17aa13a..29eb2dc 100644
--- a/crates/keys/src/error.rs
+++ b/crates/keys/src/error.rs
@@ -19,6 +19,7 @@ pub enum Error {
     Closed,
     Timeout,
     Peer,
+    Changed,
     Crypto(crypto::Error),
 }
 
@@ -54,6 +55,7 @@ impl fmt::Display for Error {
             Self::Closed => write!(f, "keys connection closed"),
             Self::Timeout => write!(f, "keys call timed out"),
             Self::Peer => write!(f, "keys socket belongs to another user"),
+            Self::Changed => write!(f, "keys serve another vault than before"),
             Self::Crypto(error) => write!(f, "{error}"),
         }
     }
diff --git a/crates/keys/src/lib.rs b/crates/keys/src/lib.rs
index d5d2f81..a3df4a9 100644
--- a/crates/keys/src/lib.rs
+++ b/crates/keys/src/lib.rs
@@ -1,19 +1,23 @@
 mod audit;
+mod cache;
 mod client;
 mod clock;
 mod error;
 pub mod file;
 mod limit;
 pub mod proto;
+mod ring;
 mod server;
 mod state;
 mod vault;
 
 pub use audit::{Audit, Op};
+pub use cache::{Cache, CacheStats};
 pub use client::Client;
 pub use clock::{Clock, MIN_GAP, Step};
 pub use error::Error;
 pub use limit::{Limits, Rate};
+pub use ring::{Connect, Connecting, Fault, Ring, RingSettings, RingStats, Working};
 pub use server::{Server, Settings, Stats, Stream};
 pub use state::{Entry, State};
 pub use vault::{Keys, MAX_RETENTION, MODE, VERSION, Vault};
diff --git a/crates/keys/src/ring.rs b/crates/keys/src/ring.rs
new file mode 100644
index 0000000..0b1f6f7
--- /dev/null
+++ b/crates/keys/src/ring.rs
@@ -0,0 +1,346 @@
+use std::cell::{Cell, RefCell};
+use std::collections::HashMap;
+use std::pin::Pin;
+use std::rc::Rc;
+use std::time::Duration;
+
+use crypto::{Secret, TenantKeys};
+use rt::{Instant, Sender};
+
+use crate::cache::{Cache, CacheStats};
+use crate::proto::{Answer, Request};
+use crate::{Client, Error};
+
+pub type Connecting = Pin<Box<dyn Future<Output = Result<Client, Error>>>>;
+pub type Connect = Rc<dyn Fn() -> Connecting>;
+type Waiters = Vec<Sender<Result<Working, Fault>>>;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct RingSettings {
+    pub node: u32,
+    pub deadline: Duration,
+    pub slow: Duration,
+    pub retry: Duration,
+    pub retry_max: Duration,
+    pub tenants: usize,
+}
+
+impl RingSettings {
+    pub fn new(node: u32) -> Self {
+        Self {
+            node,
+            deadline: Duration::from_secs(2),
+            slow: Duration::from_secs(30),
+            retry: Duration::from_millis(100),
+            retry_max: Duration::from_secs(5),
+            tenants: 65_536,
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct Working {
+    pub keys: TenantKeys,
+    pub day: u32,
+    pub oldest: u32,
+    pub today: [u8; 32],
+    pub tomorrow: [u8; 32],
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Fault {
+    Unavailable,
+    Unknown,
+    Failed,
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct RingStats {
+    pub connects: u64,
+    pub drops: u64,
+    pub slow: u64,
+    pub refused: u64,
+    pub opens: u64,
+    pub unknown: u64,
+    pub unavailable: u64,
+    pub failed: u64,
+    pub tenants: CacheStats,
+}
+
+enum Link {
+    Down { retry_at: Instant, pause: Duration },
+    Connecting,
+    Up(Rc<Client>),
+}
+
+#[derive(Clone)]
+pub struct Ring {
+    inner: Rc<Inner>,
+}
+
+struct Inner {
+    settings: RingSettings,
+    connect: Connect,
+    link: RefCell<Link>,
+    naming: RefCell<Option<Secret<16>>>,
+    day: Cell<u32>,
+    failed: Cell<bool>,
+    tenants: RefCell<Cache<(u32, u32), Working>>,
+    opening: RefCell<HashMap<(u32, u32), Waiters>>,
+    stats: Cell<RingStats>,
+}
+
+struct Pending<'a> {
+    ring: &'a Ring,
+    pause: Duration,
+}
+
+impl Drop for Pending<'_> {
+    fn drop(&mut self) {
+        let mut link = self.ring.inner.link.borrow_mut();
+        if matches!(*link, Link::Connecting) {
+            *link = Link::Down {
+                retry_at: rt::now(),
+                pause: self.pause,
+            };
+        }
+    }
+}
+
+impl Ring {
+    pub fn new(connect: Connect, settings: RingSettings) -> Self {
+        Self {
+            inner: Rc::new(Inner {
+                settings,
+                connect,
+                link: RefCell::new(Link::Down {
+                    retry_at: rt::now(),
+                    pause: settings.retry,
+                }),
+                naming: RefCell::new(None),
+                day: Cell::new(0),
+                failed: Cell::new(false),
+                tenants: RefCell::new(Cache::new(settings.tenants)),
+                opening: RefCell::new(HashMap::new()),
+                stats: Cell::new(RingStats::default()),
+            }),
+        }
+    }
+
+    pub fn failed(&self) -> bool {
+        self.inner.failed.get()
+    }
+
+    pub fn day(&self) -> u32 {
+        self.inner.day.get()
+    }
+
+    pub fn naming(&self) -> Option<[u8; 16]> {
+        self.inner.naming.borrow().as_ref().map(|key| *key.expose())
+    }
+
+    pub fn stats(&self) -> RingStats {
+        RingStats {
+            tenants: self.inner.tenants.borrow().stats(),
+            ..self.inner.stats.get()
+        }
+    }
+
+    pub async fn hello(&self) -> Result<(), Error> {
+        self.client().await.map(drop)
+    }
+
+    pub fn get(&self, tenant: u32, generation: u32) -> Option<Working> {
+        if self.failed() {
+            return None;
+        }
+        self.inner.tenants.borrow_mut().get(&(tenant, generation))
+    }
+
+    pub async fn open(&self, tenant: u32, generation: u32) -> Result<Working, Fault> {
+        if let Some(found) = self.get(tenant, generation) {
+            return Ok(found);
+        }
+        if self.failed() {
+            self.count(|s| s.refused += 1);
+            return Err(Fault::Failed);
+        }
+        let key = (tenant, generation);
+        let (reply, mut answer) = rt::channel();
+        let first = {
+            let mut opening = self.inner.opening.borrow_mut();
+            let waiting = opening.entry(key).or_default();
+            waiting.push(reply);
+            waiting.len() == 1
+        };
+        if first {
+            let ring = self.clone();
+            rt::spawn(async move {
+                let result = ring.fetch(tenant, generation).await;
+                if let Ok(working) = &result
+                    && !ring.failed()
+                {
+                    ring.inner.tenants.borrow_mut().put(key, working.clone());
+                }
+                let waiting = ring.inner.opening.borrow_mut().remove(&key);
+                for waiter in waiting.unwrap_or_default() {
+                    let _ = waiter.send(result.clone());
+                }
+            });
+        }
+        answer.recv().await.unwrap_or(Err(Fault::Unavailable))
+    }
+
+    pub async fn call(&self, request: &Request) -> Result<Answer, Error> {
+        let client = self.client().await?;
+        let settings = self.inner.settings;
+        let limit = match request {
+            Request::Hello { .. } | Request::Open { .. } | Request::Unwrap { .. } => {
+                settings.deadline
+            }
+            _ => settings.slow,
+        };
+        match client.call_within(request, limit).await {
+            Err(Error::Timeout) => {
+                self.count(|s| s.slow += 1);
+                let hello = Request::Hello {
+                    node: settings.node,
+                };
+                if client.call_within(&hello, settings.deadline).await.is_err() {
+                    self.drop_link(&client);
+                }
+                Err(Error::Timeout)
+            }
+            Err(Error::Closed) => {
+                self.drop_link(&client);
+                Err(Error::Closed)
+            }
+            other => other,
+        }
+    }
+
+    async fn fetch(&self, tenant: u32, generation: u32) -> Result<Working, Fault> {
+        self.count(|s| s.opens += 1);
+        match self.call(&Request::Open { tenant, generation }).await {
+            Ok(Answer::Keys {
+                day,
+                oldest,
+                version,
+                data,
+                hash,
+                today,
+                tomorrow,
+            }) => {
+                self.saw_day(day);
+                Ok(Working {
+                    keys: TenantKeys::new(tenant, generation, &data, version, hash),
+                    day,
+                    oldest,
+                    today,
+                    tomorrow,
+                })
+            }
+            Ok(Answer::Refused(Error::Unknown | Error::Generation)) => {
+                self.count(|s| s.unknown += 1);
+                Err(Fault::Unknown)
+            }
+            Ok(Answer::Refused(Error::Failed | Error::Unreadable))
+            | Err(Error::Failed | Error::Changed) => {
+                self.count(|s| s.failed += 1);
+                Err(Fault::Failed)
+            }
+            _ => {
+                self.count(|s| s.unavailable += 1);
+                Err(Fault::Unavailable)
+            }
+        }
+    }
+
+    async fn client(&self) -> Result<Rc<Client>, Error> {
+        if self.failed() {
+            self.count(|s| s.refused += 1);
+            return Err(Error::Failed);
+        }
+        let pause = match &*self.inner.link.borrow() {
+            Link::Up(client) if !client.is_closed() => return Ok(Rc::clone(client)),
+            Link::Up(_) => self.inner.settings.retry,
+            Link::Connecting => return Err(Error::Closed),
+            Link::Down { retry_at, pause } => {
+                if rt::now() < *retry_at {
+                    return Err(Error::Closed);
+                }
+                *pause
+            }
+        };
+        *self.inner.link.borrow_mut() = Link::Connecting;
+        let pending = Pending { ring: self, pause };
+        let connected = self.connect().await;
+        let next = match &connected {
+            Ok(client) => {
+                self.count(|s| s.connects += 1);
+                Link::Up(Rc::clone(client))
+            }
+            Err(_) => Link::Down {
+                retry_at: rt::now() + pause,
+                pause: (pause * 2).min(self.inner.settings.retry_max),
+            },
+        };
+        *self.inner.link.borrow_mut() = next;
+        drop(pending);
+        connected
+    }
+
+    async fn connect(&self) -> Result<Rc<Client>, Error> {
+        let settings = self.inner.settings;
+        let client = rt::timeout(settings.deadline, (self.inner.connect)())
+            .await
+            .map_err(|_| Error::Timeout)??;
+        let hello = Request::Hello {
+            node: settings.node,
+        };
+        let answer = client.call_within(&hello, settings.deadline).await;
+        let Ok(Answer::Hello { day, naming }) = answer else {
+            client.close();
+            return Err(answer.err().unwrap_or(Error::Bad));
+        };
+        let changed = {
+            let mut known = self.inner.naming.borrow_mut();
+            if let Some(key) = known.as_ref() {
+                !(*key == naming)
+            } else {
+                *known = Some(naming);
+                false
+            }
+        };
+        if changed {
+            client.close();
+            self.inner.failed.set(true);
+            self.inner.tenants.borrow_mut().clear();
+            return Err(Error::Changed);
+        }
+        self.saw_day(day);
+        Ok(Rc::new(client))
+    }
+
+    fn drop_link(&self, client: &Rc<Client>) {
+        let mut link = self.inner.link.borrow_mut();
+        if matches!(&*link, Link::Up(current) if Rc::ptr_eq(current, client)) {
+            client.close();
+            *link = Link::Down {
+                retry_at: rt::now(),
+                pause: self.inner.settings.retry,
+            };
+            self.count(|s| s.drops += 1);
+        }
+    }
+
+    fn saw_day(&self, day: u32) {
+        self.inner.day.set(self.inner.day.get().max(day));
+    }
+
+    fn count(&self, change: impl FnOnce(&mut RingStats)) {
+        let mut stats = self.inner.stats.get();
+        change(&mut stats);
+        self.inner.stats.set(stats);
+    }
+}
diff --git a/crates/keys/src/server.rs b/crates/keys/src/server.rs
index 0c7288a..9856d25 100644
--- a/crates/keys/src/server.rs
+++ b/crates/keys/src/server.rs
@@ -18,6 +18,7 @@ const IN_FLIGHT: usize = 1_024;
 pub trait Stream {
     fn read(&self, buf: Buf) -> impl Future<Output = (io::Result<usize>, Buf)>;
     fn write_all(&self, buf: Buf) -> impl Future<Output = (io::Result<()>, Buf)>;
+    fn shutdown(&self) -> impl Future<Output = io::Result<()>>;
 }
 
 impl Stream for rt::TcpStream {
@@ -28,6 +29,10 @@ impl Stream for rt::TcpStream {
     fn write_all(&self, buf: Buf) -> impl Future<Output = (io::Result<()>, Buf)> {
         rt::TcpStream::write_all(self, buf)
     }
+
+    fn shutdown(&self) -> impl Future<Output = io::Result<()>> {
+        rt::TcpStream::shutdown(self)
+    }
 }
 
 impl Stream for rt::UnixStream {
@@ -38,6 +43,10 @@ impl Stream for rt::UnixStream {
     fn write_all(&self, buf: Buf) -> impl Future<Output = (io::Result<()>, Buf)> {
         rt::UnixStream::write_all(self, buf)
     }
+
+    fn shutdown(&self) -> impl Future<Output = io::Result<()>> {
+        rt::UnixStream::shutdown(self)
+    }
 }
 
 #[derive(Clone, Copy, Debug)]
diff --git a/crates/keys/tests/ring.rs b/crates/keys/tests/ring.rs
new file mode 100644
index 0000000..8c47009
--- /dev/null
+++ b/crates/keys/tests/ring.rs
@@ -0,0 +1,320 @@
+use std::cell::Cell;
+use std::net::{Ipv4Addr, SocketAddr};
+use std::path::Path;
+use std::rc::Rc;
+use std::time::Duration;
+
+use crypto::{Plain, Secret};
+use keys::proto::{Answer, Frames, Request};
+use keys::{Audit, Client, Connect, Error, Fault, Ring, RingSettings, Server, Settings, Vault};
+use rt::sim::{Config, SimDriver};
+use rt::{Buf, Runtime, TcpListener, TcpStream, sleep, spawn};
+
+const DAY: u32 = 20_833;
+
+fn at(host: u8) -> SocketAddr {
+    SocketAddr::from((Ipv4Addr::new(10, 0, 0, host), 7000))
+}
+
+fn quiet() -> Config {
+    let mut config = Config::default();
+    config.faults.crash_keep = 0;
+    config
+}
+
+fn run<T: 'static>(driver: &Rc<SimDriver>, future: impl Future<Output = T> + 'static) -> T {
+    Runtime::new(driver.clone()).block_on(future)
+}
+
+async fn serve(dir: &str, host: u8) -> Rc<Server> {
+    let vault = Vault::init(Path::new(dir), DAY).await.expect("init");
+    let audit = Audit::open(Path::new(dir)).await.expect("audit");
+    let server = Rc::new(Server::start(vault, audit, Settings::default()));
+    let listener = TcpListener::bind(at(host), 16).await.expect("bind");
+    let accepting = Rc::clone(&server);
+    spawn(async move {
+        while let Ok((stream, _)) = listener.accept().await {
+            accepting.serve(stream);
+        }
+    });
+    server
+}
+
+async fn create(host: u8, tenant: u32) {
+    let client = Client::new(TcpStream::connect(at(host)).await.expect("connect"));
+    let create = Request::Create {
+        tenant,
+        generation: 1,
+        retention: 30,
+    };
+    assert!(matches!(
+        client.call(&create).await.expect("create"),
+        Answer::Keys { .. }
+    ));
+    client.close();
+}
+
+async fn naming(host: u8) -> Secret<16> {
+    let client = Client::new(TcpStream::connect(at(host)).await.expect("connect"));
+    let answer = client
+        .call(&Request::Hello { node: 1 })
+        .await
+        .expect("hello");
+    client.close();
+    match answer {
+        Answer::Hello { naming, .. } => naming,
+        _ => panic!("expected hello"),
+    }
+}
+
+fn hosts(order: &[u8]) -> (Connect, Rc<Cell<usize>>) {
+    let order: Rc<[u8]> = order.into();
+    let tries = Rc::new(Cell::new(0));
+    let counted = Rc::clone(&tries);
+    let connect: Connect = Rc::new(move || {
+        let n = counted.get();
+        counted.set(n + 1);
+        let host = order[n.min(order.len() - 1)];
+        Box::pin(async move { Ok(Client::new(TcpStream::connect(at(host)).await?)) })
+    });
+    (connect, tries)
+}
+
+fn settings() -> RingSettings {
+    RingSettings {
+        deadline: Duration::from_secs(1),
+        ..RingSettings::new(1)
+    }
+}
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum Fake {
+    Silent,
+    Closes,
+}
+
+async fn fake(host: u8, naming: Secret<16>, mode: Fake) -> Rc<Cell<u32>> {
+    let listener = TcpListener::bind(at(host), 4).await.expect("bind");
+    let naming = Rc::new(naming);
+    let closed = Rc::new(Cell::new(0));
+    let counted = Rc::clone(&closed);
+    spawn(async move {
+        while let Ok((stream, _)) = listener.accept().await {
+            let naming = Rc::clone(&naming);
+            let closed = Rc::clone(&counted);
+            spawn(async move {
+                let mut frames = Frames::default();
+                let mut greeted = false;
+                loop {
+                    let (read, buf) = stream.read(Buf::new(4096)).await;
+                    let Ok(n @ 1..) = read else {
+                        closed.set(closed.get() + 1);
+                        return;
+                    };
+                    frames.push(&buf[..n]);
+                    while let Ok(Some((op, id, body))) = frames.pop() {
+                        let answer = match Request::decode(op, &body) {
+                            Ok(Request::Hello { .. }) if !greeted => {
+                                greeted = true;
+                                Answer::Hello {
+                                    day: DAY,
+                                    naming: Secret::new(*naming.expose()),
+                                }
+                            }
+                            Ok(Request::Open { .. }) if mode == Fake::Closes => Answer::Keys {
+                                day: DAY,
+                                oldest: DAY,
+                                version: 1,
+                                data: Secret::new([5; 32]),
+                                hash: Secret::new([6; 16]),
+                                today: [7; 32],
+                                tomorrow: [8; 32],
+                            },
+                            _ => continue,
+                        };
+                        let mut out = Plain::with_capacity(256);
+                        answer.encode(id, &mut out);
+                        let (written, _) = stream.write_all(Buf::from_slice(&out)).await;
+                        written.expect("write");
+                        if matches!(answer, Answer::Keys { .. }) {
+                            drop(stream.shutdown().await);
+                        }
+                    }
+                }
+            });
+        }
+    });
+    closed
+}
+
+#[test]
+fn tenant_keys_open_once_on_first_use_and_stay_cached() {
+    let driver = Rc::new(SimDriver::new(1, quiet()));
+    run(&driver, async {
+        let _server = serve("/a", 1).await;
+        create(1, 4).await;
+        let (connect, tries) = hosts(&[1]);
+        let ring = Ring::new(connect, settings());
+        assert_eq!(ring.naming(), None);
+        assert!(ring.get(4, 1).is_none());
+        let mut tasks = Vec::new();
+        for _ in 0..20 {
+            let ring = ring.clone();
+            tasks.push(spawn(async move { ring.open(4, 1).await }));
+        }
+        let mut opened = Vec::new();
+        for task in tasks {
+            opened.push(task.await.expect("open"));
+        }
+        let label = [&4u32.to_le_bytes()[..], &1u32.to_le_bytes()].concat();
+        let sealed = opened[0].keys.seal(&label, b"mail").expect("seal");
+        for working in &opened {
+            assert_eq!(working.keys.open(&label, &sealed).expect("open"), b"mail");
+            assert_eq!((working.day, working.oldest), (DAY, DAY));
+        }
+        assert_eq!(ring.stats().opens, 1);
+        assert_eq!(tries.get(), 1);
+        assert_eq!(ring.day(), DAY);
+        assert_eq!(Some(*naming(1).await.expose()), ring.naming());
+        let cached = ring.get(4, 1).expect("cached");
+        assert_eq!(cached.keys.open(&label, &sealed).expect("open"), b"mail");
+        assert_eq!(ring.open(9, 1).await.err(), Some(Fault::Unknown));
+        assert_eq!(ring.open(4, 2).await.err(), Some(Fault::Unknown));
+        assert_eq!(ring.stats().unknown, 2);
+    });
+}
+
+#[test]
+fn keys_down_is_unavailable_and_retries_back_off() {
+    let driver = Rc::new(SimDriver::new(2, quiet()));
+    run(&driver, async {
+        let (connect, tries) = hosts(&[9]);
+        let ring = Ring::new(connect, settings());
+        assert_eq!(ring.open(4, 1).await.err(), Some(Fault::Unavailable));
+        assert_eq!(ring.open(4, 1).await.err(), Some(Fault::Unavailable));
+        assert_eq!(tries.get(), 1);
+        sleep(Duration::from_millis(150)).await;
+        assert_eq!(ring.open(4, 1).await.err(), Some(Fault::Unavailable));
+        assert_eq!(tries.get(), 2);
+        sleep(Duration::from_millis(150)).await;
+        assert_eq!(ring.open(4, 1).await.err(), Some(Fault::Unavailable));
+        assert_eq!(tries.get(), 2);
+        assert!(!ring.failed());
+    });
+}
+
+#[test]
+fn a_stuck_connection_times_out_and_the_ring_reconnects() {
+    let driver = Rc::new(SimDriver::new(3, quiet()));
+    run(&driver, async {
+        let _server = serve("/a", 1).await;
+        create(1, 4).await;
+        let closed = fake(2, naming(1).await, Fake::Silent).await;
+        let (connect, tries) = hosts(&[2, 1]);
+        let ring = Ring::new(connect, settings());
+        let begin = rt::now();
+        assert_eq!(ring.open(4, 1).await.err(), Some(Fault::Unavailable));
+        assert!(rt::now().saturating_duration_since(begin) >= Duration::from_secs(1));
+        assert_eq!(ring.stats().drops, 1);
+        sleep(Duration::from_millis(200)).await;
+        assert_eq!(closed.get(), 1);
+        let working = ring.open(4, 1).await.expect("open after reconnect");
+        assert_eq!(working.day, DAY);
+        assert_eq!(tries.get(), 2);
+        assert_eq!(ring.stats().connects, 2);
+    });
+}
+
+#[test]
+fn a_changed_naming_key_fails_the_ring_for_good() {
+    let driver = Rc::new(SimDriver::new(4, quiet()));
+    run(&driver, async {
+        let _first = serve("/a", 1).await;
+        let _second = serve("/b", 3).await;
+        fake(2, naming(1).await, Fake::Closes).await;
+        let (connect, _) = hosts(&[2, 3]);
+        let ring = Ring::new(connect, settings());
+        assert!(ring.open(4, 1).await.is_ok());
+        sleep(Duration::from_millis(200)).await;
+        assert!(ring.get(4, 1).is_some());
+        assert_eq!(ring.open(4, 2).await.err(), Some(Fault::Failed));
+        assert!(ring.failed());
+        assert!(ring.get(4, 1).is_none());
+        assert_eq!(ring.open(4, 1).await.err(), Some(Fault::Failed));
+        sleep(Duration::from_secs(10)).await;
+        assert_eq!(ring.open(4, 1).await.err(), Some(Fault::Failed));
+        assert_eq!(ring.hello().await.err(), Some(Error::Failed));
+        assert!(ring.stats().refused >= 3);
+        assert_eq!(ring.stats().failed, 1);
+    });
+}
+
+#[test]
+fn a_slow_vault_write_keeps_the_connection() {
+    let driver = Rc::new(SimDriver::new(6, quiet()));
+    run(&driver, async {
+        let _server = serve("/a", 1).await;
+        create(1, 4).await;
+        let (connect, tries) = hosts(&[1]);
+        let ring = Ring::new(
+            connect,
+            RingSettings {
+                slow: Duration::from_millis(300),
+                ..settings()
+            },
+        );
+        assert!(ring.open(4, 1).await.is_ok());
+        let slow = Request::Create {
+            tenant: 5,
+            generation: 1,
+            retention: 30,
+        };
+        assert_eq!(ring.call(&slow).await.err(), Some(Error::Timeout));
+        assert_eq!(ring.stats().slow, 1);
+        assert_eq!(ring.stats().drops, 0);
+        let patient = Ring::new(hosts(&[1]).0, settings());
+        let created = patient.call(&slow).await.expect("create");
+        assert!(matches!(created, Answer::Keys { .. }));
+        assert!(ring.open(5, 1).await.is_ok());
+        assert_eq!(tries.get(), 1);
+    });
+}
+
+#[test]
+fn a_dropped_client_closes_its_connection() {
+    let driver = Rc::new(SimDriver::new(7, quiet()));
+    run(&driver, async {
+        let closed = fake(2, Secret::new([1; 16]), Fake::Silent).await;
+        let client = Client::new(TcpStream::connect(at(2)).await.expect("connect"));
+        assert!(matches!(
+            client.call(&Request::Hello { node: 1 }).await,
+            Ok(Answer::Hello { .. })
+        ));
+        drop(client);
+        sleep(Duration::from_millis(200)).await;
+        assert_eq!(closed.get(), 1);
+    });
+}
+
+#[test]
+fn a_cancelled_open_leaves_the_ring_usable() {
+    let driver = Rc::new(SimDriver::new(5, quiet()));
+    run(&driver, async {
+        let _server = serve("/a", 1).await;
+        create(1, 4).await;
+        let silent = TcpListener::bind(at(2), 4).await.expect("bind");
+        let _held = spawn(async move {
+            let mut kept = Vec::new();
+            while let Ok((stream, _)) = silent.accept().await {
+                kept.push(stream);
+            }
+        });
+        let (connect, _) = hosts(&[2, 1]);
+        let ring = Ring::new(connect, settings());
+        let cut = rt::timeout(Duration::from_millis(100), ring.hello()).await;
+        assert!(cut.is_err());
+        sleep(Duration::from_millis(10)).await;
+        ring.hello().await.expect("hello");
+        assert!(ring.open(4, 1).await.is_ok());
+    });
+}
diff --git a/crates/rt/src/executor.rs b/crates/rt/src/executor.rs
index bb2a84e..d68338f 100644
--- a/crates/rt/src/executor.rs
+++ b/crates/rt/src/executor.rs
@@ -205,6 +205,10 @@ pub(crate) fn try_current() -> Option<Rc<Inner>> {
     CURRENT.with(|current| current.borrow().clone())
 }
 
+pub fn in_runtime() -> bool {
+    try_current().is_some()
+}
+
 pub(crate) fn driver() -> Rc<dyn Driver> {
     Rc::clone(&current().driver)
 }
diff --git a/crates/rt/src/lib.rs b/crates/rt/src/lib.rs
index 40f24ba..b90c9fb 100644
--- a/crates/rt/src/lib.rs
+++ b/crates/rt/src/lib.rs
@@ -13,7 +13,7 @@ mod time;
 
 pub use channel::{Receiver, Sender, channel};
 pub use driver::{Completion, Driver, Handle, Op, OpId, OpenOptions, Stat, Unpark};
-pub use executor::{JoinHandle, Runtime, spawn, yield_now};
+pub use executor::{JoinHandle, Runtime, in_runtime, spawn, yield_now};
 pub use file::{File, rename, sync_dir};
 pub use mesh::{Ends, Mesh, Peers};
 #[cfg(any(target_os = "macos", target_os = "linux"))]
diff --git a/crates/sched/Cargo.toml b/crates/sched/Cargo.toml
index 40bf26e..4bccde9 100644
--- a/crates/sched/Cargo.toml
+++ b/crates/sched/Cargo.toml
@@ -10,6 +10,7 @@ publish.workspace = true
 control = { path = "../control" }
 crypto = { path = "../crypto" }
 index = { path = "../index" }
+keys = { path = "../keys" }
 rt = { path = "../rt" }
 spool = { path = "../spool" }
 store = { path = "../store" }
diff --git a/crates/sched/src/intake.rs b/crates/sched/src/intake.rs
index 0ca2df3..a4b3a46 100644
--- a/crates/sched/src/intake.rs
+++ b/crates/sched/src/intake.rs
@@ -136,9 +136,15 @@ impl<T: Transport + 'static> Node<T> {
             generation: draft.generation,
             pool: draft.pool,
         };
-        let Ok(queue) = inner.queues.name(&name).await else {
-            inner.leases.give_back(&draft.tenant, taken);
-            return Err(Refusal::Failed);
+        let queue = match inner.queues.name(&name).await {
+            Ok(queue) => queue,
+            Err(error) => {
+                inner.leases.give_back(&draft.tenant, taken);
+                return Err(match error {
+                    control::Error::Keys { .. } => Refusal::Keys,
+                    _ => Refusal::Failed,
+                });
+            }
         };
         draft.queues.push(queue);
         draft.taken.push(taken);
diff --git a/crates/sched/src/keyring.rs b/crates/sched/src/keyring.rs
new file mode 100644
index 0000000..ad6b5b3
--- /dev/null
+++ b/crates/sched/src/keyring.rs
@@ -0,0 +1,32 @@
+use control::{KeyError, KeySource, Ready};
+use crypto::TenantKeys;
+use keys::{Fault, Ring};
+
+pub struct Keyring(Ring);
+
+impl Keyring {
+    pub fn new(ring: Ring) -> Self {
+        Self(ring)
+    }
+
+    pub fn ring(&self) -> &Ring {
+        &self.0
+    }
+}
+
+impl KeySource for Keyring {
+    fn get(&self, tenant: u32, generation: u32) -> Option<TenantKeys> {
+        self.0.get(tenant, generation).map(|working| working.keys)
+    }
+
+    fn ready(&self, tenant: u32, generation: u32) -> Ready<'_> {
+        Box::pin(async move {
+            match self.0.open(tenant, generation).await {
+                Ok(working) => Ok(working.keys),
+                Err(Fault::Unknown) => Err(KeyError::Unknown),
+                Err(Fault::Unavailable) => Err(KeyError::Unavailable),
+                Err(Fault::Failed) => Err(KeyError::Failed),
+            }
+        })
+    }
+}
diff --git a/crates/sched/src/lib.rs b/crates/sched/src/lib.rs
index cbbf161..e81d58e 100644
--- a/crates/sched/src/lib.rs
+++ b/crates/sched/src/lib.rs
@@ -1,6 +1,7 @@
 mod clock;
 mod error;
 mod intake;
+mod keyring;
 mod lanes;
 mod node;
 mod normal;
@@ -13,6 +14,7 @@ mod transport;
 pub use clock::Clock;
 pub use error::{Error, Refusal};
 pub use intake::{Draft, domain_of};
+pub use keyring::Keyring;
 pub use lanes::{Entry, Lanes, Limits, QueueId, Take, ranges};
 pub use node::{Arrival, Claim, Config, Note, Route, Sched, queue_core};
 pub use normal::{MAX_ADDRESS, normalize, punycode};
diff --git a/crates/sched/tests/index.rs b/crates/sched/tests/index.rs
index f255a09..0c69716 100644
--- a/crates/sched/tests/index.rs
+++ b/crates/sched/tests/index.rs
@@ -1,6 +1,8 @@
 use std::cell::{Cell, RefCell};
 use std::collections::{BTreeMap, BTreeSet};
 use std::future::poll_fn;
+use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+use std::path::Path;
 use std::pin::pin;
 use std::rc::Rc;
 use std::task::Poll;
@@ -9,12 +11,16 @@ use std::time::Duration;
 use control::{KeyError, KeySource, Naming, Object, Pool, Ready, State, Table, Tenant};
 use crypto::{Secret, Tenant as Root, TenantKeys};
 use index::{Entry, Op};
+use keys::proto::{Answer as KeyAnswer, Request};
+use keys::{
+    Audit, Client as KeyClient, Connect, Ring, RingSettings, Server, Settings as KeySettings, Vault,
+};
 use rt::sim::{Config, Disk, Faults, Machine, SimDriver, Stalls, check};
-use rt::{JoinHandle, Runtime, sleep, spawn, timeout};
+use rt::{JoinHandle, Runtime, TcpListener, TcpStream, sleep, spawn, timeout};
 use sched::{
-    Answered, Attempt, Begun, Clock, DAY, Draft, Hash, KIND_REQUEST, KIND_SUPPRESS, Meshes, Node,
-    Parts, Reason, Refusal, Running, SHAPE, Settings, Suppression, Ticket, Transport, normalize,
-    punycode, retain, retain_request,
+    Answered, Attempt, Begun, Clock, DAY, Draft, Hash, KIND_REQUEST, KIND_SUPPRESS, Keyring,
+    Meshes, Node, Parts, Reason, Refusal, Running, SHAPE, Settings, Suppression, Ticket, Transport,
+    normalize, punycode, retain, retain_request,
 };
 use spool::{Body, Head, HeadClear, Message, MessageId, SLICE, Slice};
 use store::{Client, Committer, DataFile, Options, SLOT_SIZE, Setup, Tiering, boot};
@@ -161,7 +167,61 @@ fn started_tuned(
     behind: u64,
     tuning: Option<(u64, usize)>,
 ) -> Cluster {
-    try_started(driver, cores, behind, tuning).expect("start")
+    try_started(driver, cores, behind, tuning, false).expect("start")
+}
+
+fn started_with_keys(driver: &Rc<SimDriver>, cores: u16) -> Cluster {
+    try_started(driver, cores, 0, None, true).expect("start")
+}
+
+const KEYS_DIR: &str = "/keys";
+const KEYS_AT: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 9)), 7000);
+
+fn keys_up(machine: &Machine) {
+    drop(machine.spawn_on(0, async {
+        let dir = Path::new(KEYS_DIR);
+        let vault = match Vault::open(dir).await {
+            Ok(vault) => vault,
+            Err(_) => Vault::init(dir, 20_000).await.expect("init"),
+        };
+        let audit = Audit::open(dir).await.expect("audit");
+        let server = Rc::new(Server::start(vault, audit, KeySettings::default()));
+        let listener = TcpListener::bind(KEYS_AT, 16).await.expect("bind");
+        while let Ok((stream, _)) = listener.accept().await {
+            server.serve(stream);
+        }
+    }));
+}
+
+fn keyring() -> Rc<dyn KeySource> {
+    let connect: Connect =
+        Rc::new(|| Box::pin(async { Ok(KeyClient::new(TcpStream::connect(KEYS_AT).await?)) }));
+    Rc::new(Keyring::new(Ring::new(connect, RingSettings::new(1))))
+}
+
+async fn create_keys(tenant: u32, generation: u32) {
+    let mut tries = 0;
+    let stream = loop {
+        match TcpStream::connect(KEYS_AT).await {
+            Ok(stream) => break stream,
+            Err(_) if tries < 100 => {
+                tries += 1;
+                sleep(Duration::from_millis(10)).await;
+            }
+            Err(error) => panic!("connect: {error}"),
+        }
+    };
+    let client = KeyClient::new(stream);
+    let create = Request::Create {
+        tenant,
+        generation,
+        retention: 30,
+    };
+    assert!(matches!(
+        client.call(&create).await.expect("create"),
+        KeyAnswer::Keys { .. }
+    ));
+    client.close();
 }
 
 fn try_started(
@@ -169,6 +229,7 @@ fn try_started(
     cores: u16,
     behind: u64,
     tuning: Option<(u64, usize)>,
+    real_keys: bool,
 ) -> Option<Cluster> {
     let machine = Machine::new(Rc::clone(driver), usize::from(cores) + 1, Stalls::default());
     let setup = Setup {
@@ -184,6 +245,9 @@ fn try_started(
         .ok()?
         .ok()?;
     let committer = machine.spawn_on(0, booted.committer.run());
+    if real_keys {
+        keys_up(&machine);
+    }
     let meshes = std::sync::Arc::new(Meshes::new(cores));
     let mut tasks = Vec::new();
     for start in booted.starts {
@@ -203,8 +267,9 @@ fn try_started(
                     ..Tiering::default()
                 });
             }
+            let keys: Rc<dyn KeySource> = if real_keys { keyring() } else { Rc::new(Toy) };
             let parts = Parts {
-                keys: Rc::new(Toy),
+                keys,
                 transport: Idle,
                 clock: Clock::at(rt::wall_ms().saturating_sub(behind), rt::now()),
             };
@@ -255,7 +320,7 @@ async fn wait_for(nodes: &[Node<Idle>], tenant: u32, generation: u32, state: Sta
 async fn ready(nodes: &[Node<Idle>]) {
     let pool = Object::Pool(Pool {
         id: 1,
-        addresses: vec![(std::net::IpAddr::from([192, 0, 2, 1]), 1)],
+        addresses: vec![(IpAddr::from([192, 0, 2, 1]), 1)],
     });
     let missing: Vec<Object> = {
         let control = nodes.iter().find(|n| n.controls()).expect("control core");
@@ -544,6 +609,68 @@ fn missing_keys_refuse_for_now_and_never_answer_not_suppressed() {
     cluster.stop();
 }
 
+#[test]
+fn suppressions_and_queues_use_keys_from_the_keys_process() {
+    let driver = Rc::new(SimDriver::with_disk(31, Config::default(), created(31)));
+    let cluster = started_with_keys(&driver, 2);
+    let nodes = cluster.nodes.clone();
+    let queue = cluster.run(async move {
+        create_keys(1, 1).await;
+        create_keys(2, 1).await;
+        ready(&nodes).await;
+        let lists = nodes[0].suppressions();
+        lists
+            .add(1, "kept@example.org", Reason::Manual, None)
+            .await
+            .expect("add");
+        assert_eq!(
+            lists
+                .check(1, 1, &["kept@example.org", "free@example.org"])
+                .await,
+            Ok(vec![true, false])
+        );
+        assert_eq!(
+            lists.check(2, 1, &["kept@example.org"]).await,
+            Ok(vec![false])
+        );
+        let page = lists.list(1, None, 10).await.expect("list");
+        assert_eq!(page.listed[0].address, "kept@example.org");
+        configure(&nodes, vec![tenant(3, 1, State::Active)]).await;
+        wait_for(&nodes, 3, 1, State::Active).await;
+        assert_eq!(
+            lists.check(3, 1, &["a@example.org"]).await,
+            Err(Refusal::Keys)
+        );
+        let mut draft = nodes[0].draft(1, None).expect("draft");
+        nodes[0]
+            .recipient(&mut draft, "to@example.org")
+            .await
+            .expect("recipient")
+    });
+    drop(cluster);
+    let driver = Rc::new(SimDriver::with_disk(32, Config::default(), driver.crash()));
+    let restarted = started_with_keys(&driver, 2);
+    let nodes = restarted.nodes.clone();
+    restarted.run(async move {
+        wait_for(&nodes, 1, 1, State::Active).await;
+        let lists = nodes[1].suppressions();
+        assert_eq!(
+            lists.check(1, 1, &["kept@example.org"]).await,
+            Ok(vec![true])
+        );
+        let page = lists.list(1, None, 10).await.expect("list");
+        assert_eq!(page.listed.len(), 1);
+        assert_eq!(page.listed[0].address, "kept@example.org");
+        let mut draft = nodes[1].draft(1, None).expect("draft");
+        let again = nodes[1]
+            .recipient(&mut draft, "other@example.org")
+            .await
+            .expect("recipient");
+        assert_eq!(again, queue);
+    });
+    restarted.stop();
+}
+
 #[test]
 fn listing_pages_through_every_core_in_hash_order() {
     listing(3, 4);
@@ -1217,7 +1344,7 @@ fn index_session(
     tuning: (u64, usize),
     second: bool,
 ) -> bool {
-    let Some(cluster) = try_started(driver, cores, 0, Some(tuning)) else {
+    let Some(cluster) = try_started(driver, cores, 0, Some(tuning), false) else {
         return false;
     };
     let nodes = cluster.nodes.clone();