Reset MARC files on settings update (PP-4637) (#3554)

## Description

- Adds a `marc_export_reset` Celery task that sets up the next MARC
export as a first run by removing existing files (database records and
S3 objects). The records are deleted in one transaction, and S3 deletion
is best-effort, so a failure won't leave the reset half-applied.
- Queues the reset when a library's MARC exporter settings change in the
admin interface and when a library registry pushes a new web client URL
(only if the change would result in a new web client URL).
- First-run full exports now also record a full-content delta pointing
at the same S3 object, so (1) consumers that only apply delta files
receive the complete refreshed dataset without a second upload and (2)
we avoid an extra upload and the doubled storage utilization.
- Export finalization now detects a configuration change that raced an
in-flight export (records reset, settings drift, or the library no
longer MARC-enabled), discards the stale upload, and immediately
re-queues a fresh full export.
- Cleanup and reset only delete a shared S3 object when the last record
referencing it is removed, and upload aborts are best-effort so cleanup
failures cannot cascade.
- The MARC download page labels the full-content delta as "Full content
as of <date>" instead of a date range starting at the full delta
sentinel epoch.

## Motivation and Context

MARC record exports run periodically, but a configuration change would
not trigger an update. Record updates based on those changes might not
become available until long after the change was made. Further, some
MARC consumers apply only delta files rather than reprocessing full
exports. When a library's MARC export configuration changed
(organization code, summary/genre inclusion, web client URLs), records
exported before the change were never corrected for those consumers,
leaving stale data in their catalogs indefinitely. Resetting the
library's export state on configuration changes forces a full refresh
that reaches both full-file and delta-only consumers.

[Jira PP-4637]

## How Has This Been Tested?

- New and updated tests for the new functionality.
- All tests/checks pass locally.
- [CI
tests](https://github.com/ThePalaceProject/circulation/actions/runs/29181152088)/checks
pass.

## Checklist

- [x] I have updated the documentation accordingly.
- [x] All new and existing tests passed.

Author: tdilauro

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,19 +92,50 @@ 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)
 
         except ProblemDetailException as e:
             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:

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>"