Something went wrong. Try again.
Chris’s PDS implementation
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257//! Small time-related primitives used by repository and OAuth services.
use std::sync::atomic::{AtomicU64, Ordering};
use time::OffsetDateTime;
const TID_ALPHABET: &[u8; 32] = b"234567abcdefghijklmnopqrstuvwxyz";
/// A process-local, monotonically increasing AT Protocol timestamp identifier.////// A TID stores microseconds in the high bits and a ten-bit clock identifier in/// the low bits. `cpds` uses clock id zero and an atomic compare/exchange so two/// writes in the same microsecond are still ordered.#[derive(Debug, Default)]pub struct TidClock { last: AtomicU64,}
impl TidClock { #[must_use] pub fn next(&self) -> String { self.next_value(0) }
/// Return a TID greater than a previously persisted revision. /// /// Returns `None` when the stored value is not a canonical 64-bit TID. #[must_use] pub fn next_after(&self, previous: &str) -> Option<String> { decode_tid(previous) .filter(|previous| *previous < u64::MAX) .map(|previous| self.next_value(previous)) }
fn next_value(&self, floor: u64) -> String { let wall_time = u64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000) .unwrap_or_default() << 10;
let successor = |previous: u64| { wall_time .max(floor.saturating_add(1)) .max(previous.saturating_add(1)) }; let value = self .last .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |previous| { Some(successor(previous)) }) .map_or(wall_time, successor);
encode_tid(value) }}
fn decode_tid(encoded: &str) -> Option<u64> { if encoded.len() != 13 { return None; } encoded.bytes().try_fold(0_u64, |value, byte| { let digit = TID_ALPHABET .iter() .position(|candidate| *candidate == byte)?; value .checked_mul(32)? .checked_add(u64::try_from(digit).ok()?) })}
fn encode_tid(value: u64) -> String { (0..13) .rev() .map(|digit| char::from(TID_ALPHABET[((value >> (digit * 5)) & 31) as usize])) .collect()}
pub(crate) fn now() -> i64 { OffsetDateTime::now_utc().unix_timestamp()}
pub(crate) fn format_utc_datetime(timestamp: i64) -> Result<String, time::error::ComponentRange> { let value = OffsetDateTime::from_unix_timestamp(timestamp)?; Ok(format!( "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z", value.year(), u8::from(value.month()), value.day(), value.hour(), value.minute(), value.second() ))}
#[cfg(test)]mod tests { use super::TidClock;
#[test] fn immutable_tid_encoding_preserves_every_u64_boundary_and_digit_position() { fn previous_encode_tid(mut value: u64) -> String { let mut encoded = [super::TID_ALPHABET[0]; 13]; for byte in encoded.iter_mut().rev() { *byte = super::TID_ALPHABET[(value & 31) as usize]; value >>= 5; } String::from_utf8(encoded.to_vec()).expect("the TID alphabet is ASCII") }
let mut values = vec![0, u64::MAX]; for bit in 0..64 { let boundary = 1_u64 << bit; values.extend([boundary - 1, boundary, boundary + 1, u64::MAX ^ boundary]); } for position in 0..13 { for digit in 0..32_u128 { if let Ok(value) = u64::try_from(digit << (position * 5)) { values.push(value); } } } let mut mixed = 0x4f70_656e_4149_u64; for _ in 0..10_000 { mixed = mixed .wrapping_mul(6_364_136_223_846_793_005) .wrapping_add(1_442_695_040_888_963_407); values.push(mixed); } values.sort_unstable(); values.dedup(); for value in values { let actual = super::encode_tid(value); assert_eq!(actual, previous_encode_tid(value), "{value}"); assert_eq!(actual.len(), 13); assert_eq!(super::decode_tid(&actual), Some(value)); } assert_eq!(super::encode_tid(0), "2222222222222"); assert_eq!(super::encode_tid(u64::MAX), "jzzzzzzzzzzzz"); }
#[test] fn utc_seconds_keep_previous_formatting_and_range_errors() { use time::{Date, Month, OffsetDateTime};
let minimum = Date::MIN.midnight().assume_utc().unix_timestamp(); let maximum = Date::MAX .with_hms(23, 59, 59) .unwrap() .assume_utc() .unix_timestamp(); let mut timestamps = vec![ i64::MIN, minimum - 1, minimum, -1, 0, 1, maximum, maximum + 1, i64::MAX, ]; for year in [-999, -1, 0, 1, 999, 1000, 1900, 2000, 2026] { for month in [Month::January, Month::February, Month::December] { timestamps.push( Date::from_calendar_date(year, month, 28) .unwrap() .with_hms(12, 34, 56) .unwrap() .assume_utc() .unix_timestamp(), ); } } for timestamp in timestamps { let previous = OffsetDateTime::from_unix_timestamp(timestamp).map(|value| { format!( "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z", value.year(), value.month() as u8, value.day(), value.hour(), value.minute(), value.second() ) }); assert_eq!( super::format_utc_datetime(timestamp), previous, "{timestamp}" ); } assert_eq!( super::format_utc_datetime(0).unwrap(), "1970-01-01T00:00:00Z" ); assert_eq!( super::format_utc_datetime(-1).unwrap(), "1969-12-31T23:59:59Z" ); }
#[test] fn tids_are_fixed_width_and_monotonic() { let clock = TidClock::default(); let first = clock.next(); let second = clock.next();
assert_eq!(first.len(), 13); assert_eq!(second.len(), 13); assert!(first < second); }
#[test] fn persisted_revision_is_a_monotonic_floor() { let clock = TidClock::default();
assert_eq!( clock.next_after("3zzzzzzzzzzzq").as_deref(), Some("3zzzzzzzzzzzr") ); assert!(clock.next_after("not-a-tid").is_none()); assert!(clock.next_after("jzzzzzzzzzzzz").is_none()); }
#[test] fn clock_retains_saturating_floor_and_counter_behavior() { use std::sync::atomic::Ordering;
let clock = TidClock::default(); let maximum = super::encode_tid(u64::MAX); assert_eq!(clock.next_value(u64::MAX - 1), maximum); assert_eq!(clock.last.load(Ordering::SeqCst), u64::MAX); assert_eq!(clock.next(), maximum); assert_eq!(clock.last.load(Ordering::SeqCst), u64::MAX); }
#[test] fn concurrent_clock_updates_return_distinct_published_values() { use std::{collections::HashSet, sync::atomic::Ordering};
let clock = TidClock::default(); let values = std::thread::scope(|scope| { let readers = (0..8) .map(|_| scope.spawn(|| (0..100).map(|_| clock.next()).collect::<Vec<_>>())) .collect::<Vec<_>>(); readers .into_iter() .flat_map(|reader| reader.join().unwrap()) .collect::<HashSet<_>>() }); assert_eq!(values.len(), 800); assert_eq!( values.iter().max().unwrap(), &super::encode_tid(clock.last.load(Ordering::SeqCst)) ); }}