aven_core/sync/encrypted_tail/
server.rs

1use super::*;
2use crate::db::{Database, begin_immediate};
3use crate::sync::persistence::parent_liveness::ParentState;
4use anyhow::Context as _;
5use sqlx::SqliteConnection;
6
7async fn found(conn: &mut SqliteConnection, id: &str) -> Result<Option<Accepted>> {
8    let row: Option<(i64, Vec<u8>, Vec<u8>)> = sqlx::query_as(
9        "SELECT sequence,commitment,record FROM server_e2ee_tail WHERE operation_id=?",
10    )
11    .bind(id)
12    .fetch_optional(conn)
13    .await?;
14    row.map(|(sequence, commitment, record)| {
15        Ok(Accepted {
16            mapping: Mapping {
17                operation_id: id.into(),
18                sequence,
19                commitment: commitment
20                    .try_into()
21                    .map_err(|_| anyhow::anyhow!("error encrypted-tail-storage"))?,
22            },
23            record,
24        })
25    })
26    .transpose()
27}
28async fn mapping(conn: &mut SqliteConnection, id: &str) -> Result<Option<Mapping>> {
29    let row: Option<(i64, Vec<u8>)> =
30        sqlx::query_as("SELECT sequence,commitment FROM server_e2ee_tail WHERE operation_id=?")
31            .bind(id)
32            .fetch_optional(conn)
33            .await?;
34    row.map(|(sequence, commitment)| {
35        Ok(Mapping {
36            operation_id: id.into(),
37            sequence,
38            commitment: commitment
39                .try_into()
40                .map_err(|_| anyhow::anyhow!("error encrypted-tail-storage"))?,
41        })
42    })
43    .transpose()
44}
45fn serialized_len(value: &impl Serialize) -> Result<usize> {
46    struct Count(usize);
47    impl std::io::Write for Count {
48        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
49            self.0 += bytes.len();
50            Ok(bytes.len())
51        }
52        fn flush(&mut self) -> std::io::Result<()> {
53            Ok(())
54        }
55    }
56    let mut count = Count(0);
57    serde_json::to_writer(&mut count, value)?;
58    Ok(count.0)
59}
60async fn prefix(conn: &mut SqliteConnection, id: &str) -> Result<bool> {
61    Ok(sqlx::query_scalar(
62        "SELECT EXISTS(SELECT 1 FROM server_bootstrap_prefix WHERE operation_id=?)",
63    )
64    .bind(id)
65    .fetch_one(conn)
66    .await?)
67}
68impl Database {
69    pub async fn encrypted_tail_exchange(
70        &self,
71        context: &Context,
72        bearer: &super::super::seed_claim::Secret,
73        op: Operation,
74    ) -> Result<Reply> {
75        let mut conn = self.acquire_writer().await?;
76        let mut tx = begin_immediate(&mut conn).await?;
77        let current =
78            crate::sync::seed_claim::membership::persistence::current(self, &mut tx).await?;
79        current
80            .membership
81            .authenticate(&context.authentication(bearer), false)?;
82        let binding = current.membership.publication().binding();
83        valid(
84            context.stream == binding.stream_id
85                && context.descriptor == binding.descriptor_commitment,
86        )?;
87        let n = i64::try_from(binding.prefix_count)?;
88        let high = i64::try_from(
89            crate::sync::seed_claim::membership::persistence::allocator(
90                &mut tx,
91                &current.membership,
92            )
93            .await?,
94        )?;
95        let reply = match op {
96            Operation::Features => Reply::Features(BatchFeatures {
97                count: BATCH_COUNT,
98                bytes: BATCH_BYTES,
99            }),
100            Operation::Append { record, ticket } => {
101                let e = codec::parse(&record)?;
102                valid(e.vault == context.vault && e.stream == context.stream)?;
103                ensure!(
104                    !prefix(&mut tx, &e.id).await?,
105                    super::PrefixIdentityCollision
106                );
107                if let Some(old) = found(&mut tx, &e.id).await? {
108                    Reply::Appended(old.mapping)
109                } else {
110                    ensure!(
111                        !current.membership.rotation_pending(),
112                        "error membership-rotation-pending"
113                    );
114                    valid(e.generation == current.membership.current_generation().id)?;
115                    let sequence = high
116                        .checked_add(1)
117                        .context("error encrypted-tail-sequence-exhausted")?;
118                    let mapping = Mapping {
119                        operation_id: e.id.clone(),
120                        sequence,
121                        commitment: hash(&record),
122                    };
123                    sqlx::query("INSERT INTO server_e2ee_tail(operation_id,sequence,commitment,record) VALUES(?,?,?,?)")
124                        .bind(&e.id).bind(sequence).bind(mapping.commitment.as_slice()).bind(&record).execute(&mut *tx).await?;
125                    super::attachments::server::admit(
126                        &mut tx,
127                        context,
128                        &current.membership,
129                        &e.id,
130                        &e.projection,
131                        ticket.as_ref(),
132                    )
133                    .await?;
134                    apply_parent(&mut tx, &e.id, &e.projection).await?;
135                    sqlx::query("UPDATE server_e2ee_allocator SET high_water=? WHERE singleton=1")
136                        .bind(sequence)
137                        .execute(&mut *tx)
138                        .await?;
139                    Reply::Appended(mapping)
140                }
141            }
142            Operation::Lookup {
143                operation_id,
144                expected,
145            } => {
146                valid(!operation_id.is_empty() && operation_id.len() <= 256)?;
147                if let Some(want) = &expected {
148                    valid(want.operation_id == operation_id)?;
149                }
150                if prefix(&mut tx, &operation_id).await? {
151                    valid(expected.is_none())?;
152                    Reply::Bootstrap
153                } else if let Some(record) = found(&mut tx, &operation_id).await? {
154                    valid(expected.as_ref().is_none_or(|m| m == &record.mapping))?;
155                    Reply::Found(record)
156                } else {
157                    valid(expected.is_none())?;
158                    Reply::Absent
159                }
160            }
161            Operation::Pull {
162                after,
163                limit,
164                watermark,
165            } => {
166                let watermark = watermark.unwrap_or(high);
167                valid(
168                    after >= n
169                        && after <= watermark
170                        && watermark <= high
171                        && (1..=PAGE_COUNT).contains(&limit),
172                )?;
173                let rows:Vec<(String,i64,Vec<u8>,Vec<u8>)>=sqlx::query_as("SELECT operation_id,sequence,commitment,record FROM server_e2ee_tail WHERE sequence>? AND sequence<=? ORDER BY sequence LIMIT ?")
174                    .bind(after).bind(watermark).bind(limit as i64).fetch_all(&mut *tx).await?;
175                let mut records = Vec::new();
176                let mut size = 0;
177                let mut cursor = after;
178                for (id, sequence, commitment, record) in rows {
179                    let accepted = Accepted {
180                        mapping: Mapping {
181                            operation_id: id,
182                            sequence,
183                            commitment: commitment
184                                .try_into()
185                                .map_err(|_| anyhow::anyhow!("error encrypted-tail-storage"))?,
186                        },
187                        record,
188                    };
189                    // Measure the JSON the transport sends, plus one separator.
190                    let len = serialized_len(&accepted)? + 1;
191                    if size + len > PAGE_BYTES {
192                        break;
193                    }
194                    size += len;
195                    cursor = sequence;
196                    records.push(accepted);
197                }
198                let has_more:bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_tail WHERE sequence>? AND sequence<=?)").bind(cursor).bind(watermark).fetch_one(&mut *tx).await?;
199                valid(!has_more || cursor > after)?;
200                Reply::Page(Page {
201                    after,
202                    watermark,
203                    cursor,
204                    has_more,
205                    records,
206                })
207            }
208        };
209        tx.commit().await?;
210        Ok(reply)
211    }
212
213    pub async fn encrypted_tail_batch_exchange(
214        &self,
215        context: &Context,
216        bearer: &super::super::seed_claim::Secret,
217        op: BatchOperation,
218    ) -> Result<BatchReply> {
219        let mut conn = self.acquire_writer().await?;
220        let mut tx = begin_immediate(&mut conn).await?;
221        let current =
222            crate::sync::seed_claim::membership::persistence::current(self, &mut tx).await?;
223        current
224            .membership
225            .authenticate(&context.authentication(bearer), false)?;
226        let binding = current.membership.publication().binding();
227        valid(
228            context.stream == binding.stream_id
229                && context.descriptor == binding.descriptor_commitment,
230        )?;
231        let reply = match op {
232            BatchOperation::Append { records } => {
233                valid((1..=BATCH_COUNT).contains(&records.len()))?;
234                valid(
235                    records
236                        .iter()
237                        .try_fold(0usize, |total, record| total.checked_add(record.0.len()))
238                        .is_some_and(|total| total <= BATCH_BYTES),
239                )?;
240                let mut ids = std::collections::HashSet::new();
241                let mut parsed = Vec::with_capacity(records.len());
242                for record in &records {
243                    let envelope = codec::parse(&record.0)?;
244                    valid(
245                        envelope.vault == context.vault
246                            && envelope.stream == context.stream
247                            && !matches!(envelope.projection, domain::Projection::Ref { .. })
248                            && ids.insert(envelope.id.clone()),
249                    )?;
250                    ensure!(
251                        !prefix(&mut tx, &envelope.id).await?,
252                        super::PrefixIdentityCollision
253                    );
254                    ensure!(
255                        mapping(&mut tx, &envelope.id).await?.is_none(),
256                        "error encrypted-tail-batch-known"
257                    );
258                    parsed.push(envelope);
259                }
260                ensure!(
261                    !current.membership.rotation_pending(),
262                    "error membership-rotation-pending"
263                );
264                let generation = current.membership.current_generation().id;
265                valid(
266                    parsed
267                        .iter()
268                        .all(|envelope| envelope.generation == generation),
269                )?;
270                let high = i64::try_from(
271                    crate::sync::seed_claim::membership::persistence::allocator(
272                        &mut tx,
273                        &current.membership,
274                    )
275                    .await?,
276                )?;
277                let end = high
278                    .checked_add(i64::try_from(records.len())?)
279                    .context("error encrypted-tail-sequence-exhausted")?;
280                let mut mappings = Vec::with_capacity(records.len());
281                for (offset, (record, envelope)) in records.iter().zip(&parsed).enumerate() {
282                    let sequence = high
283                        .checked_add(i64::try_from(offset)? + 1)
284                        .context("error encrypted-tail-sequence-exhausted")?;
285                    let mapping = Mapping {
286                        operation_id: envelope.id.clone(),
287                        sequence,
288                        commitment: hash(&record.0),
289                    };
290                    sqlx::query("INSERT INTO server_e2ee_tail(operation_id,sequence,commitment,record) VALUES(?,?,?,?)")
291                        .bind(&envelope.id)
292                        .bind(sequence)
293                        .bind(mapping.commitment.as_slice())
294                        .bind(&record.0)
295                        .execute(&mut *tx)
296                        .await?;
297                    super::attachments::server::admit(
298                        &mut tx,
299                        context,
300                        &current.membership,
301                        &envelope.id,
302                        &envelope.projection,
303                        None,
304                    )
305                    .await?;
306                    apply_parent(&mut tx, &envelope.id, &envelope.projection).await?;
307                    mappings.push(mapping.into());
308                }
309                sqlx::query("UPDATE server_e2ee_allocator SET high_water=? WHERE singleton=1")
310                    .bind(end)
311                    .execute(&mut *tx)
312                    .await?;
313                BatchReply::Appended(mappings)
314            }
315            BatchOperation::Resolve { operation_ids } => {
316                valid((1..=BATCH_COUNT).contains(&operation_ids.len()))?;
317                let mut ids = std::collections::HashSet::new();
318                let mut resolutions = Vec::with_capacity(operation_ids.len());
319                for operation_id in operation_ids {
320                    valid(
321                        !operation_id.is_empty()
322                            && operation_id.len() <= 256
323                            && ids.insert(operation_id.clone()),
324                    )?;
325                    resolutions.push(if prefix(&mut tx, &operation_id).await? {
326                        Resolution::Bootstrap { operation_id }
327                    } else if let Some(mapping) = mapping(&mut tx, &operation_id).await? {
328                        Resolution::Found(mapping.into())
329                    } else {
330                        Resolution::Absent { operation_id }
331                    });
332                }
333                BatchReply::Resolved(resolutions)
334            }
335        };
336        tx.commit().await?;
337        Ok(reply)
338    }
339}
340async fn apply_parent(conn: &mut SqliteConnection, id: &str, p: &domain::Projection) -> Result<()> {
341    let domain::Projection::Parent {
342        action,
343        workspace,
344        task,
345        deleted,
346        version,
347    } = p
348    else {
349        return Ok(());
350    };
351    let workspace = super::moves::server_workspace(conn, task, workspace).await?;
352    let workspace = &workspace;
353    let existing:Option<(Option<String>,bool,bool)>=sqlx::query_as("SELECT version,deleted,protected FROM server_e2ee_image_parents WHERE workspace=? AND parent=?").bind(workspace).bind(task).fetch_optional(&mut *conn).await?;
354    if existing.is_none() {
355        let count: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_parents")
356            .fetch_one(&mut *conn)
357            .await?;
358        valid(count < 262144)?;
359    }
360    let mut state = existing
361        .map(|(version, deleted, protected)| ParentState {
362            version,
363            deleted,
364            protected,
365        })
366        .unwrap_or_default();
367    state.apply(*action, id, *deleted, version.as_deref());
368    sqlx::query("INSERT INTO server_e2ee_image_parents(workspace,parent,version,deleted,protected) VALUES(?,?,?,?,?) ON CONFLICT(workspace,parent) DO UPDATE SET version=excluded.version,deleted=excluded.deleted,protected=excluded.protected")
369        .bind(workspace).bind(task).bind(state.version).bind(state.deleted).bind(state.protected).execute(&mut *conn).await?;
370    super::attachments::server::retain_parent(conn, workspace, task).await?;
371    Ok(())
372}
373
374#[cfg(test)]
375mod rotation_tests;
aven_core/sync/encrypted_tail/server.rs

aven_core/sync/encrypted_tail/
server.rs
