dpm_store/sqlite/runs/
write.rs

1//! Atomic, idempotent run writes. None of them touches the project store.
2//!
3//! Each write is one transaction under the run store's own write lock, so a heartbeat can never
4//! hold up a project command. A resent identity is looked up inside the transaction: identical
5//! content returns what was recorded, different content is refused, and nothing is written twice.
6
7use super::{
8    RunStore, Written,
9    codec::{optional_time_text, time_text},
10    read::{event_by_id, last_event, link_by_operation, load_run},
11    read_binding, schema,
12};
13use crate::{
14    RunStoreError, StoreError, StoreLineage, StoredRecord, error::database_error, schema::Layout,
15};
16use chrono::{DateTime, Utc};
17use dpm_engine::{
18    OperationFacts, RunError, authorize_run, check_link, check_run_transition, run_start_event,
19};
20use dpm_model::{
21    ActivityEntry, ActivityInput, ActorId, LifecycleEntry, LifecycleEvent, LineageId,
22    MAX_ACTIVITY_BATCH, RunId, RunLink, RunRecord, RunState, RunTransition, ValidationError,
23};
24use rusqlite::{Connection, Transaction, TransactionBehavior, params};
25use std::{collections::BTreeSet, path::Path};
26
27pub(super) fn to_json<T: serde::Serialize>(value: &T) -> Result<String, StoreError> {
28    serde_json::to_string(value).map_err(StoreError::from)
29}
30
31/// Begin a write transaction after re-checking the layout and the binding under the lock: an
32/// archive is never written, and neither is a store bound to another history than the caller's.
33fn begin<'a>(
34    connection: &'a mut Connection,
35    path: &Path,
36    expected: LineageId,
37) -> Result<Transaction<'a>, StoreError> {
38    let transaction = connection
39        .transaction_with_behavior(TransactionBehavior::Immediate)
40        .map_err(database_error(path, "begin run write"))?;
41    if schema::check_blocking(&transaction, path)? == Layout::Empty {
42        return Err(StoreError::NotInitialized(path.to_path_buf()));
43    }
44    let binding = read_binding(&transaction, path)?;
45    StoreLineage {
46        lineage_id: binding.lineage_id,
47        archived: binding.archived,
48    }
49    .check_writable(Some(expected), path)?;
50    Ok(transaction)
51}
52
53/// A run the store may take new facts for: it exists and was recorded under the lineage the store
54/// continues. A run from before a restore is a historical observation and is never written to,
55/// whether the request is new or a retry.
56fn current_run(
57    transaction: &Transaction<'_>,
58    path: &Path,
59    id: RunId,
60    expected: LineageId,
61) -> Result<RunRecord, RunStoreError> {
62    let record = load_run(transaction, path, id)?.ok_or(RunStoreError::UnknownRun(id))?;
63    if record.contract.lineage_id != expected {
64        return Err(RunStoreError::ForeignRun {
65            run: id,
66            recorded: record.contract.lineage_id,
67            current: expected,
68        });
69    }
70    Ok(record)
71}
72
73fn writable_run(
74    transaction: &Transaction<'_>,
75    path: &Path,
76    id: RunId,
77    expected: LineageId,
78    writer: &ActorId,
79    action: &'static str,
80) -> Result<RunRecord, RunStoreError> {
81    let record = current_run(transaction, path, id, expected)?;
82    authorize_run(&record, writer, action)?;
83    Ok(record)
84}
85
86fn commit(transaction: Transaction<'_>, path: &Path) -> Result<(), StoreError> {
87    transaction
88        .commit()
89        .map_err(database_error(path, "commit run write"))
90}
91
92fn insert_event(
93    transaction: &Transaction<'_>,
94    path: &Path,
95    event: &LifecycleEvent,
96) -> Result<u64, StoreError> {
97    transaction
98        .execute(
99            "INSERT INTO run_lifecycle(event_id, run_id, state, detail, observed_at, \
100             recorded_by_json, recorded_at) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7)",
101            params![
102                event.id.to_string(),
103                event.run.to_string(),
104                event.state.word(),
105                event.detail,
106                optional_time_text(event.observed_at),
107                to_json(&event.recorded_by)?,
108                time_text(event.recorded_at),
109            ],
110        )
111        .map_err(database_error(path, "append lifecycle fact"))?;
112    u64::try_from(transaction.last_insert_rowid()).map_err(|_| corrupt_sequence(path))
113}
114
115pub(super) fn corrupt_sequence(path: &Path) -> StoreError {
116    StoreError::CorruptColumn {
117        path: path.to_path_buf(),
118        record: StoredRecord::Run,
119        field: "sequence",
120        source: rusqlite::Error::IntegralValueOutOfRange(0, 0),
121    }
122}
123
124fn invalid_batch(reason: String) -> RunStoreError {
125    RunStoreError::Run(RunError::Validation(ValidationError::Invalid {
126        entity: "run activity",
127        id: "batch".into(),
128        reason,
129    }))
130}
131
132impl RunStore {
133    /// Record a run's start as its first lifecycle fact, in the history `expected`.
134    ///
135    /// The run identity is the idempotency key: a resent start that matches the recorded one
136    /// returns it unchanged, and any other content under that identity is refused.
137    pub fn start_blocking(
138        &mut self,
139        record: &RunRecord,
140        expected: LineageId,
141    ) -> Result<Written<LifecycleEntry>, RunStoreError> {
142        let path = self.path.clone();
143        let transaction = begin(&mut self.connection, &path, expected)?;
144        if load_run(&transaction, &path, record.id)?.is_some() {
145            let recorded = current_run(&transaction, &path, record.id, expected)?;
146            if recorded.request() != record.request() || recorded.recorded_by != record.recorded_by
147            {
148                return Err(RunStoreError::DuplicateRun {
149                    recorded: Box::new(recorded),
150                });
151            }
152            let first = event_by_id(&transaction, &path, run_start_event(record.id))?
153                .ok_or(RunStoreError::UnknownRun(record.id))?;
154            return Ok(Written {
155                value: first,
156                replayed: true,
157            });
158        }
159        if let Some(parent) = record.parent
160            && load_run(&transaction, &path, parent)?.is_none()
161        {
162            return Err(RunStoreError::UnknownParent(parent));
163        }
164        transaction
165            .execute(
166                "INSERT INTO runs(run_id, work_id, record_json) VALUES(?1, ?2, ?3)",
167                params![
168                    record.id.to_string(),
169                    record.work.to_string(),
170                    to_json(record)?
171                ],
172            )
173            .map_err(database_error(&path, "record run"))?;
174        let event = LifecycleEvent {
175            id: run_start_event(record.id),
176            run: record.id,
177            state: RunState::Working,
178            detail: None,
179            observed_at: record.observed_started_at,
180            recorded_by: record.recorded_by.clone(),
181            recorded_at: record.started_at,
182        };
183        let sequence = insert_event(&transaction, &path, &event)?;
184        commit(transaction, &path)?;
185        Ok(Written {
186            value: LifecycleEntry { sequence, event },
187            replayed: false,
188        })
189    }
190
191    /// Record a lifecycle transition as `writer`, at DPM's own time `now`.
192    ///
193    /// The transition identity is the idempotency key. The state machine is checked inside the
194    /// write transaction against the run's latest recorded state, so two racing reporters cannot
195    /// both end a run.
196    pub fn transition_blocking(
197        &mut self,
198        transition: &RunTransition,
199        writer: &ActorId,
200        now: DateTime<Utc>,
201        expected: LineageId,
202    ) -> Result<Written<LifecycleEntry>, RunStoreError> {
203        transition.validate().map_err(RunError::from)?;
204        let path = self.path.clone();
205        let transaction = begin(&mut self.connection, &path, expected)?;
206        if let Some(recorded) = event_by_id(&transaction, &path, transition.id)? {
207            current_run(&transaction, &path, recorded.event.run, expected)?;
208            let event = &recorded.event;
209            let same = event.run == transition.run
210                && event.state == transition.to
211                && event.detail == transition.detail
212                && event.observed_at == transition.observed_at
213                && event.recorded_by == *writer;
214            return if same {
215                Ok(Written {
216                    value: recorded,
217                    replayed: true,
218                })
219            } else {
220                Err(RunStoreError::DuplicateEvent {
221                    recorded: Box::new(recorded),
222                })
223            };
224        }
225        let record = writable_run(
226            &transaction,
227            &path,
228            transition.run,
229            expected,
230            writer,
231            "report on",
232        )?;
233        let current = last_event(&transaction, &path, record.id)?
234            .ok_or(RunStoreError::UnknownRun(record.id))?;
235        check_run_transition(record.id, current.event.state, transition.to)?;
236        let event = LifecycleEvent {
237            id: transition.id,
238            run: record.id,
239            state: transition.to,
240            detail: transition.detail.clone(),
241            observed_at: transition.observed_at,
242            recorded_by: writer.clone(),
243            recorded_at: now,
244        };
245        let sequence = insert_event(&transaction, &path, &event)?;
246        commit(transaction, &path)?;
247        Ok(Written {
248            value: LifecycleEntry { sequence, event },
249            replayed: false,
250        })
251    }
252
253    /// Record a batch of activity as `writer`, at DPM's own time `now`, and apply retention.
254    ///
255    /// All of the batch is recorded or none of it. A record is identified by its run and
256    /// `source_sequence`. One still held with the same content, including the whole of its text,
257    /// is a duplicate delivery and is answered with the recorded record; with other content it is
258    /// refused. A sequence at or below the run's high-water mark that is no longer held is refused
259    /// as expired, so a retry that outlived retention can never become a new receipt. Activity
260    /// changes no lifecycle state, whatever it reports.
261    pub fn append_activity_blocking(
262        &mut self,
263        inputs: &[ActivityInput],
264        writer: &ActorId,
265        now: DateTime<Utc>,
266        expected: LineageId,
267    ) -> Result<Vec<Written<ActivityEntry>>, RunStoreError> {
268        if inputs.is_empty() || inputs.len() > MAX_ACTIVITY_BATCH {
269            return Err(invalid_batch(format!(
270                "a batch carries 1..={MAX_ACTIVITY_BATCH} records, not {}",
271                inputs.len()
272            )));
273        }
274        for input in inputs {
275            input.validate().map_err(RunError::from)?;
276        }
277        let path = self.path.clone();
278        let limit = self.activity_limit;
279        let transaction = begin(&mut self.connection, &path, expected)?;
280        let mut authorized = BTreeSet::new();
281        let mut written = Vec::with_capacity(inputs.len());
282        for input in inputs {
283            if !authorized.contains(&input.run) {
284                writable_run(
285                    &transaction,
286                    &path,
287                    input.run,
288                    expected,
289                    writer,
290                    "record activity for",
291                )?;
292                authorized.insert(input.run);
293            }
294            written.push(super::activity::append_one(
295                &transaction,
296                &path,
297                input,
298                writer,
299                now,
300            )?);
301        }
302        super::activity::prune(&transaction, &path, limit)?;
303        commit(transaction, &path)?;
304        Ok(written)
305    }
306
307    /// Record that a committed project operation was performed by a run, as `writer`.
308    ///
309    /// `operation` carries what the project store recorded about the operation; the engine checks
310    /// it inside the write transaction against the run, so a link can never attribute someone
311    /// else's operation, another task's, or one outside the run's lifetime. An operation links to
312    /// at most one run; repeating the same link answers with the recorded one.
313    pub fn link_blocking(
314        &mut self,
315        run: RunId,
316        operation: &OperationFacts,
317        writer: &ActorId,
318        now: DateTime<Utc>,
319        expected: LineageId,
320    ) -> Result<Written<RunLink>, RunStoreError> {
321        let path = self.path.clone();
322        let transaction = begin(&mut self.connection, &path, expected)?;
323        let record = writable_run(
324            &transaction,
325            &path,
326            run,
327            expected,
328            writer,
329            "link operations to",
330        )?;
331        if let Some(linked) = link_by_operation(&transaction, &path, operation.id)? {
332            return if linked.run == run {
333                Ok(Written {
334                    value: linked,
335                    replayed: true,
336                })
337            } else {
338                Err(RunStoreError::LinkConflict {
339                    operation: operation.id,
340                    linked: linked.run,
341                })
342            };
343        }
344        let current =
345            last_event(&transaction, &path, run)?.ok_or(RunStoreError::UnknownRun(run))?;
346        let ended = current
347            .event
348            .state
349            .is_terminal()
350            .then_some(current.event.recorded_at);
351        check_link(&record, ended, operation)?;
352        let link = RunLink {
353            run,
354            operation: operation.id,
355            work: record.work,
356            linked_by: writer.clone(),
357            linked_at: now,
358        };
359        transaction
360            .execute(
361                "INSERT INTO run_links(operation_id, run_id, work_id, linked_by_json, linked_at) \
362                 VALUES(?1, ?2, ?3, ?4, ?5)",
363                params![
364                    link.operation.to_string(),
365                    link.run.to_string(),
366                    link.work.to_string(),
367                    to_json(&link.linked_by)?,
368                    time_text(link.linked_at),
369                ],
370            )
371            .map_err(database_error(&path, "record run link"))?;
372        commit(transaction, &path)?;
373        Ok(Written {
374            value: link,
375            replayed: false,
376        })
377    }
378}
dpm_store/sqlite/runs/write.rs

dpm_store/sqlite/runs/
write.rs
