Skip to content

Commit dfe9a39

Browse files
farhan-syahfarhan-syah
committed
fix(event): derive WriteOp Insert/Update from storage prior bytes
Introduce CoreLoop::emit_put_event, a shared helper consumed by every handler that executes a put-style mutation against a document engine (PointPut, PointInsert, Upsert, batched PointPut, write_batch). The helper derives the WriteOp tag — Insert when no prior row existed, Update when one was displaced — directly from the bytes returned by storage, making it structurally impossible for an emit site to disagree with what storage actually did. KV SET and ON CONFLICT DO UPDATE paths apply the same pattern: kv/crud.rs threads the prior bytes from the engine's put call into the event, so CDC and trigger consumers see an Update event with both sides populated when an existing key is overwritten. delete.rs now emits old_value for the Event Plane from the prior bytes returned by SparseEngine::delete, giving CDC/trigger subscribers the pre-delete row without a second read pass. bulk_dml.rs and truncate.rs updated to match the revised SparseEngine::delete return type (Option<Vec<u8>> → is_some check).
1 parent a492696 commit dfe9a39

12 files changed

Lines changed: 192 additions & 100 deletions

File tree

‎nodedb/src/data/executor/core_loop/event_emit.rs‎

Lines changed: 54 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,48 @@ impl CoreLoop {
2626
}
2727
}
2828

