Apply a committed `ArrayOp` entry on the local node. Decodes the op, dispatches it to the Data Plane via SPSC, and records it in the op-log so future `already_seen` checks return `true`. This is the authoritative idempotency gate — it runs on every replica after Raft commit.
(
state: &Arc<SharedState>,
tracker: &Arc<ProposeTracker>,
group_id: u64,
log_index: u64,
applied_key: u64,
array: &str,
op_bytes: &[u8],
)
| 24 | /// in the op-log so future `already_seen` checks return `true`. This is the |
| 25 | /// authoritative idempotency gate — it runs on every replica after Raft commit. |
| 26 | pub(crate) async fn apply_array_op( |
| 27 | state: &Arc<SharedState>, |
| 28 | tracker: &Arc<ProposeTracker>, |
| 29 | group_id: u64, |
| 30 | log_index: u64, |
| 31 | applied_key: u64, |
| 32 | array: &str, |
| 33 | op_bytes: &[u8], |
| 34 | ) { |
| 35 | use crate::types::{TenantId, VShardId}; |
| 36 | use nodedb_array::sync::op_codec; |
| 37 | use nodedb_array::types::coord::value::CoordValue; |
| 38 | use nodedb_cluster::array_routing::{array_vshard_for_name, vshard_for_array_coord}; |
| 39 | |
| 40 | let op = match op_codec::decode_op(op_bytes) { |
| 41 | Ok(op) => op, |
| 42 | Err(e) => { |
| 43 | warn!( |
| 44 | group_id, index = log_index, array = %array, error = %e, |
| 45 | "apply_array_op: decode failed" |
| 46 | ); |
| 47 | tracker.complete( |
| 48 | group_id, |
| 49 | log_index, |
| 50 | applied_key, |
| 51 | Err(crate::Error::Internal { |
| 52 | detail: format!("array op decode: {e}"), |
| 53 | }), |
| 54 | ); |
| 55 | return; |
| 56 | } |
| 57 | }; |
| 58 | |
| 59 | // Authoritative idempotency check: if already applied, skip Data Plane |
| 60 | // dispatch and return success so the proposer waiter is unblocked. |
| 61 | let engine = OriginApplyEngine::new( |
| 62 | Arc::clone(&state.array_sync_schemas), |
| 63 | Arc::clone(&state.array_sync_op_log), |
| 64 | ); |
| 65 | if engine.already_seen(&op.header.array, op.header.hlc) { |
| 66 | tracker.complete(group_id, log_index, applied_key, Ok(vec![])); |
| 67 | return; |
| 68 | } |
| 69 | |
| 70 | // Compute vshard for dispatch. |
| 71 | let tile_extents = state.array_sync_schemas.tile_extents(&op.header.array); |
| 72 | let vshard = if let Some(extents) = tile_extents { |
| 73 | let coord_u64: Vec<u64> = op |
| 74 | .coord |
| 75 | .iter() |
| 76 | .map(|c| match c { |
| 77 | CoordValue::Int64(v) | CoordValue::TimestampMs(v) => *v as u64, |
| 78 | CoordValue::Float64(v) => v.to_bits(), |
| 79 | CoordValue::String(_) => 0, |
| 80 | }) |
| 81 | .collect(); |
| 82 | VShardId::new(vshard_for_array_coord( |
| 83 | &op.header.array, |
no test coverage detected