diff --git a/src/palace/manager/api/admin/controller/catalog_services.py b/src/palace/manager/api/admin/controller/catalog_services.py
index 3f6abfff8d..d00ea77b4a 100644
--- a/src/palace/manager/api/admin/controller/catalog_services.py
+++ b/src/palace/manager/api/admin/controller/catalog_services.py
@@ -8,7 +8,12 @@
 )
 from palace.manager.api.admin.form_data import ProcessFormData
 from palace.manager.api.admin.problem_details import MULTIPLE_SERVICES_FOR_LIBRARY
+from palace.manager.celery.tasks.marc import (
+    MARC_EXPORT_RESET_COOLDOWN_SECONDS,
+    marc_export_reset,
+)
 from palace.manager.integration.catalog.marc.exporter import MarcExporter
+from palace.manager.integration.catalog.marc.settings import MarcExporterLibrarySettings
 from palace.manager.integration.goals import Goals
 from palace.manager.integration.settings import BaseSettings
 from palace.manager.sqlalchemy.listeners import site_configuration_has_changed
@@ -87,12 +92,32 @@ def process_post(self) -> Response | ProblemDetail:
             validated_settings = ProcessFormData.get_settings(settings_class, form_data)
             catalog_service.settings_dict = validated_settings.model_dump()
 
+            # Capture current library settings so we can detect changes after the
+            # update.
+            old_library_settings: dict[int, MarcExporterLibrarySettings] = {
+                lc.library_id: impl_cls.library_settings_load(lc)
+                for lc in catalog_service.library_configurations
+                if lc.library_id is not None
+            }
+
             # Update library settings
             if libraries_data:
                 self.process_libraries(
                     catalog_service, libraries_data, impl_cls.library_settings_class()
                 )
 
+            # Find libraries whose settings actually changed. A MARC export reset
+            # is queued for each so that the next run produces a fresh full
+            # export along with a full-content delta.
+            reset_library_ids = [
+                lc.library_id
+                for lc in catalog_service.library_configurations
+                if lc.library_id is not None
+                and lc.library_id in old_library_settings
+                and impl_cls.library_settings_load(lc)
+                != old_library_settings[lc.library_id]
+            ]
+
             # Trigger a site configuration change
             site_configuration_has_changed(self._db)
 
@@ -100,6 +125,17 @@ def process_post(self) -> Response | ProblemDetail:
             self._db.rollback()
             return e.problem_detail
 
+        # Dispatch the resets before the request transaction commits: a failed
+        # enqueue then fails the request and rolls the settings change back, so
+        # a retried save re-queues the reset. The reverse order could commit the
+        # settings and then lose a reset, leaving delta-only consumers with
+        # stale records indefinitely. The task cooldown keeps a reset from
+        # running before the transaction commits.
+        for library_id in reset_library_ids:
+            marc_export_reset.apply_async(
+                (library_id,), countdown=MARC_EXPORT_RESET_COOLDOWN_SECONDS
+            )
+
         return Response(str(catalog_service.id), response_code)
 
     def process_delete(self, service_id: int) -> Response:
diff --git a/src/palace/manager/api/controller/marc.py b/src/palace/manager/api/controller/marc.py
index 5b6450df7c..0a7d1eb11e 100644
--- a/src/palace/manager/api/controller/marc.py
+++ b/src/palace/manager/api/controller/marc.py
@@ -9,7 +9,7 @@
 from sqlalchemy.orm import Session
 
 from palace.manager.api.util.flask import get_request_library
-from palace.manager.integration.catalog.marc.exporter import MarcExporter
+from palace.manager.integration.catalog.marc.exporter import MARC_EPOCH, MarcExporter
 from palace.manager.service.integration_registry.catalog_services import (
     CatalogServicesRegistry,
 )
@@ -171,7 +171,12 @@ def download_page_body(self, session: Session, library: Library) -> str:
                     body += "<ul>"
                     for update in files.deltas:
                         update_url = self.storage_service.generate_url(update.key)
-                        update_label = f"Updates from {update.since.strftime(time_format)} to {update.created.strftime(time_format)}"
+                        # Avoid misleading label for full-content delta.
+                        update_label = (
+                            f"Full content as of {update.created.strftime(time_format)}"
+                            if update.since == MARC_EPOCH
+                            else f"Updates from {update.since.strftime(time_format)} to {update.created.strftime(time_format)}"
+                        )
                         body += f'<li><a href="{update_url}">{update_label}</a></li>'
                     body += "</ul>"
 
diff --git a/src/palace/manager/celery/tasks/marc.py b/src/palace/manager/celery/tasks/marc.py
index 158e7d96b4..a3273b9a7d 100644
--- a/src/palace/manager/celery/tasks/marc.py
+++ b/src/palace/manager/celery/tasks/marc.py
@@ -1,14 +1,22 @@
 import datetime
+import uuid
 from contextlib import ExitStack
 from tempfile import TemporaryFile
 
+from botocore.exceptions import BotoCoreError, ClientError
 from celery import shared_task
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
 
 from palace.util.datetime_helpers import utc_now
 
 from palace.manager.celery.task import Task
 from palace.manager.celery.utils import signature_with
-from palace.manager.integration.catalog.marc.exporter import LibraryInfo, MarcExporter
+from palace.manager.integration.catalog.marc.exporter import (
+    MARC_EPOCH,
+    LibraryInfo,
+    MarcExporter,
+)
 from palace.manager.integration.catalog.marc.uploader import (
     MarcUploadManager,
     UploadContext,
@@ -21,8 +29,17 @@
     Identifier,
     RecursiveEquivalencyCache,
 )
+from palace.manager.sqlalchemy.model.library import Library
 from palace.manager.sqlalchemy.model.marcfile import MarcFile
-from palace.manager.sqlalchemy.util import create
+from palace.manager.sqlalchemy.util import create, get_one
+from palace.manager.util.json import json_canonical
+
+# LibraryInfo fields that change between runs without a configuration change.
+_VOLATILE_LIBRARY_INFO_FIELDS = frozenset({"last_updated", "needs_update"})
+
+# Cooldown before a queued marc_export_reset runs, so that it cannot run before
+# the transaction that dispatched it commits.
+MARC_EXPORT_RESET_COOLDOWN_SECONDS = 30
 
 
 @shared_task(queue=QueueNames.default, bind=True)
@@ -104,6 +121,166 @@ def marc_export_collection_lock(
     )
 
 
