onetaskgraph_plugin_api/
held.rs

1//! A write held to the version of the task it writes: `--if-version`.
2//!
3//! The engine holds a write of one task to the version its caller read, and refuses the write
4//! when the task is at any other. Where that check is made is the source's to say, through
5//! [`TaskSource::holds_versions`](crate::TaskSource::holds_versions):
6//!
7//! - A source that serializes its own writes — a folder of Markdown, under the lock beside the
8//!   task's file; an in-memory source, under the mutex over its work — makes the check and the
9//!   write one step, so no competing write can land between them. It answers `true`, and reads
10//!   the version the call is held to with [`HeldVersion::expected`] inside each task write,
11//!   refusing a moved one with [`HeldVersion::moved`].
12//! - A source whose backend offers no conditional write answers `false`, and the engine reads
13//!   the task through it immediately before the write instead. The time between that read and
14//!   the mutation is the backend's residual race, which each such plugin's documentation states.
15//!
16//! The hold is task-local, so a source that never asks writes exactly as it did, and one that
17//! asks outside a held call is answered that nothing is held.
18
19use std::future::Future;
20use std::sync::{Arc, Mutex};
21
22use crate::{NativeId, SourceError};
23
24tokio::task_local! {
25    static HELD: Arc<HeldVersion>;
26}
27
28/// One task's write held to the version its caller read.
29#[derive(Debug)]
30pub struct HeldVersion {
31    id: NativeId,
32    version: String,
33    /// The version the source found instead, once it has refused the write.
34    moved: Mutex<Option<String>>,
35}
36
37impl HeldVersion {
38    /// A write of the task `id` held to `version`.
39    #[must_use]
40    pub fn new(id: NativeId, version: String) -> Self {
41        Self {
42            id,
43            version,
44            moved: Mutex::new(None),
45        }
46    }
47
48    /// Run `call` with this hold answering every [`HeldVersion::expected`] made inside it.
49    pub async fn scope<F: Future>(held: Arc<Self>, call: F) -> F::Output {
50        HELD.scope(held, call).await
51    }
52
53    /// The version the write in progress holds the task `id` to — `None` when no write in
54    /// progress is held, or the one that is holds another task.
55    #[must_use]
56    pub fn expected(id: &NativeId) -> Option<String> {
57        HELD.try_with(|held| (&held.id == id).then(|| held.version.clone()))
58            .ok()
59            .flatten()
60    }
61
62    /// Refuse the write in progress of the task `id`, which the source found at `found` rather
63    /// than at the version it is held to: the refusal the source fails its call with, writing
64    /// nothing, recorded so the engine reports it as the version having moved.
65    #[must_use]
66    pub fn moved(id: &NativeId, found: Option<String>) -> SourceError {
67        let found = found.unwrap_or_default();
68        let _ = HELD.try_with(|held| {
69            if &held.id == id {
70                *held
71                    .moved
72                    .lock()
73                    .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(found.clone());
74            }
75        });
76        SourceError::Refused {
77            message: format!(
78                "task {id} is at version {found:?}, not the version the write is held to, so \
79                 nothing was written"
80            ),
81        }
82    }
83
84    /// The version the source found the task at when it refused the write, if it did.
85    #[must_use]
86    pub fn found(&self) -> Option<String> {
87        self.moved
88            .lock()
89            .unwrap_or_else(std::sync::PoisonError::into_inner)
90            .clone()
91    }
92
93    /// The task this hold is for.
94    #[must_use]
95    pub const fn id(&self) -> &NativeId {
96        &self.id
97    }
98
99    /// The version the write is held to.
100    #[must_use]
101    pub fn version(&self) -> &str {
102        &self.version
103    }
104}
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109
110    #[tokio::test]
111    async fn a_hold_answers_only_inside_its_scope_and_only_for_its_task() {
112        let id = NativeId::from("t1");
113        assert_eq!(HeldVersion::expected(&id), None);
114        let held = Arc::new(HeldVersion::new(id.clone(), "v1".to_owned()));
115        let inside = HeldVersion::scope(Arc::clone(&held), async {
116            let other = HeldVersion::expected(&NativeId::from("t2"));
117            let refused = HeldVersion::moved(&id, Some("v2".to_owned()));
118            (HeldVersion::expected(&id), other, refused)
119        })
120        .await;
121        assert_eq!(inside.0.as_deref(), Some("v1"));
122        assert_eq!(inside.1, None);
123        assert!(matches!(inside.2, SourceError::Refused { .. }));
124        assert_eq!(held.found().as_deref(), Some("v2"));
125    }
126}
onetaskgraph_plugin_api/held.rs

onetaskgraph_plugin_api/
held.rs
