Skip to content

Commit

Permalink
Support passing in a values.Value to the chainreader GetLatestValue m…
Browse files Browse the repository at this point in the history 
…ethod (#779)

* add support for passing in a values.Value type to the contract readers GetLatestValue and QueryKey methods

---------

Co-authored-by: Sri Kidambi <[email protected]>
Co-authored-by: Cedric Cordenier <[email protected]>
  • Loading branch information
@ettec @kidambisrinivas @cedric-cordenier
3 people authored Sep 25, 2024
1 parent 96611a2 commit aded1b2
Show file tree
Hide file tree
Showing 7 changed files with 494 additions and 242 deletions.
468 changes: 244 additions & 224 deletions pkg/loop/internal/pb/contract_reader.pb.go
View file

Large diffs are not rendered by default.

2 changes: 2 additions & 0 deletions pkg/loop/internal/pb/contract_reader.proto
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ message GetLatestValueRequest {
string read_identifier = 1;
Confidence confidence = 2;
VersionedBytes params = 3;
bool as_value_type = 4;
}

// BatchGetLatestValuesRequest has arguments for [github.com/smartcontractkit/chainlink-common/pkg/types.ContractReader.BatchGetLatestValues].
Expand All @@ -32,6 +33,7 @@ message QueryKeyRequest {
BoundContract contract = 1;
QueryKeyFilter filter = 2;
LimitAndSort limit_and_sort = 3;
bool as_value_type = 4;
}

// BindRequest has arguments for [github.com/smartcontractkit/chainlink-common/pkg/types.ContractReader.Bind].
Expand Down
63 changes: 57 additions & 6 deletions pkg/loop/internal/relayer/pluginprovider/contractreader/contract_reader.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/emptypb"

"github.com/smartcontractkit/chainlink-common/pkg/loop/internal/goplugin"
Expand All @@ -21,6 +22,8 @@ import (
"github.com/smartcontractkit/chainlink-common/pkg/types"
"github.com/smartcontractkit/chainlink-common/pkg/types/query"
"github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives"
"github.com/smartcontractkit/chainlink-common/pkg/values"
valuespb "github.com/smartcontractkit/chainlink-common/pkg/values/pb"
)

var _ types.ContractReader = (*Client)(nil)
Expand All @@ -36,6 +39,7 @@ const (
JSONEncodingVersion1 EncodingVersion = iota
JSONEncodingVersion2
CBOREncodingVersion
ValuesEncodingVersion
)

const DefaultEncodingVersion = CBOREncodingVersion
Expand Down Expand Up @@ -96,6 +100,15 @@ func EncodeVersionedBytes(data any, version EncodingVersion) (*pb.VersionedBytes
if err != nil {
return nil, fmt.Errorf("%w: %w", types.ErrInvalidType, err)
}
case ValuesEncodingVersion:
val, err := values.Wrap(data)
if err != nil {
return nil, fmt.Errorf("%w: %w", types.ErrInvalidType, err)
}
bytes, err = proto.Marshal(values.Proto(val))
if err != nil {
return nil, fmt.Errorf("%w: %w", types.ErrInvalidType, err)
}
default:
return nil, fmt.Errorf("%w: unsupported encoding version %d for data %v", types.ErrInvalidEncoding, version, data)
}
Expand All @@ -121,6 +134,28 @@ func DecodeVersionedBytes(res any, vData *pb.VersionedBytes) error {
return fmt.Errorf("%w: %w", types.ErrInternal, err)
}
err = dec.Unmarshal(vData.Data, res)
case ValuesEncodingVersion:
protoValue := &valuespb.Value{}
err = proto.Unmarshal(vData.Data, protoValue)
if err != nil {
return fmt.Errorf("%w: %w", types.ErrInvalidType, err)
}

var value values.Value
value, err = values.FromProto(protoValue)
if err != nil {
return fmt.Errorf("%w: %w", types.ErrInvalidType, err)
}

valuePtr, ok := res.(*values.Value)
if ok {
*valuePtr = value
} else {
err = value.UnwrapTo(res)
if err != nil {
return fmt.Errorf("%w: %w", types.ErrInvalidType, err)
}
}
default:
return fmt.Errorf("unsupported encoding version %d for versionedData %v", vData.Version, vData.Data)
}
Expand All @@ -133,6 +168,8 @@ func DecodeVersionedBytes(res any, vData *pb.VersionedBytes) error {
}

