CRE-1601: Ring OCR plugin for shard orchestration

go.mod

@@ -7,7 +7,9 @@ require (
 	github.com/XSAM/otelsql v0.37.0
 	github.com/andybalholm/brotli v1.1.1
 	github.com/atombender/go-jsonschema v0.16.1-0.20240916205339-a74cd4e2851c
+	github.com/buraksezer/consistent v0.10.0
 	github.com/bytecodealliance/wasmtime-go/v28 v28.0.0
+	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
 	github.com/dominikbraun/graph v0.23.0
 	github.com/fxamacker/cbor/v2 v2.7.0
@@ -88,7 +90,6 @@ require (
 	github.com/buger/goterm v1.0.4 // indirect
 	github.com/buger/jsonparser v1.1.1 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.2 // indirect
-	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudevents/sdk-go/binding/format/protobuf/v2 v2.16.1 // indirect
 	github.com/cloudevents/sdk-go/v2 v2.16.1 // indirect
 	github.com/fatih/color v1.18.0 // indirect

go.sum

@@ -28,6 +28,8 @@ github.com/buger/goterm v1.0.4 h1:Z9YvGmOih81P0FbVtEYTFF6YsSgxSUKEhf/f9bTMXbY=
 github.com/buger/goterm v1.0.4/go.mod h1:HiFWV3xnkolgrBV3mY8m0X0Pumt4zg4QhbdOzQtB8tE=
 github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs=
 github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0=
+github.com/buraksezer/consistent v0.10.0 h1:hqBgz1PvNLC5rkWcEBVAL9dFMBWz6I0VgUCW25rrZlU=
+github.com/buraksezer/consistent v0.10.0/go.mod h1:6BrVajWq7wbKZlTOUPs/XVfR8c0maujuPowduSpZqmw=
 github.com/bytecodealliance/wasmtime-go/v28 v28.0.0 h1:aBU8cexP2rPZ0Qz488kvn2NXvWZHL2aG1/+n7Iv+xGc=
 github.com/bytecodealliance/wasmtime-go/v28 v28.0.0/go.mod h1:4OCU0xAW9ycwtX4nMF4zxwgJBJ5/0eMfJiHB0wAmkV4=
 github.com/cenkalti/backoff/v5 v5.0.2 h1:rIfFVxEf1QsI7E1ZHfp/B4DF/6QBAUhmgkxc0H7Zss8=

pkg/types/plugin.go

@@ -17,6 +17,7 @@ const (
 	OCR3Capability OCR2PluginType = "ocr3-capability"
 	VaultPlugin    OCR2PluginType = "vault-plugin"
 	DonTimePlugin  OCR2PluginType = "dontime"
+	RingPlugin     OCR2PluginType = "ring"
 	SecureMint     OCR2PluginType = "securemint"
 
 	CCIPCommit    OCR2PluginType = "ccip-commit"

pkg/workflows/ring/factory.go

@@ -0,0 +1,79 @@
+package ring
+
+import (
+	"context"
+	"errors"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
+	"github.com/smartcontractkit/chainlink-common/pkg/workflows/ring/pb"
+	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
+)
+
+const (
+	defaultMaxPhaseOutputBytes = 1000000 // 1 MB
+	defaultMaxReportCount      = 1
+	defaultBatchSize           = 100
+)
+
+var _ core.OCR3ReportingPluginFactory = &Factory{}
+
+type Factory struct {
+	store         *Store
+	arbiterScaler pb.ArbiterScalerClient
+	config        *ConsensusConfig
+	lggr          logger.Logger
+
+	services.StateMachine
+}
+
+func NewFactory(s *Store, arbiterScaler pb.ArbiterScalerClient, lggr logger.Logger, cfg *ConsensusConfig) (*Factory, error) {
+	if arbiterScaler == nil {
+		return nil, errors.New("arbiterScaler is required")
+	}
+	if cfg == nil {
+		cfg = &ConsensusConfig{
+			BatchSize: defaultBatchSize,
+		}
+	}
+	return &Factory{
+		store:         s,
+		arbiterScaler: arbiterScaler,
+		config:        cfg,
+		lggr:          logger.Named(lggr, "RingPluginFactory"),
+	}, nil
+}
+
+func (o *Factory) NewReportingPlugin(_ context.Context, config ocr3types.ReportingPluginConfig) (ocr3types.ReportingPlugin[[]byte], ocr3types.ReportingPluginInfo, error) {
+	plugin, err := NewPlugin(o.store, o.arbiterScaler, config, o.lggr, o.config)
+	pluginInfo := ocr3types.ReportingPluginInfo{
+		Name: "RingPlugin",
+		Limits: ocr3types.ReportingPluginLimits{
+			MaxQueryLength:       defaultMaxPhaseOutputBytes,
+			MaxObservationLength: defaultMaxPhaseOutputBytes,
+			MaxOutcomeLength:     defaultMaxPhaseOutputBytes,
+			MaxReportLength:      defaultMaxPhaseOutputBytes,
+			MaxReportCount:       defaultMaxReportCount,
+		},
+	}
+	return plugin, pluginInfo, err
+}
+
+func (o *Factory) Start(ctx context.Context) error {
+	return o.StartOnce("RingPlugin", func() error {
+		return nil
+	})
+}
+
+func (o *Factory) Close() error {
+	return o.StopOnce("RingPlugin", func() error {
+		return nil
+	})
+}
+
+func (o *Factory) Name() string { return o.lggr.Name() }
+
+func (o *Factory) HealthReport() map[string]error {
+	return map[string]error{o.Name(): o.Healthy()}
+}

pkg/workflows/ring/factory_test.go

@@ -0,0 +1,70 @@
+package ring
+
+import (
+	"context"
+	"testing"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFactory_NewFactory(t *testing.T) {
+	lggr := logger.Test(t)
+	store := NewStore()
+	arbiter := &mockArbiter{}
+
+	t.Run("with_nil_config", func(t *testing.T) {
+		f, err := NewFactory(store, arbiter, lggr, nil)
+		require.NoError(t, err)
+		require.NotNil(t, f)
+	})
+
+	t.Run("with_custom_config", func(t *testing.T) {
+		cfg := &ConsensusConfig{BatchSize: 50}
+		f, err := NewFactory(store, arbiter, lggr, cfg)
+		require.NoError(t, err)
+		require.NotNil(t, f)
+	})
+
+	t.Run("nil_arbiter_returns_error", func(t *testing.T) {
+		_, err := NewFactory(store, nil, lggr, nil)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "arbiterScaler is required")
+	})
+}
+
+func TestFactory_NewReportingPlugin(t *testing.T) {
+	lggr := logger.Test(t)
+	store := NewStore()
+	f, err := NewFactory(store, &mockArbiter{}, lggr, nil)
+	require.NoError(t, err)
+
+	config := ocr3types.ReportingPluginConfig{N: 4, F: 1}
+	plugin, info, err := f.NewReportingPlugin(context.Background(), config)
+	require.NoError(t, err)
+	require.NotNil(t, plugin)
+	require.NotEmpty(t, info.Name)
+	require.Equal(t, "RingPlugin", info.Name)
+	require.Equal(t, defaultMaxReportCount, info.Limits.MaxReportCount)
+}
+
+func TestFactory_Lifecycle(t *testing.T) {
+	lggr := logger.Test(t)
+	store := NewStore()
+	f, err := NewFactory(store, &mockArbiter{}, lggr, nil)
+	require.NoError(t, err)
+
+	err = f.Start(context.Background())
+	require.NoError(t, err)
+
+	name := f.Name()
+	require.NotEmpty(t, name)
+
+	report := f.HealthReport()
+	require.NotNil(t, report)
+	require.Contains(t, report, name)
+
+	err = f.Close()
+	require.NoError(t, err)
+}