commonware_resolver/p2p/
ingress.rs

1use crate::{ingress, Fetch, Resolver, TargetedResolver};
2use commonware_actor::{mailbox::Sender, Feedback};
3use commonware_cryptography::PublicKey;
4use commonware_utils::{vec::NonEmptyVec, Span};
5
6/// A key to fetch data for, optionally with target peers.
7pub type FetchKey<K, P, S> = ingress::FetchKey<K, S, Option<NonEmptyVec<P>>>;
8
9/// Messages that can be sent to the peer actor.
10pub type Message<K, P, S> = ingress::Message<K, S, Option<NonEmptyVec<P>>>;
11
12fn fetch_key<K, P, S>(fetch: Fetch<K, S>, targets: Option<NonEmptyVec<P>>) -> FetchKey<K, P, S> {
13    FetchKey {
14        key: fetch.key,
15        subscribers: NonEmptyVec::new((fetch.subscriber, fetch.span)),
16        metadata: targets,
17    }
18}
19
20/// A way to send messages to the peer actor.
21#[derive(Clone)]
22pub struct Mailbox<K: Span, P: Eq, S: Eq = ()> {
23    /// The channel that delivers messages to the peer actor.
24    sender: Sender<Message<K, P, S>>,
25}
26
27impl<K: Span, P: Eq, S: Eq> Mailbox<K, P, S> {
28    /// Create a new mailbox.
29    pub(super) const fn new(sender: Sender<Message<K, P, S>>) -> Self {
30        Self { sender }
31    }
32}
33
34impl<K, P, S> Resolver for Mailbox<K, P, S>
35where
36    K: Span,
37    P: PublicKey,
38    S: Clone + Eq + Send + 'static,
39{
40    type Key = K;
41    type Subscriber = S;
42
43    /// Send a fetch to the peer actor.
44    ///
45    /// If a fetch is already in progress for this key, this clears any existing
46    /// targets for that key (the fetch will try any available peer).
47    ///
48    /// If the engine has shut down, this is a no-op.
49    fn fetch<D>(&mut self, key: D) -> Feedback
50    where
51        D: Into<Fetch<Self::Key, Self::Subscriber>> + Send,
52    {
53        let Fetch {
54            key,
55            subscriber,
56            span,
57        } = key.into();
58        self.sender.enqueue(Message::Fetch(vec![FetchKey {
59            key,
60            subscribers: NonEmptyVec::new((subscriber, span)),
61            metadata: None,
62        }]))
63    }
64
65    /// Send fetches to the peer actor for a batch of keys.
66    ///
67    /// If a fetch is already in progress for any key, this clears any existing
68    /// targets for that key (the fetch will try any available peer).
69    ///
70    /// If the engine has shut down, this is a no-op.
71    fn fetch_all<D>(&mut self, keys: Vec<D>) -> Feedback
72    where
73        D: Into<Fetch<Self::Key, Self::Subscriber>> + Send,
74    {
75        self.sender.enqueue(Message::Fetch(
76            keys.into_iter()
77                .map(|key| fetch_key(key.into(), None))
78                .collect(),
79        ))
80    }
81
82    /// Send a retain request to the peer actor.
83    ///
84    /// If the engine has shut down, this is a no-op.
85    fn retain(
86        &mut self,
87        predicate: impl Fn(&Self::Key, &Self::Subscriber) -> bool + Send + 'static,
88    ) -> Feedback {
89        self.sender.enqueue(Message::Retain {
90            predicate: Box::new(predicate),
91        })
92    }
93}
94
95impl<K, P, S> TargetedResolver for Mailbox<K, P, S>
96where
97    K: Span,
98    P: PublicKey,
99    S: Clone + Eq + Send + 'static,
100{
101    type PublicKey = P;
102
103    /// Send a targeted fetch to the peer actor.
104    ///
105    /// If a fetch is already in progress for this key:
106    /// - If the existing fetch has targets, the new targets are added to the set.
107    /// - If the existing fetch has no targets, it remains unrestricted.
108    ///
109    /// To clear targeting and fall back to any peer, call [`fetch`](Self::fetch).
110    ///
111    /// Targets are automatically cleared when the fetch succeeds or is canceled.
112    /// When a peer is blocked for invalid data, only that peer is removed from
113    /// the target set.
114    ///
115    /// If the engine has shut down, this is a no-op.
116    fn fetch_targeted(
117        &mut self,
118        key: impl Into<Fetch<Self::Key, Self::Subscriber>> + Send,
119        targets: NonEmptyVec<Self::PublicKey>,
120    ) -> Feedback {
121        let Fetch {
122            key,
123            subscriber,
124            span,
125        } = key.into();
126        self.sender.enqueue(Message::Fetch(vec![FetchKey {
127            key,
128            subscribers: NonEmptyVec::new((subscriber, span)),
129            metadata: Some(targets),
130        }]))
131    }
132
133    /// Send targeted fetches to the peer actor for a batch of keys.
134    ///
135    /// If the engine has shut down, this is a no-op.
136    fn fetch_all_targeted<D>(&mut self, keys: Vec<(D, NonEmptyVec<Self::PublicKey>)>) -> Feedback
137    where
138        D: Into<Fetch<Self::Key, Self::Subscriber>> + Send,
139    {
140        self.sender.enqueue(Message::Fetch(
141            keys.into_iter()
142                .map(|(key, targets)| fetch_key(key.into(), Some(targets)))
143                .collect(),
144        ))
145    }
146}
147
148#[cfg(test)]
149mod tests {
150    use super::*;
151    use commonware_actor::mailbox::{Overflow, Policy};
152
153    type TestMessage = Message<u8, u8, u16>;
154    type TestPending = ingress::Pending<u8, u16, Option<NonEmptyVec<u8>>>;
155
156    fn fetch(key: u8, subscriber: u16, targets: Option<NonEmptyVec<u8>>) -> TestMessage {
157        Message::Fetch(vec![FetchKey {
158            key,
159            subscribers: NonEmptyVec::new((subscriber, tracing::Span::none())),
160            metadata: targets,
161        }])
162    }
163
164    fn fetch_with_subscribers(
165        key: u8,
166        subscribers: Vec<u16>,
167        targets: Option<NonEmptyVec<u8>>,
168    ) -> TestMessage {
169        Message::Fetch(vec![FetchKey {
170            key,
171            subscribers: NonEmptyVec::from_unchecked(
172                subscribers
173                    .into_iter()
174                    .map(|subscriber| (subscriber, tracing::Span::none()))
175                    .collect(),
176            ),
177            metadata: targets,
178        }])
179    }
180
181    fn subscriber_is(value: u16) -> impl Fn(&u8, &u16) -> bool + Send {
182        move |_, subscriber| *subscriber == value
183    }
184
185    fn targets(values: &[u8]) -> NonEmptyVec<u8> {
186        NonEmptyVec::from_unchecked(values.to_vec())
187    }
188
189    fn drain(pending: &mut TestPending) -> Vec<TestMessage> {
190        let mut messages = Vec::new();
191        Overflow::drain(pending, |message| {
192            messages.push(message);
193            None
194        });
195        messages
196    }
197
198    fn assert_fetch(message: &TestMessage, expected_key: u8, expected_targets: Option<&[u8]>) {
199        let Message::Fetch(keys) = message else {
200            panic!("expected fetch");
201        };
202        assert_eq!(keys.len(), 1);
203        assert_eq!(keys[0].key, expected_key);
204        match (&keys[0].metadata, expected_targets) {
205            (None, None) => {}
206            (Some(actual), Some(expected)) => assert_eq!(&actual[..], expected),
207            _ => panic!("unexpected targets"),
208        }
209    }
210
211    fn assert_fetch_keys(message: &TestMessage, expected: &[u8]) {
212        let Message::Fetch(keys) = message else {
213            panic!("expected fetch");
214        };
215        let actual: Vec<_> = keys.iter().map(|key| key.key).collect();
216        assert_eq!(actual, expected);
217    }
218
219    fn assert_fetch_subscribers(
220        message: &TestMessage,
221        expected_key: u8,
222        expected_subscribers: &[u16],
223    ) {
224        let Message::Fetch(keys) = message else {
225            panic!("expected fetch");
226        };
227        assert_eq!(keys.len(), 1);
228        assert_eq!(keys[0].key, expected_key);
229        let actual: Vec<_> = keys[0]
230            .subscribers
231            .iter()
232            .map(|(subscriber, _)| *subscriber)
233            .collect();
234        assert_eq!(actual, expected_subscribers);
235    }
236
237    #[test]
238    fn targeted_fetches_for_same_key_are_merged() {
239        let mut pending = TestPending::default();
240
241        Policy::handle(&mut pending, fetch(1, 10, Some(targets(&[2, 3]))));
242        Policy::handle(&mut pending, fetch(1, 11, Some(targets(&[3, 4]))));
243
244        let messages = drain(&mut pending);
245        assert_eq!(messages.len(), 1);
246        assert_fetch(&messages[0], 1, Some(&[2, 3, 4]));
247        assert_fetch_subscribers(&messages[0], 1, &[10, 11]);
248    }
249
250    #[test]
251    fn duplicate_fetches_for_same_key_merge_subscribers() {
252        let mut pending = TestPending::default();
253
254        Policy::handle(&mut pending, fetch_with_subscribers(1, vec![10, 11], None));
255        Policy::handle(&mut pending, fetch_with_subscribers(1, vec![11, 12], None));
256
257        let messages = drain(&mut pending);
258        assert_eq!(messages.len(), 1);
259        assert_fetch_subscribers(&messages[0], 1, &[10, 11, 12]);
260    }
261
262    #[test]
263    fn unrestricted_fetch_dominates_targeted_fetches() {
264        let mut pending = TestPending::default();
265
266        Policy::handle(&mut pending, fetch(1, 10, Some(targets(&[2]))));
267        Policy::handle(&mut pending, fetch(1, 11, None));
268        Policy::handle(&mut pending, fetch(1, 12, Some(targets(&[3]))));
269
270        let messages = drain(&mut pending);
271        assert_eq!(messages.len(), 1);
272        assert_fetch(&messages[0], 1, None);
273    }
274
275    #[test]
276    fn retain_removes_fetches_for_dropped_subscribers() {
277        let mut pending = TestPending::default();
278
279        Policy::handle(&mut pending, fetch(1, 10, None));
280        Policy::handle(&mut pending, fetch(2, 11, None));
281        Policy::handle(
282            &mut pending,
283            Message::Retain {
284                predicate: Box::new(subscriber_is(11)),
285            },
286        );
287
288        let messages = drain(&mut pending);
289        assert_eq!(messages.len(), 2);
290        assert!(matches!(messages[0], Message::Retain { .. }));
291        assert_fetch(&messages[1], 2, None);
292    }
293
294    #[test]
295    fn retain_prunes_pending_fetch_subscribers() {
296        let mut pending = TestPending::default();
297
298        Policy::handle(&mut pending, fetch_with_subscribers(1, vec![10, 11], None));
299        Policy::handle(
300            &mut pending,
301            Message::Retain {
302                predicate: Box::new(subscriber_is(11)),
303            },
304        );
305
306        let messages = drain(&mut pending);
307        assert_eq!(messages.len(), 2);
308        assert!(matches!(messages[0], Message::Retain { .. }));
309        assert_fetch_subscribers(&messages[1], 1, &[11]);
310    }
311
312    #[test]
313    fn retain_drops_pending_fetch_when_all_subscribers_are_dropped() {
314        let mut pending = TestPending::default();
315
316        Policy::handle(&mut pending, fetch_with_subscribers(1, vec![10, 11], None));
317        Policy::handle(
318            &mut pending,
319            Message::Retain {
320                predicate: Box::new(subscriber_is(12)),
321            },
322        );
323
324        let messages = drain(&mut pending);
325        assert_eq!(messages.len(), 1);
326        assert!(matches!(messages[0], Message::Retain { .. }));
327    }
328
329    #[test]
330    fn fetch_after_retain_is_retained_when_subscriber_is_dropped() {
331        let mut pending = TestPending::default();
332
333        Policy::handle(
334            &mut pending,
335            Message::Retain {
336                predicate: Box::new(|_, subscriber| *subscriber != 10),
337            },
338        );
339        Policy::handle(&mut pending, fetch(1, 10, None));
340        Policy::handle(&mut pending, fetch(2, 11, None));
341
342        let messages = drain(&mut pending);
343        assert_eq!(messages.len(), 2);
344        assert!(matches!(messages[0], Message::Retain { .. }));
345        assert_fetch_keys(&messages[1], &[1, 2]);
346    }
347}
commonware_resolver/p2p/ingress.rs

commonware_resolver/p2p/
ingress.rs
