BCF-2654 chain reader get latest value evm poc WIP

.github/workflows/integration-tests.yml

@@ -620,7 +620,7 @@ jobs:
         uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1
         with:
           repository: smartcontractkit/chainlink-solana
-          ref: develop
+          ref: BCF-2612-ChainReader
           fetch-depth: 0
           path: solanapath
       - name: Get long sha

.tool-versions

@@ -1,4 +1,4 @@
-golang 1.21.1
+golang 1.21.3
 mockery 2.35.4
 nodejs 16.16.0
 postgres 13.3

core/scripts/go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/docker/docker v24.0.6+incompatible
 	github.com/docker/go-connections v0.4.0
 	github.com/ethereum/go-ethereum v1.12.0
-	github.com/google/go-cmp v0.5.9
+	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.3.1
 	github.com/joho/godotenv v1.4.0
 	github.com/manyminds/api2go v0.0.0-20171030193247-e7b693844a6f

core/scripts/go.sum

@@ -565,8 +565,9 @@ github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
 github.com/google/go-querystring v1.1.0 h1:AnCroh3fv4ZBgVIf1Iwtovgjaw/GiKJo8M8yD/fhyJ8=
 github.com/google/go-querystring v1.1.0/go.mod h1:Kcdr2DB4koayq7X8pmAG4sNG59So17icRSOU623lUBU=

core/services/job/models.go

@@ -279,6 +279,34 @@ func (r JSONConfig) Bytes() []byte {
 	return b
 }
 
+// BytesWithPreservedJson returns raw bytes and properly marshals any potential json structure strings.
+func (r JSONConfig) BytesWithPreservedJson() []byte {
+	var retCopy = make(JSONConfig, 0)
+	for key, value := range r {
+		copiedVal := value
+		// If the value is a json structure string, unmarshal it to preserve JSON structure
+		// e.g. instead of this {"key":"{\"nestedKey\":{\"nestedValue\":123}}"}
+		//      we want this    {"key":{"nestedKey":{"nestedValue":123}}},
+		if strValue, ok := copiedVal.(string); ok {
+			if isValidJSONStruct(strValue) {
+				var parsedValue interface{}
+				if err := json.Unmarshal([]byte(strValue), &parsedValue); err == nil {
+					copiedVal = parsedValue
+				}
+			}
+		}
+		retCopy[key] = copiedVal
+	}
+
+	b, _ := json.Marshal(retCopy)
+	return b
+}
+
+func isValidJSONStruct(s string) bool {
+	var js map[string]interface{}
+	return json.Unmarshal([]byte(s), &js) == nil
+}
+
 // Value returns this instance serialized for database storage.
 func (r JSONConfig) Value() (driver.Value, error) {
 	return json.Marshal(r)

core/services/job/models_test.go

@@ -71,3 +71,69 @@ func TestOCR2OracleSpec_RelayIdentifier(t *testing.T) {
 		})
 	}
 }