func (c *Client) GetLatestValue(ctx context.Context, readIdentifier string, confidenceLevel primitives.ConfidenceLevel, params, retVal any) error {
_, asValueType := retVal.(*values.Value)

versionedParams, err := EncodeVersionedBytes(params, c.encodeWith)
if err != nil {
return err
Expand All @@ -149,6 +186,7 @@ func (c *Client) GetLatestValue(ctx context.Context, readIdentifier string, conf
ReadIdentifier: readIdentifier,
Confidence: pbConfidence,
Params: versionedParams,
AsValueType: asValueType,
},
)
if err != nil {
Expand All @@ -173,6 +211,8 @@ func (c *Client) BatchGetLatestValues(ctx context.Context, request types.BatchGe
}

func (c *Client) QueryKey(ctx context.Context, contract types.BoundContract, filter query.KeyFilter, limitAndSort query.LimitAndSort, sequenceDataType any) ([]types.Sequence, error) {
_, asValueType := sequenceDataType.(*values.Value)

pbQueryFilter, err := convertQueryFilterToProto(filter, c.encodeWith)
if err != nil {
return nil, err
Expand All @@ -192,6 +232,7 @@ func (c *Client) QueryKey(ctx context.Context, contract types.BoundContract, fil
},
Filter: pbQueryFilter,
LimitAndSort: pbLimitAndSort,
AsValueType: asValueType,
},
)
if err != nil {
Expand Down Expand Up @@ -306,12 +347,17 @@ func (c *Server) GetLatestValue(ctx context.Context, request *pb.GetLatestValueR
return nil, err
}

encodedRetVal, err := EncodeVersionedBytes(retVal, EncodingVersion(request.Params.Version))
encodeWith := EncodingVersion(request.Params.Version)
if request.AsValueType {
encodeWith = ValuesEncodingVersion
}

versionedBytes, err := EncodeVersionedBytes(retVal, encodeWith)
if err != nil {
return nil, err
}

return &pb.GetLatestValueReply{RetVal: encodedRetVal}, nil
return &pb.GetLatestValueReply{RetVal: versionedBytes}, nil
}

func (c *Server) BatchGetLatestValues(ctx context.Context, pbRequest *pb.BatchGetLatestValuesRequest) (*pb.BatchGetLatestValuesReply, error) {
Expand Down Expand Up @@ -351,7 +397,12 @@ func (c *Server) QueryKey(ctx context.Context, request *pb.QueryKeyRequest) (*pb
return nil, err
}

pbSequences, err := convertSequencesToProto(sequences, c.encodeWith)
encodeWith := c.encodeWith
if request.AsValueType {
encodeWith = ValuesEncodingVersion
}

pbSequences, err := convertSequencesToVersionedBytesProto(sequences, encodeWith)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -601,7 +652,7 @@ func convertLimitAndSortToProto(limitAndSort query.LimitAndSort) (*pb.LimitAndSo
return pbLimitAndSort, nil
}

func convertSequencesToProto(sequences []types.Sequence, version EncodingVersion) ([]*pb.Sequence, error) {
func convertSequencesToVersionedBytesProto(sequences []types.Sequence, version EncodingVersion) ([]*pb.Sequence, error) {
var pbSequences []*pb.Sequence
for _, sequence := range sequences {
versionedSequenceDataType, err := EncodeVersionedBytes(sequence.Data, version)
Expand Down Expand Up @@ -811,6 +862,8 @@ func convertLimitAndSortFromProto(limitAndSort *pb.LimitAndSort) (query.LimitAnd
}

func convertSequencesFromProto(pbSequences []*pb.Sequence, sequenceDataType any) ([]types.Sequence, error) {
sequences := make([]types.Sequence, len(pbSequences))

seqTypeOf := reflect.TypeOf(sequenceDataType)

// get the non-pointer data type for the sequence data
Expand All @@ -823,8 +876,6 @@ func convertSequencesFromProto(pbSequences []*pb.Sequence, sequenceDataType any)
return nil, fmt.Errorf("%w: sequenceDataType does not support pointers to pointers", types.ErrInvalidType)
}

sequences := make([]types.Sequence, len(pbSequences))

for idx, pbSequence := range pbSequences {
cpy := reflect.New(nonPointerType).Interface()
if err := DecodeVersionedBytes(cpy, pbSequence.Data); err != nil {
Expand Down
78 changes: 67 additions & 11 deletions pkg/loop/internal/relayer/pluginprovider/contractreader/contract_reader_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import (
"github.com/smartcontractkit/chainlink-common/pkg/types/query"
"github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives"
"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
"github.com/smartcontractkit/chainlink-common/pkg/values"

. "github.com/smartcontractkit/chainlink-common/pkg/types/interfacetests" //nolint
)
Expand Down Expand Up @@ -434,6 +435,7 @@ func (f *fakeContractReader) SetBatchLatestValues(batchCallEntry BatchCallEntry)
}

func (f *fakeContractReader) GetLatestValue(_ context.Context, readIdentifier string, confidenceLevel primitives.ConfidenceLevel, params, returnVal any) error {

if strings.HasSuffix(readIdentifier, MethodReturningAlterableUint64) {
r := returnVal.(*uint64)
for i := len(f.vals) - 1; i >= 0; i-- {
Expand Down Expand Up @@ -499,12 +501,26 @@ func (f *fakeContractReader) GetLatestValue(_ context.Context, readIdentifier st
f.lock.Lock()
defer f.lock.Unlock()
lp := params.(*LatestParams)
rv := returnVal.(*TestStruct)

if lp.I-1 >= len(f.stored) {
return errors.New("latest params index out of bounds for stored test structs")
}
*rv = f.stored[lp.I-1]

_, isValue := returnVal.(*values.Value)
if isValue {
var err error
ptrToVal := returnVal.(*values.Value)
*ptrToVal, err = values.Wrap(f.stored[lp.I-1])
if err != nil {
return err
}
} else {
rv := returnVal.(*TestStruct)
*rv = f.stored[lp.I-1]
}

return nil

}

func (f *fakeContractReader) BatchGetLatestValues(_ context.Context, request types.BatchGetLatestValuesRequest) (types.BatchGetLatestValuesResult, error) {
Expand Down Expand Up @@ -560,7 +576,9 @@ func (f *fakeContractReader) BatchGetLatestValues(_ context.Context, request typ
return result, nil
}

func (f *fakeContractReader) QueryKey(_ context.Context, _ types.BoundContract, filter query.KeyFilter, limitAndSort query.LimitAndSort, _ any) ([]types.Sequence, error) {
func (f *fakeContractReader) QueryKey(_ context.Context, _ types.BoundContract, filter query.KeyFilter, limitAndSort query.LimitAndSort, sequenceType any) ([]types.Sequence, error) {
_, isValueType := sequenceType.(*values.Value)

if filter.Key == EventName {
f.lock.Lock()
defer f.lock.Unlock()
Expand All @@ -584,17 +602,55 @@ func (f *fakeContractReader) QueryKey(_ context.Context, _ types.BoundContract,
}
}
if len(filter.Expressions) == 0 || doAppend {
sequences = append(sequences, types.Sequence{Data: trigger.testStruct})

if isValueType {
value, err := values.Wrap(trigger.testStruct)
if err != nil {
return nil, err
}
sequences = append(sequences, types.Sequence{Data: &value})
} else {
sequences = append(sequences, types.Sequence{Data: trigger.testStruct})
}
}
}

if !limitAndSort.HasSequenceSort() {
sort.Slice(sequences, func(i, j int) bool {
if sequences[i].Data.(TestStruct).Field == nil || sequences[j].Data.(TestStruct).Field == nil {
return false
}
return *sequences[i].Data.(TestStruct).Field > *sequences[j].Data.(TestStruct).Field
})
if isValueType {
if !limitAndSort.HasSequenceSort() {
sort.Slice(sequences, func(i, j int) bool {
valI := *sequences[i].Data.(*values.Value)
valJ := *sequences[j].Data.(*values.Value)

mapI := valI.(*values.Map)
mapJ := valJ.(*values.Map)

if mapI.Underlying["Field"] == nil || mapJ.Underlying["Field"] == nil {
return false
}
var iVal int32
err := mapI.Underlying["Field"].UnwrapTo(&iVal)
if err != nil {
panic(err)
}

var jVal int32
err = mapJ.Underlying["Field"].UnwrapTo(&jVal)
if err != nil {
panic(err)
}

return iVal > jVal
})
}
} else {
if !limitAndSort.HasSequenceSort() {
sort.Slice(sequences, func(i, j int) bool {
if sequences[i].Data.(TestStruct).Field == nil || sequences[j].Data.(TestStruct).Field == nil {
return false
}
return *sequences[i].Data.(TestStruct).Field > *sequences[j].Data.(TestStruct).Field
})
}
}

return sequences, nil
Expand Down
1 change: 0 additions & 1 deletion pkg/loop/internal/relayer/pluginprovider/contractreader/test/contract_reader.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,6 @@ func (c staticContractReader) GetLatestValue(_ context.Context, readName string,
return fmt.Errorf("%w: expected report context %v but got %v", types.ErrInvalidType, comp, readName)
}

//gotParams, ok := params.(*map[string]string)
gotParams, ok := params.(*map[string]any)
if !ok {
return fmt.Errorf("%w: Invalid parameter type received in GetLatestValue. Expected %T but received %T", types.ErrInvalidEncoding, c.params, params)
Expand Down
1 change: 1 addition & 0 deletions pkg/types/contract_reader.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ type ContractReader interface {
// Note that implementations should ignore extra fields in params that are not expected in the call to allow easier
// use across chains and contract versions.
// Similarly, when using a struct for returnVal, fields in the return value that are not on-chain will not be set.
// Passing in a *values.Value as the returnVal will encode the return value as an appropriate value.Value instance.
GetLatestValue(ctx context.Context, readIdentifier string, confidenceLevel primitives.ConfidenceLevel, params, returnVal any) error

// BatchGetLatestValues batches get latest value calls based on request, which is grouped by contract names that each have a slice of BatchRead.
Expand Down
Loading

0 comments on commit aded1b2

Please sign in to comment.