aven_core/sync/seed_claim/membership/
persistence.rs

1//! Bounded signed history, invitation outcomes and vault-clock transactions.
2use super::super::peer::{Authentication, RegistrationStatus, request_parts};
3use super::*;
4use crate::db::{Database, begin_immediate};
5use sqlx::SqliteConnection;
6use std::sync::Arc;
7
8pub const MAX_INVITATIONS: usize = 128;
9pub const MAX_CANDIDATES: usize = 8;
10pub(crate) struct Current {
11    pub membership: Membership,
12    pub evidence: Evidence,
13}
14/// The verified replay of one stored head. Heads are hash-linked over every
15/// record, so an unchanged head means unchanged history; storage integrity
16/// behind a cached head is established by the full replay at server start.
17#[derive(Clone, Default)]
18pub(crate) struct Cache(Arc<std::sync::Mutex<Option<Arc<Current>>>>);
19impl Cache {
20    fn get(&self, sequence: i64, head: &[u8]) -> Option<Arc<Current>> {
21        let cached = self.0.lock().unwrap_or_else(|e| e.into_inner());
22        cached
23            .as_ref()
24            .filter(|c| c.membership.sequence() as i64 == sequence && c.membership.head() == head)
25            .cloned()
26    }
27    fn set(&self, current: Arc<Current>) {
28        *self.0.lock().unwrap_or_else(|e| e.into_inner()) = Some(current);
29    }
30}
31/// Replays only when the stored head differs from the cached one; invitation
32/// bounds and projections are checked on every call.
33pub(crate) async fn current(db: &Database, conn: &mut SqliteConnection) -> Result<Arc<Current>> {
34    let (invitations, malformed): (i64, i64) = sqlx::query_as("SELECT count(*),coalesce(sum(CASE WHEN length(declaration)!=280 OR (request IS NOT NULL AND length(request)!=314) OR length(handle)!=32 OR length(inviter)!=32 THEN 1 ELSE 0 END),0) FROM server_membership_invitations")
35        .fetch_one(&mut *conn).await?;
36    ensure!(
37        invitations <= MAX_INVITATIONS as i64 && malformed == 0,
38        "error membership-limit"
39    );
40    let clock: bool = sqlx::query_scalar(
41        "SELECT EXISTS(SELECT 1 FROM server_membership_clock WHERE high_water>=0)",
42    )
43    .fetch_one(&mut *conn)
44    .await?;
45    ensure!(invitations == 0 || clock, "error enrollment-clock-missing");
46    let head: Option<(i64, Vec<u8>)> = sqlx::query_as(
47        "SELECT sequence,commitment FROM server_e2ee_membership_head WHERE singleton=1",
48    )
49    .fetch_optional(&mut *conn)
50    .await?;
51    let cached = head.and_then(|(sequence, head)| db.membership_cache.get(sequence, &head));
52    let current = match cached {
53        Some(current) => current,
54        None => {
55            let current = Arc::new(replay(conn).await?);
56            db.membership_cache.set(current.clone());
57            current
58        }
59    };
60    let orphans: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_membership_invitations i WHERE admitted_sequence IS NOT NULL AND NOT EXISTS(SELECT 1 FROM server_membership_transitions a WHERE a.handle=i.handle AND a.sequence=i.admitted_sequence))")
61        .fetch_one(&mut *conn).await?;
62    ensure!(!orphans, "error membership-history-missing");
63    let invitations: Vec<(Vec<u8>, bool, bool, bool)> = sqlx::query_as("SELECT inviter,admitted_sequence IS NOT NULL,revoked,expired FROM server_membership_invitations")
64        .fetch_all(&mut *conn).await?;
65    for (inviter, admitted, revoked, expired) in invitations {
66        let inviter: Hash = inviter
67            .try_into()
68            .map_err(|_| anyhow::anyhow!("error membership-storage"))?;
69        ensure!(
70            revoked == (!admitted && current.membership.member(&inviter).is_err())
71                && (!revoked || expired),
72            "error membership-invitation-projection"
73        );
74    }
75    Ok(current)
76}
77/// Bound stored byte lengths before materializing any attacker-influenced blobs.
78async fn replay(conn: &mut SqliteConnection) -> Result<Current> {
79    let sizes: Option<(i64, i64, i64, bool)> = sqlx::query_as("SELECT length(s.genesis),length(p.signed_record),length(p.descriptor),s.genesis_only FROM server_seed_claim s JOIN server_bootstrap_publication p ON p.singleton=s.singleton WHERE s.singleton=1")
80        .fetch_optional(&mut *conn).await?;
81    let (g, p, d, only) = sizes.context("error enrollment-requires-ready")?;
82    ensure!(
83        !only
84            && g == GENESIS_BYTES as i64
85            && p == PUBLICATION_BYTES as i64
86            && (0..=crate::sync::bootstrap_format::MAX_DESCRIPTOR_BYTES as i64).contains(&d),
87        "error membership-storage"
88    );
89    let (count, total, largest): (i64, i64, i64) = sqlx::query_as("SELECT count(*),coalesce(sum(length(record) + CASE WHEN handle IS NULL THEN 0 ELSE ? END),0),coalesce(max(length(record)),0) FROM server_membership_transitions")
90        .bind((DECLARATION_BYTES + REQUEST_BYTES) as i64).fetch_one(&mut *conn).await?;
91    ensure!(
92        (0..=MAX_TRANSITIONS as i64).contains(&count)
93            && largest <= MAX_RECORD_BYTES as i64
94            && total >= 0
95            && g + p + d + total <= MAX_CHAIN_BYTES as i64,
96        "error membership-limit"
97    );
98    let (genesis, publication, descriptor): (Vec<u8>, Vec<u8>, Vec<u8>) = sqlx::query_as("SELECT s.genesis,p.signed_record,p.descriptor FROM server_seed_claim s JOIN server_bootstrap_publication p ON p.singleton=s.singleton WHERE s.singleton=1")
99        .fetch_one(&mut *conn).await?;
100    type Row = (i64, Option<Vec<u8>>, Vec<u8>);
101    let rows: Vec<Row> = sqlx::query_as(
102        "SELECT sequence,handle,record FROM server_membership_transitions ORDER BY sequence",
103    )
104    .fetch_all(&mut *conn)
105    .await?;
106    let mut evidence = Evidence {
107        genesis,
108        publication,
109        descriptor,
110        transitions: Vec::with_capacity(rows.len()),
111    };
112    let mut membership = evidence.verify()?;
113    for (seq, handle, record) in rows {
114        ensure!(
115            seq == membership.sequence() as i64 + 1,
116            "error membership-history-missing"
117        );
118        let (declaration, request) = if let Some(handle) = handle {
119            let (declaration, request, inviter, expiry, admitted): (Vec<u8>, Vec<u8>, Vec<u8>, i64, i64) =
120                sqlx::query_as("SELECT declaration,request,inviter,expiry,admitted_sequence FROM server_membership_invitations WHERE handle=?")
121                    .bind(&handle).fetch_one(&mut *conn).await?;
122            let dec = Declaration::from_record(&membership, &declaration)?;
123            ensure!(
124                handle == dec.handle
125                    && inviter == dec.inviter
126                    && expiry >= 0
127                    && expiry as u64 == dec.expiry()
128                    && admitted == seq,
129                "error membership-storage"
130            );
131            (declaration, request)
132        } else {
133            (vec![], vec![])
134        };
135        membership = membership.append(&declaration, &request, &record)?;
136        evidence.transitions.push(EvidenceRecord {
137            declaration,
138            request,
139            record,
140        });
141    }
142    let (seq, head): (i64, Vec<u8>) = sqlx::query_as(
143        "SELECT sequence,commitment FROM server_e2ee_membership_head WHERE singleton=1",
144    )
145    .fetch_one(&mut *conn)
146    .await?;
147    ensure!(
148        seq == membership.sequence() as i64 && head == membership.head(),
149        "error membership-head-invalid"
150    );
151    Ok(Current {
152        membership,
153        evidence,
154    })
155}
156/// Unix seconds that enrollment expiry compares against.
157pub fn now() -> Result<i64> {
158    std::time::SystemTime::now()
159        .duration_since(std::time::UNIX_EPOCH)
160        .ok()
161        .and_then(|d| i64::try_from(d.as_secs()).ok())
162        .context("error enrollment-clock")
163}
164async fn observe(conn: &mut SqliteConnection, time: i64) -> Result<i64> {
165    ensure!(time >= 0, "error enrollment-clock");
166    sqlx::query("INSERT INTO server_membership_clock(singleton,high_water) VALUES(1,?) ON CONFLICT(singleton) DO UPDATE SET high_water=max(high_water,excluded.high_water)")
167        .bind(time).execute(&mut *conn).await?;
168    let high: i64 =
169        sqlx::query_scalar("SELECT high_water FROM server_membership_clock WHERE singleton=1")
170            .fetch_one(&mut *conn)
171            .await?;
172    sqlx::query("UPDATE server_membership_invitations SET expired=1 WHERE admitted_sequence IS NULL AND expiry<=?")
173        .bind(high).execute(&mut *conn).await?;
174    Ok(high)
175}
176/// Serialized outcome of cancelling a registered invitation. `Admitted` is a
177/// hint only; callers resolve admission from the verified membership chain.
178#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
179pub enum CancelStatus {
180    Admitted,
181    Cancelled,
182}
183#[derive(serde::Serialize, serde::Deserialize)]
184#[serde(deny_unknown_fields)]
185pub struct ManagementPreparation {
186    pub evidence: Evidence,
187    pub high_water: u64,
188}
189async fn append_transition(
190    conn: &mut SqliteConnection,
191    before: &Membership,
192    next: &Membership,
193    handle: Option<Hash>,
194    record: &[u8],
195) -> Result<()> {
196    sqlx::query("INSERT INTO server_membership_transitions(sequence,handle,record) VALUES(?,?,?)")
197        .bind(next.sequence() as i64)
198        .bind(handle.map(|h| h.to_vec()))
199        .bind(record)
200        .execute(&mut *conn)
201        .await?;
202    let changed = sqlx::query("UPDATE server_e2ee_membership_head SET sequence=?,commitment=? WHERE singleton=1 AND sequence=? AND commitment=?")
203        .bind(next.sequence() as i64).bind(next.head().as_slice()).bind(before.sequence() as i64).bind(before.head().as_slice()).execute(conn).await?;
204    ensure!(
205        changed.rows_affected() == 1,
206        "error membership-head-invalid"
207    );
208    Ok(())
209}
210pub(crate) async fn allocator(conn: &mut SqliteConnection, m: &Membership) -> Result<u64> {
211    let b = m.publication().binding();
212    let (n, high): (i64, i64) = sqlx::query_as(
213        "SELECT prefix_count,high_water FROM server_e2ee_allocator WHERE singleton=1 AND stream=?",
214    )
215    .bind(b.stream_id.as_slice())
216    .fetch_one(conn)
217    .await?;
218    ensure!(
219        n >= 0
220            && n as u64 == b.prefix_count
221            && high >= n
222            && high as u64 >= m.current_generation().starts_after,
223        "error membership-allocator"
224    );
225    Ok(high as u64)
226}
227impl Database {
228    pub async fn prepare_membership_management(
229        &self,
230        auth: &Authentication<'_>,
231    ) -> Result<ManagementPreparation> {
232        let mut conn = self.acquire_writer().await?;
233        let mut tx = begin_immediate(&mut conn).await?;
234        let c = current(self, &mut tx).await?;
235        c.membership.authenticate(auth, false)?;
236        let high_water = allocator(&mut tx, &c.membership).await?;
237        tx.commit().await?;
238        Ok(ManagementPreparation {
239            evidence: c.evidence.clone(),
240            high_water,
241        })
242    }
243    pub async fn apply_membership_management(
244        &self,
245        auth: &Authentication<'_>,
246        record: &[u8],
247    ) -> Result<Vec<u8>> {
248        ensure!(record.len() <= MAX_RECORD_BYTES, "error membership-limit");
249        let mut conn = self.acquire_writer().await?;
250        let mut tx = begin_immediate(&mut conn).await?;
251        let c = current(self, &mut tx).await?;
252        c.membership.authenticate(auth, false)?;
253        let (core, _, _, _) = encoding::components(record)?;
254        let (signer, action) = encoding::signer_action(core)?;
255        ensure!(matches!(action, 4 | 5), "error membership-action");
256        if !c.evidence.transitions.iter().any(|t| t.record == record) {
257            ensure!(signer == auth.device, super::Unauthorized);
258            let next = c.membership.append(&[], &[], record)?;
259            let high = allocator(&mut tx, &c.membership).await?;
260            ensure!(
261                action != 5 || next.current_generation().starts_after == high,
262                "error membership-cutoff"
263            );
264            append_transition(&mut tx, &c.membership, &next, None, record).await?;
265            for member in &c.membership.members {
266                if next.member(&member.device).is_err() {
267                    sqlx::query("UPDATE server_membership_invitations SET revoked=1,expired=1 WHERE inviter=? AND admitted_sequence IS NULL")
268                        .bind(member.device.as_slice()).execute(&mut *tx).await?;
269                    sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE device=?")
270                        .bind(member.device.as_slice())
271                        .execute(&mut *tx)
272                        .await?;
273                }
274            }
275        }
276        tx.commit().await?;
277        Ok(record.to_vec())
278    }
279}
280impl Database {
281    /// Replays the complete stored membership history and fills the cache.
282    /// Servers run this before accepting requests; storage without published
283    /// membership has nothing to verify.
284    pub async fn verify_membership_history(&self) -> Result<()> {
285        let mut conn = self.acquire_writer().await?;
286        let mut tx = begin_immediate(&mut conn).await?;
287        let published: bool =
288            sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_membership_head)")
289                .fetch_one(&mut *tx)
290                .await?;
291        if published {
292            self.membership_cache.set(Arc::new(replay(&mut tx).await?));
293            current(self, &mut tx).await?;
294        }
295        tx.commit().await?;
296        Ok(())
297    }
298    /// This read alone accepts a verified ancestor head, after current credential authentication.
299    pub async fn membership_evidence(&self, auth: &Authentication<'_>) -> Result<Evidence> {
300        let mut conn = self.acquire_writer().await?;
301        let mut tx = begin_immediate(&mut conn).await?;
302        let c = current(self, &mut tx).await?;
303        c.membership.authenticate(auth, true)?;
304        tx.commit().await?;
305        Ok(c.evidence.clone())
306    }
307    pub async fn register_membership_invitation_at(
308        &self,
309        auth: &Authentication<'_>,
310        raw: &[u8],
311        time: i64,
312    ) -> Result<RegistrationStatus> {
313        ensure!(raw.len() == DECLARATION_BYTES, "error membership-limit");
314        let mut conn = self.acquire_writer().await?;
315        let mut tx = begin_immediate(&mut conn).await?;
316        let c = current(self, &mut tx).await?;
317        c.membership.authenticate(auth, false)?;
318        let d = Declaration::from_record(&c.membership, raw)?;
319        ensure!(d.inviter == auth.device, super::Unauthorized);
320        let expiry = i64::try_from(d.expiry()).context("error enrollment-expired")?;
321        let high = observe(&mut tx, time).await?;
322        let stored: Option<Vec<u8>> = sqlx::query_scalar(
323            "SELECT declaration FROM server_membership_invitations WHERE handle=?",
324        )
325        .bind(d.handle.as_slice())
326        .fetch_optional(&mut *tx)
327        .await?;
328        if let Some(stored) = stored {
329            ensure!(stored == raw, "error enrollment-conflict");
330        } else {
331            let count: i64 =
332                sqlx::query_scalar("SELECT count(*) FROM server_membership_invitations")
333                    .fetch_one(&mut *tx)
334                    .await?;
335            ensure!(
336                count < MAX_INVITATIONS as i64 && c.membership.device_count() < MAX_DEVICES,
337                "error membership-limit"
338            );
339            ensure!(
340                expiry <= high || expiry - high <= 3600,
341                "error enrollment-expired"
342            );
343            let unfinished: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_membership_invitations WHERE inviter=? AND expired=0 AND admitted_sequence IS NULL)")
344                .bind(d.inviter.as_slice()).fetch_one(&mut *tx).await?;
345            if unfinished {
346                tx.commit().await?;
347                anyhow::bail!("error enrollment-unfinished");
348            }
349            sqlx::query("INSERT INTO server_membership_invitations(handle,inviter,declaration,expiry,expired) VALUES(?,?,?,?,?)")
350                .bind(d.handle.as_slice()).bind(d.inviter.as_slice()).bind(raw).bind(expiry).bind(expiry <= high).execute(&mut *tx).await?;
351        }
352        let (expired, consumed): (bool, bool) = sqlx::query_as("SELECT expired,admitted_sequence IS NOT NULL FROM server_membership_invitations WHERE handle=?")
353            .bind(d.handle.as_slice()).fetch_one(&mut *tx).await?;
354        tx.commit().await?;
355        Ok(if consumed {
356            RegistrationStatus::Consumed
357        } else if expired {
358            RegistrationStatus::Expired
359        } else {
360            RegistrationStatus::Open
361        })
362    }
363    pub async fn post_membership_request_at(
364        &self,
365        vault: Hash,
366        handle: Hash,
367        request: &[u8],
368        time: i64,
369    ) -> Result<()> {
370        request_parts(request, &handle)?;
371        let mut conn = self.acquire_writer().await?;
372        let mut tx = begin_immediate(&mut conn).await?;
373        let c = current(self, &mut tx).await?;
374        check(vault == c.membership.genesis.context.vault_id)?;
375        observe(&mut tx, time).await?;
376        let (stored, expired, revoked, admitted): (Option<Vec<u8>>, bool, bool, bool) = sqlx::query_as(
377            "SELECT request,expired,revoked,admitted_sequence IS NOT NULL FROM server_membership_invitations WHERE handle=?",
378        )
379        .bind(handle.as_slice())
380        .fetch_optional(&mut *tx)
381        .await?
382        .context("error enrollment-unavailable")?;
383        ensure!(
384            !revoked && (!admitted || c.membership.members.iter().any(|m| m.admission == handle)),
385            "error enrollment-revoked"
386        );
387        if let Some(stored) = stored {
388            ensure!(stored == request, "error enrollment-request-occupied");
389        } else {
390            if expired {
391                tx.commit().await?;
392                anyhow::bail!("error enrollment-expired");
393            }
394            sqlx::query("UPDATE server_membership_invitations SET request=? WHERE handle=?")
395                .bind(request)
396                .bind(handle.as_slice())
397                .execute(&mut *tx)
398                .await?;
399        }
400        tx.commit().await?;
401        Ok(())
402    }
403    pub async fn membership_mailbox(&self, vault: Hash, handle: Hash) -> Result<Mailbox> {
404        let mut conn = self.acquire_writer().await?;
405        let mut tx = begin_immediate(&mut conn).await?;
406        let c = current(self, &mut tx).await?;
407        check(vault == c.membership.genesis.context.vault_id)?;
408        let (declaration, request, admission): (Vec<u8>, Option<Vec<u8>>, Option<Vec<u8>>) = sqlx::query_as("SELECT i.declaration,i.request,a.record FROM server_membership_invitations i LEFT JOIN server_membership_transitions a ON a.handle=i.handle WHERE i.handle=?")
409            .bind(handle.as_slice()).fetch_optional(&mut *tx).await?.context("error enrollment-unavailable")?;
410        if admission.is_some() {
411            ensure!(
412                c.membership.members.iter().any(|m| m.admission == handle),
413                "error enrollment-revoked"
414            );
415        } else {
416            let d = Declaration::from_record(&c.membership, &declaration)?;
417            check(d.handle == handle)?;
418        }
419        tx.commit().await?;
420        Ok(Mailbox {
421            declaration,
422            request,
423            admission,
424        })
425    }
426    /// Terminally fences a registered, unadmitted invitation of the caller so
427    /// no new admission can commit. Serialized with admission; idempotent.
428    /// Unknown handles allocate nothing.
429    pub async fn cancel_membership_invitation_at(
430        &self,
431        auth: &Authentication<'_>,
432        handle: Hash,
433        time: i64,
434    ) -> Result<CancelStatus> {
435        let mut conn = self.acquire_writer().await?;
436        let mut tx = begin_immediate(&mut conn).await?;
437        let c = current(self, &mut tx).await?;
438        c.membership.authenticate(auth, false)?;
439        let (declaration, inviter, admitted): (Vec<u8>, Vec<u8>, bool) = sqlx::query_as("SELECT declaration,inviter,admitted_sequence IS NOT NULL FROM server_membership_invitations WHERE handle=?")
440            .bind(handle.as_slice())
441            .fetch_optional(&mut *tx)
442            .await?
443            .context("error enrollment-unavailable")?;
444        // The immutable stored row binds the exact declaration and its inviter.
445        ensure!(
446            inviter == auth.device
447                && Declaration::from_record(&c.membership, &declaration)?.handle == handle,
448            super::Unauthorized
449        );
450        let status = if admitted {
451            CancelStatus::Admitted
452        } else {
453            observe(&mut tx, time).await?;
454            sqlx::query("UPDATE server_membership_invitations SET expired=1 WHERE handle=?")
455                .bind(handle.as_slice())
456                .execute(&mut *tx)
457                .await?;
458            CancelStatus::Cancelled
459        };
460        tx.commit().await?;
461        Ok(status)
462    }
463    pub async fn admit_membership_device_at(
464        &self,
465        auth: &Authentication<'_>,
466        handle: Hash,
467        record: &[u8],
468        time: i64,
469    ) -> Result<Vec<u8>> {
470        ensure!(record.len() <= MAX_RECORD_BYTES, "error membership-limit");
471        let mut conn = self.acquire_writer().await?;
472        let mut tx = begin_immediate(&mut conn).await?;
473        let c = current(self, &mut tx).await?;
474        c.membership.authenticate(auth, false)?;
475        type Row = (Vec<u8>, Option<Vec<u8>>, Option<Vec<u8>>, bool);
476        let (declaration, request, saved, fenced): Row = sqlx::query_as("SELECT i.declaration,i.request,a.record,i.expired FROM server_membership_invitations i LEFT JOIN server_membership_transitions a ON a.handle=i.handle WHERE i.handle=?")
477            .bind(handle.as_slice()).fetch_optional(&mut *tx).await?.context("error enrollment-unavailable")?;
478        if let Some(saved) = saved {
479            ensure!(saved == record, "error enrollment-conflict");
480            observe(&mut tx, time).await?;
481        } else {
482            let d = Declaration::from_record(&c.membership, &declaration)?;
483            ensure!(
484                d.handle == handle && d.inviter == auth.device,
485                super::Unauthorized
486            );
487            let high = observe(&mut tx, time).await?;
488            // Cancelled and clock-expired rows stay terminal for new admission.
489            if fenced || high as u64 >= d.expiry() {
490                tx.commit().await?;
491                anyhow::bail!("error enrollment-expired");
492            }
493            let request = request.context("error enrollment-request-missing")?;
494            let next = c.membership.append(&declaration, &request, record)?;
495            append_transition(&mut tx, &c.membership, &next, Some(handle), record).await?;
496            sqlx::query(
497                "UPDATE server_membership_invitations SET admitted_sequence=? WHERE handle=?",
498            )
499            .bind(next.sequence() as i64)
500            .bind(handle.as_slice())
501            .execute(&mut *tx)
502            .await?;
503        }
504        tx.commit().await?;
505        Ok(record.to_vec())
506    }
507}
508#[cfg(test)]
509mod tests;
510
511impl Database {
512    pub async fn membership_checkpoint_mirror(&self) -> Result<Option<(Hash, u64, Hash, Hash)>> {
513        let mut conn = self.acquire_reader().await?;
514        type Row = (Vec<u8>, i64, Vec<u8>, Vec<u8>);
515        let row: Option<Row> = sqlx::query_as("SELECT identity,sequence,head,evidence FROM local_membership_checkpoint WHERE singleton=1").fetch_optional(&mut *conn).await?;
516        row.map(|(id, seq, head, evidence)| {
517            Ok((
518                id.try_into()
519                    .map_err(|_| anyhow::anyhow!("error membership-mirror"))?,
520                u64::try_from(seq)?,
521                head.try_into()
522                    .map_err(|_| anyhow::anyhow!("error membership-mirror"))?,
523                evidence
524                    .try_into()
525                    .map_err(|_| anyhow::anyhow!("error membership-mirror"))?,
526            ))
527        })
528        .transpose()
529    }
530    pub async fn mirror_membership_checkpoint(
531        &self,
532        identity: Hash,
533        m: &Membership,
534        evidence: Hash,
535    ) -> Result<()> {
536        let mut conn = self.acquire_writer().await?;
537        let mut tx = begin_immediate(&mut conn).await?;
538        let valid: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_peer_enrollment WHERE identity=? AND client_id=(SELECT value FROM meta WHERE key='client_id'))").bind(identity.as_slice()).fetch_one(&mut *tx).await?;
539        ensure!(valid, "error membership-identity");
540        type MirrorRow = (Vec<u8>, i64, Vec<u8>, Vec<u8>);
541        let old: Option<MirrorRow> = sqlx::query_as("SELECT identity,sequence,head,evidence FROM local_membership_checkpoint WHERE singleton=1").fetch_optional(&mut *tx).await?;
542        if let Some((id, seq, head, digest)) = old {
543            ensure!(
544                id == identity
545                    && m.head_at(u64::try_from(seq)?).is_some_and(|h| head == h)
546                    && (seq != m.sequence() as i64 || digest == evidence),
547                "error membership-mirror"
548            );
549        }
550        sqlx::query("INSERT INTO local_membership_checkpoint(singleton,identity,sequence,head,evidence) VALUES(1,?,?,?,?) ON CONFLICT(singleton) DO UPDATE SET sequence=excluded.sequence,head=excluded.head,evidence=excluded.evidence")
551            .bind(identity.as_slice()).bind(m.sequence() as i64).bind(m.head().as_slice()).bind(evidence.as_slice()).execute(&mut *tx).await?;
552        tx.commit().await?;
553        Ok(())
554    }
555}
aven_core/sync/seed_claim/membership/persistence.rs

aven_core/sync/seed_claim/membership/
persistence.rs
