fix(node): look up owned shard filters by address, not by filter

crates/quil-node/src/master_node/grpc.rs

@@ -28,6 +28,42 @@ async fn fetch_remote_app_shards(
     out
 }
 
+/// The filters this prover currently owns, for `GetShardInfo`'s
+/// `include_all == false` view.
+///
+/// The registry has two lookups and both take a bare `&[u8]`: `get_provers` is
+/// keyed by *confirmation filter*, `get_prover_info` by *address*. This site
+/// asked `get_provers` for the local address. An address is never a filter key
+/// — `SharedProverRegistry::get_provers` misses `filter_cache` and returns
+/// empty — so the owned set was always empty and `include_all == false`
+/// filtered every shard away: `qclient node prover shards` printed "No
+/// allocated shards" on a node whose `node prover status` listed twenty-seven.
+/// `GetNodeInfo` builds the same view from `get_prover_info`, and this matches
+/// its liveness predicate so the two surfaces agree allocation for allocation.
+fn owned_filters(
+    registry: &dyn quil_types::consensus::ProverRegistry,
+    address: &[u8],
+    frame_number: u64,
+) -> std::collections::HashSet<Vec<u8>> {
+    let Ok(Some(info)) = registry.get_prover_info(address) else {
+        return std::collections::HashSet::new();
+    };
+    info.allocations
+        .iter()
+        .filter(|a| {
+            // Same predicate as `GetNodeInfo`. `ExpiredEpoch` is an Active
+            // data-shard allocation that missed this epoch's re-confirm: not
+            // live, but not lost either — it comes back the moment the prover
+            // re-registers. `status` shows it as `re-confirm!`, so dropping it
+            // here would put the two surfaces back into disagreement over
+            // exactly the allocation the operator most needs to act on.
+            let eff = a.effective_status(frame_number);
+            eff.is_live() || eff == quil_types::consensus::EffectiveStatus::ExpiredEpoch
+        })
+        .map(|a| a.confirmation_filter.clone())
+        .collect()
+}
+
 pub(crate) struct GrpcArgs {
     pub config: quil_config::Config,
     pub network: u8,
@@ -1015,12 +1051,8 @@ pub(crate) fn spawn_all(
                 }
                 Err(_) => (0u64, cf),
             };
