Retry a Hub server error instead of ending the adapter's run

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

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',

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] = []

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)