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
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
//! Commit publication with final review under the global commit lock,
//! recovery-only forced publication, SSI unregistration and watermark
//! recycling. Normal commits and recovery re-drives share one index-append
//! path; unregistration and pruning delegate to the tracker and index
//! containers.
use std::ops::Bound::{Excluded, Unbounded};
use std::sync::atomic::Ordering;
use std::sync::Arc;
use dashmap::DashMap;
use linkrs_core::types::Timestamp;
use super::super::context::TransactionContext;
use super::super::error::TransactionError;
use super::super::types::*;
use super::conflict_kind::ConflictType;
use super::Certifier;
impl Certifier {
/// Publish a committed write set into the conflict indices.
///
/// Runs under the global commit lock (the same lock as the
/// certification check) to close the window between certification and
/// committed_write_sets publication. Re-checks all active (validated)
/// transactions and all committed entries since `start_timestamp` as a
/// final review.
///
/// `write_timestamp` MUST be the commit timestamp allocated by
/// `VersionManager::allocate_commit_timestamp`, never the transaction
/// start timestamp: conflict lookups compare stored timestamps against
/// later committers' start timestamps, and visibility is ordered by
/// commit time. Indexing by start time would miss conflicts from
/// long-running writers that commit after a later transaction starts.
///
/// On conflict, returns `Err` and publishes nothing.
/// Lock-free bypass for empty write sets or in-memory read-only transactions.
pub fn publish(
&self,
txn_id: TransactionId,
write_timestamp: Timestamp,
start_timestamp: Timestamp,
write_set: &WriteSet,
active_transactions: &DashMap<TransactionId, Arc<TransactionContext>>,
stats: &TransactionStats,
) -> Result<(), TransactionError> {
if write_set.is_empty() {
// SSI: still unregister read locks even for empty writes
self.ssi_tracker.unregister_reads(txn_id);
return Ok(());
}
// Lock order: commit_lock → committed_write_sets → *
let _cert_guard = self.commit_lock.lock();
let mut committed = self.committed_write_sets.lock();
// Final review under the global commit lock: because the
// certification check ran under this same lock, no conflicting
// transaction can slip between our check and this publication.
// The re-scan of active (validated) transactions and committed
// entries since our start_timestamp is defense in depth.
for entry in active_transactions.iter() {
let (other_id, other_ctx) = entry.pair();
if *other_id == txn_id {
continue;
}
if other_ctx.read_only {
continue;
}
if !other_ctx.is_write_validated() {
continue;
}
if write_set.has_conflict_with(&other_ctx.get_write_set()) {
stats.record_txn_conflict_with_type(ConflictType::WriteWrite);
log::warn!(
"certification conflict (publish) txn={} other={} type={}",
txn_id,
other_id,
ConflictType::WriteWrite
);
return Err(TransactionError::serialization_failed(format!(
"conflict {} write-write with {} (publish)",
ConflictType::WriteWrite,
other_id
)));
}
}
for (_, sets) in committed.range((Excluded(start_timestamp), Unbounded)) {
for ws in sets {
if write_set.has_conflict_with(ws) {
stats.record_txn_conflict_with_type(ConflictType::WriteWrite);
log::warn!(
"certification conflict (publish) txn={} type={}",
txn_id,
ConflictType::WriteWrite
);
return Err(TransactionError::serialization_failed(format!(
"conflict {} write-write with committed batch (publish)",
ConflictType::WriteWrite,
)));
}
}
}
committed
.entry(write_timestamp)
.or_default()
.push(write_set.clone());
self.index_write_set(txn_id, write_timestamp, write_set);
// SSI: unregister read locks and register write locks.
self.ssi_tracker.unregister_reads(txn_id);
Ok(())
}
/// Insert a write set into the conflict indices without certification.
///
/// Recovery-only path: the commit is already WAL-durable and finalized,
/// so it cannot be aborted on conflict. A conflicting entry still
/// defends later committers (they abort instead of silently losing the
/// update), which is the conservative direction. Normal commits MUST
/// use `publish` so conflicts abort before durability.
pub fn force_publish(
&self,
txn_id: TransactionId,
commit_timestamp: Timestamp,
write_set: &WriteSet,
) {
if write_set.is_empty() {
self.ssi_tracker.unregister_reads(txn_id);
return;
}
let _cert_guard = self.commit_lock.lock();
let mut committed = self.committed_write_sets.lock();
if committed
.get(&commit_timestamp)
.is_some_and(|sets| sets.iter().any(|ws| *ws == *write_set))
{
self.ssi_tracker.unregister_reads(txn_id);
return;
}
committed
.entry(commit_timestamp)
.or_default()
.push(write_set.clone());
self.index_write_set(txn_id, commit_timestamp, write_set);
self.ssi_tracker.unregister_reads(txn_id);
}
/// Append a committed write set to all four spatial indices. Shared by
/// the normal publish path and the recovery-only forced publication.
fn index_write_set(&self, txn_id: TransactionId, timestamp: Timestamp, write_set: &WriteSet) {
for vid in write_set.vertices.iter() {
self.vertex_writes.add(vid, timestamp, txn_id);
}
for edge in write_set.edges.iter() {
self.edge_writes.add(edge, timestamp, txn_id);
}
for resource in write_set.schema_resources.iter() {
self.schema_writes.add(resource, timestamp, txn_id);
}
for resource in write_set.index_resources.iter() {
self.index_writes.add(resource, timestamp, txn_id);
}
}
/// Remove all SSI read locks held by `txn_id` (on commit or abort).
pub fn unregister_reads(&self, txn_id: TransactionId) {
self.ssi_tracker.unregister_reads(txn_id);
}
/// Prune committed write sets that are no longer needed by any active
/// transaction. Entries with commit timestamps <= `oldest_active_ts`
/// are safe to remove.
///
/// The pass is batched: a full scan of every retained write set and
/// conflict-index entry runs once per [`super::PRUNE_INTERVAL`] requests
/// instead of on every commit/abort. Stale entries never affect a
/// conflict verdict — they are at or below the oldest active timestamp,
/// so `conflicts_after` and the phantom range scans cannot observe them
/// as newer — so delaying reclamation only retains bounded memory.
pub fn prune(&self, oldest_active_ts: Timestamp) {
let previous = self
.prune_countdown
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| {
Some(if remaining <= 1 {
super::PRUNE_INTERVAL
} else {
remaining - 1
})
})
.unwrap_or(1);
if previous > 1 {
return;
}
let mut committed = self.committed_write_sets.lock();
committed.retain(|ts, _| *ts > oldest_active_ts);
self.vertex_writes.prune(oldest_active_ts);
self.edge_writes.prune(oldest_active_ts);
self.schema_writes.prune(oldest_active_ts);
self.index_writes.prune(oldest_active_ts);
// SSI: prune stale read locks.
self.ssi_tracker.prune(oldest_active_ts);
}
}