From 5cc43c0b521d97b33823da00b9bbe644c46bd5c8 Mon Sep 17 00:00:00 2001 From: Robert Clarke Date: Thu, 30 Jul 2026 13:26:49 -0400 Subject: [PATCH 1/2] Revert "Survive production build latency in payout-artifact reuse (2026-07-29 re-land) (#103)" This reverts commit cd32c13831a961d385afd65955098b790c0cd384. The re-land brownouted mainnet on 2026-07-30 (generation-churn brownout); the payout-artifact reuse line is abandoned and 1.x.x returns to the #101 (e2fb33d) behavior. Co-Authored-By: Claude Fable 5 --- lab/prism/prism_coordinator.py | 661 +++++----------------- tests/test_prism_coordinator_job_cache.py | 633 ++------------------- 2 files changed, 183 insertions(+), 1111 deletions(-) diff --git a/lab/prism/prism_coordinator.py b/lab/prism/prism_coordinator.py index f2f818e..78b4547 100644 --- a/lab/prism/prism_coordinator.py +++ b/lab/prism/prism_coordinator.py @@ -106,29 +106,13 @@ # while the underlying condition persists. Any successful install or # event-driven preparation resets the backoff. PRISM_PAYOUT_ARTIFACT_REARM_BACKOFF_CAP = 16 -# How long an armed artifact keeps serving NEW reuse decisions, measured -# from its install (prepared_monotonic), never from its anchor: the reward -# window walk itself takes multiple seconds at production share volume, so -# a bound measured from the wall-clock anchor let a slow build arm an -# artifact that was already expired on arrival -- the born-expired rebuild -# livelock behind the 2026-07-29 rollback of the first anchor-scoped -# deploy. Freshness starts when the window becomes reusable. -DEFAULT_PRISM_PAYOUT_ARTIFACT_REUSE_STALENESS_SECONDS = 10.0 -# Audit-facing ceiling on how far a served window's declared anchor may -# trail the live ledger, in wall-clock terms. Deliberately much looser than -# the reuse-staleness knob: shares stamped after an anchor deterministically -# belong to the next window (never lost; bundles declare the artifact's own -# anchor, so reused windows stay audit-reproducible), and the ceiling only -# caps worst-case payout-preview lag under pathological build latency or a -# wedged pending-commit floor. An artifact whose anchor is already past the -# ceiling at install time is born expired: discarded, and counted as a -# preparation failure for re-arm backoff pacing. -DEFAULT_PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS = 300.0 -# Bounds one routine/initial admission wait behind publication-priority -# builds. A parked request past the deadline fails fast with a visible log -# line and counter instead of silently riding the client's own much longer -# timeout; the initial-job runner retries with its normal pacing. -DEFAULT_PRISM_ROUTINE_ADMISSION_DEADLINE_SECONDS = 15.0 +# How old an armed artifact's share-snapshot anchor may grow before reuse +# stops and a re-anchored rebuild is scheduled. Validity is anchor-scoped, +# not count-scoped: shares stamped after the anchor deterministically belong +# to the next window (bundles declare the artifact's own anchor, so reused +# windows stay audit-reproducible; no share is ever lost). The bound only +# caps how far a served window may trail the live ledger. +DEFAULT_PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS = 10.0 # Upper bound for one kernel transfer from the share-window spool file into # the audit builder's stdin pipe. The kernel clamps each call to the free # pipe capacity anyway; the bound only paces cancellation checkpoints. @@ -1542,11 +1526,8 @@ class PayoutLedgerArtifact: # That declaration is also what makes reuse valid while shares keep # landing: a share stamped after this anchor deterministically belongs to # the next window (it is never lost), so the artifact stays - # audit-reproducible for as long as it serves. Reuse freshness is - # measured from prepared_monotonic (install time) -- never from this - # anchor, which predates the window walk -- while the much looser - # wall-clock anchor ceiling backstops how far a served window may trail - # the live ledger. + # audit-reproducible until _usable_payout_ledger_artifact's bounded + # anchor-staleness check retires it. snapshot_anchor_ms: int | None = None # Canonical digest of shares_json. Cached bundles built before this # artifact was armed may only keep serving re-keyed lookups when their @@ -2710,15 +2691,6 @@ class CollectionIdentityUnavailable(TemplateRefreshBlocked): """Current collection work is waiting for an authorized worker identity.""" -class JobBuildAdmissionDeadlineExceeded(TemplateRefreshBlocked): - """A routine request parked behind publication priority past its bound. - - Subclasses TemplateRefreshBlocked so callers retry with their normal - pacing; the raise site logs and counts the expiry so parked admission is - visible instead of silently riding the client's own longer timeout. - """ - - class _JobBuildFailed(RuntimeError): """Internal signal used to distinguish a skipped build from a no-op.""" @@ -2859,18 +2831,10 @@ def __init__(self) -> None: "PRISM_PAYOUT_ARTIFACT_REARM_MIN_SECONDS", DEFAULT_PRISM_PAYOUT_ARTIFACT_REARM_MIN_SECONDS, ) - self.payout_artifact_reuse_staleness_seconds = env_positive_float( - "PRISM_PAYOUT_ARTIFACT_REUSE_STALENESS_SECONDS", - DEFAULT_PRISM_PAYOUT_ARTIFACT_REUSE_STALENESS_SECONDS, - ) self.payout_artifact_max_anchor_age_seconds = env_positive_float( "PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS", DEFAULT_PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS, ) - self.routine_admission_deadline_seconds = env_positive_float( - "PRISM_ROUTINE_ADMISSION_DEADLINE_SECONDS", - DEFAULT_PRISM_ROUTINE_ADMISSION_DEADLINE_SECONDS, - ) self.template_cache_seconds = env_nonnegative_float( "PRISM_TEMPLATE_CACHE_SECONDS", self.blockpoll_seconds, @@ -4079,28 +4043,8 @@ def _ensure_job_cache_state(self) -> None: "singleflight", "deferred", "subscribed", - "admission_deadline", - ) - } - if not hasattr(self, "payout_artifact_event_counts"): - # Artifact lifecycle observability (the 2026-07-29 incident was - # invisible precisely here). Guarded by - # _payout_artifact_executor_lock. - self.payout_artifact_event_counts = { - event: 0 - for event in ( - "built", - "build_aborted", - "installed", - "refreshed", - "already_current", - "discarded", - "born_expired", - "rearm_scheduled", ) } - if not hasattr(self, "_admission_deadline_last_log_monotonic"): - self._admission_deadline_last_log_monotonic: float | None = None if not hasattr(self, "job_build_cancellation_seconds"): self.job_build_cancellation_seconds = { "sum": 0.0, @@ -4319,10 +4263,10 @@ def _build_payout_ledger_artifact( ): # The floor is held this far below now only by a wedged # writer or a leaked release; an artifact anchored there - # would already be past the audit ceiling -- born - # expired -- on arrival. Abort before paying the window - # walk: the re-arm backoff paces retries and the - # share-commit liveness watchdog owns recovery. + # would already be past the reuse staleness bound on + # arrival. Abort before paying the window walk -- the + # re-arm backoff paces retries and the share-commit + # liveness watchdog owns recovery. return None records = list( self.ledger.snapshot_at_job_issue( @@ -4369,59 +4313,29 @@ def _prepare_payout_ledger_artifact( network_difficulty: int, ) -> None: """Prepare and atomically publish an artifact for a current generation.""" - build_started = time.monotonic() artifact = self._build_payout_ledger_artifact( payout_state_generation, payout_state_generation, network_difficulty, ) - build_seconds = time.monotonic() - build_started if artifact is None: # A superseding generation, a snapshot error, an empty window, or # a pathologically old pending-commit floor discarded this # attempt; that condition can persist, and each attempt walks the # reward window under the preparation lock. Back the # fence-failure re-arm off until something arms. - self._record_payout_artifact_event("build_aborted") - self._payout_artifact_log( - "payout_artifact_build_aborted", - payout_state_generation=int(payout_state_generation), - duration_seconds=round(build_seconds, 3), - ) with self._payout_artifact_executor_lock: self._payout_artifact_rearm_backoff = min( self._payout_artifact_rearm_backoff * 2, PRISM_PAYOUT_ARTIFACT_REARM_BACKOFF_CAP, ) return - self._record_payout_artifact_event("built") - self._payout_artifact_log( - "payout_artifact_built", - payout_state_generation=int(payout_state_generation), - duration_seconds=round(build_seconds, 3), - window_shares=len(artifact.shares_json), - anchor_age_ms=( - None - if artifact.snapshot_anchor_ms is None - else now_ms() - int(artifact.snapshot_anchor_ms) - ), - ) - if not self._install_payout_ledger_artifact(artifact): - # A discarded install -- born expired after a slow walk, an - # ordering loss to a fresher window, or a generation change -- - # armed nothing. It must not collapse the re-arm interval: an - # unconditional reset here is exactly the reset-then-rewalk - # livelock that rolled back the first anchor-scoped deploy. - with self._payout_artifact_executor_lock: - self._payout_artifact_rearm_backoff = min( - self._payout_artifact_rearm_backoff * 2, - PRISM_PAYOUT_ARTIFACT_REARM_BACKOFF_CAP, - ) + self._install_payout_ledger_artifact(artifact) def _install_payout_ledger_artifact( self, artifact: PayoutLedgerArtifact, - ) -> bool: + ) -> None: """Atomically publish a prepared artifact for its own generation. Snapshot-freshness-ordered and idempotent: every snapshot is taken at @@ -4434,151 +4348,87 @@ def _install_payout_ledger_artifact( artifact, and a same-window speculative rebuild re-reads unchanged state under an equal anchor -- keep the installed generation instead of re-keying bundle lookups for nothing. - - Returns True when the window is armed (installed fresh, refreshed in - place, or confirmed already current); only that outcome resets the - re-arm backoff. A discarded install returns False and leaves the - backoff alone -- in particular a BORN-EXPIRED artifact, whose anchor - aged past the audit ceiling during its own window walk. Arming (or - crediting) such an install would collapse the re-arm interval and - rewalk the reward window continuously: the livelock behind the - 2026-07-29 rollback. """ - anchor_age_ms = ( - None - if artifact.snapshot_anchor_ms is None - else now_ms() - int(artifact.snapshot_anchor_ms) - ) - if ( - anchor_age_ms is None - or anchor_age_ms > self._payout_artifact_max_anchor_age_ms() - ): - self._record_payout_artifact_event("born_expired") - self._payout_artifact_log( - "payout_artifact_born_expired", - payout_state_generation=int(artifact.payout_state_generation), - anchor_age_ms=anchor_age_ms, - window_shares=len(artifact.shares_json), - ) - return False with self._job_cache_lock: - outcome, armed_generation = ( - self._install_payout_ledger_artifact_locked(artifact) - ) - self._record_payout_artifact_event(outcome) - if outcome == "discarded": - self._payout_artifact_log( - "payout_artifact_discarded", - payout_state_generation=int(artifact.payout_state_generation), - anchor_age_ms=anchor_age_ms, - window_shares=len(artifact.shares_json), - ) - return False - if outcome in ("installed", "refreshed"): - self._payout_artifact_log( - "payout_artifact_" + outcome, - generation=armed_generation, - payout_state_generation=int(artifact.payout_state_generation), - anchor_age_ms=anchor_age_ms, - window_shares=len(artifact.shares_json), - ) + if artifact.payout_state_generation != self._payout_state_generation: + return + current = self._payout_ledger_artifact + already_current = False + if ( + current is not None + and current.payout_state_generation + == artifact.payout_state_generation + ): + current_anchor_ms = ( + -1 + if current.snapshot_anchor_ms is None + else int(current.snapshot_anchor_ms) + ) + artifact_anchor_ms = ( + -1 + if artifact.snapshot_anchor_ms is None + else int(artifact.snapshot_anchor_ms) + ) + if current_anchor_ms > artifact_anchor_ms: + return + if ( + current.network_difficulty == artifact.network_difficulty + and current.share_snapshot_sha256 is not None + and current.share_snapshot_sha256 + == artifact.share_snapshot_sha256 + ): + # The armed artifact already carries exactly this + # window; keep its generation (no lookup re-key) but + # still treat the preparation as a success below. A + # fresher anchor re-proved the same window, so advance + # the staleness clock in place -- otherwise a quiet + # share stream would cycle identical rebuilds forever + # without ever un-staling the armed artifact. The + # armed balances are kept: they are what the reuse + # fence hashes against the published payout state + # (including an accepted parent's preview patch). + if artifact_anchor_ms > current_anchor_ms: + self._payout_ledger_artifact = dataclass_replace( + current, + accepted_share_count=artifact.accepted_share_count, + prepared_monotonic=artifact.prepared_monotonic, + snapshot_anchor_ms=artifact.snapshot_anchor_ms, + ) + already_current = True + else: + template_artifacts = getattr( + self, + "_template_artifacts", + None, + ) + if ( + current.network_difficulty + != artifact.network_difficulty + and template_artifacts is not None + and current.network_difficulty + == int(template_artifacts.network_difficulty) + and artifact.network_difficulty + != int(template_artifacts.network_difficulty) + ): + # Whether a delayed pre-retarget build's anchor + # ties or leads, keep the artifact the live + # template difficulty can actually reuse: a + # wrong-difficulty install would fail every reuse + # probe on the difficulty check, which never + # re-arms, until the next synchronous build + # re-seeded the live window. + return + if not already_current: + self._payout_ledger_artifact_generation += 1 + self._payout_ledger_artifact = dataclass_replace( + artifact, + generation=self._payout_ledger_artifact_generation, + ) # An armed artifact -- installed fresh or confirmed already current # -- proves preparation can succeed again; let the next fence-failure # re-arm run at the configured floor. with self._payout_artifact_executor_lock: self._payout_artifact_rearm_backoff = 1 - return True - - def _install_payout_ledger_artifact_locked( - self, - artifact: PayoutLedgerArtifact, - ) -> tuple[str, int | None]: - """Ordering core of the install; returns (outcome, armed generation).""" - if artifact.payout_state_generation != self._payout_state_generation: - return "discarded", None - current = self._payout_ledger_artifact - if ( - current is not None - and current.payout_state_generation - == artifact.payout_state_generation - ): - current_anchor_ms = ( - -1 - if current.snapshot_anchor_ms is None - else int(current.snapshot_anchor_ms) - ) - artifact_anchor_ms = ( - -1 - if artifact.snapshot_anchor_ms is None - else int(artifact.snapshot_anchor_ms) - ) - if current_anchor_ms > artifact_anchor_ms: - return "discarded", None - if ( - current.network_difficulty == artifact.network_difficulty - and current.share_snapshot_sha256 is not None - and current.share_snapshot_sha256 - == artifact.share_snapshot_sha256 - ): - # The armed artifact already carries exactly this - # window; keep its generation (no lookup re-key) but - # still treat the preparation as a success, and advance - # the freshness clock in place either way -- otherwise a - # quiet share stream, or a pending-commit floor pinning - # the anchor so every re-prove lands at the SAME anchor, - # would cycle identical rebuilds forever without ever - # un-staling the armed artifact. No fresher window is - # constructible while the window bytes are unchanged, so - # the re-walk that just re-proved them earns the credit. - # The armed balances are kept: they are what the reuse - # fence hashes against the published payout state - # (including an accepted parent's preview patch). - anchor_advanced = artifact_anchor_ms > current_anchor_ms - self._payout_ledger_artifact = dataclass_replace( - current, - accepted_share_count=artifact.accepted_share_count, - prepared_monotonic=time.monotonic(), - snapshot_anchor_ms=( - artifact.snapshot_anchor_ms - if anchor_advanced - else current.snapshot_anchor_ms - ), - ) - if anchor_advanced: - return "refreshed", int(current.generation) - return "already_current", int(current.generation) - template_artifacts = getattr( - self, - "_template_artifacts", - None, - ) - if ( - current.network_difficulty != artifact.network_difficulty - and template_artifacts is not None - and current.network_difficulty - == int(template_artifacts.network_difficulty) - and artifact.network_difficulty - != int(template_artifacts.network_difficulty) - ): - # Whether a delayed pre-retarget build's anchor ties or - # leads, keep the artifact the live template difficulty - # can actually reuse: a wrong-difficulty install would - # fail every reuse probe on the difficulty check, which - # never re-arms, until the next synchronous build - # re-seeded the live window. - return "discarded", None - self._payout_ledger_artifact_generation += 1 - # Freshness runs from the moment the window becomes reusable, not - # from build completion: a sync-seeded artifact is constructed - # mid-bundle-build and reaches cache publication only after the - # audit builder, so stamping here keeps the budget honest at every - # install site. - self._payout_ledger_artifact = dataclass_replace( - artifact, - generation=self._payout_ledger_artifact_generation, - prepared_monotonic=time.monotonic(), - ) - return "installed", int(self._payout_ledger_artifact_generation) def _payout_artifact_preparation_loop(self) -> None: while True: @@ -4622,40 +4472,28 @@ def _schedule_payout_ledger_artifact_preparation( < effective_interval ): return - rearm_backoff = self._payout_artifact_rearm_backoff else: self._payout_artifact_rearm_backoff = 1 - rearm_backoff = None self._payout_artifact_requested = ( int(payout_state_generation), int(network_difficulty), ) self._payout_artifact_last_schedule_monotonic = time.monotonic() - if self._payout_artifact_future is None: - executor = self._payout_artifact_executor - if executor is None: - executor = ThreadPoolExecutor( - max_workers=1, - thread_name_prefix="prism-payout-artifact", - ) - self._payout_artifact_executor = executor - self._payout_artifact_future = executor.submit( - self._payout_artifact_preparation_loop - ) - if rearm_backoff is not None: - # Only debounced fence-failure re-arms log here; event-driven - # preparations ride payout publications, which are already - # visible. Recorded outside the executor lock the counter - # shares. - self._record_payout_artifact_event("rearm_scheduled") - self._payout_artifact_log( - "payout_artifact_rearm_scheduled", - payout_state_generation=int(payout_state_generation), - backoff=int(rearm_backoff), + if self._payout_artifact_future is not None: + return + executor = self._payout_artifact_executor + if executor is None: + executor = ThreadPoolExecutor( + max_workers=1, + thread_name_prefix="prism-payout-artifact", + ) + self._payout_artifact_executor = executor + self._payout_artifact_future = executor.submit( + self._payout_artifact_preparation_loop ) def _payout_artifact_max_anchor_age_ms(self) -> float: - """Audit ceiling on a served window's wall-clock anchor age.""" + """Reuse staleness bound for an armed artifact's snapshot anchor.""" return ( float( getattr( @@ -4667,50 +4505,24 @@ def _payout_artifact_max_anchor_age_ms(self) -> float: * 1000.0 ) - def _payout_artifact_reuse_staleness_seconds(self) -> float: - """Monotonic freshness budget of an armed artifact, from install.""" - return float( - getattr( - self, - "payout_artifact_reuse_staleness_seconds", - DEFAULT_PRISM_PAYOUT_ARTIFACT_REUSE_STALENESS_SECONDS, - ) - ) - - @staticmethod - def _payout_artifact_log(event: str, **fields: object) -> None: - """Single-line JSON lifecycle log; the incident was invisible here.""" - print( - "prism coordinator: " - + json.dumps({"event": event, **fields}, sort_keys=True), - flush=True, - ) - - def _record_payout_artifact_event(self, event: str) -> None: - self._ensure_job_cache_state() - with self._payout_artifact_executor_lock: - counts = self.payout_artifact_event_counts - counts[event] = int(counts.get(event, 0)) + 1 - def _usable_payout_ledger_artifact( self, payout_state_generation: int, network_difficulty: int, *, rearm_on_fence_failure: bool = True, + ignore_anchor_age: bool = False, ) -> PayoutLedgerArtifact | None: - """Return the armed artifact when reuse is valid for NEW work. + """Return the armed artifact when reuse is valid for new work. Validity is anchor-scoped: the artifact's window is exact at its own snapshot anchor (which reused bundles declare), so shares landing after the anchor never invalidate it -- they belong to the next - window by construction. Freshness is measured from the install - (prepared_monotonic), never from the anchor: the window walk itself - takes seconds at production volume, and an anchor-age freshness test - rejected every slow build on arrival -- the born-expired livelock - behind the 2026-07-29 rollback. The much looser wall-clock anchor - ceiling stays as the audit-facing backstop on how far a served - window may trail the live ledger. + window by construction. The bounded anchor-age check only caps how + far a served window may trail the live ledger. ignore_anchor_age is + for in-flight builds re-validating an artifact they already selected: + supersession and the balances fence still apply, but aging past the + bound mid-build must not scrap work that was fresh at selection. """ self._ensure_job_cache_state() with self._job_cache_lock: @@ -4727,17 +4539,14 @@ def _usable_payout_ledger_artifact( # anchor reused bundles must stamp; fail closed. return None if ( - time.monotonic() - float(artifact.prepared_monotonic) - > self._payout_artifact_reuse_staleness_seconds() - or now_ms() - int(artifact.snapshot_anchor_ms) + not ignore_anchor_age + and now_ms() - int(artifact.snapshot_anchor_ms) > self._payout_artifact_max_anchor_age_ms() ): - # The armed window outlived its freshness budget (or, under - # pathological build latency, its declared anchor crossed the - # audit ceiling). No payout event may arrive for a long time, so - # queue a bounded speculative rebuild that re-anchors the - # window; the aged artifact stays rejected here for new reuse - # decisions while in-flight builds keep the copy they selected. + # The anchor aged past the staleness bound. No payout event may + # arrive for a long time, so queue a bounded speculative rebuild + # that re-anchors the window; the aged artifact stays rejected + # here for new reuse decisions. if rearm_on_fence_failure: self._rearm_payout_ledger_artifact_after_fence_failure( payout_state_generation, @@ -4752,36 +4561,11 @@ def _usable_payout_ledger_artifact( balances_sha256 = canonical_json_sha256(artifact.prior_balances) with self._job_cache_lock: if ( - self._payout_state_generation != payout_state_generation + self._payout_ledger_artifact is not artifact + or self._payout_state_generation != payout_state_generation or self._published_payout_state.artifact is not published_artifact ): return None - latest = self._payout_ledger_artifact - if latest is not artifact: - # An equal-window freshness restamp (already_current / - # refreshed) replaces the armed object while this probe was - # hashing balances. The restamp keeps the generation, the - # window bytes, and the balances object, and only moves the - # freshness stamp (and possibly the anchor) forward, so the - # validity established above still holds for the restamped - # copy; failing closed here would turn the intentional - # pinned-floor recovery path into a spurious synchronous - # reward-window walk. A real re-key changes the generation - # or the window sha and still fails closed. - if ( - latest is None - or latest.generation != artifact.generation - or latest.payout_state_generation - != artifact.payout_state_generation - or latest.network_difficulty != artifact.network_difficulty - or latest.share_snapshot_sha256 is None - or latest.share_snapshot_sha256 - != artifact.share_snapshot_sha256 - or latest.snapshot_anchor_ms is None - or latest.prior_balances is not artifact.prior_balances - ): - return None - artifact = latest if balances_sha256 != published_artifact.prior_balances_sha256: # A candidate can carry a ledger snapshot prepared before its # payout state is published. Never keep retrying that stale @@ -4795,7 +4579,7 @@ def _rearm_payout_ledger_artifact_after_fence_failure( payout_state_generation: int, network_difficulty: int, ) -> None: - """Debounced rebuild scheduling for an artifact past its freshness budget. + """Debounced rebuild scheduling for an artifact past its anchor bound. The interval floor keeps the reward-window CTE from running continuously when artifacts age out faster than rebuilds complete. @@ -5199,14 +4983,6 @@ def _begin_routine_job_build_preparation( deferred_recorded = False subscription_recorded = False subscribed_bundle: CachedJobBundle | None = None - admission_started_monotonic = time.monotonic() - admission_deadline_seconds = float( - getattr( - self, - "routine_admission_deadline_seconds", - DEFAULT_PRISM_ROUTINE_ADMISSION_DEADLINE_SECONDS, - ) - ) while True: self._job_build_priority_changed.clear() priority_promises: tuple[Future[CachedJobBundle], ...] = () @@ -5267,38 +5043,6 @@ def remove_dead_preparation( raise _JobBuildCancelled( "coordinator stopped behind publication priority" ) - parked_seconds = time.monotonic() - admission_started_monotonic - if parked_seconds > admission_deadline_seconds: - # Unbounded parking here is how the 2026-07-29 incident hid: - # initial jobs rode the client's own 90s timeout with zero - # tracebacks. Fail fast and visibly instead; callers retry - # with their normal pacing. - with self._job_build_scheduler_lock: - if request_source == "initial": - self._record_initial_prepared_work_locked( - "admission_deadline" - ) - now_monotonic = time.monotonic() - should_log = ( - self._admission_deadline_last_log_monotonic is None - or now_monotonic - - self._admission_deadline_last_log_monotonic >= 5.0 - ) - if should_log: - self._admission_deadline_last_log_monotonic = ( - now_monotonic - ) - if should_log: - print( - "prism coordinator: job admission deadline exceeded " - f"source={request_source} " - f"parked_seconds={parked_seconds:.1f}", - flush=True, - ) - raise JobBuildAdmissionDeadlineExceeded( - "job bundle admission parked behind publication priority " - f"for {parked_seconds:.1f}s" - ) if priority_promises: for promise in priority_promises: if ( @@ -6796,8 +6540,6 @@ def _publish_payout_state_candidate( published_generation: int | None = None schedule_retry = False active_to_cancel: _FanoutCancellation | None = None - publication_born_expired: tuple[int | None, int] | None = None - publication_installed: tuple[int, int, int] | None = None publish_started = 0.0 with self._job_cache_lock: with self.lock: @@ -6901,61 +6643,17 @@ def _publish_payout_state_candidate( self._accepted_block_payout_preview_condition.notify_all() self._payout_state_generation = published_generation prepared_artifact = candidate.ledger_artifact - candidate_anchor_age_ms = ( - None - if prepared_artifact is None - or prepared_artifact.snapshot_anchor_ms is None - else now_ms() - - int(prepared_artifact.snapshot_anchor_ms) - ) if ( prepared_artifact is not None and prepared_artifact.payout_state_generation == published_generation - and candidate_anchor_age_ms is not None - and candidate_anchor_age_ms - <= self._payout_artifact_max_anchor_age_ms() ): self._payout_ledger_artifact_generation += 1 - # Restamp freshness at the install: the candidate - # artifact was built before publication, and the - # delivery-gate drain between the two can outlive - # the reuse budget -- a freshly published payout - # generation must never arm an already-stale - # artifact and force the next builds back through - # the synchronous reward-window walk. self._payout_ledger_artifact = dataclass_replace( prepared_artifact, generation=self._payout_ledger_artifact_generation, - prepared_monotonic=time.monotonic(), - ) - publication_installed = ( - self._payout_ledger_artifact_generation, - candidate_anchor_age_ms, - len(prepared_artifact.shares_json), ) else: - if ( - prepared_artifact is not None - and prepared_artifact.payout_state_generation - == published_generation - ): - # Same admission rule as - # _install_payout_ledger_artifact: candidate - # construction plus the delivery-gate drain - # can push the declared anchor past the - # audit ceiling, and arming such a - # BORN-EXPIRED artifact would fail every - # reuse probe on anchor age -- with the - # re-arm suppressed while an accepted - # preview awaits durability. Discard it; the - # post-publication probe schedules - # preparation (or the durable-confirmation - # path resumes it once the preview lands). - publication_born_expired = ( - candidate_anchor_age_ms, - len(prepared_artifact.shares_json), - ) self._payout_ledger_artifact = None self._published_payout_state = PublishedPayoutState( generation=published_generation, @@ -7021,33 +6719,6 @@ def _publish_payout_state_candidate( if published_generation is None: self._record_discarded_payout_candidate() return None - if publication_born_expired is not None: - born_expired_age_ms, born_expired_shares = publication_born_expired - self._record_payout_artifact_event("born_expired") - self._payout_artifact_log( - "payout_artifact_born_expired", - payout_state_generation=int(published_generation), - anchor_age_ms=born_expired_age_ms, - window_shares=born_expired_shares, - during_publication=True, - ) - if publication_installed is not None: - # The pointer swap above is an install site that bypasses - # _install_payout_ledger_artifact; record it in the same event - # family or the publication path stays invisible to the - # lifecycle observability this counter exists for. - installed_generation, installed_age_ms, installed_shares = ( - publication_installed - ) - self._record_payout_artifact_event("installed") - self._payout_artifact_log( - "payout_artifact_installed", - generation=int(installed_generation), - payout_state_generation=int(published_generation), - anchor_age_ms=installed_age_ms, - window_shares=installed_shares, - during_publication=True, - ) # A publication is a fresh start for speculative preparation: the # candidate artifact installs through the atomic pointer swap above # (not _install_payout_ledger_artifact), so whatever re-arm backoff @@ -7542,12 +7213,13 @@ def _job_bundle_entry_usable( if not self._job_bundle_payout_state_current(cached): return False if not cached.collection_only: - # Cached-ready-bundle freshness follows the same two-knob split - # as artifact reuse. The declared anchor is gated only by the - # audit CEILING: the anchor is frozen per template generation - # and predates the window walk, so a tight wall-clock gate here - # declared every rebuilt bundle dead on arrival once its - # template generation outlived the bound (2026-07-29 rollback). + # The bundle cache TTL and the artifact staleness bound are + # independent knobs. A cached ready bundle must not keep serving + # a window the artifact bound would already have retired -- via a + # TTL that outlives the bound, or via a same-window re-anchor + # that keeps the artifact generation while older-anchored bundles + # sit in the cache -- or jobs would keep declaring an anchor + # older than the configured limit. found_block = getattr(cached, "found_block", None) declared_anchor_ms = ( found_block.get("anchor_job_issued_at_ms") @@ -7559,26 +7231,7 @@ def _job_bundle_entry_usable( > self._payout_artifact_max_anchor_age_ms() ): return False - # Freshness proper: a bundle keyed to the currently armed - # artifact generation carries exactly the armed window, so it is - # as fresh as reuse itself. Every other ready bundle ( - # no-artifact builds and survivors of a re-key) keeps serving - # only while its own build age sits inside the reuse-staleness - # budget, so a bundle-cache TTL that outlives the budget cannot - # keep an over-age window in circulation. - if int(getattr(cached, "payout_artifact_generation", 0)) > 0: - with self._job_cache_lock: - armed = self._payout_ledger_artifact - if ( - armed is not None - and int(armed.generation) - == int(cached.payout_artifact_generation) - ): - return True - return ( - time.monotonic() - float(cached.built_monotonic) - <= self._payout_artifact_reuse_staleness_seconds() - ) + return True # Collection bundles sign a synthetic bootstrap share containing the # exact template ntime. A clock-only observation keeps the stable work # fingerprint, but it must rebuild this signed bundle instead of @@ -8431,34 +8084,26 @@ def build_shared_job_bundle( prepared_ledger_artifact: PayoutLedgerArtifact | None = None snapshot_accepted_count: int | None = None if payout_artifact is not None: - # The reuse decision was made at request preparation. Nothing - # that happens to the ARMED slot afterwards -- a fresher window - # installing (generation re-key), a same-window anchor refresh - # swapping the instance, or the artifact simply aging out of new - # reuse decisions -- invalidates the copy this build already - # selected: its window stays audit-reproducible at its declared - # anchor. Scrapping admitted work on any of those (as the first - # anchor-scoped deploy did on the generation re-key) turns - # routine churn into a rebuild storm. Only two fences gate - # completion: the payout generation and the published balances - # this artifact binds. - with self._job_cache_lock: - current_payout_state_generation = self._payout_state_generation - published_artifact = self._published_payout_state.artifact - if current_payout_state_generation != payout_state_generation: - raise JobBuildSuperseded( - "payout generation changed before construction" - ) - if published_artifact is None: - try: - published_artifact = self._current_payout_state_artifact() - except Exception as exc: - raise JobBuildSuperseded( - "published payout state unavailable before construction" - ) from exc + # The reuse decision was made at request preparation; re-validate + # supersession and the balances fence only. Aging past the anchor + # bound while queued must not scrap the build for a full snapshot + # it was armed to avoid, and the comparison is by artifact + # identity fields rather than object identity: a same-window + # rebuild refreshes the armed anchor by swapping the instance + # while the generation -- the re-key authority -- and the window + # and balances bytes stay identical. + current_artifact = self._usable_payout_ledger_artifact( + payout_state_generation, + build_request.key.network_difficulty, + ignore_anchor_age=True, + ) if ( - canonical_json_sha256(payout_artifact.prior_balances) - != published_artifact.prior_balances_sha256 + current_artifact is None + or current_artifact.payout_state_generation + != payout_artifact.payout_state_generation + or current_artifact.generation != payout_artifact.generation + or current_artifact.share_snapshot_sha256 + != payout_artifact.share_snapshot_sha256 ): raise JobBuildSuperseded( "precomputed payout artifact changed before construction" @@ -24062,8 +23707,6 @@ def tip_refresh_metrics_lines(self) -> list[str]: with self._serve_builder_metrics_lock: serve_counts = dict(self.serve_builder_counts) serve_window_counts = dict(self.serve_builder_window_cache_counts) - with self._payout_artifact_executor_lock: - payout_artifact_events = dict(self.payout_artifact_event_counts) lines.extend( [ "# HELP qbit_prism_serve_builder_events_total Persistent audit-builder daemon lifecycle and request outcomes.", @@ -24077,21 +23720,6 @@ def tip_refresh_metrics_lines(self) -> list[str]: "window_uploads", ) ], - "# HELP qbit_prism_payout_artifact_events_total Payout-ledger-artifact lifecycle outcomes (build, install, reuse pacing).", - "# TYPE qbit_prism_payout_artifact_events_total counter", - *[ - f'qbit_prism_payout_artifact_events_total{{event="{event}"}} {int(payout_artifact_events.get(event, 0))}' - for event in ( - "built", - "build_aborted", - "installed", - "refreshed", - "already_current", - "discarded", - "born_expired", - "rearm_scheduled", - ) - ], "# HELP qbit_prism_serve_builder_window_cache_total Daemon parsed share-window cache outcomes.", "# TYPE qbit_prism_serve_builder_window_cache_total counter", *[ @@ -24274,7 +23902,6 @@ def job_build_metrics_lines(self) -> list[str]: "singleflight", "deferred", "subscribed", - "admission_deadline", ) ], "# HELP qbit_prism_job_build_worker_events_total Pure builder subprocess lifecycle events.", diff --git a/tests/test_prism_coordinator_job_cache.py b/tests/test_prism_coordinator_job_cache.py index 29282ef..2dfbe91 100644 --- a/tests/test_prism_coordinator_job_cache.py +++ b/tests/test_prism_coordinator_job_cache.py @@ -24,7 +24,6 @@ from lab.prism import prism_coordinator as prism_coordinator_module from lab.prism.prism_coordinator import ( ClientState, - JobBuildAdmissionDeadlineExceeded, JobBuildSuperseded, MAX_PRISM_JOB_BUNDLE_CACHE_ENTRIES, PRISM_JOB_EXTRANONCE1_PLACEHOLDER_HEX, @@ -470,15 +469,15 @@ def test_artifact_paths_proceed_at_the_clamped_anchor_floor( server._finish_pending_share_commit(share) def test_background_build_refuses_a_pathologically_old_floor(self) -> None: - # A floor held further below now than the audit anchor ceiling (a + # A floor held further below now than the reuse staleness bound (a # wedged writer or leaked release) would arm an artifact that is - # born expired; refuse before paying the window walk so the + # dead on arrival; refuse before paying the window walk so the # re-arm backoff paces retries. ledger = AnchorRecordingLedger() server, _rpc = coordinator(ledger=ledger) install_fake_bundle_builder(server) artifacts = server.current_template_artifacts() - share = stamped_pending_share(now_ms() - 400_000) + share = stamped_pending_share(now_ms() - 30_000) self._hold_floor(server, share) try: self.assertIsNone( @@ -2574,14 +2573,15 @@ def submit(self, _fn: object) -> Future[None]: with server._payout_artifact_executor_lock: self.assertIsNone(server._payout_artifact_requested) - # Age the armed window past its reuse freshness budget. - server.payout_artifact_reuse_staleness_seconds = 10.0 + # Age the armed window past the reuse staleness bound. + server.payout_artifact_max_anchor_age_seconds = 10.0 with server._job_cache_lock: armed = server._payout_ledger_artifact assert armed is not None + assert armed.snapshot_anchor_ms is not None server._payout_ledger_artifact = dataclass_replace( armed, - prepared_monotonic=float(armed.prepared_monotonic) - 11.0, + snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 11_000, ) self.assertIsNone( @@ -2640,11 +2640,11 @@ def test_aborted_speculative_rebuilds_back_off_and_reset(self) -> None: assert artifacts is not None server.payout_artifact_rearm_min_seconds = 5.0 - # A pending-commit floor older than the audit anchor ceiling aborts - # the rebuild before the window walk (the artifact would arm born - # expired); each abort doubles the re-arm interval instead of - # retrying the reward-window walk at the floor forever. - share = stamped_pending_share(now_ms() - 400_000) + # A pathologically old pending-commit floor aborts the rebuild + # before the window walk (the artifact would arm already past the + # staleness bound); each abort doubles the re-arm interval instead + # of retrying the reward-window walk at the floor forever. + share = stamped_pending_share(now_ms() - 30_000) server._ensure_pending_share_commit_state() with server._pending_share_commit_lock: server._pending_share_commit_floor[id(share)] = [ @@ -2741,13 +2741,14 @@ def test_landed_preview_suppresses_fence_failure_rearm(self) -> None: block_height=int(rpc.template["height"]) - 1, ) - # Age the armed window past its reuse freshness budget. + # Age the armed window past the reuse staleness bound. with server._job_cache_lock: armed = server._payout_ledger_artifact assert armed is not None + assert armed.snapshot_anchor_ms is not None server._payout_ledger_artifact = dataclass_replace( armed, - prepared_monotonic=float(armed.prepared_monotonic) - 11.0, + snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 11_000, ) self.assertIsNone( @@ -2761,9 +2762,7 @@ def test_landed_preview_suppresses_fence_failure_rearm(self) -> None: # durable-confirmation call site. self.assertEqual(scheduled, []) - def test_artifact_reuse_bounded_by_install_age_and_anchor_ceiling( - self, - ) -> None: + def test_artifact_reuse_is_bounded_by_anchor_age(self) -> None: server, rpc = coordinator() install_fake_bundle_builder(server) artifacts = server.store_template_artifacts(dict(rpc.template)) @@ -2777,18 +2776,17 @@ def test_artifact_reuse_bounded_by_install_age_and_anchor_ceiling( assert armed is not None assert armed.snapshot_anchor_ms is not None - # A freshly installed window serves even when its anchor is older - # than the freshness budget and shares landed after it: post-anchor - # shares belong to the next window by construction, and freshness is - # measured from prepared_monotonic -- never from the anchor, which - # predates the window walk. (The reuse-staleness and anchor-ceiling - # env knobs wire these same attributes in __init__.) - server.payout_artifact_reuse_staleness_seconds = 10.0 + # Inside the bound the artifact serves even though shares landed + # after its anchor: they belong to the next window by construction, + # so the moved durable count is irrelevant to reuse validity. + # (PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS wires this same + # attribute in __init__.) + server.payout_artifact_max_anchor_age_seconds = 5.0 server.ledger.miners = [*server.ledger.miners, "late-1", "late-2"] with server._job_cache_lock: server._payout_ledger_artifact = dataclass_replace( armed, - snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 15_000, + snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 4_000, ) self.assertIsNotNone( server._usable_payout_ledger_artifact( @@ -2797,23 +2795,7 @@ def test_artifact_reuse_bounded_by_install_age_and_anchor_ceiling( ) ) - # Past the install-age budget the window is retired for new reuse - # decisions. - with server._job_cache_lock: - server._payout_ledger_artifact = dataclass_replace( - armed, - prepared_monotonic=float(armed.prepared_monotonic) - 11.0, - ) - self.assertIsNone( - server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - rearm_on_fence_failure=False, - ) - ) - - # The wall-clock anchor ceiling stays as the audit-facing backstop. - server.payout_artifact_max_anchor_age_seconds = 5.0 + # Past the bound the window is retired for new reuse decisions. with server._job_cache_lock: server._payout_ledger_artifact = dataclass_replace( armed, @@ -2844,7 +2826,7 @@ def test_in_flight_build_keeps_aged_artifact_it_already_selected( assert armed.snapshot_anchor_ms is not None aged = dataclass_replace( armed, - prepared_monotonic=float(armed.prepared_monotonic) - 11.0, + snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 11_000, ) server._payout_ledger_artifact = aged @@ -2924,89 +2906,27 @@ def test_in_flight_build_survives_same_window_anchor_refresh(self) -> None: selected.snapshot_anchor_ms, ) - def test_generation_current_cached_bundle_serves_past_staleness( - self, - ) -> None: - server, rpc = coordinator() - recorded = install_fake_bundle_builder(server) - first = server.store_template_artifacts(dict(rpc.template)) - assert first is not None - with server._payout_artifact_executor_lock: - server._payout_artifact_executor_shutdown = True - # Freshness budget below the bundle-cache TTL so the two gates are - # distinguishable inside the TTL window. - server.payout_artifact_reuse_staleness_seconds = 5.0 - - server.shared_job_bundle(first, mode="ready") - self.assertEqual(recorded["calls"], 1) - with server._job_cache_lock: - armed = server._payout_ledger_artifact - assert armed is not None - - # A bundle keyed to the currently armed artifact generation carries - # exactly the armed window: it keeps serving past the staleness - # budget (its window is as fresh as reuse itself), bounded upstream - # by the cache TTL. - second = server.store_template_artifacts( - base_template(height=11, prevhash="22" * 32) - ) - assert second is not None - reused = server.shared_job_bundle(second, mode="ready") - self.assertEqual(reused.payout_artifact_generation, armed.generation) - self.assertEqual(recorded["calls"], 2) - with server._job_cache_lock: - for key, entry in list(server._job_bundle_cache.items()): - if entry.payout_artifact_generation == armed.generation: - server._job_bundle_cache[key] = dataclass_replace( - entry, - built_monotonic=float(entry.built_monotonic) - 7.0, - ) - server.shared_job_bundle(second, mode="ready") - self.assertEqual(recorded["calls"], 2) - - def test_cached_no_artifact_bundle_gates_on_build_age_not_anchor( - self, - ) -> None: - # 2026-07-29 regression companion: the gate must follow build age - # (freshness of the work) inside the audit ceiling, never the - # wall-clock anchor, which is frozen per template generation. + def test_cached_ready_bundle_respects_anchor_age_bound(self) -> None: server, rpc = coordinator() recorded = install_fake_bundle_builder(server) artifacts = server.store_template_artifacts(dict(rpc.template)) assert artifacts is not None with server._payout_artifact_executor_lock: server._payout_artifact_executor_shutdown = True - server.payout_artifact_reuse_staleness_seconds = 5.0 - seeded = server.shared_job_bundle(artifacts, mode="ready") + first = server.shared_job_bundle(artifacts, mode="ready") self.assertEqual(recorded["calls"], 1) - - # Fresh build age serves from cache once the armed artifact is - # gone. - with server._job_cache_lock: - server._payout_ledger_artifact = None + # Inside the bound the cached bundle keeps serving. server.shared_job_bundle(artifacts, mode="ready") self.assertEqual(recorded["calls"], 1) - # Build age past the staleness budget rebuilds even though the - # bundle-cache TTL has not lapsed. - with server._job_cache_lock: - server._payout_ledger_artifact = None - for key, entry in list(server._job_bundle_cache.items()): - server._job_bundle_cache[key] = dataclass_replace( - entry, - built_monotonic=float(entry.built_monotonic) - 7.0, - ) - server.shared_job_bundle(artifacts, mode="ready") - self.assertEqual(recorded["calls"], 2) - - # The audit ceiling on the declared anchor stays as the backstop - # even for a freshly built entry. - aged_anchor_ms = ( - int(seeded.found_block["anchor_job_issued_at_ms"]) - 400_000 - ) + # The bundle cache TTL and the artifact staleness bound are + # independent knobs: an entry whose declared window anchor outlives + # the bound must stop serving even while its TTL is still running, + # or jobs would keep declaring an anchor older than the configured + # limit. + aged_anchor_ms = int(first.found_block["anchor_job_issued_at_ms"]) - 11_000 with server._job_cache_lock: - server._payout_ledger_artifact = None for key, entry in list(server._job_bundle_cache.items()): aged_found_block = dict(entry.found_block) aged_found_block["anchor_job_issued_at_ms"] = aged_anchor_ms @@ -3014,491 +2934,16 @@ def test_cached_no_artifact_bundle_gates_on_build_age_not_anchor( entry, found_block=aged_found_block, ) - server.shared_job_bundle(artifacts, mode="ready") - self.assertEqual(recorded["calls"], 3) - - def test_slow_window_walk_still_arms_reusable_artifact(self) -> None: - # 2026-07-29 regression: the reward-window walk takes multiple - # seconds at production volume, and freshness measured from the - # wall-clock anchor rejected every slow build on arrival. The walk - # here advances the wall clock past the freshness budget; the - # artifact must still arm, serve reuse, and reset the re-arm - # backoff. - clock = [now_ms()] - - class SlowWalkLedger(FakeLedger): - def snapshot_at_job_issue( - self, - anchor_job_issued_at_ms: int, - *, - window_weight: int | None = None, - ) -> list[FakeShare]: - result = super().snapshot_at_job_issue( - anchor_job_issued_at_ms, - window_weight=window_weight, - ) - clock[0] += 15_000 - return result - - server, rpc = coordinator(ledger=SlowWalkLedger()) - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - with server._payout_artifact_executor_lock: - server._payout_artifact_rearm_backoff = 4 - with patch( - "lab.prism.prism_coordinator.now_ms", - side_effect=lambda: clock[0], - ): - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - armed = server._payout_ledger_artifact - self.assertIsNotNone(armed) + armed = server._payout_ledger_artifact assert armed is not None assert armed.snapshot_anchor_ms is not None - self.assertGreater( - clock[0] - int(armed.snapshot_anchor_ms), - 10_000, - ) - self.assertIs( - server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ), + server._payout_ledger_artifact = dataclass_replace( armed, + snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 11_000, ) - with server._payout_artifact_executor_lock: - self.assertEqual(server._payout_artifact_rearm_backoff, 1) - metrics = server.metrics_payload() - self.assertIn( - 'qbit_prism_payout_artifact_events_total{event="installed"} 1', - metrics, - ) - self.assertIn( - 'qbit_prism_payout_artifact_events_total{event="built"} 1', - metrics, - ) - - def test_born_expired_install_paces_backoff_instead_of_resetting( - self, - ) -> None: - # A walk that outlives the audit ceiling arms nothing. Crediting it - # as a success reset the re-arm backoff to the floor and re-walked - # the reward window continuously -- the 2026-07-29 livelock. It - # must pace exactly like a failed preparation. - clock = [now_ms()] - - class GlacialWalkLedger(FakeLedger): - def snapshot_at_job_issue( - self, - anchor_job_issued_at_ms: int, - *, - window_weight: int | None = None, - ) -> list[FakeShare]: - result = super().snapshot_at_job_issue( - anchor_job_issued_at_ms, - window_weight=window_weight, - ) - clock[0] += 400_000 - return result - - server, rpc = coordinator(ledger=GlacialWalkLedger()) - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - with patch( - "lab.prism.prism_coordinator.now_ms", - side_effect=lambda: clock[0], - ): - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - self.assertIsNone(server._payout_ledger_artifact) - with server._payout_artifact_executor_lock: - self.assertEqual(server._payout_artifact_rearm_backoff, 2) - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._payout_artifact_executor_lock: - self.assertEqual(server._payout_artifact_rearm_backoff, 4) - with server._payout_artifact_executor_lock: - self.assertEqual( - server.payout_artifact_event_counts["born_expired"], - 2, - ) - - def test_equal_anchor_reprove_credits_freshness(self) -> None: - # A pending-commit floor pins the snapshot anchor, so a fence-failure - # rebuild re-proves the same window at the SAME anchor. That re-prove - # must advance the freshness clock: no fresher window is - # constructible while the floor holds, and rejecting the credit - # would re-walk the reward window (with backoff reset each round) - # until the audit ceiling -- the livelock shape this re-land - # removes. - server, rpc = coordinator() - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - share = stamped_pending_share(now_ms() - 5) - server._ensure_pending_share_commit_state() - with server._pending_share_commit_lock: - server._pending_share_commit_floor[id(share)] = [ - share, - time.monotonic(), - False, - ] - try: - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - armed = server._payout_ledger_artifact - assert armed is not None - server._payout_ledger_artifact = dataclass_replace( - armed, - prepared_monotonic=float(armed.prepared_monotonic) - 11.0, - ) - self.assertIsNone( - server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - rearm_on_fence_failure=False, - ) - ) - - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - reproved = server._payout_ledger_artifact - assert reproved is not None - self.assertEqual(reproved.generation, armed.generation) - self.assertEqual( - reproved.snapshot_anchor_ms, - armed.snapshot_anchor_ms, - ) - self.assertIsNotNone( - server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - ) - finally: - server._finish_pending_share_commit(share) - def test_publication_restamps_candidate_artifact_freshness(self) -> None: - # A payout-state candidate builds its artifact before the atomic - # publication, and the delivery-gate drain between the two can - # outlive the reuse budget. The install must restamp freshness so a - # freshly published generation never arms an already-stale artifact. - server, rpc = coordinator() - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - server._pool_ready_latched = True - real_build = server._build_payout_ledger_artifact - - def slow_publication_build(*args: object) -> object: - built = real_build(*args) - if built is None: - return None - return dataclass_replace( - built, - prepared_monotonic=time.monotonic() - 11.0, - ) - - server._build_payout_ledger_artifact = slow_publication_build # type: ignore[method-assign] - parent_hash = str(rpc.template["previousblockhash"]) - server._begin_accepted_block_payout_preview( - parent_hash, - block_height=int(rpc.template["height"]) - 1, - ) - server._publish_accepted_block_payout_preview( - parent_hash, - [ - { - "recipient_id": "miner-a", - "order_key": "miner-a", - "p2mr_program_hex": "11" * 32, - "balance_sats": 25, - } - ], - ) - - self.assertGreater(server._payout_state_generation, 0) - usable = server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - self.assertIsNotNone(usable) - - def test_usable_probe_survives_equal_window_freshness_restamp( - self, - ) -> None: - # An equal-window re-prove restamps the armed artifact by replacing - # the object while a reuse probe is hashing balances outside the - # cache lock. The probe must treat the restamped copy as the same - # armed window instead of failing closed into a synchronous - # reward-window walk: under a pinned pending-commit floor the - # restamp IS the intentional recovery path, so the race is routine. - server, rpc = coordinator() - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - armed = server._payout_ledger_artifact - assert armed is not None - - real_sha256 = prism_coordinator_module.canonical_json_sha256 - restamped_once = [False] - - def restamp_during_hash(value: object) -> str: - if not restamped_once[0]: - restamped_once[0] = True - self.assertTrue( - server._install_payout_ledger_artifact( - dataclass_replace( - armed, - generation=0, - prepared_monotonic=time.monotonic(), - ) - ) - ) - return real_sha256(value) - - with patch.object( - prism_coordinator_module, - "canonical_json_sha256", - restamp_during_hash, - ): - usable = server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - self.assertTrue(restamped_once[0]) - with server._job_cache_lock: - restamped = server._payout_ledger_artifact - assert restamped is not None - self.assertIsNot(restamped, armed) - self.assertIsNotNone(usable) - self.assertIs(usable, restamped) - self.assertEqual(restamped.generation, armed.generation) - - def test_publication_discards_born_expired_candidate_artifact( - self, - ) -> None: - # Candidate construction plus the delivery-gate drain can push the - # candidate artifact's declared anchor past the audit ceiling. The - # atomic publication must apply the same born-expired admission rule - # as _install_payout_ledger_artifact: arming the artifact would fail - # every reuse probe on anchor age -- with the re-arm suppressed - # while the accepted preview awaits durability -- instead of letting - # the post-publication probe schedule recovery. - server, rpc = coordinator() - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - server._pool_ready_latched = True - real_build = server._build_payout_ledger_artifact - ceiling_ms = server._payout_artifact_max_anchor_age_ms() - - def expired_anchor_build(*args: object) -> object: - built = real_build(*args) - if built is None: - return None - assert built.snapshot_anchor_ms is not None - return dataclass_replace( - built, - snapshot_anchor_ms=int(built.snapshot_anchor_ms) - - int(ceiling_ms) - - 1_000, - ) - - server._build_payout_ledger_artifact = expired_anchor_build # type: ignore[method-assign] - parent_hash = str(rpc.template["previousblockhash"]) - server._begin_accepted_block_payout_preview( - parent_hash, - block_height=int(rpc.template["height"]) - 1, - ) - generation_before = server._payout_state_generation - server._publish_accepted_block_payout_preview( - parent_hash, - [ - { - "recipient_id": "miner-a", - "order_key": "miner-a", - "p2mr_program_hex": "11" * 32, - "balance_sats": 25, - } - ], - ) - self.assertGreater(server._payout_state_generation, generation_before) - with server._job_cache_lock: - self.assertIsNone(server._payout_ledger_artifact) - with server._payout_artifact_executor_lock: - born_expired = int( - server.payout_artifact_event_counts.get("born_expired", 0) - ) - self.assertGreaterEqual(born_expired, 1) - self.assertIsNone( - server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - ) - - def test_publication_records_installed_event(self) -> None: - # The atomic publication arms the candidate artifact through its own - # pointer swap, not _install_payout_ledger_artifact. It must still - # count in the installed lifecycle event family, or every - # publication-path install is invisible to the observability the - # event counter exists for. - server, rpc = coordinator() - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - server._pool_ready_latched = True - with server._payout_artifact_executor_lock: - installed_before = int( - server.payout_artifact_event_counts.get("installed", 0) - ) - parent_hash = str(rpc.template["previousblockhash"]) - server._begin_accepted_block_payout_preview( - parent_hash, - block_height=int(rpc.template["height"]) - 1, - ) - server._publish_accepted_block_payout_preview( - parent_hash, - [ - { - "recipient_id": "miner-a", - "order_key": "miner-a", - "p2mr_program_hex": "11" * 32, - "balance_sats": 25, - } - ], - ) - with server._job_cache_lock: - self.assertIsNotNone(server._payout_ledger_artifact) - with server._payout_artifact_executor_lock: - installed_after = int( - server.payout_artifact_event_counts.get("installed", 0) - ) - self.assertEqual(installed_after, installed_before + 1) - - def test_generation_bump_does_not_scrap_in_flight_build(self) -> None: - # 2026-07-29 regression: every differing window bumps the armed - # artifact generation, and the in-build re-validation scrapped any - # build whose selected artifact lost the slot -- a rebuild storm at - # production share rates. A build must finish on the copy it - # selected: its window stays audit-reproducible at its declared - # anchor while the payout generation and published balances hold. - server, rpc = coordinator() - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - selected = server._payout_ledger_artifact - assert selected is not None - assert selected.snapshot_anchor_ms is not None - - fresher_window = [ - {"share_seq": seq, "miner_id": "miner-a"} for seq in (1, 2, 3, 4) - ] - fresher = dataclass_replace( - selected, - generation=0, - accepted_share_count=4, - shares_json=tuple(fresher_window), - snapshot_anchor_ms=int(selected.snapshot_anchor_ms) + 5, - share_snapshot_sha256=canonical_json_sha256(fresher_window), - ) - self.assertTrue(server._install_payout_ledger_artifact(fresher)) - with server._job_cache_lock: - bumped = server._payout_ledger_artifact - assert bumped is not None - self.assertGreater(bumped.generation, selected.generation) - - server.ledger.snapshot_calls = 0 - bundle = server.build_shared_job_bundle( - artifacts, - worker(), - payout_artifact=selected, - ) - self.assertEqual(server.ledger.snapshot_calls, 0) - self.assertEqual( - bundle.found_block["anchor_job_issued_at_ms"], - selected.snapshot_anchor_ms, - ) - self.assertEqual(bundle.shares_json, list(selected.shares_json)) - - def test_rebuilt_bundle_serves_under_aged_template_generation( - self, - ) -> None: - # 2026-07-29 regression: issued_at_ms is frozen per template - # generation and predates the walk, so a wall-clock anchor gate on - # cached bundles declared every rebuilt bundle dead on arrival once - # its template generation outlived the bound. Below the audit - # ceiling, cache service must follow build age, not anchor age. - server, rpc = coordinator() - recorded = install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - with server._payout_artifact_executor_lock: - server._payout_artifact_executor_shutdown = True - with server._job_cache_lock: - server._job_build_issued_at_ms[artifacts.generation] = ( - now_ms() - 20_000 - ) - - built = server.shared_job_bundle(artifacts, mode="ready") - self.assertEqual(recorded["calls"], 1) - self.assertEqual( - built.found_block["anchor_job_issued_at_ms"], - built.issued_at_ms, - ) - - served = server.shared_job_bundle(artifacts, mode="ready") - self.assertEqual(recorded["calls"], 1) - self.assertEqual(served.issued_at_ms, built.issued_at_ms) - - def test_admission_deadline_fails_fast_and_visibly(self) -> None: - server, _rpc = coordinator() - server.routine_admission_deadline_seconds = 0.05 - server._publication_priority_scheduled_locked = ( # type: ignore[method-assign] - lambda: True - ) - - with self.assertRaises(JobBuildAdmissionDeadlineExceeded) as raised: - server._begin_routine_job_build_preparation( - request_source="initial", - cancelled=None, - ) - - # The deadline is coordination churn, not a failure budget event: - # callers retry with their normal pacing. - self.assertIsInstance(raised.exception, TemplateRefreshBlocked) - with server._job_build_scheduler_lock: - self.assertEqual( - server.initial_job_prepared_work_counts["admission_deadline"], - 1, - ) + server.shared_job_bundle(artifacts, mode="ready") + self.assertEqual(recorded["calls"], 2) def test_reused_anchor_bundle_preview_matches_fresh_build_through_guard( self, From 6aca4fa5f27646cbbab4e113f8e6dbace03bcd61 Mon Sep 17 00:00:00 2001 From: Robert Clarke Date: Thu, 30 Jul 2026 13:26:54 -0400 Subject: [PATCH 2/2] Revert "Make payout-ledger-artifact reuse anchor-scoped so it survives production share rates (#102)" This reverts commit fcc2c3e41e3210501c459ca123f56ea18e291d50. Anchor-scoped reuse livelocked mainnet payouts on 2026-07-29 (born-expired artifact livelock); together with the #103 revert this restores the payout-artifact machinery to its #101 (e2fb33d) state: exact-count fence semantics, no staleness/rearm/anchor-age knobs, no admission-deadline machinery. Co-Authored-By: Claude Fable 5 --- lab/prism/prism_coordinator.py | 453 +++++++-------- tests/test_prism_coordinator_job_cache.py | 671 +++------------------- tests/test_prism_coordinator_vardiff.py | 6 +- 3 files changed, 293 insertions(+), 837 deletions(-) diff --git a/lab/prism/prism_coordinator.py b/lab/prism/prism_coordinator.py index 78b4547..6a889a2 100644 --- a/lab/prism/prism_coordinator.py +++ b/lab/prism/prism_coordinator.py @@ -96,23 +96,15 @@ # Minimum spacing between speculative payout-ledger-artifact rebuilds armed by # a failed reuse fence. Every fence failure wants a fresh artifact, but each # rebuild walks the full reward window; the floor keeps that CTE from running -# continuously when artifacts age out faster than a rebuild can complete. +# continuously when shares land faster than a rebuild can complete. DEFAULT_PRISM_PAYOUT_ARTIFACT_REARM_MIN_SECONDS = 5.0 # Consecutive aborted speculative rebuilds double the re-arm interval up to -# this multiplier (80s at the default floor). Rebuilds abort on generation -# supersession, snapshot errors, empty windows, or a pathologically old -# pending-commit floor; backing off stops the reward-window CTE from cycling -# -- and from holding the payout preparation lock tip builds also need -- -# while the underlying condition persists. Any successful install or -# event-driven preparation resets the backoff. +# this multiplier (80s at the default floor). Under sustained share traffic +# the exact-count publication fence keeps aborting rebuilds; backing off +# stops the reward-window CTE from cycling -- and from holding the payout +# preparation lock tip builds also need -- while writes stay continuous. Any +# successful install or event-driven preparation resets the backoff. PRISM_PAYOUT_ARTIFACT_REARM_BACKOFF_CAP = 16 -# How old an armed artifact's share-snapshot anchor may grow before reuse -# stops and a re-anchored rebuild is scheduled. Validity is anchor-scoped, -# not count-scoped: shares stamped after the anchor deterministically belong -# to the next window (bundles declare the artifact's own anchor, so reused -# windows stay audit-reproducible; no share is ever lost). The bound only -# caps how far a served window may trail the live ledger. -DEFAULT_PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS = 10.0 # Upper bound for one kernel transfer from the share-window spool file into # the audit builder's stdin pipe. The kernel clamps each call to the free # pipe capacity anyway; the bound only paces cancellation checkpoints. @@ -1511,10 +1503,6 @@ class PayoutLedgerArtifact: generation: int payout_state_generation: int network_difficulty: int - # Durable accepted-share total observed near the snapshot. Informational - # (diagnostics only): validity is scoped to snapshot_anchor_ms, and the - # count cannot be scoped to a clamped anchor because share stamping and - # writer enqueue are not atomic. accepted_share_count: int shares_json: tuple[dict[str, object], ...] = field(repr=False) prior_balances: tuple[dict[str, object], ...] = field(repr=False) @@ -1523,11 +1511,6 @@ class PayoutLedgerArtifact: # this artifact must declare it as anchor_job_issued_at_ms: an auditor # replaying qbit_audit_share_window at the declared anchor must reproduce # exactly these shares, which only holds at the snapshot's own anchor. - # That declaration is also what makes reuse valid while shares keep - # landing: a share stamped after this anchor deterministically belongs to - # the next window (it is never lost), so the artifact stays - # audit-reproducible until _usable_payout_ledger_artifact's bounded - # anchor-staleness check retires it. snapshot_anchor_ms: int | None = None # Canonical digest of shares_json. Cached bundles built before this # artifact was armed may only keep serving re-keyed lookups when their @@ -2831,10 +2814,6 @@ def __init__(self) -> None: "PRISM_PAYOUT_ARTIFACT_REARM_MIN_SECONDS", DEFAULT_PRISM_PAYOUT_ARTIFACT_REARM_MIN_SECONDS, ) - self.payout_artifact_max_anchor_age_seconds = env_positive_float( - "PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS", - DEFAULT_PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS, - ) self.template_cache_seconds = env_nonnegative_float( "PRISM_TEMPLATE_CACHE_SECONDS", self.blockpoll_seconds, @@ -3784,6 +3763,13 @@ def _ensure_job_cache_state(self) -> None: self._job_build_pending: _JobBuildRequest | None = None if not hasattr(self, "_job_build_issued_at_ms"): self._job_build_issued_at_ms: OrderedDict[int, int] = OrderedDict() + if not hasattr(self, "_job_build_anchor_counts"): + # Accepted-share count captured (bracketed) when the matching + # generation's snapshot anchor was clamped; None when the count + # read failed or moved mid-clamp. Guarded by _job_cache_lock. + self._job_build_anchor_counts: OrderedDict[int, int | None] = ( + OrderedDict() + ) if not hasattr(self, "job_build_timeout_seconds"): self.job_build_timeout_seconds = DEFAULT_PRISM_JOB_BUILD_TIMEOUT_SECONDS if not hasattr(self, "job_build_cancel_grace_seconds"): @@ -4233,16 +4219,15 @@ def _build_payout_ledger_artifact( ) -> PayoutLedgerArtifact | None: """Build a stable ledger snapshot without publishing it. - Validity is anchor-scoped, not count-scoped. The pending-commit clamp - selects the highest clean anchor: every share stamped at or below it - is already durable, so the window read at that anchor is exact and - reproducible no matter how many shares commit while the walk runs. - Shares stamped above the anchor deterministically belong to the next - window; concurrent writers therefore never invalidate this attempt. + Accepted-share counts fence both sides of the snapshot. If a writer + commits concurrently, this attempt is discarded rather than publishing + an artifact with an ambiguous cutoff; the normal inline path remains + the fail-closed fallback. """ self._ensure_job_cache_state() ledger_started = time.monotonic() try: + accepted_before, _ = self.accepted_share_stats() with self._payout_state_prepare_lock: with self._job_cache_lock: if ( @@ -4257,16 +4242,16 @@ def _build_payout_ledger_artifact( ) clamp_now_ms = now_ms() snapshot_anchor_ms = self._job_snapshot_anchor_ms(clamp_now_ms) - if ( - clamp_now_ms - snapshot_anchor_ms - > self._payout_artifact_max_anchor_age_ms() - ): - # The floor is held this far below now only by a wedged - # writer or a leaked release; an artifact anchored there - # would already be past the reuse staleness bound on - # arrival. Abort before paying the window walk -- the - # re-arm backoff paces retries and the share-commit - # liveness watchdog owns recovery. + if snapshot_anchor_ms != clamp_now_ms: + # A pending commit holds the anchor floor. Stamping and + # writer enqueue are not atomic, so a share stamped after + # the floor holder may already be durable; the global + # accepted count then cannot be scoped to the clamped + # anchor, and publishing that pairing could let reuse + # serve a window that omits a durable share. Abort before + # paying the window walk; the re-arm backoff paces + # retries and the floor drains within group-commit + # latency. return None records = list( self.ledger.snapshot_at_job_issue( @@ -4275,7 +4260,7 @@ def _build_payout_ledger_artifact( ) ) prior_balances = self.ledger.current_prior_balances() - accepted_share_count, _ = self.accepted_share_stats() + accepted_after, _ = self.accepted_share_stats() except Exception: # Artifact preparation is speculative. The synchronous bundle path # still owns errors when current work actually requires a snapshot. @@ -4285,7 +4270,7 @@ def _build_payout_ledger_artifact( "ledger_snapshot", time.monotonic() - ledger_started, ) - if not records: + if accepted_before != accepted_after or not records: return None copy_started = time.monotonic() shares_json = tuple(record.to_prism_json() for record in records) @@ -4299,7 +4284,7 @@ def _build_payout_ledger_artifact( generation=0, payout_state_generation=artifact_payout_state_generation, network_difficulty=int(network_difficulty), - accepted_share_count=accepted_share_count, + accepted_share_count=accepted_after, shares_json=shares_json, prior_balances=frozen_balances, prepared_monotonic=time.monotonic(), @@ -4319,11 +4304,10 @@ def _prepare_payout_ledger_artifact( network_difficulty, ) if artifact is None: - # A superseding generation, a snapshot error, an empty window, or - # a pathologically old pending-commit floor discarded this - # attempt; that condition can persist, and each attempt walks the - # reward window under the preparation lock. Back the - # fence-failure re-arm off until something arms. + # The publication fence (or a superseding generation) discarded + # this attempt; continuous share traffic can keep doing so, and + # each attempt walks the reward window under the preparation + # lock. Back the fence-failure re-arm off until something arms. with self._payout_artifact_executor_lock: self._payout_artifact_rearm_backoff = min( self._payout_artifact_rearm_backoff * 2, @@ -4338,16 +4322,15 @@ def _install_payout_ledger_artifact( ) -> None: """Atomically publish a prepared artifact for its own generation. - Snapshot-freshness-ordered and idempotent: every snapshot is taken at - a clean anchor (strictly below all pending commits), and the durable - window at or below an anchor is immutable, so the anchor orders - snapshots even when a build delayed in window conversion finishes - after a later snapshot installed (completion time cannot order - snapshots). Equal-anchor installs with an identical window -- every - flight waiter re-runs cache publication with the same prepared - artifact, and a same-window speculative rebuild re-reads unchanged - state under an equal anchor -- keep the installed generation instead - of re-keying bundle lookups for nothing. + Snapshot-freshness-ordered and idempotent: accepted-share counts are + append-only within a payout generation, so the count orders snapshots + even when a build delayed in window conversion finishes after a + later snapshot installed (completion time cannot order snapshots). + Equal-count installs with an identical window -- every flight waiter + re-runs cache publication with the same prepared artifact, and a + same-window speculative rebuild re-reads unchanged state under a + fresh anchor -- keep the installed generation instead of re-keying + bundle lookups for nothing. """ with self._job_cache_lock: if artifact.payout_state_generation != self._payout_state_generation: @@ -4359,65 +4342,47 @@ def _install_payout_ledger_artifact( and current.payout_state_generation == artifact.payout_state_generation ): - current_anchor_ms = ( - -1 - if current.snapshot_anchor_ms is None - else int(current.snapshot_anchor_ms) - ) - artifact_anchor_ms = ( - -1 - if artifact.snapshot_anchor_ms is None - else int(artifact.snapshot_anchor_ms) - ) - if current_anchor_ms > artifact_anchor_ms: + if ( + current.accepted_share_count + > artifact.accepted_share_count + ): return if ( - current.network_difficulty == artifact.network_difficulty - and current.share_snapshot_sha256 is not None - and current.share_snapshot_sha256 - == artifact.share_snapshot_sha256 + current.accepted_share_count + == artifact.accepted_share_count ): - # The armed artifact already carries exactly this - # window; keep its generation (no lookup re-key) but - # still treat the preparation as a success below. A - # fresher anchor re-proved the same window, so advance - # the staleness clock in place -- otherwise a quiet - # share stream would cycle identical rebuilds forever - # without ever un-staling the armed artifact. The - # armed balances are kept: they are what the reuse - # fence hashes against the published payout state - # (including an accepted parent's preview patch). - if artifact_anchor_ms > current_anchor_ms: - self._payout_ledger_artifact = dataclass_replace( - current, - accepted_share_count=artifact.accepted_share_count, - prepared_monotonic=artifact.prepared_monotonic, - snapshot_anchor_ms=artifact.snapshot_anchor_ms, - ) - already_current = True - else: - template_artifacts = getattr( - self, - "_template_artifacts", - None, - ) if ( current.network_difficulty - != artifact.network_difficulty - and template_artifacts is not None - and current.network_difficulty - == int(template_artifacts.network_difficulty) - and artifact.network_difficulty - != int(template_artifacts.network_difficulty) + == artifact.network_difficulty + and current.share_snapshot_sha256 is not None + and current.share_snapshot_sha256 + == artifact.share_snapshot_sha256 ): - # Whether a delayed pre-retarget build's anchor - # ties or leads, keep the artifact the live - # template difficulty can actually reuse: a - # wrong-difficulty install would fail every reuse - # probe on the difficulty check, which never - # re-arms, until the next synchronous build - # re-seeded the live window. - return + # The armed artifact already carries exactly this + # window; keep its generation (no lookup re-key) but + # still treat the preparation as a success below. + already_current = True + else: + template_artifacts = getattr( + self, + "_template_artifacts", + None, + ) + if ( + current.network_difficulty + != artifact.network_difficulty + and template_artifacts is not None + and current.network_difficulty + == int(template_artifacts.network_difficulty) + and artifact.network_difficulty + != int(template_artifacts.network_difficulty) + ): + # Equal counts cannot order snapshots across a + # retarget; keep the artifact the live template + # difficulty can actually reuse rather than + # letting a delayed pre-retarget build regress + # it. + return if not already_current: self._payout_ledger_artifact_generation += 1 self._payout_ledger_artifact = dataclass_replace( @@ -4492,38 +4457,13 @@ def _schedule_payout_ledger_artifact_preparation( self._payout_artifact_preparation_loop ) - def _payout_artifact_max_anchor_age_ms(self) -> float: - """Reuse staleness bound for an armed artifact's snapshot anchor.""" - return ( - float( - getattr( - self, - "payout_artifact_max_anchor_age_seconds", - DEFAULT_PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS, - ) - ) - * 1000.0 - ) - def _usable_payout_ledger_artifact( self, payout_state_generation: int, network_difficulty: int, *, rearm_on_fence_failure: bool = True, - ignore_anchor_age: bool = False, ) -> PayoutLedgerArtifact | None: - """Return the armed artifact when reuse is valid for new work. - - Validity is anchor-scoped: the artifact's window is exact at its own - snapshot anchor (which reused bundles declare), so shares landing - after the anchor never invalidate it -- they belong to the next - window by construction. The bounded anchor-age check only caps how - far a served window may trail the live ledger. ignore_anchor_age is - for in-flight builds re-validating an artifact they already selected: - supersession and the balances fence still apply, but aging past the - bound mid-build must not scrap work that was fresh at selection. - """ self._ensure_job_cache_state() with self._job_cache_lock: artifact = self._payout_ledger_artifact @@ -4534,30 +4474,26 @@ def _usable_payout_ledger_artifact( or artifact.network_difficulty != int(network_difficulty) ): return None - if artifact.snapshot_anchor_ms is None: - # Without a recorded anchor the artifact cannot declare the - # anchor reused bundles must stamp; fail closed. + if published_artifact is None: + try: + published_artifact = self._current_payout_state_artifact() + except Exception: + return None + try: + accepted_share_count, _ = self.accepted_share_stats() + except Exception: return None - if ( - not ignore_anchor_age - and now_ms() - int(artifact.snapshot_anchor_ms) - > self._payout_artifact_max_anchor_age_ms() - ): - # The anchor aged past the staleness bound. No payout event may - # arrive for a long time, so queue a bounded speculative rebuild - # that re-anchors the window; the aged artifact stays rejected - # here for new reuse decisions. + if accepted_share_count != artifact.accepted_share_count: + # The armed artifact went stale the moment a newer share became + # durable. No payout event may arrive for a long time, so queue a + # bounded speculative rebuild; exact-count reuse semantics are + # unchanged because the stale artifact stays rejected here. if rearm_on_fence_failure: self._rearm_payout_ledger_artifact_after_fence_failure( payout_state_generation, network_difficulty, ) return None - if published_artifact is None: - try: - published_artifact = self._current_payout_state_artifact() - except Exception: - return None balances_sha256 = canonical_json_sha256(artifact.prior_balances) with self._job_cache_lock: if ( @@ -4579,15 +4515,14 @@ def _rearm_payout_ledger_artifact_after_fence_failure( payout_state_generation: int, network_difficulty: int, ) -> None: - """Debounced rebuild scheduling for an artifact past its anchor bound. + """Debounced rebuild scheduling for a share-staled artifact. The interval floor keeps the reward-window CTE from running - continuously when artifacts age out faster than rebuilds complete. - Landed accepted-block previews suppress the re-arm entirely: a - speculative rebuild in that window would read database balances the - published prospective state has already superseded, and the - durable-confirmation path resumes preparation itself once the gap - closes. + continuously when shares land faster than rebuilds complete. Landed + accepted-block previews suppress the re-arm entirely: a speculative + rebuild in that window would read database balances the published + prospective state has already superseded, and the durable-confirmation + path resumes preparation itself once the gap closes. """ self._ensure_job_cache_state() with self._job_cache_lock: @@ -7143,7 +7078,7 @@ def _no_artifact_job_bundle_key( """Fallback cache identity for work built before an artifact re-key. A synchronous build caches its bundle under the no-artifact key and - then publishes its anchored window as the payout ledger artifact. The + then publishes its fenced window as the payout ledger artifact. The next lookup is keyed to that artifact and would miss the still-fresh bundle and rebuild identical work: within one payout generation and template identity the no-artifact bundle binds the same balances, @@ -7187,15 +7122,14 @@ def _job_bundle_entry_usable( cached: CachedJobBundle | None, artifacts: CachedTemplateArtifacts, ) -> bool: - """Re-validate freshness and readiness for cached bundles. + """Re-validate readiness for cached collection bundles. Readiness is monotonic in practice (the distinct accepted-miner count - only grows), so submit-capable ready bundles are served as-is while - their declared window anchor stays inside the artifact staleness - bound. A cached collection bundle is re-checked against the cheap - aggregate stats: once the pool is ready it must stop being served, or - jobs would keep collecting winning shares without submitting blocks - for up to the cache TTL. + only grows), so submit-capable ready bundles are served as-is. A cached + collection bundle is re-checked against the cheap aggregate stats: + once the pool is ready it must stop being served, or jobs would keep + collecting winning shares without submitting blocks for up to the cache + TTL. """ if cached is None: return False @@ -7213,24 +7147,6 @@ def _job_bundle_entry_usable( if not self._job_bundle_payout_state_current(cached): return False if not cached.collection_only: - # The bundle cache TTL and the artifact staleness bound are - # independent knobs. A cached ready bundle must not keep serving - # a window the artifact bound would already have retired -- via a - # TTL that outlives the bound, or via a same-window re-anchor - # that keeps the artifact generation while older-anchored bundles - # sit in the cache -- or jobs would keep declaring an anchor - # older than the configured limit. - found_block = getattr(cached, "found_block", None) - declared_anchor_ms = ( - found_block.get("anchor_job_issued_at_ms") - if isinstance(found_block, dict) - else None - ) - if declared_anchor_ms is not None and ( - now_ms() - int(declared_anchor_ms) - > self._payout_artifact_max_anchor_age_ms() - ): - return False return True # Collection bundles sign a synthetic bootstrap share containing the # exact template ntime. A clock-only observation keeps the stable work @@ -7382,12 +7298,50 @@ def _new_job_build_request( # The issued time doubles as the audit window anchor, so it # must not cover a stamped share whose commit is still in # flight: the frozen anchor stays reproducible from the - # durable ledger for every rebuild of this generation. The - # pending-commit clamp is pure anchor selection -- it freezes - # the highest anchor whose covered shares are all durable, and - # shares stamped above it deterministically belong to the next - # window -- so nothing else needs to be captured with it. - candidate_anchor_ms = self._job_snapshot_anchor_ms(now_ms()) + # durable ledger for every rebuild of this generation. + # + # The accepted count bracketing the clamp is captured with it: + # commits are single-writer FIFO in accepted_at_ms order, so at + # clamp time the total durable count equals the count of shares + # at or below the anchor. A later synchronous snapshot may only + # seed the payout ledger artifact while the live count still + # equals this value; otherwise a share newer than the frozen + # anchor is durable, and binding the inclusive count to the + # anchor-exclusive window would wedge the reuse fence open. + anchor_scoped_count: int | None = None + candidate_anchor_ms: int | None = None + # A commit racing the clamp voids one bracket. The capture is a + # pair of cheap aggregate reads, so retry a bounded number of + # times instead of leaving synchronous seeding disabled for this + # whole generation; a late re-bind after freezing is impossible + # because a failed bracket cannot tell which side of the clamp + # each racing commit landed on. + for _capture_attempt in range(3): + try: + count_before, _ = self.accepted_share_stats() + except Exception: + count_before = None + clamp_now_ms = now_ms() + candidate_anchor_ms = self._job_snapshot_anchor_ms( + clamp_now_ms + ) + if count_before is None: + break + if candidate_anchor_ms != clamp_now_ms: + # A pending commit clamps the anchor below now. Stamping + # and writer enqueue are not atomic, so a share stamped + # after the floor holder may already be durable and the + # global count cannot be scoped to this anchor; retry + # for a drained floor instead of storing an unbindable + # count. + continue + try: + count_after, _ = self.accepted_share_stats() + except Exception: + break + if int(count_after) == int(count_before): + anchor_scoped_count = int(count_after) + break with self._job_cache_lock: issued_at_ms = self._job_build_issued_at_ms.get( artifacts.generation @@ -7397,8 +7351,13 @@ def _new_job_build_request( self._job_build_issued_at_ms[artifacts.generation] = ( issued_at_ms ) + self._job_build_anchor_counts[artifacts.generation] = ( + anchor_scoped_count + ) while len(self._job_build_issued_at_ms) > 128: self._job_build_issued_at_ms.popitem(last=False) + while len(self._job_build_anchor_counts) > 128: + self._job_build_anchor_counts.popitem(last=False) build_key = JobBuildKey( best_tip_hash=artifacts.previousblockhash, previous_block_hash=artifacts.previousblockhash, @@ -8084,26 +8043,12 @@ def build_shared_job_bundle( prepared_ledger_artifact: PayoutLedgerArtifact | None = None snapshot_accepted_count: int | None = None if payout_artifact is not None: - # The reuse decision was made at request preparation; re-validate - # supersession and the balances fence only. Aging past the anchor - # bound while queued must not scrap the build for a full snapshot - # it was armed to avoid, and the comparison is by artifact - # identity fields rather than object identity: a same-window - # rebuild refreshes the armed anchor by swapping the instance - # while the generation -- the re-key authority -- and the window - # and balances bytes stay identical. - current_artifact = self._usable_payout_ledger_artifact( - payout_state_generation, - build_request.key.network_difficulty, - ignore_anchor_age=True, - ) if ( - current_artifact is None - or current_artifact.payout_state_generation - != payout_artifact.payout_state_generation - or current_artifact.generation != payout_artifact.generation - or current_artifact.share_snapshot_sha256 - != payout_artifact.share_snapshot_sha256 + self._usable_payout_ledger_artifact( + payout_state_generation, + build_request.key.network_difficulty, + ) + is not payout_artifact ): raise JobBuildSuperseded( "precomputed payout artifact changed before construction" @@ -8141,21 +8086,14 @@ def build_shared_job_bundle( raise JobBuildSuperseded( "payout generation changed before ledger snapshot" ) + accepted_before: int | None = None if resolved_mode == "ready": try: - accepted_now, _ = self.accepted_share_stats() + accepted_before, _ = self.accepted_share_stats() except Exception: # Artifact seeding is speculative; snapshot errors - # stay owned by the bundle build itself. Without an - # observed count the seed below is skipped rather - # than armed with a fabricated one. - accepted_now = None - if accepted_now is not None: - # Informational only: the window is scoped by the - # frozen anchor (shares committing during the read - # land above it and belong to the next window), so - # no bracket around the read is needed. - snapshot_accepted_count = int(accepted_now) + # stay owned by the bundle build itself. + accepted_before = None records = ( self.ledger.snapshot_at_job_issue( issued_at_ms, @@ -8164,6 +8102,33 @@ def build_shared_job_bundle( if resolved_mode == "ready" else [] ) + if accepted_before is not None: + try: + accepted_after, _ = self.accepted_share_stats() + except Exception: + pass + else: + with self._job_cache_lock: + anchor_scoped_count = ( + self._job_build_anchor_counts.get( + build_request.key.template_generation + ) + ) + # Same fence as the background preparation path -- a + # share committing during the read leaves the window + # ambiguously bound to either count -- plus an anchor + # scope check: this generation's anchor was frozen + # earlier, so a share that became durable since then + # is counted by accepted_share_stats but excluded + # from the window read at the anchor. Publishing that + # pairing would wedge the exact-count reuse fence + # open around a window missing a durable share. + if ( + anchor_scoped_count is not None + and int(accepted_after) == int(accepted_before) + and int(accepted_after) == int(anchor_scoped_count) + ): + snapshot_accepted_count = int(accepted_after) # An accepted parent's prospective carry state supersedes the # published artifact for children built on that parent; the # published balances remain the fallback for ordinary tips. @@ -8216,11 +8181,11 @@ def build_shared_job_bundle( ): # This build already paid the full ledger read; carry the window # it produced so cache publication can arm it for builds arriving - # while its anchor stays inside the staleness bound. The artifact - # carries the published payout-state balances, not this bundle's - # possibly parent-adjusted view: reuse re-applies the parent - # override itself and the reuse fence hashes the artifact - # balances against the published payout artifact. + # before the next share commit. The artifact carries the + # published payout-state balances, not this bundle's possibly + # parent-adjusted view: reuse re-applies the parent override + # itself and the reuse fence hashes the artifact balances against + # the published payout artifact. prepared_ledger_artifact = PayoutLedgerArtifact( generation=0, payout_state_generation=payout_state_generation, @@ -18356,24 +18321,18 @@ def _finish_pending_share_commit(self, pending_share: PendingShare) -> None: self._pending_share_commit_floor.pop(id(pending_share), None) def _job_snapshot_anchor_ms(self, issued_at_ms: int) -> int: - """Clamp a share-snapshot anchor below every coverable share stamp. + """Clamp a share-snapshot anchor below every pending share commit. The reward-window contract lets an auditor replay qbit_audit_share_window(anchor) against the durable ledger and expect - exactly the shares the published bundle counted. Two hazards would - violate that. A share whose accepted_at_ms is already assigned but - whose row has not committed yet (group-commit queue, in-flight batch, - or a block-candidate credit linked after landing) is invisible to the - MVCC snapshot now but joins later replays at any anchor at or above - its accepted_at_ms. And a share stamped right after this clamp - returns can land in the same millisecond as the clamp instant -- - never protected by the pending floor -- so an anchor equal to that - instant would cover it too (the window predicate is - anchor-inclusive). Anchoring strictly below every pending share and - strictly below the clamp-time millisecond keeps the issued snapshot - reproducible without making job builds wait behind the writer - connection: any stamp assigned after this call is at or above the - clamp instant, hence above the anchor. + exactly the shares the published bundle counted. A share whose + accepted_at_ms is already assigned but whose row has not committed yet + (group-commit queue, in-flight batch, or a block-candidate credit + linked after landing) would violate that: it is invisible to the MVCC + snapshot now but joins later replays at any anchor at or above its + accepted_at_ms. Anchoring strictly below every such share keeps the + issued snapshot reproducible without making job builds wait behind the + writer connection. """ self._ensure_pending_share_commit_state() stale_share_ids: list[str] = [] @@ -18402,8 +18361,8 @@ def _job_snapshot_anchor_ms(self, issued_at_ms: int) -> int: flush=True, ) if floor_ms is None: - return issued_at_ms - 1 - return min(issued_at_ms - 1, floor_ms - 1) + return issued_at_ms + return min(issued_at_ms, floor_ms - 1) def pending_share_from_submission( self, diff --git a/tests/test_prism_coordinator_job_cache.py b/tests/test_prism_coordinator_job_cache.py index 2dfbe91..c4b36f0 100644 --- a/tests/test_prism_coordinator_job_cache.py +++ b/tests/test_prism_coordinator_job_cache.py @@ -396,13 +396,10 @@ def test_payout_artifact_declares_its_own_snapshot_anchor(self) -> None: server, _rpc = coordinator(ledger=ledger) install_fake_bundle_builder(server) artifacts = server.current_template_artifacts() - clamp_now_ms = now_ms() - 6 - # Anchor selection sits strictly below the clamp instant so a share - # stamped in the same millisecond can never tie the anchor. - artifact_anchor_ms = clamp_now_ms - 1 + artifact_anchor_ms = now_ms() - 6 with patch( "lab.prism.prism_coordinator.now_ms", - return_value=clamp_now_ms, + return_value=artifact_anchor_ms, ): artifact = server._build_payout_ledger_artifact( 0, 0, artifacts.network_difficulty @@ -426,14 +423,13 @@ def test_payout_artifact_declares_its_own_snapshot_anchor(self) -> None: ) self.assertGreater(bundle.issued_at_ms, int(artifact.snapshot_anchor_ms)) - def test_artifact_paths_proceed_at_the_clamped_anchor_floor( + def test_artifact_paths_refuse_while_a_pending_commit_holds_the_floor( self, ) -> None: - # The pending-commit clamp is anchor selection, not a fence: every - # share stamped at or below the clamped anchor is already durable, - # so the window read at that anchor is exact and reproducible, and - # both artifact producers publish it. Shares stamped above the - # anchor deterministically belong to the next window. + # Stamping and writer enqueue are not atomic, so while a pending + # commit clamps the anchor a later-stamped share may already be + # durable; the global accepted count cannot be scoped to a clamped + # anchor, and neither artifact producer may bind them. ledger = AnchorRecordingLedger() server, _rpc = coordinator(ledger=ledger) install_fake_bundle_builder(server) @@ -441,75 +437,24 @@ def test_artifact_paths_proceed_at_the_clamped_anchor_floor( stamped_ms = now_ms() - 5 share = stamped_pending_share(stamped_ms) self._hold_floor(server, share) - try: - artifact = server._build_payout_ledger_artifact( - 0, 0, artifacts.network_difficulty - ) - self.assertIsNotNone(artifact) - assert artifact is not None - self.assertEqual(artifact.snapshot_anchor_ms, stamped_ms - 1) - self.assertEqual(ledger.anchors[-1], stamped_ms - 1) - - # The synchronous build proceeds at the clamped anchor and seeds - # its window for reuse under that same anchor; cache publication - # installs the seed. - bundle = server.build_shared_job_bundle(artifacts, worker()) - self.assertEqual(bundle.issued_at_ms, stamped_ms - 1) - seeded = bundle.prepared_ledger_artifact - self.assertIsNotNone(seeded) - assert seeded is not None - self.assertEqual(seeded.snapshot_anchor_ms, stamped_ms - 1) - server._install_payout_ledger_artifact(seeded) - with server._job_cache_lock: - published = server._payout_ledger_artifact - self.assertIsNotNone(published) - assert published is not None - self.assertEqual(published.snapshot_anchor_ms, stamped_ms - 1) - finally: - server._finish_pending_share_commit(share) - - def test_background_build_refuses_a_pathologically_old_floor(self) -> None: - # A floor held further below now than the reuse staleness bound (a - # wedged writer or leaked release) would arm an artifact that is - # dead on arrival; refuse before paying the window walk so the - # re-arm backoff paces retries. - ledger = AnchorRecordingLedger() - server, _rpc = coordinator(ledger=ledger) - install_fake_bundle_builder(server) - artifacts = server.current_template_artifacts() - share = stamped_pending_share(now_ms() - 30_000) - self._hold_floor(server, share) try: self.assertIsNone( server._build_payout_ledger_artifact( 0, 0, artifacts.network_difficulty ) ) + # Refused before paying the window walk. self.assertEqual(ledger.snapshot_calls, 0) - finally: - server._finish_pending_share_commit(share) - - def test_anchor_selection_excludes_same_millisecond_stamps(self) -> None: - # Millisecond granularity can hand a share stamped right after - # anchor selection the same accepted_at_ms as the clamp instant, - # and such a share is never protected by the pending floor. The - # window predicate is anchor-inclusive, so the anchor must sit - # strictly below the clamp-time millisecond or that share would - # join audit replays of a window that never contained it. - server, _rpc = coordinator(ledger=AnchorRecordingLedger()) - clamp_now = now_ms() - self.assertEqual( - server._job_snapshot_anchor_ms(clamp_now), - clamp_now - 1, - ) - share = stamped_pending_share(clamp_now - 7) - self._hold_floor(server, share) - try: - self.assertEqual( - server._job_snapshot_anchor_ms(clamp_now), - clamp_now - 8, - ) + # The synchronous build itself proceeds at the clamped anchor, + # but publishing its window for reuse is refused. + bundle = server.build_shared_job_bundle(artifacts, worker()) + self.assertEqual(bundle.issued_at_ms, stamped_ms - 1) + with server._job_cache_lock: + self.assertIsNone(server._payout_ledger_artifact) + self.assertIsNone( + server._job_build_anchor_counts.get(artifacts.generation) + ) finally: server._finish_pending_share_commit(share) @@ -2172,11 +2117,11 @@ def test_sync_ledger_snapshot_publishes_reusable_artifact(self) -> None: install_fake_bundle_builder(server) first = server.store_template_artifacts(dict(rpc.template)) assert first is not None - # Pin distinct per-generation anchors near the live clock so the - # armed window stays inside the reuse staleness bound. - first_anchor_ms = now_ms() - 20 with server._job_cache_lock: - server._job_build_issued_at_ms[first.generation] = first_anchor_ms + server._job_build_issued_at_ms[first.generation] = 1_700_000_000_000 + server._job_build_anchor_counts[first.generation] = len( + server.ledger.miners + ) inline = server.shared_job_bundle(first, mode="ready") @@ -2204,7 +2149,10 @@ def test_sync_ledger_snapshot_publishes_reusable_artifact(self) -> None: assert second is not None with server._job_cache_lock: server._job_build_issued_at_ms[second.generation] = ( - first_anchor_ms + 10 + 1_700_000_001_000 + ) + server._job_build_anchor_counts[second.generation] = len( + server.ledger.miners ) reused = server.shared_job_bundle(second, mode="ready") @@ -2218,7 +2166,7 @@ def test_sync_ledger_snapshot_publishes_reusable_artifact(self) -> None: artifact.snapshot_anchor_ms, ) - def test_sync_artifact_publication_survives_share_commit_mid_read( + def test_sync_artifact_publication_refused_when_share_commits_mid_read( self, ) -> None: class MidReadCommitLedger(FakeLedger): @@ -2232,10 +2180,8 @@ def snapshot_at_job_issue( anchor_job_issued_at_ms, window_weight=window_weight, ) - # A share becomes durable while the window read is in - # flight. The read is scoped by the frozen anchor -- the - # commit lands above it and belongs to the next window -- so - # anchor-scoped publication proceeds. + # A share becomes durable while the window read is in flight, + # so the read is ambiguously bound to either accepted count. self.miners = [*self.miners, f"mid-read-{self.snapshot_calls}"] return result @@ -2249,28 +2195,9 @@ def snapshot_at_job_issue( self.assertFalse(bundle.collection_only) self.assertEqual(bundle.payout_artifact_generation, 0) with server._job_cache_lock: - artifact = server._payout_ledger_artifact - self.assertIsNotNone(artifact) - assert artifact is not None - self.assertEqual(artifact.snapshot_anchor_ms, bundle.issued_at_ms) - self.assertEqual(list(artifact.shares_json), bundle.shares_json) + self.assertIsNone(server._payout_ledger_artifact) - # The armed window keeps serving under continuous commits: the live - # durable count has moved past the artifact's, and reuse must not - # care. - second = server.store_template_artifacts( - base_template(height=11, prevhash="22" * 32) - ) - assert second is not None - reused = server.shared_job_bundle(second, mode="ready") - self.assertEqual(reused.payout_artifact_generation, artifact.generation) - self.assertEqual(server.ledger.snapshot_calls, 1) - self.assertEqual( - reused.found_block["anchor_job_issued_at_ms"], - artifact.snapshot_anchor_ms, - ) - - def test_sync_publication_allowed_when_anchor_predates_durable_share( + def test_sync_artifact_publication_refused_when_anchor_predates_share( self, ) -> None: server, rpc = coordinator() @@ -2292,16 +2219,11 @@ def test_sync_publication_allowed_when_anchor_predates_durable_share( second = server.shared_job_bundle(artifacts, mode="ready") self.assertEqual(second.payout_artifact_generation, 0) - # The window read at the frozen anchor deterministically excludes - # the later durable share -- it belongs to the next window -- so the - # anchor-scoped publication proceeds even though the live count has - # moved past the frozen anchor. + # The before/after counts agree (4 == 4) but exceed the count that + # was scoped to the frozen anchor, so the window read at that anchor + # excludes a durable share and must not be published for reuse. with server._job_cache_lock: - republished = server._payout_ledger_artifact - self.assertIsNotNone(republished) - assert republished is not None - self.assertEqual(republished.snapshot_anchor_ms, second.issued_at_ms) - self.assertEqual(second.issued_at_ms, first.issued_at_ms) + self.assertIsNone(server._payout_ledger_artifact) def test_rearmed_artifact_is_not_shadowed_by_stale_cached_bundle( self, @@ -2338,8 +2260,8 @@ def test_rearmed_artifact_is_not_shadowed_by_stale_cached_bundle( self.assertEqual(len(rebuilt.shares_json), 4) self.assertEqual(recorded["calls"], 2) - def test_racing_commit_does_not_disable_sync_artifact_seeding(self) -> None: - class RacingStatsLedger(FakeLedger): + def test_anchor_count_capture_retries_a_racing_commit(self) -> None: + class BracketRaceLedger(FakeLedger): def __init__(self) -> None: super().__init__( miners=["miner-a", "miner-b", "miner-c", "miner-d"] @@ -2357,24 +2279,26 @@ def accepted_share_stats(self) -> dict[str, int]: "distinct_miner_count": 4, } - server, rpc = coordinator(ledger=RacingStatsLedger()) + server, rpc = coordinator(ledger=BracketRaceLedger()) install_fake_bundle_builder(server) artifacts = server.store_template_artifacts(dict(rpc.template)) assert artifacts is not None - # A commit races the informational count read. The window is scoped - # by the frozen anchor regardless of which side of the read the - # commit landed on, so seeding proceeds; the count is diagnostics, - # not a fence. + # The first bracket races a commit (3 then 4); the bounded retry + # captures a stable bracket instead of disabling synchronous + # artifact seeding for the whole generation. bundle = server.shared_job_bundle(artifacts, mode="ready") self.assertFalse(bundle.collection_only) with server._job_cache_lock: + self.assertEqual( + server._job_build_anchor_counts.get(artifacts.generation), + 4, + ) artifact = server._payout_ledger_artifact self.assertIsNotNone(artifact) assert artifact is not None - self.assertEqual(artifact.snapshot_anchor_ms, bundle.issued_at_ms) - self.assertEqual(list(artifact.shares_json), bundle.shares_json) + self.assertEqual(artifact.accepted_share_count, 4) def test_same_window_background_rebuild_keeps_artifact_generation( self, @@ -2392,8 +2316,7 @@ def test_same_window_background_rebuild_keeps_artifact_generation( # No share committed since: the speculative rebuild reads the same # window under a fresh anchor and must not spin the generation, # which would re-key bundle lookups for nothing -- but it is still a - # successful preparation, so an accumulated re-arm backoff releases - # and the fresher anchor advances the staleness clock in place. + # successful preparation, so an accumulated re-arm backoff releases. with server._payout_artifact_executor_lock: server._payout_artifact_rearm_backoff = 4 server._prepare_payout_ledger_artifact( @@ -2401,19 +2324,7 @@ def test_same_window_background_rebuild_keeps_artifact_generation( artifacts.network_difficulty, ) with server._job_cache_lock: - refreshed = server._payout_ledger_artifact - assert refreshed is not None - self.assertEqual(refreshed.generation, installed.generation) - self.assertEqual( - refreshed.share_snapshot_sha256, - installed.share_snapshot_sha256, - ) - assert installed.snapshot_anchor_ms is not None - assert refreshed.snapshot_anchor_ms is not None - self.assertGreaterEqual( - int(refreshed.snapshot_anchor_ms), - int(installed.snapshot_anchor_ms), - ) + self.assertIs(server._payout_ledger_artifact, installed) with server._payout_artifact_executor_lock: self.assertEqual(server._payout_artifact_rearm_backoff, 1) @@ -2438,12 +2349,11 @@ def test_delayed_older_snapshot_does_not_regress_artifact(self) -> None: with server._job_cache_lock: current = server._payout_ledger_artifact assert current is not None - assert current.snapshot_anchor_ms is not None + self.assertEqual(current.accepted_share_count, 3) - # A snapshot taken at an earlier anchor finishes its window - # conversion late, so its preparation timestamp is newer than the - # installed artifact's; the anchor still proves it is the older - # window. + # A snapshot read at an earlier count finishes its window conversion + # late, so its preparation timestamp is newer than the installed + # artifact's; the count still proves it is the older window. older_window = [ {"share_seq": seq, "miner_id": "miner-a"} for seq in (1, 2) ] @@ -2455,15 +2365,15 @@ def test_delayed_older_snapshot_does_not_regress_artifact(self) -> None: shares_json=tuple(older_window), prior_balances=(), prepared_monotonic=time.monotonic(), - snapshot_anchor_ms=int(current.snapshot_anchor_ms) - 5, + snapshot_anchor_ms=current.snapshot_anchor_ms, share_snapshot_sha256=canonical_json_sha256(older_window), ) server._install_payout_ledger_artifact(delayed) with server._job_cache_lock: self.assertIs(server._payout_ledger_artifact, current) - # A window snapshotted at a fresher anchor replaces regardless of - # its preparation timestamp ordering. + # A genuinely newer window replaces regardless of its preparation + # timestamp ordering. newer_window = [ {"share_seq": seq, "miner_id": "miner-a"} for seq in (1, 2, 3, 4) ] @@ -2472,7 +2382,6 @@ def test_delayed_older_snapshot_does_not_regress_artifact(self) -> None: accepted_share_count=4, shares_json=tuple(newer_window), prepared_monotonic=current.prepared_monotonic - 1.0, - snapshot_anchor_ms=int(current.snapshot_anchor_ms) + 5, share_snapshot_sha256=canonical_json_sha256(newer_window), ) server._install_payout_ledger_artifact(fresher) @@ -2516,8 +2425,8 @@ def test_difficulty_change_replaces_same_window_artifact(self) -> None: self.assertEqual(replaced.network_difficulty, retarget_difficulty) # A pre-retarget build delayed after its snapshot carries the same - # window at the old difficulty; whether its anchor trails or ties, - # the live-difficulty artifact must stay. + # count at the old difficulty; equal counts cannot order snapshots + # across a retarget, so the live-difficulty artifact must stay. delayed = dataclass_replace( installed, generation=0, @@ -2527,20 +2436,6 @@ def test_difficulty_change_replaces_same_window_artifact(self) -> None: with server._job_cache_lock: self.assertIs(server._payout_ledger_artifact, replaced) - # Even a delayed pre-retarget build that clamped a fresher anchor - # must not displace it: a wrong-difficulty install would fail every - # reuse probe on the difficulty check, which never re-arms. - assert replaced.snapshot_anchor_ms is not None - leading = dataclass_replace( - installed, - generation=0, - prepared_monotonic=time.monotonic(), - snapshot_anchor_ms=int(replaced.snapshot_anchor_ms) + 5, - ) - server._install_payout_ledger_artifact(leading) - with server._job_cache_lock: - self.assertIs(server._payout_ledger_artifact, replaced) - def test_fence_failure_rearms_artifact_preparation_with_debounce(self) -> None: server, rpc = coordinator() install_fake_bundle_builder(server) @@ -2573,16 +2468,7 @@ def submit(self, _fn: object) -> Future[None]: with server._payout_artifact_executor_lock: self.assertIsNone(server._payout_artifact_requested) - # Age the armed window past the reuse staleness bound. - server.payout_artifact_max_anchor_age_seconds = 10.0 - with server._job_cache_lock: - armed = server._payout_ledger_artifact - assert armed is not None - assert armed.snapshot_anchor_ms is not None - server._payout_ledger_artifact = dataclass_replace( - armed, - snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 11_000, - ) + server.ledger.miners = [*server.ledger.miners, "late-share"] self.assertIsNone( server._usable_payout_ledger_artifact( @@ -2633,32 +2519,44 @@ def submit(self, _fn: object) -> Future[None]: ) def test_aborted_speculative_rebuilds_back_off_and_reset(self) -> None: - ledger = AnchorRecordingLedger() + class MidReadLedger(FakeLedger): + def __init__(self) -> None: + super().__init__() + self.commit_mid_read = True + + def snapshot_at_job_issue( + self, + anchor_job_issued_at_ms: int, + *, + window_weight: int | None = None, + ) -> list[FakeShare]: + result = super().snapshot_at_job_issue( + anchor_job_issued_at_ms, + window_weight=window_weight, + ) + if self.commit_mid_read: + self.miners = [ + *self.miners, + f"mid-read-{self.snapshot_calls}", + ] + return result + + ledger = MidReadLedger() server, rpc = coordinator(ledger=ledger) install_fake_bundle_builder(server) artifacts = server.store_template_artifacts(dict(rpc.template)) assert artifacts is not None server.payout_artifact_rearm_min_seconds = 5.0 - # A pathologically old pending-commit floor aborts the rebuild - # before the window walk (the artifact would arm already past the - # staleness bound); each abort doubles the re-arm interval instead - # of retrying the reward-window walk at the floor forever. - share = stamped_pending_share(now_ms() - 30_000) - server._ensure_pending_share_commit_state() - with server._pending_share_commit_lock: - server._pending_share_commit_floor[id(share)] = [ - share, - time.monotonic(), - False, - ] + # Continuous writes abort the fenced rebuild; each abort doubles the + # re-arm interval instead of retrying the reward-window walk at the + # floor forever. server._prepare_payout_ledger_artifact(0, artifacts.network_difficulty) with server._payout_artifact_executor_lock: self.assertEqual(server._payout_artifact_rearm_backoff, 2) server._prepare_payout_ledger_artifact(0, artifacts.network_difficulty) with server._payout_artifact_executor_lock: self.assertEqual(server._payout_artifact_rearm_backoff, 4) - self.assertEqual(ledger.snapshot_calls, 0) # Elapsed time beyond the floor but inside the scaled interval must # not re-arm. @@ -2674,7 +2572,7 @@ def test_aborted_speculative_rebuilds_back_off_and_reset(self) -> None: self.assertIsNone(server._payout_artifact_requested) # A rebuild that finally arms resets the backoff to the floor. - server._finish_pending_share_commit(share) + ledger.commit_mid_read = False server._prepare_payout_ledger_artifact(0, artifacts.network_difficulty) with server._job_cache_lock: self.assertIsNotNone(server._payout_ledger_artifact) @@ -2741,15 +2639,7 @@ def test_landed_preview_suppresses_fence_failure_rearm(self) -> None: block_height=int(rpc.template["height"]) - 1, ) - # Age the armed window past the reuse staleness bound. - with server._job_cache_lock: - armed = server._payout_ledger_artifact - assert armed is not None - assert armed.snapshot_anchor_ms is not None - server._payout_ledger_artifact = dataclass_replace( - armed, - snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 11_000, - ) + server.ledger.miners = [*server.ledger.miners, "late-share"] self.assertIsNone( server._usable_payout_ledger_artifact( @@ -2762,397 +2652,6 @@ def test_landed_preview_suppresses_fence_failure_rearm(self) -> None: # durable-confirmation call site. self.assertEqual(scheduled, []) - def test_artifact_reuse_is_bounded_by_anchor_age(self) -> None: - server, rpc = coordinator() - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - armed = server._payout_ledger_artifact - assert armed is not None - assert armed.snapshot_anchor_ms is not None - - # Inside the bound the artifact serves even though shares landed - # after its anchor: they belong to the next window by construction, - # so the moved durable count is irrelevant to reuse validity. - # (PRISM_PAYOUT_ARTIFACT_MAX_ANCHOR_AGE_SECONDS wires this same - # attribute in __init__.) - server.payout_artifact_max_anchor_age_seconds = 5.0 - server.ledger.miners = [*server.ledger.miners, "late-1", "late-2"] - with server._job_cache_lock: - server._payout_ledger_artifact = dataclass_replace( - armed, - snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 4_000, - ) - self.assertIsNotNone( - server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - ) - - # Past the bound the window is retired for new reuse decisions. - with server._job_cache_lock: - server._payout_ledger_artifact = dataclass_replace( - armed, - snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 6_000, - ) - self.assertIsNone( - server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - rearm_on_fence_failure=False, - ) - ) - - def test_in_flight_build_keeps_aged_artifact_it_already_selected( - self, - ) -> None: - server, rpc = coordinator() - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - armed = server._payout_ledger_artifact - assert armed is not None - assert armed.snapshot_anchor_ms is not None - aged = dataclass_replace( - armed, - snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 11_000, - ) - server._payout_ledger_artifact = aged - - # New reuse decisions reject the aged window... - self.assertIsNone( - server._usable_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - rearm_on_fence_failure=False, - ) - ) - # ...but a build that selected it while fresh completes with it: the - # in-build re-validation checks supersession and the balances fence - # only, so queue delay past the bound cannot scrap the reuse into - # the full snapshot it was armed to avoid. - server.ledger.snapshot_calls = 0 - bundle = server.build_shared_job_bundle( - artifacts, - worker(), - payout_artifact=aged, - ) - self.assertEqual(server.ledger.snapshot_calls, 0) - self.assertEqual( - bundle.found_block["anchor_job_issued_at_ms"], - aged.snapshot_anchor_ms, - ) - - def test_in_flight_build_survives_same_window_anchor_refresh(self) -> None: - server, rpc = coordinator() - install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - armed = server._payout_ledger_artifact - assert armed is not None - assert armed.snapshot_anchor_ms is not None - # An in-flight build selected the armed artifact while it - # carried an older anchor. - selected = dataclass_replace( - armed, - snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 5, - ) - server._payout_ledger_artifact = selected - - # A same-window rebuild refreshes the anchor in place: the stored - # instance swaps while the generation -- the re-key authority -- - # and the window bytes stay identical. - server._prepare_payout_ledger_artifact( - server._payout_state_generation, - artifacts.network_difficulty, - ) - with server._job_cache_lock: - refreshed = server._payout_ledger_artifact - assert refreshed is not None - self.assertIsNot(refreshed, selected) - self.assertEqual(refreshed.generation, selected.generation) - self.assertEqual( - refreshed.share_snapshot_sha256, - selected.share_snapshot_sha256, - ) - - # The build holding the pre-refresh instance completes its reuse - # instead of being scrapped into a full snapshot by the swap. - server.ledger.snapshot_calls = 0 - bundle = server.build_shared_job_bundle( - artifacts, - worker(), - payout_artifact=selected, - ) - self.assertEqual(server.ledger.snapshot_calls, 0) - self.assertEqual( - bundle.found_block["anchor_job_issued_at_ms"], - selected.snapshot_anchor_ms, - ) - - def test_cached_ready_bundle_respects_anchor_age_bound(self) -> None: - server, rpc = coordinator() - recorded = install_fake_bundle_builder(server) - artifacts = server.store_template_artifacts(dict(rpc.template)) - assert artifacts is not None - with server._payout_artifact_executor_lock: - server._payout_artifact_executor_shutdown = True - - first = server.shared_job_bundle(artifacts, mode="ready") - self.assertEqual(recorded["calls"], 1) - # Inside the bound the cached bundle keeps serving. - server.shared_job_bundle(artifacts, mode="ready") - self.assertEqual(recorded["calls"], 1) - - # The bundle cache TTL and the artifact staleness bound are - # independent knobs: an entry whose declared window anchor outlives - # the bound must stop serving even while its TTL is still running, - # or jobs would keep declaring an anchor older than the configured - # limit. - aged_anchor_ms = int(first.found_block["anchor_job_issued_at_ms"]) - 11_000 - with server._job_cache_lock: - for key, entry in list(server._job_bundle_cache.items()): - aged_found_block = dict(entry.found_block) - aged_found_block["anchor_job_issued_at_ms"] = aged_anchor_ms - server._job_bundle_cache[key] = dataclass_replace( - entry, - found_block=aged_found_block, - ) - armed = server._payout_ledger_artifact - assert armed is not None - assert armed.snapshot_anchor_ms is not None - server._payout_ledger_artifact = dataclass_replace( - armed, - snapshot_anchor_ms=int(armed.snapshot_anchor_ms) - 11_000, - ) - - server.shared_job_bundle(artifacts, mode="ready") - self.assertEqual(recorded["calls"], 2) - - def test_reused_anchor_bundle_preview_matches_fresh_build_through_guard( - self, - ) -> None: - # The #50 payout-preview guard compares the preview computed from - # the issued job against one recomputed at landing. A reused-anchor - # bundle must therefore reproduce, byte for byte, what a fresh - # ledger read at the same anchor produces -- the audit - # reproducibility contract the artifact's snapshot_anchor_ms - # declaration documents. - programs = { - "miner-a": "aa" * 32, - "miner-b": "bb" * 32, - "miner-c": "cc" * 32, - "miner-d": "dd" * 32, - } - - class AnchorScopedLedger(FakeLedger): - """Window reads respect the anchor like the real ledger.""" - - def __init__(self) -> None: - super().__init__(miners=[]) - self.stamped: list[tuple[int, FakeShare]] = [] - - def add_share(self, miner_id: str, accepted_at_ms: int) -> None: - self.stamped.append( - ( - int(accepted_at_ms), - FakeShare( - miner_id=miner_id, - share_seq=len(self.stamped) + 1, - ), - ) - ) - - def accepted_share_stats(self) -> dict[str, int]: - self.stats_calls += 1 - return { - "accepted_share_count": len(self.stamped), - "distinct_miner_count": len( - {share.miner_id for _, share in self.stamped} - ), - } - - def snapshot_at_job_issue( - self, - anchor_job_issued_at_ms: int, - *, - window_weight: int | None = None, - ) -> list[FakeShare]: - self.snapshot_calls += 1 - return [ - share - for stamp, share in self.stamped - if stamp <= int(anchor_job_issued_at_ms) - ] - - anchor_ms = now_ms() - 50 - ledger = AnchorScopedLedger() - ledger.add_share("miner-a", anchor_ms - 30) - ledger.add_share("miner-b", anchor_ms - 20) - ledger.add_share("miner-c", anchor_ms - 10) - server, rpc = coordinator(ledger=ledger) - - def preview_bundle_builder(**kwargs: object) -> dict[str, object]: - suffix_hex = str(kwargs["coinbase_script_sig_suffix_hex"]) - weights: dict[str, int] = {} - for share in kwargs["shares"]: # type: ignore[union-attr] - miner = str(share["miner_id"]) # type: ignore[index] - weights[miner] = weights.get(miner, 0) + 1 - return { - "found_block": dict(kwargs["found_block"]), # type: ignore[call-overload] - "payout_policy_manifest": { - "accounts": [ - { - "account_type": "miner", - "recipient_id": miner, - "order_key": miner, - "p2mr_program_hex": programs[miner], - "carry_forward_balance_sats": 1_000 * weight, - } - for miner, weight in sorted(weights.items()) - ] - }, - "signed_coinbase_manifest": { - "manifest": { - "coinbase_tx_hex": synthetic_manifest_coinbase_hex( - suffix_hex - ), - } - }, - } - - server.build_audit_bundle = preview_bundle_builder # type: ignore[method-assign] - - first = server.store_template_artifacts(dict(rpc.template)) - assert first is not None - with server._job_cache_lock: - server._job_build_issued_at_ms[first.generation] = anchor_ms - - seeded_bundle = server.shared_job_bundle(first, mode="ready") - self.assertEqual(seeded_bundle.payout_artifact_generation, 0) - with server._job_cache_lock: - artifact = server._payout_ledger_artifact - assert artifact is not None - self.assertEqual(artifact.snapshot_anchor_ms, anchor_ms) - - # Shares keep landing after the anchor; the durable count moves past - # the artifact's, which must not matter for reuse. - ledger.add_share("miner-a", anchor_ms + 20) - ledger.add_share("miner-d", anchor_ms + 25) - - second = server.store_template_artifacts( - base_template(height=11, prevhash="22" * 32) - ) - assert second is not None - reused = server.shared_job_bundle(second, mode="ready") - self.assertEqual(reused.payout_artifact_generation, artifact.generation) - self.assertEqual( - reused.found_block["anchor_job_issued_at_ms"], - anchor_ms, - ) - - # Control: a fresh synchronous build pinned to the same anchor pays - # its own window read and must reproduce the reused output exactly, - # excluding the post-anchor shares deterministically. - with server._job_cache_lock: - server._payout_ledger_artifact = None - server._job_bundle_cache.clear() - third = server.store_template_artifacts( - base_template(height=12, prevhash="33" * 32) - ) - assert third is not None - with server._job_cache_lock: - server._job_build_issued_at_ms[third.generation] = anchor_ms - snapshot_calls_before = ledger.snapshot_calls - fresh = server.shared_job_bundle(third, mode="ready") - self.assertEqual(ledger.snapshot_calls, snapshot_calls_before + 1) - self.assertEqual(fresh.payout_artifact_generation, 0) - - self.assertEqual(reused.shares_json, fresh.shares_json) - self.assertEqual( - fresh.found_block["anchor_job_issued_at_ms"], - anchor_ms, - ) - assert reused.prospective_prior_balances is not None - assert fresh.prospective_prior_balances is not None - self.assertEqual( - canonical_json_text(list(reused.prospective_prior_balances)), - canonical_json_text(list(fresh.prospective_prior_balances)), - ) - - # A window at a fresher anchor covers the post-anchor shares and - # produces a different preview; built now to prove the guard - # equality below is load-bearing. - with server._job_cache_lock: - server._payout_ledger_artifact = None - server._job_bundle_cache.clear() - fourth = server.store_template_artifacts( - base_template(height=13, prevhash="44" * 32) - ) - assert fourth is not None - with server._job_cache_lock: - server._job_build_issued_at_ms[fourth.generation] = anchor_ms + 30 - divergent = server.shared_job_bundle(fourth, mode="ready") - assert divergent.prospective_prior_balances is not None - self.assertNotEqual( - canonical_json_text(list(divergent.prospective_prior_balances)), - canonical_json_text(list(reused.prospective_prior_balances)), - ) - - # The #50 guard sequence at landing: the issued preview (from the - # reused-anchor job) publishes first; the verified preview - # recomputed at the same anchor must then publish idempotently. - block_hash = "d1" * 32 - server._begin_accepted_block_payout_preview( - block_hash, - block_height=int(rpc.template["height"]), - ) - issued = server._materialize_prior_balance_preview( - reused.prospective_prior_balances - ) - server._publish_accepted_block_payout_preview(block_hash, issued) - generation_after_issued = server._payout_state_generation - verified = server._materialize_prior_balance_preview( - fresh.prospective_prior_balances - ) - server._publish_accepted_block_payout_preview(block_hash, verified) - self.assertEqual( - server._payout_state_generation, - generation_after_issued, - ) - with server._accepted_block_payout_preview_condition: - transition = server._accepted_block_payout_previews[block_hash] - self.assertEqual( - transition.preview, - server._serialize_prior_balance_preview(issued), - ) - - # A preview from any other anchor's window trips the guard exactly - # as submit_block_candidate's landing rebuild would. - with self.assertRaisesRegex(RuntimeError, "changed during retry"): - server._publish_accepted_block_payout_preview( - block_hash, - server._materialize_prior_balance_preview( - divergent.prospective_prior_balances - ), - ) - def test_new_tip_cancels_blocked_old_bundle_without_publication(self) -> None: old_tip = "11" * 32 new_tip = "22" * 32 diff --git a/tests/test_prism_coordinator_vardiff.py b/tests/test_prism_coordinator_vardiff.py index 68f8182..e5c8504 100644 --- a/tests/test_prism_coordinator_vardiff.py +++ b/tests/test_prism_coordinator_vardiff.py @@ -7150,11 +7150,9 @@ def submit() -> None: submitter.join(timeout=2) self.assertFalse(submitter.is_alive()) self.assertEqual(server._pending_share_commit_floor, {}) - # Drained floor: the anchor still sits strictly below the clamp - # instant so a same-millisecond stamp can never tie it. self.assertEqual( server._job_snapshot_anchor_ms(stamped_ms + 60_000), - stamped_ms + 60_000 - 1, + stamped_ms + 60_000, ) server.request_shutdown() writer.join(timeout=2) @@ -7179,7 +7177,7 @@ def append(self, pending: object) -> object: self.assertFalse(server._append_share_batch([entry])) self.assertTrue(entry.committed.is_set()) self.assertEqual(server._pending_share_commit_floor, {}) - self.assertEqual(server._job_snapshot_anchor_ms(10_000), 9_999) + self.assertEqual(server._job_snapshot_anchor_ms(10_000), 10_000) def test_failed_commit_releases_duplicate_key_for_exact_retry(self) -> None: server, state, healthy = submit_coordinator()