test vrf listener processes old requests on start up

core/services/vrf/v2/integration_helpers_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils"
 	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils/evmtest"
 	"github.com/smartcontractkit/chainlink/v2/core/services/chainlink"
+	"github.com/smartcontractkit/chainlink/v2/core/services/job"
 	"github.com/smartcontractkit/chainlink/v2/core/services/keystore/keys/ethkey"
 	v22 "github.com/smartcontractkit/chainlink/v2/core/services/vrf/v2"
 	"github.com/smartcontractkit/chainlink/v2/core/services/vrf/vrfcommon"
@@ -1736,3 +1737,145 @@ func testMaliciousConsumer(
 	}
 	require.Equal(t, 1, len(requests))
 }
+
+func testReplayOldRequestsOnStartUp(
+	t *testing.T,
+	ownerKey ethkey.KeyV2,
+	uni coordinatorV2UniverseCommon,
+	consumer *bind.TransactOpts,
+	consumerContract vrftesthelpers.VRFConsumerContract,
+	consumerContractAddress common.Address,
+	coordinator v22.CoordinatorV2_X,
+	coordinatorAddress common.Address,
+	batchCoordinatorAddress common.Address,
+	vrfOwnerAddress *common.Address,
+	vrfVersion vrfcommon.Version,
+	nativePayment bool,
+	assertions ...func(
+		t *testing.T,
+		coordinator v22.CoordinatorV2_X,
+		rwfe v22.RandomWordsFulfilled,
+		subID *big.Int),
+) {
+	sendingKey := cltest.MustGenerateRandomKey(t)
+	gasLanePriceWei := assets.GWei(10)
+	config, _ := heavyweight.FullTestDBV2(t, func(c *chainlink.Config, s *chainlink.Secrets) {
+		simulatedOverrides(t, assets.GWei(10), toml.KeySpecific{
+			// Gas lane.
+			Key:          ptr(sendingKey.EIP55Address),
+			GasEstimator: toml.KeySpecificGasEstimator{PriceMax: gasLanePriceWei},
+		})(c, s)
+		c.EVM[0].MinIncomingConfirmations = ptr[uint32](2)
+		c.Feature.LogPoller = ptr(true)
+		c.EVM[0].LogPollInterval = commonconfig.MustNewDuration(1 * time.Second)
+	})
+	app := cltest.NewApplicationWithConfigV2AndKeyOnSimulatedBlockchain(t, config, uni.backend, ownerKey, sendingKey)
+
+	// Create a subscription and fund with 5 LINK.
+	subID := subscribeAndAssertSubscriptionCreatedEvent(t, consumerContract, consumer, consumerContractAddress, big.NewInt(5e18), coordinator, uni.backend, nativePayment)
+
+	// Fund gas lanes.
+	sendEth(t, ownerKey, uni.backend, sendingKey.Address, 10)
+	require.NoError(t, app.Start(testutils.Context(t)))
+
+	// Create VRF Key, register it to coordinator and export
+	vrfkey, err := app.GetKeyStore().VRF().Create()
+	require.NoError(t, err)
+	registerProvingKeyHelper(t, uni, coordinator, vrfkey, &defaultMaxGasPrice)
+	keyHash := vrfkey.PublicKey.MustHash()
+
+	encodedVrfKey, err := app.GetKeyStore().VRF().Export(vrfkey.ID(), testutils.Password)
+	require.NoError(t, err)
+
+	// Shut down the node before making the randomness request
+	require.NoError(t, app.Stop())
+
+	// Make the first randomness request.
+	numWords := uint32(20)
+	requestID1, _ := requestRandomnessAndAssertRandomWordsRequestedEvent(t, consumerContract, consumer, keyHash, subID, numWords, 500_000, coordinator, uni.backend, nativePayment)
+
+	// number of blocks to mine before restarting the node
+	nBlocks := 100
+	for i := 0; i < nBlocks; i++ {
+		uni.backend.Commit()
+	}
+
+	config, db := heavyweight.FullTestDBV2(t, func(c *chainlink.Config, s *chainlink.Secrets) {
+		simulatedOverrides(t, assets.GWei(10), toml.KeySpecific{
+			// Gas lane.
+			Key:          ptr(sendingKey.EIP55Address),
+			GasEstimator: toml.KeySpecificGasEstimator{PriceMax: gasLanePriceWei},
+		})(c, s)
+		c.EVM[0].MinIncomingConfirmations = ptr[uint32](2)
+		c.Feature.LogPoller = ptr(true)
+		c.EVM[0].LogPollInterval = commonconfig.MustNewDuration(1 * time.Second)
+	})
+
+	// Start a new app and create VRF job using the same VRF key created above
+	app = cltest.NewApplicationWithConfigV2AndKeyOnSimulatedBlockchain(t, config, uni.backend, ownerKey, sendingKey)
+
+	require.NoError(t, app.Start(testutils.Context(t)))
+
+	vrfKey, err := app.GetKeyStore().VRF().Import(encodedVrfKey, testutils.Password)
+	require.NoError(t, err)
+
+	incomingConfs := 2
+	var vrfOwnerString string
+	if vrfOwnerAddress != nil {
+		vrfOwnerString = vrfOwnerAddress.Hex()
+	}
+
+	spec := testspecs.GenerateVRFSpec(testspecs.VRFSpecParams{
+		Name:                     "vrf-primary",
+		VRFVersion:               vrfVersion,
+		CoordinatorAddress:       coordinatorAddress.Hex(),
+		BatchCoordinatorAddress:  batchCoordinatorAddress.Hex(),
+		MinIncomingConfirmations: incomingConfs,
+		PublicKey:                vrfKey.PublicKey.String(),
+		FromAddresses:            []string{sendingKey.Address.String()},
+		BackoffInitialDelay:      10 * time.Millisecond,
+		BackoffMaxDelay:          time.Second,
+		V2:                       true,
+		GasLanePrice:             gasLanePriceWei,
+		VRFOwnerAddress:          vrfOwnerString,
+		EVMChainID:               testutils.SimulatedChainID.String(),
+	}).Toml()
+
+	jb, err := vrfcommon.ValidatedVRFSpec(spec)
+	require.NoError(t, err)
+	t.Log(jb.VRFSpec.PublicKey.MustHash(), vrfKey.PublicKey.MustHash())
+	err = app.JobSpawner().CreateJob(&jb)
+	require.NoError(t, err)
+
+	// Wait until all jobs are active and listening for logs
+	gomega.NewWithT(t).Eventually(func() bool {
+		jbs := app.JobSpawner().ActiveJobs()
+		for _, jb := range jbs {
+			if jb.Type == job.VRF {
+				return true
+			}
+		}
+		return false
+	}, testutils.WaitTimeout(t), 100*time.Millisecond).Should(gomega.BeTrue())
+
+	// Wait for fulfillment to be queued.
+	gomega.NewGomegaWithT(t).Eventually(func() bool {
+		uni.backend.Commit()
+		runs, err := app.PipelineORM().GetAllRuns()
+		require.NoError(t, err)
+		t.Log("runs", len(runs))
+		return len(runs) == 1
+	}, testutils.WaitTimeout(t), time.Second).Should(gomega.BeTrue())
+
+	// Mine the fulfillment that was queued.
+	mine(t, requestID1, subID, uni.backend, db, vrfVersion, testutils.SimulatedChainID)
+
+	// Assert correct state of RandomWordsFulfilled event.
+	// In particular:
+	// * success should be true
+	// * payment should be exactly the amount specified as the premium in the coordinator fee config
+	rwfe := assertRandomWordsFulfilled(t, requestID1, true, coordinator, nativePayment)
+	if len(assertions) > 0 {
+		assertions[0](t, coordinator, rwfe, subID)
+	}
+}

