diff --git a/Cargo.lock b/Cargo.lock
index 369564eb..89efe6e7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4539,7 +4539,6 @@ dependencies = [
 name = "knot-index"
 version = "2.0.0"
 dependencies = [
- "knot-cache",
  "knot-cob",
  "knot-cobs",
  "knot-git",
@@ -4551,6 +4550,8 @@ dependencies = [
  "serde",
  "tempfile",
  "thiserror 2.0.18",
+ "tokio",
+ "tracing",
 ]
 
 [[package]]
diff --git a/knot2/crates/knot-index/Cargo.toml b/knot2/crates/knot-index/Cargo.toml
index 7dac9d85..fe5502d4 100644
--- a/knot2/crates/knot-index/Cargo.toml
+++ b/knot2/crates/knot-index/Cargo.toml
@@ -10,10 +10,11 @@ knot-types = { workspace = true }
 knot-git = { workspace = true }
 knot-cob = { workspace = true }
 knot-cobs = { workspace = true }
-knot-cache = { workspace = true }
 scc = { workspace = true }
 lasso = { workspace = true }
 thiserror = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
 
 [dev-dependencies]
 knot-runtime = { workspace = true }
diff --git a/knot2/crates/knot-index/src/lib.rs b/knot2/crates/knot-index/src/lib.rs
index 2a455285..617c21ff 100644
--- a/knot2/crates/knot-index/src/lib.rs
+++ b/knot2/crates/knot-index/src/lib.rs
@@ -9,6 +9,7 @@ pub use knot_types::OfferedKey;
 
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
 
 use knot_cob::{ChangePayload, CobStore};
 use knot_cobs::{
@@ -16,9 +17,10 @@ use knot_cobs::{
     MembersCob, RegistryChange, RepoRegistryCob,
 };
 use knot_git::{Layout, Repo};
-use knot_types::{AccountDid, ClonePath, OwnerDid, RepoDid, RepoRkey};
+use knot_types::{AccountDid, ClonePath, OwnerDid, RepoDid, RepoRkey, UnixSeconds};
+use tokio::sync::watch;
 
-use intern::Interner;
+use intern::{Interner, RepoKey};
 use projections::{CollaboratorsProjection, GrantSetProjection, KeyProjection, RegistryProjection};
 
 knot_types::scalar_newtype! {
@@ -34,6 +36,269 @@ pub struct IndexCoverage {
     pub keys: Coverage,
 }
 
+macro_rules! account_list {
+    ($($name:ident),+ $(,)?) => {$(
+        #[derive(Debug, Clone, PartialEq, Eq)]
+        pub struct $name(Vec<AccountDid>);
+
+        impl $name {
+            pub fn new(accounts: Vec<AccountDid>) -> Self {
+                Self(accounts)
+            }
+
+            pub fn len(&self) -> usize {
+                self.0.len()
+            }
+
+            pub fn is_empty(&self) -> bool {
+                self.0.is_empty()
+            }
+
+            pub fn as_slice(&self) -> &[AccountDid] {
+                &self.0
+            }
+
+            pub fn into_vec(self) -> Vec<AccountDid> {
+                self.0
+            }
+        }
+    )+};
+}
+
+account_list!(
+    Pushers,
+    KeptAccounts,
+    StalePushers,
+    SuspectPushers,
+    UnreadMembers,
+    StaleMembers,
+);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum HostedCoverage {
+    Whole,
+    Partial { unread: usize },
+}
+
+impl HostedCoverage {
+    const fn over(unread: usize) -> Self {
+        match unread {
+            0 => Self::Whole,
+            unread => Self::Partial { unread },
+        }
+    }
+}
+
+struct Granted {
+    subjects: Vec<AccountDid>,
+    hosted: HostedCoverage,
+}
+
+enum Folded {
+    Grants(Vec<AccountDid>),
+    Pending,
+    Unreadable,
+}
+
+impl Folded {
+    fn into_grants(self) -> Option<Vec<AccountDid>> {
+        match self {
+            Folded::Grants(subjects) => Some(subjects),
+            Folded::Pending | Folded::Unreadable => None,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct SweepFloor(Duration);
+
+impl SweepFloor {
+    pub const DEFAULT: Self = Self::from_secs(60);
+
+    pub const fn from_secs(secs: u64) -> Self {
+        Self(Duration::from_secs(secs))
+    }
+
+    pub const fn get(self) -> Duration {
+        self.0
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct KeyTtl(Duration);
+
+impl KeyTtl {
+    pub const DEFAULT: Self = Self::from_secs(3_600);
+
+    pub const fn from_secs(secs: u64) -> Self {
+        Self(Duration::from_secs(secs))
+    }
+
+    pub const fn get(self) -> Duration {
+        self.0
+    }
+
+    pub const fn longest(self, other: Self) -> Self {
+        match self.0.as_secs() >= other.0.as_secs() {
+            true => self,
+            false => other,
+        }
+    }
+
+    pub const fn lease_from(self, now: UnixSeconds) -> KeyLease {
+        KeyLease {
+            read_at: now,
+            expires_at: after(now, self.0),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct KeyLease {
+    read_at: UnixSeconds,
+    expires_at: UnixSeconds,
+}
+
+impl KeyLease {
+    pub(crate) const fn is_live(self, now: UnixSeconds) -> bool {
+        self.expires_at.get() > now.get()
+    }
+
+    pub(crate) const fn renewal_due(self, now: UnixSeconds) -> bool {
+        let held = self.expires_at.get().saturating_sub(self.read_at.get());
+        now.get() >= self.read_at.saturating_add_secs(held / 2).get()
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct KeyReprieve {
+    retry: Duration,
+    budget: Duration,
+}
+
+impl KeyReprieve {
+    pub const DEFAULT: Self = Self::from_secs(300, 21_600);
+
+    pub const fn from_secs(retry: u64, budget: u64) -> Self {
+        Self {
+            retry: Duration::from_secs(retry),
+            budget: Duration::from_secs(budget),
+        }
+    }
+
+    pub const fn budgeted_for(self, ttl: KeyTtl) -> Self {
+        match self.budget.as_secs() >= ttl.0.as_secs() {
+            true => self,
+            false => Self {
+                retry: self.retry,
+                budget: ttl.0,
+            },
+        }
+    }
+
+    pub(crate) const fn first_failure(self, now: UnixSeconds) -> KeyLease {
+        KeyLease {
+            read_at: now,
+            expires_at: after(now, self.retry),
+        }
+    }
+
+    pub(crate) const fn extend(self, lease: KeyLease, now: UnixSeconds) -> Option<KeyLease> {
+        let horizon = after(lease.read_at, self.budget);
+        match horizon.get() > now.get() {
+            false => None,
+            true => {
+                let retry = after(now, self.retry);
+                let granted = match retry.get() < horizon.get() {
+                    true => retry,
+                    false => horizon,
+                };
+                Some(KeyLease {
+                    read_at: lease.read_at,
+                    expires_at: match lease.expires_at.get() > granted.get() {
+                        true => lease.expires_at,
+                        false => granted,
+                    },
+                })
+            }
+        }
+    }
+}
+
+pub(crate) const fn whole_secs(span: Duration) -> i64 {
+    let secs = span.as_secs();
+    match secs > i64::MAX as u64 {
+        true => i64::MAX,
+        false => secs as i64,
+    }
+}
+
+const fn after(now: UnixSeconds, span: Duration) -> UnixSeconds {
+    now.saturating_add_secs(whole_secs(span))
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum KeyRecord {
+    Stored,
+    Unheld,
+    Saturated,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum KeyReprieved {
+    Extended,
+    Pending,
+    Exhausted,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct KeyBudget(usize);
+
+impl KeyBudget {
+    pub const DEFAULT: Self = Self::from_mib(64);
+
+    pub const fn from_mib(mib: usize) -> Self {
+        Self(mib * 1024 * 1024)
+    }
+
+    pub const fn from_bytes(bytes: usize) -> Self {
+        Self(bytes)
+    }
+
+    pub const fn get(self) -> usize {
+        self.0
+    }
+}
+
+pub struct MemberWork {
+    pub unread: UnreadMembers,
+    pub due: StaleMembers,
+    pub kept: KeptAccounts,
+}
+
+pub struct KeyWork {
+    pub generation: IndexGeneration,
+    pub tracked: usize,
+    pub hosted: HostedCoverage,
+    pub pushers: Pushers,
+    pub due: StalePushers,
+    pub suspected: SuspectPushers,
+    pub complete: bool,
+    pub members: Resolved<MemberWork>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Recheck {
+    Renewals,
+    Everything,
+}
+
+fn sorted(mut subjects: Vec<AccountDid>) -> Vec<AccountDid> {
+    subjects.sort();
+    subjects.dedup();
+    subjects
+}
+
 pub struct Index {
     meta_path: PathBuf,
     layout: Layout,
@@ -43,11 +308,21 @@ pub struct Index {
     collaborators: CollaboratorsProjection,
     registry: RegistryProjection,
     keys: KeyProjection,
+    unreadable: scc::HashSet<RepoKey>,
     generation: AtomicU64,
+    generations: watch::Sender<IndexGeneration>,
 }
 
 impl Index {
     pub fn new(meta_path: impl Into<PathBuf>, layout: Layout) -> Self {
+        Self::with_key_budget(meta_path, layout, KeyBudget::DEFAULT)
+    }
+
+    pub fn with_key_budget(
+        meta_path: impl Into<PathBuf>,
+        layout: Layout,
+        budget: KeyBudget,
+    ) -> Self {
         Self {
             meta_path: meta_path.into(),
             layout,
@@ -56,8 +331,10 @@ impl Index {
             blocklist: GrantSetProjection::new(),
             collaborators: CollaboratorsProjection::new(),
             registry: RegistryProjection::new(),
-            keys: KeyProjection::new(),
+            keys: KeyProjection::new(budget),
+            unreadable: scc::HashSet::new(),
             generation: AtomicU64::new(0),
+            generations: watch::Sender::new(IndexGeneration::new(0)),
         }
     }
 
@@ -65,8 +342,13 @@ impl Index {
         IndexGeneration(self.generation.load(Ordering::Acquire))
     }
 
+    pub fn generations(&self) -> watch::Receiver<IndexGeneration> {
+        self.generations.subscribe()
+    }
+
     fn bump_generation(&self) {
         self.generation.fetch_add(1, Ordering::Release);
+        self.generations.send_replace(self.generation());
     }
 
     pub fn rebuild(&self) -> Result<(), IndexError> {
@@ -76,10 +358,27 @@ impl Index {
         Ok(())
     }
 
-    pub fn warm_collaborators(&self) {
-        self.hosted_repos().iter().for_each(|repo| {
-            let _ = self.ensure_collaborators(repo);
-        });
+    pub fn warm_collaborators(&self) -> usize {
+        self.hosted_repos()
+            .iter()
+            .filter(|repo| match self.ensure_collaborators(repo) {
+                Ok(()) => false,
+                Err(error) => {
+                    tracing::warn!(
+                        repo = repo.as_str(),
+                        %error,
+                        "collaborators unread, the knot can't open the repo"
+                    );
+                    true
+                }
+            })
+            .count()
+    }
+
+    fn unreadable_repo(&self, repo: &RepoDid) -> bool {
+        self.interner
+            .repo(repo)
+            .is_some_and(|repo| self.unreadable.contains_sync(&repo))
     }
 
     pub fn refresh_members(&self) -> Result<(), IndexError> {
@@ -132,6 +431,7 @@ impl Index {
         evacuated.iter().for_each(|repo| {
             if let Some(key) = self.interner.repo(repo) {
                 self.collaborators.drop_repo(key);
+                self.unreadable.remove_sync(&key);
             }
         });
         self.bump_generation();
@@ -146,9 +446,19 @@ impl Index {
     }
 
     pub fn refresh_collaborators(&self, repo: &RepoDid) -> Result<(), IndexError> {
+        let repo_key = self.interner.intern_repo(repo);
+        self.fold_collaborators(repo, repo_key)
+            .inspect(|()| {
+                self.unreadable.remove_sync(&repo_key);
+            })
+            .inspect_err(|_| {
+                let _ = self.unreadable.insert_sync(repo_key);
+            })
+    }
+
+    fn fold_collaborators(&self, repo: &RepoDid, repo_key: RepoKey) -> Result<(), IndexError> {
         let git = self.layout.open(repo)?;
         let store = CobStore::new(&git);
-        let repo_key = self.interner.intern_repo(repo);
         match store.list::<CollaboratorsCob>()?.as_slice() {
             [] => self.collaborators.mark_repo_empty(repo_key),
             [object] => {
@@ -220,12 +530,76 @@ impl Index {
         self.registry.hosted_repos(&self.interner)
     }
 
-    pub fn owner_of_key(&self, key: &OfferedKey) -> Resolved<Option<AccountDid>> {
-        self.keys.owner(&self.interner, key)
+    pub fn owner_of_key(&self, key: &OfferedKey, now: UnixSeconds) -> Resolved<Option<AccountDid>> {
+        self.keys.owner(&self.interner, key, now)
     }
 
-    pub fn cache_key(&self, key: OfferedKey, did: &AccountDid) {
-        self.keys.cache(&self.interner, key, did);
+    pub fn keys(&self) -> KeySet<'_> {
+        KeySet(self)
+    }
+
+    fn push_grants(&self) -> Resolved<Granted> {
+        match self.registry.coverage() {
+            Coverage::Warming => Resolved::Warming,
+            Coverage::Ready => {
+                let folded: Vec<Folded> = self
+                    .hosted_repos()
+                    .iter()
+                    .map(
+                        |repo| match (self.owner_of(repo), self.collaborators_of(repo)) {
+                            (Resolved::Ready(owner), Resolved::Ready(collaborators)) => {
+                                Folded::Grants(
+                                    owner
+                                        .map(AccountDid::from)
+                                        .into_iter()
+                                        .chain(collaborators)
+                                        .collect(),
+                                )
+                            }
+                            _ if self.unreadable_repo(repo) => Folded::Unreadable,
+                            _ => Folded::Pending,
+                        },
+                    )
+                    .collect();
+                let hosted = HostedCoverage::over(
+                    folded
+                        .iter()
+                        .filter(|repo| matches!(repo, Folded::Pending))
+                        .count(),
+                );
+                Resolved::Ready(Granted {
+                    subjects: sorted(
+                        folded
+                            .into_iter()
+                            .filter_map(Folded::into_grants)
+                            .flatten()
+                            .collect(),
+                    ),
+                    hosted,
+                })
+            }
+        }
+    }
+
+    fn member_grants(&self) -> Resolved<Vec<AccountDid>> {
+        self.member_entries()
+            .map(|entries| sorted(entries.into_iter().map(|grant| grant.subject).collect()))
+    }
+
+    fn due_among(
+        &self,
+        subjects: &[AccountDid],
+        now: UnixSeconds,
+        against: Recheck,
+    ) -> Vec<AccountDid> {
+        subjects
+            .iter()
+            .filter(|did| match against {
+                Recheck::Everything => true,
+                Recheck::Renewals => self.keys.renewal_due(&self.interner, did, now),
+            })
+            .cloned()
+            .collect()
     }
 
     pub fn coverage(&self) -> IndexCoverage {
@@ -234,7 +608,125 @@ impl Index {
             blocklist: self.blocklist.coverage(),
             collaborators: self.collaborators.coverage(),
             registry: self.registry.coverage(),
-            keys: self.keys.coverage(),
+            keys: self.keys.coverage(self.generation()),
         }
     }
 }
+
+pub struct KeySet<'a>(&'a Index);
+
+impl KeySet<'_> {
+    pub fn coverage(&self) -> Coverage {
+        self.0.keys.coverage(self.0.generation())
+    }
+
+    pub fn mark_ready(&self, generation: IndexGeneration) {
+        self.0.keys.mark_ready(generation);
+    }
+
+    pub fn mark_warming(&self) {
+        self.0.keys.mark_warming();
+    }
+
+    pub fn record(&self, did: &AccountDid, keys: Vec<OfferedKey>, lease: KeyLease) -> KeyRecord {
+        self.0.keys.record(&self.0.interner, did, keys, lease)
+    }
+
+    pub fn reprieve(
+        &self,
+        did: &AccountDid,
+        now: UnixSeconds,
+        grace: KeyReprieve,
+        exhausted: KeyLease,
+    ) -> KeyReprieved {
+        self.0
+            .keys
+            .reprieve(&self.0.interner, did, now, grace, exhausted)
+    }
+
+    pub fn retain(&self, kept: &KeptAccounts) {
+        self.0.keys.retain(&self.0.interner, kept.as_slice());
+    }
+
+    pub fn is_fresh(&self, did: &AccountDid, now: UnixSeconds) -> bool {
+        self.0.keys.is_fresh(&self.0.interner, did, now)
+    }
+
+    pub fn publisher_among(
+        &self,
+        candidates: &[AccountDid],
+        key: &OfferedKey,
+        now: UnixSeconds,
+    ) -> Option<AccountDid> {
+        self.0
+            .keys
+            .publisher_among(&self.0.interner, candidates, key, now)
+    }
+
+    pub fn note_miss(&self) {
+        self.0.keys.suspect_now();
+    }
+
+    pub fn any_unheld(&self) -> bool {
+        self.0.keys.any_unheld()
+    }
+
+    pub fn work(&self, now: UnixSeconds, floor: SweepFloor) -> Resolved<KeyWork> {
+        let index = self.0;
+        let generation = index.generation();
+        let Resolved::Ready(granted) = index.push_grants() else {
+            return Resolved::Warming;
+        };
+        let pushers = granted.subjects;
+        let against = match index.keys.take_suspicion(now, floor) {
+            true => Recheck::Everything,
+            false => Recheck::Renewals,
+        };
+        let members = index.member_grants().map(|members| {
+            let outside: Vec<AccountDid> = members
+                .into_iter()
+                .filter(|did| pushers.binary_search(did).is_err())
+                .collect();
+            let (unread, due) = index
+                .due_among(&outside, now, against)
+                .into_iter()
+                .partition(|did| !index.keys.on_file(&index.interner, did));
+            MemberWork {
+                unread: UnreadMembers(unread),
+                due: StaleMembers(due),
+                kept: KeptAccounts(pushers.iter().cloned().chain(outside).collect()),
+            }
+        });
+        let tracked = match &members {
+            Resolved::Ready(work) => work.kept.len(),
+            Resolved::Warming => pushers.len(),
+        };
+        let due = index.due_among(&pushers, now, Recheck::Renewals);
+        let suspected = match against {
+            Recheck::Renewals => Vec::new(),
+            Recheck::Everything => pushers
+                .iter()
+                .filter(|did| !index.keys.renewal_due(&index.interner, did, now))
+                .cloned()
+                .collect(),
+        };
+        let complete = granted.hosted == HostedCoverage::Whole
+            && index.keys.all_live(&index.interner, &pushers, now);
+        Resolved::Ready(KeyWork {
+            generation,
+            tracked,
+            hosted: granted.hosted,
+            pushers: Pushers(pushers),
+            due: StalePushers(due),
+            suspected: SuspectPushers(suspected),
+            complete,
+            members,
+        })
+    }
+
+    pub fn all_live(&self, pushers: &Pushers, now: UnixSeconds) -> bool {
+        self.0
+            .keys
+            .all_live(&self.0.interner, pushers.as_slice(), now)
+    }
+}
diff --git a/knot2/crates/knot-index/src/projections.rs b/knot2/crates/knot-index/src/projections.rs
index 6905a6b5..0cca826c 100644
--- a/knot2/crates/knot-index/src/projections.rs
+++ b/knot2/crates/knot-index/src/projections.rs
@@ -1,9 +1,9 @@
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::hash::{Hash, Hasher};
 use std::marker::PhantomData;
-use std::sync::Mutex;
+use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
 
-use knot_cache::{Cache, EntryCount, Lru};
 use knot_cob::{Change, ChangeId, ChangePayload, Checkpoint, CobId, CobStore, Evaluate};
 use knot_cobs::{
     CollaboratorsChange, CollaboratorsCob, Grant, GrantChange, Registration, Registry,
@@ -14,8 +14,9 @@ use knot_types::{AccountDid, ClonePath, OfferedKey, OwnerDid, RepoDid, RepoRkey,
 use crate::coverage::{Coverage, CoverageCell, Resolved};
 use crate::error::IndexError;
 use crate::intern::{AccountKey, Interner, NameKey, OwnerKey, RepoKey, RkeyKey};
-
-const KEY_CACHE_CAPACITY: usize = 16_384;
+use crate::{
+    IndexGeneration, KeyBudget, KeyLease, KeyRecord, KeyReprieve, KeyReprieved, SweepFloor,
+};
 
 #[derive(Debug, Clone, Copy)]
 struct Provenance {
@@ -818,34 +819,437 @@ impl RegistryProjection {
     }
 }
 
+enum Reading {
+    Published(Vec<Arc<OfferedKey>>),
+    Unheld,
+    Unread,
+}
+
+impl Reading {
+    fn keys(&self) -> &[Arc<OfferedKey>] {
+        match self {
+            Reading::Published(keys) => keys,
+            Reading::Unheld | Reading::Unread => &[],
+        }
+    }
+
+    fn is_published(&self) -> bool {
+        matches!(self, Reading::Published(_))
+    }
+}
+
+struct Held {
+    reading: Reading,
+    lease: KeyLease,
+}
+
+const PER_KEY_OVERHEAD: usize = 192;
+
+const PER_ACCOUNT_OVERHEAD: usize = 128;
+
+impl Held {
+    fn bytes(&self) -> usize {
+        PER_ACCOUNT_OVERHEAD
+            + self
+                .reading
+                .keys()
+                .iter()
+                .map(|key| key.as_bytes().len() + PER_KEY_OVERHEAD)
+                .sum::<usize>()
+    }
+
+    fn answers(&self, now: UnixSeconds) -> bool {
+        self.reading.is_published() && self.lease.is_live(now)
+    }
+
+    fn settled(&self, now: UnixSeconds) -> bool {
+        !matches!(self.reading, Reading::Unread) && self.lease.is_live(now)
+    }
+
+    fn renewal_due(&self, now: UnixSeconds) -> bool {
+        match self.reading {
+            Reading::Published(_) => self.lease.renewal_due(now),
+            Reading::Unheld | Reading::Unread => !self.lease.is_live(now),
+        }
+    }
+
+    fn publishes(&self, key: &OfferedKey, now: UnixSeconds) -> bool {
+        self.answers(now)
+            && self
+                .reading
+                .keys()
+                .iter()
+                .any(|published| published.as_ref() == key)
+    }
+
+    fn is_unheld(&self) -> bool {
+        matches!(self.reading, Reading::Unheld)
+    }
+}
+
+#[derive(Default)]
+struct Publishers(Vec<AccountKey>);
+
+impl Publishers {
+    fn add(&mut self, account: AccountKey) {
+        if let Err(at) = self.0.binary_search(&account) {
+            self.0.insert(at, account);
+        }
+    }
+
+    fn remove(&mut self, account: AccountKey) -> bool {
+        if let Ok(at) = self.0.binary_search(&account) {
+            self.0.remove(at);
+        }
+        self.0.is_empty()
+    }
+
+    fn to_vec(&self) -> Vec<AccountKey> {
+        self.0.clone()
+    }
+}
+
+const NEVER: u64 = u64::MAX;
+
 pub(crate) struct KeyProjection {
-    cache: Lru<OfferedKey, AccountKey>,
+    owners: scc::HashMap<Arc<OfferedKey>, Publishers>,
+    held: scc::HashMap<AccountKey, Held>,
+    budget: KeyBudget,
+    tracked_bytes: AtomicUsize,
+    unheld: AtomicUsize,
+    suspect: AtomicBool,
+    swept_at: AtomicI64,
+    ready_at: AtomicU64,
 }
 
 impl KeyProjection {
-    pub(crate) fn new() -> Self {
+    pub(crate) fn new(budget: KeyBudget) -> Self {
         Self {
-            cache: Lru::by_count(EntryCount::new(KEY_CACHE_CAPACITY as u64)),
+            owners: scc::HashMap::new(),
+            held: scc::HashMap::new(),
+            budget,
+            tracked_bytes: AtomicUsize::new(0),
+            unheld: AtomicUsize::new(0),
+            suspect: AtomicBool::new(false),
+            swept_at: AtomicI64::new(i64::MIN),
+            ready_at: AtomicU64::new(NEVER),
         }
     }
 
-    pub(crate) fn coverage(&self) -> Coverage {
-        Coverage::Ready
+    pub(crate) fn coverage(&self, generation: IndexGeneration) -> Coverage {
+        match self.ready_at.load(Ordering::Acquire) == generation.get() {
+            true => Coverage::Ready,
+            false => Coverage::Warming,
+        }
+    }
+
+    pub(crate) fn suspect_now(&self) {
+        self.suspect.store(true, Ordering::Release);
+    }
+
+    pub(crate) fn take_suspicion(&self, now: UnixSeconds, floor: SweepFloor) -> bool {
+        let held_until = |swept: i64| swept.saturating_add(crate::whole_secs(floor.get()));
+        match self.suspect.load(Ordering::Acquire) {
+            false => false,
+            true => {
+                self.swept_at
+                    .fetch_update(Ordering::AcqRel, Ordering::Acquire, |swept| {
+                        (now.get() >= held_until(swept)).then_some(now.get())
+                    })
+                    .is_ok()
+                    && self.suspect.swap(false, Ordering::AcqRel)
+            }
+        }
     }
 
-    pub(crate) fn cache(&self, interner: &Interner, key: OfferedKey, did: &AccountDid) {
-        self.cache.insert(key, interner.intern_account(did));
+    pub(crate) fn mark_ready(&self, generation: IndexGeneration) {
+        self.ready_at.store(generation.get(), Ordering::Release);
+    }
+
+    pub(crate) fn mark_warming(&self) {
+        self.ready_at.store(NEVER, Ordering::Release);
     }
 
     pub(crate) fn owner(
         &self,
         interner: &Interner,
         key: &OfferedKey,
+        now: UnixSeconds,
     ) -> Resolved<Option<AccountDid>> {
+        let publishers = self
+            .owners
+            .read_sync(key, |_, publishers| publishers.to_vec())
+            .unwrap_or_default();
         Resolved::Ready(
-            self.cache
-                .get(key)
+            publishers
+                .into_iter()
+                .find(|account| {
+                    self.held_by(*account, |held| held.publishes(key, now)) == Some(true)
+                })
                 .map(|account| interner.resolve_account(account)),
         )
     }
+
+    pub(crate) fn any_unheld(&self) -> bool {
+        self.unheld.load(Ordering::Acquire) > 0
+    }
+
+    fn track_unheld(&self, lost: bool, gained: bool) {
+        match (lost, gained) {
+            (false, true) => {
+                self.unheld.fetch_add(1, Ordering::AcqRel);
+            }
+            (true, false) => {
+                self.unheld.fetch_sub(1, Ordering::AcqRel);
+            }
+            (true, true) | (false, false) => {}
+        }
+    }
+
+    pub(crate) fn record(
+        &self,
+        interner: &Interner,
+        did: &AccountDid,
+        keys: Vec<OfferedKey>,
+        lease: KeyLease,
+    ) -> KeyRecord {
+        let account = interner.intern_account(did);
+        let incoming = Held {
+            reading: Reading::Published(keys.into_iter().map(Arc::new).collect()),
+            lease,
+        };
+        let entry = self.held.entry_sync(account);
+        let held = match &entry {
+            scc::hash_map::Entry::Occupied(slot) => slot.get().bytes(),
+            scc::hash_map::Entry::Vacant(_) => 0,
+        };
+        if self.reserve_bytes(incoming.bytes() as isize - held as isize) {
+            self.settle(entry, account, incoming);
+            return KeyRecord::Stored;
+        }
+        let unheld = Held {
+            reading: Reading::Unheld,
+            lease,
+        };
+        if !self.reserve_bytes(unheld.bytes() as isize - held as isize) {
+            return KeyRecord::Saturated;
+        }
+        self.settle(entry, account, unheld);
+        KeyRecord::Unheld
+    }
+
+    fn settle(
+        &self,
+        entry: scc::hash_map::Entry<'_, AccountKey, Held>,
+        account: AccountKey,
+        incoming: Held,
+    ) {
+        let published: HashSet<Arc<OfferedKey>> = incoming.reading.keys().iter().cloned().collect();
+        let gained = incoming.is_unheld();
+        match entry {
+            scc::hash_map::Entry::Occupied(mut slot) => {
+                let replaced = slot.insert(incoming);
+                self.track_unheld(replaced.is_unheld(), gained);
+                self.rewire(account, &published, replaced.reading.keys());
+            }
+            scc::hash_map::Entry::Vacant(slot) => {
+                let _locked = slot.insert_entry(incoming);
+                self.track_unheld(false, gained);
+                self.rewire(account, &published, &[]);
+            }
+        }
+    }
+
+    fn rewire(
+        &self,
+        account: AccountKey,
+        published: &HashSet<Arc<OfferedKey>>,
+        replaced: &[Arc<OfferedKey>],
+    ) {
+        published
+            .iter()
+            .for_each(|key| self.claim(Arc::clone(key), account));
+        replaced
+            .iter()
+            .filter(|key| !published.contains(*key))
+            .for_each(|key| self.disown(key, account));
+    }
+
+    pub(crate) fn reprieve(
+        &self,
+        interner: &Interner,
+        did: &AccountDid,
+        now: UnixSeconds,
+        grace: KeyReprieve,
+        exhausted: KeyLease,
+    ) -> KeyReprieved {
+        let account = interner.intern_account(did);
+        let mut slot = match self.held.entry_sync(account) {
+            scc::hash_map::Entry::Vacant(slot) => {
+                let pending = Held {
+                    reading: Reading::Unread,
+                    lease: grace.first_failure(now),
+                };
+                if self.reserve_bytes(pending.bytes() as isize) {
+                    let _locked = slot.insert_entry(pending);
+                }
+                return KeyReprieved::Pending;
+            }
+            scc::hash_map::Entry::Occupied(slot) => slot,
+        };
+        match grace.extend(slot.get().lease, now) {
+            Some(lease) => {
+                let carried = slot.get().reading.is_published();
+                slot.get_mut().lease = lease;
+                match carried {
+                    true => KeyReprieved::Extended,
+                    false => KeyReprieved::Pending,
+                }
+            }
+            None => {
+                let given_up = Held {
+                    reading: Reading::Published(Vec::new()),
+                    lease: exhausted,
+                };
+                let _ = self.reserve_bytes(given_up.bytes() as isize - slot.get().bytes() as isize);
+                self.track_unheld(slot.get().is_unheld(), false);
+                self.disown_all(&slot, account);
+                *slot.get_mut() = given_up;
+                KeyReprieved::Exhausted
+            }
+        }
+    }
+
+    pub(crate) fn on_file(&self, interner: &Interner, did: &AccountDid) -> bool {
+        interner
+            .account(did)
+            .is_some_and(|account| self.held.contains_sync(&account))
+    }
+
+    fn held_by(&self, account: AccountKey, ready: impl Fn(&Held) -> bool) -> Option<bool> {
+        self.held.read_sync(&account, |_, held| ready(held))
+    }
+
+    fn held_for(
+        &self,
+        interner: &Interner,
+        did: &AccountDid,
+        ready: impl Fn(&Held) -> bool,
+    ) -> Option<bool> {
+        interner
+            .account(did)
+            .and_then(|account| self.held_by(account, ready))
+    }
+
+    pub(crate) fn is_fresh(&self, interner: &Interner, did: &AccountDid, now: UnixSeconds) -> bool {
+        self.held_for(interner, did, |held| held.answers(now))
+            .unwrap_or(false)
+    }
+
+    pub(crate) fn renewal_due(
+        &self,
+        interner: &Interner,
+        did: &AccountDid,
+        now: UnixSeconds,
+    ) -> bool {
+        self.held_for(interner, did, |held| held.renewal_due(now))
+            .unwrap_or(true)
+    }
+
+    pub(crate) fn all_live(
+        &self,
+        interner: &Interner,
+        subjects: &[AccountDid],
+        now: UnixSeconds,
+    ) -> bool {
+        subjects.iter().all(|did| {
+            self.held_for(interner, did, |held| held.settled(now))
+                .unwrap_or(false)
+        })
+    }
+
+    pub(crate) fn publisher_among(
+        &self,
+        interner: &Interner,
+        candidates: &[AccountDid],
+        key: &OfferedKey,
+        now: UnixSeconds,
+    ) -> Option<AccountDid> {
+        candidates
+            .iter()
+            .find(|did| {
+                self.held_for(interner, did, |held| held.publishes(key, now))
+                    .unwrap_or(false)
+            })
+            .cloned()
+    }
+
+    pub(crate) fn retain(&self, interner: &Interner, kept: &[AccountDid]) {
+        let keep: BTreeSet<AccountKey> = kept
+            .iter()
+            .filter_map(|did| interner.account(did))
+            .collect();
+        let mut released = Vec::new();
+        self.held.iter_sync(|account, _| {
+            if !keep.contains(account) {
+                released.push(*account);
+            }
+            true
+        });
+        released
+            .into_iter()
+            .for_each(|account| self.release(account));
+    }
+
+    fn release(&self, account: AccountKey) {
+        if let scc::hash_map::Entry::Occupied(slot) = self.held.entry_sync(account) {
+            self.evict(&slot, account);
+            let _ = slot.remove();
+        }
+    }
+
+    fn evict(
+        &self,
+        slot: &scc::hash_map::OccupiedEntry<'_, AccountKey, Held>,
+        account: AccountKey,
+    ) {
+        self.track_unheld(slot.get().is_unheld(), false);
+        self.disown_all(slot, account);
+        let _ = self.reserve_bytes(-(slot.get().bytes() as isize));
+    }
+
+    fn disown_all(
+        &self,
+        slot: &scc::hash_map::OccupiedEntry<'_, AccountKey, Held>,
+        account: AccountKey,
+    ) {
+        slot.get()
+            .reading
+            .keys()
+            .iter()
+            .for_each(|key| self.disown(key, account));
+    }
+
+    fn reserve_bytes(&self, growth: isize) -> bool {
+        self.tracked_bytes
+            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |tracked| {
+                let next = tracked.saturating_add_signed(growth);
+                (growth <= 0 || next <= self.budget.get()).then_some(next)
+            })
+            .is_ok()
+    }
+
+    fn claim(&self, key: Arc<OfferedKey>, account: AccountKey) {
+        self.owners
+            .entry_sync(key)
+            .or_default()
+            .get_mut()
+            .add(account);
+    }
+
+    fn disown(&self, key: &OfferedKey, account: AccountKey) {
+        let _ = self
+            .owners
+            .remove_if_sync(key, |publishers| publishers.remove(account));
+    }
 }
diff --git a/knot2/crates/knot-index/tests/lifecycle.rs b/knot2/crates/knot-index/tests/lifecycle.rs
index a3670ad9..f493eb11 100644
--- a/knot2/crates/knot-index/tests/lifecycle.rs
+++ b/knot2/crates/knot-index/tests/lifecycle.rs
@@ -53,7 +53,7 @@ fn rebuild_folds_members_and_registry_and_collaborators_fold_on_access() {
             blocklist: Coverage::Ready,
             collaborators: Coverage::Ready,
             registry: Coverage::Ready,
-            keys: Coverage::Ready,
+            keys: Coverage::Warming,
         }
     );
 }
@@ -88,9 +88,9 @@ fn every_accessor_fails_closed_while_warming() {
     assert_eq!(index.coverage().members, Coverage::Warming);
 
     assert_eq!(
-        index.owner_of_key(&OfferedKey::from_bytes(vec![1, 2, 3])),
+        index.owner_of_key(&OfferedKey::from_bytes(vec![1, 2, 3]), at(0)),
         Resolved::Ready(None),
-        "key cache is operational from boot, never warming"
+        "a key lookup answers from the first request, even while the key set is warming"
     );
 }
 
diff --git a/knot2/crates/knot-index/tests/projections.rs b/knot2/crates/knot-index/tests/projections.rs
index 5c043877..3215ef3f 100644
--- a/knot2/crates/knot-index/tests/projections.rs
+++ b/knot2/crates/knot-index/tests/projections.rs
@@ -4,7 +4,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use knot_cob::{ChangePayload, CobHome, CobId, CobStore};
 use knot_cobs::{CollaboratorsChange, MembersChange, RegistryChange, Removal, Rename, RepoRef};
 use knot_git::{RefUpdate, Repo};
-use knot_index::{Coverage, IndexError, OfferedKey, Resolved};
+use knot_index::{Coverage, IndexError, Resolved};
 use knot_types::{ClonePath, RefName, RepoName};
 use serde::{Deserialize, Serialize};
 
@@ -538,36 +538,6 @@ fn a_repo_moved_within_one_delta_is_not_evacuated() {
     );
 }
 
-#[test]
-fn key_cache_evicts_least_recently_used() {
-    const CAP: u32 = 16_384;
-    let world = World::new();
-    let index = world.index();
-    let key = |i: u32| OfferedKey::from_bytes(i.to_le_bytes().to_vec());
-
-    (0..CAP).for_each(|i| index.cache_key(key(i), &acc("nel")));
-    assert_eq!(
-        index.owner_of_key(&key(0)),
-        Resolved::Ready(Some(acc("nel")))
-    );
-    index.cache_key(key(CAP), &acc("nel"));
-
-    assert_eq!(
-        index.owner_of_key(&key(1)),
-        Resolved::Ready(None),
-        "least-recently-used key is evicted"
-    );
-    assert_eq!(
-        index.owner_of_key(&key(0)),
-        Resolved::Ready(Some(acc("nel"))),
-        "recently-used key survives despite being inserted first"
-    );
-    assert_eq!(
-        index.owner_of_key(&key(CAP)),
-        Resolved::Ready(Some(acc("nel")))
-    );
-}
-
 fn path(raw: &str) -> ClonePath {
     ClonePath::parse(raw).unwrap()
 }
diff --git a/knot2/crates/knot-ssh/src/exec.rs b/knot2/crates/knot-ssh/src/exec.rs
index baf0e1c0..d559e297 100644
--- a/knot2/crates/knot-ssh/src/exec.rs
+++ b/knot2/crates/knot-ssh/src/exec.rs
@@ -805,7 +805,8 @@ async fn resolve_pusher<H: HttpTransport, C: Clock>(
         _ => Vec::new(),
     };
     let candidates: Vec<AccountDid> = owner.into_iter().chain(collaborators).collect();
-    if let Resolved::Ready(Some(cached)) = state.index.owner_of_key(key)
+    let now = state.atproto.now().seconds();
+    if let Resolved::Ready(Some(cached)) = state.index.owner_of_key(key, now)
         && candidates.contains(&cached)
     {
         return Some(cached);
@@ -813,8 +814,6 @@ async fn resolve_pusher<H: HttpTransport, C: Clock>(
     let _permit = state.slots.resolve.acquire().await;
     let matches = futures::stream::iter(candidates).filter_map(|did| async move {
         let keys = state.atproto.resolve_pubkeys(&did).await.ok()?;
-        keys.iter()
-            .for_each(|resolved| state.index.cache_key(resolved.clone(), &did));
         keys.iter().any(|resolved| resolved == key).then_some(did)
     });
     futures::pin_mut!(matches);
diff --git a/knot2/crates/knot-ssh/tests/ssh_push.rs b/knot2/crates/knot-ssh/tests/ssh_push.rs
index 6062be1c..50195783 100644
--- a/knot2/crates/knot-ssh/tests/ssh_push.rs
+++ b/knot2/crates/knot-ssh/tests/ssh_push.rs
@@ -1034,9 +1034,10 @@ async fn key_recognition_edge_cases() {
     .unwrap()
     .to_bytes()
     .unwrap();
-    fx.index.cache_key(
-        knot_types::OfferedKey::from_bytes(blob),
+    fx.index.keys().record(
         &AccountDid::new("did:plc:whelk").unwrap(),
+        vec![knot_types::OfferedKey::from_bytes(blob)],
+        knot_index::KeyTtl::from_secs(u32::MAX.into()).lease_from(knot_types::UnixSeconds::new(0)),
     );
     let (ok, out) = push(
         &fx.work,
diff --git a/knot2/crates/knot-types/src/ids.rs b/knot2/crates/knot-types/src/ids.rs
index 2ae9efcb..316f84c7 100644
--- a/knot2/crates/knot-types/src/ids.rs
+++ b/knot2/crates/knot-types/src/ids.rs
@@ -434,7 +434,7 @@ impl fmt::Display for CiLogsAddr {
     }
 }
 
-#[derive(Clone)]
+#[derive(Clone, PartialEq, Eq)]
 pub enum OwnerRef {
     Did(OwnerDid),
     Handle(Handle),
@@ -762,6 +762,10 @@ impl UnixMicros {
     pub const fn next(self) -> Self {
         Self(self.0.saturating_add(1))
     }
+
+    pub const fn seconds(self) -> UnixSeconds {
+        UnixSeconds((self.0 / 1_000_000) as i64)
+    }
 }
 
 impl fmt::Display for UnixSeconds {
@@ -1350,6 +1354,13 @@ mod tests {
         assert_eq!(base.to_string(), "1000");
     }
 
+    #[test]
+    fn micros_truncate_to_the_second_they_fall_in() {
+        assert_eq!(UnixMicros::new(1_999_999).seconds(), UnixSeconds::new(1));
+        assert_eq!(UnixMicros::new(2_000_000).seconds(), UnixSeconds::new(2));
+        assert_eq!(UnixMicros::new(0).seconds(), UnixSeconds::new(0));
+    }
+
     #[test]
     fn http_status_classifies_transient_codes() {
         assert!(HttpStatus::new(200).is_success());