fix(retention): verify archive marker deletion

api/services/retention/workflow_run/bundle_archive_maintenance.py

@@ -25,13 +25,20 @@ from typing import Any, TypedDict, cast
 
 import pyarrow.parquet as pq
 import sqlalchemy as sa
+from botocore.exceptions import ClientError, HTTPClientError
+from botocore.exceptions import ConnectionError as BotoCoreConnectionError
 from sqlalchemy import delete, func, inspect, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.orm import Session, sessionmaker
 
 from extensions.ext_database import db
-from libs.archive_storage import ArchiveStorage, ArchiveStorageNotConfiguredError, get_archive_storage
+from libs.archive_storage import (
+    ArchiveStorage,
+    ArchiveStorageError,
+    ArchiveStorageNotConfiguredError,
+    get_archive_storage,
+)
 from models.trigger import WorkflowTriggerLog
 from models.workflow import (
     WorkflowAppLog,
@@ -871,7 +878,7 @@ class WorkflowRunBundleArchiveMaintenance:
         table_records: dict[str, list[dict[str, Any]]],
         live_records: dict[str, list[dict[str, Any]]],
     ) -> None:
-        """Require every live row in the bundle scope to exist unchanged in the validated archive."""
+        """Require every live row to match its archive row, except for null live-only fields."""
         manifest_run_ids = {str(run_id) for run_id in manifest["run_ids"]}
         if len(manifest_run_ids) != len(manifest["run_ids"]):
             raise ValueError("archive manifest contains duplicate workflow run IDs")
@@ -899,7 +906,16 @@ class WorkflowRunBundleArchiveMaintenance:
                 )
 
             archive_subset = [archive_by_id[row_id] for row_id in live_ids]
