Merge pull request #679 from keep-network/pipe-it-all

tBTC chain extension: Incporporating the new Ethereum event subscription API and background event pull loop

cmd/start.go

@@ -216,14 +216,7 @@ func initializeExtensions(
 			return
 		}
 
-		err = tbtc.Initialize(ctx, tbtcEthereumChain)
-		if err != nil {
-			logger.Errorf(
-				"could not initialize tbtc extension: [%v]",
-				err,
-			)
-			return
-		}
+		tbtc.Initialize(ctx, tbtcEthereumChain)
 	}
 }
 

go.mod

@@ -18,9 +18,9 @@ require (
 	github.com/gogo/protobuf v1.3.1
 	github.com/google/gofuzz v1.1.0
 	github.com/ipfs/go-log v1.0.4
-	github.com/keep-network/keep-common v1.3.1-0.20210128105431-b3b56f6f88c8
+	github.com/keep-network/keep-common v1.3.1-0.20210128152700-34905d2fe019
 	github.com/keep-network/keep-core v1.3.2-0.20201229154408-59ac640ed0cb
-	github.com/keep-network/tbtc v1.1.1-0.20201117095624-38508bdb562e
+	github.com/keep-network/tbtc v1.1.1-0.20210128164215-c03b8cf351f0
 	github.com/pkg/errors v0.9.1
 	github.com/urfave/cli v1.22.1
 )

go.sum

@@ -331,14 +331,13 @@ github.com/keep-network/cli v1.20.0 h1:mEufpPsovOVdduTTkk+a2CxS3crIrGFqUsvs58gSi
 github.com/keep-network/cli v1.20.0/go.mod h1:nzsst4JjU+rGE8Q5J839fYxectxWHpLhxKNohQWtQhA=
 github.com/keep-network/go-libp2p-bootstrap v0.0.0-20200423153828-ed815bc50aec h1:2pXAsi4OUUjZKr5ds5UOF2IxXN+jVW0WetVO+czkf+A=
 github.com/keep-network/go-libp2p-bootstrap v0.0.0-20200423153828-ed815bc50aec/go.mod h1:xR8jf3/VJAjh3nWu5tFe8Yxnt2HvWsqZHfGef1P5oDk=
-github.com/keep-network/keep-common v1.2.1-0.20201116151638-8af057907255/go.mod h1:emxogTbBdey7M3jOzfxZOdfn139kN2mI2b2wA6AHKKo=
 github.com/keep-network/keep-common v1.3.0/go.mod h1:emxogTbBdey7M3jOzfxZOdfn139kN2mI2b2wA6AHKKo=
-github.com/keep-network/keep-common v1.3.1-0.20210128105431-b3b56f6f88c8 h1:yo2Gj5Wwc+oU+IeGlF0shiVMth4C3gfQzQXUowLj5MQ=
-github.com/keep-network/keep-common v1.3.1-0.20210128105431-b3b56f6f88c8/go.mod h1:emxogTbBdey7M3jOzfxZOdfn139kN2mI2b2wA6AHKKo=
+github.com/keep-network/keep-common v1.3.1-0.20210128152700-34905d2fe019 h1:WBq6cP/N3PFPo25k6AVuM1Hh46GL8DRPr2YzV97quuc=
+github.com/keep-network/keep-common v1.3.1-0.20210128152700-34905d2fe019/go.mod h1:emxogTbBdey7M3jOzfxZOdfn139kN2mI2b2wA6AHKKo=
 github.com/keep-network/keep-core v1.3.2-0.20201229154408-59ac640ed0cb h1:pDhLagUiOYWOn72UiG3jSDVDXZ4dYlsfRa0ToF/ru18=
 github.com/keep-network/keep-core v1.3.2-0.20201229154408-59ac640ed0cb/go.mod h1:4KezOJWc//c5lbtAE1ob7qYE/Gs5+a1QTjGBzzv2tu8=
