coven_database/
store_reclaim_records.rs

1use super::*;
2
3pub fn store_reclaim_journal_error(error: StoreReclaimJournalError) -> DbError {
4    DbError::from(error)
5}
6
7pub fn parse_store_reclaim_operation(
8    operation_id: ObjectHash,
9    raw: &str,
10) -> Result<DurableStoreReclaimOperation, DbError> {
11    let operation: DurableStoreReclaimOperation = serde_json::from_str(raw).map_err(|error| {
12        DbError::context(
13            format!("Store reclaim operation {operation_id} has invalid durable state"),
14            error,
15        )
16    })?;
17    operation.validate().map_err(store_reclaim_journal_error)?;
18    if operation.operation_id() != operation_id {
19        return Err(DbError::Message(format!(
20            "Store reclaim operation key {operation_id} differs from its authorization {}",
21            operation.operation_id()
22        )));
23    }
24    Ok(operation)
25}
26
27pub(crate) fn load_store_reclaim_operation_on(
28    conn: &Connection,
29    operation_id: ObjectHash,
30) -> Result<Option<DurableStoreReclaimOperation>, DbError> {
31    conn.query_row(
32        "SELECT state FROM store_reclaim_operations WHERE authorization_hash = ?1",
33        [operation_id.to_string()],
34        |row| row.get::<_, String>(0),
35    )
36    .optional()
37    .map_err(DbError::from)?
38    .map(|raw| parse_store_reclaim_operation(operation_id, &raw))
39    .transpose()
40}
41
42pub(crate) fn insert_store_reclaim_operation_on(
43    conn: &Connection,
44    operation: &DurableStoreReclaimOperation,
45) -> Result<(), DbError> {
46    operation.validate().map_err(store_reclaim_journal_error)?;
47    let state = serde_json::to_string(operation)
48        .map_err(|error| DbError::context("serialize Store reclaim operation", error))?;
49    conn.execute(
50        "INSERT INTO store_reclaim_operations (authorization_hash, state) VALUES (?1, ?2)",
51        (operation.operation_id().to_string(), state),
52    )
53    .map(|_| ())
54    .map_err(DbError::from)
55}
56
57/// Record why an operation cannot proceed, so every later cycle skips it.
58pub(crate) fn mark_store_reclaim_operation_stuck_on(
59    conn: &Connection,
60    operation_id: ObjectHash,
61    error: &str,
62) -> Result<(), DbError> {
63    // A stuck operation is one a person has to read about, so the mark carries
64    // a message even when the error's own display is empty.
65    let error = if error.trim().is_empty() {
66        format!("Store reclaim operation {operation_id} failed without a message")
67    } else {
68        error.to_string()
69    };
70    let updated = conn
71        .execute(
72            "UPDATE store_reclaim_operations SET stuck_error = ?2 WHERE authorization_hash = ?1",
73            (operation_id.to_string(), error),
74        )
75        .map_err(DbError::from)?;
76    if updated != 1 {
77        return Err(DbError::Message(format!(
78            "Store reclaim operation {operation_id} is absent and cannot be marked stuck"
79        )));
80    }
81    Ok(())
82}
83
84/// Clear a stuck mark so the next cycle runs the operation again. Refuses an
85/// operation that is not stuck, so a retry the host sends twice cannot pass as
86/// a second decision.
87pub(crate) fn clear_store_reclaim_operation_stuck_on(
88    conn: &Connection,
89    operation_id: ObjectHash,
90) -> Result<(), DbError> {
91    let updated = conn
92        .execute(
93            "UPDATE store_reclaim_operations SET stuck_error = NULL
94             WHERE authorization_hash = ?1 AND stuck_error IS NOT NULL",
95            [operation_id.to_string()],
96        )
97        .map_err(DbError::from)?;
98    if updated != 1 {
99        return Err(DbError::Message(format!(
100            "Store reclaim operation {operation_id} is not stuck"
101        )));
102    }
103    Ok(())
104}
105
106pub(crate) fn update_store_reclaim_operation_on(
107    conn: &Connection,
108    expected: &DurableStoreReclaimOperation,
109    next: &DurableStoreReclaimOperation,
110) -> Result<(), DbError> {
111    expected.validate().map_err(store_reclaim_journal_error)?;
112    next.validate().map_err(store_reclaim_journal_error)?;
113    if expected.operation_id() != next.operation_id() {
114        return Err(DbError::Message(
115            "Store reclaim transition changes its authorization identity".to_string(),
116        ));
117    }
118    let expected_state = serde_json::to_string(expected)
119        .map_err(|error| DbError::context("serialize expected Store reclaim state", error))?;
120    let next_state = serde_json::to_string(next)
121        .map_err(|error| DbError::context("serialize next Store reclaim state", error))?;
122    let updated = conn
123        .execute(
124            "UPDATE store_reclaim_operations SET state = ?3
125             WHERE authorization_hash = ?1 AND state = ?2",
126            (
127                expected.operation_id().to_string(),
128                expected_state,
129                next_state,
130            ),
131        )
132        .map_err(DbError::from)?;
133    if updated != 1 {
134        return Err(DbError::Message(
135            "Store reclaim operation changed during transition".to_string(),
136        ));
137    }
138    Ok(())
139}
coven_database/store_reclaim_records.rs

coven_database/
store_reclaim_records.rs