+
+func TestJSONConfig_BytesWithPreservedJson(t *testing.T) {
+	type testCases struct {
+		name     string
+		Input    JSONConfig
+		Expected []byte
+	}
+	tests := []testCases{
+		{
+			name: "json",
+			Input: JSONConfig{
+				"key": "{\"nestedKey\": {\"nestedValue\":123}}",
+			},
+			// regular Bytes marshals to this: {"key":"{\"nestedKey\": {\"nestedValue\":123}}"}
+			Expected: []byte(`{"key":{"nestedKey":{"nestedValue":123}}}`),
+		},
+		{
+			name: "broken json gets treated as a regular string",
+			Input: JSONConfig{
+				"key": "2324{\"nes4tedKey\":\"nestedValue\"}",
+			},
+			Expected: []byte(`{"key":"2324{\"nes4tedKey\":\"nestedValue\"}"}`),
+		},
+		{
+			name: "number",
+			Input: JSONConfig{
+				"key": 1,
+			},
+			Expected: []byte(`{"key":1}`),
+		},
+		{
+			name: "string",
+			Input: JSONConfig{
+				"key": "abc",
+			},
+			Expected: []byte(`{"key":"abc"}`),
+		},
+		{
+			name: "string number stays string number",
+			Input: JSONConfig{
+				"key1": "1",
+			},
+			Expected: []byte(`{"key1":"1"}`),
+		},
+		{
+			name: "all together",
+			Input: JSONConfig{
+				"key1": "{\"nestedKey\": {\"nestedValue\":123}}",
+				"key2": "2324{\"key\":\"value\"}",
+				"key3": 1,
+				"key4": "abc",
+			},
+			Expected: []byte(`{"key1":{"nestedKey":{"nestedValue":123}},"key2":"2324{\"key\":\"value\"}","key3":1,"key4":"abc"}`),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			result := tc.Input.BytesWithPreservedJson()
+			if !reflect.DeepEqual(result, tc.Expected) {
+				t.Errorf("Input: %v, BytesWithPreservedJson() returned unexpected result. Expected: %s, Got: %s", tc.Input, tc.Expected, result)
+			}
+		})
+
+	}
+}

core/services/ocr2/delegate.go

@@ -315,7 +315,7 @@ func (d *Delegate) cleanupEVM(jb job.Job, q pg.Queryer, relayID relay.ID) error
 		JobID:         jb.ID,
 		ContractID:    spec.ContractID,
 		New:           false,
-		RelayConfig:   spec.RelayConfig.Bytes(),
+		RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 	}
 
 	relayFilters, err := evmrelay.FilterNamesFromRelayArgs(rargs)
