Add UpstreamPulp.remote_policy for remotes created during replication

replicate() never set Remote.policy, so new remotes defaulted to immediate
and downloaded all artifacts. Let UpstreamPulp carry the intended download
policy so Capsules can replicate with on_demand.

Assisted-By: Cursor Grok 4.6
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: pablomh

CHANGES/+remote-policy.feature

@@ -0,0 +1 @@
+Added `UpstreamPulp.remote_policy` so remotes created during replication can use `on_demand` or `streamed` instead of defaulting to `immediate`.

docs/user/guides/replication.md

@@ -48,6 +48,7 @@ pulp upstream-pulp create \
 | `tls_validation` | Whether to verify the upstream server's TLS certificate. Defaults to `True`. |
 | `q_select` | A filter expression to select which upstream distributions to replicate. See [Filtering Distributions](#filtering-distributions-with-q_select). |
 | `policy` | Controls how replication manages local objects. One of `all`, `labeled`, or `nodelete`. See [Replication Policies](#replication-policies). Defaults to `all`. |
+| `remote_policy` | Download policy for remotes created during replication. One of `immediate`, `on_demand`, or `streamed`. Distinct from `policy`. When unset, remotes use Pulp's default (`immediate`). |
 
 ## Running Replication
 
@@ -151,7 +152,8 @@ pulp upstream-pulp replicate --upstream-pulp "my-upstream"
 ## Replication Policies
 
 The `policy` field controls how replication handles local objects, particularly when upstream
-distributions are removed or no longer match a `q_select` filter.
+distributions are removed or no longer match a `q_select` filter. It is not the same as a remote's
+download policy (`immediate`, `on_demand`, or `streamed`); set that with `remote_policy`.
 
 ### `all` (default)
 

pulpcore/app/migrations/0157_upstreampulp_remote_policy.py

@@ -0,0 +1,34 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0156_alter_contentartifact_relative_path_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="upstreampulp",
+            name="remote_policy",
+            field=models.TextField(
+                choices=[
+                    ("immediate", "When syncing, download all metadata and content now."),
+                    (
+                        "on_demand",
+                        "When syncing, download metadata, but do not download content now. "
+                        "Instead, download content as clients request it, and save it in Pulp "
+                        "to be served for future client requests.",
+                    ),
+                    (
+                        "streamed",
+                        "When syncing, download metadata, but do not download content now. "
+                        "Instead,download content as clients request it, but never save it in "
+                        "Pulp. This causes future requests for that same content to have to be "
+                        "downloaded again.",
+                    ),
+                ],
+                null=True,
+            ),
+        ),
+    ]

pulpcore/app/models/replica.py

@@ -11,6 +11,8 @@
 from pulpcore.app.util import get_domain_pk
 from pulpcore.plugin.models import AutoAddObjPermsMixin, BaseModel, EncryptedTextField
 
+from .repository import Remote
+
 
 class UpstreamPulp(BaseModel, AutoAddObjPermsMixin):
     ALL = "all"
@@ -59,6 +61,7 @@ class UpstreamPulp(BaseModel, AutoAddObjPermsMixin):
     sock_read_timeout = models.FloatField(
         null=True, validators=[MinValueValidator(0.0, "Timeout must be >= 0")]
     )
+    remote_policy = models.TextField(choices=Remote.POLICY_CHOICES, null=True)
 
     q_select = models.TextField(null=True)
     policy = models.TextField(choices=POLICY_CHOICES, default=ALL)

pulpcore/app/serializers/replica.py

@@ -3,7 +3,7 @@
 from rest_framework import serializers
 from rest_framework.validators import UniqueValidator
 
-from pulpcore.app.models import UpstreamPulp
+from pulpcore.app.models import Remote, UpstreamPulp
 from pulpcore.app.serializers import (
     HiddenFieldsMixin,
     IdentityField,
@@ -122,6 +122,16 @@ class UpstreamPulpSerializer(ModelSerializer, HiddenFieldsMixin):
         ),
         min_value=0.0,
     )
+    remote_policy = serializers.ChoiceField(
+        choices=Remote.POLICY_CHOICES,
+        help_text=_(
+            "Download policy for remotes created during replication. One of 'immediate', "
+            "'on_demand', or 'streamed'. Distinct from 'policy', which controls how replicate "
+            "manages local objects. Defaults to the Remote default ('immediate') when unset."
+        ),
+        required=False,
+        allow_null=True,
+    )
 
     pulp_last_updated = serializers.DateTimeField(
         help_text="Timestamp of the most recent update of the remote.", read_only=True
@@ -178,6 +188,7 @@ class Meta:
             "connect_timeout",
             "sock_connect_timeout",
             "sock_read_timeout",
+            "remote_policy",
             "pulp_last_updated",
             "hidden_fields",
             "q_select",

pulpcore/app/tasks/replica.py

@@ -28,6 +28,26 @@ def user_agent():
     return f"pulpcore/{pulp_version} ({python}, {system}) (pulp-glue {pulp_glue_version})"
 
 
+def _build_remote_settings(server):
+    """Build fields copied onto remotes created during replication."""
+    remote_settings = {
+        "ca_cert": server.ca_cert,
+        "tls_validation": server.tls_validation,
+        "client_cert": server.client_cert,
+        "client_key": server.client_key,
+        "download_concurrency": server.download_concurrency,
+        "max_retries": server.max_retries,
+        "total_timeout": server.total_timeout,
+        "connect_timeout": server.connect_timeout,
+        "sock_connect_timeout": server.sock_connect_timeout,
+        "sock_read_timeout": server.sock_read_timeout,
+    }
+    # Omit policy when unset so new remotes keep Remote.policy's default (immediate).
+    if server.remote_policy is not None:
+        remote_settings["policy"] = server.remote_policy
+    return remote_settings
+
+
 def replicate_distributions(server_pk, q_select=None, **kwargs):
     server = UpstreamPulp.objects.get(pk=server_pk)
 
@@ -58,18 +78,7 @@ def replicate_distributions(server_pk, q_select=None, **kwargs):
         }
     )
 
