diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a85c7b9..14d2c80 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,15 +25,15 @@ jobs: - name: Install system dependencies run: | sudo apt-get update - sudo apt-get install -y capnproto libcapnp-dev libclang-dev build-essential + sudo apt-get install -y capnproto libcapnp-dev libclang-dev build-essential pkg-config zlib1g-dev - name: Install Rust toolchain uses: actions-rust-lang/setup-rust-toolchain@v1 with: toolchain: nightly - components: clippy + components: clippy, rustfmt cache: true - cache-key: queueber-3 + cache-key: queueber-4 - name: Build run: cargo build --verbose @@ -44,6 +44,9 @@ jobs: - name: Run Clippy run: cargo clippy --all-targets --all-features -- -D warnings + - name: Check formatting + run: cargo fmt --all -- --check + - name: "`map_err` is forbidden!" run: | ! git grep -E '\.map_err\(' src/**/*.rs | grep -v 'Into::' diff --git a/src/dbkeys.rs b/src/dbkeys.rs index d0fefb6..76dc351 100644 --- a/src/dbkeys.rs +++ b/src/dbkeys.rs @@ -4,15 +4,15 @@ use std::backtrace::Backtrace; use crate::errors::{Error, Result}; -/// Key for items that are available to be polled: `b"available/" + id`. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct AvailableKey(Vec); +/// Key for items that are available to be polled: raw `id` bytes. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub struct AvailableKey<'a>(&'a [u8]); -/// Key for items that are currently in progress: `b"in_progress/" + id`. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct InProgressKey(Vec); +/// Key for items that are currently in progress: raw `id` bytes. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub struct InProgressKey<'a>(&'a [u8]); -/// Visibility index key: `b"visibility_index/" + visible_ts_be + b"/" + id`. +/// Visibility index key: `visible_ts_be + b"/" + id`. /// /// Notes: /// - Timestamp is encoded as 8-byte big-endian to preserve lexicographic @@ -21,83 +21,68 @@ pub struct InProgressKey(Vec); #[derive(Debug, Clone, PartialEq, Eq)] pub struct VisibilityIndexKey(Vec); -/// Lease key: `b"leases/" + lease_bytes`. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct LeaseKey(Vec); +/// Lease key: raw `lease_bytes`. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub struct LeaseKey<'a>(&'a [u8]); -/// Lease expiry index key: `b"lease_expiry/" + expiry_ts_be + b"/" + lease_bytes`. +/// Lease expiry index key: `expiry_ts_be + b"/" + lease_bytes`. /// /// Notes: /// - Timestamp is encoded as 8-byte big-endian to preserve lexicographic ordering by time. #[derive(Debug, Clone, PartialEq, Eq)] pub struct LeaseExpiryIndexKey(Vec); -impl AvailableKey { - pub const PREFIX: &'static [u8] = b"available/"; - - pub fn from_id(id: &[u8]) -> Self { - let mut key = Vec::with_capacity(Self::PREFIX.len() + id.len()); - key.extend_from_slice(Self::PREFIX); - key.extend_from_slice(id); - Self(key) +impl<'a> AvailableKey<'a> { + pub fn from_id(id: &'a [u8]) -> Self { + Self(id) } /// Returns the underlying database key bytes. pub fn as_bytes(&self) -> &[u8] { - &self.0 + self.0 } /// Returns the item id suffix contained in this key. pub fn id_suffix(&self) -> &[u8] { - &self.0[Self::PREFIX.len()..] + self.0 } - /// Returns the item id suffix from a raw `available/` key without allocating. + /// Returns the item id from a raw key without allocating. /// - /// Panics in debug builds if the provided slice does not start with the - /// expected `available/` prefix. + /// With tightened encodings, the key is the id itself. pub fn id_suffix_from_key_bytes(main_key_bytes: &[u8]) -> &[u8] { - debug_assert!(main_key_bytes.starts_with(Self::PREFIX)); - &main_key_bytes[Self::PREFIX.len()..] + main_key_bytes } } -impl AsRef<[u8]> for AvailableKey { +impl AsRef<[u8]> for AvailableKey<'_> { fn as_ref(&self) -> &[u8] { self.as_bytes() } } -impl InProgressKey { - pub const PREFIX: &'static [u8] = b"in_progress/"; - - pub fn from_id(id: &[u8]) -> Self { - let mut key = Vec::with_capacity(Self::PREFIX.len() + id.len()); - key.extend_from_slice(Self::PREFIX); - key.extend_from_slice(id); - Self(key) +impl<'a> InProgressKey<'a> { + pub fn from_id(id: &'a [u8]) -> Self { + Self(id) } pub fn as_bytes(&self) -> &[u8] { - &self.0 + self.0 } } -impl AsRef<[u8]> for InProgressKey { +impl AsRef<[u8]> for InProgressKey<'_> { fn as_ref(&self) -> &[u8] { self.as_bytes() } } impl VisibilityIndexKey { - pub const PREFIX: &'static [u8] = b"visibility_index/"; - /// Build a visibility index key from a visible-at timestamp (secs since epoch) and id. /// Timestamp is encoded as big-endian to preserve lexicographic ordering. pub fn from_visible_ts_and_id(visible_ts_secs: u64, id: &[u8]) -> Self { let ts_be = visible_ts_secs.to_be_bytes(); - let mut key = Vec::with_capacity(Self::PREFIX.len() + ts_be.len() + 1 + id.len()); - key.extend_from_slice(Self::PREFIX); + let mut key = Vec::with_capacity(ts_be.len() + 1 + id.len()); key.extend_from_slice(&ts_be); key.extend_from_slice(b"/"); key.extend_from_slice(id); @@ -110,13 +95,7 @@ impl VisibilityIndexKey { /// Parse the visible-at timestamp (secs since epoch) from a visibility index key. pub fn parse_visible_ts_secs(idx_key: &[u8]) -> Result { - debug_assert_eq!( - &idx_key[..VisibilityIndexKey::PREFIX.len()], - VisibilityIndexKey::PREFIX - ); - - let ts_start = Self::PREFIX.len(); - let ts_end = ts_start + 8; + let ts_end = 8; if idx_key.len() < ts_end + 1 { return Err(Error::AssertionFailed { msg: format!("malformed key: {:?}", idx_key), @@ -124,14 +103,14 @@ impl VisibilityIndexKey { }); } let mut ts_be = [0u8; 8]; - ts_be.copy_from_slice(&idx_key[ts_start..ts_end]); + ts_be.copy_from_slice(&idx_key[..ts_end]); Ok(u64::from_be_bytes(ts_be)) } /// Split a visibility index key into (timestamp_secs, id_slice). pub fn split_ts_and_id(idx_key: &[u8]) -> Result<(u64, &[u8])> { let ts = Self::parse_visible_ts_secs(idx_key)?; - let id_start = Self::PREFIX.len() + 8 + 1; // prefix + ts + '/' + let id_start = 8 + 1; // ts + '/' if id_start > idx_key.len() { return Err(Error::AssertionFailed { msg: format!("malformed key: {:?}", idx_key), @@ -148,34 +127,26 @@ impl AsRef<[u8]> for VisibilityIndexKey { } } -impl LeaseKey { - pub const PREFIX: &'static [u8] = b"leases/"; - - pub fn from_lease_bytes(lease: &[u8]) -> Self { - let mut key = Vec::with_capacity(Self::PREFIX.len() + lease.len()); - key.extend_from_slice(Self::PREFIX); - key.extend_from_slice(lease); - Self(key) +impl<'a> LeaseKey<'a> { + pub fn from_lease_bytes(lease: &'a [u8]) -> Self { + Self(lease) } pub fn as_bytes(&self) -> &[u8] { - &self.0 + self.0 } } -impl AsRef<[u8]> for LeaseKey { +impl AsRef<[u8]> for LeaseKey<'_> { fn as_ref(&self) -> &[u8] { self.as_bytes() } } impl LeaseExpiryIndexKey { - pub const PREFIX: &'static [u8] = b"lease_expiry/"; - pub fn from_expiry_ts_and_lease(expiry_ts_secs: u64, lease: &[u8]) -> Self { let ts_be = expiry_ts_secs.to_be_bytes(); - let mut key = Vec::with_capacity(Self::PREFIX.len() + ts_be.len() + 1 + lease.len()); - key.extend_from_slice(Self::PREFIX); + let mut key = Vec::with_capacity(ts_be.len() + 1 + lease.len()); key.extend_from_slice(&ts_be); key.extend_from_slice(b"/"); key.extend_from_slice(lease); @@ -187,8 +158,7 @@ impl LeaseExpiryIndexKey { } pub fn parse_expiry_ts_secs(idx_key: &[u8]) -> Result { - let ts_start = Self::PREFIX.len(); - let ts_end = ts_start + 8; + let ts_end = 8; if idx_key.len() < ts_end + 1 { return Err(Error::assertion_failed(&format!( "malformed key: {:?}", @@ -196,18 +166,13 @@ impl LeaseExpiryIndexKey { ))); } let mut ts_be = [0u8; 8]; - ts_be.copy_from_slice(&idx_key[ts_start..ts_end]); + ts_be.copy_from_slice(&idx_key[..ts_end]); Ok(u64::from_be_bytes(ts_be)) } pub fn split_ts_and_lease(idx_key: &[u8]) -> Result<(u64, &[u8])> { - debug_assert_eq!( - &idx_key[..LeaseExpiryIndexKey::PREFIX.len()], - LeaseExpiryIndexKey::PREFIX - ); - let ts = Self::parse_expiry_ts_secs(idx_key)?; - let lease_start = Self::PREFIX.len() + 8 + 1; // prefix + ts + '/' + let lease_start = 8 + 1; // ts + '/' if lease_start > idx_key.len() { return Err(Error::assertion_failed(&format!( "malformed key: {:?}", diff --git a/src/storage.rs b/src/storage.rs index 61cb6af..ae4abef 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -1,9 +1,8 @@ use capnp::message::{self, TypedReader}; use capnp::serialize; use rocksdb::{ - ColumnFamilyDescriptor, Direction, IteratorMode, OptimisticTransactionDB, - OptimisticTransactionOptions, Options, ReadOptions, SliceTransform, WriteBatchWithTransaction, - WriteOptions, + ColumnFamilyDescriptor, IteratorMode, OptimisticTransactionDB, OptimisticTransactionOptions, + Options, ReadOptions, SliceTransform, WriteBatchWithTransaction, WriteOptions, }; use std::path::Path; use uuid::Uuid; @@ -239,11 +238,11 @@ impl Storage { /// Return the next visibility timestamp (secs since epoch) among items in the /// visibility index, if any. This is determined by reading the first key in - /// the `visibility_index/` namespace which is ordered by big-endian timestamp. + /// the `visibility_index` CF which is ordered by big-endian timestamp. pub fn peek_next_visibility_ts_secs(&self) -> Result> { let mut iter = self .db - .prefix_iterator_cf(self.cf_visibility_index(), VisibilityIndexKey::PREFIX); + .iterator_cf(self.cf_visibility_index(), IteratorMode::Start); if let Some(kv) = iter.next() { let (idx_key, _) = kv?; Ok(Some(VisibilityIndexKey::parse_visible_ts_secs(&idx_key)?)) @@ -286,17 +285,15 @@ impl Storage { let snapshot = txn.snapshot(); let mut ro_iter = ReadOptions::default(); ro_iter.set_snapshot(&snapshot); - ro_iter.set_prefix_same_as_start(true); // Upper bound: include all visibility_index entries with timestamp <= now_secs // by setting an exclusive upper bound at (now_secs + 1). - let mut viz_upper_bound: Vec = Vec::with_capacity(VisibilityIndexKey::PREFIX.len() + 8); - viz_upper_bound.extend_from_slice(VisibilityIndexKey::PREFIX); + let mut viz_upper_bound: Vec = Vec::with_capacity(8); viz_upper_bound.extend_from_slice(&(now_secs.saturating_add(1)).to_be_bytes()); ro_iter.set_iterate_upper_bound(viz_upper_bound.clone()); let mut ro_get = ReadOptions::default(); ro_get.set_snapshot(&snapshot); // Prefix-bounded forward iterator from the prefix start using the snapshot-bound ReadOptions - let mode = IteratorMode::From(VisibilityIndexKey::PREFIX, Direction::Forward); + let mode = IteratorMode::Start; let viz_iter = txn.iterator_cf_opt(self.cf_visibility_index(), ro_iter, mode); let mut polled_items = Vec::with_capacity(n); @@ -428,14 +425,14 @@ impl Storage { let txn = self.db.transaction(); // Validate item exists in in_progress if txn - .get_pinned_for_update_cf(self.cf_in_progress(), &in_progress_key, true)? + .get_pinned_for_update_cf(self.cf_in_progress(), in_progress_key, true)? .is_none() { return Ok(false); } // Validate lease exists - let Some(lease_value) = txn.get_pinned_for_update_cf(self.cf_leases(), &lease_key, true)? + let Some(lease_value) = txn.get_pinned_for_update_cf(self.cf_leases(), lease_key, true)? else { tracing::info!("lease entry not found: {:?}", Uuid::from_bytes(*lease)); return Ok(false); @@ -523,24 +520,16 @@ impl Storage { // Restrict iterator to only keys with expiry_ts <= now by setting an // exclusive upper bound at (now + 1). let mut ro = ReadOptions::default(); - ro.set_prefix_same_as_start(true); - let mut expiry_upper_bound: Vec = - Vec::with_capacity(LeaseExpiryIndexKey::PREFIX.len() + 8); - expiry_upper_bound.extend_from_slice(LeaseExpiryIndexKey::PREFIX); + let mut expiry_upper_bound: Vec = Vec::with_capacity(8); expiry_upper_bound.extend_from_slice(&(now_secs.saturating_add(1)).to_be_bytes()); ro.set_iterate_upper_bound(expiry_upper_bound.clone()); - let mode = IteratorMode::From(LeaseExpiryIndexKey::PREFIX, Direction::Forward); + let mode = IteratorMode::Start; let iter = txn.iterator_cf_opt(self.cf_lease_expiry(), ro, mode); // Avoid deleting keys while iterating; collect expiry index keys to delete later. let mut expiry_index_keys_to_delete: Vec> = Vec::new(); for kv in iter { let (idx_key, _) = kv?; - debug_assert_eq!( - &idx_key[..LeaseExpiryIndexKey::PREFIX.len()], - LeaseExpiryIndexKey::PREFIX - ); - let _idx_val = txn .get_pinned_for_update_cf(self.cf_lease_expiry(), &idx_key, true)? .ok_or_else(|| { @@ -1468,12 +1457,8 @@ mod tests { let mut before = 0usize; { let txn = storage.db.transaction(); - let mut ro = rocksdb::ReadOptions::default(); - ro.set_prefix_same_as_start(true); - let mode = rocksdb::IteratorMode::From( - LeaseExpiryIndexKey::PREFIX, - rocksdb::Direction::Forward, - ); + let ro = rocksdb::ReadOptions::default(); + let mode = rocksdb::IteratorMode::Start; let iter = txn.iterator_cf_opt(storage.cf_lease_expiry(), ro, mode); for kv in iter { let (idx_key, _val) = kv?; @@ -1492,12 +1477,8 @@ mod tests { let mut after = 0usize; { let txn = storage.db.transaction(); - let mut ro = rocksdb::ReadOptions::default(); - ro.set_prefix_same_as_start(true); - let mode = rocksdb::IteratorMode::From( - LeaseExpiryIndexKey::PREFIX, - rocksdb::Direction::Forward, - ); + let ro = rocksdb::ReadOptions::default(); + let mode = rocksdb::IteratorMode::Start; let iter = txn.iterator_cf_opt(storage.cf_lease_expiry(), ro, mode); for kv in iter { let (idx_key, _val) = kv?; @@ -1567,10 +1548,8 @@ mod tests { // Count expiry index entries that reference this lease let mut count = 0usize; let txn = storage.db.transaction(); - let mut ro = rocksdb::ReadOptions::default(); - ro.set_prefix_same_as_start(true); - let mode = - rocksdb::IteratorMode::From(LeaseExpiryIndexKey::PREFIX, rocksdb::Direction::Forward); + let ro = rocksdb::ReadOptions::default(); + let mode = rocksdb::IteratorMode::Start; let iter = txn.iterator_cf_opt( storage .db @@ -1909,8 +1888,7 @@ mod tests { // Iterate visibility index and capture the first entry let mut ro = rocksdb::ReadOptions::default(); ro.set_prefix_same_as_start(true); - let mode = - rocksdb::IteratorMode::From(VisibilityIndexKey::PREFIX, rocksdb::Direction::Forward); + let mode = rocksdb::IteratorMode::Start; let mut viz_iter = txn.iterator_cf_opt( storage .db