Allocate a chunk of sequence values. In cluster mode: proposes to Raft leader, awaits commit. In single-node mode: directly advances the catalog counter.
(
&self,
state: &crate::control::state::SharedState,
tenant_id: u64,
sequence_name: &str,
increment: i64,
epoch: u64,
)
| 73 | /// In cluster mode: proposes to Raft leader, awaits commit. |
| 74 | /// In single-node mode: directly advances the catalog counter. |
| 75 | pub fn allocate_chunk( |
| 76 | &self, |
| 77 | state: &crate::control::state::SharedState, |
| 78 | tenant_id: u64, |
| 79 | sequence_name: &str, |
| 80 | increment: i64, |
| 81 | epoch: u64, |
| 82 | ) -> Result<RangeAllocationResponse, crate::Error> { |
| 83 | let chunk_size = self.default_chunk_size; |
| 84 | |
| 85 | // In cluster mode, propose through Raft for distributed uniqueness. |
| 86 | if let Some(proposer) = state.raft_proposer.get() { |
| 87 | let request = RangeAllocationRequest { |
| 88 | tenant_id, |
| 89 | sequence_name: sequence_name.to_string(), |
| 90 | chunk_size, |
| 91 | epoch, |
| 92 | }; |
| 93 | let payload = |
| 94 | zerompk::to_msgpack_vec(&request).map_err(|e| crate::Error::Serialization { |
| 95 | format: "msgpack".into(), |
| 96 | detail: format!("range allocation request: {e}"), |
| 97 | })?; |
| 98 | |
| 99 | // Propose to vshard 0 (system shard for metadata operations). |
| 100 | let (_group_id, _log_index) = |
| 101 | proposer(0, payload).map_err(|e| crate::Error::Dispatch { |
| 102 | detail: format!("sequence range allocation raft propose: {e}"), |
| 103 | })?; |
| 104 | |
| 105 | // Compute the allocated range based on current state. |
| 106 | // The Raft commit handler will advance the global counter. |
| 107 | let current = state |
| 108 | .sequence_registry |
| 109 | .get_def(tenant_id, sequence_name) |
| 110 | .map(|d| d.start_value) |
| 111 | .unwrap_or(1); |
| 112 | |
| 113 | let range_start = current; |
| 114 | let range_end = if increment > 0 { |
| 115 | current + chunk_size * increment - increment |
| 116 | } else { |
| 117 | current + chunk_size * increment + increment.abs() |
| 118 | }; |
| 119 | |
| 120 | return Ok(RangeAllocationResponse { |
| 121 | range_start, |
| 122 | range_end, |
| 123 | epoch, |
| 124 | }); |
| 125 | } |
| 126 | |
| 127 | // Single-node mode: allocate directly from the local counter. |
| 128 | // No Raft needed — just advance the counter by chunk_size. |
| 129 | let handle_exists = state.sequence_registry.exists(tenant_id, sequence_name); |
| 130 | if !handle_exists { |
| 131 | return Err(crate::Error::BadRequest { |
| 132 | detail: format!("sequence \"{sequence_name}\" does not exist"), |