coven_storage/cloud/
counting.rs

1//! Counting the operations a run asks of its provider.
2//!
3//! A stage's wall time says how long it waited, not what it waited on, and the
4//! two shapes want opposite fixes: one slow transfer is a size problem, two
5//! hundred fast ones are a round-trip problem no faster network will help. So
6//! every provider call is counted, and the stage timings report the count
7//! beside the time.
8//!
9//! The count is taken at the [`CloudHome`]/[`ExactSlotStorage`] boundary, which
10//! is the one place every provider call crosses, so nothing is counted twice
11//! and nothing is missed. What it counts is *operations a caller asked for*,
12//! not HTTP round trips: a listing that pages, a write that goes multipart, and
13//! a delete that reads back to prove absence are each one operation here and
14//! several requests underneath. That is the granularity the budget is written
15//! in — whether a stage does a fixed number of operations or one per commit —
16//! and a stage whose count is small while its time is large is a paging or
17//! streaming call, which is worth telling apart rather than hiding in a total.
18//!
19//! Every method is forwarded, including the ones the traits give defaults for.
20//! A decorator that inherited a default instead of forwarding it would silently
21//! replace a provider's override: Google Drive mints its own object ids in
22//! `allocate_slot`, and inheriting the logical-key default would break it. The
23//! one method that is answered rather than forwarded is
24//! [`CloudHome::provider_requests`], which is how a run finds the counter.
25//!
26//! The wrapping happens where the home is built, not where a
27//! [`CloudSyncConnection`](crate::CloudSyncConnection) is, because one home
28//! outlives and is shared by several connections — a device join opens a
29//! plaintext one to pin the Store root and walk the membership chain, then an
30//! encrypted one for everything after, over the same provider. Counting per
31//! connection would split that join's operations across two totals and start
32//! the one it reports from zero.
33
34use super::*;
35use coven_foundation::stage_timing::ProviderRequests;
36use std::sync::atomic::{AtomicU64, Ordering};
37use std::sync::Arc;
38
39/// The running total of provider operations, shared between the home that
40/// counts them and whoever reports them. Cloning shares the total rather than
41/// copying it, which is what lets the home keep counting into the same number a
42/// run is already reading.
43#[derive(Clone, Debug, Default)]
44struct ProviderRequestCount(Arc<AtomicU64>);
45
46impl ProviderRequestCount {
47    fn record(&self) {
48        self.0.fetch_add(1, Ordering::Relaxed);
49    }
50}
51
52impl ProviderRequests for ProviderRequestCount {
53    fn issued(&self) -> u64 {
54        self.0.load(Ordering::Relaxed)
55    }
56}
57
58/// A cloud home that counts what is asked of it and forwards everything else.
59pub struct CountingCloudHome {
60    inner: Arc<dyn ExactCloudHome>,
61    count: ProviderRequestCount,
62}
63
64impl CountingCloudHome {
65    /// Wraps `inner` so every operation asked of it is counted. Whoever holds
66    /// the wrapped home reaches the running total through
67    /// [`CloudHome::provider_requests`], so the counter needs no separate route
68    /// from here to the runs that report it.
69    ///
70    /// Returns the wrapper itself rather than a boxed or reference-counted
71    /// home, because the three places that build a production home hand it on
72    /// in different containers.
73    pub fn new(inner: Arc<dyn ExactCloudHome>) -> Self {
74        Self {
75            inner,
76            count: ProviderRequestCount::default(),
77        }
78    }
79
80    fn counted(&self) -> &dyn ExactCloudHome {
81        self.count.record();
82        self.inner.as_ref()
83    }
84}
85
86#[async_trait]
87impl ExactSlotStorage for CountingCloudHome {
88    async fn provider_binding(
89        &self,
90    ) -> Result<coven_protocol::objects::ResolvedProviderBinding, CloudHomeError> {
91        self.counted().provider_binding().await
92    }
93
94    async fn cross_principal_evidence(
95        &self,
96    ) -> Result<coven_protocol::provider::CrossPrincipalProviderEvidence, CloudHomeError> {
97        self.counted().cross_principal_evidence().await
98    }
99
100    async fn allocate_slot(&self, logical_key: &str) -> Result<ObjectSlot, CloudHomeError> {
101        self.counted().allocate_slot(logical_key).await
102    }
103
104    async fn list_slots(&self, prefix: &str) -> Result<Vec<ObjectSlot>, CloudHomeError> {
105        self.counted().list_slots(prefix).await
106    }
107
108    async fn create_at(
109        &self,
110        upload: &ExactUpload<'_>,
111        control: &UploadControl,
112    ) -> Result<ExactCreateOutcome, CloudHomeError> {
113        self.counted().create_at(upload, control).await
114    }
115
116    async fn create_versioned_at(
117        &self,
118        upload: &ExactUpload<'_>,
119        control: &UploadControl,
120    ) -> Result<ExactCreateOutcome, CloudHomeError> {
121        self.counted().create_versioned_at(upload, control).await
122    }
123
124    async fn read_at(&self, slot: &ObjectSlot) -> Result<Vec<u8>, CloudHomeError> {
125        self.counted().read_at(slot).await
126    }
127
128    async fn read_versioned_at(
129        &self,
130        slot: &ObjectSlot,
131    ) -> Result<CloudVersionedObject, CloudHomeError> {
132        self.counted().read_versioned_at(slot).await
133    }
134
135    async fn replace_at_if_version(
136        &self,
137        slot: &ObjectSlot,
138        expected: &CloudObjectVersion,
139        bytes: Vec<u8>,
140    ) -> Result<ConditionalWriteOutcome, CloudHomeError> {
141        self.counted()
142            .replace_at_if_version(slot, expected, bytes)
143            .await
144    }
145
146    async fn observe_at(
147        &self,
148        slot: &ObjectSlot,
149    ) -> Result<Option<coven_protocol::objects::ExactObjectRef>, CloudHomeError> {
150        self.counted().observe_at(slot).await
151    }
152
153    async fn read_range_at(
154        &self,
155        slot: &ObjectSlot,
156        start: u64,
157        end: u64,
158    ) -> Result<Vec<u8>, CloudHomeError> {
159        self.counted().read_range_at(slot, start, end).await
160    }
161
162    async fn read_at_to_file(
163        &self,
164        slot: &ObjectSlot,
165        destination: &std::path::Path,
166        progress: DownloadProgress,
167    ) -> Result<(), CloudFileReadError> {
168        self.counted()
169            .read_at_to_file(slot, destination, progress)
170            .await
171    }
172
173    async fn delete_at(&self, slot: &ObjectSlot) -> Result<(), CloudHomeError> {
174        self.counted().delete_at(slot).await
175    }
176
177    async fn delete_versioned_at(&self, slot: &ObjectSlot) -> Result<(), CloudHomeError> {
178        self.counted().delete_versioned_at(slot).await
179    }
180
181    async fn delete_and_verify_absent(&self, slot: &ObjectSlot) -> Result<(), CloudHomeError> {
182        self.counted().delete_and_verify_absent(slot).await
183    }
184}
185
186#[async_trait]
187impl CloudHome for CountingCloudHome {
188    async fn probe(&self) -> Result<(), CloudHomeError> {
189        self.counted().probe().await
190    }
191
192    async fn put_object(&self, key: &str, data: Vec<u8>) -> Result<(), CloudHomeError> {
193        self.counted().put_object(key, data).await
194    }
195
196    async fn open_multipart<'a>(
197        &'a self,
198        key: &str,
199        total_len: u64,
200    ) -> Result<BoxPartSink<'a>, CloudHomeError> {
201        self.counted().open_multipart(key, total_len).await
202    }
203
204    /// A getter, not a request.
205    fn multipart_threshold(&self) -> u64 {
206        self.inner.multipart_threshold()
207    }
208
209    /// Answered here rather than forwarded: this is the counter, so this is
210    /// what a run reporting counts is looking for.
211    fn provider_requests(&self) -> Option<Arc<dyn ProviderRequests>> {
212        Some(Arc::new(self.count.clone()))
213    }
214
215    async fn write(
216        &self,
217        key: &str,
218        body: BlobBody,
219        progress: &UploadProgress,
220    ) -> Result<(), CloudHomeError> {
221        self.counted().write(key, body, progress).await
222    }
223
224    async fn read(&self, key: &str) -> Result<Vec<u8>, CloudHomeError> {
225        self.counted().read(key).await
226    }
227
228    async fn read_range(&self, key: &str, start: u64, end: u64) -> Result<Vec<u8>, CloudHomeError> {
229        self.counted().read_range(key, start, end).await
230    }
231
232    async fn list(&self, prefix: &str) -> Result<Vec<String>, CloudHomeError> {
233        self.counted().list(prefix).await
234    }
235
236    async fn delete(&self, key: &str) -> Result<(), CloudHomeError> {
237        self.counted().delete(key).await
238    }
239
240    async fn exists(&self, key: &str) -> Result<bool, CloudHomeError> {
241        self.counted().exists(key).await
242    }
243
244    async fn set_access(
245        &self,
246        desired: CloudAccessState,
247    ) -> Result<CloudAccessOutcome, CloudHomeError> {
248        self.counted().set_access(desired).await
249    }
250}
coven_storage/cloud/counting.rs

coven_storage/cloud/
counting.rs
