coven_database/
claim.rs

1//! Durable identity and exact attempted bytes for one sequence claim (§5.1).
2
3use crate::internal_schema::WRITE_DEVICE_SQL;
4use crate::{
5    sqlite::DatabaseConnection,
6    write_encoding::{counter, damaged},
7    DbError, OperationId,
8};
9use coven_format::path::ObjectPath;
10use coven_foundation::id_source::DeviceId;
11use rusqlite::params;
12use std::io::Read;
13
14/// The original work carried by a claim. Every slot kind shares one counter.
15#[derive(Clone, Debug, PartialEq, Eq)]
16pub enum ClaimWork {
17    /// The queued originals through this write number, inclusive.
18    Writes(u64),
19    /// An admin, key-copy or snapshot-announcement step of an operation.
20    Operation(OperationId),
21}
22
23/// A create-once request, persisted before the provider is called.
24pub struct ClaimAttempt<R> {
25    /// The exact path bound by the sealed envelope.
26    pub path: ObjectPath,
27    /// The complete sealed bytes, reused for an uncertain request.
28    pub bytes: R,
29}
30
31/// Settlement of the most recent attempt.
32#[derive(Clone, Copy, Debug, PartialEq, Eq)]
33pub enum ClaimSettlement {
34    /// No authoritative read has settled the request.
35    Unknown,
36    /// The slot is still absent above the floor; resend its exact bytes.
37    Retry,
38    /// Other bytes occupy the slot, or the floor snapshot excludes this claim.
39    NotLanded,
40    /// Exact bytes or the floor snapshot establish publication.
41    Landed,
42    /// Authenticated claim evidence belongs to another copy of this device id.
43    DeviceIdInUse,
44}
45
46/// A claim retains every attempt, including attempts that lost contention.
47#[derive(Clone, Debug, PartialEq, Eq)]
48pub struct Claim {
49    /// The installation that reserved this number.
50    pub device: DeviceId,
51    /// One monotonically increasing number across every slot kind.
52    pub number: u64,
53    /// The queue batch or journal operation which owns it.
54    pub work: ClaimWork,
55    /// Attempts in their durable order; only the last can be outstanding.
56    pub attempts: Vec<ObjectPath>,
57    /// What authoritative storage reads established about the last attempt.
58    pub settlement: ClaimSettlement,
59}
60
61/// Authenticated observations supplied by the sync reader.
62pub enum ClaimRead {
63    /// The create succeeded and the subsequent authenticated floor still keeps
64    /// the attempted slot. No reread of our own ciphertext is required.
65    Created,
66    /// A slot at or above the checked floor. Absence permits exact retry.
67    Kept(Option<Box<dyn Read + Send>>),
68    /// A snapshot whose coverage includes the attempted slot below the floor.
69    Compacted {
70        /// The floor's first kept slot.
71        first: u64,
72        /// The authenticated snapshot's coverage.
73        covers: u64,
74        /// This device's last included claim in that snapshot.
75        last_claim: u64,
76        /// Digest paired with that number; absent exactly when it is zero.
77        digest: Option<[u8; 32]>,
78    },
79}
80
81fn settlement(text: String) -> rusqlite::Result<ClaimSettlement> {
82    match text.as_str() {
83        "unknown" => Ok(ClaimSettlement::Unknown),
84        "retry" => Ok(ClaimSettlement::Retry),
85        "not-landed" => Ok(ClaimSettlement::NotLanded),
86        "landed" => Ok(ClaimSettlement::Landed),
87        "device-id-in-use" => Ok(ClaimSettlement::DeviceIdInUse),
88        _ => Err(damaged(4, rusqlite::types::Type::Text)),
89    }
90}
91
92fn metadata(db: &DatabaseConnection) -> Result<Option<Claim>, DbError> {
93    db.query("SELECT device,number,through,operation,settlement FROM _coven_claim_journal WHERE singleton=1", [], |r| {
94        let work = match (r.get::<_,Option<Vec<u8>>>(2)?, r.get::<_,Option<i64>>(3)?) {
95            (Some(n),None) => ClaimWork::Writes(counter(n)?),
96            (None,Some(id)) => ClaimWork::Operation(OperationId(id)),
97            _ => return Err(damaged(2, rusqlite::types::Type::Blob)),
98        };
99        Ok(Claim { device:r.get::<_, DeviceId>(0)?, number:counter(r.get(1)?)?, work,
100            attempts:Vec::new(), settlement:settlement(r.get(4)?)? })
101    }).map(|mut rows| rows.pop())
102}
103
104pub(crate) fn current(db: &DatabaseConnection) -> Result<Option<Claim>, DbError> {
105    metadata(db)?.map(|mut claim| {
106        claim.attempts = db.query("SELECT path FROM _coven_claim_attempts WHERE device=?1 AND number=?2 ORDER BY attempt",
107            params![claim.device,claim.number.to_be_bytes().as_slice()], |r| {
108                let path:String=r.get(0)?;
109                ObjectPath::parse(&path).map_err(|e|crate::write_encoding::decode_failure(0,rusqlite::types::Type::Text,e))
110            })?;
111        Ok(claim)
112    }).transpose()
113}
114
115pub(crate) fn reserved(db: &DatabaseConnection, device: DeviceId) -> Result<u64, DbError> {
116    let previous: Option<Vec<u8>> = db.query_row(
117        "SELECT (SELECT number FROM _coven_claim_counters WHERE device=?1)",
118        [device],
119        |r| r.get(0),
120    )?;
121    Ok(previous.map(counter).transpose()?.unwrap_or(0))
122}
123
124pub(crate) fn begin<R: Read>(
125    db: &DatabaseConnection,
126    device: DeviceId,
127    work: ClaimWork,
128    number: u64,
129    prepared: crate::claim_preparation::PreparedClaim<R>,
130) -> Result<Claim, DbError> {
131    if metadata(db)?.is_some() {
132        return Err(DbError::ClaimPending);
133    }
134    // Concurrent callers may have committed another claim while this one sealed.
135    if reserved(db, device)?.checked_add(1) != Some(number) {
136        return Err(DbError::ClaimPending);
137    }
138    let (through, operation) = match work {
139        ClaimWork::Writes(through) => (Some(through.to_be_bytes().to_vec()), None),
140        ClaimWork::Operation(operation) => (None, Some(operation.0)),
141    };
142    db.internal_execute("INSERT INTO _coven_claim_counters(device,number) VALUES(?1,?2) ON CONFLICT(device) DO UPDATE SET number=excluded.number",
143        params![device,number.to_be_bytes().as_slice()])?;
144    db.internal_execute("INSERT INTO _coven_claim_journal(device,number,through,operation,settlement,singleton) VALUES(?1,?2,?3,?4,'unknown',1)",
145        params![device,number.to_be_bytes().as_slice(),through,operation])?;
146    let path = prepared.attempt.path.clone();
147    append(db, device, number, prepared)?;
148    if let ClaimWork::Writes(through) = work {
149        db.internal_execute(
150            "INSERT INTO _coven_upload_claims(device,number,claim)
151             SELECT device,number,?1 FROM _coven_uploads WHERE device=?2 AND number<=?3
152             ON CONFLICT(device,number) DO UPDATE SET claim=excluded.claim",
153            params![number.to_be_bytes().as_slice(), device, through.to_be_bytes().as_slice()],
154        )?;
155    }
156    Ok(Claim { device, number, work, attempts: vec![path], settlement: ClaimSettlement::Unknown })
157}
158
159fn append(
160    db: &DatabaseConnection,
161    device: DeviceId,
162    number: u64,
163    prepared: crate::claim_preparation::PreparedClaim<impl Read>,
164) -> Result<(), DbError> {
165    let crate::claim_preparation::PreparedClaim { mut attempt, digest, size } = prepared;
166    let ordinal: i64 = db.query_row("SELECT coalesce(max(attempt),0)+1 FROM _coven_claim_attempts WHERE device=?1 AND number=?2",
167        params![device,number.to_be_bytes()], |r|r.get(0))?;
168    db.internal_execute(
169        "INSERT INTO _coven_claim_attempts(device,number,attempt,path) VALUES(?1,?2,?3,?4)",
170        params![device, number.to_be_bytes(), ordinal, attempt.path.as_str()],
171    )?;
172    for chunk in 0i64.. {
173        let bytes = read_chunk(&mut attempt.bytes)?;
174        if bytes.is_empty() {
175            break;
176        }
177        db.internal_execute("INSERT INTO _coven_claim_chunks(device,number,attempt,chunk,bytes) VALUES(?1,?2,?3,?4,?5)",
178            params![device,number.to_be_bytes(),ordinal,chunk,bytes])?;
179    }
180    db.internal_execute(
181        "UPDATE _coven_claim_attempts SET digest=?4,size=?5 WHERE device=?1 AND number=?2 AND attempt=?3",
182        params![
183            device,
184            number.to_be_bytes(),
185            ordinal,
186            digest.as_bytes(),
187            size.to_be_bytes()
188        ],
189    )?;
190    Ok(())
191}
192
193/// The next chunk of a claim's ciphertext stream: up to one chunk's bytes,
194/// none at its end.
195pub(crate) fn read_chunk(bytes: &mut dyn Read) -> Result<Vec<u8>, DbError> {
196    let mut chunk = Vec::new();
197    bytes
198        .take(coven_format::chunks::CHUNK_SIZE as u64)
199        .read_to_end(&mut chunk)
200        .map_err(DbError::ClaimRead)?;
201    Ok(chunk)
202}
203
204/// Bounded reads of the current attempt's durable ciphertext.
205pub struct ClaimBytes<'a> {
206    db: &'a DatabaseConnection,
207    device: DeviceId,
208    number: u64,
209    attempt: i64,
210    chunk: i64,
211    ended: bool,
212}
213
214impl Iterator for ClaimBytes<'_> {
215    type Item = Result<Vec<u8>, DbError>;
216
217    fn next(&mut self) -> Option<Self::Item> {
218        if self.ended {
219            return None;
220        }
221        let bytes = self.db.query_row("SELECT (SELECT bytes FROM _coven_claim_chunks WHERE device=?1 AND number=?2 AND attempt=?3 AND chunk=?4)",
222            params![self.device,self.number.to_be_bytes(),self.attempt,self.chunk], |r|r.get::<_,Option<Vec<u8>>>(0));
223        match bytes {
224            Ok(Some(bytes)) => {
225                self.chunk += 1;
226                Some(Ok(bytes))
227            }
228            Ok(None) => {
229                self.ended = true;
230                None
231            }
232            Err(error) => {
233                self.ended = true;
234                Some(Err(error))
235            }
236        }
237    }
238}
239
240pub(crate) fn read_attempt<T, E: From<DbError>>(
241    db: &DatabaseConnection,
242    consume: impl FnOnce(ClaimBytes<'_>) -> Result<T, E>,
243) -> Result<T, E> {
244    let claim = metadata(db)?.ok_or(DbError::NoClaim)?;
245    let attempt = db.query_row(
246        "SELECT max(attempt) FROM _coven_claim_attempts WHERE device=?1 AND number=?2",
247        params![claim.device, claim.number.to_be_bytes()],
248        |r| r.get(0),
249    )?;
250    consume(ClaimBytes {
251        db,
252        device: claim.device,
253        number: claim.number,
254        attempt,
255        chunk: 0,
256        ended: false,
257    })
258}
259
260fn equal(db: &DatabaseConnection, mut received: Box<dyn Read + Send>) -> Result<bool, DbError> {
261    read_attempt(db, |mut saved| {
262        let mut equal = true;
263        loop {
264            let incoming = read_chunk(&mut received)?;
265            let stored = saved.next().transpose()?;
266            if incoming.is_empty() && stored.is_none() {
267                return Ok(equal);
268            }
269            equal &= stored.as_deref() == Some(incoming.as_slice());
270        }
271    })
272}
273
274pub(crate) fn retry(
275    db: &DatabaseConnection,
276    attempt: crate::claim_preparation::PreparedClaim<impl Read>,
277) -> Result<(), DbError> {
278    let claim = metadata(db)?.ok_or(DbError::NoClaim)?;
279    if claim.settlement != ClaimSettlement::NotLanded {
280        return Err(DbError::ClaimUnsettled);
281    }
282    append(db, claim.device, claim.number, attempt)?;
283    db.internal_execute(
284        "UPDATE _coven_claim_journal SET settlement='unknown' WHERE singleton=1",
285        [],
286    )?;
287    Ok(())
288}
289
290pub(crate) fn settle(db: &DatabaseConnection, read: ClaimRead) -> Result<ClaimSettlement, DbError> {
291    let claim = current(db)?.ok_or(DbError::NoClaim)?;
292    if matches!(
293        claim.settlement,
294        ClaimSettlement::Landed | ClaimSettlement::NotLanded | ClaimSettlement::DeviceIdInUse
295    ) {
296        return Ok(claim.settlement);
297    }
298    let attempt = claim.attempts.last().expect("a journal starts with its first attempt");
299    let (result, name) = match read {
300        ClaimRead::Created => (ClaimSettlement::Landed, "landed"),
301        ClaimRead::Kept(Some(bytes)) => {
302            if equal(db, bytes)? {
303                (ClaimSettlement::Landed, "landed")
304            } else {
305                (ClaimSettlement::NotLanded, "not-landed")
306            }
307        }
308        ClaimRead::Kept(None) => (ClaimSettlement::Retry, "retry"),
309        ClaimRead::Compacted { first, covers, last_claim, digest } => {
310            let slot = attempt.slot_number().ok_or(DbError::InvalidClaimCoverage)?;
311            if first <= slot.get() || covers < first - 1 {
312                return Err(DbError::InvalidClaimCoverage);
313            }
314            if last_claim < claim.number {
315                (ClaimSettlement::NotLanded, "not-landed")
316            } else if last_claim > claim.number {
317                (ClaimSettlement::DeviceIdInUse, "device-id-in-use")
318            } else {
319                let attempted_digest: Vec<u8> = db.query_row(
320                    "SELECT digest FROM _coven_claim_attempts WHERE device=?1 AND number=?2 ORDER BY attempt DESC LIMIT 1",
321                    params![claim.device, claim.number.to_be_bytes()],
322                    |r| r.get(0),
323                )?;
324                if digest.as_ref().map(|value| value.as_slice())
325                    == Some(attempted_digest.as_slice())
326                {
327                    (ClaimSettlement::Landed, "landed")
328                } else {
329                    (ClaimSettlement::DeviceIdInUse, "device-id-in-use")
330                }
331            }
332        }
333    };
334    db.internal_execute("UPDATE _coven_claim_journal SET settlement=?1 WHERE singleton=1", [name])?;
335    Ok(result)
336}
337
338/// The operation whose open claim the applied slots record as landed,
339/// whatever its settlement says, with the slot its last attempt claimed:
340/// this device's last recorded claim has its number and that attempt's
341/// digest. An unsettled claim of a slot applied since, or compacted under a
342/// loaded floor, is one (§5.1).
343pub(crate) fn landed_operation(
344    db: &DatabaseConnection,
345) -> Result<Option<(OperationId, coven_merge::Slot)>, DbError> {
346    let Some(claim) = metadata(db)? else {
347        return Ok(None);
348    };
349    let ClaimWork::Operation(id) = claim.work else {
350        return Ok(None);
351    };
352    let found = db.query(
353        "SELECT a.path FROM _coven_claim_attempts AS a
354         JOIN _coven_last_claims AS c ON c.device=a.device AND c.number=a.number
355             AND c.digest=a.digest
356         WHERE a.device=?1 AND a.number=?2
357             AND a.attempt=(SELECT max(attempt) FROM _coven_claim_attempts
358                 WHERE device=?1 AND number=?2)",
359        params![claim.device, claim.number.to_be_bytes()],
360        |r| r.get::<_, String>(0),
361    )?;
362    let Some(path) = found.into_iter().next() else {
363        return Ok(None);
364    };
365    let path = ObjectPath::parse(&path)
366        .map_err(|e| crate::write_encoding::decode_failure(0, rusqlite::types::Type::Text, e))?;
367    Ok(Some((id, path.slot_number().ok_or(DbError::InvalidClaimCoverage)?)))
368}
369
370/// Whether the open claim is the operation `id`'s.
371pub(crate) fn held_by(db: &DatabaseConnection, id: OperationId) -> Result<bool, DbError> {
372    Ok(metadata(db)?.is_some_and(|claim| claim.work == ClaimWork::Operation(id)))
373}
374
375/// End the open claim, settled or not, of an operation that ends without
376/// it: this device has stopped for good, and no attempt of it can count.
377pub(crate) fn drop_operation(db: &DatabaseConnection, id: OperationId) -> Result<(), DbError> {
378    if held_by(db, id)? {
379        db.internal_execute("DELETE FROM _coven_claim_journal WHERE singleton=1", [])?;
380    }
381    Ok(())
382}
383
384/// Finish only a settled attempt. The owning operation controls whether a lost
385/// claim is abandoned or retried; a landed batch removes its original queue rows.
386pub(crate) fn finish(db: &DatabaseConnection) -> Result<(), DbError> {
387    let claim = metadata(db)?.ok_or(DbError::NoClaim)?;
388    let mut writes_landed = false;
389    match claim.settlement {
390        ClaimSettlement::Unknown | ClaimSettlement::Retry => return Err(DbError::ClaimUnsettled),
391        ClaimSettlement::Landed if matches!(claim.work, ClaimWork::Writes(_)) => {
392            // A write still queued keeps its entry, with no slot; a landed
393            // one's entry may be forgotten (§16.7).
394            let unconsumed: bool = db.query_row(
395                &format!("SELECT EXISTS(SELECT 1 FROM _coven_upload_claims u JOIN _coven_writes w ON {WRITE_DEVICE_SQL}=u.device AND w.number=u.number WHERE u.device=?1 AND u.claim=?2 AND w.slot IS NULL)"),
396                params![claim.device, claim.number.to_be_bytes()],
397                |r| r.get(0),
398            )?;
399            if unconsumed {
400                return Err(DbError::ClaimUnsettled);
401            }
402            crate::upload::forget_claimed(db, claim.device, claim.number)?;
403            writes_landed = true;
404        }
405        _ => (),
406    }
407    db.internal_execute("DELETE FROM _coven_claim_journal WHERE singleton=1", [])?;
408    if writes_landed {
409        db.give_back_free_pages()?;
410    }
411    Ok(())
412}
413
414/// Path evidence distinguishes contention from an unattempted use of this id.
415#[cfg(any(test, feature = "test-utils"))]
416pub(crate) fn attempted(db: &DatabaseConnection, path: &ObjectPath) -> Result<bool, DbError> {
417    db.query_row(
418        "SELECT EXISTS(SELECT 1 FROM _coven_claim_attempts WHERE path=?1)",
419        [path.as_str()],
420        |r| r.get(0),
421    )
422}
423
424pub(crate) fn attempted_digest(
425    db: &DatabaseConnection,
426    number: u64,
427    path: &ObjectPath,
428    digest: &[u8; 32],
429) -> Result<bool, DbError> {
430    db.query_row("SELECT EXISTS(SELECT 1 FROM _coven_claim_attempts WHERE number=?1 AND path=?2 AND digest=?3)",
431        params![number.to_be_bytes(), path.as_str(), digest], |r| r.get(0))
432}
433
434/// Durable transfer state for the current attempt's exact sealed bytes.
435pub struct ClaimUpload {
436    /// The complete ciphertext length.
437    pub size: u64,
438    /// Opaque provider session retained after each confirmed part.
439    pub session: Option<Vec<u8>>,
440}
441
442pub(crate) fn upload(db: &DatabaseConnection) -> Result<ClaimUpload, DbError> {
443    let claim = metadata(db)?.ok_or(DbError::NoClaim)?;
444    db.query_row("SELECT size,session FROM _coven_claim_attempts WHERE device=?1 AND number=?2 ORDER BY attempt DESC LIMIT 1",
445        params![claim.device, claim.number.to_be_bytes()], |row| Ok(ClaimUpload {size: counter(row.get(0)?)?, session: row.get(1)?}))
446}
447
448pub(crate) fn save_upload(db: &DatabaseConnection, session: Vec<u8>) -> Result<(), DbError> {
449    let claim = metadata(db)?.ok_or(DbError::NoClaim)?;
450    db.internal_execute("UPDATE _coven_claim_attempts SET session=?3 WHERE device=?1 AND number=?2 AND attempt=(SELECT max(attempt) FROM _coven_claim_attempts WHERE device=?1 AND number=?2)",
451        params![claim.device, claim.number.to_be_bytes(), session])?;
452    Ok(())
453}
454
455pub(crate) fn range(
456    db: &DatabaseConnection,
457    offset: u64,
458    length: usize,
459) -> Result<Vec<u8>, DbError> {
460    if length == 0 {
461        return Ok(Vec::new());
462    }
463    let claim = metadata(db)?.ok_or(DbError::NoClaim)?;
464    let chunk_size = coven_format::chunks::CHUNK_SIZE as u64;
465    let end = offset.checked_add(length as u64).ok_or(DbError::DamagedDatabase)?;
466    let pieces = db.query("SELECT chunk,bytes FROM _coven_claim_chunks WHERE device=?1 AND number=?2 AND attempt=(SELECT max(attempt) FROM _coven_claim_attempts WHERE device=?1 AND number=?2) AND chunk>=?3 AND chunk<=?4 ORDER BY chunk",
467        params![claim.device, claim.number.to_be_bytes(), (offset / chunk_size) as i64, ((end-1) / chunk_size) as i64],
468        |row| Ok((row.get::<_,i64>(0)? as u64, row.get::<_,Vec<u8>>(1)?)))?;
469    let mut bytes = Vec::with_capacity(length);
470    for (index, piece) in pieces {
471        let start = offset.saturating_sub(index * chunk_size) as usize;
472        let last = (end - index * chunk_size).min(piece.len() as u64) as usize;
473        bytes.extend_from_slice(piece.get(start..last).ok_or(DbError::DamagedDatabase)?);
474    }
475    if bytes.len() != length {
476        return Err(DbError::DamagedDatabase);
477    }
478    Ok(bytes)
479}
480
481#[cfg(test)]
482#[path = "claim_tests.rs"]
483pub(crate) mod tests;
coven_database/claim.rs

coven_database/
claim.rs
