Import an array schema CRDT snapshot from a Lite peer. Proposes the schema through Raft so it is applied atomically on all replicas. Returns `SchemaImported` on successful commit.
(
&self,
msg: &ArraySchemaSyncMsg,
)
| 213 | /// Proposes the schema through Raft so it is applied atomically on all |
| 214 | /// replicas. Returns `SchemaImported` on successful commit. |
| 215 | pub async fn handle_schema( |
| 216 | &self, |
| 217 | msg: &ArraySchemaSyncMsg, |
| 218 | ) -> Result<InboundOutcome, Option<ArrayRejectMsg>> { |
| 219 | let hlc_arr: [u8; 18] = msg.schema_hlc_bytes; |
| 220 | let remote_hlc = Hlc::from_bytes(&hlc_arr); |
| 221 | |
| 222 | // In single-node mode (no raft_proposer) fall back to direct import. |
| 223 | if self.shared.raft_proposer.get().is_none() { |
| 224 | if let Err(e) = |
| 225 | self.schemas |
| 226 | .import_snapshot(&msg.array, &msg.snapshot_payload, remote_hlc) |
| 227 | { |
| 228 | warn!(array = %msg.array, error = %e, "array_inbound: schema import failed"); |
| 229 | return Err(Some(build_reject( |
| 230 | &msg.array, |
| 231 | remote_hlc, |
| 232 | ArrayRejectReason::EngineRejected, |
| 233 | format!("schema import error: {e}"), |
| 234 | ))); |
| 235 | } |
| 236 | return Ok(InboundOutcome::SchemaImported); |
| 237 | } |
| 238 | |
| 239 | let vshard_id = VShardId::new(array_vshard_for_name(&msg.array)); |
| 240 | let write = ReplicatedWrite::ArraySchema { |
| 241 | array: msg.array.clone(), |
| 242 | snapshot_payload: msg.snapshot_payload.clone(), |
| 243 | schema_hlc_bytes: hlc_arr, |
| 244 | }; |
| 245 | let entry = ReplicatedEntry::new(self.tenant_id.as_u64(), vshard_id.as_u32(), write); |
| 246 | |
| 247 | match self.propose_and_await(entry, &msg.array, remote_hlc).await { |
| 248 | Ok(()) => Ok(InboundOutcome::SchemaImported), |
| 249 | Err(Some(r)) => Err(Some(r)), |
| 250 | Err(None) => Err(None), |
| 251 | } |
| 252 | } |
| 253 | |
| 254 | // ─── Ack ───────────────────────────────────────────────────────────────── |
| 255 |