Skip to content

Commit a1caee3

Browse files
pablomhcursoragentclaude
committed
Fixes #39211 - Defer capsule distribution refreshes after sync.
Refresh capsule distributions after sync batches complete and recover once from uniqueness races, while keeping the test coverage focused on the deferred refresh flow. Co-authored-by: Cursor <cursoragent@cursor.com> Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent cd3642b commit a1caee3

7 files changed

Lines changed: 229 additions & 44 deletions

File tree

app/lib/actions/katello/capsule_content/sync_capsule.rb

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,14 @@ def plan(smart_proxy, options = {})
4949
end
5050
end
5151
end
52+
53+
repos.select { |repo| smart_proxy.pulp3_support?(repo) }.in_groups_of(Setting[:foreman_proxy_content_batch_size], false) do |repo_batch|
54+
concurrence do
55+
repo_batch.each do |repo|
56+
plan_action(Actions::Pulp3::CapsuleContent::RefreshDistribution, repo, smart_proxy)
57+
end
58+
end
59+
end
5260
end
5361
end
5462
# rubocop:enable Metrics/MethodLength

app/lib/actions/pulp3/capsule_content/generate_metadata.rb

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,9 @@ class GenerateMetadata < Pulp3::AbstractAsyncTask
55
middleware.use Actions::Middleware::ExecuteIfContentsChanged
66
def plan(repository, smart_proxy, options = {})
77
options[:contents_changed] = (options && options.key?(:contents_changed)) ? options[:contents_changed] : true
8-
sequence do
9-
unless repository.repository_type.pulp3_skip_publication
10-
plan_self(:repository_id => repository.id, :smart_proxy_id => smart_proxy.id,
11-
:options => options).output
12-
end
13-
plan_action(RefreshDistribution, repository, smart_proxy,
14-
:contents_changed => options[:contents_changed])
8+
unless repository.repository_type.pulp3_skip_publication
9+
plan_self(:repository_id => repository.id, :smart_proxy_id => smart_proxy.id,
10+
:options => options).output
1511
end
1612
end
1713

app/lib/actions/pulp3/capsule_content/refresh_distribution.rb

Lines changed: 35 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,20 +2,47 @@ module Actions
22
module Pulp3
33
module CapsuleContent
44
class RefreshDistribution < Pulp3::AbstractAsyncTask
5-
include Helpers::Presenter
6-
middleware.use Actions::Middleware::ExecuteIfContentsChanged
7-
8-
def plan(repository, smart_proxy, options = {})
5+
def plan(repository, smart_proxy)
96
plan_self(:repository_id => repository.id,
10-
:smart_proxy_id => smart_proxy.id,
11-
:options => options)
7+
:smart_proxy_id => smart_proxy.id)
128
end
139

