Fix a cache invalidation issue with Distributions serving publications indirectly via repository_version

[dralley]

Make sure we flush the cache (and update DistributedPublication) for
publications which were indirectly served via
Distribution.repository_version, rather than directly.

closes #7993

[dralley]

Not quite ready for review yet.

[dralley]

Should this whole block be in a transaction?

[gerrod3]

The new cache invalidation scenario looks correct. I don't know enough about DistributedPublications to comment on the code changes there.

[gerrod3]

Do the cache invalidation tests belong here? Should they be in their own file?

[dralley]

It would be reasonable to separate them but they're also somewhat coupled in the sense that a significant amount of logic is basically shared between the two and the scenarios are basically the same. If you feel strongly we can do that.

[pedro-psb]

I did an AI assisted review and it indicated a possible flaw where the reactivation path can lead to two active dps. Can you assert in this test that, after switching back to pub1, we only have one active dp for that distribution?

[pedro-psb]

I don't think this is helping with the race condition. If I understand correctly, the goal is ensure the invariant of only 1 active DP.

But the select for update locks a row only if it exists.
So let's say there is no active DP. It can happen the following in a scenario with 2 concurrent writers:

1. $W_1$ and $W_2$ get past the fast check and see no active DP
2. $W_1$ starts transcation. Evaluates dp=None (no row to lock on)
3. $W_2$ starts transcation. Evaluates dp=None (no row to lock on)
4. $W_1$ creates $DP_1$. Creation hook sees no other active DP to disable and activates itself
5. $W_2$ creates $DP_2$. Creation hook sees no other active DP to disable ($W_1$ transcation not done yet) and activates itself
6. $W_1$ ends transaction. $DP_1$ commited
7. $W_2$ ends transaction. $DP_2$ commited

[pedro-psb]

A sample test:

@pytest.mark.django_db(transaction=True)
def test_concurrent_set_distributed_publication_leaves_single_winner():
    """Racing ."""

    def race():
        barrier.wait()
        dist.set_distributed_publication()

    for i in range(5):
        pub = pub_factory()
        dist = dist_factory(pub=pub)
        DistributedPublication.objects.filter(distribution=dist).delete()
        barrier = threading.Barrier(2)

        threads = [threading.Thread(target=race) for _ in range(2)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        dps = DistributedPublication.objects.filter(distribution=dist, publication=pub)
        dps_count = dps.filter(expires_at__isnull=True).count()
        assert dps_count == 1, (
            f"[{i=}] exactly one DP must remain active even if both threads inserted a row"
        )

[pedro-psb]

This was probably not introduced with this changes, so if you just want to not handle it right now (maybe simplify this portion) and focus on the issue about case of serving repository_versions indirectly.

[dralley]

I've pushed a new commit based on your suggestions. Claude reasoning:

Is the feedback valid?

Yes — pedro-psb's two points are both real bugs, and they're related:

1. Concurrent-create race (line 845): select_for_update() on a .filter().first() that returns None locks nothing — you can't lock a row that doesn't exist yet. Two writers could both pass the fast path and both INSERT an active DP. His interleaving walkthrough is correct.

2. Reactivation → two active DPs (line 240): the reactivation branch used save(skip_hooks=True), so the cleanup AFTER_CREATE hook never ran. Switching back to pub1 reactivated pub1's DP but never expired the still-active pub2 DP → two active DPs for one distribution.

The other comments were already resolved earlier (transaction wrapper, keeping cache tests co-located — dralley said that's fine).

What I changed

1. publication.py — set_distributed_publication:

• Lock the Distribution row (Distribution.objects.select_for_update().get(pk=self.pk)) instead of the DP row, so concurrent writers serialize even when no DP exists yet. The second writer re-queries inside the lock and finds the committed active DP.
• On reactivation, explicitly expire the other active DP(s) for the distribution (since the AFTER_CREATE hook doesn't fire on an update), restoring the single-active invariant.

2. test_publication_retention.py:

• Strengthened the reactivation test to assert exactly one active DP remains after switching back (pedro's line-240 request) — this deterministically catches bug Adopts new content_summary layout #2.

On the concurrency test pedro sketched

I implemented it, and it passes (proving the fix), but @pytest.mark.django_db(transaction=True) errors on teardown in pulpcore's unit suite — I confirmed even a trivial transactional test fails the same way. The _populate_access_policies post_migrate receiver requires apps positionally and the transactional teardown flush doesn't supply it; that's why the suite has zero transaction=True tests. So I removed the threaded test rather than ship a CI-reddening teardown error. Enabling transactional unit tests would be a separate infra fix (making that receiver's apps/verbosity args defaulted).