(
&self,
req: &VShardEnvelope,
timeout_ms: u64,
)
| 102 | |
| 103 | impl NexarArrayDispatch { |
| 104 | async fn call_once( |
| 105 | &self, |
| 106 | req: &VShardEnvelope, |
| 107 | timeout_ms: u64, |
| 108 | ) -> ClusterResult<VShardEnvelope> { |
| 109 | // Resolve the shard's leader node from the routing table. |
| 110 | let node_id = { |
| 111 | let rt = self.routing.read().map_err(|_| { |
| 112 | nodedb_cluster::error::ClusterError::Transport { |
| 113 | detail: "routing table lock poisoned".into(), |
| 114 | } |
| 115 | })?; |
| 116 | rt.leader_for_vshard(req.vshard_id)? |
| 117 | }; |
| 118 | |
| 119 | // Short-circuit: if the target shard is owned by this node, dispatch |
| 120 | // directly to the local Data Plane instead of looping through QUIC. |
| 121 | if node_id == self.own_node_id { |
| 122 | let req_opcode = req.msg_type as u32; |
| 123 | let resp_opcode = req_opcode + 1; |
| 124 | let resp_msg_type = array_resp_msg_type(resp_opcode).ok_or_else(|| { |
| 125 | nodedb_cluster::error::ClusterError::Codec { |
| 126 | detail: format!("local dispatch: unknown response opcode {resp_opcode}"), |
| 127 | } |
| 128 | })?; |
| 129 | |
| 130 | let resp_payload = handle_array_shard_rpc( |
| 131 | req_opcode, |
| 132 | req.vshard_id, |
| 133 | &req.payload, |
| 134 | &self.local_executor, |
| 135 | ) |
| 136 | .await?; |
| 137 | |
| 138 | return Ok(VShardEnvelope::new( |
| 139 | resp_msg_type, |
| 140 | self.own_node_id, |
| 141 | req.source_node, |
| 142 | req.vshard_id, |
| 143 | resp_payload, |
| 144 | )); |
| 145 | } |
| 146 | |
| 147 | // Remote path: encode the VShardEnvelope to bytes for the RaftRpc tunnel. |
| 148 | let envelope_bytes = req.to_bytes(); |
| 149 | |
| 150 | // Wrap in RaftRpc and send. The remote node's handler decodes the |
| 151 | // envelope bytes and dispatches to the appropriate shard handler. |
| 152 | let rpc = RaftRpc::VShardEnvelope(envelope_bytes); |
| 153 | let resp_rpc = tokio::time::timeout( |
| 154 | std::time::Duration::from_millis(timeout_ms), |
| 155 | self.transport.send_rpc(node_id, rpc), |
| 156 | ) |
| 157 | .await |
| 158 | .map_err(|_| nodedb_cluster::error::ClusterError::Transport { |
| 159 | detail: format!("array shard RPC timeout ({timeout_ms}ms) to node {node_id}"), |
| 160 | })??; |
| 161 |
no test coverage detected