test(ecstore): verify multi-generation authority model (#8339)

scripts/check_test_wiring.py

@@ -1072,7 +1072,10 @@ def check_quick_checks(root: Path) -> list[str]:
         errors.append(f"{relative}: missing composite action")
         return errors
     steps = yaml_block(runs, "steps", 2) or []
-    for command in ("shellcheck --version && actionlint", "./scripts/check_error_other_format_ratchet.sh", "make script-tests"):
+    for command in (
+        "shellcheck --version && actionlint", "./scripts/check_error_other_format_ratchet.sh", "make script-tests",
+        "python3 ./scripts/test_object_generation_protocol_model.py",
+    ):
         step = workflow_step_block(steps, command, key="run", indent=4)
         if step is None:
             errors.append(f"{relative}: missing direct execution of {command}")
@@ -3517,6 +3520,8 @@ class SelfTests(unittest.TestCase):
                 "    - name: Error format ratchet\n      shell: bash\n"
                 "      run: ./scripts/check_error_other_format_ratchet.sh\n"
                 "    - name: Script tests\n      shell: bash\n      run: make script-tests\n"
+                "    - name: Object generation model\n      shell: bash\n"
+                "      run: python3 ./scripts/test_object_generation_protocol_model.py\n"
             )
             sources = {
                 ".github/workflows/ci.yml": caller.replace(
@@ -3576,6 +3581,11 @@ class SelfTests(unittest.TestCase):
                 "missing ratchet": action.replace("run: ./scripts/check_error_other_format_ratchet.sh", "run: echo skipped"),
                 "missing script tests": action.replace("run: make script-tests", "run: echo skipped"),
                 "swallowed script failure": action.replace("run: make script-tests", "run: make script-tests || true"),
+                "missing generation model": action.replace("run: python3 ./scripts/test_object_generation_protocol_model.py", "run: echo skipped"),
+                "swallowed generation model failure": action.replace(
+                    "run: python3 ./scripts/test_object_generation_protocol_model.py",
+                    "run: python3 ./scripts/test_object_generation_protocol_model.py || true",
+                ),
                 "swallowed lint failure": action.replace("&& actionlint", "&& actionlint || true"),
                 "swallowed ratchet failure": action.replace("ratchet.sh", "ratchet.sh || true"),
                 "conditional lint": action.replace("run: shellcheck", "if: false\n      run: shellcheck"),
@@ -3585,7 +3595,10 @@ class SelfTests(unittest.TestCase):
                     "name: Lint workflows", "name: |\n        run: shellcheck --version && actionlint"
                 ).replace("\n      run: shellcheck --version && actionlint\n", "\n      run: shellcheck --version && actionlint\n        || true\n"),
             }
-            for command in ("shellcheck --version && actionlint", "./scripts/check_error_other_format_ratchet.sh", "make script-tests"):
+            for command in (
+                "shellcheck --version && actionlint", "./scripts/check_error_other_format_ratchet.sh", "make script-tests",
+                "python3 ./scripts/test_object_generation_protocol_model.py",
+            ):
                 for key in ("'if' : false", '"if": false', "'continue-on-error': true", '"continue-on-error" : true'):
                     mutations[f"quoted {command} {key}"] = action.replace(f"run: {command}", f"{key}\n      run: {command}")
                 for separator in ("", "\n", "        # continued command\n"):
@@ -3606,8 +3619,9 @@ class SelfTests(unittest.TestCase):
             root = Path(tmp)
             (root / "scripts").mkdir()
             commands = ("shellcheck", "actionlint", "./scripts/check_error_other_format_ratchet.sh")
+            model_command = "python3 ./scripts/test_object_generation_protocol_model.py"
             (root / "Makefile").write_text(".PHONY: script-tests\nscript-tests:\n\texit 17\n")
-            for failing in (*commands, "make script-tests"):
+            for failing in (*commands, "make script-tests", model_command):
                 with self.subTest(command=failing):
                     run = "shellcheck --version && actionlint" if failing in ("shellcheck", "actionlint") else failing
                     step = workflow_step_block(steps, run, key="run", indent=4)
@@ -3619,6 +3633,9 @@ class SelfTests(unittest.TestCase):
                         shim = root / command
                         shim.write_text(f"#!/bin/sh\nexit {17 if command == failing else 0}\n")
                         shim.chmod(0o755)