core/services/vrf/v2/integration_v2_plus_test.go

@@ -1346,3 +1346,24 @@ func TestVRFV2PlusIntegration_CancelSubscription(t *testing.T) {
 	AssertLinkBalance(t, uni.linkContract, uni.neil.From, linkBalanceBeforeCancel.Add(linkBalanceBeforeCancel, linkAmount))
 	AssertNativeBalance(t, uni.backend, uni.neil.From, nativeBalanceBeforeCancel.Add(nativeBalanceBeforeCancel, nativeAmount))
 }
+
+func TestVRFV2PlusIntegration_ReplayOldRequestsOnStartUp(t *testing.T) {
+	t.Parallel()
+	ownerKey := cltest.MustGenerateRandomKey(t)
+	uni := newVRFCoordinatorV2PlusUniverse(t, ownerKey, 1, false)
+
+	testReplayOldRequestsOnStartUp(
+		t,
+		ownerKey,
+		uni.coordinatorV2UniverseCommon,
+		uni.vrfConsumers[0],
+		uni.consumerContracts[0],
+		uni.consumerContractAddresses[0],
+		uni.rootContract,
+		uni.rootContractAddress,
+		uni.batchCoordinatorContractAddress,
+		nil,
+		vrfcommon.V2Plus,
+		false,
+	)
+}

core/services/vrf/v2/integration_v2_test.go

@@ -2231,6 +2231,27 @@ func TestStartingCountsV1(t *testing.T) {
 	assert.Equal(t, uint64(2), countsV2[big.NewInt(0x12).String()])
 }
 
+func TestVRFV2Integration_ReplayOldRequestsOnStartUp(t *testing.T) {
+	t.Parallel()
+	ownerKey := cltest.MustGenerateRandomKey(t)
+	uni := newVRFCoordinatorV2Universe(t, ownerKey, 1)
+
+	testReplayOldRequestsOnStartUp(
+		t,
+		ownerKey,
+		uni.coordinatorV2UniverseCommon,
+		uni.vrfConsumers[0],
+		uni.consumerContracts[0],
+		uni.consumerContractAddresses[0],
+		uni.rootContract,
+		uni.rootContractAddress,
+		uni.batchCoordinatorContractAddress,
+		nil,
+		vrfcommon.V2,
+		false,
+	)
+}
+
 func FindLatestRandomnessRequestedLog(t *testing.T,
 	coordContract v22.CoordinatorV2_X,
 	keyHash [32]byte,