"""Every number the commentary claims, pinned. Plain asserts, no pytest. python3 test_snowflake.py """ import datetime as dt import random from snowflake import (BLIND, BORROW, CLAMP, RAM, REFUSE, RESUME, SKIP, Exhausted, Snowflake, Spin, Wall, run) RATE, DUR, STEP_AT, STEP_MS = 8, 200, 100.0, 50 TRACE = dict(rate=RATE, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS) tests = [] def test(fn): tests.append(fn) return fn # ---- the 64 bits --------------------------------------------------------- @test def pack_and_unpack_round_trip(): for mid_bits, seq_bits in ((10, 12), (9, 13), (11, 11), (8, 14)): g = Snowflake(machine_id=5, mid_bits=mid_bits, seq_bits=seq_bits) for ts, seq in ((0, 0), (1, 1), (1 << 20, (1 << seq_bits) - 1)): assert g.unpack(g.pack(ts, seq)) == (ts, 5, seq) @test def an_id_is_a_positive_int64(): g = Snowflake(machine_id=1023) i = g.pack((1 << 41) - 1, 4095) assert i == (1 << 63) - 1 and i.bit_length() == 63 @test def the_machine_field_is_masked_not_checked(): """1025 & 1023 == 1, so the 1025th machine is silently the 1st.""" one, alias = Snowflake(machine_id=1), Snowflake(machine_id=1025) assert one.next_id(500)[0] == alias.next_id(500)[0] == 2097156096 assert 1025 & 1023 == 1 @test def the_sequence_counts_within_a_millisecond_and_resets_across_one(): g = Snowflake(machine_id=7) assert [g.unpack(g.next_id(10)[0])[2] for _ in range(4)] == [0, 1, 2, 3] assert g.unpack(g.next_id(11)[0])[2] == 0 @test def the_4097th_id_in_a_millisecond_has_nowhere_to_go(): spin = Snowflake(machine_id=7) for _ in range(4096): spin.next_id(10) try: spin.next_id(10) raise AssertionError("expected Spin") except Spin as e: assert e.until == 11 assert spin.next_id(11)[1] == 11 # the retry lands in the next ms lender = Snowflake(machine_id=7, overflow=BORROW) for _ in range(4096): lender.next_id(10) assert lender.next_id(10)[1] == 11 # a ms that has not happened yet assert lender.borrowed == 1 @test def forty_one_bits_of_milliseconds_runs_out_in_2080(): span = 1 << 41 assert round(span / 1000 / 86400 / 365.2425, 2) == 69.68 epoch = 1288834974657 # Twitter's, 2010-11-04 died = dt.datetime.fromtimestamp((epoch + span) / 1000, dt.UTC) assert died.strftime("%Y-%m-%d") == "2080-07-10" # ---- the clock goes backwards ------------------------------------------- @test def blind_reissues_exactly_rate_times_rewind(): r = run(BLIND, **TRACE) assert (r.issued, r.refused, r.dupes) == (1600, 0, 400) assert r.dupes == RATE * STEP_MS @test def the_first_collision_is_the_first_millisecond_of_the_rewound_window(): r = run(BLIND, **TRACE) first, again, ident = r.collisions()[0] assert (first, again, ident) == (50.0, 100.0, 209743872) assert r.gen.unpack(ident) == (50, 7, 0) @test def refuse_buys_zero_duplicates_with_392_refusals(): r = run(REFUSE, **TRACE) assert (r.issued, r.refused, r.dupes) == (1208, 392, 0) assert r.issued + r.refused == RATE * DUR assert r.refused == RATE * (STEP_MS - 1) # the last ms of the window is == for step in (2, 5, 10, 50): assert run(REFUSE, rate=RATE, duration=DUR, step_at=STEP_AT, step_ms=step).refused == RATE * (step - 1) @test def clamp_buys_zero_duplicates_with_nothing(): r = run(CLAMP, **TRACE) assert (r.issued, r.refused, r.dupes, r.stall) == (1600, 0, 0, 0.0) @test def duplicates_are_rate_times_rewind_at_every_rate_and_every_rewind(): for rate in (1, 2, 4, 8, 16): for step in (2, 5, 20, 50): assert run(BLIND, rate=rate, duration=DUR, step_at=STEP_AT, step_ms=step).dupes == rate * step @test def a_one_millisecond_rewind_is_free_even_with_no_defence_at_all(): """`now == last_ts` takes the increment branch, which is the whole defence. The formula is rate x rewind for rewind >= 2 and zero at 1: a step back that lands on the millisecond the generator is already in never reaches the `now < last_ts` branch at all. """ for rate in (1, 8, 16): one = dict(rate=rate, duration=DUR, step_at=STEP_AT, step_ms=1) assert run(BLIND, **one).dupes == 0 assert run(REFUSE, **one).refused == 0 assert run(BLIND, rate=rate, duration=DUR, step_at=STEP_AT, step_ms=2).dupes == 2 * rate # ---- the headline: the defence bought ten IDs --------------------------- @test def the_supervisor_turns_the_refusal_into_400_minus_k(): for k in (1, 2, 3, 5, 10, 25, 50, 100, 200, 390, 391, 392): r = run(REFUSE, fail_after=k, **TRACE) assert r.dupes == 400 - k, (k, r.dupes) assert r.restarts == 1 for k in (393, 400, 500): # more failures than the trace can produce r = run(REFUSE, fail_after=k, **TRACE) assert (r.restarts, r.dupes) == (0, 0) @test def refusing_produced_390_where_doing_nothing_produced_400(): assert run(REFUSE, fail_after=10, **TRACE).dupes == 390 assert run(BLIND, **TRACE).dupes == 400 @test def the_tenth_refusal_lands_where_the_arithmetic_says(): r = run(REFUSE, fail_after=10, **TRACE) t, wall = r.restart_log[0] assert (t, wall) == (STEP_AT + 9 / RATE, 51) == (101.125, 51) resume = t + 1 / RATE assert (STEP_AT + STEP_MS - resume) * RATE == 390 @test def clamp_cannot_be_restarted_by_a_liveness_probe(): """It never refuses, so no K value can fire. That is the whole defence.""" for k in (1, 3, 10, 50): r = run(CLAMP, fail_after=k, **TRACE) assert (r.restarts, r.refused, r.dupes) == (0, 0, 0) @test def a_probe_cannot_help_or_hurt_a_generator_that_never_refuses(): for k in (0, 1, 3, 10, 50): assert run(BLIND, fail_after=k, **TRACE).dupes == 400 # ---- the amnesia, and the durable fix that is still wrong ---------------- @test def a_restart_inside_the_window_costs_the_rest_of_the_window(): r = run(CLAMP, restart_at=120.0, start_rule=RAM, **TRACE) assert r.dupes == 240 == 30 * RATE # wall ms 70..99, 8 IDs each @test def the_exposure_window_is_exactly_the_rewind(): for at, expect_zero in ((99.0, True), (100.0, False), (120.0, False), (149.0, False), (150.0, True), (160.0, True)): d = run(CLAMP, restart_at=at, start_rule=RAM, **TRACE).dupes assert (d == 0) is expect_zero, (at, d) @test def persisting_the_millisecond_leaves_167_duplicates(): r = run(CLAMP, restart_at=120.0, start_rule=RESUME, **TRACE) assert r.dupes == 167 spent = RATE + 20 * RATE # seq 0..167 already used in ms 99 assert spent - 1 == 167 @test def only_skipping_the_millisecond_reaches_zero_and_only_with_clamp(): assert run(CLAMP, restart_at=120.0, start_rule=SKIP, **TRACE).dupes == 0 assert run(REFUSE, restart_at=120.0, start_rule=SKIP, **TRACE).dupes == 0 assert run(BLIND, restart_at=120.0, start_rule=SKIP, **TRACE).dupes == 400 # ---- one-line variants --------------------------------------------------- class Leq(Snowflake): """`if now < self.last_ts` in next_id, with the `<` changed to `<=`.""" def next_id(self, now): if now <= self.last_ts: if self.rewind == REFUSE: self.refusals += 1 raise Exhausted("clock not ahead of last_ts") if self.rewind == CLAMP: now = self.last_ts return Snowflake.next_id(self, now) class FreeRunning(Snowflake): """`self.seq = 0` on a new millisecond, deleted. The counter never resets.""" def next_id(self, now): if now < self.last_ts and self.rewind == CLAMP: now = self.last_ts if now == self.last_ts: self.seq = (self.seq + 1) & self.seq_mask if self.seq == 0: return self._overflow(now) else: self.last_ts = now self.seq = (self.seq + 1) & self.seq_mask return self.pack(self.last_ts, self.seq), self.last_ts @test def widening_the_comparison_is_inert_for_clamp(): assert run(CLAMP, cls=Leq, **TRACE).issued == 1600 assert run(CLAMP, cls=Leq, **TRACE).dupes == 0 @test def widening_the_comparison_costs_refuse_seven_requests_in_eight(): strict, wide = run(REFUSE, **TRACE), run(REFUSE, cls=Leq, **TRACE) assert (strict.issued, wide.issued) == (1208, 150) assert wide.refused == 1450 @test def a_free_running_sequence_survives_the_rewind_with_no_defence(): assert run(BLIND, cls=FreeRunning, **TRACE).dupes == 0 assert run(BLIND, **TRACE).dupes == 400 @test def a_free_running_sequence_is_not_safe_across_a_restart_it_is_phase_shifted(): """The replacement's counter restarts at 0; the old one was at 8t mod 4096. They collide exactly when the two phases agree, which needs the restart to sit a whole number of 4096/rate = 512 ms after the clock step. In the 200 ms headline trace that is unreachable, so the restart looks harmless. Give it a trace long enough and it is not: rewind 200 ms at t=600, and a restart at t=712 (712 - 200 = 512) reissues wall ms 512..599 in full. """ assert run(BLIND, cls=FreeRunning, restart_at=120.0, **TRACE).dupes == 0 long = dict(rate=8, duration=1000, step_at=600.0, step_ms=200) for at in (700.0, 704.0, 708.0, 716.0, 720.0): assert run(BLIND, cls=FreeRunning, restart_at=at, **long).dupes == 0 assert run(BLIND, cls=FreeRunning, restart_at=712.0, **long).dupes == 704 assert (800 - 712) * 8 == 704 # wall ms 512..599, 8 IDs each assert run(BLIND, **long).dupes == 1600 # the real generator, same trace # ---- the boundary: a rewind budget measured in sequence bits ------------- @test def eighty_ids_per_ms_is_free_and_eighty_one_stalls(): assert run(CLAMP, rate=80, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS).stall == 0.0 assert round(run(CLAMP, rate=81, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS).stall, 3) == 0.432 assert 51 * 80 <= 4096 < 51 * 81 @test def the_slot_budget_predicts_the_stall_everywhere(): """rate x (rewind + 1) <= 2^seq_bits <=> the rewind is absorbed free.""" for rate in (8, 40, 79, 80, 81, 100, 160): for step in (10, 39, 40, 50): r = run(CLAMP, rate=rate, duration=DUR, step_at=STEP_AT, step_ms=step) assert (r.stall == 0.0) is (rate * (step + 1) <= 4096), (rate, step) assert r.dupes == 0 @test def one_bit_moved_off_the_machine_id_removes_the_stall(): stalls = {} for mid_bits, seq_bits in ((11, 11), (10, 12), (9, 13), (8, 14)): r = run(CLAMP, rate=100, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS, mid_bits=mid_bits, seq_bits=seq_bits) stalls[mid_bits] = round(r.stall, 3) assert stalls == {11: 31.520, 10: 10.040, 9: 0.0, 8: 0.0} @test def borrowing_never_stalls_and_runs_the_clock_forward_instead(): r = run(CLAMP, BORROW, rate=100, duration=DUR, step_at=STEP_AT, step_ms=STEP_MS) ahead = max(r.gen.unpack(i)[0] - w for _t, w, i, o in r.records if o == "ok") assert (r.stall, r.dupes, ahead) == (0.0, 0, 49) # ---- the claim that survives --------------------------------------------- @test def ids_are_sortable_to_within_one_millisecond_and_no_further(): for machines, per_ms, ms, expect_inversions in ((1, 10, 50, False), (8, 10, 50, True)): rng = random.Random(20260803) gens = [Snowflake(machine_id=m, rewind=CLAMP, overflow=BORROW) for m in range(machines)] events = [] for msec in range(ms): for m in range(machines): for off in sorted(rng.random() for _ in range(per_ms)): events.append((msec + off, gens[m].next_id(msec)[0])) events.sort() worst, high, inverted = 0.0, -1.0, 0 for t, _i in sorted(events, key=lambda e: e[1]): if t < high: inverted += 1 high = max(high, t) worst = max(worst, high - t) assert (inverted > 0) is expect_inversions assert worst < 1.0 # ---- determinism --------------------------------------------------------- @test def the_same_trace_gives_byte_identical_records(): assert run(CLAMP, **TRACE).records == run(CLAMP, **TRACE).records assert run(REFUSE, fail_after=10, **TRACE).records == \ run(REFUSE, fail_after=10, **TRACE).records @test def the_wall_clock_is_a_pure_function_of_the_simulation_counter(): w = Wall(step_at=100.0, step_ms=50) assert (w.read(99.9), w.read(100.0), w.read(149.9), w.read(150.0)) == \ (99, 50, 99, 100) assert w.reaches(100, 100.0) == 150.0 # the wall reads 100 again at t=150 if __name__ == "__main__": for fn in tests: fn() print(f" ok {fn.__name__}") print(f"\n{len(tests)} tests passed")