Add per-source ETL write-ownership guard (pull vs push)

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)

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)
 

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

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)

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)

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

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")},
+        ),
+    ]