Send a `MetadataProposeRequest` to `leader_id`. Looks up the leader's listen address via the local topology snapshot and dispatches through the existing peer transport.
(&self, leader_id: u64, data: Vec<u8>)
| 85 | /// leader's listen address via the local topology snapshot and |
| 86 | /// dispatches through the existing peer transport. |
| 87 | async fn forward_metadata_propose(&self, leader_id: u64, data: Vec<u8>) -> Result<u64> { |
| 88 | // Resolve and register the leader's address with the |
| 89 | // transport so `send_rpc` has a destination. Topology is |
| 90 | // updated by the membership / health subsystem; if the |
| 91 | // leader isn't in our local topology yet we fail loudly so |
| 92 | // the caller can fall back to its own retry policy rather |
| 93 | // than silently dropping the proposal. |
| 94 | { |
| 95 | let topo = self.topology.read().unwrap_or_else(|p| p.into_inner()); |
| 96 | let Some(node) = topo.get_node(leader_id) else { |
| 97 | return Err(crate::error::ClusterError::Transport { |
| 98 | detail: format!( |
| 99 | "metadata propose forward: leader {leader_id} not in local topology" |
| 100 | ), |
| 101 | }); |
| 102 | }; |
| 103 | let Some(addr) = node.socket_addr() else { |
| 104 | return Err(crate::error::ClusterError::Transport { |
| 105 | detail: format!( |
| 106 | "metadata propose forward: leader {leader_id} has unparseable addr {:?}", |
| 107 | node.addr |
| 108 | ), |
| 109 | }); |
| 110 | }; |
| 111 | // Idempotent: register_peer overwrites any prior mapping. |
| 112 | self.transport.register_peer(leader_id, addr); |
| 113 | } |
| 114 | |
| 115 | let req = crate::rpc_codec::RaftRpc::MetadataProposeRequest( |
| 116 | crate::rpc_codec::MetadataProposeRequest { bytes: data }, |
| 117 | ); |
| 118 | let resp = self.transport.send_rpc(leader_id, req).await?; |
| 119 | match resp { |
| 120 | crate::rpc_codec::RaftRpc::MetadataProposeResponse(r) => { |
| 121 | if r.success { |
| 122 | Ok(r.log_index) |
| 123 | } else if let Some(hint) = r.leader_hint { |
| 124 | // The receiving node was also not the leader |
| 125 | // (rare: leader changed between our local check |
| 126 | // and the forwarded RPC). Surface as NotLeader |
| 127 | // so the caller's normal retry path runs. |
| 128 | Err(crate::error::ClusterError::Raft( |
| 129 | nodedb_raft::RaftError::NotLeader { |
| 130 | leader_hint: Some(hint), |
| 131 | }, |
| 132 | )) |
| 133 | } else { |
| 134 | Err(crate::error::ClusterError::Transport { |
| 135 | detail: format!("metadata propose forward failed: {}", r.error_message), |
| 136 | }) |
| 137 | } |
| 138 | } |
| 139 | other => Err(crate::error::ClusterError::Transport { |
| 140 | detail: format!("metadata propose forward: unexpected response variant {other:?}"), |
| 141 | }), |
| 142 | } |
| 143 | } |
| 144 |
no test coverage detected