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> 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    /// On a failed `reconcile_self`, this records the error and leaves the item's
161    /// stored state exactly as it was — mirroring `enter_new`/`exit_removed`'s own
162    /// failure handling. It does NOT remove the item or call `exit` on it: that
163    /// "tear down and recover" policy is a choice for the specific `Lifecycle`
164    /// implementor to make inside its own `reconcile_self`, not something this
165    /// generic diff-and-dispatch loop should impose on every consumer. A caller
166    /// that wants recovery-via-exit (e.g. a process supervisor cleaning up a dead
167    /// child after a failed restart) can call its own cleanup logic directly from
168    /// within `reconcile_self`, ordinary function-call style, and still return
169    /// `Err` here purely for reporting.
170    fn update_existing(
171        &mut self,
172        new_map: &mut HashMap<T::Key, T>,
173        ctx: &mut T::Context,
174        output: &mut T::Output,
175        errors: &mut ReconcileErrors<T::Key, T::Error>,
176    ) {
177        let update_keys: Vec<T::Key> = new_map
178            .keys()
179            .filter(|k| self.store.contains_key(*k))
180            .cloned()
181            .collect();
182        for key in update_keys {
183            let item = new_map.remove(&key).unwrap();
184            let state = self.store.get_mut(&key).unwrap();
185            if let Err(e) = item.wrap_reconcile(state, ctx, output) {
186                errors.push((key, e));
187            }
188        }
189    }
190
191    fn enter_new(
192        &mut self,
193        mut new_map: HashMap<T::Key, T>,
194        ctx: &mut T::Context,
195        output: &mut T::Output,
196        errors: &mut ReconcileErrors<T::Key, T::Error>,
197    ) {
198        let enter_keys: Vec<T::Key> = new_map
199            .keys()
200            .filter(|k| !self.store.contains_key(*k))
201            .cloned()
202            .collect();
203        for key in enter_keys {
204            let item = new_map.remove(&key).unwrap();
205            match item.wrap_enter(ctx, output) {
206                Ok(state) => {
207                    self.store.insert(key, state);
208                }
209                Err(e) => {
210                    errors.push((key, e));
211                }
212            }
213        }
214    }
215
216    pub fn get(&self, key: &T::Key) -> Option<&T::State> {
217        self.store.get(key)
218    }
219
220    pub fn iter(&self) -> impl Iterator<Item = (&T::Key, &T::State)> {
221        self.store.iter()
222    }
223
224    pub fn iter_mut(&mut self) -> impl Iterator<Item = (&T::Key, &mut T::State)> {
225        self.store.iter_mut()
226    }
227
228    pub fn get_mut(&mut self, key: &T::Key) -> Option<&mut T::State> {
229        self.store.get_mut(key)
230    }
231}
232
233impl<T: Lifecycle> reconcile::Reconcile<T> for OptativeSet<T>
234where
235    T::Error: Debug,
236{
237    fn reconcile(
238        &mut self,
239        desired: impl IntoIterator<Item = T>,
240        ctx: &mut T::Context,
241        output: &mut T::Output,
242    ) -> ReconcileErrors<T::Key, T::Error> {
243        let mut errors = ReconcileErrors::new();
244        let mut new_map = Self::dedup_by_key(desired);
245        self.exit_removed(&new_map, ctx, output, &mut errors);
246        self.update_existing(&mut new_map, ctx, output, &mut errors);
247        self.enter_new(new_map, ctx, output, &mut errors);
248        errors
249    }
250}
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255    use std::sync::{Arc, Mutex};
256
257    #[derive(Clone)]
258    struct TestSpec {
259        id: String,
260        value: i32,
261    }
262
263    impl std::fmt::Display for TestSpec {
264        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
265            write!(f, "{}", self.id)
266        }
267    }
268
269    impl Lifecycle for TestSpec {
270        type Key = String;
271        type State = i32;
272        type Context = Arc<Mutex<Vec<String>>>;
273        type Output = ();
274        type Error = std::convert::Infallible;
275
276        fn key(&self) -> String {
277            self.id.clone()
278        }
279
280        fn enter(
281            self,
282            ctx: &mut Self::Context,
283            _output: &mut (),
284        ) -> Result<Self::State, Self::Error> {
285            ctx.lock().unwrap().push(format!("enter:{}", self.id));
286            Ok(self.value)
287        }
288
289        fn reconcile_self(
290            self,
291            state: &mut Self::State,
292            ctx: &mut Self::Context,
293            _output: &mut (),
294        ) -> Result<(), Self::Error> {
295            ctx.lock()
296                .unwrap()
297                .push(format!("reconcile_self:{}", self.id));
298            *state = self.value;
299            Ok(())
300        }
301
302        fn exit(
303            state: Self::State,
304            ctx: &mut Self::Context,
305            _output: &mut Self::Output,
306        ) -> Result<(), Self::Error> {
307            ctx.lock().unwrap().push(format!("exit:{}", state));
308            Ok(())
309        }
310    }
311
312    fn make_ctx() -> Arc<Mutex<Vec<String>>> {
313        Arc::new(Mutex::new(Vec::new()))
314    }
315
316    fn calls(ctx: &Arc<Mutex<Vec<String>>>) -> Vec<String> {
317        ctx.lock().unwrap().clone()
318    }
319
320    #[test]
321    fn new_item_calls_enter_and_stores_state() {
322        let mut ctx = make_ctx();
323        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
324        ms.reconcile(
325            vec![TestSpec {
326                id: "a".to_string(),
327                value: 42,
328            }],
329            &mut ctx,
330            &mut (),
331        );
332        assert!(calls(&ctx).contains(&"enter:a".to_string()));
333        assert_eq!(ms.get(&"a".to_string()), Some(&42));
334    }
335
336    #[test]
337    fn removed_item_calls_exit_with_old_state() {
338        let mut ctx = make_ctx();
339        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
340        ms.reconcile(
341            vec![TestSpec {
342                id: "a".to_string(),
343                value: 99,
344            }],
345            &mut ctx,
346            &mut (),
347        );
348        ms.reconcile(vec![], &mut ctx, &mut ());
349        assert!(calls(&ctx).contains(&"exit:99".to_string()));
350    }
351
352    #[test]
353    fn existing_item_calls_reconcile_self_not_enter() {
354        let mut ctx = make_ctx();
355        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
356        ms.reconcile(
357            vec![TestSpec {
358                id: "a".to_string(),
359                value: 1,
360            }],
361            &mut ctx,
362            &mut (),
363        );
364        ms.reconcile(
365            vec![TestSpec {
366                id: "a".to_string(),
367                value: 2,
368            }],
369            &mut ctx,
370            &mut (),
371        );
372        let log = calls(&ctx);
373        assert_eq!(log.iter().filter(|c| *c == "enter:a").count(), 1);
374        assert!(log.contains(&"reconcile_self:a".to_string()));
375    }
376
377    #[test]
378    fn duplicate_keys_in_batch_only_one_enter() {
379        let mut ctx = make_ctx();
380        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
381        ms.reconcile(
382            vec![
383                TestSpec {
384                    id: "a".to_string(),
385                    value: 1,
386                },
387                TestSpec {
388                    id: "a".to_string(),
389                    value: 2,
390                },
391            ],
392            &mut ctx,
393            &mut (),
394        );
395        let log = calls(&ctx);
396        assert_eq!(log.iter().filter(|c| *c == "enter:a").count(), 1);
397    }
398
399    #[test]
400    fn get_returns_state_after_enter() {
401        let mut ctx = make_ctx();
402        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
403        ms.reconcile(
404            vec![TestSpec {
405                id: "b".to_string(),
406                value: 7,
407            }],
408            &mut ctx,
409            &mut (),
410        );
411        assert_eq!(ms.get(&"b".to_string()), Some(&7));
412    }
413
414    #[test]
415    fn get_returns_updated_state_after_reconcile() {
416        let mut ctx = make_ctx();
417        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
418        ms.reconcile(
419            vec![TestSpec {
420                id: "c".to_string(),
421                value: 10,
422            }],
423            &mut ctx,
424            &mut (),
425        );
426        ms.reconcile(
427            vec![TestSpec {
428                id: "c".to_string(),
429                value: 20,
430            }],
431            &mut ctx,
432            &mut (),
433        );
434        assert_eq!(ms.get(&"c".to_string()), Some(&20));
435    }
436
437    #[test]
438    fn iter_mut_yields_mutable_state_visible_via_get() {
439        let mut ctx = make_ctx();
440        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
441        ms.reconcile(
442            vec![TestSpec {
443                id: "d".to_string(),
444                value: 5,
445            }],
446            &mut ctx,
447            &mut (),
448        );
449        for (_k, v) in ms.iter_mut() {
450            *v = 99;
451        }
452        assert_eq!(ms.get(&"d".to_string()), Some(&99));
453    }
454
455    #[test]
456    fn get_mut_returns_mutable_reference_visible_via_get() {
457        let mut ctx = make_ctx();
458        let mut ms: OptativeSet<TestSpec> = OptativeSet::new();
459        ms.reconcile(
460            vec![TestSpec {
461                id: "e".to_string(),
462                value: 3,
463            }],
464            &mut ctx,
465            &mut (),
466        );
467        if let Some(v) = ms.get_mut(&"e".to_string()) {
468            *v = 77;
469        }
470        assert_eq!(ms.get(&"e".to_string()), Some(&77));
471    }
472
473    mod enter_err {
474        use super::super::*;
475
476        #[derive(Clone)]
477        struct FallibleSpec {
478            id: String,
479            fail: bool,
480        }
481
482        impl std::fmt::Display for FallibleSpec {
483            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
484                write!(f, "{}", self.id)
485            }
486        }
487
488        #[derive(Debug, PartialEq)]
489        struct FallibleError(String);
490
491        impl Lifecycle for FallibleSpec {
492            type Key = String;
493            type State = String;
494            type Context = ();
495            type Output = ();
496            type Error = FallibleError;
497
498            fn key(&self) -> String {
499                self.id.clone()
500            }
501
502            fn enter(self, _ctx: &mut (), _output: &mut ()) -> Result<String, FallibleError> {
503                if self.fail {
504                    Err(FallibleError(format!("enter failed for {}", self.id)))
505                } else {
506                    Ok(format!("state:{}", self.id))
507                }
508            }
509
510            fn reconcile_self(
511                self,
512                state: &mut String,
513                _ctx: &mut (),
514                _output: &mut (),
515            ) -> Result<(), FallibleError> {
516                *state = format!("updated:{}", self.id);
517                Ok(())
518            }
519
520            fn exit(
521                _state: String,
522                _ctx: &mut (),
523                _output: &mut Self::Output,
524            ) -> Result<(), FallibleError> {
525                Ok(())
526            }
527        }
528
529        #[test]
530        fn enter_err_not_added_to_store_error_returned() {
531            let mut ms: OptativeSet<FallibleSpec> = OptativeSet::new();
532            let errors = ms.reconcile(
533                vec![FallibleSpec {
534                    id: "x".to_string(),
535                    fail: true,
536                }],
537                &mut (),
538                &mut (),
539            );
540            assert!(
541                ms.get(&"x".to_string()).is_none(),
542                "item must not be in store after enter Err"
543            );
544            assert_eq!(errors.len(), 1, "one error must be returned");
545            assert_eq!(errors[0].0, "x");
546            assert_eq!(errors[0].1, FallibleError("enter failed for x".to_string()));
547        }
548
549        #[test]
550        fn enter_ok_adds_item_to_store_no_errors() {
551            let mut ms: OptativeSet<FallibleSpec> = OptativeSet::new();
552            let errors = ms.reconcile(
553                vec![FallibleSpec {
554                    id: "y".to_string(),
555                    fail: false,
556                }],
557                &mut (),
558                &mut (),
559            );
560            assert_eq!(ms.get(&"y".to_string()), Some(&"state:y".to_string()));
561            assert!(errors.is_empty(), "no errors when enter returns Ok");
562        }
563    }
564
565    mod reconcile_err {
566        use super::super::*;
567
568        #[derive(Clone)]
569        struct UpdateFallibleSpec {
570            id: String,
571            fail_update: bool,
572        }
573
574        impl std::fmt::Display for UpdateFallibleSpec {
575            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
576                write!(f, "{}", self.id)
577            }
578        }
579
580        #[derive(Debug, PartialEq)]
581        struct UpdateError(String);
582
583        static EXIT_CALLED: std::sync::atomic::AtomicBool =
584            std::sync::atomic::AtomicBool::new(false);
585
586        impl Lifecycle for UpdateFallibleSpec {
587            type Key = String;
588            type State = String;
589            type Context = ();
590            type Output = ();
591            type Error = UpdateError;
592
593            fn key(&self) -> String {
594                self.id.clone()
595            }
596
597            fn enter(self, _ctx: &mut (), _output: &mut ()) -> Result<String, UpdateError> {
598                Ok(format!("state:{}", self.id))
599            }
600
601            fn reconcile_self(
602                self,
603                _state: &mut String,
604                _ctx: &mut (),
605                _output: &mut (),
606            ) -> Result<(), UpdateError> {
607                if self.fail_update {
608                    Err(UpdateError(format!("update failed for {}", self.id)))
609                } else {
610                    Ok(())
611                }
612            }
613
614            fn exit(
615                _state: String,
616                _ctx: &mut (),
617                _output: &mut Self::Output,
618            ) -> Result<(), UpdateError> {
619                EXIT_CALLED.store(true, std::sync::atomic::Ordering::SeqCst);
620                Ok(())
621            }
622        }
623
624        #[test]
625        fn reconcile_err_records_error_without_removing_or_calling_exit() {
626            EXIT_CALLED.store(false, std::sync::atomic::Ordering::SeqCst);
627            let mut ms: OptativeSet<UpdateFallibleSpec> = OptativeSet::new();
628
629            let e1 = ms.reconcile(
630                vec![UpdateFallibleSpec {
631                    id: "z".to_string(),
632                    fail_update: false,
633                }],
634                &mut (),
635                &mut (),
636            );
637            assert!(e1.is_empty());
638            assert!(ms.get(&"z".to_string()).is_some());
639
640            // A failed reconcile_self is reported like any other lifecycle error
641            // (same as a failed enter or exit elsewhere) — it does NOT remove the
642            // item from the store or call exit() on it. That symmetry (enter/
643            // reconcile_self/exit failures all "just record it, state untouched")
644            // is the point of this test: recovery-via-exit is a choice for a
645            // specific Lifecycle impl to make inside its own reconcile_self, not
646            // something OptativeSet imposes on every consumer.
647            let errors = ms.reconcile(
648                vec![UpdateFallibleSpec {
649                    id: "z".to_string(),
650                    fail_update: true,
651                }],
652                &mut (),
653                &mut (),
654            );
655            assert_eq!(errors.len(), 1);
656            assert_eq!(errors[0].0, "z");
657            assert_eq!(errors[0].1, UpdateError("update failed for z".to_string()));
658            assert!(
659                ms.get(&"z".to_string()).is_some(),
660                "a failed update must not remove the item from the store"
661            );
662            assert!(
663                !EXIT_CALLED.load(std::sync::atomic::Ordering::SeqCst),
664                "a failed update must not trigger exit()"
665            );
666
667            let e3 = ms.reconcile(
668                vec![UpdateFallibleSpec {
669                    id: "z".to_string(),
670                    fail_update: false,
671                }],
672                &mut (),
673                &mut (),
674            );
675            assert!(e3.is_empty());
676            assert!(ms.get(&"z".to_string()).is_some());
677        }
678    }
679
680    mod channel_output {
681        use super::super::*;
682        use std::sync::mpsc;
683
684        #[derive(Clone)]
685        struct ChannelOutputLifecycle {
686            id: String,
687        }
688
689        impl std::fmt::Display for ChannelOutputLifecycle {
690            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
691                write!(f, "{}", self.id)
692            }
693        }
694
695        impl Lifecycle for ChannelOutputLifecycle {
696            type Key = String;
697            type State = ();
698            type Context = ();
699            type Output = mpsc::Sender<String>;
700            type Error = std::convert::Infallible;
701
702            fn key(&self) -> String {
703                self.id.clone()
704            }
705
706            fn enter(
707                self,
708                _ctx: &mut (),
709                output: &mut mpsc::Sender<String>,
710            ) -> Result<(), Self::Error> {
711                output.send(format!("entered:{}", self.id)).unwrap();
712                Ok(())
713            }
714
715            fn reconcile_self(
716                self,
717                _state: &mut (),
718                _ctx: &mut (),
719                output: &mut mpsc::Sender<String>,
720            ) -> Result<(), Self::Error> {
721                output.send(format!("reconciled:{}", self.id)).unwrap();
722                Ok(())
723            }
724
725            fn exit(
726                _state: (),
727                _ctx: &mut (),
728                _output: &mut Self::Output,
729            ) -> Result<(), Self::Error> {
730                Ok(())
731            }
732        }
733
734        #[test]
735        fn enter_receives_output_and_can_write_to_it() {
736            let (mut tx, rx) = mpsc::channel::<String>();
737            let mut ms: OptativeSet<ChannelOutputLifecycle> = OptativeSet::new();
738            ms.reconcile(
739                vec![ChannelOutputLifecycle {
740                    id: "o1".to_string(),
741                }],
742                &mut (),
743                &mut tx,
744            );
745            drop(tx);
746            let msgs: Vec<String> = rx.try_iter().collect();
747            assert!(msgs.contains(&"entered:o1".to_string()));
748        }
749
750        #[test]
751        fn exit_does_not_receive_output() {
752            let (mut tx, rx) = mpsc::channel::<String>();
753            let mut ms: OptativeSet<ChannelOutputLifecycle> = OptativeSet::new();
754            ms.reconcile(
755                vec![ChannelOutputLifecycle {
756                    id: "o2".to_string(),
757                }],
758                &mut (),
759                &mut tx,
760            );
761            ms.reconcile(vec![], &mut (), &mut tx);
762            drop(tx);
763            let msgs: Vec<String> = rx.try_iter().collect();
764            assert!(!msgs.iter().any(|m| m.starts_with("exited:")));
765        }
766    }
767}
optative/lib.rs

optative/
lib.rs