-github.com/keep-network/tbtc v1.1.1-0.20201117095624-38508bdb562e h1:go3irX4olJUZOi59gaBX7iYIFFbJEV5/zbfCG4tOPaQ=
-github.com/keep-network/tbtc v1.1.1-0.20201117095624-38508bdb562e/go.mod h1:LS1zQDXTJrP5f6AHFQKw6/iJPdfV+WqUCfh6GTfC6Rs=
+github.com/keep-network/tbtc v1.1.1-0.20210128164215-c03b8cf351f0 h1:zXW/085lHEs1NppQKZ+oOSnOWGxT4JlYelnChNML4WU=
+github.com/keep-network/tbtc v1.1.1-0.20210128164215-c03b8cf351f0/go.mod h1:lIFIL8HRHTibQVzE0GoWdGZVfoc19KuJUbhk4W4XNT0=
 github.com/keep-network/toml v0.3.0 h1:G+NJwWR/ZiORqeLBsDXDchYoL29PXHdxOPcCueA7ctE=
 github.com/keep-network/toml v0.3.0/go.mod h1:Zeyd3lxbIlMYLREho3UK1dMP2xjqt2gLkQ5E5vM6K38=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=

pkg/chain/ethereum/tbtc.go

@@ -36,6 +36,7 @@ func WithTBTCExtension(
 		ethereumChain.client,
 		ethereumChain.nonceManager,
 		ethereumChain.miningWaiter,
+		ethereumChain.blockCounter,
 		ethereumChain.transactionMutex,
 	)
 	if err != nil {
@@ -52,129 +53,109 @@ func WithTBTCExtension(
 // on-chain notification of a new deposit creation is seen.
 func (tec *TBTCEthereumChain) OnDepositCreated(
 	handler func(depositAddress string),
-) (subscription.EventSubscription, error) {
-	return tec.tbtcSystemContract.WatchCreated(
-		func(
-			DepositContractAddress common.Address,
-			KeepAddress common.Address,
-			Timestamp *big.Int,
-			blockNumber uint64,
-		) {
-			handler(DepositContractAddress.Hex())
-		},
-		func(err error) error {
-			return fmt.Errorf("watch deposit created failed: [%v]", err)
-		},
+) subscription.EventSubscription {
+	onEvent := func(
+		DepositContractAddress common.Address,
+		KeepAddress common.Address,
+		Timestamp *big.Int,
+		blockNumber uint64,
+	) {
+		handler(DepositContractAddress.Hex())
+	}
+
+	return tec.tbtcSystemContract.Created(
 		nil,
 		nil,
-	)
+		nil,
+	).OnEvent(onEvent)
 }
 
 // OnDepositRegisteredPubkey installs a callback that is invoked when an
 // on-chain notification of a deposit's pubkey registration is seen.
 func (tec *TBTCEthereumChain) OnDepositRegisteredPubkey(
 	handler func(depositAddress string),
-) (subscription.EventSubscription, error) {
-	return tec.tbtcSystemContract.WatchRegisteredPubkey(
-		func(
-			DepositContractAddress common.Address,
-			SigningGroupPubkeyX [32]uint8,
-			SigningGroupPubkeyY [32]uint8,
-			Timestamp *big.Int,
-			blockNumber uint64,
-		) {
-			handler(DepositContractAddress.Hex())
-		},
-		func(err error) error {
-			return fmt.Errorf(
-				"watch deposit registered pubkey failed: [%v]",
-				err,
-			)
-		},
-		nil,
-	)
+) subscription.EventSubscription {
+	onEvent := func(
+		DepositContractAddress common.Address,
+		SigningGroupPubkeyX [32]uint8,
+		SigningGroupPubkeyY [32]uint8,
+		Timestamp *big.Int,
+		blockNumber uint64,
+	) {
+		handler(DepositContractAddress.Hex())
+	}
+
+	return tec.tbtcSystemContract.RegisteredPubkey(nil, nil).OnEvent(onEvent)
 }
 
 // OnDepositRedemptionRequested installs a callback that is invoked when an
 // on-chain notification of a deposit redemption request is seen.
 func (tec *TBTCEthereumChain) OnDepositRedemptionRequested(
 	handler func(depositAddress string),
-) (subscription.EventSubscription, error) {
-	return tec.tbtcSystemContract.WatchRedemptionRequested(
-		func(
-			DepositContractAddress common.Address,
-			Requester common.Address,
-			Digest [32]uint8,
-			UtxoValue *big.Int,
-			RedeemerOutputScript []uint8,
-			RequestedFee *big.Int,
-			Outpoint []uint8,
-			blockNumber uint64,
-		) {
-			handler(DepositContractAddress.Hex())
-		},
-		func(err error) error {
-			return fmt.Errorf(
-				"watch deposit redemption requested failed: [%v]",
-				err,
-			)
-		},
+) subscription.EventSubscription {
+	onEvent := func(
+		DepositContractAddress common.Address,
+		Requester common.Address,
+		Digest [32]uint8,
+		UtxoValue *big.Int,
+		RedeemerOutputScript []uint8,
+		RequestedFee *big.Int,
+		Outpoint []uint8,
+		blockNumber uint64,
+	) {
+		handler(DepositContractAddress.Hex())
+	}
+
+	return tec.tbtcSystemContract.RedemptionRequested(
 		nil,
 		nil,
 		nil,
-	)
+		nil,
+	).OnEvent(onEvent)
 }
 
 // OnDepositGotRedemptionSignature installs a callback that is invoked when an
 // on-chain notification of a deposit receiving a redemption signature is seen.
 func (tec *TBTCEthereumChain) OnDepositGotRedemptionSignature(
 	handler func(depositAddress string),
-) (subscription.EventSubscription, error) {
-	return tec.tbtcSystemContract.WatchGotRedemptionSignature(
-		func(
-			DepositContractAddress common.Address,
-			Digest [32]uint8,
-			R [32]uint8,
-			S [32]uint8,
-			Timestamp *big.Int,
-			blockNumber uint64,
-		) {
-			handler(DepositContractAddress.Hex())
-		},
-		func(err error) error {
-			return fmt.Errorf(
-				"watch deposit got redemption signature failed: [%v]",
-				err,
-			)
-		},
+) subscription.EventSubscription {
+	onEvent := func(
+		DepositContractAddress common.Address,
+		Digest [32]uint8,
+		R [32]uint8,
+		S [32]uint8,
+		Timestamp *big.Int,
+		blockNumber uint64,
+	) {
+		handler(DepositContractAddress.Hex())
+	}
+
+	return tec.tbtcSystemContract.GotRedemptionSignature(
 		nil,
 		nil,
-	)
+		nil,
+	).OnEvent(onEvent)
 }
 
 // OnDepositRedeemed installs a callback that is invoked when an
 // on-chain notification of a deposit redemption is seen.
 func (tec *TBTCEthereumChain) OnDepositRedeemed(
 	handler func(depositAddress string),
-) (subscription.EventSubscription, error) {
-	return tec.tbtcSystemContract.WatchRedeemed(
-		func(
-			DepositContractAddress common.Address,
-			Txid [32]uint8,
-			Timestamp *big.Int,
-			blockNumber uint64,
-		) {
-			handler(DepositContractAddress.Hex())
-		},
-		func(err error) error {
-			return fmt.Errorf(
-				"watch deposit redeemed failed: [%v]",
-				err,
-			)
-		},
+) subscription.EventSubscription {
+	onEvent := func(
+		DepositContractAddress common.Address,
+		Txid [32]uint8,
+		Timestamp *big.Int,
+		blockNumber uint64,
+	) {
+		handler(DepositContractAddress.Hex())
+	}
+
+	return tec.tbtcSystemContract.Redeemed(
 		nil,
 		nil,
-	)
+		nil,
+	).OnEvent(onEvent)
 }
 
 // PastDepositRedemptionRequestedEvents returns all redemption requested
@@ -384,6 +365,7 @@ func (tec *TBTCEthereumChain) getDepositContract(
 		tec.client,
 		tec.nonceManager,
 		tec.miningWaiter,
+		tec.blockCounter,
 		tec.transactionMutex,
 	)
 	if err != nil {