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
//! DML chain-integrity & integrity-tombstone helpers extracted from `impl_dml`.
//!
//! Behaviour-preserving move (issue #1634). Names and behaviour are unchanged
//! from `impl_dml`; the only adjustment is `pub(super)` visibility on the one
//! private method still called from the sibling `impl_dml` module
//! (`is_chain_integrity_broken`) so its INSERT path keeps calling it by bare
//! name.
use super::*;
use crate::storage::query::unified::UnifiedResult;
impl RedDBRuntime {
/// Issue #524 — public read of the in-memory chain tip. Returns `None`
/// when the collection is not a chain or has no rows (pre-genesis). On a
/// cold cache the first call falls back to a one-time scan so the HTTP
/// `GET /collections/:name/chain-tip` handler stays consistent with the
/// INSERT path after a restart.
pub fn chain_tip_for_collection(
&self,
collection: &str,
) -> Option<crate::runtime::blockchain_kind::ChainTipFull> {
let store = self.inner.db.store();
if !crate::runtime::blockchain_kind::is_chain(&store, collection) {
return None;
}
let mut cache = self.inner.chain_tip_cache.lock();
if let Some(existing) = cache.get(collection) {
return Some(existing.clone());
}
let scanned = crate::runtime::blockchain_kind::chain_tip_full(&store, collection)?;
cache.insert(collection.to_string(), scanned.clone());
Some(scanned)
}
/// Issue #525 — walks the chain end-to-end, recomputes each block's hash
/// against the stored fields, and returns the verification outcome. On
/// `ok == false` the integrity flag is persisted and the in-memory cache
/// is updated so subsequent INSERTs surface `ChainIntegrityBroken`.
///
/// Returns `None` when the collection is absent or not a `KIND blockchain`.
pub fn verify_chain_for_collection(
&self,
collection: &str,
) -> Option<crate::runtime::blockchain_kind::VerifyChainOutcome> {
let store = self.inner.db.store();
let outcome = crate::runtime::blockchain_kind::verify_chain_outcome(&store, collection)?;
if !outcome.ok {
crate::runtime::blockchain_kind::persist_integrity_flag(&store, collection, true);
self.inner
.chain_integrity_broken
.lock()
.insert(collection.to_string(), true);
}
Some(outcome)
}
/// Issue #525 — admin clears the `ChainIntegrityBroken` flag so the chain
/// accepts INSERTs again. Returns `false` when the collection is not a
/// chain.
pub fn clear_chain_integrity_flag(&self, collection: &str) -> bool {
let store = self.inner.db.store();
if !crate::runtime::blockchain_kind::is_chain(&store, collection) {
return false;
}
crate::runtime::blockchain_kind::persist_integrity_flag(&store, collection, false);
self.inner
.chain_integrity_broken
.lock()
.insert(collection.to_string(), false);
true
}
/// Issue #525 — INSERT-time check. Combines in-memory cache (fast path)
/// with a one-time scan of `red_config` on cold start so the flag survives
/// restart.
pub(super) fn is_chain_integrity_broken(&self, collection: &str) -> bool {
{
let cache = self.inner.chain_integrity_broken.lock();
if let Some(v) = cache.get(collection) {
return *v;
}
}
let store = self.inner.db.store();
let persisted =
crate::runtime::blockchain_kind::is_integrity_broken_persisted(&store, collection)
.unwrap_or(false);
self.inner
.chain_integrity_broken
.lock()
.insert(collection.to_string(), persisted);
persisted
}
/// Issue #765 / S6 — lazily hydrate the integrity-tombstone cache from
/// `red_config` on first access. Returns `true` when at least one
/// tombstone range is present. Subsequent calls observe the cached state
/// flag (`1` empty / `2` present) and skip the store scan.
fn ensure_integrity_tombstones_loaded(&self) -> bool {
use std::sync::atomic::Ordering;
match self
.inner
.integrity_tombstones_state
.load(Ordering::Relaxed)
{
1 => return false,
2 => return true,
_ => {}
}
// Cold: load under the cache lock so a concurrent reader cannot
// observe a half-populated vector.
let mut guard = self.inner.integrity_tombstones.lock();
if self
.inner
.integrity_tombstones_state
.load(Ordering::Relaxed)
== 0
{
let ranges = crate::runtime::integrity_tombstone::load_ranges(&self.inner.db.store());
let present = !ranges.is_empty();
*guard = ranges;
self.inner
.integrity_tombstones_state
.store(if present { 2 } else { 1 }, Ordering::Relaxed);
}
self.inner
.integrity_tombstones_state
.load(Ordering::Relaxed)
== 2
}
/// Issue #765 / S6 — durably record an integrity tombstone over the
/// inclusive RID range `[lo, hi]` of `table` (the committed rows of an
/// input stream whose end-to-end SHA-256 digest did not match). The range
/// is persisted to `red_config` (survives restart) and folded into the
/// in-memory cache so the same process filters it immediately.
pub fn record_integrity_tombstone(&self, table: &str, lo: u64, hi: u64) {
use std::sync::atomic::Ordering;
self.ensure_integrity_tombstones_loaded();
let mut guard = self.inner.integrity_tombstones.lock();
guard.push(crate::runtime::integrity_tombstone::TombstoneRange::new(
table.to_string(),
lo,
hi,
));
crate::runtime::integrity_tombstone::persist_ranges(&self.inner.db.store(), &guard);
self.inner
.integrity_tombstones_state
.store(2, Ordering::Relaxed);
}
/// Issue #765 / S6 — snapshot of the currently-cached tombstone ranges.
/// Intended for tests and forensic surfaces; the read path uses
/// [`Self::filter_integrity_tombstoned`] which avoids the clone.
pub fn integrity_tombstone_ranges(
&self,
) -> Vec<crate::runtime::integrity_tombstone::TombstoneRange> {
self.ensure_integrity_tombstones_loaded();
self.inner.integrity_tombstones.lock().clone()
}
/// Issue #765 / S6 — drop tombstoned rows from a SELECT result in place.
/// Fast no-op (one relaxed atomic load) when no tombstone has ever been
/// recorded. Clears `pre_serialized_json` when any row is removed so the
/// fast-path JSON cannot leak a filtered row back onto the wire.
pub fn filter_integrity_tombstoned(&self, result: &mut UnifiedResult) {
if !self.ensure_integrity_tombstones_loaded() {
return;
}
let guard = self.inner.integrity_tombstones.lock();
if guard.is_empty() {
return;
}
let before = result.records.len();
result.records.retain(|record| {
!crate::runtime::integrity_tombstone::record_tombstoned(&guard, record)
});
if result.records.len() != before {
result.pre_serialized_json = None;
}
}
}