diff --git a/Cargo.lock b/Cargo.lock index f6d14d4..6a680b2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -330,6 +330,21 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bit-set" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08807e080ed7f9d5433fa9b275196cfc35414f66a0c79d864dc51a0d825231a3" +dependencies = [ + "bit-vec", +] + +[[package]] +name = "bit-vec" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" + [[package]] name = "bitflags" version = "2.10.0" @@ -2277,6 +2292,7 @@ dependencies = [ "opentelemetry-resource-detectors", "opentelemetry-semantic-conventions", "opentelemetry_sdk", + "proptest", "rand 0.10.1", "reqwest 0.13.2", "rustc-hash", @@ -2525,6 +2541,25 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "proptest" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b45fcc2344c680f5025fe57779faef368840d0bd1f42f216291f0dc4ace4744" +dependencies = [ + "bit-set", + "bit-vec", + "bitflags", + "num-traits", + "rand 0.9.4", + "rand_chacha 0.9.0", + "rand_xorshift", + "regex-syntax", + "rusty-fork", + "tempfile", + "unarray", +] + [[package]] name = "prost" version = "0.14.3" @@ -2660,6 +2695,12 @@ dependencies = [ "winapi", ] +[[package]] +name = "quick-error" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0" + [[package]] name = "quick-xml" version = "0.39.2" @@ -2823,6 +2864,15 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rand_xorshift" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "513962919efc330f829edb2535844d1b912b0fbe2ca165d613e4e8788bb05a5a" +dependencies = [ + "rand_core 0.9.5", +] + [[package]] name = "rand_xoshiro" version = "0.7.0" @@ -3141,6 +3191,18 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +[[package]] +name = "rusty-fork" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc6bf79ff24e648f6da1f8d1f011e9cac26491b619e6b9280f2b47f1774e6ee2" +dependencies = [ + "fnv", + "quick-error", + "tempfile", + "wait-timeout", +] + [[package]] name = "ryu" version = "1.0.22" @@ -4193,6 +4255,12 @@ dependencies = [ "libc", ] +[[package]] +name = "unarray" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94" + [[package]] name = "unicode-bidi" version = "0.3.18" @@ -4334,6 +4402,15 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "wait-timeout" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ac3b126d3914f9849036f826e054cbabdc8519970b8998ddaf3b5bd3c65f11" +dependencies = [ + "libc", +] + [[package]] name = "walkdir" version = "2.5.0" diff --git a/opsqueue/Cargo.toml b/opsqueue/Cargo.toml index 94331b8..388cec0 100644 --- a/opsqueue/Cargo.toml +++ b/opsqueue/Cargo.toml @@ -40,7 +40,6 @@ ciborium = "0.2.2" # Webservers/clients http = "1.4.0" object_store = {version = "0.13.2", features = ["gcp", "http"]} -snowflaked = {version = "1.0.3", features = ["sync"] } tokio-tungstenite = {version = "0.29.0", optional = true} axum = { version = "0.8.9", features = ["ws", "macros"], optional = true } reqwest = { version = "0.13.2", default-features = false, features = ["json"], optional = true } @@ -82,6 +81,8 @@ crossbeam-skiplist = "0.1.3" criterion = {version = "0.8", features = ["async_tokio"]} insta = { version = "1.47.2" } assert_matches = { version = "1.5.0" } +snowflaked = {version = "1.0.3", features = ["sync"] } +proptest = "1.6.0" # [[bench]] # name = "chunks_select" @@ -91,6 +92,10 @@ assert_matches = { version = "1.5.0" } # name = "submissions_insert" # harness = false +[[bench]] +name = "snowflake_compare" +harness = false + [features] # Dependencies only in use by the server-logic: server-logic = [ diff --git a/opsqueue/benches/snowflake_compare.rs b/opsqueue/benches/snowflake_compare.rs new file mode 100644 index 0000000..442dfe0 --- /dev/null +++ b/opsqueue/benches/snowflake_compare.rs @@ -0,0 +1,95 @@ +use std::hint::black_box; +use std::time::Instant; + +use criterion::{criterion_group, criterion_main, Criterion}; +use opsqueue::common::submission::SubmissionId; +use snowflaked::sync::Generator; + +static ORACLE_GENERATOR: Generator = Generator::new(0); + +fn bench_single_thread(c: &mut Criterion) { + let mut group = c.benchmark_group("snowflake_single_thread"); + + group.bench_function("custom_submission_id_new", |b| { + b.iter(|| { + black_box(SubmissionId::new()); + }) + }); + + group.bench_function("snowflaked_generate_u64", |b| { + b.iter(|| { + let id: u64 = ORACLE_GENERATOR.generate(); + black_box(id); + }) + }); + + group.finish(); +} + +fn generate_custom_parallel(threads: usize, ids_per_thread: usize) { + let workers: Vec<_> = (0..threads) + .map(|_| { + std::thread::spawn(move || { + for _ in 0..ids_per_thread { + black_box(SubmissionId::new()); + } + }) + }) + .collect(); + + for worker in workers { + worker + .join() + .expect("parallel custom benchmark worker panicked"); + } +} + +fn generate_oracle_parallel(threads: usize, ids_per_thread: usize) { + let workers: Vec<_> = (0..threads) + .map(|_| { + std::thread::spawn(move || { + for _ in 0..ids_per_thread { + let id: u64 = ORACLE_GENERATOR.generate(); + black_box(id); + } + }) + }) + .collect(); + + for worker in workers { + worker + .join() + .expect("parallel oracle benchmark worker panicked"); + } +} + +fn bench_parallel(c: &mut Criterion) { + let mut group = c.benchmark_group("snowflake_parallel"); + let threads = 4; + let ids_per_thread = 10_000; + + group.bench_function("custom_submission_id_new_threads_4", |b| { + b.iter_custom(|iters| { + let start = Instant::now(); + for _ in 0..iters { + generate_custom_parallel(threads, ids_per_thread); + } + start.elapsed() + }); + }); + + group.bench_function("snowflaked_generate_u64_threads_4", |b| { + b.iter_custom(|iters| { + let start = Instant::now(); + for _ in 0..iters { + generate_oracle_parallel(threads, ids_per_thread); + } + start.elapsed() + }); + }); + + group.finish(); +} + +criterion_group!(benches, bench_single_thread, bench_parallel); +criterion_main!(benches); diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index e5998f6..e862906 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -1,6 +1,8 @@ //! Dealing with `Submission`s: Collections of (`Chunks`s of) operations. use std::fmt::Display; -use std::time::Duration; +use std::hint::spin_loop; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use crate::common::StrategicMetadataMap; use crate::E; @@ -12,7 +14,106 @@ use super::chunk::{ChunkCount, ChunkIndex}; pub type Metadata = Vec; -static ID_GENERATOR: snowflaked::sync::Generator = snowflaked::sync::Generator::new(0); +const TIMESTAMP_BITS: u64 = 42; +const INSTANCE_BITS: u64 = 10; +const SEQUENCE_BITS: u64 = 12; +const TIMESTAMP_SHIFT: u64 = INSTANCE_BITS + SEQUENCE_BITS; +const SEQUENCE_MASK: u64 = (1 << SEQUENCE_BITS) - 1; +const INSTANCE_MASK: u64 = (1 << INSTANCE_BITS) - 1; +const SUBMISSION_SNOWFLAKE_INSTANCE: u64 = 0; + +/// Holds the most recently generated raw snowflake value. +/// +/// This stores packed snowflake components in the exact same layout used by +/// `snowflaked` for `u64`: 42-bit timestamp, 10-bit instance, 12-bit sequence. +static LAST_SUBMISSION_ID: AtomicU64 = AtomicU64::new(0); + +fn now_unix_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system clock is set before the Unix epoch") + .as_millis() as u64 +} + +fn pack_submission_snowflake(timestamp_ms: u64, instance: u64, sequence: u64) -> u64 { + debug_assert!(timestamp_ms < (1 << TIMESTAMP_BITS)); + debug_assert!(instance <= INSTANCE_MASK); + debug_assert!(sequence <= SEQUENCE_MASK); + (timestamp_ms << TIMESTAMP_SHIFT) | (instance << SEQUENCE_BITS) | sequence +} + +fn snowflake_timestamp_ms(snowflake: u64) -> u64 { + snowflake >> TIMESTAMP_SHIFT +} + +#[cfg(test)] +fn snowflake_instance(snowflake: u64) -> u64 { + (snowflake >> SEQUENCE_BITS) & INSTANCE_MASK +} + +fn snowflake_sequence(snowflake: u64) -> u64 { + snowflake & SEQUENCE_MASK +} + +fn wait_until_next_millisecond_with( + last_millisecond: u64, + now_ms: &Now, + tick_wait: &Tick, +) -> u64 +where + Now: Fn() -> u64, + Tick: Fn(), +{ + loop { + let now = now_ms(); + if now > last_millisecond { + return now; + } + tick_wait(); + } +} + +fn generate_submission_snowflake_with( + state: &AtomicU64, + now_ms: Now, + tick_wait: Tick, +) -> u64 +where + Now: Fn() -> u64, + Tick: Fn(), +{ + let instance = SUBMISSION_SNOWFLAKE_INSTANCE; + + loop { + let last = state.load(Ordering::Relaxed); + let last_timestamp_ms = snowflake_timestamp_ms(last); + let last_sequence = snowflake_sequence(last); + let now_ms_value = now_ms(); + + let next = if now_ms_value < last_timestamp_ms { + panic!("Clock has moved backwards! This is not supported"); + } else if now_ms_value > last_timestamp_ms { + pack_submission_snowflake(now_ms_value, instance, 0) + } else if last_sequence < SEQUENCE_MASK { + pack_submission_snowflake(now_ms_value, instance, last_sequence + 1) + } else { + let next_ms = wait_until_next_millisecond_with(last_timestamp_ms, &now_ms, &tick_wait); + pack_submission_snowflake(next_ms, instance, 0) + }; + + if state + .compare_exchange_weak(last, next, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + return next; + } + } +} + +/// Generate a new snowflake value with behavior matching `snowflaked`. +fn generate_submission_snowflake() -> u64 { + generate_submission_snowflake_with(&LAST_SUBMISSION_ID, now_unix_ms, spin_loop) +} /// Uniquely identifies a [`Submission`]. /// @@ -50,20 +151,18 @@ impl Default for SubmissionId { impl SubmissionId { /// Generate a new [`SubmissionId`]. pub fn new() -> SubmissionId { - let inner: u64 = ID_GENERATOR.generate(); + let inner: u64 = generate_submission_snowflake(); SubmissionId(u63::new(inner)) } /// Access the [`std::time::SystemTime`] at which the ID was generated. pub fn system_time(self) -> std::time::SystemTime { - use snowflaked::Snowflake; let inner: u64 = self.0.into(); - let unix_timestamp_ms = inner.timestamp(); + let unix_timestamp_ms = snowflake_timestamp_ms(inner); let unix_timestamp = Duration::from_millis(unix_timestamp_ms); - ID_GENERATOR - .epoch() + UNIX_EPOCH .checked_add(unix_timestamp) - .expect("Invalid timestamp extracted from snowflake ID") + .expect("Invalid timestamp extracted from submission ID") } /// Get the time at which the submission ID was generated as a [`chrono::DateTime`]. pub fn timestamp(self) -> chrono::DateTime { @@ -137,6 +236,526 @@ impl From<&SubmissionId> for std::time::SystemTime { } } +#[cfg(test)] +mod snowflake_tests { + use super::*; + use proptest::prelude::*; + use snowflaked::Snowflake; + use std::collections::HashSet; + use std::sync::atomic::{AtomicBool, AtomicU64}; + + fn timestamp_strategy() -> BoxedStrategy { + let max = (1_u64 << TIMESTAMP_BITS) - 1; + prop_oneof![ + Just(0_u64), + Just(1_u64), + Just(2_u64), + Just(max / 2), + Just(max.saturating_sub(2)), + Just(max.saturating_sub(1)), + Just(max), + 0_u64..=max, + ] + .boxed() + } + + fn instance_strategy() -> BoxedStrategy { + prop_oneof![ + Just(0_u64), + Just(1_u64), + Just(INSTANCE_MASK.saturating_sub(1)), + Just(INSTANCE_MASK), + 0_u64..=INSTANCE_MASK, + ] + .boxed() + } + + fn sequence_strategy() -> BoxedStrategy { + prop_oneof![ + Just(0_u64), + Just(1_u64), + Just(SEQUENCE_MASK.saturating_sub(1)), + Just(SEQUENCE_MASK), + 0_u64..=SEQUENCE_MASK, + ] + .boxed() + } + + #[derive(Clone, Copy, Debug)] + struct SnowflakeModel { + last: u64, + } + + impl SnowflakeModel { + fn new(last: u64) -> Self { + Self { last } + } + + fn set_last(&mut self, last: u64) { + self.last = last; + } + + fn next_with_tick(&mut self, now_ms: u64, tick_to: Option) -> u64 { + let last_timestamp_ms = snowflake_timestamp_ms(self.last); + let last_sequence = snowflake_sequence(self.last); + + let next = if now_ms < last_timestamp_ms { + panic!("Clock has moved backwards! This is not supported"); + } else if now_ms > last_timestamp_ms { + pack_submission_snowflake(now_ms, SUBMISSION_SNOWFLAKE_INSTANCE, 0) + } else if last_sequence < SEQUENCE_MASK { + pack_submission_snowflake(now_ms, SUBMISSION_SNOWFLAKE_INSTANCE, last_sequence + 1) + } else { + let next_ms = tick_to.expect("tick_to must be provided on sequence overflow"); + assert!( + next_ms > last_timestamp_ms, + "tick_to must advance the millisecond" + ); + pack_submission_snowflake(next_ms, SUBMISSION_SNOWFLAKE_INSTANCE, 0) + }; + + self.last = next; + next + } + } + + #[test] + fn ids_are_unique_and_strictly_monotonic() { + let mut seen = HashSet::new(); + let mut previous = SubmissionId::new(); + assert!(seen.insert(previous)); + + for _ in 0..100_000 { + let current = SubmissionId::new(); + assert!( + current > previous, + "expected {current:?} to be strictly greater than {previous:?}" + ); + assert!( + seen.insert(current), + "generated a duplicate id: {current:?}" + ); + previous = current; + } + } + + #[test] + fn system_time_is_close_to_now_and_uses_unix_epoch() { + let before = SystemTime::now(); + let id = SubmissionId::new(); + let after = SystemTime::now(); + + let generated = id.system_time(); + + // The embedded timestamp has millisecond resolution, so allow the + // reconstructed time to sit anywhere within the surrounding window + // (truncated down to the millisecond) plus a small slack. + let lower = before - Duration::from_millis(1); + let upper = after + Duration::from_millis(1); + assert!( + generated >= lower && generated <= upper, + "generated time {generated:?} not within [{lower:?}, {upper:?}]" + ); + } + + #[test] + fn ids_are_unique_across_threads() { + let threads: Vec<_> = (0..8) + .map(|_| { + std::thread::spawn(|| (0..10_000).map(|_| SubmissionId::new()).collect::>()) + }) + .collect(); + + let mut seen = HashSet::new(); + for thread in threads { + for id in thread.join().expect("worker thread panicked") { + assert!( + seen.insert(id), + "generated a duplicate id across threads: {id:?}" + ); + } + } + } + + #[test] + fn packed_layout_matches_snowflaked_u64_layout() { + let id = pack_submission_snowflake(123_456, 17, 3210); + assert_eq!(snowflake_timestamp_ms(id), 123_456); + assert_eq!(snowflake_instance(id), 17); + assert_eq!(snowflake_sequence(id), 3210); + } + + #[test] + fn generated_id_is_compatible_with_snowflaked_shape() { + static REFERENCE_GENERATOR: snowflaked::sync::Generator = + snowflaked::sync::Generator::new(0); + + let ours: u64 = SubmissionId::new().into(); + let theirs: u64 = REFERENCE_GENERATOR.generate(); + + // Both IDs must use the same bit partitioning and instance value. + assert_eq!(snowflake_instance(ours), 0); + assert_eq!(snowflake_instance(theirs), theirs.instance()); + assert_eq!(snowflake_sequence(theirs), theirs.sequence()); + assert_eq!(snowflake_timestamp_ms(theirs), theirs.timestamp()); + + // Generated close in time: timestamps should be very close, though not necessarily equal. + let ours_ts = snowflake_timestamp_ms(ours); + let theirs_ts = theirs.timestamp(); + let diff = ours_ts.abs_diff(theirs_ts); + assert!( + diff <= 5, + "expected close timestamps, got ours={ours_ts}ms theirs={theirs_ts}ms" + ); + } + + #[test] + fn sequence_overflow_waits_next_millisecond_and_resets_to_zero() { + let start_ms = 42_000; + let state = AtomicU64::new(pack_submission_snowflake( + start_ms, + SUBMISSION_SNOWFLAKE_INSTANCE, + SEQUENCE_MASK, + )); + let now = AtomicU64::new(start_ms); + let advanced = AtomicBool::new(false); + + let generated = generate_submission_snowflake_with( + &state, + || now.load(Ordering::Relaxed), + || { + if !advanced.swap(true, Ordering::Relaxed) { + now.store(start_ms + 1, Ordering::Relaxed); + } + }, + ); + + assert_eq!(snowflake_timestamp_ms(generated), start_ms + 1); + assert_eq!(snowflake_instance(generated), SUBMISSION_SNOWFLAKE_INSTANCE); + assert_eq!(snowflake_sequence(generated), 0); + } + + #[test] + fn generated_ids_always_have_valid_components() { + let mut previous_timestamp = 0; + + for _ in 0..100_000 { + let id: u64 = SubmissionId::new().into(); + let timestamp = snowflake_timestamp_ms(id); + let instance = snowflake_instance(id); + let sequence = snowflake_sequence(id); + + assert_eq!(instance, SUBMISSION_SNOWFLAKE_INSTANCE); + assert!(sequence <= SEQUENCE_MASK); + assert!(timestamp >= previous_timestamp); + previous_timestamp = timestamp; + } + } + + #[test] + fn high_volume_multithreaded_generation_has_no_duplicates_after_sort() { + let threads: Vec<_> = (0..8) + .map(|_| { + std::thread::spawn(|| { + (0..25_000) + .map(|_| u64::from(SubmissionId::new())) + .collect::>() + }) + }) + .collect(); + + let mut all_ids = Vec::with_capacity(8 * 25_000); + for thread in threads { + all_ids.extend(thread.join().expect("worker thread panicked")); + } + + all_ids.sort_unstable(); + for pair in all_ids.windows(2) { + assert!( + pair[0] < pair[1], + "duplicate or non-increasing ids: {:?}", + pair + ); + } + } + + #[test] + fn submission_id_json_roundtrip_preserves_value() { + let original = SubmissionId::new(); + let json = serde_json::to_string(&original).expect("serialization failed"); + let decoded: SubmissionId = serde_json::from_str(&json).expect("deserialization failed"); + assert_eq!(decoded, original); + } + + #[test] + fn ids_across_millisecond_boundary_keep_order() { + let id_before = SubmissionId::new(); + let before_ts = snowflake_timestamp_ms(id_before.into()); + + loop { + let now_ms = now_unix_ms(); + if now_ms > before_ts { + break; + } + spin_loop(); + } + + let id_after = SubmissionId::new(); + let after_ts = snowflake_timestamp_ms(id_after.into()); + + assert!(id_after > id_before); + assert!(after_ts >= before_ts); + } + + #[test] + #[should_panic(expected = "Clock has moved backwards! This is not supported")] + fn clock_rollback_panics_like_snowflaked() { + let state = AtomicU64::new(pack_submission_snowflake( + 10_000, + SUBMISSION_SNOWFLAKE_INSTANCE, + 3, + )); + let _ = generate_submission_snowflake_with(&state, || 9_999, || {}); + } + + #[test] + fn pack_matches_snowflaked_from_parts_bit_for_bit() { + let timestamps = [0_u64, 1, 42, 1_234_567, (1 << 20) - 1, (1 << 30) - 1]; + let sequences = [0_u64, 1, 7, 127, 2048, SEQUENCE_MASK]; + + for timestamp in timestamps { + for sequence in sequences { + let ours = + pack_submission_snowflake(timestamp, SUBMISSION_SNOWFLAKE_INSTANCE, sequence); + let theirs = ::from_parts( + timestamp, + SUBMISSION_SNOWFLAKE_INSTANCE, + sequence, + ); + assert_eq!( + ours, theirs, + "bit mismatch for ts={timestamp}, seq={sequence}: ours={ours}, theirs={theirs}" + ); + } + } + } + + #[test] + fn decoder_matches_snowflaked_trait_accessors() { + for _ in 0..50_000 { + let id: u64 = SubmissionId::new().into(); + assert_eq!(snowflake_timestamp_ms(id), id.timestamp()); + assert_eq!(snowflake_instance(id), id.instance()); + assert_eq!(snowflake_sequence(id), id.sequence()); + } + } + + #[test] + fn repeated_multithreaded_stress_has_no_duplicates() { + const ROUNDS: usize = 20; + const THREADS: usize = 6; + const IDS_PER_THREAD: usize = 6_000; + + for _ in 0..ROUNDS { + let threads: Vec<_> = (0..THREADS) + .map(|_| { + std::thread::spawn(|| { + (0..IDS_PER_THREAD) + .map(|_| u64::from(SubmissionId::new())) + .collect::>() + }) + }) + .collect(); + + let mut ids = Vec::with_capacity(THREADS * IDS_PER_THREAD); + for thread in threads { + ids.extend(thread.join().expect("worker thread panicked")); + } + + ids.sort_unstable(); + for pair in ids.windows(2) { + assert!( + pair[0] < pair[1], + "duplicate or non-increasing ids: {:?}", + pair + ); + } + } + } + + #[test] + fn submission_id_u63_boundary_is_handled_correctly() { + let max_u63: u64 = u63::MAX.into(); + let max_compatible_timestamp = max_u63 >> TIMESTAMP_SHIFT; + + let max_compatible_id = + pack_submission_snowflake(max_compatible_timestamp, SUBMISSION_SNOWFLAKE_INSTANCE, 0); + let parsed = SubmissionId::try_from(max_compatible_id) + .expect("max compatible value should be accepted"); + assert_eq!(u64::from(parsed), max_compatible_id); + + let above_boundary_id = pack_submission_snowflake( + max_compatible_timestamp + 1, + SUBMISSION_SNOWFLAKE_INSTANCE, + 0, + ); + assert!( + SubmissionId::try_from(above_boundary_id).is_err(), + "value above u63 boundary should be rejected" + ); + } + + #[test] + fn model_based_scripted_transitions_match_generator() { + let state = AtomicU64::new(0); + let mut model = SnowflakeModel::new(0); + + let scripted_steps = [ + (1_000_u64, None), + (1_000_u64, None), + (1_001_u64, None), + (1_001_u64, None), + (1_050_u64, None), + ]; + + for (now_ms, tick_to) in scripted_steps { + let current = AtomicU64::new(now_ms); + let generated = generate_submission_snowflake_with( + &state, + || current.load(Ordering::Relaxed), + || { + if let Some(next_ms) = tick_to { + current.store(next_ms, Ordering::Relaxed); + } else { + panic!("unexpected wait in non-overflow scripted step"); + } + }, + ); + let expected = model.next_with_tick(now_ms, tick_to); + assert_eq!(generated, expected); + assert_eq!(state.load(Ordering::Relaxed), expected); + } + + let near_overflow = + pack_submission_snowflake(9_000, SUBMISSION_SNOWFLAKE_INSTANCE, SEQUENCE_MASK - 1); + state.store(near_overflow, Ordering::Relaxed); + model.set_last(near_overflow); + + let current = AtomicU64::new(9_000); + let generated_max = generate_submission_snowflake_with( + &state, + || current.load(Ordering::Relaxed), + || panic!("unexpected wait before reaching sequence max"), + ); + let expected_max = model.next_with_tick(9_000, None); + assert_eq!(generated_max, expected_max); + assert_eq!(snowflake_sequence(generated_max), SEQUENCE_MASK); + + let current = AtomicU64::new(9_000); + let ticked = AtomicBool::new(false); + let generated_after_wrap = generate_submission_snowflake_with( + &state, + || current.load(Ordering::Relaxed), + || { + if !ticked.swap(true, Ordering::Relaxed) { + current.store(9_001, Ordering::Relaxed); + } + }, + ); + let expected_after_wrap = model.next_with_tick(9_000, Some(9_001)); + assert_eq!(generated_after_wrap, expected_after_wrap); + assert_eq!(snowflake_timestamp_ms(generated_after_wrap), 9_001); + assert_eq!(snowflake_sequence(generated_after_wrap), 0); + } + + #[test] + fn clock_rollback_failure_mode_matches_oracle() { + let ours = std::panic::catch_unwind(|| { + let state = AtomicU64::new(pack_submission_snowflake( + 10_000, + SUBMISSION_SNOWFLAKE_INSTANCE, + 3, + )); + let _ = generate_submission_snowflake_with(&state, || 9_999, || {}); + }); + assert!( + ours.is_err(), + "expected custom generator to panic on rollback" + ); + + let oracle = std::panic::catch_unwind(|| { + let future_epoch = SystemTime::now() + .checked_add(Duration::from_millis(100)) + .expect("future epoch should fit in SystemTime"); + let generator: snowflaked::sync::Generator = snowflaked::sync::Generator::builder() + .instance(0) + .epoch(future_epoch) + .build(); + let _: u64 = generator.generate(); + }); + assert!( + oracle.is_err(), + "expected snowflaked generator to panic on rollback-like time failure" + ); + } + + proptest! { + #[test] + fn prop_pack_unpack_roundtrip_preserves_components( + timestamp in timestamp_strategy(), + instance in instance_strategy(), + sequence in sequence_strategy(), + ) { + let packed = pack_submission_snowflake(timestamp, instance, sequence); + prop_assert_eq!(snowflake_timestamp_ms(packed), timestamp); + prop_assert_eq!(snowflake_instance(packed), instance); + prop_assert_eq!(snowflake_sequence(packed), sequence); + } + + #[test] + fn prop_pack_matches_snowflaked_oracle( + timestamp in timestamp_strategy(), + instance in instance_strategy(), + sequence in sequence_strategy(), + ) { + let ours = pack_submission_snowflake(timestamp, instance, sequence); + let oracle = ::from_parts(timestamp, instance, sequence); + prop_assert_eq!(ours, oracle); + } + + #[test] + fn prop_submission_ids_are_monotonic_and_unique( + sample_size in 2_usize..1_000 + ) { + let mut previous = SubmissionId::new(); + let mut seen = HashSet::with_capacity(sample_size); + prop_assert!(seen.insert(u64::from(previous))); + + for _ in 1..sample_size { + let current = SubmissionId::new(); + prop_assert!(current > previous); + prop_assert!(seen.insert(u64::from(current))); + previous = current; + } + } + + #[test] + fn prop_submission_id_try_from_u64_boundary_behavior( + timestamp in timestamp_strategy(), + sequence in sequence_strategy(), + ) { + let packed = pack_submission_snowflake(timestamp, SUBMISSION_SNOWFLAKE_INSTANCE, sequence); + let should_fit_u63 = packed <= u64::from(u63::MAX); + let parsed = SubmissionId::try_from(packed); + + prop_assert_eq!(parsed.is_ok(), should_fit_u63); + if let Ok(parsed) = parsed { + prop_assert_eq!(u64::from(parsed), packed); + } + } + } +} + /// A submission is a group of [`Chunk`]s. /// /// This struct represents a submission that has not yet reached an end state.