liminal_server/services/
plan.rs

1//! The bound builder's durable plan and retained change stream.
2
3use super::site::{SiteError, SiteLimits};
4use super::{Lifecycle, ServiceRecord};
5use serde::{Deserialize, Serialize};
6
7/// A step's closed status vocabulary.
8#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
9#[serde(rename_all = "snake_case")]
10pub enum StepStatus {
11    Todo,
12    Doing,
13    Done,
14    Blocked,
15}
16
17/// One ordinal step; its note and title never enter diagnostics.
18#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
19#[serde(deny_unknown_fields)]
20pub struct Step {
21    /// Stable one-based ordinal, preserved for finished steps.
22    pub id: usize,
23    /// The builder's words for the work.
24    pub title: String,
25    /// Its current state.
26    pub status: StepStatus,
27    /// Optional explanation carried as data only.
28    #[serde(skip_serializing_if = "Option::is_none")]
29    pub note: Option<String>,
30    /// Epoch milliseconds of this step's last change.
31    pub at_ms: u64,
32}
33
34/// Every write retains the complete resulting step list with its actor.
35#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
36#[serde(deny_unknown_fields)]
37pub struct PlanTransition {
38    /// The admitted builder.
39    pub holder: String,
40    /// Epoch milliseconds of this write.
41    pub at_ms: u64,
42    /// The new revision.
43    pub revision: u64,
44    /// Complete steps at that revision.
45    pub steps: Vec<Step>,
46}
47
48/// Current state and complete history, restored with the service.
49#[derive(Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
50#[serde(deny_unknown_fields)]
51pub struct Plan {
52    /// Steps in ordinal order.
53    pub steps: Vec<Step>,
54    /// The sole doing step's id, or none.
55    pub current: Option<usize>,
56    /// Increases once for each accepted write.
57    pub revision: u64,
58    /// Every accepted write, oldest first.
59    #[serde(default)]
60    pub transitions: Vec<PlanTransition>,
61}
62
63/// Watchers receive the current plan; full history stays in the owning record.
64#[must_use]
65pub fn current_value(plan: &Plan) -> serde_json::Value {
66    serde_json::json!({"steps":plan.steps,"current":plan.current,"revision":plan.revision})
67}
68
69/// A site completes with its plan finished: the plan is kept on the record after completion and
70/// can no longer be written, so a step left todo, doing or blocked would stand unfinished for good.
71///
72/// # Errors
73/// Names every step that is not done.
74pub(crate) fn refuse_unfinished(plan: &Plan) -> Result<(), SiteError> {
75    let open: Vec<String> = plan
76        .steps
77        .iter()
78        .filter(|step| step.status != StepStatus::Done)
79        .map(|step| step.id.to_string())
80        .collect();
81    if open.is_empty() {
82        Ok(())
83    } else {
84        Err(SiteError::Refused(format!(
85            "plan steps not done: {}; mark them done before completing",
86            open.join(", ")
87        )))
88    }
89}
90
91fn text(value: &str, limits: &SiteLimits) -> Result<(), SiteError> {
92    if value.is_empty() || value.len() > limits.text_bytes.get() || value.contains('\0') {
93        Err(SiteError::Invalid("plan text"))
94    } else {
95        Ok(())
96    }
97}
98
99impl Plan {
100    /// Validates the complete stored plan before any retained value is served.
101    ///
102    /// # Errors
103    /// A configured bound, invalid ordinal, conflicting current or broken history.
104    pub fn validate(&self, limits: &SiteLimits) -> Result<(), SiteError> {
105        if self.transitions.len() > limits.transitions.get() {
106            return Err(SiteError::Bound("plan history"));
107        }
108        check_steps(&self.steps, limits)?;
109        let current = self
110            .steps
111            .iter()
112            .find(|step| step.status == StepStatus::Doing)
113            .map(|step| step.id);
114        if current != self.current
115            || self.revision
116                != u64::try_from(self.transitions.len()).map_err(|_| SiteError::Store)?
117        {
118            return Err(SiteError::Invalid("stored plan revision or current"));
119        }
120        let mut previous: &[Step] = &[];
121        for (index, transition) in self.transitions.iter().enumerate() {
122            if transition.revision != u64::try_from(index + 1).map_err(|_| SiteError::Store)?
123                || transition.holder.is_empty()
124                || transition.holder.len() > limits.text_bytes.get()
125                || transition.holder.chars().any(char::is_control)
126            {
127                return Err(SiteError::Invalid("stored plan transition"));
128            }
129            check_steps(&transition.steps, limits)?;
130            for done in previous
131                .iter()
132                .filter(|step| step.status == StepStatus::Done)
133            {
134                if transition
135                    .steps
136                    .get(done.id - 1)
137                    .is_none_or(|next| next.title != done.title || next.status != StepStatus::Done)
138                {
139                    return Err(SiteError::Invalid("stored finished step changed"));
140                }
141            }
142            previous = &transition.steps;
143        }
144        if previous != self.steps.as_slice() {
145            return Err(SiteError::Invalid(
146                "stored plan history differs from current",
147            ));
148        }
149        Ok(())
150    }
151    /// Replaces the list without removing or renaming any finished ordinal.
152    ///
153    /// # Errors
154    /// A configured bound, invalid title or finished step change.
155    pub fn set(
156        &mut self,
157        titles: &[String],
158        holder: &str,
159        at_ms: u64,
160        limits: &SiteLimits,
161    ) -> Result<(), SiteError> {
162        if titles.is_empty() || titles.len() > limits.steps.get() {
163            return Err(SiteError::Bound("steps"));
164        }
165        for title in titles {
166            text(title, limits)?;
167        }
168        for old in &self.steps {
169            if old.status == StepStatus::Done && titles.get(old.id - 1) != Some(&old.title) {
170                return Err(SiteError::Refused(format!(
171                    "done step {} cannot be removed or renamed",
172                    old.id
173                )));
174            }
175        }
176        let steps = titles
177            .iter()
178            .enumerate()
179            .map(|(ordinal, title)| {
180                let id = ordinal + 1;
181                self.steps
182                    .get(ordinal)
183                    .filter(|old| old.title == *title)
184                    .cloned()
185                    .unwrap_or_else(|| Step {
186                        id,
187                        title: title.clone(),
188                        status: StepStatus::Todo,
189                        note: None,
190                        at_ms,
191                    })
192            })
193            .collect();
194        self.write(steps, holder, at_ms, limits)
195    }
196
197    /// Changes one step, preserving the single doing step and terminal completion.
198    ///
199    /// # Errors
200    /// Unknown ordinal, invalid note, another doing step or a configured bound.
201    pub fn step(
202        &mut self,
203        id: usize,
204        status: StepStatus,
205        note: Option<String>,
206        holder: &str,
207        at_ms: u64,
208        limits: &SiteLimits,
209    ) -> Result<(), SiteError> {
210        if let Some(note) = &note {
211            text(note, limits)?;
212        }
213        if status == StepStatus::Doing && self.current.is_some_and(|current| current != id) {
214            return Err(SiteError::Refused(format!(
215                "step {} is already doing",
216                self.current.ok_or(SiteError::Store)?
217            )));
218        }
219        let mut steps = self.steps.clone();
220        let step = steps
221            .iter_mut()
222            .find(|step| step.id == id)
223            .ok_or_else(|| SiteError::Refused(format!("plan has no step {id}")))?;
224        if step.status == StepStatus::Done && status != StepStatus::Done {
225            return Err(SiteError::Refused(format!(
226                "done step {id} cannot change status"
227            )));
228        }
229        step.status = status;
230        step.note = note;
231        step.at_ms = at_ms;
232        self.write(steps, holder, at_ms, limits)
233    }
234
235    fn write(
236        &mut self,
237        steps: Vec<Step>,
238        holder: &str,
239        at_ms: u64,
240        limits: &SiteLimits,
241    ) -> Result<(), SiteError> {
242        if self.transitions.len() >= limits.transitions.get() {
243            return Err(SiteError::Bound("transitions"));
244        }
245        let revision = self
246            .revision
247            .checked_add(1)
248            .ok_or(SiteError::Bound("revision"))?;
249        self.current = steps
250            .iter()
251            .find(|step| step.status == StepStatus::Doing)
252            .map(|step| step.id);
253        self.transitions.push(PlanTransition {
254            holder: holder.into(),
255            at_ms,
256            revision,
257            steps: steps.clone(),
258        });
259        self.steps = steps;
260        self.revision = revision;
261        Ok(())
262    }
263}
264
265fn check_steps(steps: &[Step], limits: &SiteLimits) -> Result<(), SiteError> {
266    if steps.len() > limits.steps.get() {
267        return Err(SiteError::Bound("steps"));
268    }
269    let mut doing = false;
270    for (ordinal, step) in steps.iter().enumerate() {
271        if step.id != ordinal + 1 {
272            return Err(SiteError::Invalid("stored plan ordinal"));
273        }
274        text(&step.title, limits)?;
275        if let Some(note) = &step.note {
276            text(note, limits)?;
277        }
278        if step.status == StepStatus::Doing {
279            if doing {
280                return Err(SiteError::Invalid("stored plan has two doing steps"));
281            }
282            doing = true;
283        }
284    }
285    Ok(())
286}
287
288impl Lifecycle {
289    /// Only the signed builder with an outright site plan right can reach writes.
290    ///
291    /// # Errors
292    /// The named holder is not the currently bound builder.
293    pub fn check_plan_holder(&self, record: &ServiceRecord, holder: &str) -> Result<(), SiteError> {
294        if record.builder_bound && record.builder.as_deref() == Some(holder) {
295            return Ok(());
296        }
297        Err(SiteError::Refused(format!(
298            "pass holder {holder} is not the builder of {}",
299            record.name
300        )))
301    }
302
303    /// Writes one whole replacement after the caller's plan grant is decided.
304    ///
305    /// # Errors
306    /// Wrong builder, non-building service, invalid list or durable write failure.
307    pub(crate) fn plan_set(
308        &self,
309        service: &str,
310        holder: &str,
311        titles: &[String],
312        at_ms: u64,
313    ) -> Result<Plan, SiteError> {
314        self.write_plan(service, holder, at_ms, |plan, limits| {
315            plan.set(titles, holder, at_ms, limits)
316        })
317    }
318
319    /// Writes one step after the caller's plan grant is decided.
320    ///
321    /// # Errors
322    /// Wrong builder, non-building service, invalid transition or durable failure.
323    pub(crate) fn plan_step(
324        &self,
325        service: &str,
326        holder: &str,
327        id: usize,
328        status: StepStatus,
329        note: Option<String>,
330        at_ms: u64,
331    ) -> Result<Plan, SiteError> {
332        self.write_plan(service, holder, at_ms, |plan, limits| {
333            plan.step(id, status, note, holder, at_ms, limits)
334        })
335    }
336
337    fn write_plan(
338        &self,
339        service: &str,
340        holder: &str,
341        at_ms: u64,
342        apply: impl FnOnce(&mut Plan, &SiteLimits) -> Result<(), SiteError>,
343    ) -> Result<Plan, SiteError> {
344        let limits = &self.site_bindings()?.limits;
345        let mut registry = self.lock()?;
346        let mut record = registry
347            .records
348            .get(service)
349            .cloned()
350            .ok_or_else(|| SiteError::Refused(format!("unknown_service: {service}")))?;
351        self.check_plan_holder(&record, holder)?;
352        if record.state != super::ServiceState::Building {
353            return Err(SiteError::Invalid("plan requires building"));
354        }
355        if record.transitions.len() >= limits.transitions.get() {
356            return Err(SiteError::Bound("transitions"));
357        }
358        apply(&mut record.plan, limits)?;
359        if serde_json::to_vec(&current_value(&record.plan))
360            .map_err(|_| SiteError::Store)?
361            .len()
362            > limits.request_bytes.get()
363        {
364            return Err(SiteError::Bound("plan retained bytes"));
365        }
366        Self::site_transition(&mut record, holder, at_ms, "plan written");
367        self.keep(&mut registry, &record)?;
368        self.publish_plan(&record)?;
369        drop(registry);
370        Ok(record.plan)
371    }
372}
373
374#[cfg(test)]
375#[path = "plan_tests.rs"]
376mod tests;
liminal_server/services/plan.rs

liminal_server/services/
plan.rs
