do-memory-core 0.1.31

Core episodic learning system for AI agents with pattern extraction, reward scoring, and dual storage backend
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156

//! Two-phase commit implementation for atomic storage operations

use crate::Result;
use tracing::{debug, error, info, warn};

/// Two-phase commit strategy for episode storage
#[derive(Debug)]
pub struct TwoPhaseCommit {
    /// Phase 1 (cache write) completed
    pub phase1_complete: bool,
    /// Phase 2 (durable write) completed
    pub phase2_complete: bool,
    /// Rollback is needed
    pub rollback_needed: bool,
}

impl TwoPhaseCommit {
    /// Create a new two-phase commit transaction
    #[must_use]
    pub fn new() -> Self {
        Self {
            phase1_complete: false,
            phase2_complete: false,
            rollback_needed: false,
        }
    }

    /// Execute phase 1 - write to cache
    pub async fn phase1<F, Fut>(&mut self, operation: F) -> Result<()>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = Result<()>>,
    {
        debug!("Two-phase commit: Phase 1 (cache write)");
        match operation().await {
            Ok(()) => {
                self.phase1_complete = true;
                Ok(())
            }
            Err(e) => {
                error!("Phase 1 failed: {}", e);
                Err(e)
            }
        }
    }

    /// Execute phase 2 - write to durable storage
    pub async fn phase2<F, Fut>(&mut self, operation: F) -> Result<()>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = Result<()>>,
    {
        debug!("Two-phase commit: Phase 2 (durable write)");
        if !self.phase1_complete {
            return Err(crate::Error::Storage(
                "Cannot execute phase 2 before phase 1".to_string(),
            ));
        }

        match operation().await {
            Ok(()) => {
                self.phase2_complete = true;
                Ok(())
            }
            Err(e) => {
                error!("Phase 2 failed: {}", e);
                self.rollback_needed = true;
                Err(e)
            }
        }
    }

    /// Rollback phase 1 if phase 2 failed
    pub async fn rollback<F, Fut>(&mut self, operation: F) -> Result<()>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = Result<()>>,
    {
        if !self.rollback_needed {
            return Ok(());
        }

        warn!("Rolling back two-phase commit");
        match operation().await {
            Ok(()) => {
                info!("Rollback successful");
                Ok(())
            }
            Err(e) => {
                error!("Rollback failed: {}", e);
                Err(crate::Error::Storage(format!("Rollback failed: {e}")))
            }
        }
    }

    /// Check if commit is complete
    #[must_use]
    pub fn is_complete(&self) -> bool {
        self.phase1_complete && self.phase2_complete && !self.rollback_needed
    }
}

impl Default for TwoPhaseCommit {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_two_phase_commit_new() {
        let commit = TwoPhaseCommit::new();
        assert!(!commit.phase1_complete);
        assert!(!commit.phase2_complete);
        assert!(!commit.rollback_needed);
        assert!(!commit.is_complete());
    }

    #[tokio::test]
    async fn test_two_phase_commit_success() {
        let mut commit = TwoPhaseCommit::new();

        // Phase 1
        let result = commit.phase1(|| async { Ok(()) }).await;
        assert!(result.is_ok());
        assert!(commit.phase1_complete);

        // Phase 2
        let result = commit.phase2(|| async { Ok(()) }).await;
        assert!(result.is_ok());
        assert!(commit.phase2_complete);
        assert!(commit.is_complete());
    }

    #[tokio::test]
    async fn test_two_phase_commit_phase1_failure() {
        let mut commit = TwoPhaseCommit::new();

        let result = commit
            .phase1(|| async { Err(crate::Error::Storage("phase1 failed".into())) })
            .await;
        assert!(result.is_err());
        assert!(!commit.phase1_complete);
    }

    #[tokio::test]
    async fn test_two_phase_commit_no_phase1() {
        let mut commit = TwoPhaseCommit::new();

        let result = commit.phase2(|| async { Ok(()) }).await;
        assert!(result.is_err());
    }
}