diff --git a/every_eval_ever/cron/flat_rebuild.py b/every_eval_ever/cron/flat_rebuild.py index 995c7ff08..98867a006 100644 --- a/every_eval_ever/cron/flat_rebuild.py +++ b/every_eval_ever/cron/flat_rebuild.py @@ -1046,8 +1046,9 @@ def _commit_batch( landed = self._landed(operations) if landed: return - if attempt < store.COMMIT_ATTEMPTS and store.is_commit_conflict( - exc + if ( + attempt < store.COMMIT_ATTEMPTS + and store.is_retryable_commit_error(exc) ): store.wait_before_retry(attempt) continue diff --git a/every_eval_ever/cron/store.py b/every_eval_ever/cron/store.py index 4727c3e73..28cc14a84 100644 --- a/every_eval_ever/cron/store.py +++ b/every_eval_ever/cron/store.py @@ -94,6 +94,14 @@ #: Hub errors name the status in their message, which is how a conflict is #: recognised when the exception carries no response to read it from. _STATUS_CODE = re.compile(r'\b(409|412)\b') +#: Statuses that mean the Hub itself faulted rather than refusing the commit. +#: The write was never applied, so replaying it is safe, and the same backoff +#: that waits out a lock waits out a bad minute on the Hub. +_TRANSIENT_STATUSES = frozenset({500, 502, 503, 504}) +#: The Hub's phrasing for a server fault. Matched against the whole clause +#: rather than the bare number, because a request id or a byte count in the +#: message would otherwise read as a status. +_SERVER_STATUS = re.compile(r"server error '(500|502|503|504)\b", re.IGNORECASE) def is_commit_conflict(exc: BaseException) -> bool: @@ -119,6 +127,25 @@ def is_commit_conflict(exc: BaseException) -> bool: return _STATUS_CODE.search(text) is not None +def is_transient_hub_error(exc: BaseException) -> bool: + """Return whether the Hub faulted on its own side and may yet succeed. + + A 5xx means the Hub failed to apply the commit, not that it refused one, + so nothing landed and replaying sends the same operations at a Hub that + has had time to recover. One fault therefore costs a wait rather than the + adapter's whole run, and the run's records for the day with it. + """ + status = getattr(getattr(exc, 'response', None), 'status_code', None) + if status in _TRANSIENT_STATUSES: + return True + return _SERVER_STATUS.search(str(exc)) is not None + + +def is_retryable_commit_error(exc: BaseException) -> bool: + """Return whether waiting and sending the same commit again is worthwhile.""" + return is_commit_conflict(exc) or is_transient_hub_error(exc) + + def commit_retry_delay(attempt: int) -> float: """Return the seconds to wait before retry ``attempt`` (1 is the first). @@ -599,14 +626,19 @@ def commit( ) except Exception as exc: # noqa: BLE001 - re-raised with context last = attempt == COMMIT_ATTEMPTS - if last or not is_commit_conflict(exc): + if last or not is_retryable_commit_error(exc): raise StoreError( f'could not write to {self.repo_id}: ' f'{type(exc).__name__}: {exc}' ) from exc + cause = ( + 'lost the per-repository lock' + if is_commit_conflict(exc) + else 'hit a Hub server error' + ) print( - f'{self.repo_id}: commit attempt {attempt} lost the ' - f'per-repository lock ({type(exc).__name__}), retrying', + f'{self.repo_id}: commit attempt {attempt} {cause} ' + f'({type(exc).__name__}), retrying', file=sys.stderr, ) wait_before_retry(attempt) @@ -755,6 +787,8 @@ def state_operations(state: AdapterState) -> list[CommitOperationAdd]: 'inflight_operation', 'inflight_path', 'is_commit_conflict', + 'is_retryable_commit_error', + 'is_transient_hub_error', 'pending_fingerprints_path', 'plan_raw_upload', 'raw_prefix', diff --git a/every_eval_ever/cron/submit.py b/every_eval_ever/cron/submit.py index dd341097b..046322530 100644 --- a/every_eval_ever/cron/submit.py +++ b/every_eval_ever/cron/submit.py @@ -277,12 +277,13 @@ def publish( if ( landed is not None and attempt < store.COMMIT_ATTEMPTS - and store.is_commit_conflict(exc) + and store.is_retryable_commit_error(exc) ): # Every adapter job of a matrix publishes to this one # branch, so the Hub's per-repository commit lock is - # contended. Retried only where the datastore proved - # the batch absent, so a retry cannot duplicate it. + # contended, and the Hub itself faults from time to + # time. Retried only where the datastore proved the + # batch absent, so a retry cannot duplicate it. store.wait_before_retry(attempt) continue unresolved: list[str] = [] diff --git a/tests/test_cron_store_and_submit.py b/tests/test_cron_store_and_submit.py index d32c46ae6..4e33f607b 100644 --- a/tests/test_cron_store_and_submit.py +++ b/tests/test_cron_store_and_submit.py @@ -1138,10 +1138,12 @@ def test_a_record_and_its_sidecar_are_never_split_across_commits( def test_a_failure_before_anything_landed_reports_nothing_committed( tmp_path, + retry_waits, ) -> None: + """A Hub that faults for the whole retry budget still reports honestly.""" tree = _upload_tree(tmp_path, 2) hub = FakeHub() - hub.commit_error = RuntimeError('502 Bad Gateway') + hub.commit_error = RuntimeError("Server error '502 Bad Gateway' for url...") sub = submit.DatastoreSubmitter(hub) with pytest.raises(submit.PartialSubmissionError) as caught: @@ -1408,3 +1410,64 @@ def test_is_commit_conflict_detects_precondition_failed_messages() -> None: assert store.is_commit_conflict(exc1) assert store.is_commit_conflict(exc2) assert store.is_commit_conflict(exc3) + + +def test_a_hub_server_error_is_retried_not_reported(retry_waits) -> None: + """A 5xx never applied the commit, so sending it again is the whole fix.""" + hub = FakeHub(sha='headsha') + raw_store = store.RawStore(hub) + state = raw_store.read_state('hle') + state.fingerprints.add('a') + faulted = _conflict(500, "Server error '500 Internal Server Error' for url...") + attempts: list[dict] = [] + real_create_commit = hub.create_commit + + def fault_once(**kwargs): + attempts.append(kwargs) + if len(attempts) == 1: + raise faulted + return real_create_commit(**kwargs) + + hub.create_commit = fault_once + + raw_store.commit(store.state_operations(state), message='hle') + + assert hub.files['state/hle.fingerprints'].split() == ['a'] + assert retry_waits == [1], 'one wait, before the second attempt' + + +def test_a_hub_that_stays_broken_still_fails(retry_waits) -> None: + """The retry budget bounds a Hub outage as well as it bounds a lock.""" + hub = FakeHub(sha='headsha') + raw_store = store.RawStore(hub) + state = raw_store.read_state('hle') + hub.commit_error = _conflict(503, "Server error '503 Service Unavailable'") + + with pytest.raises(store.StoreError, match='could not write'): + raw_store.commit(store.state_operations(state), message='hle') + + assert retry_waits == list(range(1, store.COMMIT_ATTEMPTS)) + + +def test_transient_and_conflict_stay_separate_judgements() -> None: + """Each predicate answers only its own question, so neither name lies.""" + server = RuntimeError("Server error '502 Bad Gateway' for url...") + lock = RuntimeError('409 Client Error: Conflict for url...') + denied = RuntimeError("Client error '403 Forbidden' for url...") + + assert store.is_transient_hub_error(server) + assert not store.is_commit_conflict(server) + assert store.is_commit_conflict(lock) + assert not store.is_transient_hub_error(lock) + assert not store.is_retryable_commit_error(denied) + + +def test_a_request_id_holding_a_status_number_is_not_a_status() -> None: + """Hub errors carry request ids; a 500 inside one must not read as a fault.""" + exc = RuntimeError( + "Client error '403 Forbidden' for url 'https://huggingface.co/api' " + '(Request ID: Root=1-500-502503504)' + ) + + assert not store.is_transient_hub_error(exc) + assert not store.is_retryable_commit_error(exc)