From 15b65843b6c0b7a385cc765bcb58e5918d34aeb3 Mon Sep 17 00:00:00 2001
From: Vincent Gao <gaobing1230@gmail.com>
Date: Mon, 29 Jun 2026 09:22:44 +0200
Subject: [PATCH] fix: prevent non-monotonic ULIDs from timestamp/randomness
 clock skew

from_timestamp() sampled the clock twice: once for the timestamp
value and again inside randomness() to decide whether to increment
or generate fresh entropy. When a millisecond boundary fell between
those two samples, the encoded ULID contained the old timestamp but
fresh (potentially smaller) randomness, breaking lexicographic sort
order within the same millisecond.

Pass the resolved timestamp from from_timestamp() into randomness()
so both the timestamp bytes and the monotonicity decision use the
same clock sample. The randomness() parameter is optional so any
existing callers outside from_timestamp() continue to work unchanged.

Fixes #56
Fixes #45
---
 tests/test_ulid.py | 13 +++++++++++++
 ulid/__init__.py   | 10 ++++++----
 2 files changed, 19 insertions(+), 4 deletions(-)

diff --git a/tests/test_ulid.py b/tests/test_ulid.py
index 11ae933..4101f2e 100644
--- a/tests/test_ulid.py
+++ b/tests/test_ulid.py
@@ -67,8 +67,21 @@ def test_same_millisecond_monotonic_sorting() -> None:
     assert_sorted(ulids)
 
 
+def test_z_real_time_monotonic_sorting() -> None:
+    """ULIDs generated in rapid succession must be monotonically increasing.
+
+    This catches the bug where ``from_timestamp()`` samples the clock twice,
+    potentially crossing a millisecond boundary between the timestamp capture
+    and the randomness generation, which could produce a fresh (smaller) random
+    value instead of incrementing the previous one.
+    """
+    ulids = [ULID() for _ in range(5000)]
+    assert_sorted(ulids)
+
+
 @freeze_time()
 def test_same_millisecond_overflow() -> None:
+    ULID.provider.prev_timestamp = ULID.provider.timestamp()
     ULID.provider.prev_randomness = constants.MAX_RANDOMNESS
     with pytest.raises(ValueError, match="Randomness within same millisecond exhausted"):
         ULID()
diff --git a/ulid/__init__.py b/ulid/__init__.py
index d327385..7f34284 100644
--- a/ulid/__init__.py
+++ b/ulid/__init__.py
@@ -69,9 +69,10 @@ def timestamp(self, value: float | None = None) -> int:
             raise ValueError("Value exceeds maximum possible timestamp")
         return value
 
-    def randomness(self) -> bytes:
+    def randomness(self, current_timestamp: int | None = None) -> bytes:
         with self.lock:
-            current_timestamp = self.timestamp()
+            if current_timestamp is None:
+                current_timestamp = self.timestamp()
             if current_timestamp == self.prev_timestamp:
                 if self.prev_randomness == constants.MAX_RANDOMNESS:
                     raise ValueError("Randomness within same millisecond exhausted")
@@ -148,8 +149,9 @@ def from_timestamp(cls, value: float) -> Self:
             >>> ULID.from_timestamp(time.time())
             ULID(01E75QWN5HKQ0JAVX9FG1K4YP4)
         """
-        timestamp = int.to_bytes(cls.provider.timestamp(value), constants.TIMESTAMP_LEN, "big")
-        randomness = cls.provider.randomness()
+        timestamp_value = cls.provider.timestamp(value)
+        timestamp = int.to_bytes(timestamp_value, constants.TIMESTAMP_LEN, "big")
+        randomness = cls.provider.randomness(timestamp_value)
         return cls.from_bytes(timestamp + randomness)
 
     @classmethod