-            live_checksum = cls._records_checksum(live_records[table_name])
+            live_subset = []
+            for live_record, archive_record in zip(live_records[table_name], archive_subset, strict=True):
+                live_only_fields = live_record.keys() - archive_record.keys()
+                if live_only_fields and all(live_record[field] is None for field in live_only_fields):
+                    live_record = {
+                        field: value for field, value in live_record.items() if field not in live_only_fields
+                    }
+                live_subset.append(live_record)
+
+            live_checksum = cls._records_checksum(live_subset)
             archive_checksum = cls._records_checksum(archive_subset)
             if live_checksum != archive_checksum:
                 raise ValueError(
@@ -1198,8 +1214,68 @@ class WorkflowRunBundleArchiveMaintenance:
     @staticmethod
     def _delete_marker(storage: ArchiveStorage, object_prefix: str, marker_name: str) -> None:
         marker_key = f"{object_prefix}/{marker_name}"
-        if storage.object_exists(marker_key):
-            storage.delete_object(marker_key)
+        retry_delays = (0.5, 1.0)
+        retryable_error_codes = frozenset(
+            {
+                "408",
+                "429",
+                "500",
+                "502",
+                "503",
+                "504",
+                "InternalError",
+                "RequestLimitExceeded",
+                "RequestTimeout",
+                "RequestTimeoutException",
+                "ServiceUnavailable",
+                "SlowDown",
+                "Throttling",
+                "ThrottlingException",
+                "TooManyRequestsException",
+            }
+        )
+
+        def is_retryable(error: ArchiveStorageError) -> bool:
+            cause = error.__cause__
+            if isinstance(cause, BotoCoreConnectionError | HTTPClientError):
+                return True
+            if not isinstance(cause, ClientError):
+                return False
+            error_code = str(cause.response.get("Error", {}).get("Code", ""))
+            status_code = str(cause.response.get("ResponseMetadata", {}).get("HTTPStatusCode", ""))
+            return error_code in retryable_error_codes or status_code in retryable_error_codes
+
+        attempts = len(retry_delays) + 1
+        for attempt in range(attempts):
+            delete_error: ArchiveStorageError | None = None
+            try:
+                storage.delete_object(marker_key)
+            except ArchiveStorageError as error:
+                if not is_retryable(error):
+                    raise
+                delete_error = error
+                logger.warning(
+                    "Retryable archive marker delete failed; verifying marker state: key=%s, attempt=%s/%s",
+                    marker_key,
+                    attempt + 1,
+                    attempts,
+                    exc_info=True,
+                )
+
+            if not storage.object_exists(marker_key):
+                return
+            if attempt == attempts - 1:
+                raise ArchiveStorageError(
+                    f"Archive marker still exists after {attempts} delete attempts: {marker_key}"
+                ) from delete_error
+
+            logger.warning(
+                "Archive marker still exists after delete; retrying: key=%s, attempt=%s/%s",
+                marker_key,
+                attempt + 1,
+                attempts,
+            )
+            time.sleep(retry_delays[attempt])
 
     @staticmethod
     def _chunks(values: Sequence[Any], size: int) -> list[Sequence[Any]]:

api/tests/unit_tests/services/retention/workflow_run/test_bundle_archive_maintenance.py

@@ -4,9 +4,11 @@ from typing import Any, cast
 from unittest.mock import MagicMock, call, patch
 
 import pytest
+from botocore.exceptions import ClientError, ReadTimeoutError
 from sqlalchemy import event
 from sqlalchemy.orm import Session, sessionmaker
 
+from libs.archive_storage import ArchiveStorageError
 from models.workflow import WorkflowRunArchiveBundle
 from services.retention.workflow_run.bundle_archive_maintenance import (
     ARCHIVED_TABLES,
@@ -30,6 +32,12 @@ CATALOG_ID = "019f63b7-5ca4-7681-9ce0-800283608f39"
 BUNDLE_ID = "bundle-a"
 
 
+def _archive_storage_error(cause: BaseException) -> ArchiveStorageError:
+    error = ArchiveStorageError("marker delete failed")
+    error.__cause__ = cause
+    return error
+
+
 def _table_records(
     **overrides: list[dict[str, Any]],
 ) -> dict[str, list[dict[str, Any]]]:
@@ -468,6 +476,33 @@ def test_live_archive_subset_rejects_content_mismatch() -> None:
         )
 
 
+def test_live_archive_subset_accepts_null_live_field_missing_from_archive() -> None:
+    archive_records = _sample_archive_records()
+    manifest = _bundle_reference(_catalog_entry(), table_records=archive_records).manifest
+    live_records = {table_name: [dict(record) for record in records] for table_name, records in archive_records.items()}
+    live_records["workflow_node_executions"][0]["agent_workspace_binding_id"] = None
+
+    WorkflowRunBundleArchiveMaintenance._validate_live_archive_subset(
+        manifest,
+        archive_records,
+        live_records,
+    )
+
+
+def test_live_archive_subset_rejects_non_null_live_field_missing_from_archive() -> None:
+    archive_records = _sample_archive_records()
+    manifest = _bundle_reference(_catalog_entry(), table_records=archive_records).manifest
+    live_records = {table_name: [dict(record) for record in records] for table_name, records in archive_records.items()}
+    live_records["workflow_node_executions"][0]["agent_workspace_binding_id"] = "binding-1"
+
+    with pytest.raises(ValueError, match="subset content checksum mismatch for workflow_node_executions"):
+        WorkflowRunBundleArchiveMaintenance._validate_live_archive_subset(
+            manifest,
+            archive_records,
+            live_records,
+        )
+
+
 def test_live_bundle_scope_includes_archived_ids_and_indirect_children(
     unbound_session: Session, unbound_session_factory: sessionmaker[Session]
 ) -> None:
@@ -789,3 +824,132 @@ def test_mark_restored_clears_stale_delete_marker_before_releasing_restore_fence
         call.delete(storage, object_prefix, ARCHIVE_BUNDLE_DELETE_STARTED_MARKER_NAME),
         call.delete(storage, object_prefix, ARCHIVE_BUNDLE_RESTORE_STARTED_MARKER_NAME),
     ]
