fix: publish valid SRO memberships while preserving quarantined history

docs/sro-membership-source.md

@@ -120,14 +120,28 @@ static или reverse-proxy document root. Существующая retention-к
 fallback на public raw нет.
 
 Публикация записей, checkpoint организаций, load log и terminal job выполняется
-в одной транзакции. Инкрементальная публикация заменяет данные только успешно
-проверенного набора организаций; записи остальных организаций сохраняются.
-Transport/schema failure или отменённая job откатывает публикацию и checkpoints.
-Допустимый пустой ответ удаляет старые членства проверенной организации.
-Любая семантически некорректная строка остаётся в quarantine и отклоняет весь
-запуск с `sro_incomplete_membership_scan`. Это сохраняет прежние членства и
-checkpoints: неизвестный статус или неразрешённая ссылка не доказывают, что
-членство исчезло. Организация остаётся кандидатом следующего incremental запуска.
+в одной транзакции после завершения сбора всех кандидатов. Transport/schema failure,
+нарушение целостности или отменённая job откатывает публикацию и checkpoints целиком.
+Незавершённый batch остаётся диагностическим и не становится current snapshot.
+
+Завершённый сбор может закончиться `success` с quarantine, как в примере истории
+исходного контракта. Валидные строки публикуются, включая валидные членства организации,
+у которой другая строка оказалась некорректной. Неизвестный статус, отсутствующий
+SRO ID или URL, конфликт идентификаторов не исправляются догадками: исходная строка
+остаётся в quarantine с причиной. У таких организаций прежние членства, не обновлённые
+валидной строкой этого batch, сохраняются с прежними UID, payload и load batch.
+Ошибка в строке не доказывает прекращение других членств. Для полностью проверенных
+организаций исчезнувшие членства удаляются; допустимый пустой ответ удаляет их все.
+При incremental неизменённые, не опрошенные организации также сохраняются.
+
+Checkpoint полностью проверенной организации обновляется. У организации с quarantine
+прежний checkpoint удаляется в той же транзакции, чтобы следующий incremental повторил
+lookup даже при неизменившемся fingerprint. Raw artifact и staged audit сохраняются.
+Если нет ни одной валидной строки и ни одного подтверждённого пустого lookup `200`,
+quarantine отклоняет запуск с `sro_incomplete_membership_scan`; смесь invalid rows
+и неоднозначных `404` не считается подтверждённым отсутствием. Для полностью состоящего
+из `404` запуска сохраняется отдельный `sro_ambiguous_empty_scan`. В обоих случаях
+прежние записи и checkpoints остаются без изменений.
 
 Если incremental не нашёл изменившихся организаций, batch имеет нулевую дельту
 и наследует source registry date/version последней успешной публикации; ссылка
@@ -144,8 +158,16 @@ artifacts. Ошибка брокера после commit не превращае
 общий результат, включая неизменённые организации. `metadata.snapshot_records_count`
 и `snapshot_organizations_count` дают такие же общие record/distinct organization
 counts; `batch_published_records_count` и `updated_records_count` — число опубликованных
-строк текущей дельты. `parsed_count`, quarantine и candidate/queried counts относятся
-к текущей проверке; складывать их с общим published count нельзя.
+строк текущей дельты. При full с quarantine общий count также включает сохранённые
+старые членства. `retained_records_count` — число старых строк организаций с quarantine,
+которые не были обновлены валидным upsert и сохранены от удаления;
+`quarantined_organizations_count` — число организаций хотя бы с одной строкой quarantine;
+`completed_organizations_count` — число полностью проверенных организаций без quarantine,
+включая допустимые пустые результаты. Эти три поля доступны в metadata API.
+В приватном artifact `confirmed_empty_lookup_200_count` отдельно фиксирует подтверждённые
+пустые `200` для проверки допустимости публикации. `parsed_count`, quarantine и
+candidate/queried counts относятся к текущей проверке; складывать их с общим published
+count нельзя. Новая дата публикации не изменяет прежнюю lineage сохранённых записей.
 
 Источник публикует весь собственный справочник; карточка/list/dashboard для трёх
 новых источников используют весь published scope. Coverage ОПК показывается