+                    (root / "scripts/test_object_generation_protocol_model.py").write_text(
+                        f"raise SystemExit({17 if failing == model_command else 0})\n"
+                    )
                     result = subprocess.run(
                         ["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", body],
                         cwd=root, env=dict(os.environ, PATH=f"{root}{os.pathsep}{os.environ['PATH']}"),

scripts/test_object_generation_protocol_model.py

@@ -6,12 +6,16 @@
 Run with ``python3 scripts/test_object_generation_protocol_model.py``. The
 exhaustive check covers one successor slot, four durable voters, two distinct
 ballots/candidates, and every enabled prepare/accept ordering in that bound.
+Directed round schedules also cover successor recovery, tombstone continuity,
+and proposer counter allocation; they do not exhaust multi-slot interleavings.
 Durable transitions are atomic; filesystem crash atomicity is an implementation
 obligation, not something this model can establish.
 """
 
 from collections import deque
 from dataclasses import dataclass, replace
+from enum import Enum
+from itertools import combinations, product
 import unittest
 
 
@@ -140,6 +144,9 @@ def choose(voters, ballot, candidate, prepare_ids, accept_ids, adopt=True):
         return tuple(voters), None, False
     accepted = max(replies)
     value = accepted[1] if adopt and accepted[0] else candidate
+    if value == EMPTY:
+        # A recovery barrier with no accepted value does not choose an empty head.
+        return tuple(voters), None, False
     count = 0
     for disk in accept_ids:
         voters[disk], success = voters[disk].accept(ballot, value)
@@ -147,6 +154,96 @@ def choose(voters, ballot, candidate, prepare_ids, accept_ids, adopt=True):
     return tuple(voters), value, count >= QUORUM
 
 
+class Presence(Enum):
+    NEVER_PRESENT = "never-present"
+    LIVE = "live"
+    TOMBSTONE = "tombstone"
+
+
+@dataclass(frozen=True)
+class Head:
+    revision: int = 0
+    presence: Presence = Presence.NEVER_PRESENT
+    operation: str = "bootstrap"
+
+
+@dataclass(frozen=True)
+class Decision:
+    predecessor: Head
+    successor: Head
+
+
+@dataclass(frozen=True)
+class AuthorityHistory:
+    slots: tuple = ()
+    decisions: tuple = ()
+    learned: Head = Head()
+
+    def append(
+        self, expected, operation, presence, ballot, prepare_ids, accept_ids, *, learn=True,
+        recover_first=True, collapse_tombstone=False,
+    ):
+        slots = list(self.slots)
+        head = self.learned
+        if recover_first:
+            for index in range(head.revision, len(slots)):
+                # Re-propose this slot's retained preparation if the promise
+                # quorum observed no accepted value; otherwise adopt its winner.
+                slots[index], value, resolved = choose(slots[index], ballot, index, prepare_ids, accept_ids)
+                if not resolved:
+                    return replace(self, slots=tuple(slots), learned=head), None, False
+                decision = self.decisions[value]
+                if decision.predecessor != head or decision.successor.revision != index + 1:
+                    raise ValueError("recovered decisions must extend the chosen predecessor")
+                head = decision.successor
+        if collapse_tombstone and head.presence is Presence.TOMBSTONE:
+            head = Head()
+        recovered = replace(self, slots=tuple(slots), learned=head)
+        for original in self.decisions[:head.revision]:
+            if original.successor.operation == operation:
+                if original.predecessor == expected and original.successor.presence is presence:
+                    return recovered, original.successor, True
+                return recovered, None, False
+        if expected != head:
+            return recovered, None, False
+        if presence is Presence.NEVER_PRESENT:
+            raise ValueError("a successor cannot restore never-present authority")
+        successor = Head(head.revision + 1, presence, operation)
+        decisions = self.decisions + (Decision(head, successor),)
+        slots.append((Voter(),) * VOTERS)
+        slots[-1], value, chosen = choose(slots[-1], ballot, len(decisions) - 1, prepare_ids, accept_ids)
+        learned = successor if chosen and learn else head
+        updated = AuthorityHistory(tuple(slots), decisions, learned)
+        return updated, decisions[value].successor if chosen else None, chosen
+
+
+@dataclass(frozen=True)
+class ProposerSession:
+    # Only this counter survives a restart; reservation is process-local.
+    durable_counter: int = 0
+    reservation: int = 0
+
+    def reserve(self, observed_counter=0):
+        if self.reservation:
+            raise ValueError("a ballot allocation is already pending")
+        return replace(self, reservation=max(self.durable_counter, observed_counter) + 1)
+
+    def persist(self):
+        if not self.reservation:
+            raise ValueError("no ballot allocation to persist")
+        return replace(self, durable_counter=self.reservation)
+
+    def issue(self, *, require_persistence=True):
+        if not self.reservation:
+            raise ValueError("no ballot allocation to issue")
+        if require_persistence and self.durable_counter != self.reservation:
+            raise ValueError("persist the proposer counter before using a ballot")
+        return replace(self, reservation=0), self.reservation
+
+    def restart(self):
+        return ProposerSession(self.durable_counter)
+
+
 @dataclass(frozen=True)
 class Disk:
     applied_revision: int = 0
@@ -262,6 +359,172 @@ class ObjectGenerationProtocolModel(unittest.TestCase):
             self.assertEqual(current.applied_revision, revision)
         self.assertEqual(current.publish_chosen(0, "bootstrap", set()), (current, False))
 
+    def test_unlearned_predecessor_is_recovered_before_the_second_slot(self):
+        quorums = tuple(combinations(range(VOTERS), QUORUM))
+        first_accepts = quorums + tuple(combinations(range(VOTERS), QUORUM - 1))
+        schedules = 0
+        for prepare_a, accept_a, prepare_b, accept_b in product(quorums, first_accepts, quorums, quorums):
+            with self.subTest(prepare_a=prepare_a, accept_a=accept_a, prepare_b=prepare_b, accept_b=accept_b):
+                pending, _, chosen = AuthorityHistory().append(
+                    Head(), "A", Presence.LIVE, 1, prepare_a, accept_a, learn=False)
+                self.assertEqual(chosen, len(accept_a) >= QUORUM)
+                self.assertEqual(pending.learned, Head(), "the decision notification has not arrived")
+                self.assertEqual(sum(v.accepted_ballot == 1 for v in pending.slots[0]), len(accept_a))
+                recovered, head, chosen = pending.append(
+                    Head(1, Presence.LIVE, "A"), "B", Presence.LIVE, 2, prepare_b, accept_b)
+                self.assertTrue(chosen)
+                self.assertEqual(head, Head(2, Presence.LIVE, "B"))
+                self.assertEqual(recovered.decisions[1].predecessor, recovered.decisions[0].successor)
+                self.assertEqual({v.accepted_value for v in recovered.slots[0] if v.accepted_ballot == 2}, {0})
+                schedules += 1
+        print(f"multi-slot round schedules: {schedules}")
+
+    def test_recovery_failure_or_stale_expected_head_does_not_allocate_a_successor(self):
+        quorum = (0, 1, 2)
+        pending, _, chosen = AuthorityHistory().append(Head(), "A", Presence.LIVE, 1, quorum, quorum, learn=False)
+        self.assertTrue(chosen)
+        blocked, head, chosen = pending.append(Head(1, Presence.LIVE, "A"), "B", Presence.LIVE, 2, (0, 1), quorum)
+        self.assertFalse(chosen)
+        self.assertIsNone(head)
+        self.assertEqual(blocked.decisions, pending.decisions)
+        self.assertEqual(len(blocked.slots), 1)
+        recovered, _, chosen = blocked.append(Head(), "B", Presence.LIVE, 3, quorum, quorum)
+        self.assertFalse(chosen, "a stale expected predecessor is rejected after recovery")
+        self.assertEqual(recovered.learned, Head(1, Presence.LIVE, "A"))
+        self.assertEqual(recovered.decisions, pending.decisions)
+        self.assertEqual(len(recovered.slots), 1)
+
+    def test_empty_or_unobserved_minority_slot_can_retry_without_allocating_another_revision(self):
+        quorum = (0, 1, 2)
+        cases = (((0, 1), quorum, quorum), (quorum, (), quorum), (quorum, (0,), (1, 2, 3)))
+        for prepare_ids, accept_ids, recovery_ids in cases:
+            with self.subTest(prepare=prepare_ids, accept=accept_ids, recovery=recovery_ids):
+                pending, _, chosen = AuthorityHistory().append(
+                    Head(), "A", Presence.LIVE, 1, prepare_ids, accept_ids)
+                self.assertFalse(chosen)
+                self.assertTrue(all(pending.slots[0][i].accepted_ballot == 0 for i in recovery_ids))
+                retried, head, chosen = pending.append(Head(), "A", Presence.LIVE, 2, recovery_ids, recovery_ids)
+                self.assertTrue(chosen, "the quorum can choose the retained candidate in the original slot")
+                self.assertEqual(head, Head(1, Presence.LIVE, "A"))
+                self.assertEqual(len(retried.slots), 1)
+                self.assertEqual(retried.decisions, pending.decisions)
+                self.assertTrue(all(retried.slots[0][i].accepted_value == 0 for i in recovery_ids))
+                _, successor, chosen = retried.append(head, "B", Presence.LIVE, 3, quorum, quorum)
+                self.assertTrue(chosen)
+                self.assertEqual(successor.revision, 2)
+
+    def test_superseded_operation_retry_returns_its_original_successor(self):
+        quorum = (0, 1, 2)
+        history, head_a, chosen = AuthorityHistory().append(Head(), "A", Presence.LIVE, 1, quorum, quorum)
+        self.assertTrue(chosen)
+        history, head_b, chosen = history.append(head_a, "B", Presence.LIVE, 2, quorum, quorum)
+        self.assertTrue(chosen)
+        retried, result, chosen = history.append(Head(), "A", Presence.LIVE, 3, quorum, quorum)
+        self.assertTrue(chosen)
+        self.assertEqual(result, head_a)
+        self.assertEqual(retried, history, "the historical outcome cannot replace B or allocate another slot")
+        self.assertEqual(retried.learned, head_b)
+
+    def test_superseded_operation_identity_cannot_be_reused_for_different_input(self):
+        quorum = (0, 1, 2)
+        history, head_a, chosen = AuthorityHistory().append(Head(), "A", Presence.LIVE, 1, quorum, quorum)
+        self.assertTrue(chosen)
+        history, head_b, chosen = history.append(head_a, "B", Presence.LIVE, 2, quorum, quorum)
+        self.assertTrue(chosen)
+        for expected, presence in ((head_b, Presence.LIVE), (Head(), Presence.TOMBSTONE)):
+            with self.subTest(expected=expected, presence=presence):
+                rejected, result, chosen = history.append(expected, "A", presence, 3, quorum, quorum)
+                self.assertFalse(chosen)
+                self.assertIsNone(result)
+                self.assertEqual(rejected, history)
+
+    def test_skipping_predecessor_recovery_chooses_a_broken_revision_chain(self):
+        quorum = (0, 1, 2)
+        pending, _, chosen = AuthorityHistory().append(Head(), "A", Presence.LIVE, 1, quorum, quorum, learn=False)
+        self.assertTrue(chosen)
+        broken, head, chosen = pending.append(Head(), "B", Presence.LIVE, 2, quorum, quorum, recover_first=False)
+        self.assertTrue(chosen, "the next slot alone cannot validate the predecessor")
+        self.assertEqual(head.revision, 1, "the cached head caused a repeated object revision")
+        self.assertNotEqual(broken.decisions[1].predecessor, broken.decisions[0].successor)
+        with self.assertRaisesRegex(ValueError, "chosen predecessor"):
+            replace(broken, learned=Head()).append(head, "C", Presence.LIVE, 3, quorum, quorum)
+
+    def test_a_recovery_barrier_cannot_choose_an_empty_value(self):
+        voters, value, chosen = choose((Voter(),) * VOTERS, 1, EMPTY, (0, 1, 2), (0, 1, 2))
+        self.assertFalse(chosen)
+        self.assertIsNone(value)
+        self.assertTrue(all(v.accepted_ballot == 0 for v in voters))
+
+    def test_tombstone_rejects_a_delayed_never_present_create(self):
+        quorum = (0, 1, 2)
+        history, live, chosen = AuthorityHistory().append(Head(), "create", Presence.LIVE, 1, quorum, quorum)
+        self.assertTrue(chosen)
+        deleted, tombstone, chosen = history.append(live, "delete", Presence.TOMBSTONE, 2, quorum, quorum)
+        self.assertTrue(chosen)
+        self.assertEqual(tombstone, Head(2, Presence.TOMBSTONE, "delete"))
+        rejected, head, chosen = deleted.append(Head(), "stale-create", Presence.LIVE, 3, quorum, quorum)
+        self.assertFalse(chosen)
+        self.assertIsNone(head)
+        self.assertEqual(rejected, deleted)
+        recreated, head, chosen = rejected.append(tombstone, "recreate", Presence.LIVE, 4, quorum, quorum)
+        self.assertTrue(chosen)
+        self.assertEqual(head.revision, 3)
+        self.assertEqual(recreated.decisions[2].predecessor, tombstone)
+        with self.assertRaisesRegex(ValueError, "never-present"):
+            recreated.append(head, "reset", Presence.NEVER_PRESENT, 5, quorum, quorum)
+
+    def test_forgetting_tombstone_authority_allows_an_aba_create(self):
+        quorum = (0, 1, 2)
+        history, live, _ = AuthorityHistory().append(Head(), "create", Presence.LIVE, 1, quorum, quorum)
+        deleted, tombstone, chosen = history.append(live, "delete", Presence.TOMBSTONE, 2, quorum, quorum)
+        self.assertTrue(chosen)
+        broken, resurrected, chosen = deleted.append(
+            Head(), "stale-create", Presence.LIVE, 3, quorum, quorum, collapse_tombstone=True)
+        self.assertTrue(chosen, "treating the deleted head as absent admits the delayed create")
+        self.assertEqual(resurrected.revision, 1)
+        self.assertNotEqual(broken.decisions[2].predecessor, tombstone)
+
+    def test_counter_allocation_requires_persistence_and_is_consumed_once(self):
+        with self.assertRaisesRegex(ValueError, "no ballot allocation"):
+            ProposerSession().persist()
+        reserved = ProposerSession().reserve(observed_counter=7)
+        with self.assertRaisesRegex(ValueError, "already pending"):
+            reserved.reserve()
+        with self.assertRaisesRegex(ValueError, "persist the proposer counter"):
+            reserved.issue()
+        issued, ballot = reserved.persist().issue()
+        self.assertEqual(ballot, 8)
+        with self.assertRaisesRegex(ValueError, "no ballot allocation"):
+            issued.issue()
+        with self.assertRaisesRegex(ValueError, "no ballot allocation"):
+            issued.restart().issue()
+        _, next_ballot = issued.restart().reserve().persist().issue()
+        self.assertGreater(next_ballot, ballot)
+
+    def test_counter_restart_burns_persisted_allocations_before_or_after_issue(self):
+        for crash_after in ("reserve", "persist", "issue"):
+            with self.subTest(crash_after=crash_after):
+                session = ProposerSession().reserve()
+                if crash_after != "reserve":
+                    session = session.persist()
+                if crash_after == "issue":
+                    session, ballot = session.issue()
+                    self.assertEqual(ballot, 1)
+                _, next_ballot = session.restart().reserve().persist().issue()
+                self.assertEqual(next_ballot, 1 if crash_after == "reserve" else 2)
+
+    def test_issuing_before_persistence_reuses_a_ballot_after_restart(self):
+        crashed, first_ballot = ProposerSession().reserve().issue(require_persistence=False)
+        quorum = (0, 1, 2)
+        voters, original, chosen = choose((Voter(),) * VOTERS, first_ballot, 0, quorum, quorum)
+        self.assertTrue(chosen)
+        _, reused_ballot = crashed.restart().reserve().persist().issue()
+        self.assertEqual(reused_ballot, first_ballot, "the emitted ballot was not reserved durably")
+        self.assertEqual(voters[0].accept(reused_ballot, 1), (voters[0], False))
+        _, recovered, chosen = choose(voters, reused_ballot, 1, quorum, quorum)
+        self.assertTrue(chosen)
+        self.assertEqual(recovered, original, "a new operation cannot use the colliding ballot")
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)