+# TODO: The `filtered_audiences` and `filtered_genres` library settings are
+#  compared below, but changing them does not yet queue a MARC export reset.
+def _export_settings_changed(snapshot: LibraryInfo, current: LibraryInfo) -> bool:
+    """Whether a library's export-affecting settings differ between two LibraryInfo.
+
+    Ignores the fields that may change between runs without affecting the export
+    and compares the remaining fields order-independently.
+    """
+
+    def comparable(info: LibraryInfo) -> str:
+        return json_canonical(
+            info.model_dump(exclude=set(_VOLATILE_LIBRARY_INFO_FIELDS))
+        )
+
+    return comparable(snapshot) != comparable(current)
+
+
+def _export_config_changed(
+    session: Session,
+    collection_id: int,
+    snapshot: LibraryInfo,
+    current: LibraryInfo | None,
+) -> bool:
+    """Whether a library's MARC configuration changed while an export was in flight.
+
+    True when the library is no longer MARC-enabled (current is None), when its
+    settings no longer match the ones from export generation time, or when a concurrent
+    `marc_export_reset` deleted the library's MarcFile records. The settings check also
+    covers a reset that found no records to delete during a first run.
+    """
+    if current is None:
+        return True
+    if _export_settings_changed(snapshot, current):
+        return True
+    records_gone: bool = (
+        snapshot.last_updated is not None
+        and session.execute(
+            select(func.count(MarcFile.id)).where(
+                MarcFile.library_id == snapshot.library_id,
+                MarcFile.collection_id == collection_id,
+            )
+        ).scalar_one()
+        == 0
+    )
+    return records_gone
+
+
+def _finalize_uploads(
+    task: Task,
+    session: Session,
+    uploads: dict[LibraryInfo, MarcUploadManager],
+    collection_id: int,
+    start_time: datetime.datetime,
+    delta: bool,
+) -> tuple[list[LibraryInfo], list[LibraryInfo]]:
+    """Complete the uploads and create their MarcFile records.
+
+    Uploads for libraries whose configuration changed while the export was in
+    flight are discarded instead. Returns two lists: the current LibraryInfo
+    for the discarded libraries, so the caller can re-queue a fresh full export
+    for them, and the snapshot LibraryInfo for the libraries whose records were
+    created, so the caller can re-verify them after the commit.
+    """
+    registry = task.services.integration_registry.catalog_services()
+    current_infos = {
+        info.library_id: info
+        for info in MarcExporter.enabled_libraries(session, registry, collection_id)
+    }
+    drifted_libraries: list[LibraryInfo] = []
+    recorded_libraries: list[LibraryInfo] = []
+    for library, upload in uploads.items():
+        # Recording an upload after the library's configuration changed would
+        # leave stale content behind, so discard it and let the caller re-queue
+        # a fresh full export. A change landing between this check and the
+        # commit is caught by _verify_uploads_after_commit.
+        current = current_infos.get(library.library_id)
+        if _export_config_changed(session, collection_id, library, current):
+            task.log.warning(
+                f"MARC export configuration for library "
+                f"{library.library_short_name} ({library.library_id}) "
+                f"changed while this export was in flight. "
+                f"Discarding '{upload.context.s3_key}'."
+            )
+            upload.abort()
+            # A library that is no longer MARC-enabled is discarded without a re-run.
+            if current is not None:
+                drifted_libraries.append(current)
+            continue
+
+        if upload.complete():
+            recorded_libraries.append(library)
+            create(
+                session,
+                MarcFile,
+                id=upload.context.upload_uuid,
+                library_id=library.library_id,
+                collection_id=collection_id,
+                created=start_time,
+                key=upload.context.s3_key,
+                since=library.last_updated if delta else None,
+            )
+            # For first-run (or reset) libraries in a full export, create a
+            # full-content delta record pointing at the same S3 object.
+            # Delta-only consumers will receive the complete dataset without
+            # a second upload.
+            if not delta and library.last_updated is None:
+                create(
+                    session,
+                    MarcFile,
+                    library_id=library.library_id,
+                    collection_id=collection_id,
+                    created=start_time,
+                    key=upload.context.s3_key,
+                    since=MARC_EPOCH,
+                )
+            task.log.info(f"Completed upload for '{upload.context.s3_key}'")
+        else:
+            task.log.warning(
+                f"No upload for '{upload.context.s3_key}', "
+                f"because there were no records."
+            )
+    return drifted_libraries, recorded_libraries
+
+
+def _verify_uploads_after_commit(
+    task: Task,
+    collection_id: int,
+    recorded_libraries: list[LibraryInfo],
+) -> None:
+    """Detect a configuration change that landed between the finalize check and
+    the export's commit, and then queue a reset for any library affected.
+
+    Every reset trigger also changes the library's LibraryInfo, so comparing
+    settings after the commit closes the remaining race window. That is, a change
+    committed before this check is detected here, and a change committed after
+    it queues a reset task that will see (and delete) the records this export
+    just committed.
+    """
+    if not recorded_libraries:
+        return
+
+    with task.session() as session:
+        registry = task.services.integration_registry.catalog_services()
+        current_infos = {
+            info.library_id: info
+            for info in MarcExporter.enabled_libraries(session, registry, collection_id)
+        }
+
+    for library in recorded_libraries:
+        current = current_infos.get(library.library_id)
+        if current is None or _export_settings_changed(library, current):
+            task.log.warning(
+                f"MARC export configuration for library "
+                f"{library.library_short_name} ({library.library_id}) changed "
+                f"while the export was being recorded. Queueing a reset to "
+                f"discard the stale export."
+            )
+            marc_export_reset.delay(library.library_id)
+
+
 @shared_task(queue=QueueNames.default, bind=True)
 def marc_export_collection(
     task: Task,
@@ -209,30 +386,35 @@ def marc_export_collection(
 
             if no_more_works:
                 # Task is complete. Finalize the s3 uploads and create MarcFile records in DB.
-                for library, upload in uploads.items():
-                    if upload.complete():
-                        create(
-                            session,
-                            MarcFile,
-                            id=upload.context.upload_uuid,
-                            library_id=library.library_id,
-                            collection_id=collection_id,
-                            created=start_time,
-                            key=upload.context.s3_key,
-                            since=library.last_updated if delta else None,
-                        )
-                        task.log.info(f"Completed upload for '{upload.context.s3_key}'")
-                    else:
-                        task.log.warning(
-                            f"No upload for '{upload.context.s3_key}', "
-                            f"because there were no records."
-                        )
-
+                drifted_libraries, recorded_libraries = _finalize_uploads(
+                    task, session, uploads, collection_id, start_time, delta
+                )
                 task.log.info(
                     f"Finished generating MARC records for collection '{collection_name}' ({collection_id}) "
                     f"in {(utc_now() - start_time).seconds} seconds."
                 )
-                return
+
+        if no_more_works:
+            # The transaction has committed. Re-check the recorded libraries so
+            # that a configuration change landing between the finalize check and
+            # the commit cannot leave a stale export standing.
+            _verify_uploads_after_commit(task, collection_id, recorded_libraries)
+
+    if no_more_works:
+        if drifted_libraries:
+            # Re-run a full export for the libraries whose configuration changed
+            # while this export was in flight, so they get fresh content with
+            # their new settings right away rather than at the next scheduled run.
+            raise task.replace(
+                marc_export_collection.s(
+                    collection_id=collection_id,
+                    collection_name=collection_name,
+                    start_time=utc_now(),
+                    libraries=drifted_libraries,
+                    batch_size=batch_size,
+                )
+            )
+        return
 
     # This task is complete, but there are more works waiting to be exported. So we requeue ourselves
     # to process the next batch.
@@ -253,6 +435,21 @@ def marc_export_collection(
     )
 
 
+def _marc_file_key_ref_count(
+    session: Session, key: str, exclude_id: uuid.UUID | None = None
+) -> int:
+    """Count the MarcFile records referencing the given S3 key.
+
+    A first-run full export and its paired full-content delta share an S3 key,
+    so the object must only be deleted when the last record referencing it goes.
+    """
+    stmt = select(func.count(MarcFile.id)).where(MarcFile.key == key)
+    if exclude_id is not None:
+        stmt = stmt.where(MarcFile.id != exclude_id)
+    count: int = session.execute(stmt).scalar_one()
+    return count
+
+
 @shared_task(queue=QueueNames.default, bind=True)
 def marc_export_cleanup(
     task: Task,
@@ -260,6 +457,9 @@ def marc_export_cleanup(
 ) -> None:
     """
     Cleanup old MARC exports that are outdated or no longer needed.
+
+    The S3 object is deleted before the MarcFile record, so a failure leaves the
+    record in place to be retried by the next scheduled cleanup run.
     """
     storage_service = task.services.storage.public()
     registry = task.services.integration_registry.catalog_services()
@@ -272,6 +472,58 @@ def marc_export_cleanup(
                 raise task.replace(marc_export_cleanup.s())
 
             task.log.info(f"Deleting MARC export {file_record.key} ({file_record.id}).")
-            storage_service.delete(file_record.key)
+            if (
+                _marc_file_key_ref_count(
+                    session, file_record.key, exclude_id=file_record.id
+                )
+                == 0
+            ):
+                storage_service.delete(file_record.key)
             session.delete(file_record)
             session.commit()
+
+
+@shared_task(queue=QueueNames.default, bind=True)
+def marc_export_reset(task: Task, library_id: int) -> None:
+    """
+    Reset MARC export for a library so that the next export run behaves like a
+    first run, producing both a new full export and a full-content delta.
+    Deletes the MarcFile records and associated S3 objects for the library's
+    MARC-enabled collections.
+
+    The records are deleted in a single transaction, so a failure cannot leave the
+    reset half-applied. The S3 objects are then deleted best-effort: a failed
+    deletion orphans that object and logs an error, but allows the reset to continue.
+
+    Repeated configuration changes can queue duplicate tasks. We don't guard
+    against that since the task is idempotent, even under concurrent duplicates:
+    the bulk delete matches no rows the other task already removed, and the S3
+    deletions tolerate missing objects.
+    """
+    storage_service = task.services.storage.public()
+    with task.session() as session:
+        library = get_one(session, Library, id=library_id)
+        if library is None:
+            task.log.warning(
+                f"Library {library_id} not found. Skipping MARC export reset."
+            )
+            return
+
+        keys = MarcExporter.delete_files_for_reset(session, library)
+        session.commit()
+        task.log.info(
+            f"Reset MARC export for library {library_id}. "
+            f"Deleted records for {len(keys)} file(s)."
+        )
+
+        # Skip keys that other MarcFile records still reference.
+        still_referenced = set(
+            session.scalars(
+                select(MarcFile.key).where(MarcFile.key.in_(keys)).distinct()
+            )
+        )
+        for key in keys - still_referenced:
+            try:
+                storage_service.delete(key)
+            except (BotoCoreError, ClientError):
+                task.log.exception(f"Failed to delete orphaned MARC export '{key}'.")
diff --git a/src/palace/manager/integration/catalog/marc/exporter.py b/src/palace/manager/integration/catalog/marc/exporter.py
index 78bed6a308..35b8d3bc1c 100644
--- a/src/palace/manager/integration/catalog/marc/exporter.py
+++ b/src/palace/manager/integration/catalog/marc/exporter.py
@@ -4,8 +4,8 @@
 from collections.abc import Generator, Iterable, Sequence
 
 from pydantic import BaseModel, ConfigDict
-from sqlalchemy import select
 from sqlalchemy.orm import Session, aliased, raiseload, selectinload
+from sqlalchemy.sql import Select, delete, select
 
 from palace.util.datetime_helpers import utc_now
 from palace.util.log import LoggerMixin
@@ -38,6 +38,9 @@
 from palace.manager.sqlalchemy.model.marcfile import MarcFile
 from palace.manager.sqlalchemy.model.work import Work, WorkGenre
 
+# Sentinel timestamp used as `since` on a full-content delta MarcFile record.
+MARC_EPOCH = datetime.datetime(1900, 1, 1, tzinfo=datetime.UTC)
+
 
 class LibraryInfo(BaseModel):
     library_id: int
@@ -95,7 +98,11 @@ def _needs_update(
     def _web_client_urls(
         session: Session, library: Library, url: str | None = None
     ) -> tuple[str, ...]:
-        """Find web client URLs configured by the registry for this library."""
+        """Find web client URLs configured by the registry for this library.
+
+        The result is de-duplicated: each distinct URL is exported once, even
+        when a registry-supplied URL matches the manually configured one.
+        """
         urls = [
             s.web_client
             for s in session.execute(
@@ -109,7 +116,7 @@ def _web_client_urls(
         if url:
             urls.append(url)
 
-        return tuple(urls)
+        return tuple(dict.fromkeys(urls))
 
     @classmethod
     def _enabled_collections_and_libraries(
@@ -168,6 +175,95 @@ def _last_updated(
 
         return last_updated_file.created if last_updated_file else None
 
+    @staticmethod
+    def _marc_enabled_collections_query(library: Library) -> Select:
+        """Query selecting the MARC-export-enabled collections associated with a library."""
+        return (
+            select(Collection)
+            .join(Collection.integration_configuration)
+            .join(IntegrationConfiguration.library_configurations)
+            .where(
+                IntegrationLibraryConfiguration.library == library,
+                Collection.export_marc_records == True,
+            )
+        )
+
+    @staticmethod
+    def delete_files_for_reset(session: Session, library: Library) -> set[str]:
+        """Bulk-delete the MarcFile records for a library's MARC-enabled collections
+        and return the S3 keys of the deleted records. Does not commit.
+
+        Deleting these records (and their S3 objects) causes the next export run
+        to behave like a first run, producing both a full export and a
+        full-content delta for delta-only consumers.
+
+        The bulk delete tolerates concurrent duplicate resets: rows another task
+        already deleted simply match nothing.
+        """
+        enabled_collection_ids = MarcExporter._marc_enabled_collections_query(
+            library
+        ).with_only_columns(Collection.id)
+        # A single DELETE ... RETURNING keeps the keys and the deleted rows in
+        # exact correspondence. A separate SELECT would leave a window where a
+        # concurrently committed record is deleted without its key being
+        # captured, orphaning its S3 object.
+        return set(
+            session.scalars(
+                delete(MarcFile)
+                .where(
+                    MarcFile.library == library,
+                    MarcFile.collection_id.in_(enabled_collection_ids),
+                )
+                .returning(MarcFile.key)
+                .execution_options(synchronize_session=False)
+            )
+        )
+
+    @staticmethod
+    def needs_reset_for_web_client_change(
+        session: Session,
+        registry: CatalogServicesRegistry,
+        library: Library,
+        old_url: str | None,
+        new_url: str | None,
+    ) -> bool:
+        """Whether a registry web-client URL change affects the library's MARC output.
+
+        The exported set of web client URLs is every registry-supplied URL plus
+        the manual web_client_url of each MARC library configuration (see
+        _web_client_urls). A change cannot affect that set when both the old and
+        new URLs are already covered by the manual URLs (or absent), so no reset
+        is needed in that case.
+        """
+        marc_integrations = registry.configurations_query(MarcExporter).subquery()
+        marc_library_configs = (
+            session.execute(
+                select(IntegrationLibraryConfiguration)
+                .join(
+                    marc_integrations,
+                    IntegrationLibraryConfiguration.parent_id == marc_integrations.c.id,
+                )
+                .where(IntegrationLibraryConfiguration.library == library)
+            )
+            .scalars()
+            .all()
+        )
+
+        # A library with no MARC exporter configuration exports nothing, so no
+        # URL change can affect its output.
+        if not marc_library_configs:
+            return False
+
+        marc_manual_urls = {
+            MarcExporter.library_settings_load(row).web_client_url
+            for row in marc_library_configs
+        }
+
+        def covered(url: str | None) -> bool:
+            return url is None or url in marc_manual_urls
+
+        return not (covered(old_url) and covered(new_url))
+
     @classmethod
     def enabled_collections(
         cls, session: Session, registry: CatalogServicesRegistry
diff --git a/src/palace/manager/integration/catalog/marc/uploader.py b/src/palace/manager/integration/catalog/marc/uploader.py
index fe8a55ccd5..0e347e7d85 100644
--- a/src/palace/manager/integration/catalog/marc/uploader.py
+++ b/src/palace/manager/integration/catalog/marc/uploader.py
@@ -3,6 +3,7 @@
 from types import TracebackType
 from typing import IO, Literal, Self
 
+from botocore.exceptions import BotoCoreError, ClientError
 from pydantic import BaseModel
 
 from palace.util.exceptions import BasePalaceException
@@ -159,6 +160,12 @@ def complete(self) -> bool:
         return True
 
     def abort(self) -> None:
+        """Abort the in-progress multipart upload, best-effort.
+
+        The upload is being discarded, so a failed abort is logged and swallowed
+        rather than raised. The upload is left unfinalized on failure, so a later
+        call can retry the abort.
+        """
         if self._finalized:
             return
 
@@ -166,9 +173,16 @@ def abort(self) -> None:
             self._finalized = True
             return
 
-        self.storage_service.multipart_abort(
-            self.context.s3_key, self.context.upload_id
-        )
+        try:
+            self.storage_service.multipart_abort(
+                self.context.s3_key, self.context.upload_id
+            )
+        except (BotoCoreError, ClientError):
+            self.log.exception(
+                f"Failed to abort upload '{self.context.s3_key}' "
+                f"(UploadID: {self.context.upload_id})."
+            )
+            return
         self._finalized = True
 
     @property
diff --git a/src/palace/manager/integration/discovery/opds_registration.py b/src/palace/manager/integration/discovery/opds_registration.py
index a5530f55ed..dd1fe7992a 100644
--- a/src/palace/manager/integration/discovery/opds_registration.py
+++ b/src/palace/manager/integration/discovery/opds_registration.py
@@ -19,6 +19,7 @@
 )
 from palace.manager.core.problem_details import INTEGRATION_ERROR
 from palace.manager.integration.base import HasIntegrationConfiguration
+from palace.manager.integration.catalog.marc.exporter import MarcExporter
 from palace.manager.integration.goals import Goals
 from palace.manager.integration.settings import (
     BaseSettings,
@@ -514,6 +515,44 @@ def _process_registration_result(
         # communicated to the registry.
         registration.stage = desired_stage
 
+        # Reset MARC export for the library if the web client URL change affects
+        # MARC output, so delta-only consumers receive a full refresh with the
+        # updated 856 fields.
+        if (
+            web_client_url != registration.web_client
+            and registration.library is not None
+        ):
+            session = Session.object_session(registration)
+            if session is not None:
+                # Imported locally because these modules (via the service
+                # container's discovery registry) transitively import this module.
+                from palace.manager.celery.tasks.marc import (
+                    MARC_EXPORT_RESET_COOLDOWN_SECONDS,
+                    marc_export_reset,
+                )
+                from palace.manager.service.container import container_instance
+
+                catalog_registry = (
+                    container_instance().integration_registry.catalog_services()
+                )
+                if MarcExporter.needs_reset_for_web_client_change(
+                    session,
+                    catalog_registry,
+                    registration.library,
+                    old_url=registration.web_client,
+                    new_url=web_client_url,
+                ):
+                    # The task cooldown keeps the reset from running before
+                    # this transaction commits. If the transaction rolls back
+                    # instead, the reset still runs, but the cost is the same
+                    # download gap as any legitimate reset (the files are
+                    # regenerated by the next scheduled export run) and the
+                    # settings are unchanged, so the exports are never stale.
+                    marc_export_reset.apply_async(
+                        (registration.library_id,),
+                        countdown=MARC_EXPORT_RESET_COOLDOWN_SECONDS,
+                    )
+
         # Store the web client URL
         registration.web_client = web_client_url
 
diff --git a/tests/fixtures/marc.py b/tests/fixtures/marc.py
index 65b34ca21e..1d54b850bd 100644
--- a/tests/fixtures/marc.py
+++ b/tests/fixtures/marc.py
@@ -43,6 +43,9 @@ def __init__(
         self.collection2.associated_libraries = [self.library1]
         self.collection3.associated_libraries = [self.library2]
 
+        # Set by configure_export().
+        self.marc_integration: IntegrationConfiguration | None = None
+
     def integration(self) -> IntegrationConfiguration:
         return self._db.integration_configuration(
             MarcExporter, Goals.CATALOG_GOAL, name="MARC Exporter"
@@ -65,6 +68,7 @@ def works(self, collection: Collection | None = None) -> list[Work]:
 
     def configure_export(self) -> None:
         marc_integration = self.integration()
+        self.marc_integration = marc_integration
         self._db.integration_library_configuration(
             marc_integration,
             self.library1,
diff --git a/tests/manager/api/admin/controller/test_catalog_services.py b/tests/manager/api/admin/controller/test_catalog_services.py
index d9ddc56e72..e6bfaa2c4c 100644
--- a/tests/manager/api/admin/controller/test_catalog_services.py
+++ b/tests/manager/api/admin/controller/test_catalog_services.py
@@ -1,9 +1,11 @@
 import json
 from contextlib import nullcontext
+from unittest.mock import MagicMock
 
 import flask
 import pytest
 from flask import Response
+from pytest import MonkeyPatch
 from werkzeug.datastructures import ImmutableMultiDict
 
 from palace.manager.api.admin.controller.catalog_services import (
@@ -19,6 +21,10 @@
     NO_PROTOCOL_FOR_NEW_SERVICE,
     UNKNOWN_PROTOCOL,
 )
+from palace.manager.celery.tasks.marc import (
+    MARC_EXPORT_RESET_COOLDOWN_SECONDS,
+    marc_export_reset,
+)
 from palace.manager.integration.catalog.marc.exporter import MarcExporter
 from palace.manager.integration.catalog.marc.settings import MarcExporterLibrarySettings
 from palace.manager.integration.goals import Goals
@@ -318,3 +324,121 @@ def test_catalog_services_delete(
 
         none_service = get_one(db.session, IntegrationConfiguration, id=service.id)
         assert none_service is None
+
+    @pytest.mark.parametrize(
+        "changed, legacy, expect_reset",
+        [
+            pytest.param(True, False, True, id="settings-changed"),
+            pytest.param(False, False, False, id="settings-unchanged"),
+            pytest.param(False, True, False, id="legacy-dict-normalized"),
+        ],
+    )
+    def test_catalog_services_post_resets_marc_export_on_settings_change(
+        self,
+        flask_app_fixture: FlaskAppFixture,
+        controller: CatalogServicesController,
+        db: DatabaseTransactionFixture,
+        monkeypatch: MonkeyPatch,
+        changed: bool,
+        legacy: bool,
+        expect_reset: bool,
+    ):
+        mock_apply_async = MagicMock()
+        monkeypatch.setattr(marc_export_reset, "apply_async", mock_apply_async)
+
+        library = db.default_library()
+        service = db.integration_configuration(
+            MarcExporter,
+            Goals.CATALOG_GOAL,
+            name="marc",
+            libraries=[library],
+        )
+        protocol = service.protocol
+
+        # Configure initial library settings with include_summary=False.
+        lc = service.for_library(library)
+        assert lc is not None
+        if legacy:
+            # Simulate a legacy stored dict with explicit default values, which a
+            # raw dict comparison would misread as a change after normalization.
+            lc.settings_dict = {"include_summary": False, "include_genres": False}
+        else:
+            MarcExporter.library_settings_update(
+                lc,
+                MarcExporterLibrarySettings(
+                    include_summary=False, include_genres=False
+                ),
+            )
+
+        # POST with either changed or unchanged settings.
+        new_include_summary = "true" if changed else "false"
+        with flask_app_fixture.test_request_context_system_admin("/", method="POST"):
+            flask.request.form = ImmutableMultiDict(
+                [
+                    ("name", "marc"),
+                    ("id", str(service.id)),
+                    ("protocol", str(protocol)),
+                    (
+                        "libraries",
+                        json.dumps(
+                            [
+                                {
+                                    "short_name": library.short_name,
+                                    "include_summary": new_include_summary,
+                                    "include_genres": "false",
+                                }
+                            ]
+                        ),
+                    ),
+                ]
+            )
+            response = controller.process_catalog_services()
+            assert isinstance(response, Response)
+
+        if expect_reset:
+            mock_apply_async.assert_called_once_with(
+                (library.id,), countdown=MARC_EXPORT_RESET_COOLDOWN_SECONDS
+            )
+        else:
+            mock_apply_async.assert_not_called()
+
+    def test_catalog_services_post_no_reset_for_new_library(
+        self,
+        flask_app_fixture: FlaskAppFixture,
+        controller: CatalogServicesController,
+        db: DatabaseTransactionFixture,
+        monkeypatch: MonkeyPatch,
+    ):
+        """Adding a new library to a service does not trigger a reset (no prior records)."""
+        mock_apply_async = MagicMock()
+        monkeypatch.setattr(marc_export_reset, "apply_async", mock_apply_async)
+
+        service = db.integration_configuration(
+            MarcExporter, Goals.CATALOG_GOAL, name="marc"
+        )
+        protocol = service.protocol
+
+        with flask_app_fixture.test_request_context_system_admin("/", method="POST"):
+            flask.request.form = ImmutableMultiDict(
+                [
+                    ("name", "marc"),
+                    ("id", str(service.id)),
+                    ("protocol", str(protocol)),
+                    (
+                        "libraries",
+                        json.dumps(
+                            [
+                                {
+                                    "short_name": db.default_library().short_name,
+                                    "include_summary": "false",
+                                    "include_genres": "false",
+                                }
+                            ]
+                        ),
+                    ),
+                ]
+            )
+            response = controller.process_catalog_services()
+            assert isinstance(response, Response)
+
+        mock_apply_async.assert_not_called()
diff --git a/tests/manager/api/controller/test_marc.py b/tests/manager/api/controller/test_marc.py
index 175628ac36..d3d11dea59 100644
--- a/tests/manager/api/controller/test_marc.py
+++ b/tests/manager/api/controller/test_marc.py
@@ -9,7 +9,7 @@
 from palace.util.datetime_helpers import utc_now
 
 from palace.manager.api.controller.marc import MARCRecordController
-from palace.manager.integration.catalog.marc.exporter import MarcExporter
+from palace.manager.integration.catalog.marc.exporter import MARC_EPOCH, MarcExporter
 from palace.manager.integration.goals import Goals
 from palace.manager.service.integration_registry.catalog_services import (
     CatalogServicesRegistry,
@@ -244,6 +244,8 @@ def test_download_page_with_full_and_delta(
         marc_record_controller_fixture.file(
             key="delta_2", created=yesterday, since=last_week
         )
+        # A first-run full-content delta shares the full export's S3 object.
+        marc_record_controller_fixture.file(key="full", created=now, since=MARC_EPOCH)
 
         response = marc_record_controller_fixture.controller.download_page()
         html = marc_record_controller_fixture.get_response_html(response)
@@ -265,6 +267,14 @@ def test_download_page_with_full_and_delta(
             % (last_week.strftime("%B %-d, %Y"), yesterday.strftime("%B %-d, %Y"))
             in html
         )
+        # The full-content delta is labeled distinctly rather than as a
+        # date range starting at the MARC_EPOCH sentinel.
+        assert (
+            '<a href="http://s3.url/full">Full content as of %s</a>'
+            % now.strftime("%B %-d, %Y")
+            in html
+        )
+        assert "1900" not in html
 
     def test_download_page_with_exporter_but_no_collection(
         self, marc_record_controller_fixture: MARCRecordControllerFixture
diff --git a/tests/manager/celery/tasks/test_marc.py b/tests/manager/celery/tasks/test_marc.py
index 5fa37e4825..60980b3c1e 100644
--- a/tests/manager/celery/tasks/test_marc.py
+++ b/tests/manager/celery/tasks/test_marc.py
@@ -1,15 +1,17 @@
 import datetime
-from unittest.mock import ANY, call, patch
+from unittest.mock import ANY, MagicMock, call, patch
 
 import pytest
+from botocore.exceptions import ClientError
 from pymarc import MARCReader
+from pytest import MonkeyPatch
 from sqlalchemy import select
 
 from palace.util.datetime_helpers import utc_now
 
 from palace.manager.celery.tasks import marc
 from palace.manager.celery.tasks.marc import marc_export_collection_lock
-from palace.manager.integration.catalog.marc.exporter import MarcExporter
+from palace.manager.integration.catalog.marc.exporter import MARC_EPOCH, MarcExporter
 from palace.manager.integration.catalog.marc.uploader import MarcUploadManager
 from palace.manager.service.redis.models.lock import LockNotAcquired, RedisLock
 from palace.manager.sqlalchemy.model.collection import Collection
@@ -238,8 +240,15 @@ def test_normal_run(
         assert len(uploaded_files) == 2
 
         # Verify that the expected number of marc files were created in the database.
+        # A first-run full export creates both a full record and a full-content delta
+        # (MARC_EPOCH) pointing to the same S3 object; 2 libraries → 4 records, 2 S3 objects.
         marc_files = marc_export_collection_fixture.marc_files()
-        assert len(marc_files) == 2
+        assert len(marc_files) == 4
+        full_files = [mf for mf in marc_files if mf.since is None]
+        epoch_delta_files = [mf for mf in marc_files if mf.since == MARC_EPOCH]
+        assert len(full_files) == 2
+        assert len(epoch_delta_files) == 2
+        assert {mf.key for mf in full_files} == {mf.key for mf in epoch_delta_files}
         filenames = [marc_file.key for marc_file in marc_files]
 
         # Verify that the uploaded files are the expected ones.
@@ -265,7 +274,7 @@ def test_normal_run(
 
         # Because no works have been updated since the last run, no delta exports are generated
         marc_files = marc_export_collection_fixture.marc_files()
-        assert len(marc_files) == 2
+        assert len(marc_files) == 4
 
         # Update a couple works last_updated_time
         updated_works = [works[0], works[1]]
@@ -274,9 +283,9 @@ def test_normal_run(
 
         marc_export_collection_fixture.export_collection(collection, delta=True)
 
-        # Now we generate marc files
+        # Now we generate marc files (4 from the first full run + 2 new delta records).
         marc_files = marc_export_collection_fixture.marc_files()
-        assert len(marc_files) == 4
+        assert len(marc_files) == 6
         delta_marc_files = [
             marc_file
             for marc_file in marc_files
@@ -304,6 +313,184 @@ def test_normal_run(
             # Make sure records have the correct status
             assert all(record.leader.record_status == "c" for record in records)
 
+        # A subsequent full export creates only full records. The epoch delta
+        # pair is a first-run behavior and must not repeat.
+        marc_export_collection_fixture.export_collection(collection)
+        marc_files = marc_export_collection_fixture.marc_files()
+        assert len(marc_files) == 8
+        assert len([mf for mf in marc_files if mf.since == MARC_EPOCH]) == 2
+
+    @pytest.mark.parametrize(
+        "delta", [pytest.param(False, id="full"), pytest.param(True, id="delta")]
+    )
+    def test_reset_race(
+        self,
+        db: DatabaseTransactionFixture,
+        marc_exporter_fixture: MarcExporterFixture,
+        s3_service_integration_fixture: S3ServiceIntegrationFixture,
+        marc_export_collection_fixture: MarcExportCollectionFixture,
+        delta: bool,
+    ):
+        """An export whose records were reset mid-flight is discarded and re-run."""
+        marc_export_collection_fixture.setup_minio_storage()
+        collection = marc_exporter_fixture.collection1
+        assert collection.id is not None
+        marc_export_collection_fixture.works(collection)
+
+        # Craft the raced state: the scheduling-time snapshot says the libraries
+        # have prior exports, but no MarcFile records exist by completion time
+        # (as if marc_export_reset ran while the export was in flight).
+        libraries = MarcExporter.enabled_libraries(
+            db.session, marc_exporter_fixture.registry, collection.id
+        )
+        stale_libraries = [
+            info.model_copy(
+                update={"last_updated": utc_now() - datetime.timedelta(days=2)}
+            )
+            for info in libraries
+        ]
+        marc.marc_export_collection.delay(
+            collection.id,
+            collection_name=collection.name,
+            batch_size=5,
+            start_time=marc_export_collection_fixture.start_time,
+            libraries=stale_libraries,
+            delta=delta,
+        ).wait()
+
+        # The stale uploads were discarded and a fresh full export ran, creating
+        # the first-run full + epoch-delta pair for each library.
+        marc_files = marc_export_collection_fixture.marc_files()
+        assert len(marc_files) == 4
+        assert len([mf for mf in marc_files if mf.since is None]) == 2
+        assert len([mf for mf in marc_files if mf.since == MARC_EPOCH]) == 2
+
+        # Only the fresh exports' objects exist in storage.
+        uploaded = s3_service_integration_fixture.list_objects("public")
+        assert set(uploaded) == {mf.key for mf in marc_files}
+
+    def test_reset_race_first_run_settings_drift(
+        self,
+        db: DatabaseTransactionFixture,
+        marc_exporter_fixture: MarcExporterFixture,
+        s3_service_integration_fixture: S3ServiceIntegrationFixture,
+        marc_export_collection_fixture: MarcExportCollectionFixture,
+    ):
+        """A settings change during a first-run export (which the record-count
+        check can't see) is caught by the settings-drift check and re-run."""
+        marc_export_collection_fixture.setup_minio_storage()
+        collection = marc_exporter_fixture.collection1
+        assert collection.id is not None
+        marc_export_collection_fixture.works(collection)
+
+        # The export runs with a settings snapshot that no longer matches the
+        # current configuration (as if the settings changed mid-flight).
+        libraries = MarcExporter.enabled_libraries(
+            db.session, marc_exporter_fixture.registry, collection.id
+        )
+        stale_libraries = [
+            info.model_copy(update={"organization_code": "stale-org"})
+            for info in libraries
+        ]
+        marc.marc_export_collection.delay(
+            collection.id,
+            collection_name=collection.name,
+            batch_size=5,
+            start_time=marc_export_collection_fixture.start_time,
+            libraries=stale_libraries,
+        ).wait()
+
+        # The stale uploads were discarded and the re-run used current settings.
+        marc_files = marc_export_collection_fixture.marc_files()
+        assert len(marc_files) == 4
+        uploaded = s3_service_integration_fixture.list_objects("public")
+        assert set(uploaded) == {mf.key for mf in marc_files}
+        for file in uploaded:
+            data = s3_service_integration_fixture.get_object("public", file)
+            for record in MARCReader(data):
+                assert record["003"].data in {"library1-org", "library2-org"}
+
+    def test_reset_race_after_finalize_check(
+        self,
+        db: DatabaseTransactionFixture,
+        marc_exporter_fixture: MarcExporterFixture,
+        marc_export_collection_fixture: MarcExportCollectionFixture,
+        monkeypatch: MonkeyPatch,
+    ):
+        """A configuration change landing between the finalize check and the
+        export's commit is detected after the commit and queues a reset."""
+        marc_export_collection_fixture.setup_mock_storage()
+        collection = marc_exporter_fixture.collection1
+        marc_export_collection_fixture.works(collection)
+
+        mock_delay = MagicMock()
+        monkeypatch.setattr(marc.marc_export_reset, "delay", mock_delay)
+
+        # The finalize check (one call per library) sees unchanged settings; the
+        # post-commit verification sees changed ones.
+        monkeypatch.setattr(
+            marc,
+            "_export_settings_changed",
+            MagicMock(side_effect=[False, False, True, True]),
+        )
+
+        marc_export_collection_fixture.export_collection(collection)
+
+        # A reset was queued for both libraries whose stale exports were recorded.
+        assert mock_delay.call_count == 2
+        assert {call_args.args[0] for call_args in mock_delay.call_args_list} == {
+            marc_exporter_fixture.library1.id,
+            marc_exporter_fixture.library2.id,
+        }
+
+    def test_reset_race_library_disabled(
+        self,
+        db: DatabaseTransactionFixture,
+        marc_exporter_fixture: MarcExporterFixture,
+        s3_service_integration_fixture: S3ServiceIntegrationFixture,
+        marc_export_collection_fixture: MarcExportCollectionFixture,
+    ):
+        """A raced library that is no longer MARC-enabled is discarded without a re-run."""
+        marc_export_collection_fixture.setup_minio_storage()
+        collection = marc_exporter_fixture.collection1
+        assert collection.id is not None
+        marc_export_collection_fixture.works(collection)
+
+        libraries = MarcExporter.enabled_libraries(
+            db.session, marc_exporter_fixture.registry, collection.id
+        )
+        stale_libraries = [
+            info.model_copy(
+                update={"last_updated": utc_now() - datetime.timedelta(days=2)}
+            )
+            for info in libraries
+        ]
+
+        # Remove library2's MARC exporter configuration mid-flight.
+        assert marc_exporter_fixture.marc_integration is not None
+        lc = marc_exporter_fixture.marc_integration.for_library(
+            marc_exporter_fixture.library2
+        )
+        assert lc is not None
+        db.session.delete(lc)
+
+        marc.marc_export_collection.delay(
+            collection.id,
+            collection_name=collection.name,
+            batch_size=5,
+            start_time=marc_export_collection_fixture.start_time,
+            libraries=stale_libraries,
+        ).wait()
+
+        # Library1 was re-run (first-run pair); library2 was discarded silently.
+        marc_files = marc_export_collection_fixture.marc_files()
+        assert {mf.library_id for mf in marc_files} == {
+            marc_exporter_fixture.library1.id
+        }
+        assert len(marc_files) == 2
+        uploaded = s3_service_integration_fixture.list_objects("public")
+        assert set(uploaded) == {mf.key for mf in marc_files}
+
     def test_collection_no_works(
         self,
         marc_exporter_fixture: MarcExporterFixture,
@@ -374,3 +561,111 @@ def test_marc_export_cleanup(
     [not_deleted] = db.session.execute(select(MarcFile)).scalars().all()
     assert not_deleted.id == not_deleted_id
     assert mock_s3.deleted == deleted_keys
+
+
+def test_marc_export_cleanup_shared_key(
+    db: DatabaseTransactionFixture,
+    celery_fixture: CeleryFixture,
+    s3_service_fixture: S3ServiceFixture,
+    marc_exporter_fixture: MarcExporterFixture,
+    services_fixture: ServicesFixture,
+):
+    """S3 object shared by two MarcFile records is deleted only once, not twice."""
+    marc_exporter_fixture.configure_export()
+    mock_s3 = s3_service_fixture.mock_service()
+    services_fixture.services.storage.public.override(mock_s3)
+
+    shared_key = db.fresh_str()
+
+    # Simulate a first-run pair: full record + MARC_EPOCH delta sharing the same S3 key.
+    # Both belong to a collection not in the MARC-enabled set so they are cleaned up via
+    # the "disabled pair" branch of files_for_cleanup.
+    other_collection = db.collection()
+    marc_exporter_fixture.marc_file(
+        key=shared_key,
+        collection=other_collection,
+        library=marc_exporter_fixture.library1,
+    )
+    marc_exporter_fixture.marc_file(
+        key=shared_key,
+        collection=other_collection,
+        library=marc_exporter_fixture.library1,
+        since=MARC_EPOCH,
+    )
+
+    marc.marc_export_cleanup.delay(batch_size=100).wait()
+
+    # Both MarcFile records removed from the database.
+    remaining = db.session.execute(select(MarcFile)).scalars().all()
+    assert all(mf.key != shared_key for mf in remaining)
+
+    # S3 object deleted exactly once despite two records sharing the key.
+    assert mock_s3.deleted.count(shared_key) == 1
+
+
+def test_marc_export_reset(
+    db: DatabaseTransactionFixture,
+    celery_fixture: CeleryFixture,
+    s3_service_fixture: S3ServiceFixture,
+    marc_exporter_fixture: MarcExporterFixture,
+    services_fixture: ServicesFixture,
+):
+    marc_exporter_fixture.configure_export()
+    mock_s3 = s3_service_fixture.mock_service()
+    services_fixture.services.storage.public.override(mock_s3)
+
+    library1 = marc_exporter_fixture.library1
+    library2 = marc_exporter_fixture.library2
+    assert library1.id is not None
+
+    # A nonexistent library is a no-op.
+    marc.marc_export_reset.delay(library_id=-1).wait()
+
+    # First-run pair for library1 / collection1 sharing one S3 object, plus a
+    # record in another collection. Library2's record must not be affected.
+    shared_key = db.fresh_str()
+    marc_exporter_fixture.marc_file(key=shared_key)
+    marc_exporter_fixture.marc_file(key=shared_key, since=MARC_EPOCH)
+    collection2_key = marc_exporter_fixture.marc_file(
+        collection=marc_exporter_fixture.collection2
+    ).key
+    library2_file = marc_exporter_fixture.marc_file(library=library2)
+
+    marc.marc_export_reset.delay(library_id=library1.id).wait()
+
+    # Only library2's record remains.
+    remaining = db.session.execute(select(MarcFile)).scalars().all()
+    assert remaining == [library2_file]
+
+    # The shared S3 object was deleted exactly once; the other object once.
+    assert mock_s3.deleted.count(shared_key) == 1
+    assert mock_s3.deleted.count(collection2_key) == 1
+
+
+def test_marc_export_reset_s3_failure(
+    db: DatabaseTransactionFixture,
+    celery_fixture: CeleryFixture,
+    s3_service_fixture: S3ServiceFixture,
+    marc_exporter_fixture: MarcExporterFixture,
+    services_fixture: ServicesFixture,
+):
+    """A failed S3 deletion orphans the object but never leaves the reset half-applied."""
+    marc_exporter_fixture.configure_export()
+    mock_s3 = s3_service_fixture.mock_service()
+    services_fixture.services.storage.public.override(mock_s3)
+
+    library1 = marc_exporter_fixture.library1
+    assert library1.id is not None
+
+    marc_exporter_fixture.marc_file()
+    marc_exporter_fixture.marc_file(collection=marc_exporter_fixture.collection2)
+
+    with patch.object(
+        mock_s3,
+        "delete",
+        side_effect=ClientError({"Error": {}}, "DeleteObject"),
+    ):
+        marc.marc_export_reset.delay(library_id=library1.id).wait()
+
+    # All MarcFile records were deleted despite the S3 failures.
+    assert db.session.execute(select(MarcFile)).scalars().all() == []
diff --git a/tests/manager/integration/catalog/marc/test_exporter.py b/tests/manager/integration/catalog/marc/test_exporter.py
index 5f25ab7dd1..fd02d3ecb3 100644
--- a/tests/manager/integration/catalog/marc/test_exporter.py
+++ b/tests/manager/integration/catalog/marc/test_exporter.py
@@ -3,6 +3,7 @@
 
 import pytest
 from freezegun import freeze_time
+from sqlalchemy import select
 from sqlalchemy.exc import InvalidRequestError
 
 from palace.util.datetime_helpers import datetime_utc, utc_now
@@ -10,10 +11,14 @@
 from palace.manager.core.classifier import Classifier
 from palace.manager.integration.catalog.marc.exporter import LibraryInfo, MarcExporter
 from palace.manager.integration.catalog.marc.settings import MarcExporterLibrarySettings
+from palace.manager.integration.goals import Goals
 from palace.manager.sqlalchemy.model.classification import Genre
 from palace.manager.sqlalchemy.model.discovery_service_registration import (
     DiscoveryServiceRegistration,
 )
+from palace.manager.sqlalchemy.model.integration import (
+    IntegrationLibraryConfiguration,
+)
 from palace.manager.sqlalchemy.model.marcfile import MarcFile
 from palace.manager.sqlalchemy.model.work import WorkGenre
 from palace.manager.sqlalchemy.util import create, get_one_or_create
@@ -76,6 +81,11 @@ def test__web_client_urls(
             "http://web-client",
         )
 
+        # Duplicate URLs are collapsed; each distinct URL is exported once.
+        assert web_client_urls(url="http://web-client/registry") == (
+            "http://web-client/registry",
+        )
+
     def test__enabled_collections_and_libraries(
         self,
         db: DatabaseTransactionFixture,
@@ -630,3 +640,150 @@ def test_enabled_libraries_includes_filtering_settings(
         # Library2 should have no filtering
         assert library2_info.filtered_audiences == ()
         assert library2_info.filtered_genres == ()
+
+    def test_delete_files_for_reset(
+        self,
+        db: DatabaseTransactionFixture,
+        marc_exporter_fixture: MarcExporterFixture,
+    ) -> None:
+        library1 = marc_exporter_fixture.library1
+        library2 = marc_exporter_fixture.library2
+        collection1 = marc_exporter_fixture.collection1
+        collection2 = marc_exporter_fixture.collection2
+        collection1.export_marc_records = True
+        collection2.export_marc_records = True
+
+        # No records, nothing to delete.
+        assert MarcExporter.delete_files_for_reset(db.session, library1) == set()
+
+        # Create MarcFile records for library1 across two collections, with a
+        # second record sharing the first record's key (a first-run pair).
+        key1 = marc_exporter_fixture.marc_file(
+            library=library1, collection=collection1
+        ).key
+        marc_exporter_fixture.marc_file(
+            library=library1, collection=collection1, key=key1
+        )
+        key2 = marc_exporter_fixture.marc_file(
+            library=library1, collection=collection2
+        ).key
+        # Record for library2 on collection1 must not be affected.
+        library2_file = marc_exporter_fixture.marc_file(
+            library=library2, collection=collection1
+        )
+
+        # The shared key is returned once, and only library2's record survives.
+        assert MarcExporter.delete_files_for_reset(db.session, library1) == {
+            key1,
+            key2,
+        }
+        remaining = db.session.execute(select(MarcFile)).scalars().all()
+        assert remaining == [library2_file]
+
+    @pytest.mark.parametrize(
+        "old_url, new_url, marc_configs, expected",
+        [
+            pytest.param(
+                "http://old", "http://new", (None,), True, id="url-change-no-manual"
+            ),
+            pytest.param(None, "http://new", (None,), True, id="url-added"),
+            pytest.param("http://old", None, (None,), True, id="url-removed"),
+            pytest.param(
+                None,
+                "http://manual",
+                ("http://manual",),
+                False,
+                id="added-url-covered-by-manual",
+            ),
+            pytest.param(
+                "http://old",
+                "http://manual",
+                ("http://manual",),
+                True,
+                id="old-url-not-covered",
+            ),
+            pytest.param(
+                "http://manual",
+                None,
+                ("http://manual",),
+                False,
+                id="removed-url-covered-by-manual",
+            ),
+            pytest.param(
+                "http://manual1",
+                "http://manual2",
+                ("http://manual1", "http://manual2"),
+                False,
+                id="both-urls-covered-by-manual",
+            ),
+            pytest.param(
+                "http://old", "http://new", (), False, id="no-marc-configuration"
+            ),
+        ],
+    )
+    def test_needs_reset_for_web_client_change(
+        self,
+        db: DatabaseTransactionFixture,
+        marc_exporter_fixture: MarcExporterFixture,
+        old_url: str | None,
+        new_url: str | None,
+        marc_configs: tuple[str | None, ...],
+        expected: bool,
+    ) -> None:
+        """Each marc_configs entry is a MARC library configuration's manual
+        web_client_url (None for a configuration without one); an empty tuple
+        means the library has no MARC exporter configuration at all."""
+        library1 = marc_exporter_fixture.library1
+
+        for idx, manual_url in enumerate(marc_configs):
+            marc_integration = db.integration_configuration(
+                MarcExporter, Goals.CATALOG_GOAL, name=f"MARC Exporter {idx}"
+            )
+            db.integration_library_configuration(
+                marc_integration,
+                library1,
+                MarcExporterLibrarySettings(web_client_url=manual_url),
+            )
+
+        assert (
+            MarcExporter.needs_reset_for_web_client_change(
+                db.session,
+                marc_exporter_fixture.registry,
+                library1,
+                old_url=old_url,
+                new_url=new_url,
+            )
+            == expected
+        )
+
+    def test_needs_reset_for_web_client_change_other_protocol(
+        self,
+        db: DatabaseTransactionFixture,
+        marc_exporter_fixture: MarcExporterFixture,
+    ) -> None:
+        """A matching URL on a non-MARC catalog integration doesn't count as covered."""
+        library1 = marc_exporter_fixture.library1
+        marc_integration = db.integration_configuration(
+            MarcExporter, Goals.CATALOG_GOAL, name="MARC Exporter"
+        )
+        db.integration_library_configuration(marc_integration, library1)
+        other = db.integration_configuration(
+            "fake protocol", Goals.CATALOG_GOAL, name="other"
+        )
+        # Created directly: the db fixture helper resolves the protocol through
+        # the registry, which doesn't know "fake protocol".
+        other_lc, _ = create(
+            db.session,
+            IntegrationLibraryConfiguration,
+            parent=other,
+            library=library1,
+        )
+        other_lc.settings_dict = {"web_client_url": "http://manual"}
+
+        assert MarcExporter.needs_reset_for_web_client_change(
+            db.session,
+            marc_exporter_fixture.registry,
+            library1,
+            old_url=None,
+            new_url="http://manual",
+        )
diff --git a/tests/manager/integration/catalog/marc/test_uploader.py b/tests/manager/integration/catalog/marc/test_uploader.py
index 0ae3cadb50..70287cdbef 100644
--- a/tests/manager/integration/catalog/marc/test_uploader.py
+++ b/tests/manager/integration/catalog/marc/test_uploader.py
@@ -1,10 +1,11 @@
 from functools import partial
 from io import BytesIO
 from tempfile import TemporaryFile
-from unittest.mock import create_autospec
+from unittest.mock import create_autospec, patch
 from uuid import UUID, uuid4
 
 import pytest
+from botocore.exceptions import ClientError
 
 from palace.util.datetime_helpers import datetime_utc
 
@@ -147,6 +148,25 @@ def test_abort(self, marc_upload_manager_fixture: MarcUploadManagerFixture):
         # calling abort again is a no-op
         uploader.abort()
 
+        # A failed abort is logged and swallowed, and the upload stays unfinalized.
+        uploader = marc_upload_manager_fixture.create_uploader()
+        uploader.begin_upload()
+        with patch.object(
+            marc_upload_manager_fixture.mock_s3_service,
+            "multipart_abort",
+            side_effect=ClientError({"Error": {}}, "AbortMultipartUpload"),
+        ):
+            uploader.abort()
+        assert not uploader.finalized
+
+        # A later call can retry the abort and succeed.
+        uploader.abort()
+        assert uploader.finalized
+        assert (
+            uploader.context.s3_key
+            in marc_upload_manager_fixture.mock_s3_service.aborted
+        )
+
     def test_complete(self, marc_upload_manager_fixture: MarcUploadManagerFixture):
         # If the upload hasn't started, the upload isn't aborted, but it is finalized
         uploader = marc_upload_manager_fixture.create_uploader()
diff --git a/tests/manager/integration/discovery/test_opds_registration.py b/tests/manager/integration/discovery/test_opds_registration.py
index a9a668d596..66960617c8 100644
--- a/tests/manager/integration/discovery/test_opds_registration.py
+++ b/tests/manager/integration/discovery/test_opds_registration.py
@@ -5,7 +5,7 @@
 from dataclasses import dataclass
 from functools import partial
 from typing import Any
-from unittest.mock import MagicMock
+from unittest.mock import ANY, MagicMock
 
 import pytest
 from Crypto.Cipher import PKCS1_OAEP
@@ -18,8 +18,13 @@
     REMOTE_INTEGRATION_FAILED,
     SHARED_SECRET_DECRYPTION_ERROR,
 )
+from palace.manager.celery.tasks.marc import (
+    MARC_EXPORT_RESET_COOLDOWN_SECONDS,
+    marc_export_reset,
+)
 from palace.manager.core.config import CannotLoadConfiguration
 from palace.manager.core.problem_details import INTEGRATION_ERROR, INVALID_INPUT
+from palace.manager.integration.catalog.marc.exporter import MarcExporter
 from palace.manager.integration.discovery.opds_registration import (
     OpdsRegistrationService,
 )
@@ -664,6 +669,11 @@ def test__process_registration_result(
         encryptor = MagicMock()
 
         reg = MagicMock(spec=DiscoveryServiceRegistration)
+        # A mock isn't a mapped ORM object, so keep the MARC-export reset path
+        # (which needs the registration's session) from being taken. That path
+        # is covered separately with real registrations in
+        # test__process_registration_result_resets_marc_on_web_client_change.
+        reg.library = None
 
         # Result must be a dictionary.
         with pytest.raises(ProblemDetailException) as excinfo:
@@ -717,6 +727,74 @@ def test__process_registration_result(
 
         assert SHARED_SECRET_DECRYPTION_ERROR == excinfo.value.problem_detail
 
+    @pytest.mark.parametrize(
+        "existing_web_client, catalog_web_client, needs_reset, expect_delay",
+        [
+            pytest.param(None, "http://new-url", True, True, id="none-to-url"),
+            pytest.param(
+                "http://old-url", "http://new-url", True, True, id="url-changed"
+            ),
+            pytest.param(
+                "http://old-url",
+                "http://new-url",
+                False,
+                False,
+                id="url-changed-no-reset-needed",
+            ),
+            pytest.param(
+                "http://same-url", "http://same-url", True, False, id="url-unchanged"
+            ),
+            pytest.param(None, None, True, False, id="both-none"),
+        ],
+    )
+    def test__process_registration_result_resets_marc_on_web_client_change(
+        self,
+        remote_registry_fixture: RemoteRegistryFixture,
+        monkeypatch: MonkeyPatch,
+        existing_web_client: str | None,
+        catalog_web_client: str | None,
+        needs_reset: bool,
+        expect_delay: bool,
+    ):
+        mock_needs_reset = MagicMock(return_value=needs_reset)
+        monkeypatch.setattr(
+            MarcExporter, "needs_reset_for_web_client_change", mock_needs_reset
+        )
+        mock_apply_async = MagicMock()
+        monkeypatch.setattr(marc_export_reset, "apply_async", mock_apply_async)
+
+        registration = remote_registry_fixture.create_registration()
+        registration.web_client = existing_web_client
+
+        catalog: dict[str, Any] = {}
+        if catalog_web_client is not None:
+            catalog["links"] = [
+                {"href": catalog_web_client, "rel": "self", "type": "text/html"}
+            ]
+
+        m = remote_registry_fixture.registry._process_registration_result
+        m(registration, catalog, MagicMock(), RegistrationStage.TESTING)
+
+        # The reset check is only consulted when the web client URL changed.
+        if existing_web_client != catalog_web_client:
+            mock_needs_reset.assert_called_once_with(
+                remote_registry_fixture.db.session,
+                ANY,
+                registration.library,
+                old_url=existing_web_client,
+                new_url=catalog_web_client,
+            )
+        else:
+            mock_needs_reset.assert_not_called()
+
+        if expect_delay:
+            mock_apply_async.assert_called_once_with(
+                (registration.library_id,),
+                countdown=MARC_EXPORT_RESET_COOLDOWN_SECONDS,
+            )
+        else:
+            mock_apply_async.assert_not_called()
+
 
 class TestLibraryRegistrationScript:
     def test_constructor(