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],
)
| 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, |
no test coverage detected