-            let provers = self.registry.get_provers(&self.self_address).unwrap_or_default();
-            let allocated_filters: std::collections::HashSet<Vec<u8>> = provers
-                .iter()
-                .filter(|pr| pr.address == self.self_address)
-                .flat_map(|pr| pr.allocations.iter().filter(|a| a.is_live(frame_number)).map(|a| a.confirmation_filter.clone()))
-                .collect();
+            let allocated_filters =
+                owned_filters(self.registry.as_ref(), &self.self_address, frame_number);
             let local_get_sizes = quil_engine::shard_info::local_app_shard_get_sizes(self.crdt.clone(), self.shards_store.clone());
             let local_result = quil_engine::shard_info::get_shard_info(
                 include_all, &self.self_address, &allocated_filters, difficulty, frame_number,
@@ -1686,3 +1718,168 @@ pub(crate) fn spawn_all(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use quil_types::consensus::{
+        EffectiveStatus, ProverAllocationInfo, ProverInfo, ProverRegistry, ProverShardSummary,
+        ProverStatus,
+    };
+    use quil_types::error::Result;
+
+    const ADDRESS: [u8; 32] = [7u8; 32];
+    const FILTER: [u8; 4] = [1, 2, 3, 4];
+
+    /// A registry that keys `get_provers` by filter, the way the production
+    /// `SharedProverRegistry` does.
+    ///
+    /// `quil_engine::test_support::TestProverRegistry` returns every prover for
+    /// any filter, which is exactly why an address handed to a filter-keyed
+    /// lookup passed unnoticed: under that stub the buggy call and the correct
+    /// one are indistinguishable.
+    struct FilterKeyedRegistry {
+        provers: Vec<ProverInfo>,
+    }
+
+    impl FilterKeyedRegistry {
+        fn under(&self, filter: &[u8]) -> Vec<ProverInfo> {
+            self.provers
+                .iter()
+                .filter(|p| {
+                    p.allocations
+                        .iter()
+                        .any(|a| a.confirmation_filter == filter)
+                })
+                .cloned()
+                .collect()
+        }
+    }
+
+    impl ProverRegistry for FilterKeyedRegistry {
+        fn get_prover_info(&self, address: &[u8]) -> Result<Option<ProverInfo>> {
+            Ok(self.provers.iter().find(|p| p.address == address).cloned())
+        }
+        fn get_provers(&self, filter: &[u8]) -> Result<Vec<ProverInfo>> {
+            Ok(self.under(filter))
+        }
+        fn get_provers_by_status(
+            &self,
+            filter: &[u8],
+            status: ProverStatus,
+        ) -> Result<Vec<ProverInfo>> {
+            Ok(self
+                .under(filter)
+                .into_iter()
+                .filter(|p| p.status == status)
+                .collect())
+        }
+        fn get_active_provers(&self, filter: &[u8], _frame: u64) -> Result<Vec<ProverInfo>> {
+            Ok(self.under(filter))
+        }
+        fn get_prover_count(&self, filter: &[u8]) -> Result<usize> {
+            Ok(self.under(filter).len())
+        }
+        fn get_next_prover(&self, _input: &[u8; 32], _filter: &[u8], _frame: u64) -> Result<Vec<u8>> {
+            Ok(Vec::new())
+        }
+        fn get_ordered_provers(
+            &self,
+            _input: &[u8; 32],
+            _filter: &[u8],
+            _frame: u64,
+        ) -> Result<Vec<Vec<u8>>> {
+            Ok(Vec::new())
+        }
+        fn get_prover_shard_summaries(&self, _frame: u64) -> Result<Vec<ProverShardSummary>> {
+            Ok(Vec::new())
+        }
+    }
+
+    fn allocation(filter: &[u8], status: ProverStatus, join_frame: u64) -> ProverAllocationInfo {
+        ProverAllocationInfo {
+            status,
+            confirmation_filter: filter.to_vec(),
+            rejection_filter: Vec::new(),
+            join_frame_number: join_frame,
+            leave_frame_number: 0,
+            pause_frame_number: 0,
+            resume_frame_number: 0,
+            kick_frame_number: 0,
+            join_confirm_frame_number: join_frame + 1,
+            join_reject_frame_number: 0,
+            leave_confirm_frame_number: 0,
+            leave_reject_frame_number: 0,
+            last_active_frame_number: join_frame,
+            epoch: 0,
+            ring: 0,
+            vertex_address: Vec::new(),
+        }
+    }
+
+    fn registry(allocs: Vec<ProverAllocationInfo>) -> FilterKeyedRegistry {
+        FilterKeyedRegistry {
+            provers: vec![ProverInfo {
+                public_key: Vec::new(),
+                address: ADDRESS.to_vec(),
+                status: ProverStatus::Active,
+                kick_frame_number: 0,
+                allocations: allocs,
+                available_storage: 1 << 30,
+                seniority: 0,
+                delegate_address: Vec::new(),
+            }],
+        }
+    }
+
+    /// The regression: the local prover holds one Active allocation, and the
+    /// owned set has to contain its filter. Against the old
+    /// `get_provers(&self_address)` this is empty — an address is not a filter
+    /// key — and `GetShardInfo(include_all: false)` returns nothing.
+    #[test]
+    fn an_active_allocation_is_owned() {
+        let reg = registry(vec![allocation(&FILTER, ProverStatus::Active, 100)]);
+        let owned = owned_filters(&reg, &ADDRESS, 200);
+        assert_eq!(owned.len(), 1, "the prover's own allocation went missing");
+        assert!(owned.contains(&FILTER.to_vec()));
+    }
+
+    /// Only live allocations count: a Kicked slot is not owned, and neither is
+    /// a Joining one that ran past its grace window without being confirmed.
+    #[test]
+    fn dead_allocations_are_not_owned() {
+        let stale = [9u8; 4];
+        let reg = registry(vec![
+            allocation(&FILTER, ProverStatus::Kicked, 100),
+            allocation(&stale, ProverStatus::Joining, 100),
+        ]);
+        // Frame 5000 is epoch 6; a join proposed in epoch 0 had to be
+        // confirmed in epoch 1, so it reads as implicitly rejected.
+        assert!(owned_filters(&reg, &ADDRESS, 5_000).is_empty());
+    }
+
+    /// A stale-epoch allocation is still owned. It is an Active data-shard
+    /// allocation that missed this epoch's re-confirm — recoverable, and shown
+    /// by `node prover status` as `re-confirm!`. Dropping it here would put
+    /// `shards` and `status` back into disagreement over precisely the
+    /// allocation the operator most needs to act on.
+    #[test]
+    fn a_stale_epoch_allocation_is_still_owned() {
+        let reg = registry(vec![allocation(&FILTER, ProverStatus::Active, 100)]);
+        assert_eq!(
+            reg.provers[0].allocations[0].effective_status(5_000),
+            EffectiveStatus::ExpiredEpoch,
+            "fixture no longer produces the state under test"
+        );
+        assert!(owned_filters(&reg, &ADDRESS, 5_000).contains(&FILTER.to_vec()));
+    }
+
+    /// An address the registry has never seen yields nothing rather than
+    /// erroring; a node that is not a prover simply owns no filters.
+    #[test]
+    fn an_unknown_address_owns_nothing() {
+        let reg = registry(vec![allocation(&FILTER, ProverStatus::Active, 100)]);
+        assert!(owned_filters(&reg, &[0u8; 32], 200).is_empty());
+    }
+}
+