fix(genesis): map consume race to GenesisChallengeError

src/modulr_core/genesis/challenge.py

@@ -118,7 +118,12 @@ def verify_genesis_challenge_signature(
 
 
 class GenesisChallengeService:
-    """Issue and verify one-shot genesis challenges (SQLite-backed)."""
+    """
+    Issue and verify one-shot genesis challenges stored in SQLite.
+
+    Callers should use one shared connection (or equivalent isolation) per request
+    so issue/verify and consume see consistent rows.
+    """
 
     def __init__(
         self,
@@ -127,11 +132,30 @@ def __init__(
         challenge_repo: GenesisChallengeRepository,
         clock: Callable[[], int],
     ) -> None:
+        """
+        Args:
+            genesis_repo: Singleton ``core_genesis`` row (instance id, completion).
+            challenge_repo: ``genesis_challenge`` rows.
+            clock: Callable returning current Unix seconds (inject for tests).
+        """
         self._genesis = genesis_repo
         self._challenges = challenge_repo
         self._clock = clock
 
     def issue(self, *, subject_signing_pubkey_hex: str) -> IssuedGenesisChallenge:
+        """
+        Create a new challenge bound to the operator-submitted signing pubkey.
+
+        Args:
+            subject_signing_pubkey_hex: Lowercase 64-hex Ed25519 public key to bind.
+
+        Returns:
+            Challenge id, canonical body (UTF-8 text to sign), and expiry metadata.
+
+        Raises:
+            GenesisChallengeError: If genesis is already complete or inputs are
+                invalid (via body builder).
+        """
         snap = self._genesis.get()
         if snap.genesis_complete:
             raise GenesisChallengeError("genesis already complete")
@@ -162,6 +186,18 @@ def issue(self, *, subject_signing_pubkey_hex: str) -> IssuedGenesisChallenge:
         )
 
     def verify_and_consume(self, *, challenge_id: str, signature_hex: str) -> None:
+        """
+        Verify the Ed25519 signature and mark the challenge consumed (one shot).
+
+        Args:
+            challenge_id: 64 lowercase hex challenge id (same as body ``nonce``).
+            signature_hex: 128 lowercase hex Ed25519 signature over ``UTF-8(body)``.
+
+        Raises:
+            GenesisChallengeError: Unknown id, already consumed, expired, bad
+                signature, genesis already complete, or another request consumed
+                the row after verification (concurrent duplicate submit / race).
+        """
         snap = self._genesis.get()
         if snap.genesis_complete:
             raise GenesisChallengeError("genesis already complete")
@@ -178,5 +214,6 @@ def verify_and_consume(self, *, challenge_id: str, signature_hex: str) -> None:
             signature_hex=signature_hex,
             expected_subject_pubkey_hex=row.subject_signing_pubkey_hex,
         )
-        self._challenges.mark_consumed(challenge_id, consumed_at=now)
+        if not self._challenges.mark_consumed(challenge_id, consumed_at=now):
+            raise GenesisChallengeError("challenge already consumed")
         self._genesis.touch(updated_at=now)

src/modulr_core/repositories/genesis_challenge.py

@@ -20,7 +20,7 @@ class GenesisChallengeRow:
 
 
 class GenesisChallengeRepository:
-    """CRUD for ``genesis_challenge`` (singleton row table, keyed by challenge_id)."""
+    """Persistence for ``genesis_challenge`` (one row per ``challenge_id``)."""
 
     def __init__(self, conn: sqlite3.Connection) -> None:
         self._conn = conn
@@ -67,16 +67,27 @@ def get_by_id(self, challenge_id: str) -> GenesisChallengeRow | None:
             return None
         return _row_to_challenge(row)
 
-    def mark_consumed(self, challenge_id: str, *, consumed_at: int) -> None:
+    def mark_consumed(self, challenge_id: str, *, consumed_at: int) -> bool:
+        """
+        Set ``consumed_at`` on a pending challenge (conditional single-use consume).
+
+        Args:
+            challenge_id: Primary key of the challenge row.
+            consumed_at: Unix seconds at consume time.
+
+        Returns:
+            True if exactly one row was updated. False if the row was missing or
+            ``consumed_at`` was already set (including lost races between readers
+            and this update).
+        """
         cur = self._conn.execute(
             """
             UPDATE genesis_challenge SET consumed_at = ?
             WHERE challenge_id = ? AND consumed_at IS NULL
             """,
             (consumed_at, challenge_id),
         )
-        if cur.rowcount != 1:
-            raise RuntimeError("challenge not found or already consumed")
+        return cur.rowcount == 1
 
 
 def _row_to_challenge(row: sqlite3.Row | tuple[Any, ...]) -> GenesisChallengeRow:

tests/test_genesis_challenge.py

@@ -178,3 +178,47 @@ def test_get_or_create_instance_id_stable() -> None:
     b = repo.get_or_create_instance_id(updated_at=20)
     assert a == b
     assert repo.get().instance_id == a
+
+
+def test_mark_consumed_returns_false_on_second_call() -> None:
+    """Second consume on the same row is a no-op and returns False."""
+    conn = _conn()
+    _, pk = _test_keypair()
+    repo = GenesisChallengeRepository(conn)
+    cid = "c" * 64
+    repo.insert(
+        challenge_id=cid,
+        subject_signing_pubkey_hex=pk,
+        body="x",
+        issued_at=1,
+        expires_at=400,
+    )
+    assert repo.mark_consumed(cid, consumed_at=2) is True
+    assert repo.mark_consumed(cid, consumed_at=3) is False
+
+
+def test_verify_and_consume_failed_mark_maps_to_already_consumed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    If the conditional consume updates zero rows after a valid signature,
+    surface ``challenge already consumed`` (e.g. concurrent verify won).
+    """
+    conn = _conn()
+    g_repo = CoreGenesisRepository(conn)
+    c_repo = GenesisChallengeRepository(conn)
+    priv, pk = _test_keypair()
+    svc = GenesisChallengeService(
+        genesis_repo=g_repo,
+        challenge_repo=c_repo,
+        clock=lambda: 1_000,
+    )
+    issued = svc.issue(subject_signing_pubkey_hex=pk)
+    sig = priv.sign(issued.body.encode("utf-8")).hex()
+
+    def _no_op_mark(_cid: str, *, consumed_at: int) -> bool:
+        return False
+
+    monkeypatch.setattr(c_repo, "mark_consumed", _no_op_mark)
+    with pytest.raises(GenesisChallengeError, match="challenge already consumed"):
+        svc.verify_and_consume(challenge_id=issued.challenge_id, signature_hex=sig)