29+
/// Emit a point write/overwrite/update event derived from the new bytes
30+
/// produced by the handler and the prior bytes returned from storage.
31+
///
32+
/// Shared by every handler that runs a put-style mutation against a
33+
/// document engine: PointPut, Upsert (both branches), batched PointPut,
34+
/// columnar-row overwrite. Each of these knows its *new* bytes and
35+
/// receives *prior* bytes from the storage API; the Event Plane
36+
/// payload (`new_value` / `old_value`) is derived from both after
37+
/// applying the strict→msgpack shim, and the `WriteOp` tag is computed
38+
/// from their presence.
39+
pub(in crate::data::executor) fn emit_put_event(
40+
&mut self,
41+
task: &super::super::task::ExecutionTask,
42+
tid: u32,
43+
collection: &str,
44+
row_id: &str,
45+
new_stored: &[u8],
46+
prior_stored: Option<&[u8]>,
47+
) {
48+
let new_converted = self.resolve_event_payload(tid, collection, new_stored);
49+
let old_converted =
50+
prior_stored.and_then(|p| self.resolve_event_payload(tid, collection, p));
51+
let old_bytes: Option<&[u8]> = match (prior_stored, old_converted.as_deref()) {
52+
(Some(_), Some(c)) => Some(c),
53+
(Some(raw), None) => Some(raw),
54+
(None, _) => None,
55+
};
56+
let op = if old_bytes.is_some() {
57+
crate::event::WriteOp::Update
58+
} else {
59+
crate::event::WriteOp::Insert
60+
};
61+
self.emit_write_event(
62+
task,
63+
collection,
64+
op,
65+
row_id,
66+
Some(new_converted.as_deref().unwrap_or(new_stored)),
67+
old_bytes,
68+
);
69+
}
70+
2971
/// Set the Event Plane producer (called after open, before event loop).
3072
pub fn set_event_producer(&mut self, producer: crate::event::bus::EventProducer) {
3173
self.event_producer = Some(producer);
@@ -34,9 +76,18 @@ impl CoreLoop {
3476
/// Emit a write event to the Event Plane.
3577
///
3678
/// Called after a successful write (PointPut, PointDelete, PointUpdate,
37-
/// BatchInsert, BulkDelete, etc.). The Data Plane NEVER blocks here —
38-
/// if the ring buffer is full, the event is dropped and the Event Plane
39-
/// will detect the gap via sequence numbers and replay from WAL.
79+
/// BatchInsert, BulkDelete, atomic KV ops, etc.). The Data Plane NEVER
80+
/// blocks here — if the ring buffer is full, the event is dropped and
81+
/// the Event Plane will detect the gap via sequence numbers and replay
82+
/// from WAL.
83+
///
84+
/// Prefer [`CoreLoop::emit_put_event`] for any handler that performs a
85+
/// put-style mutation against a document engine — it derives the
86+
/// Insert/Update tag from the prior bytes returned by storage so the
87+
/// emit site cannot disagree with what the row actually did. This
88+
/// lower-level entry point stays for paths where the op is structurally
89+
/// determined by the operation itself (kv-atomic increment, CAS, plain
90+
/// delete) rather than by inspecting pre/post state.
4091
pub(in crate::data::executor) fn emit_write_event(
4192
&mut self,
4293
task: &super::super::task::ExecutionTask,

‎nodedb/src/data/executor/handlers/bulk_dml.rs‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -322,7 +322,13 @@ impl CoreLoop {
322322
// Delete each matching document with full cascade.
323323
let mut affected = 0u64;
324324
for doc_id in &matching_ids {
325-
if self.sparse.delete(tid, collection, doc_id).unwrap_or(false) {
325+
if self
326+
.sparse
327+
.delete(tid, collection, doc_id)
328+
.ok()
329+
.flatten()
330+
.is_some()
331+
{
326332
// Cascade: inverted index.
327333
if let Err(e) = self.inverted.remove_document(
328334
crate::types::TenantId::new(tid),

‎nodedb/src/data/executor/handlers/kv/crud.rs‎

Lines changed: 25 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -70,23 +70,23 @@ impl CoreLoop {
7070
}
7171

7272
let now_ms = current_ms();
73-
let _old = self
73+
let old = self
7474
.kv_engine
7575
.put(tid, collection, key, value, ttl_ms, now_ms);
7676
if let Some(ref m) = self.metrics {
7777
m.record_kv_put();
7878
}
7979

80-
// Emit write event to Event Plane.
80+
// RESP SET / UPSERT semantics: if `old` is present the write
81+
// replaced an existing row and the Event Plane must see an Update
82+
// event with both sides populated. Otherwise it was a fresh
83+
// Insert.
8184
let key_str = String::from_utf8_lossy(key);
82-
self.emit_write_event(
83-
task,
84-
collection,
85-
crate::event::WriteOp::Insert,
86-
&key_str,
87-
Some(value),
88-
None,
89-
);
85+
let (op, old_slice): (_, Option<&[u8]>) = match old.as_deref() {
86+
Some(o) => (crate::event::WriteOp::Update, Some(o)),
87+
None => (crate::event::WriteOp::Insert, None),
88+
};
89+
self.emit_write_event(task, collection, op, &key_str, Some(value), old_slice);
9090

9191
self.response_ok(task)
9292
}
@@ -135,8 +135,7 @@ impl CoreLoop {
135135
);
136136
}
137137

138-
let _old = self
139-
.kv_engine
138+
self.kv_engine
140139
.put(tid, collection, key, value, ttl_ms, now_ms);
141140
if let Some(ref m) = self.metrics {
142141
m.record_kv_put();
@@ -183,8 +182,7 @@ impl CoreLoop {
183182
return self.response_ok(task);
184183
}
185184

186-
let _old = self
187-
.kv_engine
185+
self.kv_engine
188186
.put(tid, collection, key, value, ttl_ms, now_ms);
189187
if let Some(ref m) = self.metrics {
190188
m.record_kv_put();
@@ -235,10 +233,10 @@ impl CoreLoop {
235233
let now_ms = current_ms();
236234
let existing_bytes = self.kv_engine.get(tid, collection, key, now_ms);
237235

238-
let stored_bytes: Vec<u8> = match existing_bytes {
236+
let stored_bytes: Vec<u8> = match &existing_bytes {
239237
None => value.to_vec(),
240238
Some(existing_raw) => {
241-
let existing_val = match nodedb_types::value_from_msgpack(&existing_raw) {
239+
let existing_val = match nodedb_types::value_from_msgpack(existing_raw) {
242240
Ok(v) => v,
243241
Err(_) => {
244242
return self.response_error(
@@ -283,21 +281,28 @@ impl CoreLoop {
283281
}
284282
};
285283

286-
let _old = self
287-
.kv_engine
284+
self.kv_engine
288285
.put(tid, collection, key, &stored_bytes, ttl_ms, now_ms);
289286
if let Some(ref m) = self.metrics {
290287
m.record_kv_put();
291288
}
292289

290+
// `ON CONFLICT DO UPDATE` onto an existing row is an Update from
291+
// every downstream consumer's perspective; a fresh key with no
292+
// prior value is an Insert. The pre-write `existing_bytes` probe
293+
// above is the source of truth.
293294
let key_str = String::from_utf8_lossy(key);
295+
let (op, old_slice): (_, Option<&[u8]>) = match existing_bytes.as_deref() {
296+
Some(o) => (crate::event::WriteOp::Update, Some(o)),
297+
None => (crate::event::WriteOp::Insert, None),
298+
};
294299
self.emit_write_event(
295300
task,
296301
collection,
297-
crate::event::WriteOp::Insert,
302+
op,
298303
&key_str,
299304
Some(&stored_bytes),
300-
None,
305+
old_slice,
301306
);
302307

303308
self.response_ok(task)

‎nodedb/src/data/executor/handlers/point/apply_put.rs‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,14 +13,19 @@ impl CoreLoop {
1313
///
1414
/// Stores the document, auto-indexes text fields, updates column stats,
1515
/// and populates the document cache. Does NOT commit the transaction.
16+
///
17+
/// Returns the prior stored bytes when this put replaced an existing row,
18+
/// or `None` when it was a fresh insert. The caller threads the prior
19+
/// bytes into `emit_write_event` so the Event Plane's `WriteOp` tag
20+
/// reflects the actual mutation.
1621
pub(in crate::data::executor) fn apply_point_put(
1722
&mut self,
1823
txn: &WriteTransaction,
1924
tid: u32,
2025
collection: &str,
2126
document_id: &str,
2227
value: &[u8],
23-
) -> crate::Result<()> {
28+
) -> crate::Result<Option<Vec<u8>>> {
2429
// Evaluate generated columns before encoding.
2530
let config_key = (crate::types::TenantId::new(tid), collection.to_string());
2631
let value = if let Some(config) = self.doc_configs.get(&config_key)
@@ -60,7 +65,8 @@ impl CoreLoop {
6065
value.to_vec()
6166
};
6267

63-
self.sparse
68+
let prior = self
69+
.sparse
6470
.put_in_txn(txn, tid, collection, document_id, &stored)?;
6571

6672
// Text indexing and stats use the original JSON input, not the stored
@@ -282,7 +288,7 @@ impl CoreLoop {
282288
}
283289
}
284290

285-
Ok(())
291+
Ok(prior)
286292
}
287293
}
288294

‎nodedb/src/data/executor/handlers/point/delete.rs‎

Lines changed: 17 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ impl CoreLoop {
1717
) -> Response {
1818
debug!(core = self.core_id, %collection, %document_id, "point delete");
1919
match self.sparse.delete(tid, collection, document_id) {
20-
Ok(_) => {
20+
Ok(prior) => {
2121
// Cascade 1: Remove from full-text inverted index.
2222
if let Err(e) = self.inverted.remove_document(
2323
crate::types::TenantId::new(tid),
@@ -76,15 +76,22 @@ impl CoreLoop {
7676

7777
self.checkpoint_coordinator.mark_dirty("sparse", 1);
7878

79-
// Emit delete event to Event Plane.
80-
self.emit_write_event(
81-
task,
82-
collection,
83-
crate::event::WriteOp::Delete,
84-
document_id,
85-
None,
86-
None, // old_value: would require reading before delete; future batch adds this.
87-
);
79+
// Emit delete event to Event Plane if the row actually
80+
// existed. `sparse.delete` returns the prior bytes — we
81+
// thread them through so CDC/trigger consumers see the
82+
// pre-delete state as `old_value`. A delete against a
83+
// non-existent key is a true no-op and emits nothing.
84+
if let Some(prior_bytes) = prior.as_deref() {
85+
let old_converted = self.resolve_event_payload(tid, collection, prior_bytes);
86+
self.emit_write_event(
87+
task,
88+
collection,
89+
crate::event::WriteOp::Delete,
90+
document_id,
91+
None,
92+
Some(old_converted.as_deref().unwrap_or(prior_bytes)),
93+
);
94+
}
8895

8996
self.response_ok(task)
9097
}

‎nodedb/src/data/executor/handlers/point/insert.rs‎

Lines changed: 4 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,9 @@ impl CoreLoop {
6464
Err(e) => return self.response_error(task, e),
6565
}
6666

67+
// `apply_point_put` returns prior bytes if any — for PointInsert this
68+
// must be `None` because the probe above already rejected the
69+
// conflict case. We intentionally drop it.
6770
if let Err(e) = self.apply_point_put(&txn, tid, collection, document_id, value) {
6871
return self.response_error(task, e);
6972
}
@@ -80,15 +83,7 @@ impl CoreLoop {
8083

8184
self.checkpoint_coordinator.mark_dirty("sparse", 1);
8285

83-
let event_value = self.resolve_event_payload(tid, collection, value);
84-
self.emit_write_event(
85-
task,
86-
collection,
87-
crate::event::WriteOp::Insert,
88-
document_id,
89-
Some(event_value.as_deref().unwrap_or(value)),
90-
None,
91-
);
86+
self.emit_put_event(task, tid, collection, document_id, value, None);
9287

9388
self.response_ok(task)
9489
}

‎nodedb/src/data/executor/handlers/point/put.rs‎

Lines changed: 15 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -31,14 +31,17 @@ impl CoreLoop {
3131
}
3232
};
3333

34-
if let Err(e) = self.apply_point_put(&txn, tid, collection, document_id, value) {
35-
return self.response_error(
36-
task,
37-
ErrorCode::Internal {
38-
detail: e.to_string(),
39-
},
40-
);
41-
}
34+
let prior = match self.apply_point_put(&txn, tid, collection, document_id, value) {
35+
Ok(p) => p,
36+
Err(e) => {
37+
return self.response_error(
38+
task,
39+
ErrorCode::Internal {
40+
detail: e.to_string(),
41+
},
42+
);
43+
}
44+
};
4245

4346
if let Err(e) = txn.commit() {
4447
return self.response_error(
@@ -51,18 +54,10 @@ impl CoreLoop {
5154

5255
self.checkpoint_coordinator.mark_dirty("sparse", 1);
5356

54-
// Emit write event to Event Plane (after successful commit).
55-
// For strict collections, convert Binary Tuple → msgpack so the
56-
// Event Plane can deserialize the payload for trigger dispatch.
57-
let event_value = self.resolve_event_payload(tid, collection, value);
58-
self.emit_write_event(
59-
task,
60-
collection,
61-
crate::event::WriteOp::Insert,
62-
document_id,
63-
Some(event_value.as_deref().unwrap_or(value)),
64-
None,
65-
);
57+
// Emit write event to Event Plane. Insert vs Update is derived
58+
// from whether `prior` was present — a PointPut onto an existing
59+
// row is an Update from every downstream consumer's perspective.
60+
self.emit_put_event(task, tid, collection, document_id, value, prior.as_deref());
6661

6762
self.response_ok(task)
6863
}

‎nodedb/src/data/executor/handlers/point/update.rs‎

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -204,21 +204,22 @@ impl CoreLoop {
204204
.sparse
205205
.put(tid, collection, document_id, &updated_bytes)
206206
{
207-
Ok(()) => {
207+
Ok(_prior) => {
208208
self.doc_cache
209209
.put(tid, collection, document_id, &updated_bytes);
210210

211-
// Emit update event to Event Plane.
212-
// Convert Binary Tuple → msgpack for strict collections.
213-
let new_ev = self.resolve_event_payload(tid, collection, &updated_bytes);
214-
let old_ev = self.resolve_event_payload(tid, collection, &current_bytes);
215-
self.emit_write_event(
211+
// Emit update event to Event Plane. `current_bytes`
212+
// is the pre-update row already read above; the
213+
// helper derives `WriteOp::Update` from the Some
214+
// prior + Some new pair and handles strict→msgpack
215+
// conversion on both sides.
216+
self.emit_put_event(
216217
task,
218+
tid,
217219
collection,
218-
crate::event::WriteOp::Update,
219220
document_id,
220-
Some(new_ev.as_deref().unwrap_or(&updated_bytes)),
221-
Some(old_ev.as_deref().unwrap_or(&current_bytes)),
221+
&updated_bytes,
222+
Some(&current_bytes),
222223
);
223224

224225
if returning {

‎nodedb/src/data/executor/handlers/transaction/sub_plan.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -374,7 +374,7 @@ impl CoreLoop {
374374
encode_for_storage(value)?
375375
};
376376
match self.sparse.put(tid, collection, document_id, &stored) {
377-
Ok(()) => {
377+
Ok(_prior) => {
378378
// Auto-index text fields (includes nested block content).
379379
if let Some(doc) = super::super::super::doc_format::decode_document(value) {
380380
let text_content = super::super::document::extract_indexable_text(&doc);

‎nodedb/src/data/executor/handlers/truncate.rs‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,13 @@ impl CoreLoop {
3737
// Delete each document with full cascade.
3838
let mut truncated = 0u64;
3939
for doc_id in &all_ids {
40-
if self.sparse.delete(tid, collection, doc_id).unwrap_or(false) {
40+
if self
41+
.sparse
42+
.delete(tid, collection, doc_id)
43+
.ok()
44+
.flatten()
45+
.is_some()
46+
{
4147
if let Err(e) = self.inverted.remove_document(
4248
crate::types::TenantId::new(tid),
4349
collection,

0 commit comments

Comments
 (0)