From cbbec9d5d4ba48d80a767916e1f3ff356783ee66 Mon Sep 17 00:00:00 2001 From: Bolek <1416262+bolekk@users.noreply.github.com> Date: Tue, 22 Sep 2026 08:26:45 +0000 Subject: [PATCH] fix(cresettings): enforce single active job per config type (#23789) The single-instance check was broken in two ways: it was global, so a shard_assignment job could not run alongside a settings job, and the activeJobID pointer was never stored, so the 'already active' error could never actually fire. Track the active job ID per config type (settings / shard_assignment) in a sync.Map. ServicesForSpec validates the config type and claims the slot atomically via LoadOrStore, so one settings job and one shard_assignment job may coexist while a second job of the same type is rejected. OnDeleteJob frees the slot only if it is held by the deleted job's ID, so deleting a rejected duplicate cannot evict the active job. --- core/services/cresettings/delegate.go | 44 ++++---- core/services/cresettings/delegate_test.go | 118 +++++++++++++++++++++ 2 files changed, 144 insertions(+), 18 deletions(-) create mode 100644 core/services/cresettings/delegate_test.go diff --git a/core/services/cresettings/delegate.go b/core/services/cresettings/delegate.go index f9f7bb35827..eb2b2db2412 100644 --- a/core/services/cresettings/delegate.go +++ b/core/services/cresettings/delegate.go @@ -1,12 +1,12 @@ // cresettings jobs are used to distribute updates for CRE settings overrides. // See: https://pkg.go.dev/github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings -// Only one Job of type CRESettings may run at a time. Attempts to create a second job will fail. +// Only one Job of type CRESettings may run at a time per config type. Attempts to create a second job of the same config type will fail. package cresettings import ( "context" "fmt" - "sync/atomic" + "sync" "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/loop" @@ -29,7 +29,8 @@ type delegate struct { atomicSettings *loop.AtomicSettings shardAssignmentSettings *loop.AtomicSettings - activeJobID atomic.Pointer[int32] + // activeJobIDs maps config type to the active job ID. + activeJobIDs sync.Map } func (d *delegate) JobType() job.Type { @@ -39,18 +40,16 @@ func (d *delegate) JobType() job.Type { func (d *delegate) BeforeJobCreated(j job.Job) {} func (d *delegate) ServicesForSpec(ctx context.Context, j job.Job) ([]job.ServiceCtx, error) { - if activeJobID := d.activeJobID.Load(); activeJobID != nil { - return nil, fmt.Errorf("another %s job is already active: %d", job.CRESettings, activeJobID) + spec := j.CRESettingsSpec + configType := d.configType(spec) + switch configType { + case ConfigTypeSettings, ConfigTypeShardAssignment: + default: + return nil, fmt.Errorf("unknown config_type %q", configType) } - spec := j.CRESettingsSpec - configType := ConfigTypeSettings - if spec.Settings != "" { - if ct, ok := extractConfigType(spec.Settings); ok { - configType = ct - } else { - d.lggr.Infow("No config_type specified, defaulting to settings") - } + if activeJobID, loaded := d.activeJobIDs.LoadOrStore(configType, j.ID); loaded { + return nil, fmt.Errorf("another %s job with config_type %q is already active: %d", job.CRESettings, configType, activeJobID.(int32)) } switch configType { @@ -71,9 +70,6 @@ func (d *delegate) ServicesForSpec(ctx context.Context, j job.Job) ([]job.Servic return nil, fmt.Errorf("failed to update settings: %w", err) } d.lggr.Infow("Updated settings", "hash", spec.Hash, "settings", spec.Settings) - - default: - return nil, fmt.Errorf("unknown config_type %q", configType) } return nil, nil @@ -84,8 +80,20 @@ func (d *delegate) AfterJobCreated(j job.Job) {} func (d *delegate) BeforeJobDeleted(j job.Job) {} func (d *delegate) OnDeleteJob(ctx context.Context, jb job.Job) error { - if !d.activeJobID.CompareAndSwap(&jb.ID, nil) { - d.lggr.Errorf("job %d was not active", jb.ID) + configType := d.configType(jb.CRESettingsSpec) + if !d.activeJobIDs.CompareAndDelete(configType, jb.ID) { + d.lggr.Errorf("job %d was not the active %s job for config_type %q", jb.ID, job.CRESettings, configType) } return nil } + +func (d *delegate) configType(spec *job.CRESettingsSpec) string { + if spec == nil || spec.Settings == "" { + return ConfigTypeSettings + } + if ct, ok := extractConfigType(spec.Settings); ok { + return ct + } + d.lggr.Infow("No config_type specified, defaulting to settings") + return ConfigTypeSettings +} diff --git a/core/services/cresettings/delegate_test.go b/core/services/cresettings/delegate_test.go new file mode 100644 index 00000000000..9027722a254 --- /dev/null +++ b/core/services/cresettings/delegate_test.go @@ -0,0 +1,118 @@ +package cresettings + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/pkg/loop" + "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink/v2/core/services/job" +) + +const ( + settingsToml = `Foo = "bar" +` + shardAssignmentToml = `config_type = "shard_assignment" +static_default_assignment = [0, 1] +` +) + +func newTestDelegate(t *testing.T) *delegate { + t.Helper() + return NewDelegate(logger.TestLogger(t), &loop.AtomicSettings{}, &loop.AtomicSettings{}) +} + +func cresettingsJob(id int32, settings string) job.Job { + return job.Job{ + ID: id, + Type: job.CRESettings, + CRESettingsSpec: &job.CRESettingsSpec{Settings: settings}, + } +} + +func TestServicesForSpecOneJobPerConfigType(t *testing.T) { + t.Parallel() + + ctx := context.Background() + d := newTestDelegate(t) + + _, err := d.ServicesForSpec(ctx, cresettingsJob(1, settingsToml)) + require.NoError(t, err) + + _, err = d.ServicesForSpec(ctx, cresettingsJob(2, shardAssignmentToml)) + require.NoError(t, err) + + _, err = d.ServicesForSpec(ctx, cresettingsJob(3, settingsToml)) + require.ErrorContains(t, err, fmt.Sprintf("another %s job with config_type %q is already active: 1", job.CRESettings, ConfigTypeSettings)) + + _, err = d.ServicesForSpec(ctx, cresettingsJob(4, shardAssignmentToml)) + require.ErrorContains(t, err, fmt.Sprintf("another %s job with config_type %q is already active: 2", job.CRESettings, ConfigTypeShardAssignment)) +} + +func TestServicesForSpecDefaultConfigTypeSharesSettingsSlot(t *testing.T) { + t.Parallel() + + ctx := context.Background() + d := newTestDelegate(t) + + _, err := d.ServicesForSpec(ctx, cresettingsJob(1, shardAssignmentToml)) + require.NoError(t, err) + + _, err = d.ServicesForSpec(ctx, cresettingsJob(2, settingsToml)) + require.NoError(t, err) + + // spec with no config_type defaults to settings, so it collides with job 2 + _, err = d.ServicesForSpec(ctx, cresettingsJob(3, `Foo = "baz" +`)) + require.ErrorContains(t, err, "already active: 2") +} + +func TestServicesForSpecUnknownConfigType(t *testing.T) { + t.Parallel() + + ctx := context.Background() + d := newTestDelegate(t) + + _, err := d.ServicesForSpec(ctx, cresettingsJob(1, `config_type = "unknown" +Foo = "bar" +`)) + require.ErrorContains(t, err, `unknown config_type "unknown"`) + + // unknown config_type must not claim a slot + _, err = d.ServicesForSpec(ctx, cresettingsJob(2, settingsToml)) + require.NoError(t, err) +} + +func TestOnDeleteJobFreesSlotPerConfigType(t *testing.T) { + t.Parallel() + + ctx := context.Background() + d := newTestDelegate(t) + + _, err := d.ServicesForSpec(ctx, cresettingsJob(1, settingsToml)) + require.NoError(t, err) + _, err = d.ServicesForSpec(ctx, cresettingsJob(2, shardAssignmentToml)) + require.NoError(t, err) + + // deleting a rejected duplicate must not free the active job's slot + _, err = d.ServicesForSpec(ctx, cresettingsJob(3, settingsToml)) + require.ErrorContains(t, err, "already active: 1") + require.NoError(t, d.OnDeleteJob(ctx, cresettingsJob(3, settingsToml))) + _, err = d.ServicesForSpec(ctx, cresettingsJob(4, settingsToml)) + require.ErrorContains(t, err, "already active: 1") + + // deleting the active settings job frees only the settings slot + require.NoError(t, d.OnDeleteJob(ctx, cresettingsJob(1, settingsToml))) + _, err = d.ServicesForSpec(ctx, cresettingsJob(5, settingsToml)) + require.NoError(t, err) + + // shard assignment slot is unaffected + _, err = d.ServicesForSpec(ctx, cresettingsJob(6, shardAssignmentToml)) + require.ErrorContains(t, err, "already active: 2") + require.NoError(t, d.OnDeleteJob(ctx, cresettingsJob(2, shardAssignmentToml))) + _, err = d.ServicesForSpec(ctx, cresettingsJob(7, shardAssignmentToml)) + require.NoError(t, err) +}