Single-node fallback: dispatch directly to the Data Plane, bypassing Raft. Used when `raft_proposer` is absent (development / unit tests).
(
&self,
op: ArrayOp,
)
| 127 | /// Single-node fallback: dispatch directly to the Data Plane, bypassing |
| 128 | /// Raft. Used when `raft_proposer` is absent (development / unit tests). |
| 129 | pub(super) async fn apply_op_direct( |
| 130 | &self, |
| 131 | op: ArrayOp, |
| 132 | ) -> Result<InboundOutcome, Option<ArrayRejectMsg>> { |
| 133 | let data_plane_op = self.op_to_data_plane_plan(&op)?; |
| 134 | let vshard = self.vshard_for_op(&op); |
| 135 | |
| 136 | let dispatch_result = |
| 137 | crate::control::server::dispatch_utils::dispatch_to_data_plane_with_source( |
| 138 | self.shared(), |
| 139 | self.tenant_id(), |
| 140 | vshard, |
| 141 | data_plane_op, |
| 142 | TraceId::ZERO, |
| 143 | crate::event::EventSource::CrdtSync, |
| 144 | ) |
| 145 | .await; |
| 146 | |
| 147 | if let Err(e) = dispatch_result { |
| 148 | warn!( |
| 149 | array = %op.header.array, |
| 150 | error = %e, |
| 151 | "array_inbound: Data Plane dispatch failed" |
| 152 | ); |
| 153 | return Err(Some(build_reject( |
| 154 | &op.header.array, |
| 155 | op.header.hlc, |
| 156 | ArrayRejectReason::EngineRejected, |
| 157 | format!("dispatch error: {e}"), |
| 158 | ))); |
| 159 | } |
| 160 | |
| 161 | if let Err(e) = self.engine().record_applied(&op) { |
| 162 | error!( |
| 163 | array = %op.header.array, |
| 164 | hlc = ?op.header.hlc, |
| 165 | error = %e, |
| 166 | "array_inbound: op applied but op-log append failed (replay may re-apply)" |
| 167 | ); |
| 168 | } |
| 169 | |
| 170 | if let Some(observer) = self.apply_observer() { |
| 171 | observer.on_op_applied(&op); |
| 172 | } |
| 173 | |
| 174 | Ok(InboundOutcome::Applied) |
| 175 | } |
| 176 | |
| 177 | /// Compute the vShard that owns this op's tile. |
| 178 | /// |
no test coverage detected