From 67433fbbcceef700cb211a44000f5cf55908b6b6 Mon Sep 17 00:00:00 2001 From: Tobias Macey Date: Thu, 2 Jul 2026 18:15:33 -0400 Subject: [PATCH 1/2] Add per-(etl_source, resource_type) ETL write-ownership guard Pull-ETL batch loaders (load_courses, load_programs, canvas stale sweep) full-sync prune/unpublish anything not present in the current batch. If a push pipeline (webhook or Dagster asset) also writes the same (etl_source, resource_type), each side treats the other's writes as missing, causing unpublish/republish flapping. There was no source-of-truth flag. Add ETLSourceOwnership (etl_source, resource_type, mode: pull|push), defaulting to pull so existing sources are unaffected. Wire pull_write_allowed into load_courses/load_programs (skip write+prune entirely for a push-owned pair) and the canvas stale-course sweep (skip prune only). Add assert_push_allowed for the upcoming generic learning_resources webhook loader (PR #3557) to call before writing. This is the prerequisite for cutting any source (SEE/Sloan first, already dual-modeled as a Dagster asset) from pull to push; no source is cut over by this change. Co-Authored-By: Claude Sonnet 5 --- learning_resources/admin.py | 10 +++ learning_resources/etl/loaders.py | 17 ++++ learning_resources/etl/loaders_test.py | 54 +++++++++++++ learning_resources/etl/ownership.py | 80 +++++++++++++++++++ learning_resources/etl/ownership_test.py | 80 +++++++++++++++++++ learning_resources/factories.py | 12 +++ .../migrations/0116_etlsourceownership.py | 50 ++++++++++++ learning_resources/models.py | 29 +++++++ learning_resources/tasks.py | 10 +++ learning_resources/tasks_test.py | 35 +++++++- 10 files changed, 376 insertions(+), 1 deletion(-) create mode 100644 learning_resources/etl/ownership.py create mode 100644 learning_resources/etl/ownership_test.py create mode 100644 learning_resources/migrations/0116_etlsourceownership.py 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..2656c712c7 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, @@ -659,6 +660,15 @@ def sync_canvas_courses(canvas_course_ids, overwrite): canvas_readable_ids.append(resource_readable_id) if not canvas_course_ids: + if not pull_write_allowed( + ETLSource.canvas.name, LearningResourceType.course.name + ): + log.info( + "Skipping canvas stale-course prune: %s/%s is push-owned", + ETLSource.canvas.name, + LearningResourceType.course.name, + ) + return stale_courses = LearningResource.objects.filter( etl_source=ETLSource.canvas.name ).exclude(readable_id__in=canvas_readable_ids) diff --git a/learning_resources/tasks_test.py b/learning_resources/tasks_test.py index 0797411892..301d73b0ba 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,38 @@ 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_stale_prune_when_push_owned(settings, mocker): + """ + sync_canvas_courses should not 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_bucket = mocker.Mock() + mock_bucket.objects.filter.return_value = [] + mocker.patch( + "learning_resources.tasks.get_bucket_by_name", return_value=mock_bucket + ) + + 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) + + lr_stale.refresh_from_db() + assert lr_stale.published is True + + @pytest.mark.parametrize( ("etl_source", "archive_path", "overwrite"), [ From 970d8666d6c305fb696f04d41c4ee30a0f730302 Mon Sep 17 00:00:00 2001 From: Tobias Macey Date: Thu, 2 Jul 2026 18:34:47 -0400 Subject: [PATCH 2/2] Fix canvas ownership guard to also block ingestion, not just prune The ownership check in sync_canvas_courses only guarded the stale- course prune block, not the S3 archive ingestion loop above it, so a push-owned canvas source would still receive pull-side writes - defeating the point of the guard. Move the pull_write_allowed check to the top of the function, matching load_courses/load_programs: skip the entire sync (ingestion + prune) when push-owned. Flagged by Sentry Seer's automated review on PR #3565. Co-Authored-By: Claude Sonnet 5 --- learning_resources/tasks.py | 16 +++++++--------- learning_resources/tasks_test.py | 15 +++++++-------- 2 files changed, 14 insertions(+), 17 deletions(-) diff --git a/learning_resources/tasks.py b/learning_resources/tasks.py index 2656c712c7..c269f66ae9 100644 --- a/learning_resources/tasks.py +++ b/learning_resources/tasks.py @@ -624,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) @@ -660,15 +667,6 @@ def sync_canvas_courses(canvas_course_ids, overwrite): canvas_readable_ids.append(resource_readable_id) if not canvas_course_ids: - if not pull_write_allowed( - ETLSource.canvas.name, LearningResourceType.course.name - ): - log.info( - "Skipping canvas stale-course prune: %s/%s is push-owned", - ETLSource.canvas.name, - LearningResourceType.course.name, - ) - return stale_courses = LearningResource.objects.filter( etl_source=ETLSource.canvas.name ).exclude(readable_id__in=canvas_readable_ids) diff --git a/learning_resources/tasks_test.py b/learning_resources/tasks_test.py index 301d73b0ba..1615276bc4 100644 --- a/learning_resources/tasks_test.py +++ b/learning_resources/tasks_test.py @@ -838,10 +838,10 @@ 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_stale_prune_when_push_owned(settings, mocker): +def test_sync_canvas_courses_skips_entirely_when_push_owned(settings, mocker): """ - sync_canvas_courses should not unpublish/delete stale courses once canvas - courses are push-owned. + 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( @@ -850,11 +850,8 @@ def test_sync_canvas_courses_skips_stale_prune_when_push_owned(settings, mocker) mode=ETLSourceOwnership.Mode.PUSH, ) mocker.patch("learning_resources.tasks.resource_unpublished_actions") - mock_bucket = mocker.Mock() - mock_bucket.objects.filter.return_value = [] - mocker.patch( - "learning_resources.tasks.get_bucket_by_name", return_value=mock_bucket - ) + 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", @@ -866,6 +863,8 @@ def test_sync_canvas_courses_skips_stale_prune_when_push_owned(settings, mocker) 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