optative/
lib.rs

1use std::collections::HashMap;
2use std::fmt::Debug;
3use std::hash::Hash;
4
5pub mod reconcile;
6pub use reconcile::{Reconcile, ReconcileErrors};
7
8pub struct LifecycleContext {
9    pub display_name: String,
10    pub metadata: serde_json::Map<String, serde_json::Value>,
11}
12
13pub trait Lifecycle {
14    type Key: Hash + Eq + Clone + serde::Serialize + serde::de::DeserializeOwned;
15    type State;
16    type Context;
17    type Output;
18    type Error;
19
20    fn key(&self) -> Self::Key;
21
22    /// Human-readable label for logs and traces. Defaults to empty; override
23    /// when you want lifecycle events to identify themselves by something
24    /// other than `key`.
25    fn display_name(&self) -> String {
26        String::new()
27    }
28
29    fn enter(
30        self,
31        ctx: &mut Self::Context,
32        output: &mut Self::Output,
33    ) -> Result<Self::State, Self::Error>;
34
35    fn reconcile_self(
36        self,
37        state: &mut Self::State,
38        ctx: &mut Self::Context,
39        output: &mut Self::Output,
40    ) -> Result<(), Self::Error>;
41
42    fn exit(
43        state: Self::State,
44        ctx: &mut Self::Context,
45        output: &mut Self::Output,
46    ) -> Result<(), Self::Error>;
47
48    fn enhance_lifecycle_context(&self, _ctx: &mut LifecycleContext) {}
49
50    fn enhance_lifecycle_state_context(_state: &Self::State, _ctx: &mut LifecycleContext) {}
51
52    fn lifecycle_context(&self) -> LifecycleContext {
53        let mut ctx = LifecycleContext {
54            display_name: self.display_name(),
55            metadata: serde_json::Map::new(),
56        };
57        self.enhance_lifecycle_context(&mut ctx);
58        ctx
59    }
60
61    fn lifecycle_state_context(state: &Self::State) -> LifecycleContext {
62        let mut ctx = LifecycleContext {
63            display_name: String::new(),
64            metadata: serde_json::Map::new(),
65        };
66        Self::enhance_lifecycle_state_context(state, &mut ctx);
67        ctx
68    }
69
70    fn wrap_enter(
71        self,
72        ctx: &mut Self::Context,
73        output: &mut Self::Output,
74    ) -> Result<Self::State, Self::Error>
75    where
76        Self: Sized,
77    {
78        self.enter(ctx, output)
79    }
80
81    fn wrap_reconcile(
82        self,
83        state: &mut Self::State,
84        ctx: &mut Self::Context,
85        output: &mut Self::Output,
86    ) -> Result<(), Self::Error>
87    where
88        Self: Sized,
89    {
90        self.reconcile_self(state, ctx, output)
91    }
92
93    fn wrap_exit(
94        state: Self::State,
95        ctx: &mut Self::Context,
96        output: &mut Self::Output,
97    ) -> Result<(), Self::Error> {
98        Self::exit(state, ctx, output)
99    }
100}
101
102pub struct OptativeSet<T: Lifecycle> {
103    store: HashMap<T::Key, T::State>,
104}
105
106impl<T: Lifecycle> Default for OptativeSet<T> {
107    fn default() -> Self {
108        Self {
109            store: HashMap::new(),
110        }
111    }
112}
113
114impl<T: Lifecycle + 'static> OptativeSet<T>
115where
116    T::Error: Debug,
117{
118    pub fn new() -> Self {
119        Self::default()
120    }
121
122    /// Seed the set with `(key, state)` pairs without calling `enter`. For
123    /// items that already exist outside the lifecycle (processes, files,
124    /// windows); the next `reconcile` treats them as present.
125    pub fn with_initial_state(items: impl IntoIterator<Item = (T::Key, T::State)>) -> Self {
126        Self {
127            store: items.into_iter().collect(),
128        }
129    }
130
131    fn dedup_by_key(items: impl IntoIterator<Item = T>) -> HashMap<T::Key, T> {
132        let mut map = HashMap::new();
133        for item in items {
134            map.insert(item.key(), item);
135        }
136        map
137    }
138
139    fn exit_removed(
140        &mut self,
141        new_map: &HashMap<T::Key, T>,
142        ctx: &mut T::Context,
143        output: &mut T::Output,
144        errors: &mut ReconcileErrors<T::Key, T::Error>,
145    ) {
146        let exit_keys: Vec<T::Key> = self
147            .store
148            .keys()
149            .filter(|k| !new_map.contains_key(*k))
150            .cloned()
151            .collect();
152        for key in exit_keys {
153            let state = self.store.remove(&key).unwrap();
154            if let Err(e) = T::wrap_exit(state, ctx, output) {
155                errors.push((key, e));
156            }
157        }
158    }
159
160    fn update_existing(
161        &mut self,
162        new_map: &mut HashMap<T::Key, T>,
163        ctx: &mut T::Context,
164        output: &mut T::Output,
165        errors: &mut ReconcileErrors<T::Key, T::Error>,
166    ) {
167        let update_keys: Vec<T::Key> = new_map
168            .keys()
169            .filter(|k| self.store.contains_key(*k))
170            .cloned()
171            .collect();
172        for key in update_keys {
173            let item = new_map.remove(&key).unwrap();
174            let state = self.store.get_mut(&key).unwrap();
175            if let Err(e) = item.wrap_reconcile(state, ctx, output) {
176                let old_state = self.store.remove(&key).unwrap();
177                if let Err(exit_e) = T::wrap_exit(old_state, ctx, output) {
178                    errors.push((key.clone(), exit_e));
179                }
180                errors.push((key, e));
181            }
182        }
183    }
184
185    fn enter_new(
186        &mut self,
187        mut new_map: HashMap<T::Key, T>,
188        ctx: &mut T::Context,
189        output: &mut T::Output,
190        errors: &mut ReconcileErrors<T::Key, T::Error>,
191    ) {
192        let enter_keys: Vec<T::Key> = new_map
193            .keys()
194            .filter(|k| !self.store.contains_key(*k))
195            .cloned()
196            .collect();
197        for key in enter_keys {
198            let item = new_map.remove(&key).unwrap();
199            match item.wrap_enter(ctx, output) {
200                Ok(state) => {
201                    self.store.insert(key, state);
202                }
203                Err(e) => {
204                    errors.push((key, e));
205                }
206            }
207        }
208    }
209
210    pub fn get(&self, key: &T::Key) -> Option<&T::State> {
211        self.store.get(key)
212    }
213
214    pub fn iter(&self) -> impl Iterator<Item = (&T::Key, &T::State)> {
215        self.store.iter()
216    }
217
218    pub fn iter_mut(&mut self) -> impl Iterator<Item = (&T::Key, &mut T::State)> {
219        self.store.iter_mut()
220    }
221
222    pub fn get_mut(&mut self, key: &T::Key) -> Option<&mut T::State> {
223        self.store.get_mut(key)
224    }
225}
226
227impl<T: Lifecycle + 'static> reconcile::Reconcile<T> for OptativeSet<T>
228where
229    T::Error: Debug,
230{
231    fn reconcile(
232        &mut self,
233        desired: impl IntoIterator<Item = T>,
234        ctx: &mut T::Context,
235        output: &mut T::Output,
236    ) -> ReconcileErrors<T::Key, T::Error> {
237        let mut errors = ReconcileErrors::new();
238        let mut new_map = Self::dedup_by_key(desired);
239        self.exit_removed(&new_map, ctx, output, &mut errors);
240        self.update_existing(&mut new_map, ctx, output, &mut errors);
241        self.enter_new(new_map, ctx, output, &mut errors);
242        errors
243    }
244}
245
246#[cfg(test)]
247mod tests {
248    use super::*;
249    use std::sync::{Arc, Mutex};
250
251    #[derive(Clone)]
252    struct TestSpec {
253        id: String,
254        value: i32,
255    }
256
257    impl std::fmt::Display for TestSpec {
258        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
259            write!(f, "{}", self.id)
260        }
261    }
262
263    impl Lifecycle for TestSpec {
264        type Key = String;
265        type State = i32;
266        type Context = Arc<Mutex<Vec<String>>>;
267        type Output = ();
268        type Error = std::convert::Infallible;
269
270        fn key(&self) -> String {
271            self.id.clone()
272        }
273
274        fn enter(
275            self,
276            ctx: &mut Self::Context,
277            _output: &mut (),
278        ) -> Result<Self::State, Self::Error> {
279            ctx.lock().unwrap().push(format!("enter:{}", self.id));
280            Ok(self.value)
281        }
282
283        fn reconcile_self(
284            self,
285            state: &mut Self::State,
286            ctx: &mut Self::Context,
287            _output: &mut (),
288        ) -> Result<(), Self::Error> {
289            ctx.lock()
290                .unwrap()
291                .push(format!("reconcile_self:{}", self.id));
292            *state = self.value;
293            Ok(())
294        }
295
296        fn exit(
297            state: Self::State,
298            ctx: &mut Self::Context,
299            _output: &mut Self::Output,
300        ) -> Result<(), Self::Error> {
301            ctx.lock().unwrap().push(format!("exit:{}", state));
302            Ok(())
303        }
304    }
305
306    fn make_ctx() -> Arc<Mutex<Vec<String>>> {
307        Arc::new(Mutex::new(Vec::new()))
308    }
309
310    fn calls(ctx: &Arc<Mutex<Vec<String>>>) -> Vec<String> {
311        ctx.lock().unwrap().clone()
312    }
313
314    #[test]
315    fn new_item_calls_enter_and_stores_state() {
316        let mut ctx = make_ctx();
317        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
318        ms.reconcile(
319            vec![TestSpec {
320                id: "a".to_string(),
321                value: 42,
322            }],
323            &mut ctx,
324            &mut (),
325        );
326        assert!(calls(&ctx).contains(&"enter:a".to_string()));
327        assert_eq!(ms.get(&"a".to_string()), Some(&42));
328    }
329
330    #[test]
331    fn removed_item_calls_exit_with_old_state() {
332        let mut ctx = make_ctx();
333        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
334        ms.reconcile(
335            vec![TestSpec {
336                id: "a".to_string(),
337                value: 99,
338            }],
339            &mut ctx,
340            &mut (),
341        );
342        ms.reconcile(vec![], &mut ctx, &mut ());
343        assert!(calls(&ctx).contains(&"exit:99".to_string()));
344    }
345
346    #[test]
347    fn existing_item_calls_reconcile_self_not_enter() {
348        let mut ctx = make_ctx();
349        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
350        ms.reconcile(
351            vec![TestSpec {
352                id: "a".to_string(),
353                value: 1,
354            }],
355            &mut ctx,
356            &mut (),
357        );
358        ms.reconcile(
359            vec![TestSpec {
360                id: "a".to_string(),
361                value: 2,
362            }],
363            &mut ctx,
364            &mut (),
365        );
366        let log = calls(&ctx);
367        assert_eq!(log.iter().filter(|c| *c == "enter:a").count(), 1);
368        assert!(log.contains(&"reconcile_self:a".to_string()));
369    }
370
371    #[test]
372    fn duplicate_keys_in_batch_only_one_enter() {
373        let mut ctx = make_ctx();
374        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
375        ms.reconcile(
376            vec![
377                TestSpec {
378                    id: "a".to_string(),
379                    value: 1,
380                },
381                TestSpec {
382                    id: "a".to_string(),
383                    value: 2,
384                },
385            ],
386            &mut ctx,
387            &mut (),
388        );
389        let log = calls(&ctx);
390        assert_eq!(log.iter().filter(|c| *c == "enter:a").count(), 1);
391    }
392
393    #[test]
394    fn get_returns_state_after_enter() {
395        let mut ctx = make_ctx();
396        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
397        ms.reconcile(
398            vec![TestSpec {
399                id: "b".to_string(),
400                value: 7,
401            }],
402            &mut ctx,
403            &mut (),
404        );
405        assert_eq!(ms.get(&"b".to_string()), Some(&7));
406    }
407
408    #[test]
409    fn get_returns_updated_state_after_reconcile() {
410        let mut ctx = make_ctx();
411        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
412        ms.reconcile(
413            vec![TestSpec {
414                id: "c".to_string(),
415                value: 10,
416            }],
417            &mut ctx,
418            &mut (),
419        );
420        ms.reconcile(
421            vec![TestSpec {
422                id: "c".to_string(),
423                value: 20,
424            }],
425            &mut ctx,
426            &mut (),
427        );
428        assert_eq!(ms.get(&"c".to_string()), Some(&20));
429    }
430
431    #[test]
432    fn iter_mut_yields_mutable_state_visible_via_get() {
433        let mut ctx = make_ctx();
434        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
435        ms.reconcile(
436            vec![TestSpec {
437                id: "d".to_string(),
438                value: 5,
439            }],
440            &mut ctx,
441            &mut (),
442        );
443        for (_k, v) in ms.iter_mut() {
444            *v = 99;
445        }
446        assert_eq!(ms.get(&"d".to_string()), Some(&99));
447    }
448
449    #[test]
450    fn get_mut_returns_mutable_reference_visible_via_get() {
451        let mut ctx = make_ctx();
452        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
453        ms.reconcile(
454            vec![TestSpec {
455                id: "e".to_string(),
456                value: 3,
457            }],
458            &mut ctx,
459            &mut (),
460        );
461        if let Some(v) = ms.get_mut(&"e".to_string()) {
462            *v = 77;
463        }
464        assert_eq!(ms.get(&"e".to_string()), Some(&77));
465    }
466
467    mod enter_err {
468        use super::super::*;
469
470        #[derive(Clone)]
471        struct FallibleSpec {
472            id: String,
473            fail: bool,
474        }
475
476        impl std::fmt::Display for FallibleSpec {
477            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
478                write!(f, "{}", self.id)
479            }
480        }
481
482        #[derive(Debug, PartialEq)]
483        struct FallibleError(String);
484
485        impl Lifecycle for FallibleSpec {
486            type Key = String;
487            type State = String;
488            type Context = ();
489            type Output = ();
490            type Error = FallibleError;
491
492            fn key(&self) -> String {
493                self.id.clone()
494            }
495
496            fn enter(self, _ctx: &mut (), _output: &mut ()) -> Result<String, FallibleError> {
497                if self.fail {
498                    Err(FallibleError(format!("enter failed for {}", self.id)))
499                } else {
500                    Ok(format!("state:{}", self.id))
501                }
502            }
503
504            fn reconcile_self(
505                self,
506                state: &mut String,
507                _ctx: &mut (),
508                _output: &mut (),
509            ) -> Result<(), FallibleError> {
510                *state = format!("updated:{}", self.id);
511                Ok(())
512            }
513
514            fn exit(
515                _state: String,
516                _ctx: &mut (),
517                _output: &mut Self::Output,
518            ) -> Result<(), FallibleError> {
519                Ok(())
520            }
521        }
522
523        #[test]
524        fn enter_err_not_added_to_store_error_returned() {
525            let mut ms: OptativeSet<FallibleSpec> = OptativeSet::new();
526            let errors = ms.reconcile(
527                vec![FallibleSpec {
528                    id: "x".to_string(),
529                    fail: true,
530                }],
531                &mut (),
532                &mut (),
533            );
534            assert!(
535                ms.get(&"x".to_string()).is_none(),
536                "item must not be in store after enter Err"
537            );
538            assert_eq!(errors.len(), 1, "one error must be returned");
539            assert_eq!(errors[0].0, "x");
540            assert_eq!(errors[0].1, FallibleError("enter failed for x".to_string()));
541        }
542
543        #[test]
544        fn enter_ok_adds_item_to_store_no_errors() {
545            let mut ms: OptativeSet<FallibleSpec> = OptativeSet::new();
546            let errors = ms.reconcile(
547                vec![FallibleSpec {
548                    id: "y".to_string(),
549                    fail: false,
550                }],
551                &mut (),
552                &mut (),
553            );
554            assert_eq!(ms.get(&"y".to_string()), Some(&"state:y".to_string()));
555            assert!(errors.is_empty(), "no errors when enter returns Ok");
556        }
557    }
558
559    mod reconcile_err {
560        use super::super::*;
561
562        #[derive(Clone)]
563        struct UpdateFallibleSpec {
564            id: String,
565            fail_update: bool,
566        }
567
568        impl std::fmt::Display for UpdateFallibleSpec {
569            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
570                write!(f, "{}", self.id)
571            }
572        }
573
574        #[derive(Debug, PartialEq)]
575        struct UpdateError(String);
576
577        static EXIT_CALLED: std::sync::atomic::AtomicBool =
578            std::sync::atomic::AtomicBool::new(false);
579
580        impl Lifecycle for UpdateFallibleSpec {
581            type Key = String;
582            type State = String;
583            type Context = ();
584            type Output = ();
585            type Error = UpdateError;
586
587            fn key(&self) -> String {
588                self.id.clone()
589            }
590
591            fn enter(self, _ctx: &mut (), _output: &mut ()) -> Result<String, UpdateError> {
592                Ok(format!("state:{}", self.id))
593            }
594
595            fn reconcile_self(
596                self,
597                _state: &mut String,
598                _ctx: &mut (),
599                _output: &mut (),
600            ) -> Result<(), UpdateError> {
601                if self.fail_update {
602                    Err(UpdateError(format!("update failed for {}", self.id)))
603                } else {
604                    Ok(())
605                }
606            }
607
608            fn exit(
609                _state: String,
610                _ctx: &mut (),
611                _output: &mut Self::Output,
612            ) -> Result<(), UpdateError> {
613                EXIT_CALLED.store(true, std::sync::atomic::Ordering::SeqCst);
614                Ok(())
615            }
616        }
617
618        #[test]
619        fn reconcile_err_exit_called_entry_removed_error_returned() {
620            EXIT_CALLED.store(false, std::sync::atomic::Ordering::SeqCst);
621            let mut ms: OptativeSet<UpdateFallibleSpec> = OptativeSet::new();
622
623            let e1 = ms.reconcile(
624                vec![UpdateFallibleSpec {
625                    id: "z".to_string(),
626                    fail_update: false,
627                }],
628                &mut (),
629                &mut (),
630            );
631            assert!(e1.is_empty());
632            assert!(ms.get(&"z".to_string()).is_some());
633
634            let errors = ms.reconcile(
635                vec![UpdateFallibleSpec {
636                    id: "z".to_string(),
637                    fail_update: true,
638                }],
639                &mut (),
640                &mut (),
641            );
642            assert_eq!(errors.len(), 1);
643            assert_eq!(errors[0].0, "z");
644            assert_eq!(errors[0].1, UpdateError("update failed for z".to_string()));
645            assert!(ms.get(&"z".to_string()).is_none());
646            assert!(EXIT_CALLED.load(std::sync::atomic::Ordering::SeqCst));
647
648            EXIT_CALLED.store(false, std::sync::atomic::Ordering::SeqCst);
649            let e3 = ms.reconcile(
650                vec![UpdateFallibleSpec {
651                    id: "z".to_string(),
652                    fail_update: false,
653                }],
654                &mut (),
655                &mut (),
656            );
657            assert!(e3.is_empty());
658            assert!(ms.get(&"z".to_string()).is_some());
659        }
660    }
661
662    mod channel_output {
663        use super::super::*;
664        use std::sync::mpsc;
665
666        #[derive(Clone)]
667        struct ChannelOutputLifecycle {
668            id: String,
669        }
670
671        impl std::fmt::Display for ChannelOutputLifecycle {
672            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
673                write!(f, "{}", self.id)
674            }
675        }
676
677        impl Lifecycle for ChannelOutputLifecycle {
678            type Key = String;
679            type State = ();
680            type Context = ();
681            type Output = mpsc::Sender<String>;
682            type Error = std::convert::Infallible;
683
684            fn key(&self) -> String {
685                self.id.clone()
686            }
687
688            fn enter(
689                self,
690                _ctx: &mut (),
691                output: &mut mpsc::Sender<String>,
692            ) -> Result<(), Self::Error> {
693                output.send(format!("entered:{}", self.id)).unwrap();
694                Ok(())
695            }
696
697            fn reconcile_self(
698                self,
699                _state: &mut (),
700                _ctx: &mut (),
701                output: &mut mpsc::Sender<String>,
702            ) -> Result<(), Self::Error> {
703                output.send(format!("reconciled:{}", self.id)).unwrap();
704                Ok(())
705            }
706
707            fn exit(
708                _state: (),
709                _ctx: &mut (),
710                _output: &mut Self::Output,
711            ) -> Result<(), Self::Error> {
712                Ok(())
713            }
714        }
715
716        #[test]
717        fn enter_receives_output_and_can_write_to_it() {
718            let (mut tx, rx) = mpsc::channel::<String>();
719            let mut ms: OptativeSet<ChannelOutputLifecycle> = OptativeSet::new();
720            ms.reconcile(
721                vec![ChannelOutputLifecycle {
722                    id: "o1".to_string(),
723                }],
724                &mut (),
725                &mut tx,
726            );
727            drop(tx);
728            let msgs: Vec<String> = rx.try_iter().collect();
729            assert!(msgs.contains(&"entered:o1".to_string()));
730        }
731
732        #[test]
733        fn exit_does_not_receive_output() {
734            let (mut tx, rx) = mpsc::channel::<String>();
735            let mut ms: OptativeSet<ChannelOutputLifecycle> = OptativeSet::new();
736            ms.reconcile(
737                vec![ChannelOutputLifecycle {
738                    id: "o2".to_string(),
739                }],
740                &mut (),
741                &mut tx,
742            );
743            ms.reconcile(vec![], &mut (), &mut tx);
744            drop(tx);
745            let msgs: Vec<String> = rx.try_iter().collect();
746            assert!(!msgs.iter().any(|m| m.starts_with("exited:")));
747        }
748    }
749}
optative/lib.rs

optative/
lib.rs
