Implement same ms monotonic sort order

Close #1

Author: mdomke

CHANGELOG.rst

@@ -10,7 +10,10 @@ Versions follow `Semantic Versioning <http://www.semver.org>`_
 Changed
 ~~~~~~~
 * Added Python 3.13 to the trove classifiers.
-* Optimze ``@property`` to ``@cached_property`` `<@WH-2099 <https://github.com/WH-2099>`_.
+* Optimze ``@property`` to ``@cached_property`` `@WH-2099 <https://github.com/WH-2099>`_.
+* When generating ULIDs within the same millisecond, the library will ensure monotonic sort order by
+  incrementing the randomness component by 1 bit. This process is descrbied in the
+  `spec <https://github.com/ulid/spec/blob/master/README.md#monotonicity>`_.
 
 Fixed
 ~~~~~

tests/test_ulid.py

@@ -61,6 +61,19 @@ def test_ulid_monotonic_sorting(tick: int) -> None:
     assert_sorted([v.bytes for v in ulids])
 
 
+@freeze_time()
+def test_same_millisecond_monotonic_sorting() -> None:
+    ulids = [ULID() for _ in range(1000)]
+    assert_sorted(ulids)
+
+
+@freeze_time()
+def test_same_millisecond_overflow() -> None:
+    ULID.provider.prev_randomness = constants.MAX_RANDOMNESS
+    with pytest.raises(ValueError, match="Randomness within same millisecond exhausted"):
+        ULID()
+
+
 def assert_sorted(seq: list) -> None:
     last = seq[0]
     for item in seq[1:]:
@@ -154,6 +167,11 @@ def test_ulid_from_timestamp() -> None:
     assert ulid1.timestamp == ulid2.timestamp
 
 
+def test_ulid_from_timestamp_overflow() -> None:
+    with pytest.raises(ValueError, match="Value exceeds maximum possible timestamp"):
+        ULID.from_timestamp(constants.MAX_TIMESTAMP + 1)
+
+
 Params = Union[bytes, str, int, float]
 
 

ulid/__init__.py

@@ -6,6 +6,7 @@
 import uuid
 from datetime import datetime
 from datetime import timezone
+from threading import Lock
 from typing import Any
 from typing import cast
 from typing import Generic
@@ -53,8 +54,42 @@ def wrapped(cls: Any, value: T) -> R:
         return wrapped
 
 
+class ValueProvider:
+    def __init__(self) -> None:
+        self.lock = Lock()
+        self.prev_timestamp = constants.MIN_TIMESTAMP
+        self.prev_randomness = constants.MIN_RANDOMNESS
+
+    def timestamp(self, value: float | None = None) -> int:
+        if value is None:
+            value = time.time_ns() // constants.NANOSECS_IN_MILLISECS
+        elif isinstance(value, float):
+            value = int(value * constants.MILLISECS_IN_SECS)
+        if value > constants.MAX_TIMESTAMP:
+            raise ValueError("Value exceeds maximum possible timestamp")
+        return value
+
+    def randomness(self) -> bytes:
+        with self.lock:
+            current_timestamp = self.timestamp()
+            if current_timestamp == self.prev_timestamp:
+                if self.prev_randomness == constants.MAX_RANDOMNESS:
+                    raise ValueError("Randomness within same millisecond exhausted")
+                randomness = (int.from_bytes(self.prev_randomness) + 1).to_bytes(
+                    constants.RANDOMNESS_LEN, byteorder="big"
+                )
+            else:
+                randomness = os.urandom(constants.RANDOMNESS_LEN)
+
+            self.prev_randomness = randomness
+            self.prev_timestamp = current_timestamp
+        return randomness
+
+
 @functools.total_ordering
 class ULID:
+    provider = ValueProvider()
+
     """The :class:`ULID` object consists of a timestamp part of 48 bits and of 80 random bits.
 
     .. code-block:: text
@@ -82,9 +117,7 @@ class ULID:
     def __init__(self, value: bytes | None = None) -> None:
         if value is not None and len(value) != constants.BYTES_LEN:
             raise ValueError("ULID has to be exactly 16 bytes long.")
-        self.bytes: bytes = (
-            value or ULID.from_timestamp(time.time_ns() // constants.NANOSECS_IN_MILLISECS).bytes
-        )
+        self.bytes: bytes = value or ULID.from_timestamp(self.provider.timestamp()).bytes
 
     @classmethod
     @validate_type(datetime)
@@ -113,10 +146,8 @@ def from_timestamp(cls, value: float) -> Self:
             >>> ULID.from_timestamp(time.time())
             ULID(01E75QWN5HKQ0JAVX9FG1K4YP4)
         """
-        if isinstance(value, float):
-            value = int(value * constants.MILLISECS_IN_SECS)
-        timestamp = int.to_bytes(value, constants.TIMESTAMP_LEN, "big")
-        randomness = os.urandom(constants.RANDOMNESS_LEN)
+        timestamp = int.to_bytes(cls.provider.timestamp(value), constants.TIMESTAMP_LEN, "big")
+        randomness = cls.provider.randomness()
         return cls.from_bytes(timestamp + randomness)
 
     @classmethod

ulid/constants.py

@@ -1,6 +1,12 @@
 MILLISECS_IN_SECS = 1000
 NANOSECS_IN_MILLISECS = 1000000
 
+MIN_RANDOMNESS = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+MAX_RANDOMNESS = b"\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff"
+
+MIN_TIMESTAMP = 0
+MAX_TIMESTAMP = 2**48 - 1
+
 TIMESTAMP_LEN = 6
 RANDOMNESS_LEN = 10
 BYTES_LEN = TIMESTAMP_LEN + RANDOMNESS_LEN