src/apps/core/serializers.py

@@ -85,6 +85,9 @@ class SnapshotRunMetadataSerializer(serializers.Serializer):
     found_memberships_count = serializers.IntegerField(read_only=True)
     published_records_count = serializers.IntegerField(read_only=True)
     quarantined_records_count = serializers.IntegerField(read_only=True)
+    quarantined_organizations_count = serializers.IntegerField(read_only=True)
+    completed_organizations_count = serializers.IntegerField(read_only=True)
+    retained_records_count = serializers.IntegerField(read_only=True)
     missing_admission_dates_count = serializers.IntegerField(read_only=True)
     http_errors_count = serializers.IntegerField(read_only=True)
     parse_errors_count = serializers.IntegerField(read_only=True)

src/apps/parsers/registry_snapshots.py

@@ -147,22 +147,41 @@ def _invalidate_committed_snapshot_cache() -> None:
         )
 
 
-def _remove_obsolete_records(source, retained_ids, scope_organization_ids) -> None:
+def _remove_obsolete_records(
+    source, retained_ids, scope_organization_ids, preserve_organization_ids
+) -> int:
     obsolete = OrganizationSourceRecord.objects.filter(source=source)
     if scope_organization_ids is not None:
         obsolete = obsolete.filter(
             extension__organization_id__in=scope_organization_ids
         )
+    retained_count = 0
+    if preserve_organization_ids:
+        protected = obsolete.filter(
+            extension__organization_id__in=preserve_organization_ids
+        )
+        retained_count = protected.exclude(external_id__in=retained_ids).count()
+        obsolete = obsolete.exclude(
+            extension__organization_id__in=preserve_organization_ids
+        )
     obsolete.exclude(external_id__in=retained_ids).delete()
+    return retained_count
 
 
-def _set_published_count(artifact, source, published, scope_organization_ids) -> None:
+def _set_published_count(
+    artifact, source, published, scope_organization_ids, retained_count=None
+) -> None:
     artifact.published_count = published
-    if scope_organization_ids is not None:
+    if scope_organization_ids is not None or retained_count is not None:
         artifact.metadata = {**artifact.metadata, "updated_records_count": published}
         artifact.published_count = OrganizationSourceRecord.objects.filter(
             source=source
         ).count()
+    if retained_count is not None:
+        artifact.metadata = {
+            **artifact.metadata,
+            "retained_records_count": retained_count,
+        }
     organizations_count = (
         OrganizationSourceRecord.objects.filter(source=source)
         .order_by()
@@ -183,8 +202,13 @@ def publish_snapshot(
     *,
     on_publish: Callable[[ParserSourceArtifact, SnapshotResult], None] | None = None,
     scope_organization_ids: list[uuid.UUID] | None = None,
+    preserve_organization_ids: list[uuid.UUID] | None = None,
 ) -> SnapshotResult:
-    """Publish complete staging in one transaction, preserving stable record IDs."""
+    """Publish validated staging atomically; optionally retain uncertain old rows.
+
+    Preservation affects only obsolete deletion for the selected organizations.
+    Valid staged rows are still upserted, including their existing stable IDs.
+    """
     source = artifact.source
     descriptor = get_source_group_descriptor(source)
     staged = ParserStagedRecord.objects.filter(
@@ -253,13 +277,21 @@ def publish_snapshot(
                 raise SnapshotValidationError("organization_resolution_changed")
             published += len(inputs)
         retained_ids = staged.values_list("external_id", flat=True)
-        _remove_obsolete_records(source, retained_ids, scope_organization_ids)
+        retained_count = _remove_obsolete_records(
+            source, retained_ids, scope_organization_ids, preserve_organization_ids
+        )
         descriptor.extension_model.objects.filter(records__isnull=True).delete()
         # Deletion can change counts for extensions whose remaining rows were upserted
         # before old records were removed.
         _refresh_extension_counts(descriptor)
         staged.update(disposition=ParserStagedRecord.Disposition.PUBLISHED)
-        _set_published_count(artifact, source, published, scope_organization_ids)
+        _set_published_count(
+            artifact,
+            source,
+            published,
+            scope_organization_ids,
+            retained_count if preserve_organization_ids is not None else None,
+        )
         artifact.status = ParserSourceArtifact.Status.PUBLISHED
         artifact.save()
         result = _artifact_result(artifact)

src/apps/parsers/sro_membership.py

@@ -447,6 +447,7 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress):
     raw_count, not_found, missing_dates = 0, 0, 0
     parse_errors = 0
     recognized_lookup_404, successful_lookup_200 = 0, 0
+    confirmed_empty_lookup_200 = 0
     reasons, dates = Counter(), set()
 
     def fetch(url):
@@ -480,6 +481,7 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress):
             rows, version = parse_sro_lookup(page, expected_lookup_value=value)
             recognized_lookup_404 += page.status_code == 404
             successful_lookup_200 += page.status_code == 200
+            confirmed_empty_lookup_200 += page.status_code == 200 and not rows
             if version:
                 dates.add(version)
             if len(dates) > 1:
@@ -547,6 +549,7 @@ def _collect_sro_candidates(artifact, candidates, http, raw, on_progress):
             not_found_organizations_count=not_found,
             recognized_lookup_404_count=recognized_lookup_404,
             successful_lookup_200_count=successful_lookup_200,
+            confirmed_empty_lookup_200_count=confirmed_empty_lookup_200,
             raw_memberships_count=raw_count,
             found_memberships_count=raw_count,
             quarantined_records_count=sum(reasons.values()),
@@ -609,6 +612,34 @@ def _save_sro_checkpoints(candidates, artifact) -> None:
         )
 
 