@@ -552,7 +552,7 @@ func (d *Delegate) newServicesGenericPlugin(
 		JobID:         spec.ID,
 		ContractID:    spec.ContractID,
 		New:           d.isNewlyCreatedJob,
-		RelayConfig:   spec.RelayConfig.Bytes(),
+		RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 		ProviderType:  cconf.ProviderType,
 	}, types.PluginArgs{
 		TransmitterID: spec.TransmitterID.String,
@@ -664,7 +664,7 @@ func (d *Delegate) newServicesMercury(
 			JobID:         jb.ID,
 			ContractID:    spec.ContractID,
 			New:           d.isNewlyCreatedJob,
-			RelayConfig:   spec.RelayConfig.Bytes(),
+			RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 			ProviderType:  string(spec.PluginType),
 		}, types.PluginArgs{
 			TransmitterID: transmitterID,
@@ -785,7 +785,7 @@ func (d *Delegate) newServicesDKG(
 			JobID:         jb.ID,
 			ContractID:    spec.ContractID,
 			New:           d.isNewlyCreatedJob,
-			RelayConfig:   spec.RelayConfig.Bytes(),
+			RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 		}, types.PluginArgs{
 			TransmitterID: spec.TransmitterID.String,
 			PluginConfig:  spec.PluginConfig.Bytes(),
@@ -870,7 +870,7 @@ func (d *Delegate) newServicesOCR2VRF(
 			JobID:         jb.ID,
 			ContractID:    spec.ContractID,
 			New:           d.isNewlyCreatedJob,
-			RelayConfig:   spec.RelayConfig.Bytes(),
+			RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 		}, types.PluginArgs{
 			TransmitterID: transmitterID,
 			PluginConfig:  spec.PluginConfig.Bytes(),
@@ -884,7 +884,7 @@ func (d *Delegate) newServicesOCR2VRF(
 			ExternalJobID: jb.ExternalJobID,
 			JobID:         jb.ID,
 			ContractID:    cfg.DKGContractAddress,
-			RelayConfig:   spec.RelayConfig.Bytes(),
+			RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 		}, types.PluginArgs{
 			TransmitterID: transmitterID,
 			PluginConfig:  spec.PluginConfig.Bytes(),
@@ -1327,7 +1327,7 @@ func (d *Delegate) newServicesOCR2Functions(
 				ExternalJobID: jb.ExternalJobID,
 				JobID:         jb.ID,
 				ContractID:    spec.ContractID,
-				RelayConfig:   spec.RelayConfig.Bytes(),
+				RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 				New:           d.isNewlyCreatedJob,
 			},
 			types.PluginArgs{

core/services/ocr2/plugins/median/plugin.go

@@ -27,6 +27,7 @@ func (p *Plugin) NewMedianFactory(ctx context.Context, provider types.MedianProv
 	var ctxVals loop.ContextValues
 	ctxVals.SetValues(ctx)
 	lggr := logger.With(p.Logger, ctxVals.Args()...)
+
 	factory := median.NumericalMedianFactory{
 		ContractTransmitter:       provider.MedianContract(),
 		DataSource:                dataSource,

core/services/ocr2/plugins/median/services.go

@@ -5,9 +5,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math/big"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/smartcontractkit/libocr/offchainreporting2/reportingplugin/median"
 	libocr "github.com/smartcontractkit/libocr/offchainreporting2plus"
+	ocr2types "github.com/smartcontractkit/libocr/offchainreporting2plus/types"
 
 	"github.com/smartcontractkit/chainlink-relay/pkg/loop"
 	"github.com/smartcontractkit/chainlink-relay/pkg/types"
@@ -22,6 +26,8 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/plugins"
 )
 
+const ContractName = "offchain_aggregator"
+
 type MedianConfig interface {
 	JobPipelineMaxSuccessfulRuns() uint64
 	plugins.RegistrarConfig
@@ -44,6 +50,21 @@ func (m *medianConfig) JobPipelineMaxSuccessfulRuns() uint64 {
 	return m.jobPipelineMaxSuccessfulRuns
 }
 
+// This wrapper avoids the need to modify the signature of NewMedianFactory in all of the non-evm
+// relay repos as well as its primary definition in chainlink-relay. Once ChainReader is implemented
+// and working on all 4 blockchain families, we can remove the original MedianContract() method from
+// MedianProvider and pass medianContract as a separate param to NewMedianFactory
+type medianProviderWrapper struct {
+	types.MedianProvider
+	contract median.MedianContract
+}
+
+// Override relay's implementation of MedianContract with product plugin's implementation of
+// MedianContract, making use of product-agnostic ChainReader to read the contract instead of relay MedianContract
+func (m medianProviderWrapper) MedianContract() median.MedianContract {
+	return m.contract
+}
+
 func NewMedianServices(ctx context.Context,
 	jb job.Job,
 	isNewlyCreatedJob bool,
@@ -73,7 +94,7 @@ func NewMedianServices(ctx context.Context,
 		JobID:         jb.ID,
 		ContractID:    spec.ContractID,
 		New:           isNewlyCreatedJob,
-		RelayConfig:   spec.RelayConfig.Bytes(),
+		RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 		ProviderType:  string(spec.PluginType),
 	}, types.PluginArgs{
 		TransmitterID: spec.TransmitterID.String,
@@ -111,6 +132,13 @@ func NewMedianServices(ctx context.Context,
 		CreatedAt:    time.Now(),
 	}, lggr)
 
+	if medianProvider.ChainReader() != nil {
+		medianProvider = medianProviderWrapper{
+			medianProvider, // attach newer MedianContract which uses ChainReader
+			newMedianContract(provider.ChainReader(), common.HexToAddress(spec.ContractID)),
+		}
+	}
+
 	if cmdName := env.MedianPluginCmd.Get(); cmdName != "" {
 
 		// use unique logger names so we can use it to register a loop
@@ -152,3 +180,48 @@ func NewMedianServices(ctx context.Context,
 	}
 	return
 }
+
+type medianContract struct {
+	chainReader types.ChainReader
+	contract    types.BoundContract
+}
+
+type latestTransmissionDetailsResponse struct {
+	ConfigDigest    ocr2types.ConfigDigest
+	Epoch           uint32
+	Round           uint8
+	LatestAnswer    *big.Int
+	LatestTimestamp uint64
+}
+
+type latestRoundRequested struct {
+	ConfigDigest ocr2types.ConfigDigest
+	Epoch        uint32
+	Round        uint8
+}
+
+func (m *medianContract) LatestTransmissionDetails(ctx context.Context) (configDigest ocr2types.ConfigDigest, epoch uint32, round uint8, latestAnswer *big.Int, latestTimestamp time.Time, err error) {
+	var resp latestTransmissionDetailsResponse
+
+	err = m.chainReader.GetLatestValue(ctx, m.contract, "LatestTransmissionDetails", nil, &resp)
+	if err != nil {
+		return
+	}
+	return resp.ConfigDigest, resp.Epoch, resp.Round, resp.LatestAnswer, time.Unix(int64(resp.LatestTimestamp), 0), err
+}
+
+func (m *medianContract) LatestRoundRequested(ctx context.Context, lookback time.Duration) (configDigest ocr2types.ConfigDigest, epoch uint32, round uint8, err error) {
+	var resp latestRoundRequested
+
+	err = m.chainReader.GetLatestValue(ctx, m.contract, "LatestRoundReported", map[string]string{}, &resp)
+	if err != nil {
+		return
+	}
+
+	return resp.ConfigDigest, resp.Epoch, resp.Round, err
+}
+
+func newMedianContract(chainReader types.ChainReader, address common.Address) *medianContract {
+	contract := types.BoundContract{Address: address.String(), Name: ContractName, Pending: true}
+	return &medianContract{chainReader, contract}
+}

core/services/ocr2/plugins/ocr2keeper/util.go

@@ -52,7 +52,7 @@ func EVMProvider(db *sqlx.DB, chain evm.Chain, lggr logger.Logger, spec job.Job,
 			ExternalJobID: spec.ExternalJobID,
 			JobID:         oSpec.ID,
 			ContractID:    oSpec.ContractID,
-			RelayConfig:   oSpec.RelayConfig.Bytes(),
+			RelayConfig:   oSpec.RelayConfig.BytesWithPreservedJson(),
 		},
 		types.PluginArgs{
 			TransmitterID: oSpec.TransmitterID.String,

core/services/ocrbootstrap/delegate.go

@@ -109,7 +109,7 @@ func (d *Delegate) ServicesForSpec(jb job.Job) (services []job.ServiceCtx, err e
 	ctx := ctxVals.ContextWithValues(context.Background())
 
 	var routerFields relayConfigRouterContractFields
-	if err = json.Unmarshal(spec.RelayConfig.Bytes(), &routerFields); err != nil {
+	if err = json.Unmarshal(spec.RelayConfig.BytesWithPreservedJson(), &routerFields); err != nil {
 		return nil, err
 	}
 
@@ -124,12 +124,12 @@ func (d *Delegate) ServicesForSpec(jb job.Job) (services []job.ServiceCtx, err e
 				ExternalJobID: jb.ExternalJobID,
 				JobID:         jb.ID,
 				ContractID:    spec.ContractID,
-				RelayConfig:   spec.RelayConfig.Bytes(),
+				RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 				New:           d.isNewlyCreatedJob,
 				ProviderType:  string(types.Functions),
 			},
 			types.PluginArgs{
-				PluginConfig: spec.RelayConfig.Bytes(), // contains all necessary fields for config provider
+				PluginConfig: spec.RelayConfig.BytesWithPreservedJson(), // contains all necessary fields for config provider
 			},
 		)
 	} else {
@@ -138,7 +138,7 @@ func (d *Delegate) ServicesForSpec(jb job.Job) (services []job.ServiceCtx, err e
 			JobID:         jb.ID,
 			ContractID:    spec.ContractID,
 			New:           d.isNewlyCreatedJob,
-			RelayConfig:   spec.RelayConfig.Bytes(),
+			RelayConfig:   spec.RelayConfig.BytesWithPreservedJson(),
 		})
 	}