hashiverse_lib/client/post_bundle/
live_post_bundle_feedback_manager.rs

1//! # Production [`PostBundleFeedbackManager`] — cache, fetch, heal
2//!
3//! Mirrors
4//! [`crate::client::post_bundle::live_post_bundle_manager`] but for feedback: checks
5//! `BUCKET_POST_BUNDLE_FEEDBACK` in
6//! [`crate::client::client_storage::client_storage::ClientStorage`] (with an age-based
7//! TTL since feedback mutates), falls back to `GetPostBundleFeedbackV1` against DHT
8//! peers, and spawns
9//! [`crate::client::post_bundle::post_bundle_feedback_healing`] in the background to
10//! reconcile divergent copies. Cache propagation is guided by the same
11//! [`crate::client::caching::cache_radius_tracker`] used by the post-bundle side.
12
13use crate::anyhow_assert_eq;
14use crate::client::caching::cache_radius_tracker::CacheRadiusTracker;
15use crate::client::caching::post_bundle_cache_uploader;
16use crate::client::client_storage::client_storage::{ClientStorage, BUCKET_POST_BUNDLE_FEEDBACK};
17use crate::client::peer_tracker::peer_tracker::{PeerTracker, SharedPeerTracker};
18use crate::client::post_bundle::post_bundle_feedback_healing;
19use crate::client::post_bundle::post_bundle_feedback_manager::PostBundleFeedbackManager;
20use crate::protocol::payload::payload::{CacheRequestTokenV1, GetPostBundleFeedbackResponseV1, GetPostBundleFeedbackV1, PayloadRequestKind, PayloadResponseKind};
21use crate::protocol::posting::encoded_post_bundle_feedback::EncodedPostBundleFeedbackV1;
22use crate::protocol::peer::Peer;
23use crate::protocol::rpc;
24use crate::tools::buckets::BucketLocation;
25use crate::tools::config::{CLIENT_POST_BUNDLE_FEEDBACK_CACHE_DURATION};
26use crate::tools::runtime_services::RuntimeServices;
27use crate::tools::time::TimeMillis;
28use crate::tools::tools::LeadingAgreementBits;
29use crate::tools::types::Id;
30use crate::tools::{config, json};
31use bytes::Bytes;
32use log::{info, trace, warn};
33use scopeguard::defer;
34use std::collections::HashMap;
35use std::sync::Arc;
36use tokio::sync::Mutex;
37
38pub struct LivePostBundleFeedbackManager {
39    runtime_services: Arc<RuntimeServices>,
40    client_storage: Arc<dyn ClientStorage>,
41    peer_tracker: SharedPeerTracker,
42    sponsor_id: Id,
43    post_bundle_feedback_inflight: Mutex<HashMap<Id, Arc<Mutex<()>>>>,
44    post_bundle_feedback_cache_radius_tracker: CacheRadiusTracker,
45}
46
47impl LivePostBundleFeedbackManager {
48    pub fn new(runtime_services: Arc<RuntimeServices>, sponsor_id: Id, client_storage: Arc<dyn ClientStorage>, peer_tracker: SharedPeerTracker) -> Self {
49        Self {
50            runtime_services,
51            client_storage,
52            peer_tracker,
53            sponsor_id,
54            post_bundle_feedback_inflight: Mutex::new(HashMap::new()),
55            post_bundle_feedback_cache_radius_tracker: CacheRadiusTracker::new(CLIENT_POST_BUNDLE_FEEDBACK_CACHE_DURATION.const_mul(5)), // We let these cache radius caches last longer than our own local cache of the bundles...
56        }
57    }
58}
59
60#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
61#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
62impl PostBundleFeedbackManager for LivePostBundleFeedbackManager {
63    async fn get_post_bundle_feedback(&self, bucket_location: BucketLocation, time_millis: TimeMillis) -> anyhow::Result<EncodedPostBundleFeedbackV1> {
64        let post_bundle_location_id = bucket_location.location_id;
65        let get_from_cache = async |location_id: Id, time_millis: TimeMillis, reason: &str| -> Option<EncodedPostBundleFeedbackV1> {
66            let result: anyhow::Result<Option<EncodedPostBundleFeedbackV1>> = try {
67                let raw = self.client_storage.get(BUCKET_POST_BUNDLE_FEEDBACK, &location_id.to_hex_str(), time_millis).await?;
68                if let Some(raw) = raw {
69                    let feedback = EncodedPostBundleFeedbackV1::from_bytes(Bytes::from(raw))?;
70                    let duration = time_millis - feedback.header.time_millis;
71                    if duration < CLIENT_POST_BUNDLE_FEEDBACK_CACHE_DURATION {
72                        trace!("Using cached PostBundleFeedback for {} (age {}) at {}", location_id, duration, reason);
73                        Some(feedback)
74                    } else {
75                        trace!("Cached PostBundleFeedback for {} expired (age {}) at {}", location_id, duration, reason);
76                        None
77                    }
78                } else {
79                    None
80                }
81            };
82
83            result.unwrap_or_else(|e| {
84                warn!("discarding problematic cached PostBundleFeedback: {}", e);
85                None
86            })
87        };
88
89        // Check the cache
90        if let Some(cached) = get_from_cache(post_bundle_location_id, time_millis, "preflight").await {
91            return Ok(cached);
92        }
93
94        // Inflight deduplication: get or create a per-key mutex
95        let key_lock = {
96            let mut inflight = self.post_bundle_feedback_inflight.lock().await;
97            inflight.entry(post_bundle_location_id).or_insert_with(|| Arc::new(Mutex::new(()))).clone()
98        };
99        let _inflight_guard = key_lock.lock().await;
100        defer!(if let Ok(mut m) = self.post_bundle_feedback_inflight.try_lock() { m.remove(&post_bundle_location_id); });
101
102        // Re-check the cache - a concurrent caller may have just finished fetching and stored the result.
103        if let Some(cached) = get_from_cache(post_bundle_location_id, time_millis, "postflight").await {
104            return Ok(cached);
105        }
106
107        let cache_radius = self.post_bundle_feedback_cache_radius_tracker.get(post_bundle_location_id, time_millis);
108
109        // We need to refresh...
110        let mut peers_visited = Vec::new();
111        let mut accepted_post_bundle_feedbacks = AcceptedPostBundleFeedbacks::new(post_bundle_location_id, time_millis);
112        let mut cache_request_tokens: Vec<CacheRequestTokenV1> = Vec::new();
113        let mut positive_responder_leading_agreement_bits: Vec<LeadingAgreementBits> = Vec::new();
114        {
115            let mut peer_iter = PeerTracker::iterate_to_location(&self.peer_tracker, post_bundle_location_id, 2 * config::REDUNDANT_SERVERS_PER_POST, cache_radius).await?;
116            while let Some((peer, leading_agreement_bits)) = peer_iter.next_peer() {
117                if accepted_post_bundle_feedbacks.accepted_originator_ids().contains(&peer.id) {
118                    continue;
119                }
120
121                let result: anyhow::Result<()> = try {
122                    info!("Requesting PostBundleFeedback with leading_agreement_bits={} from peer {}", leading_agreement_bits, peer);
123
124                    let request = GetPostBundleFeedbackV1 {
125                        bucket_location: bucket_location.clone(),
126                        peers_visited: peers_visited.clone(),
127                        already_retrieved_peer_ids: accepted_post_bundle_feedbacks.accepted_originator_ids().to_vec(),
128                    };
129                    let request = json::struct_to_bytes(&request)?;
130                    let response = rpc::rpc::rpc_server_known(&self.runtime_services, &self.sponsor_id, &peer, PayloadRequestKind::GetPostBundleFeedbackV1, request).await?;
131                    anyhow_assert_eq!(&PayloadResponseKind::GetPostBundleFeedbackResponseV1, &response.response_request_kind);
132                    let response = GetPostBundleFeedbackResponseV1::from_bytes(response.bytes)?;
133
134                    peers_visited.push(peer.clone());
135                    peer_iter.add_peers(response.peers_nearer);
136
137                    if let Some(token) = response.cache_request_token {
138                        cache_request_tokens.push(token);
139                    }
140
141                    let mut found_bundle_from_this_peer = false;
142
143                    for cached_bytes in response.post_bundle_feedbacks_cached {
144                        match accepted_post_bundle_feedbacks.offer_cached(cached_bytes) {
145                            Ok(accepted) => found_bundle_from_this_peer |= accepted,
146                            Err(e) => warn!("Rejected cached PostBundleFeedback relayed by peer {}: {}", peer, e),
147                        }
148                    }
149
150                    if let Some(post_bundle_feedback_raw) = response.encoded_post_bundle_feedback {
151                        found_bundle_from_this_peer |= accepted_post_bundle_feedbacks.offer_direct(post_bundle_feedback_raw, &peer)?;
152                    }
153
154                    if found_bundle_from_this_peer {
155                        positive_responder_leading_agreement_bits.push(leading_agreement_bits);
156                    }
157
158                    if accepted_post_bundle_feedbacks.quorum_count() >= config::REDUNDANT_SERVERS_PER_POST {
159                        break;
160                    }
161                };
162
163                if let Err(e) = result {
164                    warn!("Error retrieving PostBundleFeedback from peer {}: {}", peer, e);
165                    peer_iter.remove_peer(&peer);
166                }
167            }
168
169            let result = PeerTracker::flush(&self.peer_tracker).await;
170            if let Err(e) = result {
171                warn!("Error flushing peer tracker: {}", e);
172            }
173        }
174
175        trace!("Discovered {} post bundle feedbacks after visiting {} peers", accepted_post_bundle_feedbacks.quorum_count(), peers_visited.len());
176
177        // Store our new cache radius - the peer "furthest out" from our bundle's location_id
178        if let Some(new_radius) = positive_responder_leading_agreement_bits.iter().copied().min() {
179            self.post_bundle_feedback_cache_radius_tracker.update(post_bundle_location_id, new_radius, time_millis);
180        }
181
182        // Merge all collected bundles, taking the best-pow entry per (post_id, feedback_type).
183        // NB: the result is an unsigned client view (its header no longer covers its feedbacks_bytes) - fine for our own cache, never forward it.
184        let best_encoded_post_bundle_feedback = EncodedPostBundleFeedbackV1::merge(accepted_post_bundle_feedbacks.bundles())
185            .ok_or_else(|| anyhow::anyhow!("No post bundle feedbacks discovered for {}", post_bundle_location_id))?;
186
187        // Cache it
188        trace!("Caching PostBundleFeedback from post_bundle_location_id={} with length {}", post_bundle_location_id, best_encoded_post_bundle_feedback.feedbacks_bytes.len());
189        let encoded_post_bundle_feedback_bytes = best_encoded_post_bundle_feedback.to_bytes()?;
190        self.client_storage.put(BUCKET_POST_BUNDLE_FEEDBACK, &post_bundle_location_id.to_hex_str(), encoded_post_bundle_feedback_bytes.to_vec(), time_millis).await?;
191
192        // Kick off healing
193        post_bundle_feedback_healing::heal_post_bundle_feedbacks(
194            self.runtime_services.clone(),
195            self.sponsor_id,
196            &peers_visited,
197            accepted_post_bundle_feedbacks.bundles(),
198            &best_encoded_post_bundle_feedback,
199        );
200
201        // Kick off remote cache populating
202        post_bundle_cache_uploader::upload_post_bundle_feedback_caches(
203            self.runtime_services.clone(),
204            self.sponsor_id,
205            cache_request_tokens,
206            accepted_post_bundle_feedbacks.bundle_bytes_for_upload(),
207        );
208
209        Ok(best_encoded_post_bundle_feedback)
210    }
211}
212
213/// The feedback bundles accepted so far while walking towards a bucket's location.  Mirrors
214/// [`crate::client::post_bundle::live_post_bundle_manager`]'s `AcceptedPostBundles`: every accepted bundle comes from a genuine
215/// (server-PoW) originator, is for this location, and counts once towards the quorum - so free self-signed bundles can neither
216/// end the walk early nor mask a genuine originator's copy via `already_retrieved_peer_ids`.
217struct AcceptedPostBundleFeedbacks {
218    location_id: Id,
219    time_millis: TimeMillis,
220    accepted_originator_ids: Vec<Id>,
221    bundles: Vec<EncodedPostBundleFeedbackV1>,
222    /// The intact originator-signed bytes of every accepted, non-empty bundle that passes full verify(): the only thing we may forward to caches.
223    bundle_bytes_for_upload: Vec<(Id, Bytes)>,
224}
225
226impl AcceptedPostBundleFeedbacks {
227    fn new(location_id: Id, time_millis: TimeMillis) -> Self {
228        Self { location_id, time_millis, accepted_originator_ids: Vec::new(), bundles: Vec::new(), bundle_bytes_for_upload: Vec::new() }
229    }
230
231    /// Originators whose bundles we have accepted - passed to servers so they don't send us those again.
232    fn accepted_originator_ids(&self) -> &[Id] {
233        &self.accepted_originator_ids
234    }
235
236    /// How many distinct servers' views we have accepted.
237    fn quorum_count(&self) -> usize {
238        self.accepted_originator_ids.len()
239    }
240
241    fn bundles(&self) -> &[EncodedPostBundleFeedbackV1] {
242        &self.bundles
243    }
244
245    fn bundle_bytes_for_upload(&self) -> Vec<(Id, Bytes)> {
246        self.bundle_bytes_for_upload.clone()
247    }
248
249    /// A cached snapshot relayed by some peer.  Servers fully verify feedback before caching it, so a relayed copy must pass full
250    /// verification, and an empty one proves nothing.  Returns whether it was accepted; errors if it was unacceptable.
251    fn offer_cached(&mut self, encoded_post_bundle_feedback_bytes: Bytes) -> anyhow::Result<bool> {
252        let encoded_post_bundle_feedback = EncodedPostBundleFeedbackV1::from_bytes(encoded_post_bundle_feedback_bytes.clone())?;
253        if self.accepted_originator_ids.contains(&encoded_post_bundle_feedback.header.peer.id) {
254            return Ok(false);
255        }
256        encoded_post_bundle_feedback.verify_for_location(&self.location_id)?;
257        self.verify_not_from_the_future(&encoded_post_bundle_feedback)?;
258        if encoded_post_bundle_feedback.feedbacks_bytes.is_empty() {
259            trace!("Ignoring empty cached PostBundleFeedback from peer.id={}", encoded_post_bundle_feedback.header.peer.id);
260            return Ok(false);
261        }
262
263        trace!("Accepted cached PostBundleFeedback from peer.id={} with length {}", encoded_post_bundle_feedback.header.peer.id, encoded_post_bundle_feedback.feedbacks_bytes.len());
264        self.accept(encoded_post_bundle_feedback_bytes, encoded_post_bundle_feedback);
265        Ok(true)
266    }
267
268    /// The bundle `responding_peer` holds itself.  It may be empty, and may hold individual entries with bad PoW (which merge drops).
269    /// Returns whether it was accepted; errors if it was unacceptable.
270    fn offer_direct(&mut self, encoded_post_bundle_feedback_bytes: Bytes, responding_peer: &Peer) -> anyhow::Result<bool> {
271        let encoded_post_bundle_feedback = EncodedPostBundleFeedbackV1::from_bytes(encoded_post_bundle_feedback_bytes.clone())?;
272        if encoded_post_bundle_feedback.header.peer.id != responding_peer.id {
273            anyhow::bail!("PostBundleFeedback from peer {} claims to be published by {}", responding_peer.id, encoded_post_bundle_feedback.header.peer.id);
274        }
275        if self.accepted_originator_ids.contains(&encoded_post_bundle_feedback.header.peer.id) {
276            return Ok(false);
277        }
278        encoded_post_bundle_feedback.verify_origin_for_location(&self.location_id)?;
279        self.verify_not_from_the_future(&encoded_post_bundle_feedback)?;
280
281        trace!("Accepted PostBundleFeedback from peer.id={} with length {}", encoded_post_bundle_feedback.header.peer.id, encoded_post_bundle_feedback.feedbacks_bytes.len());
282        self.accept(encoded_post_bundle_feedback_bytes, encoded_post_bundle_feedback);
283        Ok(true)
284    }
285
286    /// merge() keeps the newest header, and our local cache judges freshness by it - so a future-dated bundle would pin a stale view in it.
287    fn verify_not_from_the_future(&self, encoded_post_bundle_feedback: &EncodedPostBundleFeedbackV1) -> anyhow::Result<()> {
288        let time_millis_latest_allowed = self.time_millis + config::CLIENT_POST_TIMESTAMP_DELTA_THRESHOLD;
289        if encoded_post_bundle_feedback.header.time_millis > time_millis_latest_allowed {
290            anyhow::bail!("PostBundleFeedback from peer.id={} is dated {}, later than the allowed {}", encoded_post_bundle_feedback.header.peer.id, encoded_post_bundle_feedback.header.time_millis, time_millis_latest_allowed);
291        }
292        Ok(())
293    }
294
295    fn accept(&mut self, encoded_post_bundle_feedback_bytes: Bytes, encoded_post_bundle_feedback: EncodedPostBundleFeedbackV1) {
296        let originator_id = encoded_post_bundle_feedback.header.peer.id;
297        self.accepted_originator_ids.push(originator_id);
298        if !encoded_post_bundle_feedback.feedbacks_bytes.is_empty() && encoded_post_bundle_feedback.verify().is_ok() {
299            self.bundle_bytes_for_upload.push((originator_id, encoded_post_bundle_feedback_bytes));
300        }
301        self.bundles.push(encoded_post_bundle_feedback);
302    }
303}
304
305#[cfg(test)]
306mod tests {
307    use super::*;
308    use crate::protocol::posting::encoded_post_bundle_feedback::test_fixtures::{make_bundle_signed_by, make_forged_feedback, make_valid_feedback};
309    use crate::protocol::posting::encoded_post_feedback::EncodedPostFeedbackV1;
310    use crate::tools::pow_generator::single_threaded_pow_generator::SingleThreadedPowGenerator;
311    use crate::tools::server_id::ServerId;
312    use crate::tools::time_provider::time_provider::RealTimeProvider;
313    use crate::tools::time::MILLIS_IN_MILLISECOND;
314    use crate::tools::types::Pow;
315
316    async fn make_genuine_server_id() -> anyhow::Result<ServerId> {
317        ServerId::new("genuine", &RealTimeProvider, config::SERVER_KEY_POW_MIN, true, &SingleThreadedPowGenerator::new()).await
318    }
319
320    async fn make_unworked_server_id() -> anyhow::Result<ServerId> {
321        loop {
322            let server_id = ServerId::new("unworked", &RealTimeProvider, Pow(0), true, &SingleThreadedPowGenerator::new()).await?;
323            if server_id.pow < config::SERVER_KEY_POW_MIN {
324                return Ok(server_id);
325            }
326        }
327    }
328
329    fn bundle_bytes(server_id: &ServerId, location_id: Id, encoded_post_feedbacks: &[EncodedPostFeedbackV1]) -> anyhow::Result<Bytes> {
330        make_bundle_signed_by(server_id, location_id, TimeMillis(1000), encoded_post_feedbacks)?.to_bytes()
331    }
332
333    // Quorum stuffing: free self-signed bundles, and empty cached ones, must not end the walk.
334    #[tokio::test]
335    async fn self_signed_and_empty_cached_bundles_do_not_reach_quorum() -> anyhow::Result<()> {
336        let location_id = Id::random();
337        let mut accepted_post_bundle_feedbacks = AcceptedPostBundleFeedbacks::new(location_id, TimeMillis(1000));
338        for _ in 0..config::REDUNDANT_SERVERS_PER_POST {
339            let unworked_server_id = make_unworked_server_id().await?;
340            assert!(accepted_post_bundle_feedbacks.offer_cached(bundle_bytes(&unworked_server_id, location_id, &[make_valid_feedback(Id::random(), 1)?])?).is_err());
341            assert!(!accepted_post_bundle_feedbacks.offer_cached(bundle_bytes(&make_genuine_server_id().await?, location_id, &[])?)?);
342        }
343        assert_eq!(accepted_post_bundle_feedbacks.quorum_count(), 0);
344        assert!(accepted_post_bundle_feedbacks.accepted_originator_ids().is_empty());
345        Ok(())
346    }
347
348    #[tokio::test]
349    async fn cached_bundle_must_be_for_this_location_and_fully_valid() -> anyhow::Result<()> {
350        let location_id = Id::random();
351        let server_id = make_genuine_server_id().await?;
352        let mut accepted_post_bundle_feedbacks = AcceptedPostBundleFeedbacks::new(location_id, TimeMillis(1000));
353        assert!(accepted_post_bundle_feedbacks.offer_cached(bundle_bytes(&server_id, Id::random(), &[make_valid_feedback(Id::random(), 1)?])?).is_err());
354        assert!(accepted_post_bundle_feedbacks.offer_cached(bundle_bytes(&server_id, location_id, &[make_forged_feedback(Id::random(), 1)])?).is_err());
355        assert_eq!(accepted_post_bundle_feedbacks.quorum_count(), 0);
356        Ok(())
357    }
358
359    // Masking: a forged bundle claiming a genuine originator's id must not land that id in already_retrieved_peer_ids.
360    #[tokio::test]
361    async fn forged_bundle_cannot_claim_a_genuine_originator() -> anyhow::Result<()> {
362        let location_id = Id::random();
363        let genuine_server_id = make_genuine_server_id().await?;
364        let impostor_server_id = make_genuine_server_id().await?;
365        let mut forged_bundle = make_bundle_signed_by(&impostor_server_id, location_id, TimeMillis(1000), &[make_valid_feedback(Id::random(), 1)?])?;
366        forged_bundle.header.peer = genuine_server_id.to_peer(&RealTimeProvider)?;
367        forged_bundle.header.signature_generate(&impostor_server_id.keys.signature_key);
368
369        let mut accepted_post_bundle_feedbacks = AcceptedPostBundleFeedbacks::new(location_id, TimeMillis(1000));
370        assert!(accepted_post_bundle_feedbacks.offer_cached(forged_bundle.to_bytes()?).is_err());
371        assert!(!accepted_post_bundle_feedbacks.accepted_originator_ids().contains(&genuine_server_id.id));
372
373        // The genuine copy is then still accepted
374        assert!(accepted_post_bundle_feedbacks.offer_cached(bundle_bytes(&genuine_server_id, location_id, &[make_valid_feedback(Id::random(), 1)?])?)?);
375        assert_eq!(accepted_post_bundle_feedbacks.accepted_originator_ids(), &[genuine_server_id.id]);
376        Ok(())
377    }
378
379    #[tokio::test]
380    async fn direct_bundle_must_name_the_responding_server() -> anyhow::Result<()> {
381        let location_id = Id::random();
382        let responding_server_id = make_genuine_server_id().await?;
383        let other_server_id = make_genuine_server_id().await?;
384        let mut accepted_post_bundle_feedbacks = AcceptedPostBundleFeedbacks::new(location_id, TimeMillis(1000));
385        let responding_peer = responding_server_id.to_peer(&RealTimeProvider)?;
386        assert!(accepted_post_bundle_feedbacks.offer_direct(bundle_bytes(&other_server_id, location_id, &[make_valid_feedback(Id::random(), 1)?])?, &responding_peer).is_err());
387        assert!(accepted_post_bundle_feedbacks.offer_direct(bundle_bytes(&responding_server_id, Id::random(), &[])?, &responding_peer).is_err(), "and be for this location");
388        assert_eq!(accepted_post_bundle_feedbacks.quorum_count(), 0);
389        Ok(())
390    }
391
392    // A server's own copy counts even when empty or holding a bad entry (merge drops it), but only fully valid, non-empty copies are uploaded.
393    #[tokio::test]
394    async fn direct_bundles_count_once_and_only_valid_ones_are_uploaded() -> anyhow::Result<()> {
395        let location_id = Id::random();
396        let mut accepted_post_bundle_feedbacks = AcceptedPostBundleFeedbacks::new(location_id, TimeMillis(1000));
397
398        let empty_server_id = make_genuine_server_id().await?;
399        assert!(accepted_post_bundle_feedbacks.offer_direct(bundle_bytes(&empty_server_id, location_id, &[])?, &empty_server_id.to_peer(&RealTimeProvider)?)?);
400
401        let poisoned_server_id = make_genuine_server_id().await?;
402        let valid_feedback = make_valid_feedback(Id::random(), 1)?;
403        assert!(accepted_post_bundle_feedbacks.offer_direct(bundle_bytes(&poisoned_server_id, location_id, &[valid_feedback.clone(), make_forged_feedback(Id::random(), 2)])?, &poisoned_server_id.to_peer(&RealTimeProvider)?)?);
404
405        let honest_server_id = make_genuine_server_id().await?;
406        let honest_bundle_bytes = bundle_bytes(&honest_server_id, location_id, &[make_valid_feedback(Id::random(), 1)?])?;
407        assert!(accepted_post_bundle_feedbacks.offer_direct(honest_bundle_bytes.clone(), &honest_server_id.to_peer(&RealTimeProvider)?)?);
408        assert!(!accepted_post_bundle_feedbacks.offer_cached(honest_bundle_bytes.clone())?, "an originator counts once");
409
410        assert_eq!(accepted_post_bundle_feedbacks.quorum_count(), 3);
411        assert_eq!(accepted_post_bundle_feedbacks.bundle_bytes_for_upload(), vec![(honest_server_id.id, honest_bundle_bytes)]);
412        let merged = EncodedPostBundleFeedbackV1::merge(accepted_post_bundle_feedbacks.bundles()).expect("non-empty input");
413        assert_eq!(merged.get_post_pow_for_feedback_type(&valid_feedback.post_id, 1), valid_feedback.pow, "the poisoned server's honest entry still counts");
414        Ok(())
415    }
416
417    #[tokio::test]
418    async fn future_dated_bundles_are_rejected() -> anyhow::Result<()> {
419        let location_id = Id::random();
420        let time_millis_now = TimeMillis(1000);
421        let time_millis_latest_allowed = time_millis_now + config::CLIENT_POST_TIMESTAMP_DELTA_THRESHOLD;
422        let mut accepted_post_bundle_feedbacks = AcceptedPostBundleFeedbacks::new(location_id, time_millis_now);
423
424        let cached_server_id = make_genuine_server_id().await?;
425        let future_cached_bundle = make_bundle_signed_by(&cached_server_id, location_id, time_millis_latest_allowed + MILLIS_IN_MILLISECOND, &[make_valid_feedback(Id::random(), 1)?])?;
426        assert!(accepted_post_bundle_feedbacks.offer_cached(future_cached_bundle.to_bytes()?).is_err());
427
428        let direct_server_id = make_genuine_server_id().await?;
429        let future_direct_bundle = make_bundle_signed_by(&direct_server_id, location_id, time_millis_latest_allowed + MILLIS_IN_MILLISECOND, &[])?;
430        assert!(accepted_post_bundle_feedbacks.offer_direct(future_direct_bundle.to_bytes()?, &direct_server_id.to_peer(&RealTimeProvider)?).is_err());
431        assert_eq!(accepted_post_bundle_feedbacks.quorum_count(), 0);
432
433        // Right at the bound, both are still accepted
434        let edge_cached_bundle = make_bundle_signed_by(&cached_server_id, location_id, time_millis_latest_allowed, &[make_valid_feedback(Id::random(), 1)?])?;
435        assert!(accepted_post_bundle_feedbacks.offer_cached(edge_cached_bundle.to_bytes()?)?);
436        let edge_direct_bundle = make_bundle_signed_by(&direct_server_id, location_id, time_millis_latest_allowed, &[])?;
437        assert!(accepted_post_bundle_feedbacks.offer_direct(edge_direct_bundle.to_bytes()?, &direct_server_id.to_peer(&RealTimeProvider)?)?);
438        assert_eq!(accepted_post_bundle_feedbacks.quorum_count(), 2);
439        Ok(())
440    }
441}
hashiverse_lib/client/post_bundle/live_post_bundle_feedback_manager.rs

hashiverse_lib/client/post_bundle/
live_post_bundle_feedback_manager.rs
