use super::super::{Context, hash, valid};
use super::{codec::Descriptor, *};
#[cfg(test)]
use crate::sync::encrypted_tail::attachments::codec::DescriptorAuthority;
use crate::sync::seed_claim::membership::Membership;
use crate::{
attachments::lifecycle::LifecyclePolicy,
db::{Database, begin_immediate},
sync::seed_claim::Secret,
};
use anyhow::{Context as _, Result, ensure};
use sqlx::SqliteConnection;
async fn load(
conn: &mut SqliteConnection,
object: &[u8; 32],
commitment: &[u8; 32],
) -> Result<(Descriptor, bool)> {
let (bytes, complete): (Vec<u8>, bool) =
sqlx::query_as("SELECT descriptor,complete FROM server_e2ee_images WHERE object=?")
.bind(object.as_slice())
.fetch_optional(conn)
.await?
.context("error encrypted-image-unknown")?;
valid(hash(&bytes) == *commitment)?;
let d = Descriptor::decode(&bytes)?;
valid(d.object == *object)?;
Ok((d, complete))
}
/// Stamps newly unreferenced images and clears the stamp from images that are
/// referenced or ticketed again. Only the background prune runs this full
/// scan, so an image's grace period starts at the first prune pass that
/// observes it unreferenced.
async fn refresh(conn: &mut SqliteConnection, now: i64) -> Result<()> {
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?;
Ok(())
}
/// Clears the unreferenced stamp of images a live parent references, so a
/// parent that is deleted again later gets a full grace period.
pub(in crate::sync::encrypted_tail) async fn retain_parent(
conn: &mut SqliteConnection,
workspace: &str,
parent: &str,
) -> Result<()> {
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?;
Ok(())
}
async fn protected(
conn: &mut SqliteConnection,
object: &[u8; 32],
workspace: &str,
) -> Result<bool> {
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?)
}
async fn scope(conn: &mut SqliteConnection, object: &[u8; 32], workspace: &str) -> Result<()> {
valid(
sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM server_e2ee_image_scopes WHERE object=? AND workspace=?)",
)
.bind(object.as_slice())
.bind(workspace)
.fetch_one(conn)
.await?,
)
}
async fn reserve(
conn: &mut SqliteConnection,
d: &Descriptor,
workspace: &str,
device: &[u8; 32],
quota: i64,
now: i64,
) -> Result<()> {
sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE expires_at<=?")
.bind(now)
.execute(&mut *conn)
.await?;
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?;
if existing {
return Ok(());
}
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=?")
.bind(now).bind(d.object.as_slice()).fetch_one(&mut *conn).await?;
if reactivating {
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>?))")
.bind(now).fetch_one(&mut *conn).await?;
valid(active < 128)?;
}
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?;
if !counted {
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?;
ensure!(
used.checked_add(d.byte_size()).is_some_and(|n| n <= quota),
"error attachment-quota-exceeded"
);
}
let count: i64 =
sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_tickets WHERE expires_at>?")
.bind(now)
.fetch_one(&mut *conn)
.await?;
valid(count < 128)?;
let mut id = [0; 32];
getrandom::fill(&mut id).map_err(|_| anyhow::anyhow!("error encrypted-image-entropy"))?;
let expires = now
.checked_add(crate::attachments::lifecycle::LEASE_TTL.as_secs() as i64)
.context("error encrypted-image-clock")?;
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")
.bind(id.as_slice()).bind(d.object.as_slice()).bind(workspace).bind(device.as_slice()).bind(expires).execute(&mut *conn).await?;
sqlx::query("UPDATE server_e2ee_images SET unreferenced_at=NULL WHERE object=?")
.bind(d.object.as_slice())
.execute(conn)
.await?;
Ok(())
}
async fn ticket(
conn: &mut SqliteConnection,
object: &[u8; 32],
workspace: &str,
device: &[u8; 32],
ticket: &Ticket,
now: i64,
) -> Result<()> {
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?)
}
async fn status(
conn: &mut SqliteConnection,
d: &Descriptor,
workspace: &str,
device: &[u8; 32],
now: i64,
) -> Result<Reply> {
let complete: bool =
sqlx::query_scalar("SELECT complete FROM server_e2ee_images WHERE object=?")
.bind(d.object.as_slice())
.fetch_one(&mut *conn)
.await?;
let indices: Vec<i64> = sqlx::query_scalar(
"SELECT chunk_index FROM server_e2ee_image_chunks WHERE object=? ORDER BY chunk_index",
)
.bind(d.object.as_slice())
.fetch_all(&mut *conn)
.await?;
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?;
let (reservation, expires_at) = match t {
Some((id, e)) => (
Some(
id.try_into()
.map_err(|_| anyhow::anyhow!("error encrypted-image-storage"))?,
),
Some(e),
),
None => (None, None),
};
Ok(Reply::Status(Status {
complete,
missing: (0..d.artifact.chunks.len())
.filter(|i| !indices.contains(&(*i as i64)))
.collect(),
reservation,
expires_at,
}))
}
async fn eligible_object(
conn: &mut SqliteConnection,
m: &Membership,
d: &Descriptor,
) -> Result<()> {
// Unknown generations fail before querying historical provenance.
valid(m.generations().iter().any(|g| g.id == d.generation))?;
let admitted = if d.generation != m.current_generation().id {
sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_images WHERE object=? AND origin IS NOT NULL AND descriptor=?)")
.bind(d.object.as_slice()).bind(d.encode()?).fetch_one(conn).await?
} else {
false
};
aven_protocol::tail::eligible_image(m, d.generation, admitted)?;
Ok(())
}
impl Database {
/// Deletes the stored bytes of at most `limit` images that have been
/// unreferenced and unticketed for at least `grace`, keeping their
/// descriptors. Returns how many images were pruned.
pub async fn prune_encrypted_images(
&self,
grace: std::time::Duration,
limit: usize,
) -> Result<usize> {
let mut conn = self.acquire_writer().await?;
let mut tx = begin_immediate(&mut conn).await?;
let now = chrono::Utc::now().timestamp();
refresh(&mut tx, now).await?;
let cutoff = now
.checked_sub(i64::try_from(grace.as_secs())?)
.context("error encrypted-image-clock")?;
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?;
// Objects with live tickets were excluded by `refresh`, so no
// uploader can still write to a pruned object.
for object in &objects {
sqlx::query("UPDATE server_e2ee_images SET complete=0 WHERE object=?")
.bind(object)
.execute(&mut *tx)
.await?;
sqlx::query("DELETE FROM server_e2ee_image_chunks WHERE object=?")
.bind(object)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(objects.len())
}
pub async fn encrypted_image_exchange(
&self,
context: &Context,
bearer: &Secret,
op: Operation,
policy: LifecyclePolicy,
) -> Result<Reply> {
let mut conn = self.acquire_writer().await?;
let mut tx = begin_immediate(&mut conn).await?;
let current =
crate::sync::seed_claim::membership::persistence::current(self, &mut tx).await?;
current
.membership
.authenticate(&context.authentication(bearer), false)?;
ensure!(
!current.membership.rotation_pending()
|| matches!(&op, Operation::Status { .. } | Operation::Read { .. }),
"error membership-rotation-pending"
);
let binding = current.membership.publication().binding();
valid(
context.stream == binding.stream_id
&& context.descriptor == binding.descriptor_commitment,
)?;
let now = chrono::Utc::now().timestamp();
let reply = match op {
Operation::Declare {
workspace,
descriptor,
} => {
workspace.parse::<crate::ids::WorkspaceId>()?;
let d = Descriptor::decode(&descriptor)?;
valid(d.vault == context.vault && d.stream == context.stream)?;
eligible_object(&mut tx, ¤t.membership, &d).await?;
let old: Option<Vec<u8>> =
sqlx::query_scalar("SELECT descriptor FROM server_e2ee_images WHERE object=?")
.bind(d.object.as_slice())
.fetch_optional(&mut *tx)
.await?;
if let Some(old) = old {
valid(old == descriptor)?;
} else {
let total: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_images")
.fetch_one(&mut *tx)
.await?;
valid(total < 65536)?;
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?;
}
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?;
if !known {
let count: i64 =
sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_scopes")
.fetch_one(&mut *tx)
.await?;
valid(count < 262144)?;
}
sqlx::query(
"INSERT OR IGNORE INTO server_e2ee_image_scopes(object,workspace) VALUES(?,?)",
)
.bind(d.object.as_slice())
.bind(&workspace)
.execute(&mut *tx)
.await?;
load(&mut tx, &d.object, &hash(&descriptor)).await?;
reserve(
&mut tx,
&d,
&workspace,
&context.device,
policy.quota_bytes,
now,
)
.await?;
status(&mut tx, &d, &workspace, &context.device, now).await?
}
other => dispatch_existing(&mut tx, context, ¤t.membership, other, now).await?,
};
tx.commit().await?;
Ok(reply)
}
}
async fn dispatch_existing(
conn: &mut SqliteConnection,
context: &Context,
membership: &Membership,
op: Operation,
now: i64,
) -> Result<Reply> {
let (workspace, object, commitment) = match &op {
Operation::Status {
workspace,
object,
descriptor_commitment,
}
| Operation::Put {
workspace,
object,
descriptor_commitment,
..
}
| Operation::Complete {
workspace,
object,
descriptor_commitment,
..
}
| Operation::Read {
workspace,
object,
descriptor_commitment,
..
}
| Operation::Release {
workspace,
object,
descriptor_commitment,
..
} => (workspace, *object, *descriptor_commitment),
_ => unreachable!(),
};
workspace.parse::<crate::ids::WorkspaceId>()?;
scope(conn, &object, workspace).await?;
let (d, _) = load(conn, &object, &commitment).await?;
valid(d.vault == context.vault && d.stream == context.stream)?;
eligible_object(conn, membership, &d).await?;
match &op {
Operation::Status { .. } => status(conn, &d, workspace, &context.device, now).await,
Operation::Read { index, .. } => {
valid(*index < d.artifact.chunks.len())?;
let admitted: bool = sqlx::query_scalar(
"SELECT origin IS NOT NULL FROM server_e2ee_images WHERE object=?",
)
.bind(object.as_slice())
.fetch_one(&mut *conn)
.await?;
valid(admitted)?;
let bytes: 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(conn)
.await?;
Ok(match bytes {
Some(b) => {
d.verify_chunk(*index, &b)?;
Reply::Chunk(b)
}
None => Reply::Unavailable,
})
}
Operation::Put { reservation, .. }
| Operation::Complete { reservation, .. }
| Operation::Release { reservation, .. } => {
ticket(
conn,
&object,
workspace,
&context.device,
&Ticket {
reservation: *reservation,
},
now,
)
.await?;
match &op {
Operation::Put { index, record, .. } => {
d.verify_chunk(*index, record)?;
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?;
valid(old.as_ref().is_none_or(|b| b == record))?;
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?;
}
Operation::Complete { .. } => {
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?;
d.verify(&records)?;
sqlx::query("UPDATE server_e2ee_images SET complete=1 WHERE object=?")
.bind(object.as_slice())
.execute(&mut *conn)
.await?;
}
Operation::Release { .. } => {
sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE reservation=?")
.bind(reservation.as_slice())
.execute(&mut *conn)
.await?;
}
_ => unreachable!(),
}
Ok(Reply::Done)
}
_ => unreachable!(),
}
}
pub(in crate::sync::encrypted_tail) async fn admit(
conn: &mut SqliteConnection,
context: &Context,
membership: &Membership,
id: &str,
p: &super::super::domain::Projection,
t: Option<&Ticket>,
) -> Result<()> {
use super::super::domain::Projection;
let now = chrono::Utc::now().timestamp();
match p {
Projection::Move {
source,
target,
tasks,
} => {
valid(t.is_none())?;
super::super::moves::admit(conn, source, target, tasks).await?;
}
Projection::Ref {
workspace,
task,
reference,
descriptor,
deleted,
version,
} => {
let authored_workspace = workspace;
let routed = super::super::moves::server_workspace(conn, task, workspace).await?;
let workspace = &routed;
let d = Descriptor::decode(descriptor)?;
eligible_object(conn, membership, &d).await?;
let (_, complete) = load(conn, &d.object, &hash(descriptor)).await?;
valid(complete && d.vault == context.vault && d.stream == context.stream)?;
let records: Vec<Vec<u8>> = sqlx::query_scalar(
"SELECT bytes FROM server_e2ee_image_chunks WHERE object=? ORDER BY chunk_index",
)
.bind(d.object.as_slice())
.fetch_all(&mut *conn)
.await?;
d.verify(&records)?;
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?;
valid(!exists)?;
if !protected(conn, &d.object, workspace).await? {
ticket(
conn,
&d.object,
authored_workspace,
&context.device,
t.context("error encrypted-image-reservation-required")?,
now,
)
.await?;
}
let count: i64 =
sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_references")
.fetch_one(&mut *conn)
.await?;
valid(count < 262144)?;
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?;
if state.is_none() {
let count: i64 =
sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_parents")
.fetch_one(&mut *conn)
.await?;
valid(count < 262144)?;
}
let (known, state_deleted, sticky) = state.unwrap_or((None, false, false));
let mut state = crate::sync::persistence::parent_liveness::ParentState {
version: known.clone(),
deleted: state_deleted,
protected: sticky,
};
state.protect_hint(*deleted, version.as_deref());
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?;
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?;
sqlx::query("UPDATE server_e2ee_images SET origin=COALESCE(origin,?) WHERE object=?")
.bind(id)
.bind(d.object.as_slice())
.execute(&mut *conn)
.await?;
if let Some(t) = t {
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?;
}
sqlx::query(
"INSERT OR IGNORE INTO server_e2ee_image_scopes(object, workspace) VALUES (?, ?)",
)
.bind(d.object.as_slice())
.bind(workspace)
.execute(&mut *conn)
.await?;
let scopes: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_scopes")
.fetch_one(&mut *conn)
.await?;
valid(scopes <= 262144)?;
retain_parent(conn, workspace, task).await?;
}
Projection::Unref {
workspace,
task,
reference,
} => {
let workspace = super::super::moves::server_workspace(conn, task, workspace).await?;
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();
valid(changed == 1)?;
}
_ => valid(t.is_none())?,
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn invalid_workspaces_refuse_authenticated_declare_and_existing_dispatch() {
use crate::ids::InvalidWorkspaceId;
use crate::sync::seed_claim::membership::test_support::Fixture;
use aven_protocol::artifact::catalog::{Artifact, Chunk};
let f = Fixture::new().await;
let membership = Membership::from_publication(
f.seed.genesis(),
&f.package.descriptor,
f.publication.record(),
)
.unwrap();
let binding = membership.publication().binding();
let context = Context {
vault: binding.vault_id,
genesis: binding.genesis_commitment,
device: f.seed.genesis().device_id(),
head: membership.head(),
stream: binding.stream_id,
descriptor: binding.descriptor_commitment,
};
// A structurally valid, not-yet-uploaded object recipe.
let descriptor = Descriptor {
vault: context.vault,
stream: context.stream,
generation: membership.current_generation().id,
object: [44; 32],
artifact: Artifact {
total: 1,
aggregate: [45; 32],
chunks: vec![Chunk {
length: 223,
hash: [46; 32],
nonce: [47; 24],
}],
},
}
.encode()
.unwrap();
for op in [
Operation::Declare {
workspace: "invalid".into(),
descriptor: descriptor.clone(),
},
Operation::Status {
workspace: "0123456789ABCDEI".into(),
object: [44; 32],
descriptor_commitment: hash(&descriptor),
},
] {
let error = f
.db
.encrypted_image_exchange(&context, f.seed.bearer(), op, LifecyclePolicy::default())
.await
.err()
.unwrap();
assert!(error.is::<InvalidWorkspaceId>());
assert_eq!(
error.to_string(),
"workspace ID must be 16 Crockford Base32 characters"
);
}
let mut conn = f.db.acquire_reader().await.unwrap();
let counts: (i64, i64, i64) = sqlx::query_as(
"SELECT (SELECT count(*) FROM server_e2ee_images), (SELECT count(*) FROM server_e2ee_image_scopes), (SELECT count(*) FROM server_e2ee_image_tickets)",
).fetch_one(&mut *conn).await.unwrap();
assert_eq!(counts, (0, 0, 0));
}
#[tokio::test]
async fn retired_declaration_cannot_bypass_active_object_limit() {
let root = tempfile::tempdir().unwrap();
let db = Database::open(&root.path().join("test.sqlite"))
.await
.unwrap();
let a = crate::sync::encrypted_tail::tests::authority();
let mut conn = db.acquire_writer().await.unwrap();
let mut tx = begin_immediate(&mut conn).await.unwrap();
let (retired, _) = Descriptor::seal(&a, b"x").unwrap();
sqlx::query("INSERT INTO server_e2ee_images(object,byte_size,descriptor) VALUES(?,?,?)")
.bind(retired.object.as_slice())
.bind(retired.byte_size())
.bind(retired.encode().unwrap())
.execute(&mut *tx)
.await
.unwrap();
// These are unadmitted staging objects, not fabricated accepted references.
for _ in 0..128 {
let (d, records) = Descriptor::seal(&a, b"x").unwrap();
sqlx::query(
"INSERT INTO server_e2ee_images(object,byte_size,descriptor) VALUES(?,?,?)",
)
.bind(d.object.as_slice())
.bind(d.byte_size())
.bind(d.encode().unwrap())
.execute(&mut *tx)
.await
.unwrap();
sqlx::query(
"INSERT INTO server_e2ee_image_chunks(object,chunk_index,bytes) VALUES(?,0,?)",
)
.bind(d.object.as_slice())
.bind(&records[0])
.execute(&mut *tx)
.await
.unwrap();
}
assert!(
reserve(
&mut tx,
&retired,
"0000000000000000",
&a.context.device,
i64::MAX,
123
)
.await
.is_err()
);
let tickets: i64 = sqlx::query_scalar("SELECT count(*) FROM server_e2ee_image_tickets")
.fetch_one(&mut *tx)
.await
.unwrap();
assert_eq!(tickets, 0);
}
}