Remote dispatch via `ExecuteRequest` RPC.
(args: RemoteDispatchArgs<'_>)
| 111 | |
| 112 | /// Remote dispatch via `ExecuteRequest` RPC. |
| 113 | async fn dispatch_remote(args: RemoteDispatchArgs<'_>) -> Result<Vec<Vec<u8>>, Error> { |
| 114 | let RemoteDispatchArgs { |
| 115 | plan, |
| 116 | shared, |
| 117 | node_id, |
| 118 | vshard_id, |
| 119 | tenant_id, |
| 120 | database_id, |
| 121 | trace_id, |
| 122 | deadline_ms, |
| 123 | version_set, |
| 124 | } = args; |
| 125 | let transport = shared.cluster_transport.as_ref().ok_or(Error::Internal { |
| 126 | detail: "gateway: cluster transport not available for remote dispatch".into(), |
| 127 | })?; |
| 128 | |
| 129 | // Encode the plan. |
| 130 | let plan_bytes = plan_wire::encode(&plan).map_err(|e| Error::Internal { |
| 131 | detail: format!("gateway: plan encode failed: {e}"), |
| 132 | })?; |
| 133 | |
| 134 | // Build descriptor version entries. |
| 135 | let descriptor_versions: Vec<nodedb_cluster::rpc_codec::DescriptorVersionEntry> = version_set |
| 136 | .iter() |
| 137 | .map( |
| 138 | |(name, version)| nodedb_cluster::rpc_codec::DescriptorVersionEntry { |
| 139 | collection: name.clone(), |
| 140 | version: *version, |
| 141 | }, |
| 142 | ) |
| 143 | .collect(); |
| 144 | |
| 145 | let req = RaftRpc::ExecuteRequest(ExecuteRequest { |
| 146 | plan_bytes, |
| 147 | tenant_id: tenant_id.as_u64(), |
| 148 | database_id: database_id.as_u64(), |
| 149 | deadline_remaining_ms: deadline_ms, |
| 150 | trace_id: trace_id.0, |
| 151 | descriptor_versions, |
| 152 | }); |
| 153 | |
| 154 | debug!( |
| 155 | node_id, |
| 156 | vshard_id, |
| 157 | tenant_id = tenant_id.as_u64(), |
| 158 | "gateway: dispatching ExecuteRequest to remote node" |
| 159 | ); |
| 160 | |
| 161 | let resp_rpc = transport.send_rpc(node_id, req).await.map_err(|e| { |
| 162 | // Transport failure means the target node is unreachable — |
| 163 | // we do NOT know who the new leader is. Use leader_node = 0 |
| 164 | // so the retry loop does NOT re-entrench the unreachable node |
| 165 | // as leader in the routing table. The next retry will route |
| 166 | // locally (leader == 0 → local) and let the local Raft state |
| 167 | // resolve to the actual leader. |
| 168 | Error::NotLeader { |
| 169 | vshard_id: VShardId::new((vshard_id % VShardId::COUNT as u64) as u32), |
| 170 | leader_node: 0, |
no test coverage detected