diff --git a/learning_resources/admin.py b/learning_resources/admin.py index dd18f6ac36..a53d266dcd 100644 --- a/learning_resources/admin.py +++ b/learning_resources/admin.py @@ -270,6 +270,15 @@ class ContentSummarizerConfigurationAdmin(admin.ModelAdmin): ) +class ETLSourceOwnershipAdmin(admin.ModelAdmin): + """ETLSourceOwnership Admin""" + + model = models.ETLSourceOwnership + list_display = ("etl_source", "resource_type", "mode", "updated_on") + list_filter = ("etl_source", "mode") + search_fields = ("etl_source", "resource_type") + + admin.site.register(models.LearningResourceTopic, LearningResourceTopicAdmin) admin.site.register(models.LearningResourceInstructor, LearningResourceInstructorAdmin) admin.site.register(models.LearningResource, LearningResourceAdmin) @@ -285,3 +294,4 @@ class ContentSummarizerConfigurationAdmin(admin.ModelAdmin): admin.site.register( models.ContentSummarizerConfiguration, ContentSummarizerConfigurationAdmin ) +admin.site.register(models.ETLSourceOwnership, ETLSourceOwnershipAdmin) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index fedaf7a3d6..516bfb4391 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -32,6 +32,7 @@ ResourceNextRunConfig, ) from learning_resources.etl.exceptions import ExtractException +from learning_resources.etl.ownership import pull_write_allowed from learning_resources.etl.utils import most_common_topics from learning_resources.models import ( ContentFile, @@ -632,6 +633,14 @@ def load_courses( Returns: A list of course LearningResources """ + if not pull_write_allowed(etl_source, LearningResourceType.course.name): + log.info( + "Skipping pull-ETL write for %s/%s: ownership is push", + etl_source, + LearningResourceType.course.name, + ) + return [] + blocklist = load_course_blocklist() duplicates = load_course_duplicates(etl_source) @@ -888,6 +897,14 @@ def load_programs( For MITx Online data, each deferred child program may map to either PROGRAM_PROGRAMS or PROGRAM_COURSES based on child `display_mode`. """ + if not pull_write_allowed(etl_source, LearningResourceType.program.name): + log.info( + "Skipping pull-ETL write for %s/%s: ownership is push", + etl_source, + LearningResourceType.program.name, + ) + return [] + blocklist = load_course_blocklist() duplicates = load_course_duplicates(etl_source) diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index c0a195dd43..39fb9482b7 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -69,6 +69,7 @@ from learning_resources.factories import ( ContentFileFactory, CourseFactory, + ETLSourceOwnershipFactory, LearningResourceContentTagFactory, LearningResourceDepartmentFactory, LearningResourceFactory, @@ -89,6 +90,7 @@ from learning_resources.models import ( ContentFile, Course, + ETLSourceOwnership, LearningResource, LearningResourceImage, LearningResourceOfferor, @@ -1415,6 +1417,31 @@ def test_load_courses(mocker, mock_blocklist, mock_duplicates, prune): assert course_to_unpublish.learning_resource.published is not prune +def test_load_courses_skips_write_when_push_owned(mocker): + """load_courses should no-op (no writes, no prune) for a push-owned source""" + ETLSourceOwnershipFactory.create( + etl_source=ETLSource.see.name, + resource_type=LearningResourceType.course.name, + mode=ETLSourceOwnership.Mode.PUSH, + ) + course_to_unpublish = CourseFactory.create(etl_source=ETLSource.see.name) + + mock_load_course = mocker.patch( + "learning_resources.etl.loaders.load_course", autospec=True + ) + + result = load_courses( + ETLSource.see.name, + [{"readable_id": "some-course"}], + config=CourseLoaderConfig(prune=True), + ) + + assert result == [] + mock_load_course.assert_not_called() + course_to_unpublish.refresh_from_db() + assert course_to_unpublish.learning_resource.published is True + + def test_load_programs(mocker, mock_blocklist, mock_duplicates): """Test that load_programs calls the expected functions""" program_data = [{"courses": [{"platform": "a"}, {}], "id": 5}] @@ -1434,6 +1461,33 @@ def test_load_programs(mocker, mock_blocklist, mock_duplicates): mock_duplicates.assert_called_once_with("mitx") +def test_load_programs_skips_write_when_push_owned(mocker): + """load_programs should no-op (no writes, no prune) for a push-owned source""" + ETLSourceOwnershipFactory.create( + etl_source=ETLSource.see.name, + resource_type=LearningResourceType.program.name, + mode=ETLSourceOwnership.Mode.PUSH, + ) + program_to_unpublish = ProgramFactory.create( + learning_resource__etl_source=ETLSource.see.name + ) + + mock_load_program = mocker.patch( + "learning_resources.etl.loaders.load_program", autospec=True + ) + + result = load_programs( + ETLSource.see.name, + [{"courses": [], "id": 1}], + config=ProgramLoaderConfig(prune=True), + ) + + assert result == [] + mock_load_program.assert_not_called() + program_to_unpublish.refresh_from_db() + assert program_to_unpublish.learning_resource.published is True + + @pytest.fixture def mitxonline_program_children_fixture(mocker, settings): """ diff --git a/learning_resources/etl/ownership.py b/learning_resources/etl/ownership.py new file mode 100644 index 0000000000..22a5d35825 --- /dev/null +++ b/learning_resources/etl/ownership.py @@ -0,0 +1,80 @@ +""" +Per-(etl_source, resource_type) write-ownership guard. + +Batch pull-ETL loaders (load_courses, load_programs, the canvas stale-course +sweep) implement full-sync unpublish/delete: anything not present in the +current batch is treated as removed upstream. If a push pipeline (webhook or +Dagster asset) also writes the same (etl_source, resource_type), each side's +full sync treats the other side's writes as missing, causing +unpublish/republish flapping. + +ETLSourceOwnership records which pipeline currently owns writes for a given +pair. A missing row defaults to "pull" so existing sources are unaffected +until a row is explicitly created to cut a source over to push. Pull loaders +should call `pull_write_allowed` (or the stricter `assert_pull_allowed`) +before writing/pruning; push loaders should call `assert_push_allowed` +before writing, so a push pipeline can't silently write into a pair that +pull-ETL still owns and will prune on its next run. +""" + +import logging + +from learning_resources.models import ETLSourceOwnership + +log = logging.getLogger(__name__) + +PULL = ETLSourceOwnership.Mode.PULL +PUSH = ETLSourceOwnership.Mode.PUSH + + +class OwnershipError(Exception): + """Raised when a loader writes to an (etl_source, resource_type) it does not own.""" + + +def get_ownership_mode(etl_source: str, resource_type: str) -> str: + """ + Return the write-ownership mode for an (etl_source, resource_type) pair. + + Defaults to PULL when no row exists. + """ + row = ETLSourceOwnership.objects.filter( + etl_source=etl_source, resource_type=resource_type + ).first() + return row.mode if row else PULL + + +def is_push_owned(etl_source: str, resource_type: str) -> bool: + """Whether an (etl_source, resource_type) pair is push-owned.""" + return get_ownership_mode(etl_source, resource_type) == PUSH + + +def pull_write_allowed(etl_source: str, resource_type: str) -> bool: + """Whether a pull-ETL loader may write/prune this (etl_source, resource_type).""" + return not is_push_owned(etl_source, resource_type) + + +def assert_pull_allowed(etl_source: str, resource_type: str) -> None: + """Hard guard for pull loaders: raise if this pair has been cut over to push.""" + if is_push_owned(etl_source, resource_type): + msg = ( + f"Refusing pull write: {etl_source}/{resource_type} is push-owned. " + "This source has been cut over to a push pipeline; the pull-ETL " + "loader must not write or prune it." + ) + raise OwnershipError(msg) + + +def assert_push_allowed(etl_source: str, resource_type: str) -> None: + """ + Guard for push (webhook/event) loaders: raise unless this + (etl_source, resource_type) has been explicitly cut over to push + ownership. Prevents a push loader from writing into a pair that pull-ETL + still owns and will prune/overwrite on its next scheduled run. + """ + if not is_push_owned(etl_source, resource_type): + msg = ( + f"Refusing push write: {etl_source}/{resource_type} is pull-owned. " + "Create an ETLSourceOwnership row set to push before enabling " + "push writes for this source." + ) + raise OwnershipError(msg) diff --git a/learning_resources/etl/ownership_test.py b/learning_resources/etl/ownership_test.py new file mode 100644 index 0000000000..ff9f34ce54 --- /dev/null +++ b/learning_resources/etl/ownership_test.py @@ -0,0 +1,80 @@ +"""Tests for learning_resources.etl.ownership""" + +import pytest + +from learning_resources.constants import LearningResourceType +from learning_resources.etl.constants import ETLSource +from learning_resources.etl.ownership import ( + OwnershipError, + assert_pull_allowed, + assert_push_allowed, + get_ownership_mode, + is_push_owned, + pull_write_allowed, +) +from learning_resources.factories import ETLSourceOwnershipFactory +from learning_resources.models import ETLSourceOwnership + +pytestmark = pytest.mark.django_db + + +def test_defaults_to_pull_when_no_row(): + """No ETLSourceOwnership row means pull ownership.""" + assert get_ownership_mode(ETLSource.see.name, LearningResourceType.course.name) == ( + ETLSourceOwnership.Mode.PULL + ) + assert is_push_owned(ETLSource.see.name, LearningResourceType.course.name) is False + assert ( + pull_write_allowed(ETLSource.see.name, LearningResourceType.course.name) is True + ) + + +def test_push_owned_row_flips_ownership(): + """A push row for one (etl_source, resource_type) only affects that pair.""" + ETLSourceOwnershipFactory.create( + etl_source=ETLSource.see.name, + resource_type=LearningResourceType.course.name, + mode=ETLSourceOwnership.Mode.PUSH, + ) + + assert is_push_owned(ETLSource.see.name, LearningResourceType.course.name) is True + assert ( + pull_write_allowed(ETLSource.see.name, LearningResourceType.course.name) + is False + ) + + # A different resource_type for the same source is unaffected. + assert is_push_owned(ETLSource.see.name, LearningResourceType.program.name) is False + # A different source is unaffected. + assert ( + is_push_owned(ETLSource.mitxonline.name, LearningResourceType.course.name) + is False + ) + + +def test_assert_pull_allowed_raises_when_push_owned(): + """assert_pull_allowed should raise once a pair is cut over to push.""" + ETLSourceOwnershipFactory.create( + etl_source=ETLSource.see.name, + resource_type=LearningResourceType.course.name, + mode=ETLSourceOwnership.Mode.PUSH, + ) + with pytest.raises(OwnershipError): + assert_pull_allowed(ETLSource.see.name, LearningResourceType.course.name) + + # No-op when pull-owned (the default). + assert_pull_allowed(ETLSource.mitxonline.name, LearningResourceType.course.name) + + +def test_assert_push_allowed_raises_when_pull_owned(): + """assert_push_allowed should raise for a pair that is still pull-owned.""" + with pytest.raises(OwnershipError): + assert_push_allowed(ETLSource.mitxonline.name, LearningResourceType.course.name) + + ETLSourceOwnershipFactory.create( + etl_source=ETLSource.see.name, + resource_type=LearningResourceType.course.name, + mode=ETLSourceOwnership.Mode.PUSH, + ) + # No-op once explicitly cut over to push. + assert_push_allowed(ETLSource.see.name, LearningResourceType.course.name) diff --git a/learning_resources/factories.py b/learning_resources/factories.py index 283aa4b1ed..ca8cf7a54a 100644 --- a/learning_resources/factories.py +++ b/learning_resources/factories.py @@ -952,3 +952,15 @@ class ContentSummarizerConfigurationFactory(DjangoModelFactory): class Meta: model = models.ContentSummarizerConfiguration django_get_or_create = ("platform", "llm_model") + + +class ETLSourceOwnershipFactory(DjangoModelFactory): + """Factory for ETLSourceOwnership""" + + etl_source = factory.Faker("word") + resource_type = constants.LearningResourceType.course.name + mode = models.ETLSourceOwnership.Mode.PULL + + class Meta: + model = models.ETLSourceOwnership + django_get_or_create = ("etl_source", "resource_type") diff --git a/learning_resources/migrations/0116_etlsourceownership.py b/learning_resources/migrations/0116_etlsourceownership.py new file mode 100644 index 0000000000..888d78d3cc --- /dev/null +++ b/learning_resources/migrations/0116_etlsourceownership.py @@ -0,0 +1,50 @@ +# Generated by Django 4.2.30 on 2026-07-02 00:00 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("learning_resources", "0115_videoplaylist_parent_learning_resource"), + ] + + operations = [ + migrations.CreateModel( + name="ETLSourceOwnership", + fields=[ + ( + "id", + models.AutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("created_on", models.DateTimeField(auto_now_add=True, db_index=True)), + ("updated_on", models.DateTimeField(auto_now=True)), + ("etl_source", models.CharField(max_length=32)), + ("resource_type", models.CharField(max_length=32)), + ( + "mode", + models.CharField( + choices=[ + ("pull", "Pull (Celery ETL)"), + ("push", "Push (webhook/event pipeline)"), + ], + default="pull", + max_length=8, + ), + ), + ], + options={ + "verbose_name": "ETL source ownership", + "verbose_name_plural": "ETL source ownerships", + "abstract": False, + }, + ), + migrations.AlterUniqueTogether( + name="etlsourceownership", + unique_together={("etl_source", "resource_type")}, + ), + ] diff --git a/learning_resources/models.py b/learning_resources/models.py index 02505787f8..b53bfdbf95 100644 --- a/learning_resources/models.py +++ b/learning_resources/models.py @@ -1589,3 +1589,32 @@ class ContentSummarizerConfiguration(TimestampedModel): default=list, ) is_active = models.BooleanField(default=True) + + +class ETLSourceOwnership(TimestampedModel): + """ + Declares which pipeline owns writes for an (etl_source, resource_type) pair. + + A missing row means "pull" (the historical default: Celery ETL is the + source of truth and its full-sync loaders may prune/unpublish anything + they didn't just write). Adding a "push" row hands write ownership to a + webhook/event pipeline (e.g. Dagster) and tells pull-ETL loaders to skip + writing and pruning that pair entirely, so the two writers stop treating + each other's rows as missing. See learning_resources/etl/ownership.py. + """ + + class Mode(models.TextChoices): + PULL = "pull", "Pull (Celery ETL)" + PUSH = "push", "Push (webhook/event pipeline)" + + etl_source = models.CharField(max_length=32) + resource_type = models.CharField(max_length=32) + mode = models.CharField(max_length=8, choices=Mode.choices, default=Mode.PULL) + + class Meta: + unique_together = ("etl_source", "resource_type") + verbose_name = "ETL source ownership" + verbose_name_plural = "ETL source ownerships" + + def __str__(self): + return f"{self.etl_source}/{self.resource_type}: {self.mode}" diff --git a/learning_resources/tasks.py b/learning_resources/tasks.py index c5a3524a19..c269f66ae9 100644 --- a/learning_resources/tasks.py +++ b/learning_resources/tasks.py @@ -34,6 +34,7 @@ load_learning_materials, load_run_dependent_values, ) +from learning_resources.etl.ownership import pull_write_allowed from learning_resources.etl.pipelines import ocw_courses_etl from learning_resources.etl.utils import ( get_bucket_by_name, @@ -623,6 +624,13 @@ def sync_canvas_courses(canvas_course_ids, overwrite): Args: overwrite (bool): Whether to overwrite existing content files """ + if not pull_write_allowed(ETLSource.canvas.name, LearningResourceType.course.name): + log.info( + "Skipping pull-ETL write for %s/%s: ownership is push", + ETLSource.canvas.name, + LearningResourceType.course.name, + ) + return bucket = get_bucket_by_name(settings.COURSE_ARCHIVE_BUCKET_NAME) s3_prefix = get_s3_prefix_for_source(ETLSource.canvas.name) diff --git a/learning_resources/tasks_test.py b/learning_resources/tasks_test.py index 0797411892..1615276bc4 100644 --- a/learning_resources/tasks_test.py +++ b/learning_resources/tasks_test.py @@ -16,10 +16,11 @@ from learning_resources.etl.constants import MARKETING_PAGE_FILE_TYPE, ETLSource from learning_resources.factories import ( ContentFileFactory, + ETLSourceOwnershipFactory, LearningResourceFactory, LearningResourceRunFactory, ) -from learning_resources.models import ContentFile, LearningResource +from learning_resources.models import ContentFile, ETLSourceOwnership, LearningResource from learning_resources.tasks import ( cleanup_deleted_content_files, get_ocw_data, @@ -837,6 +838,37 @@ def test_sync_canvas_courses(settings, mocker, django_assert_num_queries, canvas assert mock_ingest_course.call_count == 2 +def test_sync_canvas_courses_skips_entirely_when_push_owned(settings, mocker): + """ + sync_canvas_courses should not ingest archives or unpublish/delete stale + courses once canvas courses are push-owned. + """ + settings.CANVAS_COURSE_BUCKET_PREFIX = "canvas/" + ETLSourceOwnershipFactory.create( + etl_source=ETLSource.canvas.name, + resource_type=LearningResourceType.course.name, + mode=ETLSourceOwnership.Mode.PUSH, + ) + mocker.patch("learning_resources.tasks.resource_unpublished_actions") + mock_get_bucket = mocker.patch("learning_resources.tasks.get_bucket_by_name") + mock_ingest_course = mocker.patch("learning_resources.tasks.ingest_canvas_course") + + lr_stale = LearningResourceFactory.create( + readable_id="course3", + etl_source=ETLSource.canvas.name, + published=True, + test_mode=True, + resource_type="course", + ) + + sync_canvas_courses(canvas_course_ids=None, overwrite=False) + + mock_get_bucket.assert_not_called() + mock_ingest_course.assert_not_called() + lr_stale.refresh_from_db() + assert lr_stale.published is True + + @pytest.mark.parametrize( ("etl_source", "archive_path", "overwrite"), [