+def _sro_publication_scope(artifact, candidates):
+    """Keep uncertain old memberships without discarding independently valid rows."""
+    staged = artifact.staged_records.all()
+    if staged.filter(organization_id__isnull=True).exists():
+        raise SnapshotValidationError("organization_resolution_changed")
+    quarantined_ids = set(
+        staged.filter(disposition=ParserStagedRecord.Disposition.QUARANTINED)
+        .order_by()
+        .values_list("organization_id", flat=True)
+        .distinct()
+    )
+    completed = [org for org in candidates if org.uid not in quarantined_ids]
+    artifact.metadata.update(
+        quarantined_organizations_count=len(quarantined_ids),
+        completed_organizations_count=len(completed),
+    )
+    if (
+        quarantined_ids
+        and not staged.filter(
+            disposition=ParserStagedRecord.Disposition.STAGED
+        ).exists()
+        and not artifact.metadata["confirmed_empty_lookup_200_count"]
+    ):
+        # Invalid rows plus an ambiguous 404 do not establish any valid coverage.
+        raise SnapshotValidationError("sro_incomplete_membership_scan")
+    return completed, list(quarantined_ids)
+
+
 def refresh_sro_membership(
     *,
     load_batch: int,
@@ -641,11 +672,6 @@ def refresh_sro_membership(
                     )
             finally:
                 save_artifact_file(artifact, handle)
-        if artifact.quarantined_count:
-            # A semantic failure is not evidence that old memberships disappeared.
-            # Keep raw/quarantine diagnostics, but never advance any checkpoint or
-            # replace the previous complete dataset with this incomplete scan.
-            raise SnapshotValidationError("sro_incomplete_membership_scan")
         if (
             candidates
             and artifact.metadata["recognized_lookup_404_count"] == len(candidates)
@@ -654,6 +680,7 @@ def refresh_sro_membership(
             # An all-404 scan cannot distinguish absence from a site-wide outage.
             # Raw responses stay available, but previous rows/checkpoints survive.
             raise SnapshotValidationError("sro_ambiguous_empty_scan")
+        completed, quarantined_ids = _sro_publication_scope(artifact, candidates)
         if mode == "incremental" and not candidates:
             previous = (
                 ParserSourceArtifact.objects.filter(
@@ -675,7 +702,12 @@ def refresh_sro_membership(
         artifact.save()
 
         def finalize(published_artifact, result):
-            _save_sro_checkpoints(candidates, published_artifact)
+            # An unchanged fingerprint must not suppress retry after a full sweep
+            # found invalid rows. Deletion shares the publication transaction.
+            SroOrganizationLookup.objects.filter(
+                organization_id__in=quarantined_ids
+            ).delete()
+            _save_sro_checkpoints(completed, published_artifact)
             if on_publish:
                 on_publish(published_artifact, result)
 
@@ -684,6 +716,7 @@ def refresh_sro_membership(
             scope_organization_ids=[org.uid for org in candidates]
             if mode == "incremental"
             else None,
+            preserve_organization_ids=quarantined_ids,
             on_publish=finalize,
         )
     except Exception as exc:

src/core/openapi.json

@@ -11787,6 +11787,21 @@
             "type": "integer",
             "readOnly": true
           },
+          "quarantined_organizations_count": {
+            "title": "Quarantined organizations count",
+            "type": "integer",
+            "readOnly": true
+          },
+          "completed_organizations_count": {
+            "title": "Completed organizations count",
+            "type": "integer",
+            "readOnly": true
+          },
+          "retained_records_count": {
+            "title": "Retained records count",
+            "type": "integer",
+            "readOnly": true
+          },
           "missing_admission_dates_count": {
             "title": "Missing admission dates count",
             "type": "integer",
@@ -11966,6 +11981,21 @@
             "type": "integer",
             "readOnly": true
           },
+          "quarantined_organizations_count": {
+            "title": "Quarantined organizations count",
+            "type": "integer",
+            "readOnly": true
+          },
+          "completed_organizations_count": {
+            "title": "Completed organizations count",
+            "type": "integer",
+            "readOnly": true
+          },
+          "retained_records_count": {
+            "title": "Retained records count",
+            "type": "integer",
+            "readOnly": true
+          },
           "missing_admission_dates_count": {
             "title": "Missing admission dates count",
             "type": "integer",

tests/apps/parsers/test_snapshot_public_contract.py

@@ -319,6 +319,9 @@ def test_sro_job_and_history_expose_typed_actual_counters_only(admin_client):
         "found_memberships_count": 2,
         "snapshot_records_count": 10,
         "snapshot_organizations_count": 8,
+        "quarantined_organizations_count": 1,
+        "completed_organizations_count": 2,
+        "retained_records_count": 3,
         "http_errors_count": 1,
         "parse_errors_count": 2,
         "upstream_registry_date": None,
@@ -334,6 +337,13 @@ def test_sro_job_and_history_expose_typed_actual_counters_only(admin_client):
     assert payload["meta"]["candidate_organizations_count"] == 7
     assert payload["meta"]["source"] == "sro_membership_check"
     assert payload["meta"]["upstream_registry_date"] is None
+    preservation_counters = {
+        "quarantined_organizations_count": 1,
+        "completed_organizations_count": 2,
+        "retained_records_count": 3,
+    }
+    for key, value in preservation_counters.items():
+        assert payload["meta"][key] == value
     assert "private_loader_setting" not in payload["meta"]
     log = ParserLoadLog.objects.create(
         source="sro_membership_check", batch_id=1, status="success", records_count=10
@@ -352,6 +362,8 @@ def test_sro_job_and_history_expose_typed_actual_counters_only(admin_client):
     assert response.data["meta"]["http_errors_count"] == 1
     assert response.data["meta"]["parse_errors_count"] == 2
     assert "private_loader_setting" not in response.data["meta"]
+    for key, value in preservation_counters.items():
+        assert response.data["meta"][key] == value
     schema = admin_client.get(reverse("schema-swagger-ui"), {"format": "openapi"}).data
     assert (
         schema["definitions"]["SnapshotRunMetadata"]["properties"][
@@ -363,3 +375,6 @@ def test_sro_job_and_history_expose_typed_actual_counters_only(admin_client):
         "organizations_count"
         in schema["definitions"]["SnapshotJobResult"]["properties"]
     )
+    for key in preservation_counters:
+        for name in ("SnapshotRunMetadata", "SnapshotJobResult"):
+            assert schema["definitions"][name]["properties"][key]["type"] == "integer"

tests/apps/parsers/test_sro_quarantine_publication.py

@@ -0,0 +1,314 @@
+"""Complete collection can publish valid memberships without erasing uncertain ones."""
+
+from types import SimpleNamespace
+from uuid import UUID, uuid4
+
+import pytest
+from apps.core.models import BackgroundJob, JobStatus
+from apps.parsers.models import (
+    ParserLoadLog,
+    ParserSourceArtifact,
+    ParserStagedRecord,
+    SroOrganizationLookup,
+)
+from apps.parsers.registry_snapshots import SnapshotValidationError
+from apps.parsers.source_cards import SourceCardService
+from apps.parsers.sro_membership import _sro_candidates, refresh_sro_membership
+from apps.parsers.tasks_registry_snapshots import _run_snapshot
+from django.utils import timezone
+from organizations.models import Organization, OrganizationSourceRecord
+
+from tests.apps.parsers.test_sro_membership import FixtureSite, lookup_html
+from tests.apps.parsers.test_sro_upstream_empty import MixedFixtureSite
+
+
+@pytest.fixture
+def organizations(db, settings, tmp_path):
+    settings.SRO_DEV_COLLECTION_ENABLED = True
+    settings.MEDIA_ROOT = str(tmp_path / "media")
+    settings.PARSER_PRIVATE_ARTIFACT_ROOT = str(tmp_path / "private")
+    return [
+        Organization.objects.create(
+            uid=UUID(int=index),
+            name=f"АО Смешанная фикстура {index}",
+            inn=f"{index}234567890",
+            ogrn=f"{index}027700132195",
+            okpo=f"0012345{index}",
+            directory_imported_at=timezone.now(),
+            opk_registry_membership=True,
+        )
+        for index in (1, 2)
+    ]
+
+
+def unresolved_html(organization, *, mixed=True):
+    html = lookup_html(
+        organization, ("001", "002") if mixed else ("002",), status="Исключен"
+    )
+    return html.replace(
+        b' href="https://moskva.reestr-sro.ru/sro-v-proektirovanii/sro-id-002/"',
+        b"",
+    )
+
+
+def seed(organizations):
+    first, _ = organizations
+    return refresh_sro_membership(
+        load_batch=1,
+        mode="full",
+        client=FixtureSite(
+            organizations, {first.ogrn: lookup_html(first, ("001", "002", "003"))}
+        ),
+    )
+
+
+def published_rows():
+    return list(OrganizationSourceRecord.objects.order_by("uid").values())
+
+
+def checkpoints():
+    return list(SroOrganizationLookup.objects.order_by("organization_id").values())
+
+
+@pytest.mark.parametrize("mode", ("full", "incremental"))
+def test_valid_sibling_upsert_retains_uncertain_old_rows_and_retries_org(
+    organizations, mode
+):
+    uncertain, empty = organizations
+    seed(organizations)
+    old = {row["external_id"]: row for row in published_rows()}
+    if mode == "incremental":
+        for organization in organizations:
+            organization.name += " обновлённая"
+            organization.save(update_fields=["name"])
+    artifact, result = refresh_sro_membership(
+        load_batch=2,
+        mode=mode,
+        client=FixtureSite(
+            organizations,
+            {
+                uncertain.ogrn: unresolved_html(uncertain),
+                empty.ogrn: lookup_html(empty, ()),
+            },
+        ),
+    )
+    current = {row["external_id"]: row for row in published_rows()}
+    assert len(current) == result.published == artifact.published_count == 3
+    changed = current[f"{uncertain.uid}:001"]
+    assert changed["uid"] == old[f"{uncertain.uid}:001"]["uid"]
+    assert changed["status"] == "inactive"
+    assert changed["load_batch"] == 2
+    for identifier in ("002", "003"):
+        key = f"{uncertain.uid}:{identifier}"
+        assert current[key] == old[key]
+    assert not OrganizationSourceRecord.objects.filter(
+        extension__organization=empty
+    ).exists()
+    assert artifact.metadata["retained_records_count"] == 2
+    assert artifact.metadata["batch_published_records_count"] == 1
+    assert artifact.metadata["snapshot_records_count"] == 3
+    assert artifact.metadata["snapshot_organizations_count"] == 1
+    assert artifact.metadata["quarantined_organizations_count"] == 1
+    assert artifact.metadata["completed_organizations_count"] == 1
+    assert artifact.quarantined_count == result.quarantined == 1
+    assert (
+        artifact.staged_records.get(disposition="quarantined").reason
+        == "sro_page_unresolved"
+    )
+    assert not SroOrganizationLookup.objects.filter(organization=uncertain).exists()
+    assert (
+        SroOrganizationLookup.objects.get(organization=empty).artifact_id
+        == artifact.uid
+    )
+    assert [org.uid for org in _sro_candidates("incremental")] == [uncertain.uid]
+    assert SourceCardService.get_card("sro-membership-check")["records_count"] == 3
+
+
+def test_first_load_publishes_valid_memberships_of_mixed_organization(organizations):
+    uncertain, unused = organizations
+    unused.delete()
+    artifact, result = refresh_sro_membership(
+        load_batch=1,
+        mode="full",
+        client=FixtureSite([uncertain], {uncertain.ogrn: unresolved_html(uncertain)}),
+    )
+    assert result.published == result.quarantined == 1
+    assert artifact.metadata["retained_records_count"] == 0
+    assert artifact.metadata["completed_organizations_count"] == 0
+    assert artifact.metadata["quarantined_organizations_count"] == 1
+    assert OrganizationSourceRecord.objects.get().payload["sro_id"] == "001"
+    assert not SroOrganizationLookup.objects.exists()
+    assert [org.uid for org in _sro_candidates("incremental")] == [uncertain.uid]
+
+
+def test_all_invalid_plus_ambiguous_404_is_not_success(organizations):
+    uncertain, negative = organizations
+    seed(organizations)
+    before, before_checkpoints = published_rows(), checkpoints()
+    site = MixedFixtureSite(organizations, {negative.ogrn})
+    site.overrides[uncertain.ogrn] = unresolved_html(uncertain, mixed=False)
+    with pytest.raises(
+        SnapshotValidationError, match="^sro_incomplete_membership_scan$"
+    ):
+        refresh_sro_membership(load_batch=2, mode="full", client=site)
+    assert published_rows() == before
+    assert checkpoints() == before_checkpoints
+
+
+def test_confirmed_zero_with_quarantine_can_publish_preserving_old_memberships(
+    organizations,
+):
+    uncertain, empty = organizations
+    seed(organizations)
+    artifact, result = refresh_sro_membership(
+        load_batch=2,
+        mode="full",
+        client=FixtureSite(
+            organizations,
+            {
+                uncertain.ogrn: unresolved_html(uncertain, mixed=False),
+                empty.ogrn: lookup_html(empty, ()),
+            },
+        ),
+    )
+    assert result.published == 3
+    assert artifact.metadata["batch_published_records_count"] == 0
+    assert artifact.metadata["retained_records_count"] == 3
+    assert not SroOrganizationLookup.objects.filter(organization=uncertain).exists()
+    assert (
+        SroOrganizationLookup.objects.get(organization=empty).artifact_id
+        == artifact.uid
+    )
+
+
+@pytest.mark.parametrize(
+    "error",
+    (
+        SnapshotValidationError("sro_upstream_transport_error"),
+        b"<html>upstream schema changed</html>",
+    ),
+    ids=("transport", "schema"),
+)
+def test_incomplete_collection_does_not_publish_valid_rows_or_clear_checkpoints(
+    organizations, error
+):
+    uncertain, failed = organizations
+    seed(organizations)
+    before, before_checkpoints = published_rows(), checkpoints()
+    with pytest.raises(SnapshotValidationError):
+        refresh_sro_membership(
+            load_batch=2,
+            mode="full",
+            client=FixtureSite(
+                organizations,
+                {uncertain.ogrn: unresolved_html(uncertain), failed.ogrn: error},
+            ),
+        )
+    assert published_rows() == before
+    assert checkpoints() == before_checkpoints
+    artifact = ParserSourceArtifact.objects.get(load_batch=2)
+    assert artifact.status == ParserSourceArtifact.Status.REJECTED
+    assert not artifact.staged_records.filter(disposition="published").exists()
+
+
+def test_cancelled_finalization_rolls_back_rows_and_quarantined_checkpoint_deletion(
+    organizations,
+):
+    uncertain, complete = organizations
+    seed(organizations)
+    before, before_checkpoints = published_rows(), checkpoints()
+
+    def revoked(artifact, result):
+        assert not SroOrganizationLookup.objects.filter(organization=uncertain).exists()
+        assert (
+            SroOrganizationLookup.objects.get(organization=complete).artifact_id
+            == artifact.uid
+        )
+        raise SnapshotValidationError("snapshot_job_no_longer_active")
+
+    with pytest.raises(
+        SnapshotValidationError, match="^snapshot_job_no_longer_active$"
+    ):
+        refresh_sro_membership(
+            load_batch=2,
+            mode="full",
+            client=FixtureSite(
+                organizations, {uncertain.ogrn: unresolved_html(uncertain)}
+            ),
+            on_publish=revoked,
+        )
+    assert published_rows() == before
+    assert checkpoints() == before_checkpoints
+
+
+def test_task_counts_include_retained_records_but_batch_counts_only_upserts(
+    organizations,
+):
+    uncertain, _ = organizations
+    seed(organizations)
+    task = SimpleNamespace(
+        request=SimpleNamespace(id=str(uuid4())),
+        name="parsers.sro_membership_check.refresh",
+    )
+
+    def refresh(**kwargs):
+        return refresh_sro_membership(
+            mode="full",
+            client=FixtureSite(
+                organizations, {uncertain.ogrn: unresolved_html(uncertain)}
+            ),
+            **kwargs,
+        )
+
+    result = _run_snapshot(
+        task, source="sro_membership_check", refresh=refresh, requested_by_id=None
+    )
+    assert result["records_count"] == result["published_records_count"] == 4
+    assert result["batch_published_records_count"] == 2
+    assert result["retained_records_count"] == 2
+    assert result["quarantined_records_count"] == 1
+    job = BackgroundJob.objects.get(task_id=task.request.id)
+    assert job.status == JobStatus.SUCCESS
+    assert job.meta["published_records_count"] == 4
+    log = ParserLoadLog.objects.get(pk=result["load_id"])
+    assert log.status == ParserLoadLog.Status.SUCCESS
+    assert log.records_count == 4
+    assert (
+        ParserStagedRecord.objects.filter(
+            artifact_id=result["artifact_id"], disposition="published"
+        ).count()
+        == 2
+    )
+
+
+def test_successful_incremental_retry_removes_retained_obsolete_memberships(
+    organizations,
+):
+    uncertain, unchanged = organizations
+    seed(organizations)
+    refresh_sro_membership(
+        load_batch=2,
+        mode="full",
+        client=FixtureSite(organizations, {uncertain.ogrn: unresolved_html(uncertain)}),
+    )
+    before = {row["external_id"]: row for row in published_rows()}
+    assert len(before) == 4
+    assert not SroOrganizationLookup.objects.filter(organization=uncertain).exists()
+    assert [org.uid for org in _sro_candidates("incremental")] == [uncertain.uid]
+
+    artifact, result = refresh_sro_membership(
+        load_batch=3, mode="incremental", client=FixtureSite([uncertain])
+    )
+    after = {row["external_id"]: row for row in published_rows()}
+    assert result.published == len(after) == 2
+    assert set(after) == {f"{uncertain.uid}:001", f"{unchanged.uid}:001"}
+    assert after[f"{uncertain.uid}:001"]["uid"] == before[f"{uncertain.uid}:001"]["uid"]
+    assert after[f"{uncertain.uid}:001"]["load_batch"] == 3
+    assert after[f"{unchanged.uid}:001"] == before[f"{unchanged.uid}:001"]
+    assert artifact.metadata["retained_records_count"] == 0
+    assert artifact.metadata["batch_published_records_count"] == 1
+    assert (
+        SroOrganizationLookup.objects.get(organization=uncertain).artifact_id
+        == artifact.uid
+    )
+    assert _sro_candidates("incremental") == []