(
local_vshard_id: u32,
payload: &[u8],
executor: &Arc<dyn ArrayLocalExecutor>,
)
| 103 | } |
| 104 | |
| 105 | async fn handle_put( |
| 106 | local_vshard_id: u32, |
| 107 | payload: &[u8], |
| 108 | executor: &Arc<dyn ArrayLocalExecutor>, |
| 109 | ) -> Result<Vec<u8>> { |
| 110 | let req: ArrayShardPutReq = |
| 111 | zerompk::from_msgpack(payload).map_err(|e| ClusterError::Codec { |
| 112 | detail: format!("ArrayShardPutReq decode: {e}"), |
| 113 | })?; |
| 114 | |
| 115 | // Validate routing before dispatching: the PUT must have been sent to |
| 116 | // the correct shard. A mismatch means the coordinator used a stale |
| 117 | // routing table and the write must be rejected so the caller can retry |
| 118 | // with a refreshed routing table rather than silently writing to the |
| 119 | // wrong shard. |
| 120 | validate_put_routing(&req, local_vshard_id)?; |
| 121 | |
| 122 | let applied_lsn = executor.exec_put(&req).await?; |
| 123 | let resp = ArrayShardPutResp { |
| 124 | shard_id: local_vshard_id, |
| 125 | applied_lsn, |
| 126 | }; |
| 127 | serialise(resp) |
| 128 | } |
| 129 | |
| 130 | async fn handle_delete( |
| 131 | local_vshard_id: u32, |
no test coverage detected