| 1 |
|
| 2 |
|
| 3 |
|
| 4 |
|
| 5 |
|
| 6 |
|
| 7 |
|
| 8 |
|
| 9 |
|
| 10 |
|
| 11 |
|
| 12 |
|
| 13 |
|
| 14 |
|
| 15 |
|
| 16 |
|
| 17 |
|
| 18 |
|
| 19 |
|
| 20 |
use rusqlite::Connection; |
| 21 |
|
| 22 |
use crate::conflict::canonical_payload; |
| 23 |
use crate::error::Result; |
| 24 |
use crate::types::{ChangeEntry, Hlc, PulledChange}; |
| 25 |
|
| 26 |
|
| 27 |
#[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| 28 |
pub(crate) enum LosingSide { |
| 29 |
|
| 30 |
Local, |
| 31 |
|
| 32 |
Remote, |
| 33 |
} |
| 34 |
|
| 35 |
impl LosingSide { |
| 36 |
fn as_str(self) -> &'static str { |
| 37 |
match self { |
| 38 |
LosingSide::Local => "local", |
| 39 |
LosingSide::Remote => "remote", |
| 40 |
} |
| 41 |
} |
| 42 |
} |
| 43 |
|
| 44 |
|
| 45 |
|
| 46 |
|
| 47 |
|
| 48 |
|
| 49 |
|
| 50 |
pub(crate) const MAX_STASH_ROWS: i64 = 1_000; |
| 51 |
|
| 52 |
|
| 53 |
|
| 54 |
fn hlc_text(hlc: &Hlc) -> String { |
| 55 |
format!("{}:{}:{}", hlc.wall_ms, hlc.counter, hlc.node) |
| 56 |
} |
| 57 |
|
| 58 |
|
| 59 |
|
| 60 |
|
| 61 |
|
| 62 |
|
| 63 |
|
| 64 |
pub(crate) fn stash_loser( |
| 65 |
conn: &Connection, |
| 66 |
scope: &str, |
| 67 |
side: LosingSide, |
| 68 |
losing: &ChangeEntry, |
| 69 |
losing_device: crate::ids::DeviceId, |
| 70 |
winning: &ChangeEntry, |
| 71 |
) -> Result<bool> { |
| 72 |
if canonical_payload(losing.data.as_ref()) == canonical_payload(winning.data.as_ref()) { |
| 73 |
return Ok(false); |
| 74 |
} |
| 75 |
|
| 76 |
let payload = losing |
| 77 |
.data |
| 78 |
.as_ref() |
| 79 |
.map(serde_json::to_string) |
| 80 |
.transpose() |
| 81 |
.map_err(|e| crate::error::SyncKitError::Internal(format!("stash payload: {e}")))?; |
| 82 |
|
| 83 |
conn.execute( |
| 84 |
"INSERT INTO sync_conflict_stash |
| 85 |
(table_name, row_id, scope, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc) |
| 86 |
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", |
| 87 |
rusqlite::params![ |
| 88 |
losing.table, |
| 89 |
losing.row_id, |
| 90 |
scope, |
| 91 |
side.as_str(), |
| 92 |
payload, |
| 93 |
hlc_text(&losing.hlc), |
| 94 |
losing_device.to_string(), |
| 95 |
hlc_text(&winning.hlc), |
| 96 |
], |
| 97 |
)?; |
| 98 |
|
| 99 |
tracing::debug!( |
| 100 |
table = %losing.table, |
| 101 |
row_id = %losing.row_id, |
| 102 |
side = side.as_str(), |
| 103 |
"stashed the losing side of a conflict" |
| 104 |
); |
| 105 |
Ok(true) |
| 106 |
} |
| 107 |
|
| 108 |
|
| 109 |
|
| 110 |
|
| 111 |
|
| 112 |
|
| 113 |
|
| 114 |
|
| 115 |
|
| 116 |
pub(crate) fn stash_superseded( |
| 117 |
conn: &Connection, |
| 118 |
scope: &str, |
| 119 |
dropped: &PulledChange, |
| 120 |
committed: &Hlc, |
| 121 |
) -> Result<()> { |
| 122 |
let payload = dropped |
| 123 |
.entry |
| 124 |
.data |
| 125 |
.as_ref() |
| 126 |
.map(serde_json::to_string) |
| 127 |
.transpose() |
| 128 |
.map_err(|e| crate::error::SyncKitError::Internal(format!("stash payload: {e}")))?; |
| 129 |
|
| 130 |
conn.execute( |
| 131 |
"INSERT INTO sync_conflict_stash |
| 132 |
(table_name, row_id, scope, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc) |
| 133 |
VALUES (?1, ?2, ?3, 'remote', ?4, ?5, ?6, ?7)", |
| 134 |
rusqlite::params![ |
| 135 |
dropped.entry.table, |
| 136 |
dropped.entry.row_id, |
| 137 |
scope, |
| 138 |
payload, |
| 139 |
hlc_text(&dropped.entry.hlc), |
| 140 |
dropped.device_id.to_string(), |
| 141 |
hlc_text(committed), |
| 142 |
], |
| 143 |
)?; |
| 144 |
|
| 145 |
tracing::debug!( |
| 146 |
table = %dropped.entry.table, |
| 147 |
row_id = %dropped.entry.row_id, |
| 148 |
"stashed a remote change superseded by the committed HLC" |
| 149 |
); |
| 150 |
Ok(()) |
| 151 |
} |
| 152 |
|
| 153 |
|
| 154 |
|
| 155 |
|
| 156 |
pub(crate) fn trim_stash(conn: &Connection) -> Result<usize> { |
| 157 |
let removed = conn.execute( |
| 158 |
"DELETE FROM sync_conflict_stash WHERE id NOT IN |
| 159 |
(SELECT id FROM sync_conflict_stash ORDER BY id DESC LIMIT ?1)", |
| 160 |
rusqlite::params![MAX_STASH_ROWS], |
| 161 |
)?; |
| 162 |
if removed > 0 { |
| 163 |
tracing::debug!(removed, "trimmed the conflict stash"); |
| 164 |
} |
| 165 |
Ok(removed) |
| 166 |
} |
| 167 |
|
| 168 |
#[cfg(test)] |
| 169 |
mod tests { |
| 170 |
use super::*; |
| 171 |
use crate::ids::DeviceId; |
| 172 |
use uuid::Uuid; |
| 173 |
|
| 174 |
fn hlc(wall_ms: i64, counter: u32, node: &str) -> Hlc { |
| 175 |
Hlc { |
| 176 |
wall_ms, |
| 177 |
counter, |
| 178 |
node: DeviceId::new(Uuid::parse_str(node).unwrap()), |
| 179 |
} |
| 180 |
} |
| 181 |
|
| 182 |
#[test] |
| 183 |
fn hlc_text_is_wall_counter_node() { |
| 184 |
let h = hlc(1_700_000_000_123, 7, "3f2504e0-4f89-41d3-9a0c-0305e82c3301"); |
| 185 |
assert_eq!( |
| 186 |
hlc_text(&h), |
| 187 |
"1700000000123:7:3f2504e0-4f89-41d3-9a0c-0305e82c3301" |
| 188 |
); |
| 189 |
} |
| 190 |
|
| 191 |
#[test] |
| 192 |
fn hlc_text_ordering_tracks_hlc_ordering() { |
| 193 |
|
| 194 |
|
| 195 |
|
| 196 |
|
| 197 |
let node_a = "00000000-0000-0000-0000-00000000000a"; |
| 198 |
let node_b = "00000000-0000-0000-0000-00000000000b"; |
| 199 |
let fixtures = [ |
| 200 |
hlc(1_700_000_000_000, 0, node_a), |
| 201 |
hlc(1_700_000_000_000, 0, node_b), |
| 202 |
hlc(1_700_000_000_000, 1, node_a), |
| 203 |
hlc(1_700_000_000_000, 9, node_b), |
| 204 |
hlc(1_700_000_000_001, 0, node_a), |
| 205 |
hlc(1_899_999_999_999, 4, node_b), |
| 206 |
]; |
| 207 |
|
| 208 |
for (i, left) in fixtures.iter().enumerate() { |
| 209 |
for (j, right) in fixtures.iter().enumerate() { |
| 210 |
assert_eq!( |
| 211 |
hlc_text(left).cmp(&hlc_text(right)), |
| 212 |
left.cmp(right), |
| 213 |
"lexical order disagreed with Hlc order for fixtures {i} and {j}" |
| 214 |
); |
| 215 |
} |
| 216 |
} |
| 217 |
} |
| 218 |
|
| 219 |
#[test] |
| 220 |
fn hlc_text_distinguishes_every_component() { |
| 221 |
let node_a = "00000000-0000-0000-0000-00000000000a"; |
| 222 |
let node_b = "00000000-0000-0000-0000-00000000000b"; |
| 223 |
let base = hlc(1_700_000_000_000, 0, node_a); |
| 224 |
assert_ne!( |
| 225 |
hlc_text(&base), |
| 226 |
hlc_text(&hlc(1_700_000_000_001, 0, node_a)) |
| 227 |
); |
| 228 |
assert_ne!( |
| 229 |
hlc_text(&base), |
| 230 |
hlc_text(&hlc(1_700_000_000_000, 1, node_a)) |
| 231 |
); |
| 232 |
assert_ne!( |
| 233 |
hlc_text(&base), |
| 234 |
hlc_text(&hlc(1_700_000_000_000, 0, node_b)) |
| 235 |
); |
| 236 |
} |
| 237 |
} |
| 238 |
|