hashiverse_lib/client/caching/
post_bundle_cache_uploader.rs

1//! # Fire-and-forget cache replication
2//!
3//! After a successful post or feedback submission, the server returns a set of
4//! `CachePostBundle[Feedback]Token` values that authorise the client to forward the
5//! freshly-written bundle to *other* servers for caching. This module spawns those
6//! forwards as background tasks so the original submission can return to the user
7//! without blocking on network propagation.
8//!
9//! Bundles that already carry the remote server as an originator are filtered out —
10//! there's no point re-uploading data to a server we already know holds it.
11
12use std::sync::Arc;
13use bytes::Bytes;
14use crate::anyhow_assert_eq;
15use crate::protocol::payload::payload::{
16    CachePostBundleFeedbackResponseV1, CachePostBundleFeedbackV1, CachePostBundleResponseV1,
17    CachePostBundleV1, CacheRequestTokenV1, PayloadRequestKind, PayloadResponseKind,
18};
19use crate::protocol::rpc;
20use crate::tools::runtime_services::RuntimeServices;
21use crate::tools::tools::spawn_background_task;
22use crate::tools::types::Id;
23use crate::tools::json;
24use log::warn;
25
26/// Fire-and-forget: for each token, uploads all collected bundles whose originator is not
27/// already cached at the token's server.
28pub fn upload_post_bundle_caches(
29    runtime_services: Arc<RuntimeServices>,
30    sponsor_id: Id,
31    tokens: Vec<CacheRequestTokenV1>,
32    bundles: Vec<(Id, Bytes)>,
33) {
34    if tokens.is_empty() || bundles.is_empty() {
35        return;
36    }
37
38    spawn_background_task(async move {
39        for token in &tokens {
40            let now = runtime_services.time_provider.current_time_millis();
41            if token.is_expired(now) {
42                continue;
43            }
44
45            let filtered: Vec<&[u8]> = bundles_not_already_cached(&token.already_cached_peer_ids, &bundles).into_iter().map(|bytes| bytes.as_ref()).collect();
46
47            if filtered.is_empty() {
48                continue;
49            }
50
51            let result: anyhow::Result<()> = async {
52                let request_bytes = CachePostBundleV1::new_to_bytes(token, &filtered)?;
53                let response = rpc::rpc::rpc_server_known(
54                    &runtime_services,
55                    &sponsor_id,
56                    &token.peer,
57                    PayloadRequestKind::CachePostBundleV1,
58                    request_bytes,
59                )
60                .await?;
61                anyhow_assert_eq!(&PayloadResponseKind::CachePostBundleResponseV1, &response.response_request_kind);
62                json::bytes_to_struct::<CachePostBundleResponseV1>(&response.bytes)?;
63                Ok(())
64            }
65            .await;
66
67            if let Err(e) = result {
68                warn!("Failed to upload post bundle cache to {}: {}", token.peer, e);
69            }
70        }
71    });
72}
73
74/// Fire-and-forget: for each token, uploads every collected feedback bundle whose originator is
75/// not already cached at the token's server, one `CachePostBundleFeedbackV1` per bundle.
76///
77/// `bundles` must be the intact, originator-signed bytes (they have to pass `verify()` on the
78/// receiving server) — never the client's merged view from `EncodedPostBundleFeedbackV1::merge`.
79pub fn upload_post_bundle_feedback_caches(
80    runtime_services: Arc<RuntimeServices>,
81    sponsor_id: Id,
82    tokens: Vec<CacheRequestTokenV1>,
83    bundles: Vec<(Id, Bytes)>,
84) {
85    if tokens.is_empty() || bundles.is_empty() {
86        return;
87    }
88
89    spawn_background_task(async move {
90        for token in &tokens {
91            for feedback_bytes in bundles_not_already_cached(&token.already_cached_peer_ids, &bundles) {
92                let now = runtime_services.time_provider.current_time_millis();
93                if token.is_expired(now) {
94                    break;
95                }
96
97                let result: anyhow::Result<()> = async {
98                    let request_bytes = CachePostBundleFeedbackV1::new_to_bytes(token, feedback_bytes)?;
99                    let response = rpc::rpc::rpc_server_known(
100                        &runtime_services,
101                        &sponsor_id,
102                        &token.peer,
103                        PayloadRequestKind::CachePostBundleFeedbackV1,
104                        request_bytes,
105                    )
106                    .await?;
107                    anyhow_assert_eq!(&PayloadResponseKind::CachePostBundleFeedbackResponseV1, &response.response_request_kind);
108                    json::bytes_to_struct::<CachePostBundleFeedbackResponseV1>(&response.bytes)?;
109                    Ok(())
110                }
111                .await;
112
113                if let Err(e) = result {
114                    warn!("Failed to upload post bundle feedback cache to {}: {}", token.peer, e);
115                }
116            }
117        }
118    });
119}
120
121/// The bundles whose originator the token's server does not already hold in its cache.
122fn bundles_not_already_cached<'a>(already_cached_peer_ids: &[Id], bundles: &'a [(Id, Bytes)]) -> Vec<&'a Bytes> {
123    bundles.iter().filter(|(originator_id, _)| !already_cached_peer_ids.contains(originator_id)).map(|(_, bundle_bytes)| bundle_bytes).collect()
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129
130    #[test]
131    fn bundles_not_already_cached_skips_cached_originators() {
132        let cached_originator_id = Id::random();
133        let uncached_originator_id = Id::random();
134        let cached_bundle_bytes = Bytes::from_static(b"cached");
135        let uncached_bundle_bytes = Bytes::from_static(b"uncached");
136        let bundles = vec![(cached_originator_id, cached_bundle_bytes), (uncached_originator_id, uncached_bundle_bytes.clone())];
137
138        assert_eq!(bundles_not_already_cached(&[cached_originator_id], &bundles), vec![&uncached_bundle_bytes]);
139        assert_eq!(bundles_not_already_cached(&[], &bundles).len(), 2);
140        assert!(bundles_not_already_cached(&[cached_originator_id, uncached_originator_id], &bundles).is_empty());
141    }
142}
hashiverse_lib/client/caching/post_bundle_cache_uploader.rs

hashiverse_lib/client/caching/
post_bundle_cache_uploader.rs
