aven_core/sync/encrypted_tail/attachments/
server.rs

1use super::super::{Context, hash, valid};
2use super::{codec::Descriptor, *};
3use crate::sync::seed_claim::membership::Membership;
4use crate::{
5    attachments::lifecycle::LifecyclePolicy,
6    db::{Database, begin_immediate},
7    sync::seed_claim::Secret,
8};
9use anyhow::{Context as _, Result, ensure};
10use sqlx::SqliteConnection;
11
12async fn load(
13    conn: &mut SqliteConnection,
14    object: &[u8; 32],
15    commitment: &[u8; 32],
16) -> Result<(Descriptor, bool)> {
17    let (bytes, complete): (Vec<u8>, bool) =
18        sqlx::query_as("SELECT descriptor,complete FROM server_e2ee_images WHERE object=?")
19            .bind(object.as_slice())
20            .fetch_optional(conn)
21            .await?
22            .context("error encrypted-image-unknown")?;
23    valid(hash(&bytes) == *commitment)?;
24    let d = Descriptor::decode(&bytes)?;
25    valid(d.object == *object)?;
26    Ok((d, complete))
27}
28/// Stamps newly unreferenced images and clears the stamp from images that are
29/// referenced or ticketed again. Only the background prune runs this full
30/// scan, so an image's grace period starts at the first prune pass that
31/// observes it unreferenced.
32async fn refresh(conn: &mut SqliteConnection, now: i64) -> Result<()> {
33    sqlx::query("UPDATE server_e2ee_images SET unreferenced_at=CASE WHEN EXISTS(SELECT 1 FROM server_e2ee_image_references r JOIN server_e2ee_image_parents p ON p.workspace=r.workspace AND p.parent=r.parent WHERE r.object=server_e2ee_images.object AND r.deleted=0 AND (p.deleted=0 OR p.protected=1 OR p.version IS NULL)) OR EXISTS(SELECT 1 FROM server_e2ee_image_tickets t WHERE t.object=server_e2ee_images.object AND t.expires_at>?) THEN NULL ELSE COALESCE(unreferenced_at,?) END").bind(now).bind(now).execute(conn).await?;
34    Ok(())
35}
36/// Clears the unreferenced stamp of images a live parent references, so a
37/// parent that is deleted again later gets a full grace period.
38pub(in crate::sync::encrypted_tail) async fn retain_parent(
39    conn: &mut SqliteConnection,
40    workspace: &str,
41    parent: &str,
42) -> Result<()> {
43    sqlx::query("UPDATE server_e2ee_images SET unreferenced_at=NULL WHERE unreferenced_at IS NOT NULL AND object IN (SELECT r.object FROM server_e2ee_image_references r JOIN server_e2ee_image_parents p ON p.workspace=r.workspace AND p.parent=r.parent WHERE r.workspace=? AND r.parent=? AND r.deleted=0 AND (p.deleted=0 OR p.protected=1 OR p.version IS NULL))").bind(workspace).bind(parent).execute(conn).await?;
44    Ok(())
45}
46async fn protected(
47    conn: &mut SqliteConnection,
48    object: &[u8; 32],
49    workspace: &str,
50) -> Result<bool> {
51    Ok(sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_references r JOIN server_e2ee_image_parents p ON p.workspace=r.workspace AND p.parent=r.parent WHERE r.object=? AND r.workspace=? AND r.deleted=0 AND (p.deleted=0 OR p.protected=1 OR p.version IS NULL))").bind(object.as_slice()).bind(workspace).fetch_one(conn).await?)
52}
53async fn scope(conn: &mut SqliteConnection, object: &[u8; 32], workspace: &str) -> Result<()> {
54    valid(
55        sqlx::query_scalar(
56            "SELECT EXISTS(SELECT 1 FROM server_e2ee_image_scopes WHERE object=? AND workspace=?)",
57        )
58        .bind(object.as_slice())
59        .bind(workspace)
60        .fetch_one(conn)
61        .await?,
62    )
63}
64async fn reserve(
65    conn: &mut SqliteConnection,
66    d: &Descriptor,
67    workspace: &str,
68    device: &[u8; 32],
69    quota: i64,
70    now: i64,
71) -> Result<()> {
72    sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE expires_at<=?")
73        .bind(now)
74        .execute(&mut *conn)
75        .await?;
76    let existing: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_tickets WHERE object=? AND workspace=? AND device=? AND expires_at>?)").bind(d.object.as_slice()).bind(workspace).bind(device.as_slice()).bind(now).fetch_one(&mut *conn).await?;
77    if existing {
78        return Ok(());
79    }
80    let reactivating: bool = sqlx::query_scalar("SELECT origin IS NULL AND NOT EXISTS(SELECT 1 FROM server_e2ee_image_chunks c WHERE c.object=i.object) AND NOT EXISTS(SELECT 1 FROM server_e2ee_image_tickets t WHERE t.object=i.object AND t.expires_at>?) FROM server_e2ee_images i WHERE object=?")
81        .bind(now).bind(d.object.as_slice()).fetch_one(&mut *conn).await?;
82    if reactivating {
83        let active: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_images i WHERE origin IS NULL AND (EXISTS(SELECT 1 FROM server_e2ee_image_chunks c WHERE c.object=i.object) OR EXISTS(SELECT 1 FROM server_e2ee_image_tickets t WHERE t.object=i.object AND t.expires_at>?))")
84            .bind(now).fetch_one(&mut *conn).await?;
85        valid(active < 128)?;
86    }
87    let counted = protected(conn, &d.object, workspace).await? || sqlx::query_scalar::<_,bool>("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_tickets WHERE object=? AND workspace=? AND expires_at>?)").bind(d.object.as_slice()).bind(workspace).bind(now).fetch_one(&mut *conn).await?;
88    if !counted {
89        let used: i64 = sqlx::query_scalar("SELECT COALESCE(SUM(i.byte_size),0) FROM server_e2ee_images i WHERE EXISTS(SELECT 1 FROM server_e2ee_image_references r JOIN server_e2ee_image_parents p ON p.workspace=r.workspace AND p.parent=r.parent WHERE r.object=i.object AND r.workspace=? AND r.deleted=0 AND (p.deleted=0 OR p.protected=1 OR p.version IS NULL)) OR EXISTS(SELECT 1 FROM server_e2ee_image_tickets t WHERE t.object=i.object AND t.workspace=? AND t.expires_at>?)").bind(workspace).bind(workspace).bind(now).fetch_one(&mut *conn).await?;
90        ensure!(
91            used.checked_add(d.byte_size()).is_some_and(|n| n <= quota),
92            "error attachment-quota-exceeded"
93        );
94    }
95    let count: i64 =
96        sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_tickets WHERE expires_at>?")
97            .bind(now)
98            .fetch_one(&mut *conn)
99            .await?;
100    valid(count < 128)?;
101    let mut id = [0; 32];
102    getrandom::fill(&mut id).map_err(|_| anyhow::anyhow!("error encrypted-image-entropy"))?;
103    let expires = now
104        .checked_add(crate::attachments::lifecycle::LEASE_TTL.as_secs() as i64)
105        .context("error encrypted-image-clock")?;
106    sqlx::query("INSERT INTO server_e2ee_image_tickets(reservation,object,workspace,device,expires_at) VALUES(?,?,?,?,?) ON CONFLICT(object,workspace,device) DO UPDATE SET reservation=excluded.reservation,expires_at=excluded.expires_at")
107        .bind(id.as_slice()).bind(d.object.as_slice()).bind(workspace).bind(device.as_slice()).bind(expires).execute(&mut *conn).await?;
108    sqlx::query("UPDATE server_e2ee_images SET unreferenced_at=NULL WHERE object=?")
109        .bind(d.object.as_slice())
110        .execute(conn)
111        .await?;
112    Ok(())
113}
114async fn ticket(
115    conn: &mut SqliteConnection,
116    object: &[u8; 32],
117    workspace: &str,
118    device: &[u8; 32],
119    ticket: &Ticket,
120    now: i64,
121) -> Result<()> {
122    valid(sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_tickets WHERE object=? AND workspace=? AND device=? AND reservation=? AND expires_at>?)").bind(object.as_slice()).bind(workspace).bind(device.as_slice()).bind(ticket.reservation.as_slice()).bind(now).fetch_one(conn).await?)
123}
124async fn status(
125    conn: &mut SqliteConnection,
126    d: &Descriptor,
127    workspace: &str,
128    device: &[u8; 32],
129    now: i64,
130) -> Result<Reply> {
131    let complete: bool =
132        sqlx::query_scalar("SELECT complete FROM server_e2ee_images WHERE object=?")
133            .bind(d.object.as_slice())
134            .fetch_one(&mut *conn)
135            .await?;
136    let indices: Vec<i64> = sqlx::query_scalar(
137        "SELECT chunk_index FROM server_e2ee_image_chunks WHERE object=? ORDER BY chunk_index",
138    )
139    .bind(d.object.as_slice())
140    .fetch_all(&mut *conn)
141    .await?;
142    let t:Option<(Vec<u8>,i64)>=sqlx::query_as("SELECT reservation,expires_at FROM server_e2ee_image_tickets WHERE object=? AND workspace=? AND device=? AND expires_at>?").bind(d.object.as_slice()).bind(workspace).bind(device.as_slice()).bind(now).fetch_optional(conn).await?;
143    let (reservation, expires_at) = match t {
144        Some((id, e)) => (
145            Some(
146                id.try_into()
147                    .map_err(|_| anyhow::anyhow!("error encrypted-image-storage"))?,
148            ),
149            Some(e),
150        ),
151        None => (None, None),
152    };
153    Ok(Reply::Status(Status {
154        complete,
155        missing: (0..d.artifact.chunks.len())
156            .filter(|i| !indices.contains(&(*i as i64)))
157            .collect(),
158        reservation,
159        expires_at,
160    }))
161}
162async fn eligible_object(
163    conn: &mut SqliteConnection,
164    m: &Membership,
165    d: &Descriptor,
166) -> Result<()> {
167    valid(m.generations().iter().any(|g| g.id == d.generation))?;
168    if d.generation != m.current_generation().id {
169        let admitted: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_images WHERE object=? AND origin IS NOT NULL AND descriptor=?)")
170            .bind(d.object.as_slice()).bind(d.encode()?).fetch_one(conn).await?;
171        valid(admitted)?;
172    }
173    Ok(())
174}
175impl Database {
176    /// Deletes the stored bytes of at most `limit` images that have been
177    /// unreferenced and unticketed for at least `grace`, keeping their
178    /// descriptors. Returns how many images were pruned.
179    pub async fn prune_encrypted_images(
180        &self,
181        grace: std::time::Duration,
182        limit: usize,
183    ) -> Result<usize> {
184        let mut conn = self.acquire_writer().await?;
185        let mut tx = begin_immediate(&mut conn).await?;
186        let now = chrono::Utc::now().timestamp();
187        refresh(&mut tx, now).await?;
188        let cutoff = now
189            .checked_sub(i64::try_from(grace.as_secs())?)
190            .context("error encrypted-image-clock")?;
191        let objects:Vec<Vec<u8>>=sqlx::query_scalar("SELECT object FROM server_e2ee_images WHERE unreferenced_at<=? AND (complete=1 OR EXISTS(SELECT 1 FROM server_e2ee_image_chunks c WHERE c.object=server_e2ee_images.object)) ORDER BY object LIMIT ?").bind(cutoff).bind(i64::try_from(limit)?).fetch_all(&mut *tx).await?;
192        // Objects with live tickets were excluded by `refresh`, so no
193        // uploader can still write to a pruned object.
194        for object in &objects {
195            sqlx::query("UPDATE server_e2ee_images SET complete=0 WHERE object=?")
196                .bind(object)
197                .execute(&mut *tx)
198                .await?;
199            sqlx::query("DELETE FROM server_e2ee_image_chunks WHERE object=?")
200                .bind(object)
201                .execute(&mut *tx)
202                .await?;
203        }
204        tx.commit().await?;
205        Ok(objects.len())
206    }
207    pub async fn encrypted_image_exchange(
208        &self,
209        context: &Context,
210        bearer: &Secret,
211        op: Operation,
212        policy: LifecyclePolicy,
213    ) -> Result<Reply> {
214        let mut conn = self.acquire_writer().await?;
215        let mut tx = begin_immediate(&mut conn).await?;
216        let current =
217            crate::sync::seed_claim::membership::persistence::current(self, &mut tx).await?;
218        current
219            .membership
220            .authenticate(&context.authentication(bearer), false)?;
221        ensure!(
222            !current.membership.rotation_pending()
223                || matches!(&op, Operation::Status { .. } | Operation::Read { .. }),
224            "error membership-rotation-pending"
225        );
226        let binding = current.membership.publication().binding();
227        valid(
228            context.stream == binding.stream_id
229                && context.descriptor == binding.descriptor_commitment,
230        )?;
231        let now = chrono::Utc::now().timestamp();
232        let reply = match op {
233            Operation::Declare {
234                workspace,
235                descriptor,
236            } => {
237                workspace.parse::<crate::ids::WorkspaceId>()?;
238                let d = Descriptor::decode(&descriptor)?;
239                valid(d.vault == context.vault && d.stream == context.stream)?;
240                eligible_object(&mut tx, &current.membership, &d).await?;
241                let old: Option<Vec<u8>> =
242                    sqlx::query_scalar("SELECT descriptor FROM server_e2ee_images WHERE object=?")
243                        .bind(d.object.as_slice())
244                        .fetch_optional(&mut *tx)
245                        .await?;
246                if let Some(old) = old {
247                    valid(old == descriptor)?;
248                } else {
249                    let total: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_images")
250                        .fetch_one(&mut *tx)
251                        .await?;
252                    valid(total < 65536)?;
253                    sqlx::query("INSERT INTO server_e2ee_images(object,byte_size,descriptor,complete,unreferenced_at) VALUES(?,?,?,0,?)").bind(d.object.as_slice()).bind(d.byte_size()).bind(&descriptor).bind(now).execute(&mut *tx).await?;
254                }
255                let known:bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_scopes WHERE object=? AND workspace=?)").bind(d.object.as_slice()).bind(&workspace).fetch_one(&mut *tx).await?;
256                if !known {
257                    let count: i64 =
258                        sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_scopes")
259                            .fetch_one(&mut *tx)
260                            .await?;
261                    valid(count < 262144)?;
262                }
263                sqlx::query(
264                    "INSERT OR IGNORE INTO server_e2ee_image_scopes(object,workspace) VALUES(?,?)",
265                )
266                .bind(d.object.as_slice())
267                .bind(&workspace)
268                .execute(&mut *tx)
269                .await?;
270                load(&mut tx, &d.object, &hash(&descriptor)).await?;
271                reserve(
272                    &mut tx,
273                    &d,
274                    &workspace,
275                    &context.device,
276                    policy.quota_bytes,
277                    now,
278                )
279                .await?;
280                status(&mut tx, &d, &workspace, &context.device, now).await?
281            }
282            other => dispatch_existing(&mut tx, context, &current.membership, other, now).await?,
283        };
284        tx.commit().await?;
285        Ok(reply)
286    }
287}
288async fn dispatch_existing(
289    conn: &mut SqliteConnection,
290    context: &Context,
291    membership: &Membership,
292    op: Operation,
293    now: i64,
294) -> Result<Reply> {
295    let (workspace, object, commitment) = match &op {
296        Operation::Status {
297            workspace,
298            object,
299            descriptor_commitment,
300        }
301        | Operation::Put {
302            workspace,
303            object,
304            descriptor_commitment,
305            ..
306        }
307        | Operation::Complete {
308            workspace,
309            object,
310            descriptor_commitment,
311            ..
312        }
313        | Operation::Read {
314            workspace,
315            object,
316            descriptor_commitment,
317            ..
318        }
319        | Operation::Release {
320            workspace,
321            object,
322            descriptor_commitment,
323            ..
324        } => (workspace, *object, *descriptor_commitment),
325        _ => unreachable!(),
326    };
327    workspace.parse::<crate::ids::WorkspaceId>()?;
328    scope(conn, &object, workspace).await?;
329    let (d, _) = load(conn, &object, &commitment).await?;
330    valid(d.vault == context.vault && d.stream == context.stream)?;
331    eligible_object(conn, membership, &d).await?;
332    match &op {
333        Operation::Status { .. } => status(conn, &d, workspace, &context.device, now).await,
334        Operation::Read { index, .. } => {
335            valid(*index < d.artifact.chunks.len())?;
336            let admitted: bool = sqlx::query_scalar(
337                "SELECT origin IS NOT NULL FROM server_e2ee_images WHERE object=?",
338            )
339            .bind(object.as_slice())
340            .fetch_one(&mut *conn)
341            .await?;
342            valid(admitted)?;
343            let bytes: Option<Vec<u8>> = sqlx::query_scalar(
344                "SELECT bytes FROM server_e2ee_image_chunks WHERE object=? AND chunk_index=?",
345            )
346            .bind(object.as_slice())
347            .bind(*index as i64)
348            .fetch_optional(conn)
349            .await?;
350            Ok(match bytes {
351                Some(b) => {
352                    d.verify_chunk(*index, &b)?;
353                    Reply::Chunk(b)
354                }
355                None => Reply::Unavailable,
356            })
357        }
358        Operation::Put { reservation, .. }
359        | Operation::Complete { reservation, .. }
360        | Operation::Release { reservation, .. } => {
361            ticket(
362                conn,
363                &object,
364                workspace,
365                &context.device,
366                &Ticket {
367                    reservation: *reservation,
368                },
369                now,
370            )
371            .await?;
372            match &op {
373                Operation::Put { index, record, .. } => {
374                    d.verify_chunk(*index, record)?;
375                    let old:Option<Vec<u8>>=sqlx::query_scalar("SELECT bytes FROM server_e2ee_image_chunks WHERE object=? AND chunk_index=?").bind(object.as_slice()).bind(*index as i64).fetch_optional(&mut *conn).await?;
376                    valid(old.as_ref().is_none_or(|b| b == record))?;
377                    sqlx::query("INSERT OR IGNORE INTO server_e2ee_image_chunks(object,chunk_index,bytes) VALUES(?,?,?)").bind(object.as_slice()).bind(*index as i64).bind(record).execute(&mut *conn).await?;
378                }
379                Operation::Complete { .. } => {
380                    let records:Vec<Vec<u8>>=sqlx::query_scalar("SELECT bytes FROM server_e2ee_image_chunks WHERE object=? ORDER BY chunk_index").bind(object.as_slice()).fetch_all(&mut *conn).await?;
381                    d.verify(&records)?;
382                    sqlx::query("UPDATE server_e2ee_images SET complete=1 WHERE object=?")
383                        .bind(object.as_slice())
384                        .execute(&mut *conn)
385                        .await?;
386                }
387                Operation::Release { .. } => {
388                    sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE reservation=?")
389                        .bind(reservation.as_slice())
390                        .execute(&mut *conn)
391                        .await?;
392                }
393                _ => unreachable!(),
394            }
395            Ok(Reply::Done)
396        }
397        _ => unreachable!(),
398    }
399}
400
401pub(in crate::sync::encrypted_tail) async fn admit(
402    conn: &mut SqliteConnection,
403    context: &Context,
404    membership: &Membership,
405    id: &str,
406    p: &super::super::domain::Projection,
407    t: Option<&Ticket>,
408) -> Result<()> {
409    use super::super::domain::Projection;
410    let now = chrono::Utc::now().timestamp();
411    match p {
412        Projection::Move {
413            source,
414            target,
415            tasks,
416        } => {
417            valid(t.is_none())?;
418            super::super::moves::admit(conn, source, target, tasks).await?;
419        }
420        Projection::Ref {
421            workspace,
422            task,
423            reference,
424            descriptor,
425            deleted,
426            version,
427        } => {
428            let authored_workspace = workspace;
429            let routed = super::super::moves::server_workspace(conn, task, workspace).await?;
430            let workspace = &routed;
431            let d = Descriptor::decode(descriptor)?;
432            eligible_object(conn, membership, &d).await?;
433            let (_, complete) = load(conn, &d.object, &hash(descriptor)).await?;
434            valid(complete && d.vault == context.vault && d.stream == context.stream)?;
435            let records: Vec<Vec<u8>> = sqlx::query_scalar(
436                "SELECT bytes FROM server_e2ee_image_chunks WHERE object=? ORDER BY chunk_index",
437            )
438            .bind(d.object.as_slice())
439            .fetch_all(&mut *conn)
440            .await?;
441            d.verify(&records)?;
442            let exists:bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_image_references WHERE workspace=? AND reference=?)").bind(workspace).bind(reference).fetch_one(&mut *conn).await?;
443            valid(!exists)?;
444            if !protected(conn, &d.object, workspace).await? {
445                ticket(
446                    conn,
447                    &d.object,
448                    authored_workspace,
449                    &context.device,
450                    t.context("error encrypted-image-reservation-required")?,
451                    now,
452                )
453                .await?;
454            }
455            let count: i64 =
456                sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_references")
457                    .fetch_one(&mut *conn)
458                    .await?;
459            valid(count < 262144)?;
460            let state: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?;
461            if state.is_none() {
462                let count: i64 =
463                    sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_parents")
464                        .fetch_one(&mut *conn)
465                        .await?;
466                valid(count < 262144)?;
467            }
468            let (known, state_deleted, sticky) = state.unwrap_or((None, false, false));
469            let mut state = crate::sync::persistence::parent_liveness::ParentState {
470                version: known.clone(),
471                deleted: state_deleted,
472                protected: sticky,
473            };
474            state.protect_hint(*deleted, version.as_deref());
475            sqlx::query("INSERT INTO server_e2ee_image_parents(workspace,parent,version,deleted,protected) VALUES(?,?,?,?,?) ON CONFLICT(workspace,parent) DO UPDATE SET protected=excluded.protected").bind(workspace).bind(task).bind(known).bind(state_deleted).bind(state.protected).execute(&mut *conn).await?;
476            sqlx::query("INSERT INTO server_e2ee_image_references(workspace,reference,parent,deleted,object) VALUES(?,?,?,0,?)").bind(workspace).bind(reference).bind(task).bind(d.object.as_slice()).execute(&mut *conn).await?;
477            sqlx::query("UPDATE server_e2ee_images SET origin=COALESCE(origin,?) WHERE object=?")
478                .bind(id)
479                .bind(d.object.as_slice())
480                .execute(&mut *conn)
481                .await?;
482            if let Some(t) = t {
483                sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE reservation=? AND device=? AND workspace=? AND object=?").bind(t.reservation.as_slice()).bind(context.device.as_slice()).bind(authored_workspace).bind(d.object.as_slice()).execute(&mut *conn).await?;
484            }
485            sqlx::query(
486                "INSERT OR IGNORE INTO server_e2ee_image_scopes(object, workspace) VALUES (?, ?)",
487            )
488            .bind(d.object.as_slice())
489            .bind(workspace)
490            .execute(&mut *conn)
491            .await?;
492            let scopes: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_scopes")
493                .fetch_one(&mut *conn)
494                .await?;
495            valid(scopes <= 262144)?;
496            retain_parent(conn, workspace, task).await?;
497        }
498        Projection::Unref {
499            workspace,
500            task,
501            reference,
502        } => {
503            let workspace = super::super::moves::server_workspace(conn, task, workspace).await?;
504            let changed=sqlx::query("UPDATE server_e2ee_image_references SET deleted=1 WHERE workspace=? AND reference=? AND parent=?").bind(&workspace).bind(reference).bind(task).execute(&mut *conn).await?.rows_affected();
505            valid(changed == 1)?;
506        }
507        _ => valid(t.is_none())?,
508    }
509    Ok(())
510}
511
512#[cfg(test)]
513mod tests {
514    use super::*;
515
516    #[tokio::test]
517    async fn retired_declaration_cannot_bypass_active_object_limit() {
518        let root = tempfile::tempdir().unwrap();
519        let db = Database::open(&root.path().join("test.sqlite"))
520            .await
521            .unwrap();
522        let a = crate::sync::encrypted_tail::tests::authority();
523        let mut conn = db.acquire_writer().await.unwrap();
524        let mut tx = begin_immediate(&mut conn).await.unwrap();
525        let (retired, _) = Descriptor::seal(&a, b"x").unwrap();
526        sqlx::query("INSERT INTO server_e2ee_images(object,byte_size,descriptor) VALUES(?,?,?)")
527            .bind(retired.object.as_slice())
528            .bind(retired.byte_size())
529            .bind(retired.encode().unwrap())
530            .execute(&mut *tx)
531            .await
532            .unwrap();
533        // These are unadmitted staging objects, not fabricated accepted references.
534        for _ in 0..128 {
535            let (d, records) = Descriptor::seal(&a, b"x").unwrap();
536            sqlx::query(
537                "INSERT INTO server_e2ee_images(object,byte_size,descriptor) VALUES(?,?,?)",
538            )
539            .bind(d.object.as_slice())
540            .bind(d.byte_size())
541            .bind(d.encode().unwrap())
542            .execute(&mut *tx)
543            .await
544            .unwrap();
545            sqlx::query(
546                "INSERT INTO server_e2ee_image_chunks(object,chunk_index,bytes) VALUES(?,0,?)",
547            )
548            .bind(d.object.as_slice())
549            .bind(&records[0])
550            .execute(&mut *tx)
551            .await
552            .unwrap();
553        }
554        assert!(
555            reserve(
556                &mut tx,
557                &retired,
558                "0000000000000000",
559                &a.context.device,
560                i64::MAX,
561                123
562            )
563            .await
564            .is_err()
565        );
566        let tickets: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_tickets")
567            .fetch_one(&mut *tx)
568            .await
569            .unwrap();
570        assert_eq!(tickets, 0);
571    }
572}
aven_core/sync/encrypted_tail/attachments/server.rs

aven_core/sync/encrypted_tail/attachments/
server.rs
