[pull] develop from smartcontractkit:develop

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
+}

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)
+}