sozu_command_lib/
protected_transaction.rs

1//! Two-phase coordination of the protected health-check headers of a cluster
2//! across the main process and every worker.
3//!
4//! The main-process SetProtectedHealthCheck task uses [`Transaction`](crate::protected_transaction::Transaction), while each
5//! worker uses [`WorkerStore`](crate::protected_transaction::WorkerStore) for its staged and committed versions. Their causal
6//! rules are described below.
7//!
8//! Main side, [`Transaction`](crate::protected_transaction::Transaction): it proposes a
9//! [`ProtectedHeaderChange`](crate::secret::ProtectedHeaderChange) in the
10//! [`ProtectedHeaderJournal`](crate::secret::ProtectedHeaderJournal) and walks every targeted worker through
11//! *prepare* (validate and stage), *commit* (apply) and, when anything fails,
12//! *abort* (drop the staged proposal, or undo the applied one). The journal
13//! confirms the proposal only when every worker acknowledged the commit; it
14//! rejects it otherwise, once every worker is known to be coherent again. A
15//! worker that cannot be made coherent is reported, to be restarted: a
16//! restarted worker holds no value, which is coherent and probes nothing.
17//!
18//! A transaction carries a cluster id, a
19//! [`ProtectedHeaderVersion`](crate::secret::ProtectedHeaderVersion), a phase
20//! and per-worker acknowledgement flags. It never carries a value, a
21//! fingerprint key or an encoded order, so a task may persist it.
22//!
23//! Worker side, [`WorkerStore`](crate::protected_transaction::WorkerStore): it resolves only versions it holds, applies
24//! them in version order whatever the order of arrival, and keeps the state
25//! before the applied version until it is released, so that an abort after a
26//! commit restores exactly that state.
27
28use std::collections::{BTreeMap, HashMap};
29
30use serde::{Deserialize, Serialize};
31
32use crate::{
33    proto::command::HealthCheckConfig,
34    secret::{
35        ProtectedHeaderChange, ProtectedHeaderJournal, ProtectedHeaderSet, ProtectedHeaderVersion,
36        Resolution, VersionExhausted,
37    },
38};
39
40/// Identity of a worker within one transaction.
41pub type WorkerId = u32;
42
43/// Where a [`Transaction`] stands.
44#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
45pub enum Phase {
46    /// Every worker is asked to validate and stage the proposal.
47    Preparing,
48    /// Every worker staged it and is asked to apply it.
49    Committing,
50    /// Something failed: every worker is asked to drop or undo the proposal.
51    Aborting,
52}
53
54/// What the owner of a [`Transaction`] sends next.
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub enum Order {
57    Commit,
58    Abort,
59}
60
61/// How a finished [`Transaction`] ended.
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub enum Outcome {
64    /// Every worker acknowledged the commit; the journal confirmed the change.
65    Committed,
66    /// Every worker acknowledged the abort; the journal dropped the change.
67    Aborted,
68    /// The journal dropped the change, but these workers did not acknowledge
69    /// the abort and may still hold the proposed values: restart them.
70    Incoherent(Vec<WorkerId>),
71}
72
73/// What the owner of a [`Transaction`] does after an event.
74#[derive(Debug, Clone, PartialEq, Eq)]
75pub enum Step {
76    /// Nothing changed that needs an action.
77    Wait,
78    /// Send this order to these workers, which are the whole target set.
79    Send(Order, Vec<WorkerId>),
80    /// Reply to the client with this outcome.
81    Done(Outcome),
82}
83
84/// Why a [`Transaction`] could not begin. Nothing was proposed.
85#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
86pub enum BeginError {
87    #[error("a protected health check change needs at least one worker")]
88    NoWorker,
89    #[error(transparent)]
90    Exhausted(#[from] VersionExhausted),
91}
92
93/// One in-flight change of the protected headers of a cluster, as the main
94/// process sees it. Persistable: see the module documentation.
95#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
96pub struct Transaction {
97    cluster_id: String,
98    version: ProtectedHeaderVersion,
99    phase: Phase,
100    /// Every targeted worker, and whether it acknowledged the current phase.
101    workers: BTreeMap<WorkerId, bool>,
102    /// Workers whose abort failed. Only meaningful in [`Phase::Aborting`].
103    unrecoverable: Vec<WorkerId>,
104    finished: bool,
105}
106
107impl Transaction {
108    /// Propose `change` for `cluster_id` and target `workers`. Refused before
109    /// the journal is touched when there is no worker.
110    pub fn begin(
111        journal: &mut ProtectedHeaderJournal,
112        cluster_id: &str,
113        change: ProtectedHeaderChange,
114        workers: impl IntoIterator<Item = WorkerId>,
115    ) -> Result<Self, BeginError> {
116        let workers: BTreeMap<WorkerId, bool> = workers.into_iter().map(|w| (w, false)).collect();
117        if workers.is_empty() {
118            return Err(BeginError::NoWorker);
119        }
120        let version = journal.propose(cluster_id, change)?;
121        Ok(Self {
122            cluster_id: cluster_id.to_owned(),
123            version,
124            phase: Phase::Preparing,
125            workers,
126            unrecoverable: Vec::new(),
127            finished: false,
128        })
129    }
130
131    pub fn cluster_id(&self) -> &str {
132        &self.cluster_id
133    }
134
135    pub fn version(&self) -> ProtectedHeaderVersion {
136        self.version
137    }
138
139    pub fn phase(&self) -> Phase {
140        self.phase
141    }
142
143    pub fn is_finished(&self) -> bool {
144        self.finished
145    }
146
147    /// Every targeted worker, in order: the recipients of the prepare order.
148    pub fn workers(&self) -> Vec<WorkerId> {
149        self.workers.keys().copied().collect()
150    }
151
152    /// `worker` staged `version`.
153    pub fn prepared(&mut self, worker: WorkerId, version: ProtectedHeaderVersion) -> Step {
154        if !self.acknowledge(Phase::Preparing, worker, version) {
155            return Step::Wait;
156        }
157        if self.all_acknowledged() {
158            self.enter(Phase::Committing);
159            return Step::Send(Order::Commit, self.workers());
160        }
161        Step::Wait
162    }
163
164    /// `worker` applied `version`; the last acknowledgement confirms it.
165    pub fn committed(
166        &mut self,
167        journal: &mut ProtectedHeaderJournal,
168        worker: WorkerId,
169        version: ProtectedHeaderVersion,
170    ) -> Step {
171        if !self.acknowledge(Phase::Committing, worker, version) {
172            return Step::Wait;
173        }
174        if self.all_acknowledged() {
175            self.finished = true;
176            // `Ignored` is legitimate: a newer version accepted first already
177            // superseded this one, and the journal keeps the newer state.
178            let _ = journal.accept(&self.cluster_id, self.version);
179            return Step::Done(Outcome::Committed);
180        }
181        Step::Wait
182    }
183
184    /// `worker` refused `version`, or the order to it could not be queued.
185    /// Before the commit is confirmed that starts the abort of every worker; a
186    /// failed abort marks the worker as unrecoverable.
187    pub fn failed(
188        &mut self,
189        journal: &mut ProtectedHeaderJournal,
190        worker: WorkerId,
191        version: ProtectedHeaderVersion,
192    ) -> Step {
193        if self.finished || version != self.version || !self.workers.contains_key(&worker) {
194            return Step::Wait;
195        }
196        match self.phase {
197            Phase::Preparing | Phase::Committing => self.begin_abort(),
198            Phase::Aborting => {
199                if !self.unrecoverable.contains(&worker) {
200                    self.unrecoverable.push(worker);
201                }
202                self.workers.insert(worker, true);
203                self.finish_abort(journal)
204            }
205        }
206    }
207
208    /// `worker` dropped or undid `version`.
209    pub fn aborted(
210        &mut self,
211        journal: &mut ProtectedHeaderJournal,
212        worker: WorkerId,
213        version: ProtectedHeaderVersion,
214    ) -> Step {
215        if !self.acknowledge(Phase::Aborting, worker, version) {
216            return Step::Wait;
217        }
218        self.finish_abort(journal)
219    }
220
221    /// The deadline passed. A transaction not yet aborting starts to abort;
222    /// one already aborting gives up on the workers that did not answer.
223    pub fn timed_out(&mut self, journal: &mut ProtectedHeaderJournal) -> Step {
224        if self.finished {
225            return Step::Wait;
226        }
227        match self.phase {
228            Phase::Preparing | Phase::Committing => self.begin_abort(),
229            Phase::Aborting => {
230                let silent: Vec<WorkerId> = self
231                    .workers
232                    .iter()
233                    .filter(|(_, acknowledged)| !**acknowledged)
234                    .map(|(worker, _)| *worker)
235                    .collect();
236                for worker in silent {
237                    if !self.unrecoverable.contains(&worker) {
238                        self.unrecoverable.push(worker);
239                    }
240                    self.workers.insert(worker, true);
241                }
242                self.finish_abort(journal)
243            }
244        }
245    }
246
247    /// Record the acknowledgement of `worker` for `phase`; `false` when it is
248    /// stale, foreign, out of phase or a repeat, in which case nothing changed.
249    fn acknowledge(
250        &mut self,
251        phase: Phase,
252        worker: WorkerId,
253        version: ProtectedHeaderVersion,
254    ) -> bool {
255        if self.finished || self.phase != phase || version != self.version {
256            return false;
257        }
258        match self.workers.get_mut(&worker) {
259            Some(acknowledged) if !*acknowledged => {
260                *acknowledged = true;
261                true
262            }
263            _ => false,
264        }
265    }
266
267    fn all_acknowledged(&self) -> bool {
268        self.workers.values().all(|acknowledged| *acknowledged)
269    }
270
271    fn enter(&mut self, phase: Phase) {
272        self.phase = phase;
273        for acknowledged in self.workers.values_mut() {
274            *acknowledged = false;
275        }
276    }
277
278    /// Ask every worker to drop or undo the proposal: one that failed to
279    /// prepare holds nothing, and a repeated abort is harmless.
280    fn begin_abort(&mut self) -> Step {
281        self.enter(Phase::Aborting);
282        Step::Send(Order::Abort, self.workers())
283    }
284
285    fn finish_abort(&mut self, journal: &mut ProtectedHeaderJournal) -> Step {
286        if !self.all_acknowledged() {
287            return Step::Wait;
288        }
289        self.finished = true;
290        journal.reject(&self.cluster_id, self.version);
291        if self.unrecoverable.is_empty() {
292            Step::Done(Outcome::Aborted)
293        } else {
294            Step::Done(Outcome::Incoherent(self.unrecoverable.to_owned()))
295        }
296    }
297}
298
299/// The configuration of a cluster together with the values its headers name.
300#[derive(Clone)]
301pub struct Payload {
302    pub config: HealthCheckConfig,
303    pub headers: ProtectedHeaderSet,
304}
305
306/// What a worker holds at one version; `None` is an absent cluster.
307#[derive(Clone, Default)]
308struct Held {
309    version: Option<ProtectedHeaderVersion>,
310    payload: Option<Payload>,
311}
312
313#[derive(Default)]
314struct ClusterVersions {
315    applied: Held,
316    /// The state before `applied`, restored by an abort of `applied`.
317    previous: Held,
318    staged: BTreeMap<ProtectedHeaderVersion, Option<Payload>>,
319}
320
321/// Why a worker refused to stage a version.
322#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
323pub enum StageError {
324    #[error("protected health check version is older than the one already applied")]
325    Stale,
326}
327
328/// What a commit did on a worker.
329#[derive(Debug, Clone, Copy, PartialEq, Eq)]
330pub enum Commit {
331    /// The version is now what the cluster holds.
332    Applied,
333    /// A newer version is already applied: nothing is overwritten.
334    Superseded,
335    /// The worker never staged this version, or already resolved it.
336    Unknown,
337}
338
339/// The versioned protected headers one worker holds. No serde, no `Clone`,
340/// and `Debug` prints counts only.
341#[derive(Default)]
342pub struct WorkerStore {
343    clusters: HashMap<String, ClusterVersions>,
344}
345
346impl WorkerStore {
347    /// Stage `payload` as `version` of `cluster`; `None` stages a removal.
348    /// Nothing is applied. A version at or below the applied one is refused,
349    /// and staging the same version again keeps the first payload.
350    pub fn prepare(
351        &mut self,
352        cluster: &str,
353        version: ProtectedHeaderVersion,
354        payload: Option<Payload>,
355    ) -> Result<(), StageError> {
356        let versions = self.clusters.entry(cluster.to_owned()).or_default();
357        if versions
358            .applied
359            .version
360            .is_some_and(|applied| version <= applied)
361        {
362            return Err(StageError::Stale);
363        }
364        versions.staged.entry(version).or_insert(payload);
365        Ok(())
366    }
367
368    /// Apply a staged `version`. Versions apply in order: one older than the
369    /// applied version overwrites nothing, but it becomes the state an abort
370    /// of the applied version restores when it is newer than the one kept.
371    pub fn commit(&mut self, cluster: &str, version: ProtectedHeaderVersion) -> Commit {
372        let Some(versions) = self.clusters.get_mut(cluster) else {
373            return Commit::Unknown;
374        };
375        let Some(payload) = versions.staged.remove(&version) else {
376            return Commit::Unknown;
377        };
378        let held = Held {
379            version: Some(version),
380            payload,
381        };
382        if versions
383            .applied
384            .version
385            .is_none_or(|applied| version > applied)
386        {
387            versions.previous = std::mem::replace(&mut versions.applied, held);
388            Commit::Applied
389        } else {
390            if versions.previous.version.is_none_or(|kept| version > kept) {
391                versions.previous = held;
392            }
393            Commit::Superseded
394        }
395    }
396
397    /// Drop a staged `version`; when it is the applied one, restore the state
398    /// that preceded it. Anything else is `Ignored`.
399    pub fn abort(&mut self, cluster: &str, version: ProtectedHeaderVersion) -> Resolution {
400        let Some(versions) = self.clusters.get_mut(cluster) else {
401            return Resolution::Ignored;
402        };
403        let mut changed = versions.staged.remove(&version).is_some();
404        if versions.applied.version == Some(version) {
405            versions.applied = std::mem::take(&mut versions.previous);
406            changed = true;
407        }
408        self.forget_if_empty(cluster);
409        if changed {
410            Resolution::Applied
411        } else {
412            Resolution::Ignored
413        }
414    }
415
416    /// The transaction that applied `version` succeeded everywhere: the state
417    /// before it can no longer be wanted, and is dropped.
418    pub fn release(&mut self, cluster: &str, version: ProtectedHeaderVersion) {
419        if let Some(versions) = self.clusters.get_mut(cluster)
420            && versions.applied.version == Some(version)
421        {
422            versions.previous = Held::default();
423        }
424        self.forget_if_empty(cluster);
425    }
426
427    /// The payload the cluster currently holds, if any.
428    pub fn effective(&self, cluster: &str) -> Option<&Payload> {
429        self.clusters.get(cluster)?.applied.payload.as_ref()
430    }
431
432    /// The cluster is gone: every held, kept and staged value goes with it.
433    pub fn remove_cluster(&mut self, cluster: &str) {
434        self.clusters.remove(cluster);
435    }
436
437    fn forget_if_empty(&mut self, cluster: &str) {
438        if self.clusters.get(cluster).is_some_and(|v| {
439            v.applied.version.is_none() && v.previous.version.is_none() && v.staged.is_empty()
440        }) {
441            self.clusters.remove(cluster);
442        }
443    }
444}
445
446impl std::fmt::Debug for WorkerStore {
447    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
448        f.debug_struct("WorkerStore")
449            .field("clusters", &self.clusters.len())
450            .field(
451                "staged",
452                &self
453                    .clusters
454                    .values()
455                    .map(|v| v.staged.len())
456                    .sum::<usize>(),
457            )
458            .finish()
459    }
460}
461
462#[cfg(test)]
463mod tests {
464    use zeroize::Zeroizing;
465
466    use super::*;
467    use crate::{
468        proto::command::HealthCheckHeaderValue,
469        secret::{FingerprintKey, ProtectedHeader, SecretValue},
470    };
471
472    const SENTINEL: &str = "SYNTHETIC-TXN-SECRET-77aa";
473
474    fn set(value: &str) -> ProtectedHeaderSet {
475        let key = FingerprintKey::from_parts(Zeroizing::new([5; 32]), [6; 16]);
476        let mut carrier = HealthCheckHeaderValue {
477            name: "authorization".to_owned(),
478            value: value.as_bytes().to_vec(),
479        };
480        let secret = SecretValue::take_from_ingress(&mut carrier).unwrap();
481        ProtectedHeaderSet::new(vec![ProtectedHeader::new(
482            &key,
483            "authorization".to_owned(),
484            secret,
485        )])
486    }
487
488    fn replace(value: &str) -> ProtectedHeaderChange {
489        ProtectedHeaderChange::Replace(set(value))
490    }
491
492    fn payload(value: &str) -> Payload {
493        Payload {
494            config: HealthCheckConfig::default(),
495            headers: set(value),
496        }
497    }
498
499    fn held_ptr(journal: &ProtectedHeaderJournal, cluster: &str) -> Option<Vec<u8>> {
500        journal
501            .effective(cluster)
502            .and_then(|s| s.headers().first())
503            .map(|h| h.value().expose_secret_bytes().to_vec())
504    }
505
506    /// Drive a transaction through prepare and commit with every worker acking.
507    fn commit_all(txn: &mut Transaction, journal: &mut ProtectedHeaderJournal) -> Step {
508        let version = txn.version();
509        let mut last = Step::Wait;
510        for worker in txn.workers() {
511            txn.prepared(worker, version);
512        }
513        for worker in txn.workers() {
514            last = txn.committed(journal, worker, version);
515        }
516        last
517    }
518
519    #[test]
520    fn begin_without_worker_touches_nothing() {
521        let mut journal = ProtectedHeaderJournal::default();
522        let error = Transaction::begin(&mut journal, "app", replace("a"), []).unwrap_err();
523        assert_eq!(error, BeginError::NoWorker);
524        assert!(journal.effective("app").is_none());
525        // The next valid request progresses and gets the first version.
526        let txn = Transaction::begin(&mut journal, "app", replace("a"), [1]).unwrap();
527        assert_eq!(txn.phase(), Phase::Preparing);
528        assert!(journal.effective("app").is_some());
529    }
530
531    #[test]
532    fn all_workers_commit_confirms_the_journal() {
533        let mut journal = ProtectedHeaderJournal::default();
534        let mut txn = Transaction::begin(&mut journal, "app", replace("a"), [1, 2]).unwrap();
535        let version = txn.version();
536
537        assert_eq!(txn.prepared(1, version), Step::Wait);
538        assert_eq!(
539            txn.prepared(2, version),
540            Step::Send(Order::Commit, vec![1, 2])
541        );
542        assert_eq!(txn.phase(), Phase::Committing);
543        assert_eq!(txn.committed(&mut journal, 1, version), Step::Wait);
544        // Not yet success: one worker has not acknowledged.
545        assert!(!txn.is_finished());
546        assert_eq!(
547            txn.committed(&mut journal, 2, version),
548            Step::Done(Outcome::Committed)
549        );
550        assert!(txn.is_finished());
551        // Confirmed: rejecting it now finds nothing to drop.
552        assert_eq!(journal.reject("app", version), Resolution::Ignored);
553        assert!(journal.effective("app").is_some());
554    }
555
556    #[test]
557    fn zero_acknowledgement_failure_aborts_everyone_and_drops_the_proposal() {
558        let mut journal = ProtectedHeaderJournal::default();
559        let mut txn = Transaction::begin(&mut journal, "app", replace("a"), [1, 2]).unwrap();
560        let version = txn.version();
561
562        assert_eq!(
563            txn.failed(&mut journal, 1, version),
564            Step::Send(Order::Abort, vec![1, 2])
565        );
566        assert_eq!(txn.aborted(&mut journal, 1, version), Step::Wait);
567        assert_eq!(
568            txn.aborted(&mut journal, 2, version),
569            Step::Done(Outcome::Aborted)
570        );
571        assert!(journal.effective("app").is_none(), "no value is kept");
572    }
573
574    #[test]
575    fn one_success_one_failure_never_commits() {
576        let mut journal = ProtectedHeaderJournal::default();
577        let mut txn = Transaction::begin(&mut journal, "app", replace("a"), [1, 2]).unwrap();
578        let version = txn.version();
579        assert_eq!(txn.prepared(1, version), Step::Wait);
580        assert_eq!(
581            txn.failed(&mut journal, 2, version),
582            Step::Send(Order::Abort, vec![1, 2])
583        );
584        // A late prepare from the other worker cannot revive the commit.
585        assert_eq!(txn.prepared(2, version), Step::Wait);
586        assert_eq!(txn.committed(&mut journal, 1, version), Step::Wait);
587        txn.aborted(&mut journal, 1, version);
588        assert_eq!(
589            txn.aborted(&mut journal, 2, version),
590            Step::Done(Outcome::Aborted)
591        );
592        assert!(journal.effective("app").is_none());
593    }
594
595    #[test]
596    fn failure_after_a_partial_commit_aborts_and_restores() {
597        let mut journal = ProtectedHeaderJournal::default();
598        let mut txn = Transaction::begin(&mut journal, "app", replace("a"), [1, 2]).unwrap();
599        let version = txn.version();
600        txn.prepared(1, version);
601        txn.prepared(2, version);
602        assert_eq!(txn.committed(&mut journal, 1, version), Step::Wait);
603        // Timeout (or a failure, or an unqueueable commit) after worker 1 applied.
604        assert_eq!(
605            txn.timed_out(&mut journal),
606            Step::Send(Order::Abort, vec![1, 2])
607        );
608        assert_eq!(txn.phase(), Phase::Aborting);
609        // The commit acknowledgement of worker 2 arriving now is out of phase.
610        assert_eq!(txn.committed(&mut journal, 2, version), Step::Wait);
611        txn.aborted(&mut journal, 1, version);
612        assert_eq!(
613            txn.aborted(&mut journal, 2, version),
614            Step::Done(Outcome::Aborted)
615        );
616        assert!(journal.effective("app").is_none());
617    }
618
619    #[test]
620    fn a_failed_abort_is_reported_not_hidden() {
621        let mut journal = ProtectedHeaderJournal::default();
622        let mut txn = Transaction::begin(&mut journal, "app", replace("a"), [1, 2, 3]).unwrap();
623        let version = txn.version();
624        txn.failed(&mut journal, 1, version);
625        assert_eq!(txn.aborted(&mut journal, 1, version), Step::Wait);
626        // Worker 2's abort fails; worker 3 never answers.
627        assert_eq!(txn.failed(&mut journal, 2, version), Step::Wait);
628        assert_eq!(
629            txn.timed_out(&mut journal),
630            Step::Done(Outcome::Incoherent(vec![2, 3]))
631        );
632        assert!(journal.effective("app").is_none(), "main never commits");
633    }
634
635    #[test]
636    fn stale_foreign_and_duplicate_acknowledgements_change_nothing() {
637        let mut journal = ProtectedHeaderJournal::default();
638        let mut first = Transaction::begin(&mut journal, "app", replace("a"), [1]).unwrap();
639        let old = first.version();
640        assert!(matches!(
641            commit_all(&mut first, &mut journal),
642            Step::Done(_)
643        ));
644
645        let mut second = Transaction::begin(&mut journal, "app", replace("b"), [1, 2]).unwrap();
646        let version = second.version();
647        let before = second.clone();
648        assert_eq!(second.prepared(1, old), Step::Wait, "older version");
649        assert_eq!(second.prepared(9, version), Step::Wait, "foreign worker");
650        assert_eq!(second.committed(&mut journal, 1, version), Step::Wait);
651        assert_eq!(second.aborted(&mut journal, 1, version), Step::Wait);
652        assert_eq!(second.failed(&mut journal, 1, old), Step::Wait);
653        assert_eq!(second, before);
654        assert_eq!(second.prepared(1, version), Step::Wait);
655        assert_eq!(second.prepared(1, version), Step::Wait, "duplicate");
656        assert_eq!(
657            second.prepared(2, version),
658            Step::Send(Order::Commit, vec![1, 2])
659        );
660        // The finished first transaction ignores everything.
661        let done = first.clone();
662        assert_eq!(first.failed(&mut journal, 1, old), Step::Wait);
663        assert_eq!(first.timed_out(&mut journal), Step::Wait);
664        assert_eq!(first, done);
665        assert!(journal.effective("app").is_some());
666    }
667
668    #[test]
669    fn a_then_b_completes_in_either_order_and_the_newest_wins() {
670        for a_first in [true, false] {
671            let mut journal = ProtectedHeaderJournal::default();
672            let mut a = Transaction::begin(&mut journal, "app", replace("a"), [1]).unwrap();
673            let mut b = Transaction::begin(&mut journal, "app", replace("b"), [1]).unwrap();
674            let b_ptr = held_ptr(&journal, "app");
675            if a_first {
676                commit_all(&mut a, &mut journal);
677                commit_all(&mut b, &mut journal);
678            } else {
679                commit_all(&mut b, &mut journal);
680                commit_all(&mut a, &mut journal);
681            }
682            assert!(a.is_finished() && b.is_finished());
683            assert_eq!(held_ptr(&journal, "app"), b_ptr, "B stays effective");
684            // Nothing is left pending: both versions are resolved.
685            assert_eq!(journal.reject("app", a.version()), Resolution::Ignored);
686            assert_eq!(journal.reject("app", b.version()), Resolution::Ignored);
687        }
688    }
689
690    #[test]
691    fn an_aborted_ancestor_leaves_a_live_descendant_alone() {
692        let mut journal = ProtectedHeaderJournal::default();
693        let mut a = Transaction::begin(&mut journal, "app", replace("a"), [1]).unwrap();
694        let mut b = Transaction::begin(&mut journal, "app", replace("b"), [1]).unwrap();
695        let b_ptr = held_ptr(&journal, "app");
696        let version = a.version();
697        a.failed(&mut journal, 1, version);
698        assert_eq!(
699            a.aborted(&mut journal, 1, version),
700            Step::Done(Outcome::Aborted)
701        );
702        assert_eq!(held_ptr(&journal, "app"), b_ptr);
703        assert!(matches!(
704            commit_all(&mut b, &mut journal),
705            Step::Done(Outcome::Committed)
706        ));
707    }
708
709    #[test]
710    fn a_persisted_transaction_carries_no_value() {
711        let mut journal = ProtectedHeaderJournal::default();
712        let txn = Transaction::begin(&mut journal, "app", replace(SENTINEL), [1, 2]).unwrap();
713        let json = serde_json::to_string(&txn).unwrap();
714        assert!(!json.contains(SENTINEL), "{json}");
715        assert!(!json.contains("authorization"), "{json}");
716        assert!(!format!("{txn:?}").contains(SENTINEL));
717        assert!(!format!("{journal:?}").contains(SENTINEL));
718        let back: Transaction = serde_json::from_str(&json).unwrap();
719        assert_eq!(back, txn);
720    }
721
722    fn versions(count: usize) -> Vec<ProtectedHeaderVersion> {
723        let mut journal = ProtectedHeaderJournal::default();
724        (0..count)
725            .map(|_| journal.propose("v", ProtectedHeaderChange::Clear).unwrap())
726            .collect()
727    }
728
729    fn ptr(store: &WorkerStore, cluster: &str) -> Option<Vec<u8>> {
730        store
731            .effective(cluster)
732            .and_then(|p| p.headers.headers().first())
733            .map(|h| h.value().expose_secret_bytes().to_vec())
734    }
735
736    #[test]
737    fn worker_stages_without_applying_and_commit_applies() {
738        let v = versions(1)[0];
739        let mut store = WorkerStore::default();
740        store.prepare("app", v, Some(payload("a"))).unwrap();
741        assert!(store.effective("app").is_none(), "staged is not applied");
742        assert_eq!(store.commit("app", v), Commit::Applied);
743        assert!(store.effective("app").is_some());
744        assert_eq!(store.commit("app", v), Commit::Unknown, "already resolved");
745    }
746
747    #[test]
748    fn worker_abort_drops_staged_and_restores_the_applied_predecessor() {
749        let v = versions(2);
750        let mut store = WorkerStore::default();
751        store.prepare("app", v[0], Some(payload("a"))).unwrap();
752        store.commit("app", v[0]);
753        let before = ptr(&store, "app");
754
755        store.prepare("app", v[1], Some(payload("b"))).unwrap();
756        assert_eq!(store.abort("app", v[1]), Resolution::Applied);
757        assert_eq!(ptr(&store, "app"), before);
758
759        store.prepare("app", v[1], Some(payload("b"))).unwrap();
760        store.commit("app", v[1]);
761        assert_ne!(ptr(&store, "app"), before);
762        // Undo after the commit restores exactly the previous state.
763        assert_eq!(store.abort("app", v[1]), Resolution::Applied);
764        assert_eq!(ptr(&store, "app"), before);
765        assert_eq!(store.abort("app", v[1]), Resolution::Ignored, "repeat");
766    }
767
768    #[test]
769    fn worker_never_lets_an_older_version_overwrite_a_newer_one() {
770        let v = versions(2);
771        let mut store = WorkerStore::default();
772        store.prepare("app", v[0], Some(payload("a"))).unwrap();
773        store.prepare("app", v[1], Some(payload("b"))).unwrap();
774        assert_eq!(store.commit("app", v[1]), Commit::Applied);
775        let newest = ptr(&store, "app");
776        assert_eq!(store.commit("app", v[0]), Commit::Superseded);
777        assert_eq!(ptr(&store, "app"), newest);
778        assert_eq!(
779            store.prepare("app", v[0], Some(payload("a"))),
780            Err(StageError::Stale)
781        );
782        // B fails afterwards: A, which was superseded, is what comes back.
783        assert_eq!(store.abort("app", v[1]), Resolution::Applied);
784        assert!(store.effective("app").is_some(), "A is restored, not lost");
785        assert_ne!(ptr(&store, "app"), newest);
786    }
787
788    #[test]
789    fn worker_release_forgets_the_predecessor() {
790        let v = versions(2);
791        let mut store = WorkerStore::default();
792        store.prepare("app", v[0], Some(payload("a"))).unwrap();
793        store.commit("app", v[0]);
794        store.prepare("app", v[1], Some(payload("b"))).unwrap();
795        store.commit("app", v[1]);
796        let newest = ptr(&store, "app");
797        store.release("app", v[1]);
798        // A stale abort of another version changes nothing.
799        assert_eq!(store.abort("app", v[0]), Resolution::Ignored);
800        assert_eq!(ptr(&store, "app"), newest);
801        // The predecessor is gone: the owner never aborts after a release,
802        // and if it did the worker would hold nothing rather than old values.
803        assert_eq!(store.abort("app", v[1]), Resolution::Applied);
804        assert!(store.effective("app").is_none());
805    }
806
807    #[test]
808    fn worker_clear_and_removal_erase_future_values() {
809        let v = versions(3);
810        let mut store = WorkerStore::default();
811        store.prepare("app", v[0], Some(payload("a"))).unwrap();
812        store.commit("app", v[0]);
813        let held = store.effective("app").unwrap().headers.clone();
814        assert_eq!(held.len(), 1);
815
816        let cleared = Payload {
817            config: HealthCheckConfig::default(),
818            headers: ProtectedHeaderSet::new(Vec::new()),
819        };
820        store.prepare("app", v[1], Some(cleared)).unwrap();
821        store.commit("app", v[1]);
822        assert!(store.effective("app").unwrap().headers.is_empty());
823        // The snapshot taken before the clear keeps its exact bytes.
824        assert_eq!(
825            held.headers()[0].value().expose_secret_bytes(),
826            b"a".as_slice()
827        );
828
829        store.prepare("app", v[2], None).unwrap();
830        store.commit("app", v[2]);
831        assert!(store.effective("app").is_none());
832        store.remove_cluster("app");
833        assert_eq!(
834            format!("{store:?}"),
835            "WorkerStore { clusters: 0, staged: 0 }"
836        );
837    }
838
839    #[test]
840    fn worker_debug_prints_counts_only() {
841        let v = versions(1)[0];
842        let mut store = WorkerStore::default();
843        store.prepare("app", v, Some(payload(SENTINEL))).unwrap();
844        store.commit("app", v);
845        let shown = format!("{store:?}");
846        assert!(!shown.contains(SENTINEL) && !shown.contains("authorization"));
847    }
848}
sozu_command_lib/protected_transaction.rs

sozu_command_lib/
protected_transaction.rs