+
+
+@pytest.mark.parametrize(
+    "delete_cause",
+    [
+        ReadTimeoutError(endpoint_url="https://storage.example.com"),
+        ClientError({"Error": {"Code": "SlowDown"}}, "DeleteObject"),
+        ClientError(
+            {
+                "Error": {"Code": "Unknown"},
+                "ResponseMetadata": {"HTTPStatusCode": 503},
+            },
+            "DeleteObject",
+        ),
+    ],
+    ids=["read-timeout", "slow-down", "http-503"],
+)
+def test_delete_marker_accepts_retryable_delete_error_when_head_confirms_absent(
+    delete_cause: BaseException,
+) -> None:
+    storage = MagicMock()
+    storage.delete_object.side_effect = _archive_storage_error(delete_cause)
+    storage.object_exists.return_value = False
+
+    with patch("services.retention.workflow_run.bundle_archive_maintenance.time.sleep") as sleep:
+        WorkflowRunBundleArchiveMaintenance._delete_marker(storage, "bundle-prefix", "_MARKER")
+
+    storage.delete_object.assert_called_once_with("bundle-prefix/_MARKER")
+    storage.object_exists.assert_called_once_with("bundle-prefix/_MARKER")
+    sleep.assert_not_called()
+
+
+def test_delete_marker_retries_when_marker_remains_then_succeeds() -> None:
+    storage = MagicMock()
+    storage.delete_object.side_effect = [
+        _archive_storage_error(ReadTimeoutError(endpoint_url="https://storage.example.com")),
+        None,
+    ]
+    storage.object_exists.side_effect = [True, False]
+
+    with patch("services.retention.workflow_run.bundle_archive_maintenance.time.sleep") as sleep:
+        WorkflowRunBundleArchiveMaintenance._delete_marker(storage, "bundle-prefix", "_MARKER")
+
+    assert storage.delete_object.call_count == 2
+    assert storage.object_exists.call_count == 2
+    sleep.assert_called_once_with(0.5)
+
+
+def test_delete_marker_fails_closed_when_marker_persists() -> None:
+    storage = MagicMock()
+    storage.object_exists.return_value = True
+
+    with (
+        patch("services.retention.workflow_run.bundle_archive_maintenance.time.sleep") as sleep,
+        pytest.raises(ArchiveStorageError, match="still exists after 3 delete attempts"),
+    ):
+        WorkflowRunBundleArchiveMaintenance._delete_marker(storage, "bundle-prefix", "_MARKER")
+
+    assert storage.delete_object.call_count == 3
+    assert storage.object_exists.call_count == 3
+    assert sleep.call_args_list == [call(0.5), call(1.0)]
+
+
+def test_delete_marker_fails_closed_when_head_fails() -> None:
+    storage = MagicMock()
+    storage.object_exists.side_effect = ArchiveStorageError("marker HEAD failed")
+
+    with (
+        patch("services.retention.workflow_run.bundle_archive_maintenance.time.sleep") as sleep,
+        pytest.raises(ArchiveStorageError, match="marker HEAD failed"),
+    ):
+        WorkflowRunBundleArchiveMaintenance._delete_marker(storage, "bundle-prefix", "_MARKER")
+
+    storage.delete_object.assert_called_once_with("bundle-prefix/_MARKER")
+    storage.object_exists.assert_called_once_with("bundle-prefix/_MARKER")
+    sleep.assert_not_called()
+
+
+def test_delete_marker_does_not_retry_non_retryable_delete_error() -> None:
+    storage = MagicMock()
+    storage.delete_object.side_effect = _archive_storage_error(
+        ClientError(
+            {
+                "Error": {"Code": "AccessDenied"},
+                "ResponseMetadata": {"HTTPStatusCode": 403},
+            },
+            "DeleteObject",
+        )
+    )
+
+    with pytest.raises(ArchiveStorageError, match="marker delete failed"):
+        WorkflowRunBundleArchiveMaintenance._delete_marker(storage, "bundle-prefix", "_MARKER")
+
+    storage.delete_object.assert_called_once_with("bundle-prefix/_MARKER")
+    storage.object_exists.assert_not_called()
+
+
+def test_delete_bundle_with_deleted_marker_clears_stale_delete_started_marker(
+    sqlite_session: Session, sqlite_session_factory: sessionmaker[Session]
+) -> None:
+    entry = _catalog_entry()
+    _persist_catalog(sqlite_session, entry)
+    archive_records = _sample_archive_records()
+    bundle_ref = _bundle_reference(entry, table_records=archive_records)
+    storage = MagicMock()
+    existing_markers = {
+        f"{bundle_ref.object_prefix}/{ARCHIVE_BUNDLE_DELETED_MARKER_NAME}",
+        f"{bundle_ref.object_prefix}/{ARCHIVE_BUNDLE_DELETE_STARTED_MARKER_NAME}",
+    }
+    storage.object_exists.side_effect = lambda key: key in existing_markers
+    storage.delete_object.side_effect = lambda key: existing_markers.discard(key)
+    maintenance = WorkflowRunBundleArchiveMaintenance(session_factory=sqlite_session_factory)
+
+    with (
+        patch.object(
+            maintenance,
+            "_validate_archive_object",
+            return_value=(bundle_ref.manifest, archive_records, 123),
+        ),
+        patch.object(maintenance, "_load_live_bundle_records", return_value=_table_records()),
+    ):
+        result = maintenance._delete_bundle(sqlite_session, storage, bundle_ref)
+
+    assert result.success
+    assert existing_markers == {f"{bundle_ref.object_prefix}/{ARCHIVE_BUNDLE_DELETED_MARKER_NAME}"}
+    assert storage.delete_object.call_args_list == [
+        call(f"{bundle_ref.object_prefix}/{ARCHIVE_BUNDLE_DELETE_STARTED_MARKER_NAME}"),
+        call(f"{bundle_ref.object_prefix}/{ARCHIVE_BUNDLE_RESTORED_MARKER_NAME}"),
+    ]