Core execution path: route → dispatch with retry → fuse.
(
&self,
ctx: &QueryContext,
plan: PhysicalPlan,
version_set: GatewayVersionSet,
)
| 205 | |
| 206 | /// Core execution path: route → dispatch with retry → fuse. |
| 207 | async fn execute_with_version_set( |
| 208 | &self, |
| 209 | ctx: &QueryContext, |
| 210 | plan: PhysicalPlan, |
| 211 | version_set: GatewayVersionSet, |
| 212 | ) -> Result<Vec<Vec<u8>>, Error> { |
| 213 | // Hold the routing guard only for the route computation, then drop it |
| 214 | // before any await points so the future remains Send. |
| 215 | let routes = { |
| 216 | let routing_guard = self |
| 217 | .shared |
| 218 | .cluster_routing |
| 219 | .as_ref() |
| 220 | .map(|rw| rw.read().unwrap_or_else(|p| p.into_inner())); |
| 221 | let routing = routing_guard.as_deref(); |
| 222 | route_plan(plan, self.shared.node_id, routing, ctx.database_id) |
| 223 | // routing_guard dropped here |
| 224 | }; |
| 225 | |
| 226 | let deadline_ms = default_deadline_ms(&self.shared); |
| 227 | // Gateway-level byte ceiling: per-route `dispatch_to_data_plane` |
| 228 | // already caps each shard's payload; this additionally caps the |
| 229 | // scatter-gather *sum* so an N-shard fan-out can't accumulate |
| 230 | // N × cap across routes. |
| 231 | let max_total_bytes = self.shared.tuning.network.max_query_result_bytes as usize; |
| 232 | let mut all_payloads: Vec<Vec<u8>> = Vec::new(); |
| 233 | let mut accumulated_bytes: usize = 0; |
| 234 | |
| 235 | for route in routes { |
| 236 | let initial_decision = route.decision.clone(); |
| 237 | let vshard_id_for_retry = crate::types::VShardId::new(route.vshard_id); |
| 238 | let plan_for_retry = route.plan.clone(); |
| 239 | let vshard_id_u32 = route.vshard_id; |
| 240 | |
| 241 | let routing_ref = self.shared.cluster_routing.as_deref(); |
| 242 | |
| 243 | let retry_counter = Arc::clone(&self.not_leader_retry_count); |
| 244 | let version_set_for_route = version_set.clone(); |
| 245 | let payloads = retry_not_leader(routing_ref, move |attempt| { |
| 246 | if attempt > 0 { |
| 247 | retry_counter.fetch_add(1, Ordering::Relaxed); |
| 248 | } |
| 249 | let plan = plan_for_retry.clone(); |
| 250 | let shared = Arc::clone(&self.shared); |
| 251 | let tenant_id = ctx.tenant_id; |
| 252 | let database_id = ctx.database_id; |
| 253 | let trace_id = ctx.trace_id; |
| 254 | let version_set = version_set_for_route.clone(); |
| 255 | async move { |
| 256 | let decision = { |
| 257 | let routing_guard = shared |
| 258 | .cluster_routing |
| 259 | .as_ref() |
| 260 | .map(|rw| rw.read().unwrap_or_else(|p| p.into_inner())); |
| 261 | let raft_snapshot: Vec<nodedb_cluster::GroupStatus> = shared |
| 262 | .raft_status_fn |
| 263 | .as_ref() |
| 264 | .map(|f| f()) |
no test coverage detected