Merge pull request #6602 from smartcontractkit/release/1.4.1

VERSION

@@ -1 +1 @@
-1.4.0
+1.4.1

core/chains/evm/client/node.go

@@ -469,6 +469,9 @@ func (n *node) EthSubscribe(ctx context.Context, channel chan<- *evmtypes.Head,
 	lggr.Debug("RPC call: evmclient.Client#EthSubscribe")
 	start := time.Now()
 	sub, err := n.ws.rpc.EthSubscribe(ctx, channel, args...)
+	if err == nil {
+		n.registerSub(sub)
+	}
 	duration := time.Since(start)
 
 	n.logResult(lggr, err, duration, n.getRPCDomain(), "EthSubscribe",
@@ -477,9 +480,6 @@ func (n *node) EthSubscribe(ctx context.Context, channel chan<- *evmtypes.Head,
 		"err", err,
 	)
 
-	if sub != nil {
-		n.registerSub(sub)
-	}
 	return sub, err
 }
 
@@ -904,6 +904,9 @@ func (n *node) SubscribeFilterLogs(ctx context.Context, q ethereum.FilterQuery,
 	lggr.Debug("RPC call: evmclient.Client#SubscribeFilterLogs")
 	start := time.Now()
 	sub, err = n.ws.geth.SubscribeFilterLogs(ctx, q, ch)
+	if err == nil {
+		n.registerSub(sub)
+	}
 	err = n.wrapWS(err)
 	duration := time.Since(start)
 
@@ -913,10 +916,6 @@ func (n *node) SubscribeFilterLogs(ctx context.Context, q ethereum.FilterQuery,
 		"err", err,
 	)
 
-	if sub != nil {
-		n.registerSub(sub)
-	}
-
 	return
 }
 

core/chains/evm/client/node_lifecycle_test.go

@@ -12,6 +12,7 @@ import (
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
 
+	evmtypes "github.com/smartcontractkit/chainlink/core/chains/evm/types"
 	"github.com/smartcontractkit/chainlink/core/internal/testutils"
 	"github.com/smartcontractkit/chainlink/core/logger"
 )
@@ -157,10 +158,15 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) {
 		dial(t, n)
 		defer n.Close()
 
+		_, err := n.EthSubscribe(testutils.Context(t), make(chan *evmtypes.Head))
+		assert.Error(t, err)
+
 		n.wg.Add(1)
 		n.aliveLoop()
 
 		assert.Equal(t, NodeStateUnreachable, n.State())
+		// sc-39341: ensure failed EthSubscribe didn't register a (*rpc.ClientSubscription)(nil) which would lead to a panic on Unsubscribe
+		assert.Len(t, n.subs, 0)
 	})
 
 	t.Run("if remote RPC connection is closed transitions to unreachable", func(t *testing.T) {

core/services/fluxmonitorv2/validate.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/pelletier/go-toml"
 	"github.com/pkg/errors"
+
 	"github.com/smartcontractkit/chainlink/core/services/job"
 	"github.com/smartcontractkit/chainlink/core/store/models"
 	"github.com/smartcontractkit/chainlink/core/utils"
@@ -31,25 +32,7 @@ func ValidatedFluxMonitorSpec(config ValidationConfig, ts string) (job.Job, erro
 	}
 	err = tree.Unmarshal(&spec)
 	if err != nil {
-		var specIntThreshold job.FluxMonitorSpecIntThreshold
-		err = tree.Unmarshal(&specIntThreshold)
-		if err != nil {
-			return jb, err
-		}
-		spec = job.FluxMonitorSpec{
-			ContractAddress:     specIntThreshold.ContractAddress,
-			Threshold:           float32(specIntThreshold.Threshold),
-			AbsoluteThreshold:   float32(specIntThreshold.AbsoluteThreshold),
-			PollTimerPeriod:     specIntThreshold.PollTimerPeriod,
-			PollTimerDisabled:   specIntThreshold.PollTimerDisabled,
-			IdleTimerPeriod:     specIntThreshold.IdleTimerPeriod,
-			IdleTimerDisabled:   specIntThreshold.IdleTimerDisabled,
-			DrumbeatSchedule:    specIntThreshold.DrumbeatSchedule,
-			DrumbeatRandomDelay: specIntThreshold.DrumbeatRandomDelay,
-			DrumbeatEnabled:     specIntThreshold.DrumbeatEnabled,
-			MinPayment:          specIntThreshold.MinPayment,
-			EVMChainID:          specIntThreshold.EVMChainID,
-		}
+		return jb, err
 	}
 	jb.FluxMonitorSpec = &spec
 

core/services/fluxmonitorv2/validate_test.go

@@ -8,6 +8,8 @@ import (
 	"github.com/smartcontractkit/chainlink/core/assets"
 	"github.com/smartcontractkit/chainlink/core/services/job"
 	"github.com/smartcontractkit/chainlink/core/store/models"
+	"github.com/smartcontractkit/chainlink/core/utils/tomlutils"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -67,8 +69,8 @@ answer1 [type=median index=0];
 				assert.Equal(t, "fluxmonitor", j.Type.String())
 				assert.Equal(t, uint32(1), j.SchemaVersion)
 				assert.Equal(t, "0x3cCad4715152693fE3BC4460591e3D3Fbd071b42", j.FluxMonitorSpec.ContractAddress.String())
-				assert.Equal(t, float32(0.5), spec.Threshold)
-				assert.Equal(t, float32(0), spec.AbsoluteThreshold)
+				assert.Equal(t, tomlutils.Float32(0.5), spec.Threshold)
+				assert.Equal(t, tomlutils.Float32(0), spec.AbsoluteThreshold)
 				assert.Equal(t, true, spec.IdleTimerDisabled)
 				assert.Equal(t, 1*time.Minute, spec.PollTimerPeriod)
 				assert.Equal(t, false, spec.PollTimerDisabled)

core/services/job/models.go

@@ -24,6 +24,7 @@ import (
 	"github.com/smartcontractkit/chainlink/core/store/models"
 	"github.com/smartcontractkit/chainlink/core/utils"
 	"github.com/smartcontractkit/chainlink/core/utils/stringutils"
+	"github.com/smartcontractkit/chainlink/core/utils/tomlutils"
 )
 
 const (
@@ -372,34 +373,14 @@ func (s *CronSpec) SetID(value string) error {
 	return nil
 }
 
-// Need to also try integer thresholds until
-// https://github.com/pelletier/go-toml/issues/571 is addressed.
-// The UI's TOML.stringify({"threshold": 1.0}) (https://github.com/iarna/iarna-toml)
-// will return "threshold = 1" since ts/js doesn't know the
-// difference between 1.0 and 1, so we need to address it on the backend.
-type FluxMonitorSpecIntThreshold struct {
-	ContractAddress     ethkey.EIP55Address `toml:"contractAddress"`
-	Threshold           int                 `toml:"threshold"`
-	AbsoluteThreshold   int                 `toml:"absoluteThreshold"`
-	PollTimerPeriod     time.Duration
-	PollTimerDisabled   bool
-	IdleTimerPeriod     time.Duration
-	IdleTimerDisabled   bool
-	DrumbeatSchedule    string
-	DrumbeatRandomDelay time.Duration
-	DrumbeatEnabled     bool
-	MinPayment          *assets.Link
-	EVMChainID          *utils.Big `toml:"evmChainID"`
-}
-
 type FluxMonitorSpec struct {
 	ID              int32               `toml:"-"`
 	ContractAddress ethkey.EIP55Address `toml:"contractAddress"`
-	Threshold       float32             `toml:"threshold,float"`
+	Threshold       tomlutils.Float32   `toml:"threshold,float"`
 	// AbsoluteThreshold is the maximum absolute change allowed in a fluxmonitored
 	// value before a new round should be kicked off, so that the current value
 	// can be reported on-chain.
-	AbsoluteThreshold   float32 `toml:"absoluteThreshold,float"`
+	AbsoluteThreshold   tomlutils.Float32 `toml:"absoluteThreshold,float"`
 	PollTimerPeriod     time.Duration
 	PollTimerDisabled   bool
 	IdleTimerPeriod     time.Duration
@@ -435,7 +416,7 @@ type VRFSpec struct {
 	BatchFulfillmentEnabled bool `toml:"batchFulfillmentEnabled"`
 	// BatchFulfillmentGasMultiplier is used to determine the final gas estimate for the batch
 	// fulfillment.
-	BatchFulfillmentGasMultiplier float64 `toml:"batchFulfillmentGasMultiplier"`
+	BatchFulfillmentGasMultiplier tomlutils.Float64 `toml:"batchFulfillmentGasMultiplier"`
 
 	CoordinatorAddress       ethkey.EIP55Address   `toml:"coordinatorAddress"`
 	PublicKey                secp256k1.PublicKey   `toml:"publicKey"`

core/services/vrf/listener_v2_types.go

@@ -119,7 +119,7 @@ func (lsn *listenerV2) processBatch(
 	totalGasLimitBumped := batchFulfillmentGasEstimate(
 		uint64(len(batch.proofs)),
 		maxCallbackGasLimit,
-		lsn.job.VRFSpec.BatchFulfillmentGasMultiplier,
+		float64(lsn.job.VRFSpec.BatchFulfillmentGasMultiplier),
 	)
 	ll := l.With("numRequestsInBatch", len(batch.reqIDs),
 		"requestIDs", batch.reqIDs,

core/utils/tomlutils/toml.go

@@ -0,0 +1,35 @@
+package tomlutils
+
+import (
+	"strconv"
+)
+
+// Float32 represents float32 values for TOML
+type Float32 float32
+
+// UnmarshalText parses the value as a proper float32
+func (t *Float32) UnmarshalText(text []byte) error {
+	f32, err := strconv.ParseFloat(string(text), 32)
+	if err != nil {
+		return err
+	}
+
+	*t = Float32(f32)
+
+	return nil
+}
+
+// Float64 represents float64 values for TOML
+type Float64 float64
+
+// UnmarshalText parses the value as a proper float64
+func (t *Float64) UnmarshalText(text []byte) error {
+	f32, err := strconv.ParseFloat(string(text), 64)
+	if err != nil {
+		return err
+	}
+
+	*t = Float64(f32)
+
+	return nil
+}

core/utils/tomlutils/toml_test.go

@@ -0,0 +1,71 @@
+package tomlutils
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestUtils_TomlFloat32_Success_Decimal(t *testing.T) {
+	t.Parallel()
+
+	var tomlF32 Float32
+
+	err := tomlF32.UnmarshalText([]byte("0.23"))
+
+	assert.Nil(t, err)
+	assert.Equal(t, tomlF32, Float32(0.23))
+}
+
+func TestUtils_TomlFloat32_Success_Integer(t *testing.T) {
+	t.Parallel()
+
+	var tomlF32 Float32
+
+	err := tomlF32.UnmarshalText([]byte("13"))
+
+	assert.Nil(t, err)
+	assert.Equal(t, tomlF32, Float32(13))
+}
+
+func TestUtils_TomlFloat32_Failure(t *testing.T) {
+	t.Parallel()
+
+	var tomlF32 Float32
+
+	err := tomlF32.UnmarshalText([]byte("1s"))
+
+	assert.NotNil(t, err)
+}
+
+func TestUtils_TomlFloat64_Success_Decimal(t *testing.T) {
+	t.Parallel()
+
+	var tomlF64 Float64
+
+	err := tomlF64.UnmarshalText([]byte("2.82"))
+
+	assert.Nil(t, err)
+	assert.Equal(t, tomlF64, Float64(2.82))
+}
+
+func TestUtils_TomlFloat64_Success_Integer(t *testing.T) {
+	t.Parallel()
+
+	var tomlF64 Float64
+
+	err := tomlF64.UnmarshalText([]byte("3"))
+
+	assert.Nil(t, err)
+	assert.Equal(t, tomlF64, Float64(3))
+}
+
+func TestUtils_TomlFloat64_Failure(t *testing.T) {
+	t.Parallel()
+
+	var tomlF64 Float64
+
+	err := tomlF64.UnmarshalText([]byte("1s"))
+
+	assert.NotNil(t, err)
+}

core/web/jobs_controller_test.go

@@ -16,11 +16,12 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	p2ppeer "github.com/libp2p/go-libp2p-core/peer"
 	"github.com/pelletier/go-toml"
-	"github.com/smartcontractkit/sqlx"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/guregu/null.v4"
 
+	"github.com/smartcontractkit/sqlx"
+
 	"github.com/smartcontractkit/chainlink/core/internal/cltest"
 	"github.com/smartcontractkit/chainlink/core/internal/testutils"
 	"github.com/smartcontractkit/chainlink/core/services/directrequest"
@@ -29,6 +30,7 @@ import (
 	"github.com/smartcontractkit/chainlink/core/services/keystore/keys/p2pkey"
 	"github.com/smartcontractkit/chainlink/core/services/pg"
 	"github.com/smartcontractkit/chainlink/core/testdata/testspecs"
+	"github.com/smartcontractkit/chainlink/core/utils/tomlutils"
 	"github.com/smartcontractkit/chainlink/core/web"
 	"github.com/smartcontractkit/chainlink/core/web/presenters"
 )
@@ -301,8 +303,8 @@ func TestJobController_Create_HappyPath(t *testing.T) {
 				assert.Equal(t, ethkey.EIP55Address("0x3cCad4715152693fE3BC4460591e3D3Fbd071b42"), jb.FluxMonitorSpec.ContractAddress)
 				assert.Equal(t, time.Second, jb.FluxMonitorSpec.IdleTimerPeriod)
 				assert.Equal(t, false, jb.FluxMonitorSpec.IdleTimerDisabled)
-				assert.Equal(t, float32(0.5), jb.FluxMonitorSpec.Threshold)
-				assert.Equal(t, float32(0), jb.FluxMonitorSpec.AbsoluteThreshold)
+				assert.Equal(t, tomlutils.Float32(0.5), jb.FluxMonitorSpec.Threshold)
+				assert.Equal(t, tomlutils.Float32(0), jb.FluxMonitorSpec.AbsoluteThreshold)
 			},
 		},
 		{

core/web/presenters/job.go

@@ -100,8 +100,8 @@ func NewFluxMonitorSpec(spec *job.FluxMonitorSpec) *FluxMonitorSpec {
 	}
 	return &FluxMonitorSpec{
 		ContractAddress:     spec.ContractAddress,
-		Threshold:           spec.Threshold,
-		AbsoluteThreshold:   spec.AbsoluteThreshold,
+		Threshold:           float32(spec.Threshold),
+		AbsoluteThreshold:   float32(spec.AbsoluteThreshold),
 		PollTimerPeriod:     spec.PollTimerPeriod.String(),
 		PollTimerDisabled:   spec.PollTimerDisabled,
 		IdleTimerPeriod:     spec.IdleTimerPeriod.String(),

core/web/resolver/spec.go

@@ -669,7 +669,7 @@ func (r *VRFSpecResolver) BatchFulfillmentEnabled() bool {
 
 // BatchFulfillmentGasMultiplier resolves the spec's batch fulfillment gas multiplier.
 func (r *VRFSpecResolver) BatchFulfillmentGasMultiplier() float64 {
-	return r.spec.BatchFulfillmentGasMultiplier
+	return float64(r.spec.BatchFulfillmentGasMultiplier)
 }
 
 // ChunkSize resolves the spec's chunk size.

docs/CHANGELOG.md

@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ...
 
+## [1.4.1] - 2022-05-11
+
+### Fixed
+
+- Ensure failed EthSubscribe didn't register a (*rpc.ClientSubscription)(nil) which would lead to a panic on Unsubscribe
+- Fixes parsing of float values on job specs
+
 ## [1.4.0] - 2022-05-02
 
 ### Added