MCPcopy Create free account
hub / github.com/NodeDB-Lab/nodedb / apply_op

Method apply_op

nodedb/src/control/array_sync/inbound.rs:321–397  ·  view source on GitHub ↗

Validate a decoded op, then route it through Raft (or directly to the Data Plane in single-node mode) before returning. # Fast-path idempotency The idempotency check here is a *pre-proposal fast-path*: it avoids wasting a Raft round-trip for ops the local replica already knows about. The authoritative check happens in the distributed applier (after Raft commit) so replicas that missed the origin

(
        &self,
        op: ArrayOp,
        raw_op_bytes: &[u8],
    )

Source from the content-addressed store, hash-verified

319 /// (after Raft commit) so replicas that missed the original entry still
320 /// accept it on re-delivery.
321 pub(super) async fn apply_op(
322 &self,
323 op: ArrayOp,
324 raw_op_bytes: &[u8],
325 ) -> Result<InboundOutcome, Option<ArrayRejectMsg>> {
326 // 1. Shape validation.
327 if let Err(e) = op.validate_shape() {
328 return Err(Some(build_reject(
329 &op.header.array,
330 op.header.hlc,
331 ArrayRejectReason::ShapeInvalid,
332 format!("shape validation: {e}"),
333 )));
334 }
335
336 // 2. Schema HLC gating.
337 match self.engine.schema_hlc(&op.header.array) {
338 None => {
339 return Err(Some(build_reject(
340 &op.header.array,
341 op.header.hlc,
342 ArrayRejectReason::ArrayUnknown,
343 format!("array '{}' not known to this replica", op.header.array),
344 )));
345 }
346 Some(local_schema) if op.header.schema_hlc > local_schema => {
347 return Err(Some(build_reject(
348 &op.header.array,
349 op.header.hlc,
350 ArrayRejectReason::SchemaTooNew,
351 format!(
352 "op schema_hlc {:?} > local {:?}; request schema sync",
353 op.header.schema_hlc, local_schema
354 ),
355 )));
356 }
357 Some(_) => {}
358 }
359
360 // 3. Fast-path idempotency check (before proposing).
361 if self.engine.already_seen(&op.header.array, op.header.hlc) {
362 return Ok(InboundOutcome::Idempotent);
363 }
364
365 // 4. In single-node mode (no raft_proposer): apply directly to the
366 // Data Plane, matching the pre-Raft behaviour. This path is only
367 // exercised when the cluster stack has not been started (development,
368 // single-node Origin, unit tests without a raft setup).
369 if self.shared.raft_proposer.get().is_none() {
370 return self.apply_op_direct(op).await;
371 }
372
373 // 5. Multi-node path: propose through Raft.
374 let hlc_bytes = op.header.hlc.to_bytes();
375 let write = ReplicatedWrite::ArrayOp {
376 array: op.header.array.clone(),
377 op_bytes: raw_op_bytes.to_vec(),
378 schema_hlc_bytes: hlc_bytes,

Callers 3

handle_deltaMethod · 0.80
handle_delta_batchMethod · 0.80
handle_snapshot_chunkMethod · 0.80

Calls 14

build_rejectFunction · 0.85
validate_shapeMethod · 0.80
apply_op_directMethod · 0.80
vshard_for_opMethod · 0.80
propose_and_awaitMethod · 0.80
on_op_appliedMethod · 0.80
schema_hlcMethod · 0.45
already_seenMethod · 0.45
getMethod · 0.45
to_bytesMethod · 0.45
cloneMethod · 0.45
to_vecMethod · 0.45

Tested by

no test coverage detected