-    remote_settings = {
-        "ca_cert": server.ca_cert,
-        "tls_validation": server.tls_validation,
-        "client_cert": server.client_cert,
-        "client_key": server.client_key,
-        "download_concurrency": server.download_concurrency,
-        "max_retries": server.max_retries,
-        "total_timeout": server.total_timeout,
-        "connect_timeout": server.connect_timeout,
-        "sock_connect_timeout": server.sock_connect_timeout,
-        "sock_read_timeout": server.sock_read_timeout,
-    }
+    remote_settings = _build_remote_settings(server)
 
     try:
         task_group = TaskGroup.current()

pulpcore/tests/functional/api/test_replication.py

@@ -281,6 +281,80 @@ def test_replication_remote_settings_propagation(
     assert remote.max_retries == 2
 
 
+@pytest.mark.parallel
+def test_replication_remote_policy(
+    domain_factory,
+    bindings_cfg,
+    pulpcore_bindings,
+    file_bindings,
+    monitor_task,
+    monitor_task_group,
+    pulp_settings,
+    gen_object_with_cleanup,
+    file_distribution_factory,
+    file_publication_factory,
+    file_repository_factory,
+    tmp_path,
+    add_domain_objects_to_cleanup,
+):
+    """Remotes created by replicate() inherit UpstreamPulp.remote_policy when set."""
+    source_domain = domain_factory()
+    add_domain_objects_to_cleanup(source_domain)
+
+    repository = file_repository_factory(pulp_domain=source_domain.name)
+    file_path = tmp_path / "file.txt"
+    file_path.write_text("DEADBEEF")
+    monitor_task(
+        file_bindings.ContentFilesApi.create(
+            file=str(file_path),
+            relative_path="file.txt",
+            repository=repository.pulp_href,
+            pulp_domain=source_domain.name,
+        ).task
+    )
+    publication = file_publication_factory(
+        pulp_domain=source_domain.name, repository=repository.pulp_href
+    )
+    file_distribution_factory(pulp_domain=source_domain.name, publication=publication.pulp_href)
+
+    replica_domain = domain_factory()
+    add_domain_objects_to_cleanup(replica_domain)
+
+    upstream_pulp_body = {
+        "name": str(uuid.uuid4()),
+        "base_url": bindings_cfg.host,
+        "api_root": pulp_settings.API_ROOT,
+        "domain": source_domain.name,
+        "username": bindings_cfg.username,
+        "password": bindings_cfg.password,
+        "remote_policy": "on_demand",
+    }
+    upstream_pulp = gen_object_with_cleanup(
+        pulpcore_bindings.UpstreamPulpsApi, upstream_pulp_body, pulp_domain=replica_domain.name
+    )
+
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(
+        upstream_pulp.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate()
+    )
+    monitor_task_group(response.task_group)
+
+    result = file_bindings.RemotesFileApi.list(pulp_domain=replica_domain.name)
+    assert result.count == 1
+    remote = result.results[0]
+    assert remote.policy == "on_demand"
+
+    pulpcore_bindings.UpstreamPulpsApi.partial_update(
+        upstream_pulp.pulp_href, {"remote_policy": "streamed"}
+    )
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(
+        upstream_pulp.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate()
+    )
+    monitor_task_group(response.task_group)
+
+    remote = file_bindings.RemotesFileApi.list(pulp_domain=replica_domain.name).results[0]
+    assert remote.policy == "streamed"
+
+
 @pytest.mark.parallel
 def test_replication_with_repo_based_distribution(
     domain_factory,

pulpcore/tests/unit/test_replica_remote_policy.py

@@ -0,0 +1,35 @@
+from types import SimpleNamespace
+
+from pulpcore.app.tasks.replica import _build_remote_settings
+
+
+def _server(**overrides):
+    base = {
+        "ca_cert": "api-ca",
+        "tls_validation": True,
+        "client_cert": "api-cert",
+        "client_key": "api-key",
+        "download_concurrency": 10,
+        "max_retries": 3,
+        "total_timeout": 30,
+        "connect_timeout": 5,
+        "sock_connect_timeout": 5,
+        "sock_read_timeout": 5,
+        "remote_policy": None,
+    }
+    base.update(overrides)
+    return SimpleNamespace(**base)
+
+
+def test_build_remote_settings_omits_policy_when_unset():
+    settings = _build_remote_settings(_server())
+
+    assert "policy" not in settings
+    assert settings["ca_cert"] == "api-ca"
+    assert settings["download_concurrency"] == 10
+
+
+def test_build_remote_settings_includes_policy_when_set():
+    settings = _build_remote_settings(_server(remote_policy="on_demand"))
+
+    assert settings["policy"] == "on_demand"