Merge pull request #89 from skel84/issue-83-m9-bundle-commit

feat(core): implement atomic bundle commit

Author: skel84

crates/allocdb-bench/src/lib.rs

@@ -707,6 +707,7 @@ fn open_hotspot_engine(
         shard_id: 0,
         max_resources: 1,
         max_reservations: hotspot_reservation_capacity(options.hotspot_rounds),
+        max_bundle_size: 1,
         max_operations: hotspot_operation_capacity(
             options.hotspot_rounds,
             options.hotspot_contenders,
@@ -727,6 +728,7 @@ fn open_retry_engine(
         shard_id: 0,
         max_resources: retry_resource_capacity(options.retry_table_capacity),
         max_reservations: 1,
+        max_bundle_size: 1,
         max_operations: options.retry_table_capacity,
         max_ttl_slots: RETRY_TTL_SLOTS,
         max_client_retry_window_slots: RETRY_WINDOW_SLOTS,

crates/allocdb-core/src/command.rs

@@ -6,14 +6,20 @@ pub struct CommandContext {
     pub request_slot: Slot,
 }
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Eq, PartialEq)]
 pub struct ClientRequest {
     pub operation_id: OperationId,
     pub client_id: ClientId,
     pub command: Command,
 }
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+impl AsRef<ClientRequest> for ClientRequest {
+    fn as_ref(&self) -> &ClientRequest {
+        self
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
 pub enum Command {
     CreateResource {
         resource_id: ResourceId,
@@ -23,6 +29,11 @@ pub enum Command {
         holder_id: HolderId,
         ttl_slots: u64,
     },
+    ReserveBundle {
+        resource_ids: Vec<ResourceId>,
+        holder_id: HolderId,
+        ttl_slots: u64,
+    },
     Confirm {
         reservation_id: ReservationId,
         holder_id: HolderId,
@@ -37,9 +48,20 @@ pub enum Command {
     },
 }
 
+impl AsRef<Command> for Command {
+    fn as_ref(&self) -> &Command {
+        self
+    }
+}
+
 impl Command {
     #[must_use]
-    pub fn fingerprint(self) -> u128 {
+    ///
+    /// # Panics
+    ///
+    /// Panics only if the in-memory bundle length cannot fit into `u128`, which cannot happen on
+    /// supported targets because slice lengths are already bounded by `usize`.
+    pub fn fingerprint(&self) -> u128 {
         let mut state = 0x6c62_272e_07bb_0142_62b8_2175_6295_c58du128;
 
         match self {
@@ -55,29 +77,45 @@ impl Command {
                 state = mix(state, 2);
                 state = mix(state, resource_id.get());
                 state = mix(state, holder_id.get());
-                mix(state, u128::from(ttl_slots))
+                mix(state, u128::from(*ttl_slots))
+            }
+            Self::ReserveBundle {
+                resource_ids,
+                holder_id,
+                ttl_slots,
+            } => {
+                state = mix(state, 3);
+                state = mix(
+                    state,
+                    u128::try_from(resource_ids.len()).expect("bundle len must fit u128"),
+                );
+                for resource_id in resource_ids {
+                    state = mix(state, resource_id.get());
+                }
+                state = mix(state, holder_id.get());
+                mix(state, u128::from(*ttl_slots))
             }
             Self::Confirm {
                 reservation_id,
                 holder_id,
             } => {
-                state = mix(state, 3);
+                state = mix(state, 4);
                 state = mix(state, reservation_id.get());
                 mix(state, holder_id.get())
             }
             Self::Release {
                 reservation_id,
                 holder_id,
             } => {
-                state = mix(state, 4);
+                state = mix(state, 5);
                 state = mix(state, reservation_id.get());
                 mix(state, holder_id.get())
             }
             Self::Expire {
                 reservation_id,
                 deadline_slot,
             } => {
-                state = mix(state, 5);
+                state = mix(state, 6);
                 state = mix(state, reservation_id.get());
                 mix(state, u128::from(deadline_slot.get()))
             }

crates/allocdb-core/src/command_codec.rs

@@ -9,11 +9,12 @@ pub enum CommandCodecError {
 }
 
 #[must_use]
-pub fn encode_client_request(request: ClientRequest) -> Vec<u8> {
+pub fn encode_client_request(request: impl AsRef<ClientRequest>) -> Vec<u8> {
+    let request = request.as_ref();
     let mut bytes = Vec::new();
     bytes.extend_from_slice(&request.operation_id.get().to_le_bytes());
     bytes.extend_from_slice(&request.client_id.get().to_le_bytes());
-    encode_command(&mut bytes, request.command);
+    encode_command(&mut bytes, &request.command);
     bytes
 }
 
@@ -36,7 +37,8 @@ pub fn decode_client_request(bytes: &[u8]) -> Result<ClientRequest, CommandCodec
 }
 
 #[must_use]
-pub fn encode_internal_command(command: Command) -> Vec<u8> {
+pub fn encode_internal_command(command: impl AsRef<Command>) -> Vec<u8> {
+    let command = command.as_ref();
     let mut bytes = Vec::new();
     encode_command(&mut bytes, command);
     bytes
@@ -54,7 +56,7 @@ pub fn decode_internal_command(bytes: &[u8]) -> Result<Command, CommandCodecErro
     Ok(command)
 }
 
-fn encode_command(bytes: &mut Vec<u8>, command: Command) {
+fn encode_command(bytes: &mut Vec<u8>, command: &Command) {
     match command {
         Command::CreateResource { resource_id } => {
             bytes.push(1);
@@ -70,27 +72,44 @@ fn encode_command(bytes: &mut Vec<u8>, command: Command) {
             bytes.extend_from_slice(&holder_id.get().to_le_bytes());
             bytes.extend_from_slice(&ttl_slots.to_le_bytes());
         }
+        Command::ReserveBundle {
+            resource_ids,
+            holder_id,
+            ttl_slots,
+        } => {
+            bytes.push(3);
+            bytes.extend_from_slice(
+                &u32::try_from(resource_ids.len())
+                    .expect("bundle len must fit u32")
+                    .to_le_bytes(),
+            );
+            for resource_id in resource_ids {
+                bytes.extend_from_slice(&resource_id.get().to_le_bytes());
+            }
+            bytes.extend_from_slice(&holder_id.get().to_le_bytes());
+            bytes.extend_from_slice(&ttl_slots.to_le_bytes());
+        }
         Command::Confirm {
             reservation_id,
             holder_id,
         } => {
-            bytes.push(3);
+            bytes.push(4);
             bytes.extend_from_slice(&reservation_id.get().to_le_bytes());
             bytes.extend_from_slice(&holder_id.get().to_le_bytes());
         }
         Command::Release {
             reservation_id,
             holder_id,
         } => {
-            bytes.push(4);
+            bytes.push(5);
             bytes.extend_from_slice(&reservation_id.get().to_le_bytes());
             bytes.extend_from_slice(&holder_id.get().to_le_bytes());
         }
         Command::Expire {
             reservation_id,
             deadline_slot,
         } => {
-            bytes.push(5);
+            bytes.push(6);
             bytes.extend_from_slice(&reservation_id.get().to_le_bytes());
             bytes.extend_from_slice(&deadline_slot.get().to_le_bytes());
         }
@@ -107,15 +126,28 @@ fn decode_command(cursor: &mut Cursor<'_>) -> Result<Command, CommandCodecError>
             holder_id: HolderId(cursor.read_u128()?),
             ttl_slots: cursor.read_u64()?,
         }),
-        3 => Ok(Command::Confirm {
+        3 => {
+            let resource_count = usize::try_from(cursor.read_u32()?)
+                .map_err(|_| CommandCodecError::InvalidLayout)?;
+            let mut resource_ids = Vec::with_capacity(resource_count);
+            for _ in 0..resource_count {
+                resource_ids.push(ResourceId(cursor.read_u128()?));
+            }
+            Ok(Command::ReserveBundle {
+                resource_ids,
+                holder_id: HolderId(cursor.read_u128()?),
+                ttl_slots: cursor.read_u64()?,
+            })
+        }
+        4 => Ok(Command::Confirm {
             reservation_id: ReservationId(cursor.read_u128()?),
             holder_id: HolderId(cursor.read_u128()?),
         }),
-        4 => Ok(Command::Release {
+        5 => Ok(Command::Release {
             reservation_id: ReservationId(cursor.read_u128()?),
             holder_id: HolderId(cursor.read_u128()?),
         }),
-        5 => Ok(Command::Expire {
+        6 => Ok(Command::Expire {
             reservation_id: ReservationId(cursor.read_u128()?),
             deadline_slot: Slot(cursor.read_u64()?),
         }),
@@ -160,6 +192,10 @@ impl<'a> Cursor<'a> {
         Ok(u64::from_le_bytes(self.read_exact::<8>()?))
     }
 
+    fn read_u32(&mut self) -> Result<u32, CommandCodecError> {
+        Ok(u32::from_le_bytes(self.read_exact::<4>()?))
+    }
+
     fn read_u128(&mut self) -> Result<u128, CommandCodecError> {
         Ok(u128::from_le_bytes(self.read_exact::<16>()?))
     }
@@ -187,7 +223,7 @@ mod tests {
             },
         };
 
-        let decoded = decode_client_request(&encode_client_request(request)).unwrap();
+        let decoded = decode_client_request(&encode_client_request(&request)).unwrap();
         assert_eq!(decoded, request);
     }
 
@@ -198,10 +234,26 @@ mod tests {
             deadline_slot: Slot(42),
         };
 
-        let decoded = decode_internal_command(&encode_internal_command(command)).unwrap();
+        let decoded = decode_internal_command(&encode_internal_command(&command)).unwrap();
         assert_eq!(decoded, command);
     }
 
+    #[test]
+    fn reserve_bundle_round_trips() {
+        let request = ClientRequest {
+            operation_id: OperationId(11),
+            client_id: ClientId(12),
+            command: Command::ReserveBundle {
+                resource_ids: vec![ResourceId(13), ResourceId(14)],
+                holder_id: HolderId(15),
+                ttl_slots: 16,
+            },
+        };
+
+        let decoded = decode_client_request(&encode_client_request(&request)).unwrap();
+        assert_eq!(decoded, request);
+    }
+
     #[test]
     fn decoder_rejects_truncated_payload() {
         let request = ClientRequest {
@@ -211,7 +263,7 @@ mod tests {
                 resource_id: ResourceId(3),
             },
         };
-        let mut bytes = encode_client_request(request);
+        let mut bytes = encode_client_request(&request);
         bytes.pop();
 
         assert_eq!(

crates/allocdb-core/src/config.rs

@@ -3,6 +3,7 @@ pub struct Config {
     pub shard_id: u64,
     pub max_resources: u32,
     pub max_reservations: u32,
+    pub max_bundle_size: u32,
     pub max_operations: u32,
     pub max_ttl_slots: u64,
     pub max_client_retry_window_slots: u64,
@@ -16,6 +17,8 @@ pub enum ConfigError {
     HistoryWindowTooLarge,
     OperationWindowTooLarge,
     BucketCapacityTooLarge,
+    BundleSizeTooLarge,
+    ReservationMemberTableTooLarge,
     WheelTooLarge,
 }
 
@@ -33,6 +36,18 @@ impl Config {
             .expect("validated operation window must fit in u64")
     }
 
+    /// Returns the fixed-capacity reservation-member table size.
+    ///
+    /// # Panics
+    ///
+    /// Panics if called before [`Self::validate`] has confirmed that the derived table size fits
+    /// `usize`.
+    #[must_use]
+    pub fn max_reservation_members(&self) -> usize {
+        usize::try_from(u64::from(self.max_reservations) * u64::from(self.max_bundle_size))
+            .expect("validated reservation-member table size must fit usize")
+    }
+
     /// Validates that the configured capacities and retention windows are internally consistent.
     ///
     /// # Errors
@@ -52,6 +67,10 @@ impl Config {
             return Err(ConfigError::ZeroCapacity("max_operations"));
         }
 
+        if self.max_bundle_size == 0 {
+            return Err(ConfigError::ZeroCapacity("max_bundle_size"));
+        }
+
         if self.max_ttl_slots == 0 {
             return Err(ConfigError::ZeroCapacity("max_ttl_slots"));
         }
@@ -60,6 +79,10 @@ impl Config {
             return Err(ConfigError::ZeroCapacity("max_expiration_bucket_len"));
         }
 
+        if self.max_bundle_size > self.max_resources {
+            return Err(ConfigError::BundleSizeTooLarge);
+        }
+
         if self.reservation_history_window_slots > self.max_ttl_slots {
             return Err(ConfigError::HistoryWindowTooLarge);
         }
@@ -75,6 +98,16 @@ impl Config {
             return Err(ConfigError::BucketCapacityTooLarge);
         }
 
+        let Some(max_reservation_members) =
+            u64::from(self.max_reservations).checked_mul(u64::from(self.max_bundle_size))
+        else {
+            return Err(ConfigError::ReservationMemberTableTooLarge);
+        };
+
+        if usize::try_from(max_reservation_members).is_err() {
+            return Err(ConfigError::ReservationMemberTableTooLarge);
+        }
+
         let Some(_) = self.max_ttl_slots.checked_add(1) else {
             return Err(ConfigError::WheelTooLarge);
         };

crates/allocdb-core/src/fixed_map.rs

@@ -270,7 +270,7 @@ impl<K: FixedKey, V> FixedMap<K, V> {
     }
 }
 
-fn hash_u128(value: u128) -> u64 {
+pub(crate) fn hash_u128(value: u128) -> u64 {
     let bytes = value.to_le_bytes();
     let lower = u64::from_le_bytes(bytes[..8].try_into().expect("slice has exact size"));
     let upper = u64::from_le_bytes(bytes[8..].try_into().expect("slice has exact size"));

crates/allocdb-core/src/lib.rs

@@ -17,6 +17,7 @@ pub use config::{Config, ConfigError};
 pub use ids::{ClientId, HolderId, Lsn, OperationId, ReservationId, ResourceId, Slot};
 pub use result::{CommandOutcome, ResultCode};
 pub use state_machine::{
-    AllocDb, HealthMetrics, OperationRecord, ReservationLookupError, ReservationRecord,
-    ReservationState, ResourceRecord, ResourceState, SlotOverflowError, SlotOverflowKind,
+    AllocDb, HealthMetrics, OperationRecord, ReservationLookupError, ReservationMemberRecord,
+    ReservationRecord, ReservationState, ResourceRecord, ResourceState, SlotOverflowError,
+    SlotOverflowKind,
 };