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.has_device(inviter))
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}
176pub use aven_protocol::claim::membership::ManagementPreparation;
177pub use aven_protocol::wire::enrollment::CancelStatus;
178async fn append_transition(
179    conn: &mut SqliteConnection,
180    before: &Membership,
181    next: &Membership,
182    handle: Option<Hash>,
183    record: &[u8],
184) -> Result<()> {
185    sqlx::query("INSERT INTO server_membership_transitions(sequence,handle,record) VALUES(?,?,?)")
186        .bind(next.sequence() as i64)
187        .bind(handle.map(|h| h.to_vec()))
188        .bind(record)
189        .execute(&mut *conn)
190        .await?;
191    let changed = sqlx::query("UPDATE server_e2ee_membership_head SET sequence=?,commitment=? WHERE singleton=1 AND sequence=? AND commitment=?")
192        .bind(next.sequence() as i64).bind(next.head().as_slice()).bind(before.sequence() as i64).bind(before.head().as_slice()).execute(conn).await?;
193    ensure!(
194        changed.rows_affected() == 1,
195        "error membership-head-invalid"
196    );
197    Ok(())
198}
199pub(crate) async fn allocator(conn: &mut SqliteConnection, m: &Membership) -> Result<u64> {
200    let b = m.publication().binding();
201    let (n, high): (i64, i64) = sqlx::query_as(
202        "SELECT prefix_count,high_water FROM server_e2ee_allocator WHERE singleton=1 AND stream=?",
203    )
204    .bind(b.stream_id.as_slice())
205    .fetch_one(conn)
206    .await?;
207    ensure!(
208        n >= 0
209            && n as u64 == b.prefix_count
210            && high >= n
211            && high as u64 >= m.current_generation().starts_after,
212        "error membership-allocator"
213    );
214    Ok(high as u64)
215}
216impl Database {
217    pub async fn prepare_membership_management(
218        &self,
219        auth: &Authentication<'_>,
220    ) -> Result<ManagementPreparation> {
221        let mut conn = self.acquire_writer().await?;
222        let mut tx = begin_immediate(&mut conn).await?;
223        let c = current(self, &mut tx).await?;
224        c.membership.authenticate(auth, false)?;
225        let high_water = allocator(&mut tx, &c.membership).await?;
226        tx.commit().await?;
227        Ok(ManagementPreparation {
228            evidence: c.evidence.clone(),
229            high_water,
230        })
231    }
232    pub async fn apply_membership_management(
233        &self,
234        auth: &Authentication<'_>,
235        record: &[u8],
236    ) -> Result<Vec<u8>> {
237        ensure!(record.len() <= MAX_RECORD_BYTES, "error membership-limit");
238        let mut conn = self.acquire_writer().await?;
239        let mut tx = begin_immediate(&mut conn).await?;
240        let c = current(self, &mut tx).await?;
241        c.membership.authenticate(auth, false)?;
242        let (signer, action) = Membership::record_action(record)?;
243        ensure!(matches!(action, 4 | 5), "error membership-action");
244        if !c.evidence.transitions.iter().any(|t| t.record == record) {
245            ensure!(signer == auth.device, super::Unauthorized);
246            let next = c.membership.append(&[], &[], record)?;
247            let high = allocator(&mut tx, &c.membership).await?;
248            ensure!(
249                action != 5 || next.current_generation().starts_after == high,
250                "error membership-cutoff"
251            );
252            append_transition(&mut tx, &c.membership, &next, None, record).await?;
253            for device in c.membership.devices() {
254                if !next.has_device(device) {
255                    sqlx::query("UPDATE server_membership_invitations SET revoked=1,expired=1 WHERE inviter=? AND admitted_sequence IS NULL")
256                        .bind(device.as_slice()).execute(&mut *tx).await?;
257                    sqlx::query("DELETE FROM server_e2ee_image_tickets WHERE device=?")
258                        .bind(device.as_slice())
259                        .execute(&mut *tx)
260                        .await?;
261                }
262            }
263        }
264        tx.commit().await?;
265        Ok(record.to_vec())
266    }
267}
268impl Database {
269    /// Replays the complete stored membership history and fills the cache.
270    /// Servers run this before accepting requests; storage without published
271    /// membership has nothing to verify.
272    pub async fn verify_membership_history(&self) -> Result<()> {
273        let mut conn = self.acquire_writer().await?;
274        let mut tx = begin_immediate(&mut conn).await?;
275        let published: bool =
276            sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_e2ee_membership_head)")
277                .fetch_one(&mut *tx)
278                .await?;
279        if published {
280            self.membership_cache.set(Arc::new(replay(&mut tx).await?));
281            current(self, &mut tx).await?;
282        }
283        tx.commit().await?;
284        Ok(())
285    }
286    /// This read alone accepts a verified ancestor head, after current credential authentication.
287    pub async fn membership_evidence(&self, auth: &Authentication<'_>) -> Result<Evidence> {
288        let mut conn = self.acquire_writer().await?;
289        let mut tx = begin_immediate(&mut conn).await?;
290        let c = current(self, &mut tx).await?;
291        c.membership.authenticate(auth, true)?;
292        tx.commit().await?;
293        Ok(c.evidence.clone())
294    }
295    pub async fn register_membership_invitation_at(
296        &self,
297        auth: &Authentication<'_>,
298        raw: &[u8],
299        time: i64,
300    ) -> Result<RegistrationStatus> {
301        ensure!(raw.len() == DECLARATION_BYTES, "error membership-limit");
302        let mut conn = self.acquire_writer().await?;
303        let mut tx = begin_immediate(&mut conn).await?;
304        let c = current(self, &mut tx).await?;
305        c.membership.authenticate(auth, false)?;
306        let d = Declaration::from_record(&c.membership, raw)?;
307        ensure!(d.inviter() == auth.device, super::Unauthorized);
308        let expiry = i64::try_from(d.expiry()).context("error enrollment-expired")?;
309        let high = observe(&mut tx, time).await?;
310        let stored: Option<Vec<u8>> = sqlx::query_scalar(
311            "SELECT declaration FROM server_membership_invitations WHERE handle=?",
312        )
313        .bind(d.handle().as_slice())
314        .fetch_optional(&mut *tx)
315        .await?;
316        if let Some(stored) = stored {
317            ensure!(stored == raw, "error enrollment-conflict");
318        } else {
319            let count: i64 =
320                sqlx::query_scalar("SELECT count(*) FROM server_membership_invitations")
321                    .fetch_one(&mut *tx)
322                    .await?;
323            ensure!(
324                count < MAX_INVITATIONS as i64 && c.membership.device_count() < MAX_DEVICES,
325                "error membership-limit"
326            );
327            ensure!(
328                expiry <= high || expiry - high <= 3600,
329                "error enrollment-expired"
330            );
331            let unfinished: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM server_membership_invitations WHERE inviter=? AND expired=0 AND admitted_sequence IS NULL)")
332                .bind(d.inviter().as_slice()).fetch_one(&mut *tx).await?;
333            if unfinished {
334                tx.commit().await?;
335                anyhow::bail!("error enrollment-unfinished");
336            }
337            sqlx::query("INSERT INTO server_membership_invitations(handle,inviter,declaration,expiry,expired) VALUES(?,?,?,?,?)")
338                .bind(d.handle().as_slice()).bind(d.inviter().as_slice()).bind(raw).bind(expiry).bind(expiry <= high).execute(&mut *tx).await?;
339        }
340        let (expired, consumed): (bool, bool) = sqlx::query_as("SELECT expired,admitted_sequence IS NOT NULL FROM server_membership_invitations WHERE handle=?")
341            .bind(d.handle().as_slice()).fetch_one(&mut *tx).await?;
342        tx.commit().await?;
343        Ok(if consumed {
344            RegistrationStatus::Consumed
345        } else if expired {
346            RegistrationStatus::Expired
347        } else {
348            RegistrationStatus::Open
349        })
350    }
351    pub async fn post_membership_request_at(
352        &self,
353        vault: Hash,
354        handle: Hash,
355        request: &[u8],
356        time: i64,
357    ) -> Result<()> {
358        request_parts(request, &handle)?;
359        let mut conn = self.acquire_writer().await?;
360        let mut tx = begin_immediate(&mut conn).await?;
361        let c = current(self, &mut tx).await?;
362        check(vault == c.membership.genesis().context().vault_id)?;
363        observe(&mut tx, time).await?;
364        let (stored, expired, revoked, admitted): (Option<Vec<u8>>, bool, bool, bool) = sqlx::query_as(
365            "SELECT request,expired,revoked,admitted_sequence IS NOT NULL FROM server_membership_invitations WHERE handle=?",
366        )
367        .bind(handle.as_slice())
368        .fetch_optional(&mut *tx)
369        .await?
370        .context("error enrollment-unavailable")?;
371        ensure!(
372            !revoked && (!admitted || c.membership.has_admission(handle)),
373            "error enrollment-revoked"
374        );
375        if let Some(stored) = stored {
376            ensure!(stored == request, "error enrollment-request-occupied");
377        } else {
378            if expired {
379                tx.commit().await?;
380                anyhow::bail!("error enrollment-expired");
381            }
382            sqlx::query("UPDATE server_membership_invitations SET request=? WHERE handle=?")
383                .bind(request)
384                .bind(handle.as_slice())
385                .execute(&mut *tx)
386                .await?;
387        }
388        tx.commit().await?;
389        Ok(())
390    }
391    pub async fn membership_mailbox(&self, vault: Hash, handle: Hash) -> Result<Mailbox> {
392        let mut conn = self.acquire_writer().await?;
393        let mut tx = begin_immediate(&mut conn).await?;
394        let c = current(self, &mut tx).await?;
395        check(vault == c.membership.genesis().context().vault_id)?;
396        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=?")
397            .bind(handle.as_slice()).fetch_optional(&mut *tx).await?.context("error enrollment-unavailable")?;
398        if admission.is_some() {
399            ensure!(
400                c.membership.has_admission(handle),
401                "error enrollment-revoked"
402            );
403        } else {
404            let d = Declaration::from_record(&c.membership, &declaration)?;
405            check(d.handle() == handle)?;
406        }
407        tx.commit().await?;
408        Ok(Mailbox {
409            declaration,
410            request,
411            admission,
412        })
413    }
414    /// Terminally fences a registered, unadmitted invitation of the caller so
415    /// no new admission can commit. Serialized with admission; idempotent.
416    /// Unknown handles allocate nothing.
417    pub async fn cancel_membership_invitation_at(
418        &self,
419        auth: &Authentication<'_>,
420        handle: Hash,
421        time: i64,
422    ) -> Result<CancelStatus> {
423        let mut conn = self.acquire_writer().await?;
424        let mut tx = begin_immediate(&mut conn).await?;
425        let c = current(self, &mut tx).await?;
426        c.membership.authenticate(auth, false)?;
427        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=?")
428            .bind(handle.as_slice())
429            .fetch_optional(&mut *tx)
430            .await?
431            .context("error enrollment-unavailable")?;
432        // The immutable stored row binds the exact declaration and its inviter.
433        ensure!(
434            inviter == auth.device
435                && Declaration::from_record(&c.membership, &declaration)?.handle() == handle,
436            super::Unauthorized
437        );
438        let status = if admitted {
439            CancelStatus::Admitted
440        } else {
441            observe(&mut tx, time).await?;
442            sqlx::query("UPDATE server_membership_invitations SET expired=1 WHERE handle=?")
443                .bind(handle.as_slice())
444                .execute(&mut *tx)
445                .await?;
446            CancelStatus::Cancelled
447        };
448        tx.commit().await?;
449        Ok(status)
450    }
451    pub async fn admit_membership_device_at(
452        &self,
453        auth: &Authentication<'_>,
454        handle: Hash,
455        record: &[u8],
456        time: i64,
457    ) -> Result<Vec<u8>> {
458        ensure!(record.len() <= MAX_RECORD_BYTES, "error membership-limit");
459        let mut conn = self.acquire_writer().await?;
460        let mut tx = begin_immediate(&mut conn).await?;
461        let c = current(self, &mut tx).await?;
462        c.membership.authenticate(auth, false)?;
463        type Row = (Vec<u8>, Option<Vec<u8>>, Option<Vec<u8>>, bool);
464        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=?")
465            .bind(handle.as_slice()).fetch_optional(&mut *tx).await?.context("error enrollment-unavailable")?;
466        if let Some(saved) = saved {
467            ensure!(saved == record, "error enrollment-conflict");
468            observe(&mut tx, time).await?;
469        } else {
470            let d = Declaration::from_record(&c.membership, &declaration)?;
471            ensure!(
472                d.handle() == handle && d.inviter() == auth.device,
473                super::Unauthorized
474            );
475            let high = observe(&mut tx, time).await?;
476            // Cancelled and clock-expired rows stay terminal for new admission.
477            if fenced || high as u64 >= d.expiry() {
478                tx.commit().await?;
479                anyhow::bail!("error enrollment-expired");
480            }
481            let request = request.context("error enrollment-request-missing")?;
482            let next = c.membership.append(&declaration, &request, record)?;
483            append_transition(&mut tx, &c.membership, &next, Some(handle), record).await?;
484            sqlx::query(
485                "UPDATE server_membership_invitations SET admitted_sequence=? WHERE handle=?",
486            )
487            .bind(next.sequence() as i64)
488            .bind(handle.as_slice())
489            .execute(&mut *tx)
490            .await?;
491        }
492        tx.commit().await?;
493        Ok(record.to_vec())
494    }
495}
496#[cfg(test)]
497mod tests;
498
499impl Database {
500    pub async fn membership_checkpoint_mirror(&self) -> Result<Option<(Hash, u64, Hash, Hash)>> {
501        let mut conn = self.acquire_reader().await?;
502        type Row = (Vec<u8>, i64, Vec<u8>, Vec<u8>);
503        let row: Option<Row> = sqlx::query_as("SELECT identity,sequence,head,evidence FROM local_membership_checkpoint WHERE singleton=1").fetch_optional(&mut *conn).await?;
504        row.map(|(id, seq, head, evidence)| {
505            Ok((
506                id.try_into()
507                    .map_err(|_| anyhow::anyhow!("error membership-mirror"))?,
508                u64::try_from(seq)?,
509                head.try_into()
510                    .map_err(|_| anyhow::anyhow!("error membership-mirror"))?,
511                evidence
512                    .try_into()
513                    .map_err(|_| anyhow::anyhow!("error membership-mirror"))?,
514            ))
515        })
516        .transpose()
517    }
518    pub async fn mirror_membership_checkpoint(
519        &self,
520        identity: Hash,
521        m: &Membership,
522        evidence: Hash,
523    ) -> Result<()> {
524        let mut conn = self.acquire_writer().await?;
525        let mut tx = begin_immediate(&mut conn).await?;
526        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?;
527        ensure!(valid, "error membership-identity");
528        type MirrorRow = (Vec<u8>, i64, Vec<u8>, Vec<u8>);
529        let old: Option<MirrorRow> = sqlx::query_as("SELECT identity,sequence,head,evidence FROM local_membership_checkpoint WHERE singleton=1").fetch_optional(&mut *tx).await?;
530        if let Some((id, seq, head, digest)) = old {
531            ensure!(
532                id == identity
533                    && m.head_at(u64::try_from(seq)?).is_some_and(|h| head == h)
534                    && (seq != m.sequence() as i64 || digest == evidence),
535                "error membership-mirror"
536            );
537        }
538        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")
539            .bind(identity.as_slice()).bind(m.sequence() as i64).bind(m.head().as_slice()).bind(evidence.as_slice()).execute(&mut *tx).await?;
540        tx.commit().await?;
541        Ok(())
542    }
543}
aven_core/sync/seed_claim/membership/persistence.rs

aven_core/sync/seed_claim/membership/
persistence.rs
