diff --git a/api/crossref/views.py b/api/crossref/views.py index 9baeb37d562..8d4f1ebf422 100644 --- a/api/crossref/views.py +++ b/api/crossref/views.py @@ -7,6 +7,7 @@ from api.crossref.permissions import RequestComesFromMailgun from osf.models import Preprint, NotificationTypeEnum +from osf.models.base import Guid from website import settings from website.preprints.tasks import mint_doi_on_crossref_fail @@ -48,6 +49,24 @@ def post(self, request): if record.get('status').lower() == 'success' and doi: msg = record.find('msg').text created = bool(msg == 'Successfully added') + # Unversioned DOIs (no _vN suffix, e.g. 10.31233/osf.io/tnaqp) are routing + # aliases that always resolve to the latest version via OSF's GUID routing. + # Store them as 'doi_unversioned' on the v1 preprint so we can track which + # preprint series have had their unversioned DOI registered. + _, version = Guid.split_guid(guid) if guid else (None, None) + if not version: + logger.info(f'Unversioned DOI confirmed by CrossRef: {doi}') + if guid: + v1_preprint = Preprint.objects.filter( + versioned_guids__guid___id=guid, + versioned_guids__version=1, + ).first() + if v1_preprint: + v1_preprint.set_identifier_value(category='doi_unversioned', value=doi) + Preprint.objects.filter(id=v1_preprint.id).update(doi_resync_queued_at=None) + dois_processed += 1 + continue + legacy_doi = preprint.get_identifier(category='legacy_doi') if created or legacy_doi: # Sets preprint_doi_created and saves the preprint @@ -56,6 +75,7 @@ def post(self, request): elif 'possible preprint/vor pair' not in msg.lower(): # Directly updates the identifier preprint.set_identifier_value(category='doi', value=doi) + Preprint.objects.filter(id=preprint.id).update(doi_resync_queued_at=None) dois_processed += 1 @@ -67,9 +87,14 @@ def post(self, request): if 'Relation target DOI does not exist' in record.find('msg').text: logger.warning('Related publication DOI does not exist, sending metadata again without it...') mint_doi_on_crossref_fail.apply_async(kwargs={'preprint_id': preprint._id}) - # This error occurs when a single preprint is being updated several times in a row with the same metadata [#PLAT-944] - elif 'less or equal to previously submitted version' in record.find('msg').text and record_count == 2: - break + # This error occurs when a single preprint is being updated several times in a row + # with the same metadata [#PLAT-944]. Previously this broke out of the loop when + # record_count == 2 (single DOI submitted twice). Now batches legitimately contain + # 2 records (versioned + unversioned DOI), so we continue instead of break to allow + # the remaining record to be processed. + elif 'less or equal to previously submitted version' in record.find('msg').text: + dois_processed += 1 + continue else: unexpected_errors = True logger.info(f'Creation success email received from CrossRef for preprints: {guids}') diff --git a/api_tests/crossref/views/test_crossref_email_response.py b/api_tests/crossref/views/test_crossref_email_response.py index 196c0debd2a..e39fb4b4176 100644 --- a/api_tests/crossref/views/test_crossref_email_response.py +++ b/api_tests/crossref/views/test_crossref_email_response.py @@ -220,3 +220,91 @@ def test_confirmation_marks_legacy_doi_as_deleted(self, app, url, preprint): app.post(url, context_data) assert preprint.identifiers.get(category='legacy_doi').deleted + + def test_unversioned_doi_confirmation_skips_identifier_update(self, app, url, preprint): + versioned_doi = settings.DOI_FORMAT.format( + prefix=preprint.provider.doi_prefix, guid=preprint._id + ) + preprint.set_identifier_value(category='doi', value=versioned_doi) + + base_guid = preprint.get_guid()._id # no _vN suffix + unversioned_doi = settings.DOI_FORMAT.format( + prefix=preprint.provider.doi_prefix, guid=base_guid + ) + dual_confirmation_xml = """ + + + 1390675999 + {batch_id} + + {versioned_doi} + Successfully updated + + + {unversioned_doi} + Successfully added + + + 2 + 2 + 0 + 0 + + + """.format( + batch_id=preprint._id, + versioned_doi=versioned_doi, + unversioned_doi=unversioned_doi, + ) + + context_data = self.make_mailgun_payload(crossref_response=dual_confirmation_xml) + with capture_notifications(expect_none=True): + app.post(url, context_data) + + preprint.reload() + assert preprint.get_identifier_value('doi') == versioned_doi + assert preprint.get_identifier_value('doi_unversioned') == unversioned_doi + + def test_unversioned_doi_confirmation_update_does_not_store_doi_unversioned(self, app, url, preprint): + versioned_doi = settings.DOI_FORMAT.format( + prefix=preprint.provider.doi_prefix, guid=preprint._id + ) + preprint.set_identifier_value(category='doi', value=versioned_doi) + + base_guid = preprint.get_guid()._id + unversioned_doi = settings.DOI_FORMAT.format( + prefix=preprint.provider.doi_prefix, guid=base_guid + ) + update_confirmation_xml = """ + + + 1390676000 + {batch_id} + + {versioned_doi} + Successfully updated + + + {unversioned_doi} + Successfully updated + + + 2 + 2 + 0 + 0 + + + """.format( + batch_id=preprint._id, + versioned_doi=versioned_doi, + unversioned_doi=unversioned_doi, + ) + + context_data = self.make_mailgun_payload(crossref_response=update_confirmation_xml) + with capture_notifications(expect_none=True): + app.post(url, context_data) + + preprint.reload() + assert preprint.get_identifier_value('doi') == versioned_doi + assert preprint.get_identifier_value('doi_unversioned') == unversioned_doi diff --git a/osf/management/commands/resync_preprint_dois_v1.py b/osf/management/commands/resync_preprint_dois_v1.py new file mode 100644 index 00000000000..0be7d967c29 --- /dev/null +++ b/osf/management/commands/resync_preprint_dois_v1.py @@ -0,0 +1,317 @@ +import logging +from dataclasses import dataclass + +from django.contrib.contenttypes.models import ContentType +from django.core.management.base import BaseCommand, CommandError +from django.db.models import Q +from django.utils import timezone + +from framework.celery_tasks import app +from osf.models import Preprint, PreprintProvider, Identifier +from osf.models.base import VersionedGuidMixin +from osf.management.commands.sync_doi_metadata import async_request_identifier_update +from website import settings + +logger = logging.getLogger(__name__) + + +@dataclass +class ResyncPassResult: + total: int + queued: int + errored: int + + @property + def remaining(self): + # Preprints that still need this pass after accounting for what was just queued. + return max(0, self.total - self.queued) + + +def get_in_flight_doi_resync_count(): + # Preprints with a DOI resync request dispatched to Crossref that hasn't been confirmed or gone stale. + cutoff = timezone.now() - settings.PREPRINT_DOI_RESYNC_INFLIGHT_TIMEOUT + return Preprint.objects.filter(doi_resync_queued_at__gte=cutoff).count() + + +def get_available_doi_resync_capacity(): + return max(0, settings.PREPRINT_DOI_RESYNC_MAX_IN_FLIGHT - get_in_flight_doi_resync_count()) + + +def _not_in_flight_query(): + cutoff = timezone.now() - settings.PREPRINT_DOI_RESYNC_INFLIGHT_TIMEOUT + return Q(doi_resync_queued_at__isnull=True) | Q(doi_resync_queued_at__lt=cutoff) + + +def _batch_for_pass(preprints_to_update, dry_run, batch_size, capacity): + """Slice the queryset for one pass. + + A dry run just previews the backlog and ignores in-flight capacity. A live run is + capped by both `batch_size` and the capacity which is left in the shared in-flight for this tick. + """ + if dry_run: + return preprints_to_update[:batch_size] if batch_size else preprints_to_update.iterator() + + if capacity is None: + capacity = get_available_doi_resync_capacity() + effective_batch_size = min(batch_size, capacity) if batch_size else capacity + return preprints_to_update[:effective_batch_size] if effective_batch_size else [] + + +def get_preprints_needing_v1_doi(provider_id=None): + content_type = ContentType.objects.get_for_model(Preprint) + + already_versioned_ids = Identifier.objects.filter( + content_type=content_type, + category='doi', + deleted__isnull=True, + value__contains=VersionedGuidMixin.GUID_VERSION_DELIMITER, + ).values_list('object_id', flat=True) + + public_query = Q(is_published=True, is_public=True, deleted__isnull=True) + withdrawn_query = Q(date_withdrawn__isnull=False, ever_public=True) + + qs = Preprint.objects.filter( + versioned_guids__version=1, + ).filter( + public_query | withdrawn_query + ).filter( + _not_in_flight_query() + ).exclude( + id__in=already_versioned_ids + ).exclude( + Q(provider__doi_prefix__isnull=True) | Q(provider__doi_prefix='') + ).exclude( + tags__name='qatest', + tags__system=True, + ).distinct().order_by('id') + + if provider_id: + qs = qs.filter(provider___id=provider_id) + + return qs + + +def resync_preprint_dois_v1(dry_run=True, batch_size=1000, provider_id=None, capacity=None): + preprints_to_update = get_preprints_needing_v1_doi(provider_id=provider_id) + + total = preprints_to_update.count() + logger.info( + f'{"[DRY RUN] " if dry_run else ""}' + f'{total} preprints need v1 DOI resync' + + (f' (provider={provider_id})' if provider_id else '') + ) + + preprints_iterable = _batch_for_pass(preprints_to_update, dry_run, batch_size, capacity) + + queued = 0 + errored = 0 + for preprint in preprints_iterable: + if dry_run: + logger.info(f'[DRY RUN] Would resync DOI for preprint {preprint._id}') + queued += 1 + continue + + try: + async_request_identifier_update.apply_async(kwargs={'preprint_id': preprint._id}) + Preprint.objects.filter(id=preprint.id).update(doi_resync_queued_at=timezone.now()) + logger.info(f'Queued DOI resync for preprint {preprint._id}') + queued += 1 + except Exception: + logger.exception(f'Failed to queue DOI resync for preprint {preprint._id}') + errored += 1 + + logger.info( + f'{"[DRY RUN] " if dry_run else ""}' + f'Done: {queued} preprints queued, {errored} errored' + ) + result = ResyncPassResult(total=total, queued=queued, errored=errored) + if not dry_run and batch_size: + logger.info( + f'Estimated remaining after this batch: ~{result.remaining}. ' + f'Re-run this command until 0 preprints remain.' + ) + return result + + +def get_preprints_needing_unversioned_doi(provider_id=None): + content_type = ContentType.objects.get_for_model(Preprint) + + already_has_unversioned = Identifier.objects.filter( + content_type=content_type, + category='doi_unversioned', + deleted__isnull=True, + ).values_list('object_id', flat=True) + + has_versioned_doi = Identifier.objects.filter( + content_type=content_type, + category='doi', + deleted__isnull=True, + value__contains=VersionedGuidMixin.GUID_VERSION_DELIMITER, + ).values_list('object_id', flat=True) + + public_query = Q(is_published=True, is_public=True, deleted__isnull=True) + withdrawn_query = Q(date_withdrawn__isnull=False, ever_public=True) + + qs = Preprint.objects.filter( + versioned_guids__version=1, + id__in=has_versioned_doi, + ).filter( + public_query | withdrawn_query + ).filter( + _not_in_flight_query() + ).exclude( + id__in=already_has_unversioned + ).exclude( + Q(provider__doi_prefix__isnull=True) | Q(provider__doi_prefix='') + ).exclude( + tags__name='qatest', + tags__system=True, + ).distinct().order_by('id') + + if provider_id: + qs = qs.filter(provider___id=provider_id) + + return qs + + +def register_missing_unversioned_dois(dry_run=True, batch_size=1000, provider_id=None, capacity=None): + preprints_to_update = get_preprints_needing_unversioned_doi(provider_id=provider_id) + + total = preprints_to_update.count() + logger.info( + f'{"[DRY RUN] " if dry_run else ""}' + f'{total} preprints need unversioned DOI registration' + + (f' (provider={provider_id})' if provider_id else '') + ) + + preprints_iterable = _batch_for_pass(preprints_to_update, dry_run, batch_size, capacity) + + queued = 0 + errored = 0 + for preprint in preprints_iterable: + if dry_run: + logger.info(f'[DRY RUN] Would register unversioned DOI for preprint {preprint._id}') + queued += 1 + continue + + try: + async_request_identifier_update.apply_async(kwargs={'preprint_id': preprint._id}) + Preprint.objects.filter(id=preprint.id).update(doi_resync_queued_at=timezone.now()) + logger.info(f'Queued unversioned DOI registration for preprint {preprint._id}') + queued += 1 + except Exception: + logger.exception(f'Failed to queue unversioned DOI registration for preprint {preprint._id}') + errored += 1 + + logger.info( + f'{"[DRY RUN] " if dry_run else ""}' + f'Unversioned DOI pass done: {queued} queued, {errored} errored' + ) + result = ResyncPassResult(total=total, queued=queued, errored=errored) + if not dry_run and batch_size: + logger.info( + f'Estimated unversioned remaining after this batch: ~{result.remaining}. ' + f'Re-run until 0 preprints remain.' + ) + return result + + +@app.task(bind=True, name='osf.management.commands.resync_preprint_dois_v1', max_retries=0) +def resync_preprint_dois_v1_task(self, batch_size=1000, dry_run=False, provider_id=None): + capacity = None if dry_run else get_available_doi_resync_capacity() + + v1_result = resync_preprint_dois_v1( + dry_run=dry_run, + batch_size=batch_size, + provider_id=provider_id, + capacity=capacity, + ) + if capacity is not None: + capacity = max(0, capacity - v1_result.queued) + + unversioned_result = register_missing_unversioned_dois( + dry_run=dry_run, + batch_size=batch_size, + provider_id=provider_id, + capacity=capacity, + ) + + if not dry_run and (v1_result.remaining > 0 or unversioned_result.remaining > 0): + logger.info( + f'Rescheduling DOI resync dispatcher in ' + f'{settings.PREPRINT_DOI_RESYNC_DISPATCH_INTERVAL} ' + f'({v1_result.remaining} v1 + {unversioned_result.remaining} unversioned remaining)' + ) + self.apply_async( + kwargs={ + 'batch_size': batch_size, + 'dry_run': dry_run, + 'provider_id': provider_id, + }, + countdown=settings.PREPRINT_DOI_RESYNC_DISPATCH_INTERVAL.total_seconds(), + ) + + +class Command(BaseCommand): + help = ( + 'Resync DOIs for version-1 preprints that are missing the versioned DOI suffix (_v1), ' + 'and register missing unversioned DOIs. ' + 'Use --provider (e.g. socarxiv) to restrict the run to a single preprint provider; ' + 'without it, all providers are processed. ' + 'With --dry_run, previews the backlog synchronously and exits. ' + 'Otherwise, starts the resync_preprint_dois_v1_task dispatcher once: it queues an initial ' + 'batch (bounded by --batch_size and the shared in-flight capacity, see ' + 'PREPRINT_DOI_RESYNC_MAX_IN_FLIGHT) and self-reschedules every ' + 'PREPRINT_DOI_RESYNC_DISPATCH_INTERVAL until the backlog is drained, then stops on its ' + 'own -- no need to re-run this command.' + ) + + def add_arguments(self, parser): + super().add_arguments(parser) + parser.add_argument( + '--dry_run', + action='store_true', + dest='dry_run', + help='Log what would be done without submitting to Crossref.', + ) + parser.add_argument( + '--batch_size', + '-b', + type=int, + default=1000, + help='Maximum number of preprints to process per dispatch tick (default: 1000).', + ) + parser.add_argument( + '--provider', + '-p', + type=str, + default=None, + dest='provider_id', + help='Restrict to a single provider _id (e.g. socarxiv).', + ) + + def handle(self, *args, **options): + dry_run = options['dry_run'] + batch_size = options['batch_size'] + provider_id = options['provider_id'] + + if provider_id and not PreprintProvider.objects.filter(_id=provider_id).exists(): + raise CommandError(f'Unknown preprint provider: {provider_id}') + + if dry_run: + # Synchronous preview of both passes; nothing gets queued or dispatched. + resync_preprint_dois_v1(dry_run=True, batch_size=batch_size, provider_id=provider_id) + register_missing_unversioned_dois(dry_run=True, batch_size=batch_size, provider_id=provider_id) + return + + resync_preprint_dois_v1_task.apply_async( + kwargs={ + 'batch_size': batch_size, + 'dry_run': False, + 'provider_id': provider_id, + }, + ) + self.stdout.write( + 'Started the DOI resync dispatcher. It will self-reschedule on its own until the ' + 'backlog is drained -- no need to re-run this command. Check logs/Sentry for progress.' + ) diff --git a/osf/migrations/0057_preprint_doi_resync_queued_at.py b/osf/migrations/0057_preprint_doi_resync_queued_at.py new file mode 100644 index 00000000000..409a7377f6f --- /dev/null +++ b/osf/migrations/0057_preprint_doi_resync_queued_at.py @@ -0,0 +1,17 @@ +from django.db import migrations +import osf.utils.fields + + +class Migration(migrations.Migration): + + dependencies = [ + ('osf', '0056_merge_20260918'), + ] + + operations = [ + migrations.AddField( + model_name='preprint', + name='doi_resync_queued_at', + field=osf.utils.fields.NonNaiveDateTimeField(default=None, null=True, blank=True), + ), + ] diff --git a/osf/models/preprint.py b/osf/models/preprint.py index 24616e83a76..b22830b11b6 100644 --- a/osf/models/preprint.py +++ b/osf/models/preprint.py @@ -233,6 +233,7 @@ class Preprint(DirtyFieldsMixin, VersionedGuidMixin, IdentifierMixin, Reviewable identifiers = GenericRelation(Identifier, related_query_name='preprints') preprint_doi_created = NonNaiveDateTimeField(default=None, null=True, blank=True) + doi_resync_queued_at = NonNaiveDateTimeField(default=None, null=True, blank=True) date_withdrawn = NonNaiveDateTimeField(default=None, null=True, blank=True) withdrawal_justification = models.TextField(default='', blank=True) ever_public = models.BooleanField(default=False, blank=True) diff --git a/osf_tests/test_user.py b/osf_tests/test_user.py index 99b130f24eb..ab05051340c 100644 --- a/osf_tests/test_user.py +++ b/osf_tests/test_user.py @@ -1354,7 +1354,7 @@ def test_add_system_tag(self, user): assert len(user.system_tags) == 1 - tag = Tag.all_tags.get(name=tag_name.lower(), system=True) + tag = Tag.all_tags.get(name=tag_name, system=True) assert tag in user.all_tags.all() def test_add_system_tag_instance(self, user): diff --git a/tests/identifiers/test_crossref.py b/tests/identifiers/test_crossref.py index 173e8f9a5a0..8eb8749c32e 100644 --- a/tests/identifiers/test_crossref.py +++ b/tests/identifiers/test_crossref.py @@ -96,6 +96,18 @@ def test_crossref_build_doi(self, crossref_client, preprint): assert crossref_client.build_doi(preprint) == settings.DOI_FORMAT.format(prefix=doi_prefix, guid=preprint._id) + def test_crossref_build_unversioned_doi(self, crossref_client, preprint): + doi_prefix = preprint.provider.doi_prefix + base_guid = preprint.get_guid()._id + + expected = settings.DOI_FORMAT.format(prefix=doi_prefix, guid=base_guid) + assert crossref_client.build_unversioned_doi(preprint) == expected + assert '_v' not in expected + + def test_crossref_build_unversioned_doi_matches_base_guid_not_versioned(self, crossref_client, preprint, preprint_version): + assert crossref_client.build_doi(preprint_version) != crossref_client.build_unversioned_doi(preprint_version) + assert crossref_client.build_unversioned_doi(preprint) == crossref_client.build_unversioned_doi(preprint_version) + def test_crossref_build_doi_versioned(self, crossref_client, preprint_version): doi_prefix = preprint_version.provider.doi_prefix @@ -325,6 +337,57 @@ def test_metadata_for_non_included_relation(self, crossref_client, preprint): root_without_relation = lxml.etree.fromstring(xml_without_relation) assert root_without_relation.find('.//{%s}intra_work_relation' % crossref.CROSSREF_RELATIONS) is None + def test_metadata_includes_unversioned_doi_entry(self, crossref_client, preprint): + crossref_xml = crossref_client.build_metadata(preprint, include_unversioned_doi=True) + root = lxml.etree.fromstring(crossref_xml) + + posted_contents = root.findall('.//{%s}posted_content' % crossref.CROSSREF_NAMESPACE) + assert len(posted_contents) == 2 + + versioned_dois = posted_contents[0].findall('.//{%s}doi' % crossref.CROSSREF_NAMESPACE) + assert len(versioned_dois) == 1 + assert versioned_dois[0].text == crossref_client.build_doi(preprint) + + unversioned_dois = posted_contents[1].findall('.//{%s}doi' % crossref.CROSSREF_NAMESPACE) + assert len(unversioned_dois) == 1 + assert unversioned_dois[0].text == crossref_client.build_unversioned_doi(preprint) + assert '_v' not in unversioned_dois[0].text + + unversioned_resource = posted_contents[1].find('.//{%s}resource' % crossref.CROSSREF_NAMESPACE) + base_guid = preprint.get_guid()._id + assert unversioned_resource.text == settings.DOMAIN + base_guid + + def test_metadata_unversioned_doi_uses_latest_version_metadata(self, crossref_client, preprint, preprint_version): + crossref_xml = crossref_client.build_metadata(preprint, include_unversioned_doi=True) + root = lxml.etree.fromstring(crossref_xml) + + posted_contents = root.findall('.//{%s}posted_content' % crossref.CROSSREF_NAMESPACE) + assert len(posted_contents) == 2 + + unversioned_resource = posted_contents[1].find('.//{%s}resource' % crossref.CROSSREF_NAMESPACE) + base_guid = preprint.get_guid()._id + assert unversioned_resource.text == settings.DOMAIN + base_guid + + def test_metadata_unversioned_doi_has_no_version_relations(self, crossref_client, preprint, preprint_version): + crossref_xml = crossref_client.build_metadata(preprint_version, include_unversioned_doi=True) + root = lxml.etree.fromstring(crossref_xml) + + posted_contents = root.findall('.//{%s}posted_content' % crossref.CROSSREF_NAMESPACE) + unversioned_content = posted_contents[1] + + relations = unversioned_content.findall('.//{%s}intra_work_relation' % crossref.CROSSREF_RELATIONS) + is_version_of_relations = [ + r for r in relations if r.get('relationship-type') == 'isVersionOf' + ] + assert len(is_version_of_relations) == 0 + + def test_metadata_bulk_does_not_include_unversioned_doi(self, crossref_client, preprint, preprint_version): + crossref_xml = crossref_client.build_metadata([preprint, preprint_version]) + root = lxml.etree.fromstring(crossref_xml) + + posted_contents = root.findall('.//{%s}posted_content' % crossref.CROSSREF_NAMESPACE) + assert len(posted_contents) == 2 # one per preprint, no extras + def test_metadata_for_affiliated_institutions(self, crossref_client, preprint): institution = InstitutionFactory() institution.ror_uri = 'http://ror.org/WHATisITgoodFOR/' diff --git a/tests/identifiers/test_resync_preprint_dois_v1.py b/tests/identifiers/test_resync_preprint_dois_v1.py new file mode 100644 index 00000000000..bee528a808f --- /dev/null +++ b/tests/identifiers/test_resync_preprint_dois_v1.py @@ -0,0 +1,339 @@ +import pytest +from datetime import timedelta +from unittest import mock +from django.core.management import call_command +from django.core.management.base import CommandError +from django.utils import timezone + +from osf.models import Preprint +from osf_tests.factories import PreprintFactory, PreprintProviderFactory +from osf.management.commands.resync_preprint_dois_v1 import ( + get_preprints_needing_v1_doi, + resync_preprint_dois_v1, + resync_preprint_dois_v1_task, +) +from website import settings + +pytestmark = pytest.mark.django_db + + +@pytest.fixture() +def provider(): + p = PreprintProviderFactory() + p.doi_prefix = '10.31219' + p.save() + return p + + +@pytest.fixture() +def preprint(provider): + pp = PreprintFactory(provider=provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=provider.doi_prefix, guid=pp.get_guid()._id) + pp.set_identifier_values(doi=old_doi, save=True) + return pp + + +@pytest.fixture() +def preprint_with_v1_doi(provider): + pp = PreprintFactory(provider=provider, is_published=True) + v1_doi = settings.DOI_FORMAT.format(prefix=provider.doi_prefix, guid=pp._id) + pp.set_identifier_values(doi=v1_doi, save=True) + return pp + + +class TestGetPreprrintsNeedingV1Doi: + + def test_includes_public_preprint_without_versioned_doi(self, preprint): + qs = get_preprints_needing_v1_doi() + assert preprint in qs + + def test_excludes_preprint_with_versioned_doi(self, preprint_with_v1_doi): + qs = get_preprints_needing_v1_doi() + assert preprint_with_v1_doi not in qs + + def test_excludes_preprint_with_no_doi_if_private(self, provider): + private_preprint = PreprintFactory(provider=provider, is_published=False) + private_preprint.is_public = False + private_preprint.save() + qs = get_preprints_needing_v1_doi() + assert private_preprint not in qs + + def test_includes_withdrawn_preprint_with_ever_public(self, provider): + pp = PreprintFactory(provider=provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=provider.doi_prefix, guid=pp.get_guid()._id) + pp.set_identifier_values(doi=old_doi, save=True) + pp.date_withdrawn = timezone.now() + pp.ever_public = True + pp.save() + qs = get_preprints_needing_v1_doi() + assert pp in qs + + def test_excludes_withdrawn_preprint_never_public(self, provider): + pp = PreprintFactory(provider=provider, is_published=False) + Preprint.objects.filter(pk=pp.pk).update(date_withdrawn=timezone.now()) + qs = get_preprints_needing_v1_doi() + assert pp not in qs + + def test_excludes_version_2_preprint(self, preprint): + from tests.utils import capture_notifications + with capture_notifications(): + v2 = PreprintFactory.create_version(preprint, is_published=True, set_doi=False) + old_doi = settings.DOI_FORMAT.format(prefix=preprint.provider.doi_prefix, guid=v2.get_guid()._id) + v2.set_identifier_values(doi=old_doi, save=True) + qs = get_preprints_needing_v1_doi() + assert v2 not in qs + + def test_excludes_qatest_tagged_preprint(self, preprint): + preprint.add_system_tag('qatest') + qs = get_preprints_needing_v1_doi() + assert preprint not in qs + + def test_excludes_deleted_preprint(self, preprint): + preprint.deleted = timezone.now() + preprint.save() + qs = get_preprints_needing_v1_doi() + assert preprint not in qs + + def test_provider_filter_limits_results(self, preprint, provider): + other_provider = PreprintProviderFactory() + other_provider.doi_prefix = '10.12345' + other_provider.save() + other_preprint = PreprintFactory(provider=other_provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=other_provider.doi_prefix, guid=other_preprint.get_guid()._id) + other_preprint.set_identifier_values(doi=old_doi, save=True) + + qs = get_preprints_needing_v1_doi(provider_id=provider._id) + assert preprint in qs + assert other_preprint not in qs + + def test_preprint_with_no_doi_identifier_is_included(self, provider): + pp = PreprintFactory(provider=provider, is_published=True, set_doi=False) + qs = get_preprints_needing_v1_doi() + assert pp in qs + + def test_excludes_preprint_with_fresh_in_flight_marker(self, preprint): + preprint.doi_resync_queued_at = timezone.now() + preprint.save() + qs = get_preprints_needing_v1_doi() + assert preprint not in qs + + def test_includes_preprint_with_stale_in_flight_marker(self, preprint): + stale_cutoff = timezone.now() - settings.PREPRINT_DOI_RESYNC_INFLIGHT_TIMEOUT + preprint.doi_resync_queued_at = stale_cutoff - timedelta(minutes=1) + preprint.save() + qs = get_preprints_needing_v1_doi() + assert preprint in qs + + +class TestResyncPreprintDoisV1: + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_dry_run_does_not_queue_tasks(self, mock_task, preprint): + resync_preprint_dois_v1(dry_run=True) + mock_task.apply_async.assert_not_called() + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_live_run_queues_task_for_each_preprint(self, mock_task, preprint): + resync_preprint_dois_v1(dry_run=False) + mock_task.apply_async.assert_called_once_with(kwargs={'preprint_id': preprint._id}) + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_batch_size_limits_processed_count(self, mock_task, provider): + preprints = [] + for _ in range(5): + pp = PreprintFactory(provider=provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=provider.doi_prefix, guid=pp.get_guid()._id) + pp.set_identifier_values(doi=old_doi, save=True) + preprints.append(pp) + + resync_preprint_dois_v1(dry_run=False, batch_size=2) + assert mock_task.apply_async.call_count == 2 + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_skips_provider_without_doi_prefix(self, mock_task, provider): + no_prefix_provider = PreprintProviderFactory() + no_prefix_provider.doi_prefix = '' + no_prefix_provider.save() + pp = PreprintFactory(provider=no_prefix_provider, is_published=True) + old_doi = '10.000/old-doi' + pp.set_identifier_values(doi=old_doi, save=True) + + resync_preprint_dois_v1(dry_run=False) + queued_ids = [ + call.kwargs['kwargs']['preprint_id'] + for call in mock_task.apply_async.call_args_list + ] + assert pp._id not in queued_ids + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_provider_filter_is_applied(self, mock_task, preprint, provider): + other_provider = PreprintProviderFactory() + other_provider.doi_prefix = '10.99999' + other_provider.save() + other_pp = PreprintFactory(provider=other_provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=other_provider.doi_prefix, guid=other_pp.get_guid()._id) + other_pp.set_identifier_values(doi=old_doi, save=True) + + resync_preprint_dois_v1(dry_run=False, provider_id=provider._id) + + queued_ids = [ + call.kwargs['kwargs']['preprint_id'] + for call in mock_task.apply_async.call_args_list + ] + assert preprint._id in queued_ids + assert other_pp._id not in queued_ids + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_already_versioned_doi_is_not_queued(self, mock_task, preprint_with_v1_doi): + resync_preprint_dois_v1(dry_run=False) + queued_ids = [ + call.kwargs['kwargs']['preprint_id'] + for call in mock_task.apply_async.call_args_list + ] + assert preprint_with_v1_doi._id not in queued_ids + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_queuing_marks_preprint_in_flight(self, mock_task, preprint): + assert preprint.doi_resync_queued_at is None + resync_preprint_dois_v1(dry_run=False) + preprint.refresh_from_db() + assert preprint.doi_resync_queued_at is not None + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_dry_run_does_not_mark_in_flight(self, mock_task, preprint): + resync_preprint_dois_v1(dry_run=True) + preprint.refresh_from_db() + assert preprint.doi_resync_queued_at is None + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_capacity_limits_queued_count(self, mock_task, provider): + for _ in range(5): + pp = PreprintFactory(provider=provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=provider.doi_prefix, guid=pp.get_guid()._id) + pp.set_identifier_values(doi=old_doi, save=True) + + result = resync_preprint_dois_v1(dry_run=False, batch_size=1000, capacity=2) + assert mock_task.apply_async.call_count == 2 + assert result.queued == 2 + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_zero_capacity_queues_nothing(self, mock_task, preprint): + result = resync_preprint_dois_v1(dry_run=False, capacity=0) + mock_task.apply_async.assert_not_called() + assert result.queued == 0 + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_in_flight_preprint_reduces_available_capacity(self, mock_task, provider): + already_queued = PreprintFactory(provider=provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=provider.doi_prefix, guid=already_queued.get_guid()._id) + already_queued.set_identifier_values(doi=old_doi, save=True) + already_queued.doi_resync_queued_at = timezone.now() + already_queued.save() + + needs_resync = PreprintFactory(provider=provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=provider.doi_prefix, guid=needs_resync.get_guid()._id) + needs_resync.set_identifier_values(doi=old_doi, save=True) + + with mock.patch( + 'osf.management.commands.resync_preprint_dois_v1.settings.PREPRINT_DOI_RESYNC_MAX_IN_FLIGHT', 1 + ): + result = resync_preprint_dois_v1(dry_run=False) + + mock_task.apply_async.assert_not_called() + assert result.queued == 0 + + +class TestResyncPreprintDoisV1Task: + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_reschedules_while_work_remains(self, mock_task, provider): + for _ in range(3): + pp = PreprintFactory(provider=provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=provider.doi_prefix, guid=pp.get_guid()._id) + pp.set_identifier_values(doi=old_doi, save=True) + + with mock.patch.object(resync_preprint_dois_v1_task, 'apply_async') as mock_reschedule: + resync_preprint_dois_v1_task(batch_size=1, dry_run=False) + + mock_reschedule.assert_called_once() + assert mock_reschedule.call_args.kwargs['countdown'] == ( + settings.PREPRINT_DOI_RESYNC_DISPATCH_INTERVAL.total_seconds() + ) + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_does_not_reschedule_once_backlog_is_empty(self, mock_task, preprint): + with mock.patch.object(resync_preprint_dois_v1_task, 'apply_async') as mock_reschedule: + resync_preprint_dois_v1_task(batch_size=1000, dry_run=False) + + mock_reschedule.assert_not_called() + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_dry_run_never_reschedules(self, mock_task, provider): + for _ in range(3): + pp = PreprintFactory(provider=provider, is_published=True) + old_doi = settings.DOI_FORMAT.format(prefix=provider.doi_prefix, guid=pp.get_guid()._id) + pp.set_identifier_values(doi=old_doi, save=True) + + with mock.patch.object(resync_preprint_dois_v1_task, 'apply_async') as mock_reschedule: + resync_preprint_dois_v1_task(batch_size=1, dry_run=True) + + mock_reschedule.assert_not_called() + + +class TestResyncPreprintDoisV1Command: + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.resync_preprint_dois_v1_task.apply_async') + def test_live_run_starts_dispatcher_once(self, mock_apply_async, preprint): + call_command('resync_preprint_dois_v1', batch_size=50, provider_id=preprint.provider._id) + + mock_apply_async.assert_called_once_with( + kwargs={ + 'batch_size': 50, + 'dry_run': False, + 'provider_id': preprint.provider._id, + }, + ) + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.resync_preprint_dois_v1_task.apply_async') + def test_live_run_uses_default_options(self, mock_apply_async): + call_command('resync_preprint_dois_v1') + + mock_apply_async.assert_called_once_with( + kwargs={'batch_size': 1000, 'dry_run': False, 'provider_id': None}, + ) + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + @mock.patch('osf.management.commands.resync_preprint_dois_v1.resync_preprint_dois_v1_task.apply_async') + def test_live_run_does_not_dispatch_to_crossref_synchronously(self, mock_apply_async, mock_identifier_task, preprint): + # handle() should hand off to the task rather than talking to Crossref itself + call_command('resync_preprint_dois_v1') + + mock_identifier_task.apply_async.assert_not_called() + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.register_missing_unversioned_dois') + @mock.patch('osf.management.commands.resync_preprint_dois_v1.resync_preprint_dois_v1') + @mock.patch('osf.management.commands.resync_preprint_dois_v1.resync_preprint_dois_v1_task.apply_async') + def test_dry_run_previews_both_passes_without_dispatching( + self, mock_apply_async, mock_v1_pass, mock_unversioned_pass, preprint + ): + call_command('resync_preprint_dois_v1', dry_run=True, batch_size=25, provider_id=preprint.provider._id) + + mock_v1_pass.assert_called_once_with(dry_run=True, batch_size=25, provider_id=preprint.provider._id) + mock_unversioned_pass.assert_called_once_with(dry_run=True, batch_size=25, provider_id=preprint.provider._id) + mock_apply_async.assert_not_called() + + @pytest.mark.parametrize('dry_run', [True, False]) + @mock.patch('osf.management.commands.resync_preprint_dois_v1.resync_preprint_dois_v1_task.apply_async') + def test_unknown_provider_raises_and_dispatches_nothing(self, mock_apply_async, dry_run, preprint): + with pytest.raises(CommandError, match='Unknown preprint provider: not-a-provider'): + call_command('resync_preprint_dois_v1', dry_run=dry_run, provider_id='not-a-provider') + + mock_apply_async.assert_not_called() + + @mock.patch('osf.management.commands.resync_preprint_dois_v1.async_request_identifier_update') + def test_dry_run_does_not_queue_or_mark_in_flight(self, mock_identifier_task, preprint): + call_command('resync_preprint_dois_v1', dry_run=True) + + mock_identifier_task.apply_async.assert_not_called() + preprint.refresh_from_db() + assert preprint.doi_resync_queued_at is None diff --git a/website/identifiers/clients/crossref.py b/website/identifiers/clients/crossref.py index 309a994abc0..abd0fc2f7ec 100644 --- a/website/identifiers/clients/crossref.py +++ b/website/identifiers/clients/crossref.py @@ -41,13 +41,25 @@ def build_doi(self, preprint): prefix = preprint.provider.doi_prefix return settings.DOI_FORMAT.format(prefix=prefix, guid=preprint._id) - def build_metadata(self, preprint, include_relation=True): + def build_unversioned_doi(self, preprint): + prefix = preprint.provider.doi_prefix + return settings.DOI_FORMAT.format(prefix=prefix, guid=preprint.get_guid()._id) + + def _get_latest_published_version(self, preprint): + latest = preprint.get_preprint_versions( + date_published__isnull=False, + include_rejected=False, + ).first() + return latest or preprint + + def build_metadata(self, preprint, include_relation=True, include_unversioned_doi=False): """Return the crossref metadata XML document for a given preprint as a string for DOI minting purposes :param preprint: the preprint, or list of preprints to build metadata for """ if isinstance(preprint, (list, QuerySet)): preprints = preprint + include_unversioned_doi = False # not supported for bulk batches else: preprints = [preprint] @@ -74,6 +86,10 @@ def build_metadata(self, preprint, include_relation=True): for preprint in preprints: body.append(self.build_posted_content(preprint, element, include_relation)) + if include_unversioned_doi: + latest = self._get_latest_published_version(preprints[0]) + body.append(self.build_unversioned_posted_content(latest, element)) + root = element.doi_batch( head, body, @@ -82,10 +98,16 @@ def build_metadata(self, preprint, include_relation=True): root.attrib['{%s}schemaLocation' % XSI] = CROSSREF_SCHEMA_LOCATION return lxml.etree.tostring(root) - def build_posted_content(self, preprint, element, include_relation): + @staticmethod + def _doi_suffix(doi): + return doi.split('/', 1)[-1].lower() + + def build_posted_content(self, preprint, element, include_relation, doi_override=None, resource_override=None, include_version_relations=True): """Build the element for a single preprint preprint - preprint to build posted_content for element - namespace element to use when building parts of the XML structure + doi_override - if provided, use this DOI value instead of the preprint's own DOI + resource_override - if provided, use this URL as the instead of the default """ from osf.models import SpamStatus @@ -138,7 +160,7 @@ def build_posted_content(self, preprint, element, include_relation): preprint_versions = preprint.get_preprint_versions( versioned_guids__version__lt=preprint.version, include_rejected=False, - ) + ) if include_version_relations else [] if preprint_versions: for previous_version in preprint_versions: @@ -157,15 +179,29 @@ def build_posted_content(self, preprint, element, include_relation): posted_content.append(relations_program) minted_doi = preprint.get_identifier_value('doi') - doi = minted_doi or self.build_doi(preprint) + if minted_doi and self._doi_suffix(minted_doi) == self._doi_suffix(self.build_unversioned_doi(preprint)): + minted_doi = None + doi = doi_override or minted_doi or self.build_doi(preprint) + resource_url = resource_override if resource_override is not None else settings.DOMAIN + preprint._id doi_data = [ element.doi(doi), - element.resource(settings.DOMAIN + preprint._id) + element.resource(resource_url) ] posted_content.append(element.doi_data(*doi_data)) return posted_content + def build_unversioned_posted_content(self, preprint, element): + base_guid = preprint.get_guid()._id + return self.build_posted_content( + preprint, + element, + include_relation=False, + include_version_relations=False, + doi_override=self.build_unversioned_doi(preprint), + resource_override=settings.DOMAIN + base_guid, + ) + def _process_crossref_name(self, contributor): # Adapted from logic used in `api/citations/utils.py` # If the user has a family and given name, use those @@ -251,7 +287,7 @@ def _build_url(self, **query): def create_identifier(self, preprint, category, include_relation=True): if category == 'doi': - metadata = self.build_metadata(preprint, include_relation) + metadata = self.build_metadata(preprint, include_relation, include_unversioned_doi=True) doi = self.build_doi(preprint) username, password = self.get_credentials() logger.info(f'Sending metadata for DOI {doi}:\n{metadata}') diff --git a/website/settings/defaults.py b/website/settings/defaults.py index 41e4190390d..111375f851d 100644 --- a/website/settings/defaults.py +++ b/website/settings/defaults.py @@ -393,6 +393,13 @@ def parent_dir(path): # Crossref has a second metadata api that uses JSON with different features CROSSREF_JSON_API_URL = 'https://api.crossref.org/' +# Max preprints to work on same time and wait for the confirmation +PREPRINT_DOI_RESYNC_MAX_IN_FLIGHT = 1000 +# Reschedule interval while there's still eligible work and free capacity. +PREPRINT_DOI_RESYNC_DISPATCH_INTERVAL = timedelta(minutes=5) +# Preprint that hasn't received a Crossref confirmation within this window is treated as `stale` and it becomes eligible to be resync again. +PREPRINT_DOI_RESYNC_INFLIGHT_TIMEOUT = timedelta(hours=24) + # Leave as `None` for production, test/staging/local envs must set SHARE_PROVIDER_PREPEND = None @@ -508,6 +515,7 @@ class CeleryConfig: 'osf.management.commands.daily_reporters_go', 'osf.management.commands.monthly_reporters_go', 'osf.management.commands.fetch_cedar_metadata_templates', + 'osf.management.commands.resync_preprint_dois_v1', 'osf.metrics.reporters', 'scripts.remove_after_use.merge_notification_subscription_provider_ct', # Items below are celery task names, not python module paths. @@ -673,6 +681,7 @@ class CeleryConfig: 'osf.management.commands.archive_registrations_on_IA', 'osf.management.commands.approve_pending_schema_responses', 'osf.management.commands.sync_doi_metadata', + 'osf.management.commands.resync_preprint_dois_v1', 'api.providers.tasks', 'api.users.tasks', 'osf.utils.download_telemetry',