1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
//! Resolve data identified by a fixed-length key.
#![doc(
html_logo_url = "https://commonware.xyz/imgs/rustdoc_logo.svg",
html_favicon_url = "https://commonware.xyz/favicon.ico"
)]
commonware_macros::stability_scope!(BETA {
use commonware_actor::Feedback;
use commonware_cryptography::PublicKey;
use commonware_utils::{channel::oneshot, vec::NonEmptyVec, Span};
use core::cmp::Ordering;
pub mod delivery;
mod ingress;
pub mod opaque;
pub mod p2p;
mod subscribers;
/// A key to fetch data for a subscriber.
#[derive(Clone, Debug)]
pub struct Fetch<K, S = ()> {
/// The peer-visible key.
pub key: K,
/// Subscriber attached to the key.
pub subscriber: S,
/// Trace span carried from issuance to delivery.
pub span: tracing::Span,
}
impl<K: PartialEq, S: PartialEq> PartialEq for Fetch<K, S> {
fn eq(&self, other: &Self) -> bool {
self.key == other.key && self.subscriber == other.subscriber
}
}
impl<K: Eq, S: Eq> Eq for Fetch<K, S> {}
impl<K: PartialOrd, S: PartialOrd> PartialOrd for Fetch<K, S> {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
match self.key.partial_cmp(&other.key)? {
Ordering::Equal => self.subscriber.partial_cmp(&other.subscriber),
ordering => Some(ordering),
}
}
}
impl<K: Ord, S: Ord> Ord for Fetch<K, S> {
fn cmp(&self, other: &Self) -> Ordering {
self.key
.cmp(&other.key)
.then_with(|| self.subscriber.cmp(&other.subscriber))
}
}
impl<K, S: Default> From<K> for Fetch<K, S> {
fn from(key: K) -> Self {
Self {
key,
subscriber: S::default(),
span: tracing::Span::none(),
}
}
}
/// Data delivered for a resolved fetch.
#[derive(Clone, Debug)]
pub struct Delivery<K, S> {
/// The peer-visible key used to validate the response.
pub key: K,
/// Subscribers that were still retained when the response arrived, each
/// paired with the trace span of the fetch that requested it.
pub subscribers: NonEmptyVec<(S, tracing::Span)>,
}
impl<K: PartialEq, S: PartialEq> PartialEq for Delivery<K, S> {
fn eq(&self, other: &Self) -> bool {
self.key == other.key
&& self.subscribers.len() == other.subscribers.len()
&& self
.subscribers
.iter()
.zip(other.subscribers.iter())
.all(|((a, _), (b, _))| a == b)
}
}
impl<K: Eq, S: Eq> Eq for Delivery<K, S> {}
impl<K: PartialOrd, S: PartialOrd> PartialOrd for Delivery<K, S> {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
match self.key.partial_cmp(&other.key)? {
Ordering::Equal => self
.subscribers
.iter()
.map(|(subscriber, _)| subscriber)
.partial_cmp(other.subscribers.iter().map(|(subscriber, _)| subscriber)),
ordering => Some(ordering),
}
}
}
impl<K: Ord, S: Ord> Ord for Delivery<K, S> {
fn cmp(&self, other: &Self) -> Ordering {
self.key.cmp(&other.key).then_with(|| {
self.subscribers
.iter()
.map(|(subscriber, _)| subscriber)
.cmp(other.subscribers.iter().map(|(subscriber, _)| subscriber))
})
}
}
/// Notified when data is available, and must validate it.
pub trait Consumer: Clone + Send + 'static {
/// Type used to key data requested from peers.
type Key: Span;
/// Type of data to retrieve.
type Value;
/// Type used to track subscribers on fetch keys.
type Subscriber: Clone + Eq + Send + 'static;
/// Deliver data to the consumer.
///
/// Returns a receiver that resolves to `true` if the data is valid for the key.
///
/// The returned receiver may be dropped before completion if the application
/// cancels the fetch via [`Resolver::retain`]. When this happens, the
/// resolver discards the validation result.
///
/// Implementations of [`Resolver`] must only invoke `deliver` for keys that were
/// previously requested via [`Resolver::fetch`] (or [`TargetedResolver`] variants).
///
/// `delivery` contains the peer-visible key and the retained subscribers
/// for the fetch. Subscribers decide who should observe a valid response;
/// they do not define peer validity.
fn deliver(
&mut self,
delivery: Delivery<Self::Key, Self::Subscriber>,
value: Self::Value,
) -> oneshot::Receiver<bool>;
}
/// Responsible for fetching data and notifying a `Consumer`.
pub trait Resolver: Clone + Send + 'static {
/// Type used to key data requested from peers.
type Key: Span;
/// Type used to track subscribers on fetch keys.
///
/// Implementations that also own the [`Consumer`] should supply subscribers to
/// [`Consumer::deliver`] when a fetch resolves.
type Subscriber: Clone + Eq + Send + 'static;
/// Initiate a fetch.
///
/// The resolver fetches and delivers the key. The subscriber is
/// retained and supplied to [`Consumer::deliver`] when the fetch resolves.
/// If multiple subscribers are attached to the same key,
/// the fetch is retained as long as at least one subscriber satisfies the
/// latest [`retain`](Self::retain) predicate.
///
/// Passing a bare key is supported when `Subscriber: Default`.
fn fetch<F>(&mut self, key: F) -> Feedback
where
F: Into<Fetch<Self::Key, Self::Subscriber>> + Send;
/// Initiate fetches for a batch of keys.
fn fetch_all<F>(&mut self, keys: Vec<F>) -> Feedback
where
F: Into<Fetch<Self::Key, Self::Subscriber>> + Send;
/// Retain only fetch subscribers satisfying the predicate.
///
/// The predicate receives the peer-visible key and subscriber.
///
/// Fetches not retained are canceled. If response validation is in
/// progress, cancellation may drop the [`Consumer::deliver`] future
/// before it reports whether the data was valid.
fn retain(
&mut self,
predicate: impl Fn(&Self::Key, &Self::Subscriber) -> bool + Send + 'static,
) -> Feedback;
}
/// Extension for resolvers that accept target peer hints.
pub trait TargetedResolver: Resolver {
/// Type used to identify peers for targeted fetch hints.
type PublicKey: PublicKey;
/// Initiate a fetch with target peer hints.
///
/// Implementations define whether target hints persist through retries,
/// merge with existing in-progress fetches, or are discarded.
fn fetch_targeted(
&mut self,
fetch: impl Into<Fetch<Self::Key, Self::Subscriber>> + Send,
targets: NonEmptyVec<Self::PublicKey>,
) -> Feedback;
/// Initiate fetches for multiple keys, each with their own target hints.
///
/// See [`fetch_targeted`](Self::fetch_targeted) for details on target behavior.
fn fetch_all_targeted<F>(
&mut self,
keys: Vec<(F, NonEmptyVec<Self::PublicKey>)>,
) -> Feedback
where
F: Into<Fetch<Self::Key, Self::Subscriber>> + Send;
}
});