hayson_engine/buffer/
merge.rs

1use serde_json::Value;
2
3use crate::table::{Table, TableError};
4
5use super::entry::{BufferEntry, BufferOp};
6
7#[derive(Debug, Clone, PartialEq)]
8pub struct Conflict {
9    pub table: String,
10    pub key: String,
11    pub versions: Vec<(String, BufferEntry)>,
12}
13
14#[derive(Debug, Clone, PartialEq, Default)]
15pub struct MergeResult {
16    pub clean: Vec<BufferEntry>,
17    pub conflicts: Vec<Conflict>,
18}
19
20/// K-way merges every buffer entry by `buffered_at` — this is the relative order WITHIN one merge
21/// operation only. It never looks at what is already committed in the database: whatever comes out
22/// as `clean` still has to go through `apply_clean_entries`, which stages each entry through the
23/// table's normal `put`/`patch`/`delete` path and so always lands after the current end of the log,
24/// no matter how early `buffered_at` claims to be — this is what satisfies the "never retroactive"
25/// rule, not anything in this function.
26pub fn merge(mut entries: Vec<BufferEntry>) -> MergeResult {
27    let order: Vec<usize> = {
28        let mut idx: Vec<usize> = (0..entries.len()).collect();
29        idx.sort_by(|&a, &b| entries[a].buffered_at.cmp(&entries[b].buffered_at));
30        idx
31    };
32    entries = order.into_iter().map(|i| entries[i].clone()).collect();
33
34    let mut conflicted_keys: Vec<(String, String)> = Vec::new();
35    for entry in &entries {
36        let group_key = (entry.table.clone(), entry.key.clone());
37        if conflicted_keys.contains(&group_key) {
38            continue;
39        }
40        let versions: Vec<&BufferEntry> = entries
41            .iter()
42            .filter(|other| other.table == entry.table && other.key == entry.key)
43            .collect();
44        let distinct_sources: std::collections::HashSet<&str> = versions.iter().map(|e| e.source.as_str()).collect();
45        let first = versions.first().map(|e| (e.op, &e.data));
46        let all_same_content = versions.iter().all(|e| Some((e.op, &e.data)) == first);
47        if distinct_sources.len() > 1 && !all_same_content {
48            conflicted_keys.push(group_key);
49        }
50    }
51
52    let mut clean = Vec::new();
53    let mut conflicts: Vec<Conflict> = Vec::new();
54    for entry in entries {
55        let group_key = (entry.table.clone(), entry.key.clone());
56        if conflicted_keys.contains(&group_key) {
57            match conflicts
58                .iter_mut()
59                .find(|c| c.table == entry.table && c.key == entry.key)
60            {
61                Some(conflict) => conflict.versions.push((entry.source.clone(), entry)),
62                None => conflicts.push(Conflict {
63                    table: entry.table.clone(),
64                    key: entry.key.clone(),
65                    versions: vec![(entry.source.clone(), entry)],
66                }),
67            }
68        } else {
69            clean.push(entry);
70        }
71    }
72
73    MergeResult { clean, conflicts }
74}
75
76/// The integration point with the live database: stages every clean entry through `Table`'s
77/// `put_from_buffer`/`patch_from_buffer`/`delete_from_buffer`, in the order `merge` produced —
78/// never through a lower-level path that could smuggle in a historical `_seq`, and never through
79/// the trigger-firing `put`/`patch`/`delete` (the entry's originating live write already fired
80/// them once). A `TableError::Unique` here means the entry was clean against other buffers but
81/// violates a unique constraint against the CURRENT registry state (which may have changed since
82/// the entry was buffered) — that also becomes a `Conflict`, not a silent error. Any other
83/// `TableError` (validation, not-found on a patch, ...) is surfaced to the caller instead of being
84/// folded into a conflict, since it is not something conflict resolution can fix.
85pub fn apply_clean_entries(table: &mut Table, clean: Vec<BufferEntry>) -> Result<Vec<Conflict>, TableError> {
86    let mut conflicts = Vec::new();
87    for entry in clean {
88        let outcome = match entry.op {
89            BufferOp::Put => {
90                let document = Value::Object(entry.data.clone().unwrap_or_default());
91                table.put_from_buffer(&document, false).map(|_| ())
92            }
93            BufferOp::Patch => {
94                let changes = Value::Object(entry.data.clone().unwrap_or_default());
95                table.patch_from_buffer(&entry.key, &changes, false).map(|_| ())
96            }
97            BufferOp::Delete => table.delete_from_buffer(&entry.key, false).map(|_| ()),
98        };
99        match outcome {
100            Ok(()) => {}
101            Err(TableError::Unique(_)) => conflicts.push(Conflict {
102                table: entry.table.clone(),
103                key: entry.key.clone(),
104                versions: vec![(entry.source.clone(), entry)],
105            }),
106            Err(other) => return Err(other),
107        }
108    }
109    Ok(conflicts)
110}
111
112#[cfg(test)]
113mod tests {
114    use serde_json::{json, Map};
115    use tempfile::tempdir;
116
117    use super::*;
118    use crate::schema::parse_schema;
119
120    fn users_schema() -> crate::schema::TableSchema {
121        parse_schema(&json!({
122            "schemaVersion": 1,
123            "table": "users",
124            "primaryKey": "id",
125            "fields": {
126                "id": { "type": "string" },
127                "name": { "type": "string", "required": true },
128                "email": { "type": "string", "unique": true },
129            },
130        }))
131        .unwrap()
132    }
133
134    fn put(key: &str, source: &str, buffered_at: i64, name: &str) -> BufferEntry {
135        let mut data = Map::new();
136        data.insert("id".into(), Value::String(key.into()));
137        data.insert("name".into(), Value::String(name.into()));
138        BufferEntry {
139            table: "users".into(),
140            key: key.into(),
141            op: BufferOp::Put,
142            data: Some(data),
143            buffered_at,
144            source: source.into(),
145        }
146    }
147
148    fn self_audit_trigger() -> crate::catalog::TriggerEntry {
149        crate::catalog::TriggerEntry {
150            name: "users_insert_self_audit".into(),
151            table: "users".into(),
152            event: crate::catalog::TriggerEvent::Insert,
153            effect: crate::catalog::TriggerEffect {
154                table: "users".into(),
155                fields: vec![
156                    crate::catalog::TriggerFieldMapping {
157                        target: "id".into(),
158                        source: crate::catalog::TriggerValueSource::Constant(json!("audit-u1")),
159                    },
160                    crate::catalog::TriggerFieldMapping {
161                        target: "name".into(),
162                        source: crate::catalog::TriggerValueSource::Constant(json!("audit")),
163                    },
164                ],
165            },
166        }
167    }
168
169    #[test]
170    fn merging_a_buffer_never_fires_a_trigger_the_live_write_already_fired() {
171        let dir = tempdir().unwrap();
172        let path = dir.path().join("users.jsonl");
173        let mut table = Table::open(users_schema(), &path).unwrap();
174        table.set_triggers(vec![self_audit_trigger()]);
175
176        let entry = put("u1", "branch-a", 1, "Ania");
177        let result = merge(vec![entry]);
178        assert!(result.conflicts.is_empty());
179
180        apply_clean_entries(&mut table, result.clean).unwrap();
181
182        assert!(table.get("u1").unwrap().is_some());
183        assert!(table.get("audit-u1").unwrap().is_none());
184    }
185
186    #[test]
187    fn two_folders_without_conflicts_merge_in_buffered_at_order() {
188        let a = put("u2", "branch-a", 200, "Piotr");
189        let b = put("u1", "branch-b", 100, "Ania");
190        let result = merge(vec![a.clone(), b.clone()]);
191
192        assert!(result.conflicts.is_empty());
193        assert_eq!(result.clean, vec![b, a]);
194    }
195
196    #[test]
197    fn the_same_key_changed_differently_in_two_buffers_is_a_conflict_not_clean() {
198        let a = put("u1", "branch-a", 100, "Ania");
199        let b = put("u1", "branch-b", 200, "Zosia");
200        let result = merge(vec![a.clone(), b.clone()]);
201
202        assert!(result.clean.is_empty());
203        assert_eq!(result.conflicts.len(), 1);
204        let conflict = &result.conflicts[0];
205        assert_eq!(conflict.key, "u1");
206        assert_eq!(conflict.versions.len(), 2);
207    }
208
209    #[test]
210    fn a_buffer_entry_earlier_than_an_already_committed_write_never_jumps_ahead_of_it() {
211        let dir = tempdir().unwrap();
212        let path = dir.path().join("users.jsonl");
213        let mut table = Table::open(users_schema(), &path).unwrap();
214
215        table.put(&json!({ "id": "base", "name": "Committed" }), false).unwrap();
216
217        let earlier = put("late-arriving", "branch-a", 1, "FromTheOldBranch");
218        let result = merge(vec![earlier]);
219        assert!(result.conflicts.is_empty());
220
221        apply_clean_entries(&mut table, result.clean).unwrap();
222
223        let raw = std::fs::read_to_string(&path).unwrap();
224        let lines: Vec<&str> = raw.lines().collect();
225        assert_eq!(lines.len(), 2);
226        assert!(lines[0].contains("\"base\""));
227        assert!(lines[1].contains("\"late-arriving\""));
228    }
229
230    #[test]
231    fn a_write_clean_across_buffers_still_conflicts_on_unique_against_current_db_state() {
232        let dir = tempdir().unwrap();
233        let path = dir.path().join("users.jsonl");
234        let mut table = Table::open(users_schema(), &path).unwrap();
235
236        table
237            .put(&json!({ "id": "u1", "name": "Ania", "email": "a@x.io" }), false)
238            .unwrap();
239
240        let mut data = Map::new();
241        data.insert("id".into(), Value::String("u2".into()));
242        data.insert("name".into(), Value::String("Zosia".into()));
243        data.insert("email".into(), Value::String("a@x.io".into()));
244        let entry = BufferEntry {
245            table: "users".into(),
246            key: "u2".into(),
247            op: BufferOp::Put,
248            data: Some(data),
249            buffered_at: 1,
250            source: "branch-a".into(),
251        };
252
253        let result = merge(vec![entry]);
254        assert!(result.conflicts.is_empty());
255
256        let conflicts = apply_clean_entries(&mut table, result.clean).unwrap();
257        assert_eq!(conflicts.len(), 1);
258        assert_eq!(conflicts[0].key, "u2");
259        assert_eq!(table.get("u2").unwrap(), None);
260    }
261}
hayson_engine/buffer/merge.rs

hayson_engine/buffer/
merge.rs
