aven_core/sync/client/
tail.rs

1//! Encrypted ordinary-task transport: record rounds and image transfers.
2use anyhow::{Result, ensure};
3use aven_protocol::refusal::Tail as Refusal;
4use serde::Serialize;
5
6use super::errors::is_stale;
7use super::exchange::{self, Link};
8use super::keys::ProtectedLocalKeyStore;
9use super::keys::peer::{PublishingBlocked, TailSnapshot};
10use crate::db::Database;
11use crate::sync::{
12    encrypted_tail::{self as tail, Accepted, Context, Operation, Reply},
13    seed_claim::Secret,
14};
15mod images;
16pub use aven_protocol::wire::tail::Envelope;
17pub use aven_protocol::wire::tail::{BATCH_PATH, PATH};
18pub use images::{DrainSnapshot, IMAGES_PATH, ImageTransfer, Round};
19pub struct Client {
20    link: Link,
21    origin: url::Url,
22    locator: String,
23}
24
25pub enum PushStep {
26    Appended,
27    BatchAppended(usize),
28    Image(Option<ImageTransfer>),
29    Empty,
30}
31
32impl Client {
33    pub fn new(origin: &str, link: Link) -> Result<Self> {
34        Ok(Self {
35            link,
36            origin: super::origin::endpoint(origin, PATH)?,
37            locator: origin.into(),
38        })
39    }
40    fn enrollment(&self) -> Result<super::enrollment::Client> {
41        super::enrollment::Client::new(&self.locator, self.link.clone())
42    }
43    pub async fn exchange(
44        &self,
45        context: &Context,
46        bearer: &Secret,
47        operation: Operation,
48    ) -> Result<Reply> {
49        let response_limit = match &operation {
50            Operation::Append { .. } | Operation::Features => tail::CONTROL_LIMIT,
51            Operation::Lookup { .. } => tail::APPEND_LIMIT,
52            Operation::Pull { .. } => tail::RESPONSE_LIMIT,
53        };
54        let request_limit = if matches!(&operation, Operation::Append { .. }) {
55            tail::APPEND_LIMIT
56        } else {
57            tail::CONTROL_LIMIT
58        };
59        self.exchange_to(
60            PATH,
61            context,
62            bearer,
63            operation,
64            request_limit,
65            response_limit,
66        )
67        .await
68    }
69    async fn exchange_to<O: Serialize, R: serde::de::DeserializeOwned>(
70        &self,
71        path: &str,
72        context: &Context,
73        bearer: &Secret,
74        operation: O,
75        request_limit: usize,
76        response_limit: usize,
77    ) -> Result<R> {
78        let mut endpoint = self.origin.clone();
79        endpoint.set_path(path);
80        let mut correlation = [0; 32];
81        getrandom::fill(&mut correlation)
82            .map_err(|_| anyhow::anyhow!("error encrypted-tail-entropy"))?;
83        let compact = path == BATCH_PATH;
84        let bytes = if compact {
85            serde_json::to_vec(&tail::batch::Envelope {
86                context: context.clone(),
87                correlation,
88                operation,
89            })?
90        } else {
91            serde_json::to_vec(&Envelope {
92                context: context.clone(),
93                correlation,
94                operation,
95            })?
96        };
97        ensure!(bytes.len() <= request_limit, "error encrypted-tail-limit");
98        let bytes = exchange::post_json(&self.link, &endpoint, Some(bearer), bytes, response_limit)
99            .await
100            .map_err(|failure| match failure {
101                exchange::Failure::Hosting(hosting) => super::errors::hosting_error(hosting),
102                exchange::Failure::Network => {
103                    anyhow::anyhow!("error encrypted-tail-network outcome-unknown")
104                }
105                exchange::Failure::SecureTransport => {
106                    anyhow::anyhow!("error encrypted-tail-tls outcome-unknown")
107                }
108                exchange::Failure::Malformed => anyhow::anyhow!("error encrypted-tail-http"),
109                exchange::Failure::TooLarge => anyhow::anyhow!("error encrypted-tail-limit"),
110                exchange::Failure::RequestBodyLimit => {
111                    anyhow::anyhow!("error sync-request-body-limit")
112                }
113                exchange::Failure::Refused { code, .. } => match Refusal::classify(code.as_deref())
114                {
115                    Refusal::Malformed => {
116                        anyhow::anyhow!("error encrypted-tail-malformed")
117                    }
118                    Refusal::BatchKnown => {
119                        anyhow::anyhow!("error encrypted-tail-batch-known")
120                    }
121                    Refusal::Stale => crate::sync::seed_claim::membership::StaleContext.into(),
122                    // The server authenticated the credential but refused
123                    // this device, as enrollment does for a removed device.
124                    Refusal::Unauthorized => {
125                        crate::sync::seed_claim::membership::Unauthorized.into()
126                    }
127                    Refusal::PrefixIdentityCollision => tail::PrefixIdentityCollision.into(),
128                    Refusal::Quota => {
129                        anyhow::anyhow!("error attachment-quota-exceeded")
130                    }
131                    Refusal::Unknown => {
132                        anyhow::anyhow!("error encrypted-tail-refused outcome-unknown")
133                    }
134                },
135            })?;
136        let response: Envelope<R> = if compact {
137            let reply: tail::batch::Envelope<R> = serde_json::from_slice(&bytes)
138                .map_err(|_| anyhow::anyhow!("error encrypted-tail-http"))?;
139            Envelope {
140                context: reply.context,
141                correlation: reply.correlation,
142                operation: reply.operation,
143            }
144        } else {
145            serde_json::from_slice(&bytes)
146                .map_err(|_| anyhow::anyhow!("error encrypted-tail-http"))?
147        };
148        ensure!(
149            response.context == *context && response.correlation == correlation,
150            "error encrypted-tail-context"
151        );
152        Ok(response.operation)
153    }
154    pub async fn batch_exchange(
155        &self,
156        context: &Context,
157        bearer: &Secret,
158        operation: tail::BatchOperation,
159    ) -> Result<tail::BatchReply> {
160        let limit = if matches!(operation, tail::BatchOperation::Append { .. }) {
161            tail::BATCH_APPEND_LIMIT
162        } else {
163            tail::BATCH_CONTROL_LIMIT
164        };
165        self.exchange_to(
166            BATCH_PATH,
167            context,
168            bearer,
169            operation,
170            limit,
171            tail::BATCH_CONTROL_LIMIT,
172        )
173        .await
174    }
175
176    /// Every unresolved operation is looked up before any upload or resend.
177    async fn reconcile_frozen(
178        &self,
179        a: &tail::Authority,
180        bearer: &Secret,
181        db: &Database,
182        blob_dir: &std::path::Path,
183        batch_count: usize,
184    ) -> Result<bool> {
185        let frozen = db.encrypted_tail_frozen_records(a).await?;
186        let mut responses = Vec::with_capacity(frozen.len());
187        if frozen.len() > 1 && batch_count > 1 {
188            let tail::BatchReply::Resolved(resolutions) = self
189                .batch_exchange(
190                    &a.context,
191                    bearer,
192                    tail::BatchOperation::Resolve {
193                        operation_ids: frozen.iter().map(|(id, _)| id.clone()).collect(),
194                    },
195                )
196                .await?
197            else {
198                anyhow::bail!("error encrypted-tail-reply")
199            };
200            ensure!(
201                resolutions.len() == frozen.len(),
202                "error encrypted-tail-mapping"
203            );
204            for ((id, record), resolution) in frozen.iter().zip(resolutions) {
205                responses.push(match resolution {
206                    tail::Resolution::Found(mapping) => {
207                        ensure!(mapping.operation_id == *id, "error encrypted-tail-mapping");
208                        Reply::Found(Accepted {
209                            mapping: mapping.into(),
210                            record: record.clone(),
211                        })
212                    }
213                    tail::Resolution::Absent { operation_id } => {
214                        ensure!(operation_id == *id, "error encrypted-tail-mapping");
215                        Reply::Absent
216                    }
217                    tail::Resolution::Bootstrap { .. } => {
218                        anyhow::bail!("error encrypted-tail-accepted-identity")
219                    }
220                });
221            }
222        } else {
223            for (id, _) in &frozen {
224                responses.push(
225                    self.exchange(
226                        &a.context,
227                        bearer,
228                        Operation::Lookup {
229                            operation_id: id.clone(),
230                            expected: None,
231                        },
232                    )
233                    .await?,
234                );
235            }
236        }
237        crate::sync::crash::Crash::Tail.at("after-resolve");
238        let mut absences = Vec::new();
239        for ((id, record), response) in frozen.iter().zip(responses) {
240            match response {
241                Reply::Found(mut accepted) => {
242                    ensure!(
243                        accepted.mapping.operation_id == *id,
244                        "error encrypted-tail-accepted-identity"
245                    );
246                    db.observe_encrypted_tail(a, &accepted.mapping).await?;
247                    crate::sync::crash::Crash::Tail.at("batch-observed");
248                    use sha2::{Digest, Sha256};
249                    if Sha256::digest(&accepted.record).as_slice() != accepted.mapping.commitment {
250                        let Reply::Found(found) = self
251                            .exchange(
252                                &a.context,
253                                bearer,
254                                Operation::Lookup {
255                                    operation_id: id.clone(),
256                                    expected: Some(accepted.mapping.clone()),
257                                },
258                            )
259                            .await?
260                        else {
261                            anyhow::bail!("error encrypted-tail-accepted-unavailable")
262                        };
263                        ensure!(
264                            found.mapping == accepted.mapping,
265                            "error encrypted-tail-mapping"
266                        );
267                        accepted = found;
268                    }
269                    db.verify_encrypted_tail_outcome(a, &accepted).await?;
270                }
271                Reply::Absent => absences.push(a.confirm_absent(record, &Reply::Absent)?),
272                _ => anyhow::bail!("error encrypted-tail-accepted-identity"),
273            }
274        }
275        if !absences.is_empty()
276            && !db
277                .reconcile_encrypted_tail_absences(a, &absences, blob_dir)
278                .await?
279        {
280            return Ok(false);
281        }
282        Ok(!a.rotation_pending())
283    }
284    #[cfg(any(test, feature = "test-support"))]
285    pub async fn push(
286        &self,
287        inputs: &TailSnapshot,
288        db: &Database,
289        blob_dir: &std::path::Path,
290    ) -> Result<PushStep> {
291        Ok(self
292            .push_in_run(inputs, db, blob_dir, None, &mut 1, 1)
293            .await?
294            .0)
295    }
296    /// Dispatches the next ordered head. An unavailable local image source or
297    /// failed image transfer leaves that head pending and stops this round's push
298    /// phase instead of appending its Ref. Fails with `PublishingBlocked`
299    /// before any upload or append while a withdrawal rotation is required.
300    async fn push_in_run(
301        &self,
302        inputs: &TailSnapshot,
303        db: &Database,
304        blob_dir: &std::path::Path,
305        preflight: Option<tail::Preflight>,
306        batch_count: &mut usize,
307        remaining: usize,
308    ) -> Result<(PushStep, Option<tail::Preflight>)> {
309        inputs.require_publishing_ready()?;
310        let (a, bearer) = (&inputs.authority, &inputs.bearer);
311        if *batch_count == 0 && db.encrypted_tail_has_batch_work(a).await? {
312            *batch_count = match self.exchange(&a.context, bearer, Operation::Features).await {
313                Ok(Reply::Features(features))
314                    if features.count >= tail::BATCH_COUNT
315                        && features.bytes >= tail::BATCH_BYTES =>
316                {
317                    tail::BATCH_COUNT
318                }
319                Ok(Reply::Features(_)) => 1,
320                Err(error) if super::errors::has_code(&error, "encrypted-tail-malformed") => 1,
321                Err(error) => return Err(error),
322                _ => anyhow::bail!("error encrypted-tail-reply"),
323            };
324        }
325        if !self
326            .reconcile_frozen(a, bearer, db, blob_dir, *batch_count)
327            .await?
328        {
329            return Ok((PushStep::Empty, preflight));
330        }
331        // A missing local source leaves its head pending without blocking pulls.
332        let (prepared, preflight) = match db
333            .prepare_encrypted_batch_in_run(
334                a,
335                blob_dir,
336                preflight.clone(),
337                (*batch_count).max(1).min(remaining),
338            )
339            .await
340        {
341            Err(error) if error.is::<tail::attachments::ImageSourceUnavailable>() => {
342                return Ok((PushStep::Image(Some(ImageTransfer::Failed)), preflight));
343            }
344            prepared => prepared?,
345        };
346        let Some(tail::Push { record, upload }) = prepared else {
347            return Ok((PushStep::Empty, preflight));
348        };
349        let frozen = db.encrypted_tail_frozen_records(a).await?;
350        if frozen.len() > 1 && *batch_count > 1 {
351            inputs.require_publishing_ready()?;
352            let tail::BatchReply::Appended(mappings) = self
353                .batch_exchange(
354                    &a.context,
355                    bearer,
356                    tail::BatchOperation::Append {
357                        records: frozen
358                            .iter()
359                            .map(|(_, r)| tail::BatchRecord(r.clone()))
360                            .collect(),
361                    },
362                )
363                .await?
364            else {
365                anyhow::bail!("error encrypted-tail-reply")
366            };
367            crate::sync::crash::Crash::Tail.at("after-append");
368            crate::sync::crash::Crash::Tail.at("after-batch-append");
369            ensure!(
370                mappings.len() == frozen.len(),
371                "error encrypted-tail-mapping"
372            );
373            use sha2::{Digest, Sha256};
374            for ((id, record), mapping) in frozen.iter().zip(&mappings) {
375                ensure!(
376                    mapping.operation_id == *id
377                        && Sha256::digest(record).as_slice() == mapping.commitment,
378                    "error encrypted-tail-mapping"
379                );
380            }
381            ensure!(
382                mappings
383                    .windows(2)
384                    .all(|pair| pair[0].sequence.checked_add(1) == Some(pair[1].sequence)),
385                "error encrypted-tail-mapping"
386            );
387            let mappings: Vec<tail::Mapping> = mappings.into_iter().map(Into::into).collect();
388            db.accept_encrypted_tail_batch(a, &mappings).await?;
389            return Ok((PushStep::BatchAppended(mappings.len()), preflight));
390        }
391        let is_image = upload.is_some();
392        let ticket = match upload {
393            Some(upload) => match self.upload_prepared_image(inputs, upload).await {
394                Ok(ticket) => Some(ticket),
395                Err(error)
396                    if is_stale(&error)
397                        || super::errors::is_hosting_refusal(&error)
398                        || error.is::<PublishingBlocked>()
399                        || super::errors::has_code(&error, "attachment-quota-exceeded")
400                        // A proxy body limit refuses every retry identically.
401                        || super::errors::has_code(&error, "sync-request-body-limit") =>
402                {
403                    return Err(error);
404                }
405                Err(_) => {
406                    return Ok((PushStep::Image(Some(ImageTransfer::Failed)), preflight));
407                }
408            },
409            None => None,
410        };
411        inputs.require_publishing_ready()?;
412        let Reply::Appended(mapping) = self
413            .exchange(&a.context, bearer, Operation::Append { ticket, record })
414            .await?
415        else {
416            anyhow::bail!("error encrypted-tail-reply")
417        };
418        crate::sync::crash::Crash::Tail.at("after-append");
419        let frozen = db.observe_encrypted_tail(a, &mapping).await?;
420        use sha2::{Digest, Sha256};
421        let accepted = if Sha256::digest(&frozen).as_slice() == mapping.commitment {
422            Accepted {
423                mapping,
424                record: frozen,
425            }
426        } else {
427            let Reply::Found(record) = self
428                .exchange(
429                    &a.context,
430                    bearer,
431                    Operation::Lookup {
432                        operation_id: mapping.operation_id.clone(),
433                        expected: Some(mapping.clone()),
434                    },
435                )
436                .await?
437            else {
438                anyhow::bail!("error encrypted-tail-accepted-unavailable")
439            };
440            ensure!(record.mapping == mapping, "error encrypted-tail-mapping");
441            record
442        };
443        db.verify_encrypted_tail_outcome(a, &accepted).await?;
444        Ok((
445            if is_image {
446                PushStep::Image(None)
447            } else {
448                PushStep::Appended
449            },
450            preflight,
451        ))
452    }
453    /// Reads one authorized page without preparing uploads. True refers only to
454    /// this remote watermark, never to unresolved local work or overall readiness.
455    #[cfg(any(test, feature = "test-support"))]
456    pub async fn pull_only_round(
457        &self,
458        store: &ProtectedLocalKeyStore,
459        db: &Database,
460    ) -> Result<bool> {
461        let enrollment = self.enrollment()?;
462        enrollment.refresh(store, db).await?;
463        retry_stale!(
464            self.pull_only_round_once(store, db).await,
465            enrollment.refresh(store, db).await,
466        )
467    }
468    #[cfg(any(test, feature = "test-support"))]
469    async fn pull_only_round_once(
470        &self,
471        store: &ProtectedLocalKeyStore,
472        db: &Database,
473    ) -> Result<bool> {
474        let inputs = store.tail_inputs(db, &self.locator).await?;
475        db.validate_encrypted_attachment_integrity(&inputs.authority)
476            .await?;
477        self.pull(&inputs.authority, &inputs.bearer, db).await
478    }
479    async fn pull(&self, a: &tail::Authority, bearer: &Secret, db: &Database) -> Result<bool> {
480        let state = db.encrypted_round_state(a).await?;
481        let (after, watermark) = (state.cursor, state.initial_watermark);
482        let Reply::Page(page) = self
483            .exchange(
484                &a.context,
485                bearer,
486                Operation::Pull {
487                    after,
488                    limit: tail::PAGE_COUNT,
489                    watermark,
490                },
491            )
492            .await?
493        else {
494            anyhow::bail!("error encrypted-tail-reply")
495        };
496        ensure!(
497            page.after == after && watermark.is_none_or(|w| page.watermark == w),
498            "error encrypted-tail-cursor"
499        );
500        db.apply_encrypted_tail_page(a, &page).await?;
501        Ok(!page.has_more)
502    }
503}
aven_core/sync/client/tail.rs

aven_core/sync/client/
tail.rs
