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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
//! Phase 4.2 optimistic auto-commit write path.
//!
//! Multiple concurrent writers can build their staged copies in
//! parallel; at commit time we validate only that *the records this
//! writer touched* have not been modified since our snapshot. Disjoint
//! writers pass validation without retry.
//!
//! See [`Database::execute_mutating_optimistic`] for the contract; the
//! merge mechanics live in [`crate::replay`].
use std::any::Any;
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::time::Instant;
use anyhow::{anyhow, Result};
use lora_compiler::CompiledQuery;
use lora_executor::{LoraValue, MutableExecutionContext, MutableExecutor, Row};
use lora_store::{
GraphStorage, GraphStorageMut, MutationEvent, MutationRecorder, MutationWriteSet,
};
use crate::database::Database;
use crate::transaction::BufferingRecorder;
use crate::wal::write_scope::ensure_wal_query_can_start;
use super::replay::{
install_recorder_if_inmemory, merge_events_into, validate_write_set_unchanged,
};
impl<S> Database<S>
where
S: GraphStorage + GraphStorageMut + Any + Clone + Send + Sync + 'static,
{
/// Optimistic auto-commit write path with per-record conflict
/// detection (Phase 4.2). Multiple concurrent writers can build
/// their staged copies in parallel; at commit time we validate
/// only that *the records this writer touched* have not been
/// modified since our snapshot. Disjoint writers pass validation
/// without retry.
///
/// The merge: a winning writer publishes
/// `current + my_events_replayed`, not `my_snapshot + my_writes`.
/// That preserves any concurrent (disjoint) writer's updates that
/// landed between our snapshot and our commit.
///
/// Per-record locks are acquired (sorted by id) for the write set
/// so two writers with overlapping write sets serialize at the
/// commit boundary rather than racing on validation.
pub(crate) fn execute_mutating_optimistic(
&self,
params: BTreeMap<String, LoraValue>,
deadline: Option<Instant>,
compiled: &Arc<CompiledQuery>,
) -> Result<Vec<Row>> {
// Cap retries so a livelock under heavy contention surfaces
// as an error. With per-record validation, retries happen
// only when this writer's own write set actually overlaps
// a concurrent writer's; the typical case is one iteration.
const MAX_RETRIES: usize = 64;
for _ in 0..MAX_RETRIES {
let snapshot = self.store.load_full();
let mut staged: S = (*snapshot).clone();
// Buffer mutation events tx-locally. Replayed into the
// durable WAL only on the winning commit; a losing
// retry leaves no on-disk trace.
let buffer = Arc::new(Mutex::new(Vec::<MutationEvent>::new()));
let buffering_rec: Arc<dyn MutationRecorder> =
Arc::new(BufferingRecorder::new(buffer.clone()));
install_recorder_if_inmemory(&mut staged, Some(buffering_rec));
let exec_result = {
let mut executor = MutableExecutor::with_deadline(
MutableExecutionContext {
storage: &mut staged,
params: params.clone(),
},
deadline,
);
executor.execute_compiled_rows(compiled)
};
let rows = match exec_result {
Ok(rows) => rows,
Err(e) => return Err(anyhow::Error::from(e)),
};
install_recorder_if_inmemory(&mut staged, None);
// Drain the buffered events. Build the write set from
// them — this is the set of records we're going to
// validate and apply at commit time.
let events: Vec<MutationEvent> = std::mem::take(&mut buffer.lock().unwrap());
if events.is_empty() {
// Mutating-shape query that didn't actually mutate
// (e.g., MATCH that found nothing to SET). Nothing
// to publish — the rows are valid against snapshot.
return Ok(rows);
}
let mut write_set = MutationWriteSet::new();
write_set.extend_from_events(events.iter());
// Brief commit critical section: WAL append + state
// publish must serialize on the writer Mutex so the
// WAL records appear in commit order. The per-record
// `LockTable` exists in `Database` but is intentionally
// not used here — the writer Mutex already provides
// single-writer-at-commit-time semantics, so per-record
// locks would be redundant overhead. They become
// load-bearing in a future phase that drops the writer
// Mutex in favour of ArcSwap CAS for true concurrent
// commits; until then the table is plumbed but idle.
let _commit_lock = self
.writer
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
// Fast path: nothing else committed since our snapshot.
// `staged` already reflects exactly what we want to
// publish — no merge needed. This restores Phase 3's
// single-writer cost (one mutation pass, one graph
// clone) when there's no concurrent activity.
//
// Slow path (concurrent writer landed): validate per
// record, then rebuild publish state by replaying our
// events onto `current`. Roughly doubles per-write cost
// but is the price of preserving concurrent updates.
let current = self.store.load_full();
let publish_state: S = if Arc::ptr_eq(¤t, &snapshot) {
staged
} else {
if !validate_write_set_unchanged(&*snapshot, &*current, &write_set) {
drop(_commit_lock);
continue; // retry from a fresh snapshot
}
let mut merged: S = (*current).clone();
if !merge_events_into(&mut merged, &events) {
// The replay couldn't apply (e.g. id allocation
// collision the validation missed). Retry.
drop(_commit_lock);
continue;
}
merged
};
let mut publish_state = publish_state;
// Durable WAL append. Order matters: WAL goes first so
// a crash between WAL and publish replays cleanly.
let mut wrote_commit = false;
if let Some(rec) = self.wal.as_ref() {
ensure_wal_query_can_start(rec)?;
wrote_commit = rec.commit_events(events)?.wrote();
}
// Reinstall the durable recorder on the merged state so
// the post-publish live store keeps observing mutations.
if let Some(rec) = self.wal.as_ref() {
install_recorder_if_inmemory(
&mut publish_state,
Some(rec.clone() as Arc<dyn MutationRecorder>),
);
}
self.store.store(Arc::new(publish_state));
if wrote_commit {
if let Some(rec) = self.wal.as_ref() {
let live = self.store.load_full();
self.observe_snapshot_commit_if_needed(&*live, rec)?;
}
}
return Ok(rows);
}
Err(anyhow!(
"auto-commit write conflict: exceeded {MAX_RETRIES} retries"
))
}
}