1410
def invoke_external_task
15-
smart_proxy = ::SmartProxy.unscoped.find(input[:smart_proxy_id])
16-
repo = ::Katello::Repository.find(input[:repository_id])
1711
repo.backend_service(smart_proxy).with_mirror_adapter.refresh_distributions
1812
end
13+
14+
def rescue_external_task(error)
15+
if distribution_uniqueness_conflict?(error) && !retried_distribution_refresh?
16+
# A concurrent RefreshDistribution created this distribution first.
17+
# Re-invoke so refresh_distributions finds the existing distribution
18+
# and issues a partial_update instead. Dynflow will poll the new task.
19+
# Only retry once; if the follow-up refresh still conflicts, fall back
20+
# to the parent error handling rather than replaying indefinitely.
21+
output[:retried_distribution_refresh] = true
22+
self.external_task = invoke_external_task
23+
else
24+
super
25+
end
26+
end
27+
28+
private
29+
30+
def repo
31+
@repo ||= ::Katello::Repository.find(input[:repository_id])
32+
end
33+
34+
def smart_proxy
35+
@smart_proxy ||= super
36+
end
37+
38+
def distribution_uniqueness_conflict?(error)
39+
error.is_a?(::Katello::Errors::Pulp3Error) &&
40+
::Katello::Pulp3::DistributionConflict.create_race?(error)
41+
end
42+
43+
def retried_distribution_refresh?
44+
output[:retried_distribution_refresh]
45+
end
1946
end
2047
end
2148
end
Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
module Katello
2+
module Pulp3
3+
module DistributionConflict
4+
CREATE_RACE_PATTERN = /
5+
["']base_path["'].*?
6+
(
7+
must\ be\ unique | code=['"]unique |
8+
Overlaps\ with\ existing\ distribution
9+
)
10+
/mx
11+
12+
def self.create_race?(error_or_message)
13+
message = error_or_message.respond_to?(:message) ? error_or_message.message : error_or_message.to_s
14+
message.match?(CREATE_RACE_PATTERN)
15+
end
16+
end
17+
end
18+
end

app/services/katello/pulp3/repository.rb

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -240,7 +240,7 @@ def sync(options = {})
240240

241241
def sync_url_params(_sync_options)
242242
params = {remote: repo.remote_href, mirror: repo.root.mirroring_policy == Katello::RootRepository::MIRRORING_POLICY_CONTENT}
243-
params[:skip_types] = skip_types if (skip_types && repo.root.mirroring_policy != Katello::RootRepository::MIRRORING_POLICY_COMPLETE)
243+
params[:skip_types] = skip_types if skip_types && repo.root.mirroring_policy != Katello::RootRepository::MIRRORING_POLICY_COMPLETE
244244
params
245245
end
246246

@@ -291,8 +291,7 @@ def refresh_distributions
291291
create_distribution(relative_path)
292292
rescue api.client_module::ApiError => e
293293
# Now it seems there is a distribution. Fetch it and save the reference.
294-
if e.message.include?("\"base_path\":[\"This field must be unique.\"]") ||
295-
e.message.include?("\"base_path\":[\"Overlaps with existing distribution\"")
294+
if ::Katello::Pulp3::DistributionConflict.create_race?(e)
296295
dist = lookup_distributions(base_path: repo.relative_path).first
297296
save_distribution_references([dist.pulp_href])
298297
return update_distribution
@@ -551,7 +550,7 @@ def custom_cdn_ca_cert
551550
end
552551

553552
def append_proxy_cacert(options)
554-
if root.http_proxy&.cacert&.present? && options.key?(:ca_cert)
553+
if root.http_proxy&.cacert.present? && options.key?(:ca_cert)
555554
options[:ca_cert] = [options[:ca_cert], root.http_proxy.cacert].compact.join("\n")
556555
end
557556
options

test/actions/katello/capsule_content_test.rb

Lines changed: 14 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -64,10 +64,7 @@ class SyncTest < TestBase
6464
assert_equal repo.id, input[:repository_id]
6565
end
6666

67-
assert_tree_planned_with(tree, Actions::Pulp3::CapsuleContent::RefreshDistribution) do |input|
68-
assert_equal capsule_content.smart_proxy.id, input[:smart_proxy_id]
69-
assert_equal repo.id, input[:repository_id]
70-
end
67+
assert_refresh_boundary(tree, [repo.id])
7168
end
7269

7370
it 'plans correctly for a pulp3 yum repo without the proper plugin' do
@@ -104,10 +101,7 @@ class SyncTest < TestBase
104101
assert_equal repo.id, input[:repository_id]
105102
end
106103

107-
assert_tree_planned_with(tree, Actions::Pulp3::CapsuleContent::RefreshDistribution) do |input|
108-
assert_equal capsule_content.smart_proxy.id, input[:smart_proxy_id]
109-
assert_equal repo.id, input[:repository_id]
110-
end
104+
assert_refresh_boundary(tree, [repo.id])
111105
end
112106

113107
it 'plans correctly for a pulp3 docker repo' do
@@ -121,10 +115,7 @@ class SyncTest < TestBase
121115
assert_equal repo.id, input[:repository_id]
122116
end
123117

124-
assert_tree_planned_with(tree, Actions::Pulp3::CapsuleContent::RefreshDistribution) do |input|
125-
assert_equal capsule_content.smart_proxy.id, input[:smart_proxy_id]
126-
assert_equal repo.id, input[:repository_id]
127-
end
118+
assert_refresh_boundary(tree, [repo.id])
128119
end
129120

130121
it 'plans correctly for a pulp3 ansible collection repo' do
@@ -134,10 +125,7 @@ class SyncTest < TestBase
134125
repo = katello_repositories(:pulp3_ansible_collection_1)
135126
tree = plan_action_tree(action_class, capsule_content.smart_proxy, :repository_id => repo.id)
136127
assert_tree_planned_steps(tree, ::Actions::Pulp3::ContentGuard::Refresh)
137-
assert_tree_planned_with(tree, Actions::Pulp3::CapsuleContent::RefreshDistribution) do |input|
138-
assert_equal capsule_content.smart_proxy.id, input[:smart_proxy_id]
139-
assert_equal repo.id, input[:repository_id]
140-
end
128+
assert_refresh_boundary(tree, [repo.id])
141129
end
142130

143131
it 'plans correctly for a pulp3 apt repo' do
@@ -164,10 +152,7 @@ class SyncTest < TestBase
164152
assert_equal repo.id, input[:repository_id]
165153
end
166154

167-
assert_tree_planned_with(tree, Actions::Pulp3::CapsuleContent::RefreshDistribution) do |input|
168-
assert_equal capsule_content.smart_proxy.id, input[:smart_proxy_id]
169-
assert_equal repo.id, input[:repository_id]
170-
end
155+
assert_refresh_boundary(tree, [repo.id])
171156
end
172157

173158
it 'plans correctly for a pulp2 apt repo' do
@@ -183,6 +168,7 @@ class SyncTest < TestBase
183168
}
184169

185170
assert_tree_planned_with(tree, ::Actions::Pulp3::Orchestration::Repository::RefreshRepos, options)
171+
refute_tree_planned(tree, ::Actions::Pulp3::CapsuleContent::RefreshDistribution)
186172
end
187173

188174
it 'plans correctly for a pulp yum repo' do
@@ -220,6 +206,7 @@ class SyncTest < TestBase
220206
with_pulp3_features(capsule_content.smart_proxy)
221207
capsule_content.smart_proxy.add_lifecycle_environment(dev_environment)
222208
repos_in_dev = Katello::Repository.in_environment(dev_environment).pluck(:pulp_id)
209+
repo_ids_in_dev = Katello::Repository.in_environment(dev_environment).pluck(:id)
223210

224211
tree = plan_action_tree(action_class, capsule_content.smart_proxy, :environment_id => dev_environment.id)
225212
options = { smart_proxy_id: capsule_content.smart_proxy.id,
@@ -242,14 +229,16 @@ class SyncTest < TestBase
242229
assert_includes repos_in_dev, repo.pulp_id
243230
end
244231

232+
assert_refresh_boundary(tree, repo_ids_in_dev)
233+
end
234+
235+
def assert_refresh_boundary(tree, expected_repo_ids)
236+
planned_repo_ids = []
245237
assert_tree_planned_with(tree, Actions::Pulp3::CapsuleContent::RefreshDistribution) do |input|
246238
assert_equal capsule_content.smart_proxy.id, input[:smart_proxy_id]
247-
repo = Katello::Repository.find(input[:repository_id])
248-
assert_includes repos_in_dev, repo.pulp_id
249-
repos_in_dev.delete(repo.pulp_id)
239+
planned_repo_ids << input[:repository_id]
250240
end
251-
252-
assert_empty repos_in_dev
241+
assert_equal expected_repo_ids.uniq.sort, planned_repo_ids.sort
253242
end
254243

255244
it 'fails when trying to sync to the default capsule' do
Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,148 @@
1+
require 'katello_test_helper'
2+
3+
module ::Actions::Pulp3::CapsuleContent
4+
class DistributionConflictTest < ActiveSupport::TestCase
5+
it 'matches both async task and direct api race messages' do
6+
async_error = ::Katello::Errors::Pulp3Error.new(
7+
"{'base_path': [ErrorDetail(string='This field must be unique.', code='unique')]}"
8+
)
9+
api_error = RuntimeError.new('{"base_path":["Overlaps with existing distribution"]}')
10+
11+
assert ::Katello::Pulp3::DistributionConflict.create_race?(async_error)
12+
assert ::Katello::Pulp3::DistributionConflict.create_race?(api_error)
13+
refute ::Katello::Pulp3::DistributionConflict.create_race?("Remote artifacts cannot be exported")
14+
end
15+
end
16+
17+
class GenerateMetadataTest < ActiveSupport::TestCase
18+
include Dynflow::Testing
19+
include Support::Actions::Fixtures
20+
include Support::CapsuleSupport
21+
include Support::Actions::RemoteAction
22+
23+
let(:proxy) { capsule_content.smart_proxy }
24+
25+
before do
26+
set_user
27+
SmartProxy.any_instance.stubs(:ping_pulp).returns({})
28+
SmartProxy.any_instance.stubs(:ping_pulp3).returns({})
29+
SmartProxy.any_instance.stubs(:pulp3_configuration).returns(nil)
30+
end
31+
32+
# pulp3_skip_publication: true - plan_self is skipped, tree is empty
33+
it 'does not plan RefreshDistribution inline for publication-less repos (e.g. docker)' do
34+
repo = katello_repositories(:pulp3_docker_1)
35+
tree = plan_action_tree(::Actions::Pulp3::CapsuleContent::GenerateMetadata,
36+
repo, proxy)
37+
38+
refute_tree_planned(tree, ::Actions::Pulp3::CapsuleContent::RefreshDistribution)
39+
end
40+
41+
# pulp3_skip_publication: false - plan_self IS called to create a publication
42+
it 'does not plan RefreshDistribution inline for publication-based repos (e.g. file)' do
43+
repo = katello_repositories(:pulp3_file_1)
44+
tree = plan_action_tree(::Actions::Pulp3::CapsuleContent::GenerateMetadata,
45+
repo, proxy)
46+
47+
refute_tree_planned(tree, ::Actions::Pulp3::CapsuleContent::RefreshDistribution)
48+
end
49+
end
50+
51+
class RefreshDistributionTest < ActiveSupport::TestCase
52+
include Dynflow::Testing
53+
include Support::Actions::Fixtures
54+
include Support::CapsuleSupport
55+
include Support::Actions::RemoteAction
56+
57+
let(:proxy) { capsule_content.smart_proxy }
58+
let(:repo) { katello_repositories(:pulp3_docker_1) }
59+
60+
before do
61+
set_user
62+
SmartProxy.any_instance.stubs(:ping_pulp).returns({})
63+
SmartProxy.any_instance.stubs(:ping_pulp3).returns({})
64+
SmartProxy.any_instance.stubs(:pulp3_configuration).returns(nil)
65+
end
66+
67+
def build_action_with_output
68+
action = create_action(::Actions::Pulp3::CapsuleContent::RefreshDistribution)
69+
action.stubs(:input).returns('repository_id' => repo.id, 'smart_proxy_id' => proxy.id)
70+
action_output = {}
71+
action.stubs(:output).returns(action_output)
72+
[action, action_output]
73+
end
74+
75+
it 'recovers from a concurrent distribution creation by retrying as an update' do
76+
action, action_output = build_action_with_output
77+
78+
mock_task = mock('pulp_task')
79+
action.expects(:invoke_external_task).returns(mock_task)
80+
action.expects(:external_task=).with(mock_task)
81+
82+
error = ::Katello::Errors::Pulp3Error.new(
83+
"{'base_path': [ErrorDetail(string='This field must be unique.', code='unique')]}"
84+
)
85+
action.rescue_external_task(error)
86+
assert action_output[:retried_distribution_refresh]
87+
end
88+
89+
it 'recovers when both name and base_path are reported as non-unique (real Pulp error format)' do
90+
action, action_output = build_action_with_output
91+
92+
mock_task = mock('pulp_task')
93+
action.expects(:invoke_external_task).returns(mock_task)
94+
action.expects(:external_task=).with(mock_task)
95+
96+
error = ::Katello::Errors::Pulp3Error.new(
97+
"{'name': [ErrorDetail(string='This field must be unique.', code='unique')], " \
98+
"'base_path': [ErrorDetail(string='This field must be unique.', code='unique')]}"
99+
)
100+
action.rescue_external_task(error)
101+
assert action_output[:retried_distribution_refresh]
102+
end
103+
104+
# Non-Pulp3Errors fall through to the parent's rescue_external_task,
105+
# which delegates to Dynflow's default handler. We verify here that
106+
# RefreshDistribution does not intercept them for its own retry logic.
107+
it 'does not attempt recovery for non-Pulp3Errors' do
108+
action, _action_output = build_action_with_output
109+
action.expects(:invoke_external_task).never
110+
111+
action.rescue_external_task(RuntimeError.new("connection refused"))
112+
end
113+
114+
it 'recovers from an overlap conflict (base_path overlaps existing distribution)' do
115+
action, action_output = build_action_with_output
116+
117+
mock_task = mock('pulp_task')
118+
action.expects(:invoke_external_task).returns(mock_task)
119+
action.expects(:external_task=).with(mock_task)
120+
121+
error = ::Katello::Errors::Pulp3Error.new(
122+
"{'base_path': ['Overlaps with existing distribution']}"
123+
)
124+
action.rescue_external_task(error)
125+
assert action_output[:retried_distribution_refresh]
126+
end
127+
128+
it 're-raises the conflict after the one allowed retry is exhausted' do
129+
action, action_output = build_action_with_output
130+
action_output[:retried_distribution_refresh] = true
131+
action.expects(:invoke_external_task).never
132+
133+
error = ::Katello::Errors::Pulp3Error.new(
134+
"{'base_path': [ErrorDetail(string='This field must be unique.', code='unique')]}"
135+
)
136+
137+
assert_raises(::Katello::Errors::Pulp3Error) { action.rescue_external_task(error) }
138+
end
139+
140+
it 're-raises Pulp3Errors unrelated to distribution uniqueness' do
141+
action, _action_output = build_action_with_output
142+
action.expects(:invoke_external_task).never
143+
144+
error = ::Katello::Errors::Pulp3Error.new("Remote artifacts cannot be exported")
145+
assert_raises(::Katello::Errors::Pulp3Error) { action.rescue_external_task(error) }
146+
end
147+
end
148+
end

0 commit comments

Comments
 (0)