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
// SPDX-License-Identifier: BUSL-1.1
//! CRDT constraint-install handlers.
//!
//! A committed `ConstraintChange` on the per-vshard data Raft log decodes to a
//! `SetConstraints` / `DropConstraints` op and lands here so every replica
//! installs the same constraint set into its per-core (`!Send`) CRDT validator,
//! keyed by collection. The installed set is in-memory: it is rebuilt on
//! restart from Raft-log replay of these entries.
use tracing::{debug, warn};
use crate::bridge::envelope::{ErrorCode, Response};
use crate::data::executor::core_loop::CoreLoop;
use crate::data::executor::task::ExecutionTask;
impl CoreLoop {
/// Install a collection's constraint set into the tenant CRDT validator.
///
/// Each blob is a zerompk-encoded `nodedb_crdt::Constraint`. Decode is
/// loud: a malformed blob fails the whole op rather than silently dropping
/// a constraint, which would weaken the invariant set on this replica.
pub(in crate::data::executor) fn execute_crdt_set_constraints(
&mut self,
task: &ExecutionTask,
collection: &str,
constraint_version: u64,
constraints: &[Vec<u8>],
) -> Response {
debug!(core = self.core_id, %collection, constraint_version, count = constraints.len(), "crdt set constraints");
let mut decoded = Vec::with_capacity(constraints.len());
for blob in constraints {
match zerompk::from_msgpack::<nodedb_crdt::Constraint>(blob) {
Ok(c) => decoded.push(c),
Err(e) => {
warn!(core = self.core_id, error = %e, "crdt constraint decode failed");
return self.response_error(
task,
ErrorCode::Internal {
detail: format!("constraint decode failed: {e}"),
},
);
}
}
}
let tenant_id = task.request.tenant_id;
let engine = match self.get_crdt_engine(task.request.database_id, tenant_id) {
Ok(e) => e,
Err(e) => {
warn!(core = self.core_id, error = %e, "failed to create CRDT engine");
return self.response_error(
task,
ErrorCode::Internal {
detail: e.to_string(),
},
);
}
};
// A `false` return means the incoming version is older than the one
// already installed: a stale duplicate was correctly ignored by the
// fence, which is success, not an error.
if engine.set_collection_constraints(collection, constraint_version, decoded) {
self.checkpoint_coordinator.mark_dirty("crdt", 1);
} else {
debug!(core = self.core_id, %collection, constraint_version, "stale constraint version ignored");
}
self.response_ok(task)
}
/// Remove every constraint scoped to `collection` from the tenant CRDT
/// validator.
pub(in crate::data::executor) fn execute_crdt_drop_constraints(
&mut self,
task: &ExecutionTask,
collection: &str,
constraint_version: u64,
) -> Response {
debug!(core = self.core_id, %collection, constraint_version, "crdt drop constraints");
let tenant_id = task.request.tenant_id;
let engine = match self.get_crdt_engine(task.request.database_id, tenant_id) {
Ok(e) => e,
Err(e) => {
warn!(core = self.core_id, error = %e, "failed to create CRDT engine");
return self.response_error(
task,
ErrorCode::Internal {
detail: e.to_string(),
},
);
}
};
// `false` means a stale (older-version) drop was correctly ignored by
// the fence — success, not an error.
if engine.drop_collection_constraints(collection, constraint_version) {
self.checkpoint_coordinator.mark_dirty("crdt", 1);
} else {
debug!(core = self.core_id, %collection, constraint_version, "stale constraint version ignored");
}
self.response_ok(task)
}
/// Read the constraint set installed in this replica's CRDT validator for
/// `collection`. Read-only — no `mark_dirty`. The installed
/// `Vec<nodedb_crdt::Constraint>` is zerompk-encoded into the response
/// payload so a caller can inspect exactly what this node's validator holds
/// (catalog replication does not prove the validator installed — the
/// validator itself must be read).
pub(in crate::data::executor) fn execute_crdt_read_constraints(
&mut self,
task: &ExecutionTask,
collection: &str,
) -> Response {
debug!(core = self.core_id, %collection, "crdt read constraints");
let tenant_id = task.request.tenant_id;
let engine = match self.get_crdt_engine(task.request.database_id, tenant_id) {
Ok(e) => e,
Err(e) => {
warn!(core = self.core_id, error = %e, "failed to create CRDT engine");
return self.response_error(
task,
ErrorCode::Internal {
detail: e.to_string(),
},
);
}
};
let constraints = engine.constraints_for_collection(collection);
match zerompk::to_msgpack_vec(&constraints) {
Ok(bytes) => self.response_with_payload(task, bytes),
Err(e) => self.response_error(
task,
ErrorCode::Internal {
detail: format!("constraint encode failed: {e}"),
},
),
}
}
}