aven_core/sync/client/
tail.rs

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

aven_core/sync/client/
tail.rs
