diff --git a/cmd/maintainer.go b/cmd/maintainer.go index b8d81e2aa5..d67c243f7a 100644 --- a/cmd/maintainer.go +++ b/cmd/maintainer.go @@ -6,10 +6,14 @@ import ( "github.com/spf13/cobra" + "github.com/keep-network/keep-core/build" "github.com/keep-network/keep-core/config" "github.com/keep-network/keep-core/pkg/bitcoin/electrum" + "github.com/keep-network/keep-core/pkg/chain" "github.com/keep-network/keep-core/pkg/chain/ethereum" + "github.com/keep-network/keep-core/pkg/clientinfo" "github.com/keep-network/keep-core/pkg/maintainer" + "github.com/keep-network/keep-core/pkg/maintainer/spv" ) // MaintainerCommand contains the definition of the maintainer command-line @@ -61,7 +65,7 @@ func maintainers(cmd *cobra.Command, args []string) error { ) } - _, tbtcChain, _, _, _, err := ethereum.Connect( + _, tbtcChain, blockCounter, _, _, err := ethereum.Connect( ctx, clientConfig.Ethereum, ) @@ -72,14 +76,51 @@ func maintainers(cmd *cobra.Command, args []string) error { ) } + metricsRecorder := initializeMaintainerMetrics(ctx, blockCounter) + maintainer.Initialize( ctx, clientConfig.Maintainer, btcChain, btcDiffChain, tbtcChain, + metricsRecorder, ) <-ctx.Done() return fmt.Errorf("unexpected context cancellation") } + +// initializeMaintainerMetrics sets up the client info registry and performance +// metrics for the maintainer command. It returns a metrics recorder wired to +// the SPV maintainer, or nil when the client info endpoint is not configured +// (in which case metrics recording is disabled). +func initializeMaintainerMetrics( + ctx context.Context, + blockCounter chain.BlockCounter, +) spv.MetricsRecorder { + registry, isConfigured := clientinfo.Initialize( + ctx, + clientConfig.ClientInfo.Port, + ) + if !isConfigured { + logger.Infof("client info endpoint not configured") + return nil + } + + perfMetrics := clientinfo.NewPerformanceMetrics(ctx, registry) + + registry.RegisterMetricClientInfo(build.Version) + registry.ObserveEthConnectivity( + blockCounter, + clientConfig.ClientInfo.EthereumMetricsTick, + ) + registry.RegisterEthChainInfoSource(blockCounter) + + logger.Infof( + "enabled client info endpoint on port [%v]", + clientConfig.ClientInfo.Port, + ) + + return perfMetrics +} diff --git a/cmd/maintainer_test.go b/cmd/maintainer_test.go new file mode 100644 index 0000000000..2fb664fc12 --- /dev/null +++ b/cmd/maintainer_test.go @@ -0,0 +1,32 @@ +package cmd + +import ( + "context" + "testing" +) + +// TestInitializeMaintainerMetricsDisabledWhenPortUnset verifies the metrics +// on/off gate: when the client info port is 0 (unset), the maintainer boot path +// must report the endpoint as not configured and return a genuinely nil +// recorder, leaving SPV metrics recording disabled. This is the sole production +// switch that turns the SPV proof-submission counters on or off, so inverting +// the gate would silently change maintainer behavior. +func TestInitializeMaintainerMetricsDisabledWhenPortUnset(t *testing.T) { + // clientConfig is a package-level global; restore it so the test does not + // leak state into other tests in this package. + originalPort := clientConfig.ClientInfo.Port + defer func() { clientConfig.ClientInfo.Port = originalPort }() + + clientConfig.ClientInfo.Port = 0 + + // The block counter is only touched on the configured (enabled) path, so a + // nil value is safe for the disabled path under test. + recorder := initializeMaintainerMetrics(context.Background(), nil) + + if recorder != nil { + t.Errorf( + "expected a nil recorder when the client info port is unset, got [%v]", + recorder, + ) + } +} diff --git a/cmd/maintainercli.go b/cmd/maintainercli.go index e674375fc3..87cda788bf 100644 --- a/cmd/maintainercli.go +++ b/cmd/maintainercli.go @@ -354,6 +354,7 @@ var submitDepositSweepProofCommand = cobra.Command{ requiredConfirmations, btcChain, tbtcChain, + nil, ); err != nil { return fmt.Errorf("failed to submit deposit sweep proof [%v]", err) } @@ -444,6 +445,7 @@ var submitRedemptionProofCommand = cobra.Command{ requiredConfirmations, btcChain, tbtcChain, + nil, ); err != nil { return fmt.Errorf("failed to submit redemption proof [%v]", err) } diff --git a/config/category.go b/config/category.go index f6b3f2ab0c..3fadf3ab35 100644 --- a/config/category.go +++ b/config/category.go @@ -30,6 +30,7 @@ var StartCmdCategories = []Category{ var MaintainerCategories = []Category{ Ethereum, BitcoinElectrum, + ClientInfo, Maintainer, } diff --git a/docs/performance-metrics.adoc b/docs/performance-metrics.adoc index af2a7132bc..cbb5686f11 100644 --- a/docs/performance-metrics.adoc +++ b/docs/performance-metrics.adoc @@ -197,6 +197,42 @@ For each action type, the following metrics are available: *Description*: Total count of coordination duration samples *Labels*: None +=== SPV Proof Submission Metrics (Maintainer) + +These metrics are recorded by the SPV maintainer (the `maintainer` command) when +it submits Bitcoin SPV proofs to the host chain. They require the client info +endpoint to be configured. + +==== `performance_deposit_sweep_proof_submissions_total` +*Type*: Counter +*Description*: Total number of deposit sweep proof submissions attempted +*Labels*: None + +==== `performance_deposit_sweep_proof_submissions_success_total` +*Type*: Counter +*Description*: Total number of successful deposit sweep proof submissions +*Labels*: None + +==== `performance_deposit_sweep_proof_submissions_failed_total` +*Type*: Counter +*Description*: Total number of failed deposit sweep proof submissions +*Labels*: None + +==== `performance_redemption_proof_submissions_total` +*Type*: Counter +*Description*: Total number of redemption proof submissions attempted +*Labels*: None + +==== `performance_redemption_proof_submissions_success_total` +*Type*: Counter +*Description*: Total number of successful redemption proof submissions +*Labels*: None + +==== `performance_redemption_proof_submissions_failed_total` +*Type*: Counter +*Description*: Total number of failed redemption proof submissions +*Labels*: None + === Network Metrics ==== `performance_incoming_message_queue_size` diff --git a/pkg/beacon/dkg/marshalling.go b/pkg/beacon/dkg/marshaling.go similarity index 100% rename from pkg/beacon/dkg/marshalling.go rename to pkg/beacon/dkg/marshaling.go diff --git a/pkg/beacon/dkg/marshalling_test.go b/pkg/beacon/dkg/marshaling_test.go similarity index 100% rename from pkg/beacon/dkg/marshalling_test.go rename to pkg/beacon/dkg/marshaling_test.go diff --git a/pkg/beacon/dkg/result/marshalling.go b/pkg/beacon/dkg/result/marshaling.go similarity index 100% rename from pkg/beacon/dkg/result/marshalling.go rename to pkg/beacon/dkg/result/marshaling.go diff --git a/pkg/beacon/dkg/result/marshalling_test.go b/pkg/beacon/dkg/result/marshaling_test.go similarity index 100% rename from pkg/beacon/dkg/result/marshalling_test.go rename to pkg/beacon/dkg/result/marshaling_test.go diff --git a/pkg/beacon/gjkr/member.go b/pkg/beacon/gjkr/member.go index 55cf54c1d9..95f8e81c2a 100644 --- a/pkg/beacon/gjkr/member.go +++ b/pkg/beacon/gjkr/member.go @@ -100,10 +100,9 @@ type CommitmentsVerifyingMember struct { // Shares calculated for the current member by peer group members which passed // the validation. // - // receivedQualifiedSharesS are defined as `s_ji` and receivedQualifiedSharesT are - // defined as `t_ji` across the protocol specification. - // TODO remove receivedQualifiedSharesT - exists only for unit tests purpose - receivedQualifiedSharesS, receivedQualifiedSharesT map[group.MemberIndex]*big.Int + // receivedQualifiedSharesS are defined as `s_ji` across the protocol + // specification. + receivedQualifiedSharesS map[group.MemberIndex]*big.Int // Commitments to secret shares polynomial coefficients received from // other group members. receivedPeerCommitments map[group.MemberIndex][]*bn256.G1 @@ -285,7 +284,6 @@ func (cm *CommittingMember) InitializeCommitmentsVerification() *CommitmentsVeri return &CommitmentsVerifyingMember{ CommittingMember: cm, receivedQualifiedSharesS: make(map[group.MemberIndex]*big.Int), - receivedQualifiedSharesT: make(map[group.MemberIndex]*big.Int), receivedPeerCommitments: make(map[group.MemberIndex][]*bn256.G1), } } diff --git a/pkg/beacon/gjkr/protocol.go b/pkg/beacon/gjkr/protocol.go index 8cedfbcd0b..148d7a369b 100644 --- a/pkg/beacon/gjkr/protocol.go +++ b/pkg/beacon/gjkr/protocol.go @@ -433,7 +433,6 @@ func (cvm *CommitmentsVerifyingMember) VerifyReceivedSharesAndCommitmentsMessage break } cvm.receivedQualifiedSharesS[commitmentsMessage.senderID] = shareS - cvm.receivedQualifiedSharesT[commitmentsMessage.senderID] = shareT break } } @@ -766,7 +765,6 @@ func (sjm *SharesJustifyingMember) discardReceivedShares( memberID group.MemberIndex, ) { delete(sjm.receivedQualifiedSharesS, memberID) - delete(sjm.receivedQualifiedSharesT, memberID) } // Inspects evidence log looking for ephemeral public key message sent in phase diff --git a/pkg/beacon/gjkr/protocol_accusations_test.go b/pkg/beacon/gjkr/protocol_accusations_test.go index 016119dad8..eec68f9267 100644 --- a/pkg/beacon/gjkr/protocol_accusations_test.go +++ b/pkg/beacon/gjkr/protocol_accusations_test.go @@ -134,7 +134,7 @@ func TestResolveSecretSharesAccusations(t *testing.T) { } for testName, test := range tests { t.Run(testName, func(t *testing.T) { - members, err := initializeSharesJustifyingMemberGroup( + members, receivedSharesT, err := initializeSharesJustifyingMemberGroup( dishonestThreshold, groupSize, ) @@ -145,7 +145,7 @@ func TestResolveSecretSharesAccusations(t *testing.T) { accuser := findSharesJustifyingMemberByID(members, test.accuserID) modifiedShareS := accuser.receivedQualifiedSharesS[test.accusedID] - modifiedShareT := accuser.receivedQualifiedSharesT[test.accusedID] + modifiedShareT := receivedSharesT[test.accuserID][test.accusedID] if test.modifyShareS != nil { modifiedShareS = test.modifyShareS(modifiedShareS) @@ -422,7 +422,7 @@ func TestResolveSecretSharesAccusationsIncorrectAccussedMemberId(t *testing.T) { for testName, test := range tests { t.Run(testName, func(t *testing.T) { - members, err := initializeSharesJustifyingMemberGroup( + members, _, err := initializeSharesJustifyingMemberGroup( dishonestThreshold, groupSize, ) @@ -540,19 +540,20 @@ func findCoefficientsJustifyingMemberByID( return nil } -// InitializeSharesJustifyingMemberGroup generates a group of members and simulates -// shares calculation and commitments sharing betwen members (Phases 3 and 4). -// It generates coefficients for each group member, calculates commitments and -// shares for each peer member individually. At the end it stores values for each -// member just like they would be received from peers. +// initializeSharesJustifyingMemberGroup initializes a group of shares +// justifying members with simulated received shares and commitments. It also +// returns the received `t_ji` shares keyed by receiver then sender member +// index; these are not stored on the member (only `s_ji` is production state) +// but some accusation tests need them to reconstruct the peer shares message. func initializeSharesJustifyingMemberGroup(dishonestThreshold, groupSize int) ( []*SharesJustifyingMember, + map[group.MemberIndex]map[group.MemberIndex]*big.Int, error, ) { commitmentsVerifyingMembers, err := initializeCommitmentsVerifiyingMembersGroup(dishonestThreshold, groupSize) if err != nil { - return nil, fmt.Errorf("group initialization failed [%s]", err) + return nil, nil, fmt.Errorf("group initialization failed [%s]", err) } var sharesJustifyingMembers []*SharesJustifyingMember @@ -567,14 +568,18 @@ func initializeSharesJustifyingMemberGroup(dishonestThreshold, groupSize int) ( groupCoefficientsB := make(map[group.MemberIndex][]*big.Int, groupSize) groupCommitments := make(map[group.MemberIndex][]*bn256.G1, groupSize) + // receivedSharesT keeps the `t_ji` shares received by each member from its + // peers, keyed by receiver then sender member index. + receivedSharesT := make(map[group.MemberIndex]map[group.MemberIndex]*big.Int) + for _, m := range sharesJustifyingMembers { memberCoefficientsA, err := generatePolynomial(dishonestThreshold) if err != nil { - return nil, fmt.Errorf("polynomial generation failed [%s]", err) + return nil, nil, fmt.Errorf("polynomial generation failed [%s]", err) } memberCoefficientsB, err := generatePolynomial(dishonestThreshold) if err != nil { - return nil, fmt.Errorf("polynomial generation failed [%s]", err) + return nil, nil, fmt.Errorf("polynomial generation failed [%s]", err) } // polynomial is of degree dishonestThreshold so it has @@ -598,13 +603,16 @@ func initializeSharesJustifyingMemberGroup(dishonestThreshold, groupSize int) ( for _, p := range sharesJustifyingMembers { if m.ID != p.ID { p.receivedQualifiedSharesS[m.ID] = m.evaluateMemberShare(p.ID, groupCoefficientsA[m.ID]) - p.receivedQualifiedSharesT[m.ID] = m.evaluateMemberShare(p.ID, groupCoefficientsB[m.ID]) + if receivedSharesT[p.ID] == nil { + receivedSharesT[p.ID] = make(map[group.MemberIndex]*big.Int) + } + receivedSharesT[p.ID][m.ID] = m.evaluateMemberShare(p.ID, groupCoefficientsB[m.ID]) p.receivedPeerCommitments[m.ID] = groupCommitments[m.ID] } } } - return sharesJustifyingMembers, nil + return sharesJustifyingMembers, receivedSharesT, nil } // initializePointsJustifyingMemberGroup generates a group of members and diff --git a/pkg/beacon/gjkr/protocol_commitments_test.go b/pkg/beacon/gjkr/protocol_commitments_test.go index 14b144be3c..dd0d837ad7 100644 --- a/pkg/beacon/gjkr/protocol_commitments_test.go +++ b/pkg/beacon/gjkr/protocol_commitments_test.go @@ -277,12 +277,6 @@ func assertValidSharesAndCommitments( len(verifyingMember.receivedQualifiedSharesS), ) } - if len(verifyingMember.receivedQualifiedSharesT) != expectedReceivedSharesLength { - t.Errorf("\nexpected: %v received shares T\nactual: %v\n", - expectedReceivedSharesLength, - len(verifyingMember.receivedQualifiedSharesT), - ) - } if len(verifyingMember.receivedPeerCommitments) != groupSize-1 { t.Errorf("\nexpected: %v received commitments\nactual: %v\n", expectedReceivedSharesLength, diff --git a/pkg/beacon/gjkr/protocol_sharing_test.go b/pkg/beacon/gjkr/protocol_sharing_test.go index 51f38096c3..39f72ddd90 100644 --- a/pkg/beacon/gjkr/protocol_sharing_test.go +++ b/pkg/beacon/gjkr/protocol_sharing_test.go @@ -188,7 +188,7 @@ func initializeQualifiedMembersGroup(dishonestThreshold, groupSize int) ( []*QualifiedMember, error, ) { - sharesJustifyingMembers, err := initializeSharesJustifyingMemberGroup( + sharesJustifyingMembers, _, err := initializeSharesJustifyingMemberGroup( dishonestThreshold, groupSize, ) diff --git a/pkg/beacon/gjkr/protocol_test.go b/pkg/beacon/gjkr/protocol_test.go index 25e23363dc..a5d4631442 100644 --- a/pkg/beacon/gjkr/protocol_test.go +++ b/pkg/beacon/gjkr/protocol_test.go @@ -69,12 +69,6 @@ func TestRoundTrip(t *testing.T) { len(member.receivedQualifiedSharesS), ) } - if len(member.receivedQualifiedSharesT) != groupSize-1 { - t.Fatalf("\nexpected: %d received shares T\nactual: %d\n", - groupSize-1, - len(member.receivedQualifiedSharesT), - ) - } member.CombineMemberShares() } diff --git a/pkg/beacon/node.go b/pkg/beacon/node.go index 6bd4054d81..5cf2b6408d 100644 --- a/pkg/beacon/node.go +++ b/pkg/beacon/node.go @@ -52,6 +52,28 @@ func (n *node) IsInGroup(groupPublicKey []byte) bool { return len(n.groupRegistry.GetGroup(groupPublicKey)) > 0 } +// setBroadcastChannelFilter installs the given membership filter on the +// broadcast channel so that only messages from operators in the group are +// accepted. It returns an error if the filter cannot be set; callers must abort +// on that error instead of proceeding on an unfiltered channel, which would +// accept messages from operators outside the group. +func setBroadcastChannelFilter( + channelLogger *zap.SugaredLogger, + channel net.BroadcastChannel, + filter net.BroadcastChannelFilter, +) error { + if err := channel.SetFilter(filter); err != nil { + channelLogger.Errorf( + "could not set filter for channel [%v]: [%v]", + channel.Name(), + err, + ) + return err + } + + return nil +} + // JoinDKGIfEligible takes a seed value and undergoes the process of the // distributed key generation if this node's operator proves to be eligible for // the group generated by that seed. This is an interactive on-chain process, @@ -133,13 +155,12 @@ func (n *node) JoinDKGIfEligible( signing, ) - err = broadcastChannel.SetFilter(membershipValidator.IsInGroup) - if err != nil { - dkgLogger.Errorf( - "could not set filter for channel [%v]: [%v]", - broadcastChannel.Name(), - err, - ) + if err = setBroadcastChannelFilter( + dkgLogger, + broadcastChannel, + membershipValidator.IsInGroup, + ); err != nil { + return } for _, index := range indexes { @@ -358,13 +379,12 @@ func (n *node) GenerateRelayEntry( n.beaconChain.Signing(), ) - err = channel.SetFilter(membershipValidator.IsInGroup) - if err != nil { - relayLogger.Errorf( - "could not set filter for channel [%v]: [%v]", - channel.Name(), - err, - ) + if err = setBroadcastChannelFilter( + relayLogger, + channel, + membershipValidator.IsInGroup, + ); err != nil { + return } blockCounter, err := n.beaconChain.BlockCounter() diff --git a/pkg/beacon/node_test.go b/pkg/beacon/node_test.go index f204ee5042..44a23565f8 100644 --- a/pkg/beacon/node_test.go +++ b/pkg/beacon/node_test.go @@ -5,11 +5,72 @@ import ( "math/big" "testing" + "go.uber.org/zap" + "github.com/keep-network/keep-core/pkg/chain/local_v1" + "github.com/keep-network/keep-core/pkg/net" + "github.com/keep-network/keep-core/pkg/operator" ) var relayEntryTimeout = uint64(15) +// filterErrorChannel is a broadcast channel whose SetFilter result is +// controllable, used to exercise the membership-filter abort path. +type filterErrorChannel struct { + net.BroadcastChannel + setFilterErr error +} + +func (c *filterErrorChannel) SetFilter(net.BroadcastChannelFilter) error { + return c.setFilterErr +} + +func (c *filterErrorChannel) Name() string { + return "test-channel" +} + +// TestSetBroadcastChannelFilter verifies that the membership filter is required +// before a node proceeds on a group channel: when the filter cannot be set the +// helper surfaces the error so the caller aborts, rather than proceeding on an +// unfiltered channel that would accept messages from operators outside the +// group. +func TestSetBroadcastChannelFilter(t *testing.T) { + filter := func(*operator.PublicKey) bool { return true } + + tests := map[string]struct { + setFilterErr error + expectError bool + }{ + "filter set successfully": { + setFilterErr: nil, + expectError: false, + }, + "filter cannot be set": { + setFilterErr: fmt.Errorf("cannot set filter"), + expectError: true, + }, + } + + for testName, test := range tests { + t.Run(testName, func(t *testing.T) { + channel := &filterErrorChannel{setFilterErr: test.setFilterErr} + + err := setBroadcastChannelFilter( + zap.NewNop().Sugar(), + channel, + filter, + ) + + if test.expectError && err == nil { + t.Fatal("expected an error, got nil") + } + if !test.expectError && err != nil { + t.Fatalf("unexpected error: [%v]", err) + } + }) + } +} + func TestMonitorRelayEntryOnChain_EntrySubmitted(t *testing.T) { localChain := local_v1.Connect(5, 3) diff --git a/pkg/beacon/registry/marshalling.go b/pkg/beacon/registry/marshaling.go similarity index 100% rename from pkg/beacon/registry/marshalling.go rename to pkg/beacon/registry/marshaling.go diff --git a/pkg/beacon/registry/marshalling_test.go b/pkg/beacon/registry/marshaling_test.go similarity index 100% rename from pkg/beacon/registry/marshalling_test.go rename to pkg/beacon/registry/marshaling_test.go diff --git a/pkg/bitcoin/electrum/electrum.go b/pkg/bitcoin/electrum/electrum.go index e670646e4a..d03a0eeb4f 100644 --- a/pkg/bitcoin/electrum/electrum.go +++ b/pkg/bitcoin/electrum/electrum.go @@ -1264,7 +1264,7 @@ func connectWithRetry( return result, err } -func requestWithRetry[K interface{}]( +func requestWithRetry[K any]( c *Connection, requestFn func(ctx context.Context, client *electrum.Client) (K, error), requestName string, diff --git a/pkg/chain/address.go b/pkg/chain/address.go index b856527b6d..81496adc1a 100644 --- a/pkg/chain/address.go +++ b/pkg/chain/address.go @@ -39,7 +39,7 @@ func (a Addresses) String() string { } var sb strings.Builder - var i = 0 + var i int sb.WriteString("[") for i = 0; i < len(a)-1; i++ { diff --git a/pkg/chain/ethereum/ethereum_timestamp_test.go b/pkg/chain/ethereum/ethereum_timestamp_test.go new file mode 100644 index 0000000000..34d33109ee --- /dev/null +++ b/pkg/chain/ethereum/ethereum_timestamp_test.go @@ -0,0 +1,238 @@ +package ethereum + +import ( + "context" + "errors" + "math/big" + "testing" + + "github.com/ethereum/go-ethereum/core/types" + "github.com/keep-network/keep-common/pkg/chain/ethereum/ethutil" +) + +// timestampMockClient is a minimal ethutil.EthereumClient used to exercise the +// timestamp-based block search. It embeds the interface so it satisfies the +// full contract while only the two methods used by GetBlockNumberByTimestamp +// are implemented; any other call would panic, which keeps the test honest +// about what the searched code actually touches. +type timestampMockClient struct { + ethutil.EthereumClient + // blockTimes maps a block number to its timestamp. + blockTimes map[uint64]uint64 + // latest is the number of the current (highest) block. + latest uint64 +} + +func newTimestampMockClient(baseTime, spacing, latest uint64) *timestampMockClient { + blockTimes := make(map[uint64]uint64) + for n := uint64(0); n <= latest; n++ { + blockTimes[n] = baseTime + n*spacing + } + return ×tampMockClient{blockTimes: blockTimes, latest: latest} +} + +func (m *timestampMockClient) header(number uint64) (*types.Header, error) { + time, ok := m.blockTimes[number] + if !ok { + return nil, errBlockOutOfRange + } + return &types.Header{ + Number: new(big.Int).SetUint64(number), + Time: time, + }, nil +} + +// HeaderByNumber returns the latest header when number is nil, matching the +// behavior currentBlock relies on. +func (m *timestampMockClient) HeaderByNumber( + _ context.Context, + number *big.Int, +) (*types.Header, error) { + if number == nil { + return m.header(m.latest) + } + return m.header(number.Uint64()) +} + +func (m *timestampMockClient) BlockByNumber( + _ context.Context, + number *big.Int, +) (*types.Block, error) { + header, err := m.header(number.Uint64()) + if err != nil { + return nil, err + } + return types.NewBlockWithHeader(header), nil +} + +var errBlockOutOfRange = errors.New("block out of range") + +func TestGetBlockNumberByTimestamp(t *testing.T) { + const ( + baseTime = uint64(1_600_000_000) + spacing = uint64(12) + latest = uint64(100) + ) + latestTime := baseTime + latest*spacing + + tests := map[string]struct { + timestamp uint64 + expectedBlock uint64 + expectingError bool + }{ + "timestamp of the latest block": { + timestamp: latestTime, + expectedBlock: latest, + }, + "timestamp exactly matching a middle block": { + timestamp: baseTime + 50*spacing, + expectedBlock: 50, + }, + "timestamp closer to the lower block": { + // Between block 50 (t=+600) and 51 (t=+612); +605 is 5s from 50 + // and 7s from 51, so the lower block wins. + timestamp: baseTime + 50*spacing + 5, + expectedBlock: 50, + }, + "timestamp closer to the higher block": { + // +607 is 7s from 50 and 5s from 51, so the higher block wins. + timestamp: baseTime + 50*spacing + 7, + expectedBlock: 51, + }, + "timestamp equidistant between two blocks": { + // +606 is 6s from both 50 and 51; closerBlock returns the greater + // block number on a tie. + timestamp: baseTime + 50*spacing + 6, + expectedBlock: 51, + }, + "timestamp of the earliest block": { + timestamp: baseTime, + expectedBlock: 0, + }, + "timestamp in the future": { + timestamp: latestTime + 1, + expectingError: true, + }, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + bc := &baseChain{ + client: newTimestampMockClient(baseTime, spacing, latest), + } + + block, err := bc.GetBlockNumberByTimestamp(test.timestamp) + + if test.expectingError { + if err == nil { + t.Fatalf("expected an error but got none") + } + return + } + + if err != nil { + t.Fatalf("unexpected error: [%v]", err) + } + + if block != test.expectedBlock { + t.Errorf( + "unexpected block number\nexpected: [%d]\nactual: [%d]", + test.expectedBlock, + block, + ) + } + }) + } +} + +// TestGetBlockNumberByTimestamp_ForwardCompensation exercises the forward +// compensation loop in GetBlockNumberByTimestamp. When the actual block spacing +// (15s) is greater than the assumed averageBlockTime (13s), the initial backward +// jump overshoots below the target timestamp, so the search must walk forward +// block by block to converge. The main table above uses a 12s spacing, where the +// backward jump always lands at or after the target and the forward loop never +// runs. +func TestGetBlockNumberByTimestamp_ForwardCompensation(t *testing.T) { + const ( + baseTime = uint64(1_600_000_000) + spacing = uint64(15) + latest = uint64(100) + ) + + bc := &baseChain{ + client: newTimestampMockClient(baseTime, spacing, latest), + } + + // Target a point 7s after block 50 (t=+757), between block 50 (t=+750) and + // block 51 (t=+765). The backward jump from the tip lands on block 43 + // (t=+645, before the target), so the forward loop walks 43->51 and the + // closer-block tie-break then selects block 50 (7s away vs. 8s). + timestamp := baseTime + 50*spacing + 7 + expectedBlock := uint64(50) + + block, err := bc.GetBlockNumberByTimestamp(timestamp) + if err != nil { + t.Fatalf("unexpected error: [%v]", err) + } + + if block != expectedBlock { + t.Errorf( + "unexpected block number\nexpected: [%d]\nactual: [%d]", + expectedBlock, + block, + ) + } +} + +func TestCloserBlock(t *testing.T) { + block := func(number, time uint64) *types.Block { + return types.NewBlockWithHeader(&types.Header{ + Number: new(big.Int).SetUint64(number), + Time: time, + }) + } + + tests := map[string]struct { + timestamp uint64 + b1, b2 *types.Block + expectedNumber uint64 + }{ + "first block closer": { + timestamp: 100, + b1: block(5, 100), + b2: block(6, 110), + expectedNumber: 5, + }, + "second block closer": { + timestamp: 110, + b1: block(5, 100), + b2: block(6, 110), + expectedNumber: 6, + }, + "equidistant returns the greater block number (b2)": { + timestamp: 105, + b1: block(5, 100), + b2: block(6, 110), + expectedNumber: 6, + }, + "equidistant returns the greater block number (b1)": { + timestamp: 105, + b1: block(6, 110), + b2: block(5, 100), + expectedNumber: 6, + }, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + result := closerBlock(test.timestamp, test.b1, test.b2) + if result.NumberU64() != test.expectedNumber { + t.Errorf( + "unexpected block number\nexpected: [%d]\nactual: [%d]", + test.expectedNumber, + result.NumberU64(), + ) + } + }) + } +} diff --git a/pkg/clientinfo/performance.go b/pkg/clientinfo/performance.go index d48c1c6b4d..9f0209858a 100644 --- a/pkg/clientinfo/performance.go +++ b/pkg/clientinfo/performance.go @@ -37,15 +37,12 @@ type PerformanceMetrics struct { registry *Registry cancel context.CancelFunc - // Counters track cumulative counts of events countersMutex sync.RWMutex counters map[string]*counter - // Histograms track distributions of values (like durations) histogramsMutex sync.RWMutex histograms map[string]*histogram - // Gauges track current values (like queue sizes) gaugesMutex sync.RWMutex gauges map[string]*gauge } @@ -102,7 +99,16 @@ func (pm *PerformanceMetrics) Stop() { // registerAllMetrics registers all performance metrics with 0 values // so they appear in the /metrics endpoint even before operations occur. func (pm *PerformanceMetrics) registerAllMetrics() { - // Register all counter metrics with 0 initial value + pm.registerCounterMetrics() + pm.registerWalletActionMetrics() + pm.registerHistogramMetrics() + pm.registerGaugeMetrics() +} + +// registerCounterMetrics registers all counter metrics with 0 initial values. +// Map entries are populated before observers are registered so that observer +// callbacks never read the map while it is being written concurrently. +func (pm *PerformanceMetrics) registerCounterMetrics() { counters := []string{ MetricDKGJoinedTotal, MetricDKGFailedTotal, @@ -119,6 +125,9 @@ func (pm *PerformanceMetrics) registerAllMetrics() { MetricRedemptionProofSubmissionsTotal, MetricRedemptionProofSubmissionsSuccessTotal, MetricRedemptionProofSubmissionsFailedTotal, + MetricDepositSweepProofSubmissionsTotal, + MetricDepositSweepProofSubmissionsSuccessTotal, + MetricDepositSweepProofSubmissionsFailedTotal, MetricWalletActionsTotal, MetricWalletActionSuccessTotal, MetricWalletActionFailedTotal, @@ -147,14 +156,12 @@ func (pm *PerformanceMetrics) registerAllMetrics() { counters = append(counters, NetworkJoinFailureMetricName(reason)) } - // First, initialize all counters in the map pm.countersMutex.Lock() for _, name := range counters { pm.counters[name] = &counter{value: 0} } pm.countersMutex.Unlock() - // Then, register observers (this prevents concurrent map read/write) for _, name := range counters { metricName := name // Capture for closure pm.registry.ObserveApplicationSource( @@ -175,7 +182,11 @@ func (pm *PerformanceMetrics) registerAllMetrics() { ) } - // Register per-action type wallet metrics +} + +// registerWalletActionMetrics registers per-action-type wallet counters and +// duration histograms with 0 initial values. +func (pm *PerformanceMetrics) registerWalletActionMetrics() { // For each action type, register: total, success_total, failed_total, duration_seconds for _, actionType := range GetAllWalletActionTypes() { actionCounters := []string{ @@ -236,8 +247,12 @@ func (pm *PerformanceMetrics) registerAllMetrics() { ) } - // Register all duration/histogram metrics with 0 initial values - // Note: These use the actual metric names as used in the codebase +} + +// registerHistogramMetrics registers standalone duration/histogram metrics with +// 0 initial values. +func (pm *PerformanceMetrics) registerHistogramMetrics() { + // These use the actual metric names as used in the codebase. durationMetrics := []string{ MetricDKGDurationSeconds, MetricSigningDurationSeconds, @@ -249,7 +264,6 @@ func (pm *PerformanceMetrics) registerAllMetrics() { MetricNetworkHandshakeDurationSeconds, } - // First, initialize all histograms in the map pm.histogramsMutex.Lock() for _, name := range durationMetrics { pm.histograms[name] = &histogram{ @@ -258,7 +272,6 @@ func (pm *PerformanceMetrics) registerAllMetrics() { } pm.histogramsMutex.Unlock() - // Then, register observers (this prevents concurrent map read/write) for _, name := range durationMetrics { metricName := name sources := map[string]Source{ @@ -295,7 +308,10 @@ func (pm *PerformanceMetrics) registerAllMetrics() { pm.registry.ObserveApplicationSource("performance", sources) } - // Register all gauge metrics with 0 initial value +} + +// registerGaugeMetrics registers all gauge metrics with 0 initial values. +func (pm *PerformanceMetrics) registerGaugeMetrics() { gauges := []string{ MetricWalletDispatcherActiveActions, MetricIncomingMessageQueueSize, @@ -309,14 +325,12 @@ func (pm *PerformanceMetrics) registerAllMetrics() { MetricSwapUtilizationPercent, } - // First, initialize all gauges in the map pm.gaugesMutex.Lock() for _, name := range gauges { pm.gauges[name] = &gauge{value: 0} } pm.gaugesMutex.Unlock() - // Then, register observers (this prevents concurrent map read/write) for _, name := range gauges { metricName := name // Capture for closure pm.registry.ObserveApplicationSource( @@ -433,7 +447,7 @@ func (pm *PerformanceMetrics) SetGauge(name string, value float64) { // observeSystemMetrics periodically collects and updates system metrics // including CPU utilization, memory usage, and goroutine count. func (pm *PerformanceMetrics) observeSystemMetrics(ctx context.Context) { - ticker := time.NewTicker(60 * time.Second) // Update every 10 seconds + ticker := time.NewTicker(60 * time.Second) // Update every 60 seconds defer ticker.Stop() var lastMemStats runtime.MemStats @@ -634,6 +648,11 @@ const ( MetricRedemptionProofSubmissionsSuccessTotal = "redemption_proof_submissions_success_total" MetricRedemptionProofSubmissionsFailedTotal = "redemption_proof_submissions_failed_total" + // Deposit Sweep Proof Submission Metrics (SPV maintainer) + MetricDepositSweepProofSubmissionsTotal = "deposit_sweep_proof_submissions_total" + MetricDepositSweepProofSubmissionsSuccessTotal = "deposit_sweep_proof_submissions_success_total" + MetricDepositSweepProofSubmissionsFailedTotal = "deposit_sweep_proof_submissions_failed_total" + // Wallet Action Metrics (aggregate) MetricWalletActionsTotal = "wallet_actions_total" MetricWalletActionSuccessTotal = "wallet_action_success_total" diff --git a/pkg/maintainer/maintainer.go b/pkg/maintainer/maintainer.go index ea2fbfabcf..279615d9ac 100644 --- a/pkg/maintainer/maintainer.go +++ b/pkg/maintainer/maintainer.go @@ -17,6 +17,7 @@ func Initialize( btcChain bitcoin.Chain, btcDiffChain btcdiff.Chain, spvChain spv.Chain, + metricsRecorder spv.MetricsRecorder, ) { // If none of the maintainers was specified in the config (i.e. no option was // provided to the `maintainer` command), all maintainers should be launched. @@ -43,6 +44,7 @@ func Initialize( spvChain, btcDiffChain, btcChain, + metricsRecorder, ) } diff --git a/pkg/maintainer/spv/deposit_sweep.go b/pkg/maintainer/spv/deposit_sweep.go index f8405e1576..5608ccc949 100644 --- a/pkg/maintainer/spv/deposit_sweep.go +++ b/pkg/maintainer/spv/deposit_sweep.go @@ -9,6 +9,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/keep-network/keep-core/pkg/bitcoin" "github.com/keep-network/keep-core/pkg/chain" + "github.com/keep-network/keep-core/pkg/clientinfo" ) // SubmitDepositSweepProof prepares deposit sweep proof for the given @@ -19,6 +20,7 @@ func SubmitDepositSweepProof( requiredConfirmations uint, btcChain bitcoin.Chain, spvChain Chain, + metricsRecorder MetricsRecorder, ) error { return submitDepositSweepProof( transactionHash, @@ -26,7 +28,7 @@ func SubmitDepositSweepProof( btcChain, spvChain, bitcoin.AssembleSpvProof, - getGlobalMetricsRecorder(), + metricsRecorder, ) } @@ -36,18 +38,16 @@ func submitDepositSweepProof( btcChain bitcoin.Chain, spvChain Chain, spvProofAssembler spvProofAssembler, - metricsRecorder interface { - IncrementCounter(name string, value float64) - }, + metricsRecorder MetricsRecorder, ) error { // Record proof submission attempt if metricsRecorder != nil { - metricsRecorder.IncrementCounter("deposit_sweep_proof_submissions_total", 1) + metricsRecorder.IncrementCounter(clientinfo.MetricDepositSweepProofSubmissionsTotal, 1) } if requiredConfirmations == 0 { if metricsRecorder != nil { - metricsRecorder.IncrementCounter("deposit_sweep_proof_submissions_failed_total", 1) + metricsRecorder.IncrementCounter(clientinfo.MetricDepositSweepProofSubmissionsFailedTotal, 1) } return fmt.Errorf( "provided required confirmations count must be greater than 0", @@ -61,7 +61,7 @@ func submitDepositSweepProof( ) if err != nil { if metricsRecorder != nil { - metricsRecorder.IncrementCounter("deposit_sweep_proof_submissions_failed_total", 1) + metricsRecorder.IncrementCounter(clientinfo.MetricDepositSweepProofSubmissionsFailedTotal, 1) } return fmt.Errorf( "failed to assemble transaction spv proof: [%v]", @@ -76,7 +76,7 @@ func submitDepositSweepProof( ) if err != nil { if metricsRecorder != nil { - metricsRecorder.IncrementCounter("deposit_sweep_proof_submissions_failed_total", 1) + metricsRecorder.IncrementCounter(clientinfo.MetricDepositSweepProofSubmissionsFailedTotal, 1) } return fmt.Errorf( "error while parsing transaction inputs: [%v]", @@ -91,7 +91,7 @@ func submitDepositSweepProof( vault, ); err != nil { if metricsRecorder != nil { - metricsRecorder.IncrementCounter("deposit_sweep_proof_submissions_failed_total", 1) + metricsRecorder.IncrementCounter(clientinfo.MetricDepositSweepProofSubmissionsFailedTotal, 1) } return fmt.Errorf( "failed to submit deposit sweep proof with reimbursement: [%v]", @@ -101,7 +101,7 @@ func submitDepositSweepProof( // Record successful proof submission if metricsRecorder != nil { - metricsRecorder.IncrementCounter("deposit_sweep_proof_submissions_success_total", 1) + metricsRecorder.IncrementCounter(clientinfo.MetricDepositSweepProofSubmissionsSuccessTotal, 1) } return nil @@ -118,17 +118,12 @@ func parseDepositSweepTransactionInputs( common.Address, error, ) { - // Represents the main UTXO of the deposit sweep transaction. Nil if there - // was no main UTXO. var mainUTXO *bitcoin.UnspentTransactionOutput = nil - // Stores the vault address of the deposits. Each deposit should have the - // same value of vault. The zero-filled value indicates there was no vault - // value set for the deposits. + // Each deposit must have the same vault value. The zero-filled value + // indicates there was no vault set for the deposits. var vault = common.Address{} - // This flag checks if at least one deposit input has been found during - // deposit processing. var depositAlreadyProcessed = false // Perform a sanity check: a deposit sweep transaction must have exactly one @@ -256,20 +251,11 @@ func getUnprovenDepositSweepTransactions( []*bitcoin.Transaction, error, ) { - blockCounter, err := spvChain.BlockCounter() + startBlock, err := unprovenSearchStartBlock(historyDepth, spvChain) if err != nil { - return nil, fmt.Errorf("failed to get block counter: [%v]", err) + return nil, err } - currentBlock, err := blockCounter.CurrentBlock() - if err != nil { - return nil, fmt.Errorf("failed to get current block: [%v]", err) - } - - // Calculate the starting block of the range in which the events will be - // searched for. - startBlock := currentBlock - historyDepth - events, err := spvChain.PastDepositRevealedEvents( &tbtc.DepositRevealedEventFilter{ @@ -310,40 +296,28 @@ func getUnprovenDepositSweepTransactions( continue } - walletTransactions, err := btcChain.GetTransactionsForPublicKeyHash( + unproven, err := collectUnprovenWalletTransactions( walletPublicKeyHash, transactionLimit, - ) - if err != nil { - return nil, fmt.Errorf( - "failed to get transactions for wallet: [%v]", - err, - ) - } - - for _, transaction := range walletTransactions { - isUnproven, err := - isUnprovenDepositSweepTransaction( + btcChain, + func(transaction *bitcoin.Transaction) (bool, error) { + return isUnprovenDepositSweepTransaction( transaction, walletPublicKeyHash, btcChain, spvChain, ) - if err != nil { - return nil, fmt.Errorf( - "failed to check if transaction is an unproven deposit sweep "+ - "transaction: [%v]", - err, - ) - } - - if isUnproven { - unprovenDepositSweepTransactions = append( - unprovenDepositSweepTransactions, - transaction, - ) - } + }, + false, + ) + if err != nil { + return nil, err } + + unprovenDepositSweepTransactions = append( + unprovenDepositSweepTransactions, + unproven..., + ) } return unprovenDepositSweepTransactions, nil diff --git a/pkg/maintainer/spv/deposit_sweep_test.go b/pkg/maintainer/spv/deposit_sweep_test.go index dc61256ccf..4d44e73c18 100644 --- a/pkg/maintainer/spv/deposit_sweep_test.go +++ b/pkg/maintainer/spv/deposit_sweep_test.go @@ -10,9 +10,24 @@ import ( "github.com/keep-network/keep-core/internal/testutils" "github.com/keep-network/keep-core/pkg/bitcoin" + "github.com/keep-network/keep-core/pkg/clientinfo" "github.com/keep-network/keep-core/pkg/tbtc" ) +// fakeMetricsRecorder is a test double for MetricsRecorder that accumulates the +// counter increments it receives, so tests can assert on them. +type fakeMetricsRecorder struct { + counters map[string]float64 +} + +func newFakeMetricsRecorder() *fakeMetricsRecorder { + return &fakeMetricsRecorder{counters: make(map[string]float64)} +} + +func (f *fakeMetricsRecorder) IncrementCounter(name string, value float64) { + f.counters[name] += value +} + func TestSubmitDepositSweepProof(t *testing.T) { bytesFromHex := func(str string) []byte { value, err := hex.DecodeString(str) @@ -90,13 +105,15 @@ func TestSubmitDepositSweepProof(t *testing.T) { return nil, nil, fmt.Errorf("error while assembling spv proof") } + recorder := newFakeMetricsRecorder() + err := submitDepositSweepProof( depositSweepTransaction.Hash(), requiredConfirmations, btcChain, spvChain, mockSpvProofAssembler, - getGlobalMetricsRecorder(), + recorder, ) if err != nil { t.Fatal(err) @@ -132,6 +149,116 @@ func TestSubmitDepositSweepProof(t *testing.T) { expectedVault[:], submittedProof.vault[:], ) + + // A successful submission must record one attempt and one success, and + // must not record a failure. + testutils.AssertIntsEqual( + t, + "total submissions counter", + 1, + int(recorder.counters[clientinfo.MetricDepositSweepProofSubmissionsTotal]), + ) + testutils.AssertIntsEqual( + t, + "success counter", + 1, + int(recorder.counters[clientinfo.MetricDepositSweepProofSubmissionsSuccessTotal]), + ) + testutils.AssertIntsEqual( + t, + "failed counter", + 0, + int(recorder.counters[clientinfo.MetricDepositSweepProofSubmissionsFailedTotal]), + ) +} + +// TestSubmitDepositSweepProofRecordsFailureMetrics verifies that a rejected +// submission (here, zero required confirmations) records both an attempt and a +// failure, but no success. This is the metrics seam that the maintainer boot +// path activates in production by passing a real recorder. +func TestSubmitDepositSweepProofRecordsFailureMetrics(t *testing.T) { + recorder := newFakeMetricsRecorder() + + err := submitDepositSweepProof( + bitcoin.Hash{}, + 0, // zero required confirmations forces an early failure + nil, + nil, + nil, + recorder, + ) + if err == nil { + t.Fatal("expected an error for zero required confirmations") + } + + testutils.AssertIntsEqual( + t, + "total submissions counter", + 1, + int(recorder.counters[clientinfo.MetricDepositSweepProofSubmissionsTotal]), + ) + testutils.AssertIntsEqual( + t, + "failed counter", + 1, + int(recorder.counters[clientinfo.MetricDepositSweepProofSubmissionsFailedTotal]), + ) + testutils.AssertIntsEqual( + t, + "success counter", + 0, + int(recorder.counters[clientinfo.MetricDepositSweepProofSubmissionsSuccessTotal]), + ) +} + +// TestSubmitDepositSweepProofRecordsAssembleFailureMetrics verifies that a +// failure occurring after the initial attempt is counted as a failure and not a +// success. Here the SPV proof assembler errors out, exercising a deeper +// failed-counter branch than the early zero-confirmations reject. The +// parse-inputs and on-chain-submit branches share this same guard idiom; the +// local chain double cannot be forced to fail the on-chain submit, so that +// branch is left to the shared pattern. +func TestSubmitDepositSweepProofRecordsAssembleFailureMetrics(t *testing.T) { + recorder := newFakeMetricsRecorder() + + failingAssembler := func( + bitcoin.Hash, + uint, + bitcoin.Chain, + ) (*bitcoin.Transaction, *bitcoin.SpvProof, error) { + return nil, nil, fmt.Errorf("error while assembling spv proof") + } + + err := submitDepositSweepProof( + bitcoin.Hash{}, + 6, // non-zero, so the failure comes from proof assembly + nil, + nil, + failingAssembler, + recorder, + ) + if err == nil { + t.Fatal("expected an error from the failing proof assembler") + } + + testutils.AssertIntsEqual( + t, + "total submissions counter", + 1, + int(recorder.counters[clientinfo.MetricDepositSweepProofSubmissionsTotal]), + ) + testutils.AssertIntsEqual( + t, + "failed counter", + 1, + int(recorder.counters[clientinfo.MetricDepositSweepProofSubmissionsFailedTotal]), + ) + testutils.AssertIntsEqual( + t, + "success counter", + 0, + int(recorder.counters[clientinfo.MetricDepositSweepProofSubmissionsSuccessTotal]), + ) } func TestGetUnprovenDepositSweepTransactions(t *testing.T) { diff --git a/pkg/maintainer/spv/moved_funds_sweep.go b/pkg/maintainer/spv/moved_funds_sweep.go index 417a1f5347..e070710abb 100644 --- a/pkg/maintainer/spv/moved_funds_sweep.go +++ b/pkg/maintainer/spv/moved_funds_sweep.go @@ -11,11 +11,15 @@ import ( // SubmitMovedFundsSweepProof prepares moved funds sweep proof for the given // transaction and submits it to the on-chain contract. If the number of // required confirmations is `0`, an error is returned. +// The metricsRecorder parameter is accepted to satisfy transactionProofSubmitter +// but is currently unused: moved funds sweep proof submissions are not yet +// instrumented with metrics. func SubmitMovedFundsSweepProof( transactionHash bitcoin.Hash, requiredConfirmations uint, btcChain bitcoin.Chain, spvChain Chain, + _ MetricsRecorder, ) error { return submitMovedFundsSweepProof( transactionHash, @@ -137,20 +141,11 @@ func getUnprovenMovedFundsSweepTransactions( []*bitcoin.Transaction, error, ) { - blockCounter, err := spvChain.BlockCounter() + startBlock, err := unprovenSearchStartBlock(historyDepth, spvChain) if err != nil { - return nil, fmt.Errorf("failed to get block counter: [%v]", err) + return nil, err } - currentBlock, err := blockCounter.CurrentBlock() - if err != nil { - return nil, fmt.Errorf("failed to get current block: [%v]", err) - } - - // Calculate the starting block of the range in which the events will be - // searched for. - startBlock := currentBlock - historyDepth - events, err := spvChain.PastMovingFundsCommitmentSubmittedEvents( &tbtc.MovingFundsCommitmentSubmittedEventFilter{ @@ -211,45 +206,31 @@ func getUnprovenMovedFundsSweepTransactions( // When wallet makes a moved funds sweep transaction, it transfers // funds to itself. Therefore we can search all the transactions that // pay to the wallet's public key hash. - walletTransactions, err := btcChain.GetTransactionsForPublicKeyHash( + // + // A wallet can have only one unproven moved funds sweep transaction at + // a time, so we stop at the first match. + unproven, err := collectUnprovenWalletTransactions( walletPublicKeyHash, transactionLimit, - ) - if err != nil { - return nil, fmt.Errorf( - "failed to get transactions for wallet: [%v]", - err, - ) - } - - for _, transaction := range walletTransactions { - isUnproven, err := - isUnprovenMovedFundsSweepTransaction( + btcChain, + func(transaction *bitcoin.Transaction) (bool, error) { + return isUnprovenMovedFundsSweepTransaction( transaction, walletPublicKeyHash, btcChain, spvChain, ) - if err != nil { - return nil, fmt.Errorf( - "failed to check if transaction is an unproven moved "+ - "funds sweep transaction: [%v]", - err, - ) - } - - if isUnproven { - unprovenMovedFundsSweepTransactions = append( - unprovenMovedFundsSweepTransactions, - transaction, - ) - - // A wallet can have only one unproven moved funds sweep - // transaction at a time. If we found such transaction, we don't - // have to look at this wallet's transactions anymore. - break - } + }, + true, + ) + if err != nil { + return nil, err } + + unprovenMovedFundsSweepTransactions = append( + unprovenMovedFundsSweepTransactions, + unproven..., + ) } return unprovenMovedFundsSweepTransactions, nil diff --git a/pkg/maintainer/spv/moving_funds.go b/pkg/maintainer/spv/moving_funds.go index 81d1e13e51..011e17e80c 100644 --- a/pkg/maintainer/spv/moving_funds.go +++ b/pkg/maintainer/spv/moving_funds.go @@ -11,11 +11,15 @@ import ( // SubmitMovingFundsProof prepares moving funds proof for the given // transaction and submits it to the on-chain contract. If the number of // required confirmations is `0`, an error is returned. +// The metricsRecorder parameter is accepted to satisfy transactionProofSubmitter +// but is currently unused: moving funds proof submissions are not yet +// instrumented with metrics. func SubmitMovingFundsProof( transactionHash bitcoin.Hash, requiredConfirmations uint, btcChain bitcoin.Chain, spvChain Chain, + _ MetricsRecorder, ) error { return submitMovingFundsProof( transactionHash, @@ -133,20 +137,11 @@ func getUnprovenMovingFundsTransactions( []*bitcoin.Transaction, error, ) { - blockCounter, err := spvChain.BlockCounter() + startBlock, err := unprovenSearchStartBlock(historyDepth, spvChain) if err != nil { - return nil, fmt.Errorf("failed to get block counter: [%v]", err) + return nil, err } - currentBlock, err := blockCounter.CurrentBlock() - if err != nil { - return nil, fmt.Errorf("failed to get current block: [%v]", err) - } - - // Calculate the starting block of the range in which the events will be - // searched for. - startBlock := currentBlock - historyDepth - // The `MovingFundsCommitmentSubmitted` event can only be emitted once for // a given wallet. Therefore there will always be only one event for a wallet. // We do not have to worry about duplicate events for the same wallet. @@ -197,41 +192,29 @@ func getUnprovenMovingFundsTransactions( // source wallet. targetWalletPublicKeyHash := targetWallets[0] - walletTransactions, err := btcChain.GetTransactionsForPublicKeyHash( + unproven, err := collectUnprovenWalletTransactions( targetWalletPublicKeyHash, transactionLimit, - ) - if err != nil { - return nil, fmt.Errorf( - "failed to get transactions for wallet: [%v]", - err, - ) - } - - for _, transaction := range walletTransactions { - isUnproven, err := - isUnprovenMovingFundsTransaction( + btcChain, + func(transaction *bitcoin.Transaction) (bool, error) { + return isUnprovenMovingFundsTransaction( transaction, walletPublicKeyHash, targetWallets, btcChain, spvChain, ) - if err != nil { - return nil, fmt.Errorf( - "failed to check if transaction is an unproven moving funds "+ - "transaction: [%v]", - err, - ) - } - - if isUnproven { - unprovenMovingFundsTransactions = append( - unprovenMovingFundsTransactions, - transaction, - ) - } + }, + false, + ) + if err != nil { + return nil, err } + + unprovenMovingFundsTransactions = append( + unprovenMovingFundsTransactions, + unproven..., + ) } return unprovenMovingFundsTransactions, nil diff --git a/pkg/maintainer/spv/redemptions.go b/pkg/maintainer/spv/redemptions.go index e504860f81..6665cf1f4a 100644 --- a/pkg/maintainer/spv/redemptions.go +++ b/pkg/maintainer/spv/redemptions.go @@ -9,13 +9,6 @@ import ( "github.com/keep-network/keep-core/pkg/tbtc" ) -// getGlobalMetricsRecorder returns the global metrics recorder if set. -func getGlobalMetricsRecorder() interface { - IncrementCounter(name string, value float64) -} { - return getMetricsRecorder() -} - // SubmitRedemptionProof prepares redemption proof for the given transaction // and submits it to the on-chain contract. If the number of required // confirmations is `0`, an error is returned. @@ -24,6 +17,7 @@ func SubmitRedemptionProof( requiredConfirmations uint, btcChain bitcoin.Chain, spvChain Chain, + metricsRecorder MetricsRecorder, ) error { return submitRedemptionProof( transactionHash, @@ -31,7 +25,7 @@ func SubmitRedemptionProof( btcChain, spvChain, bitcoin.AssembleSpvProof, - getGlobalMetricsRecorder(), + metricsRecorder, ) } @@ -41,9 +35,7 @@ func submitRedemptionProof( btcChain bitcoin.Chain, spvChain Chain, spvProofAssembler spvProofAssembler, - metricsRecorder interface { - IncrementCounter(name string, value float64) - }, + metricsRecorder MetricsRecorder, ) error { // Record proof submission attempt if metricsRecorder != nil { @@ -167,20 +159,11 @@ func getUnprovenRedemptionTransactions( []*bitcoin.Transaction, error, ) { - blockCounter, err := spvChain.BlockCounter() + startBlock, err := unprovenSearchStartBlock(historyDepth, spvChain) if err != nil { - return nil, fmt.Errorf("failed to get block counter: [%v]", err) + return nil, err } - currentBlock, err := blockCounter.CurrentBlock() - if err != nil { - return nil, fmt.Errorf("failed to get current block: [%v]", err) - } - - // Calculate the starting block of the range in which the events will be - // searched for. - startBlock := currentBlock - historyDepth - events, err := spvChain.PastRedemptionRequestedEvents( &tbtc.RedemptionRequestedEventFilter{ @@ -221,40 +204,28 @@ func getUnprovenRedemptionTransactions( continue } - walletTransactions, err := btcChain.GetTransactionsForPublicKeyHash( + unproven, err := collectUnprovenWalletTransactions( walletPublicKeyHash, transactionLimit, - ) - if err != nil { - return nil, fmt.Errorf( - "failed to get transactions for wallet: [%v]", - err, - ) - } - - for _, transaction := range walletTransactions { - isUnproven, err := - isUnprovenRedemptionTransaction( + btcChain, + func(transaction *bitcoin.Transaction) (bool, error) { + return isUnprovenRedemptionTransaction( transaction, walletPublicKeyHash, btcChain, spvChain, ) - if err != nil { - return nil, fmt.Errorf( - "failed to check if transaction is an unproven redemption "+ - "transaction: [%v]", - err, - ) - } - - if isUnproven { - unprovenRedemptionTransactions = append( - unprovenRedemptionTransactions, - transaction, - ) - } + }, + false, + ) + if err != nil { + return nil, err } + + unprovenRedemptionTransactions = append( + unprovenRedemptionTransactions, + unproven..., + ) } return unprovenRedemptionTransactions, nil diff --git a/pkg/maintainer/spv/redemptions_test.go b/pkg/maintainer/spv/redemptions_test.go index 4f10a3a208..1271285a89 100644 --- a/pkg/maintainer/spv/redemptions_test.go +++ b/pkg/maintainer/spv/redemptions_test.go @@ -6,10 +6,100 @@ import ( "github.com/go-test/deep" "github.com/keep-network/keep-core/internal/testutils" "github.com/keep-network/keep-core/pkg/bitcoin" + "github.com/keep-network/keep-core/pkg/clientinfo" "github.com/keep-network/keep-core/pkg/tbtc" "testing" ) +// TestSubmitRedemptionProofRecordsFailureMetrics verifies that a rejected +// submission (here, zero required confirmations) records both an attempt and a +// failure, but no success. This is the metrics seam that the maintainer boot +// path activates in production by passing a real recorder. +func TestSubmitRedemptionProofRecordsFailureMetrics(t *testing.T) { + recorder := newFakeMetricsRecorder() + + err := submitRedemptionProof( + bitcoin.Hash{}, + 0, // zero required confirmations forces an early failure + nil, + nil, + nil, + recorder, + ) + if err == nil { + t.Fatal("expected an error for zero required confirmations") + } + + testutils.AssertIntsEqual( + t, + "total submissions counter", + 1, + int(recorder.counters[clientinfo.MetricRedemptionProofSubmissionsTotal]), + ) + testutils.AssertIntsEqual( + t, + "failed counter", + 1, + int(recorder.counters[clientinfo.MetricRedemptionProofSubmissionsFailedTotal]), + ) + testutils.AssertIntsEqual( + t, + "success counter", + 0, + int(recorder.counters[clientinfo.MetricRedemptionProofSubmissionsSuccessTotal]), + ) +} + +// TestSubmitRedemptionProofRecordsAssembleFailureMetrics verifies that a +// failure occurring after the initial attempt is counted as a failure and not a +// success. Here the SPV proof assembler errors out, exercising a deeper +// failed-counter branch than the early zero-confirmations reject. The +// parse-inputs and on-chain-submit branches share this same guard idiom; the +// local chain double cannot be forced to fail the on-chain submit, so that +// branch is left to the shared pattern. +func TestSubmitRedemptionProofRecordsAssembleFailureMetrics(t *testing.T) { + recorder := newFakeMetricsRecorder() + + failingAssembler := func( + bitcoin.Hash, + uint, + bitcoin.Chain, + ) (*bitcoin.Transaction, *bitcoin.SpvProof, error) { + return nil, nil, fmt.Errorf("error while assembling spv proof") + } + + err := submitRedemptionProof( + bitcoin.Hash{}, + 6, // non-zero, so the failure comes from proof assembly + nil, + nil, + failingAssembler, + recorder, + ) + if err == nil { + t.Fatal("expected an error from the failing proof assembler") + } + + testutils.AssertIntsEqual( + t, + "total submissions counter", + 1, + int(recorder.counters[clientinfo.MetricRedemptionProofSubmissionsTotal]), + ) + testutils.AssertIntsEqual( + t, + "failed counter", + 1, + int(recorder.counters[clientinfo.MetricRedemptionProofSubmissionsFailedTotal]), + ) + testutils.AssertIntsEqual( + t, + "success counter", + 0, + int(recorder.counters[clientinfo.MetricRedemptionProofSubmissionsSuccessTotal]), + ) +} + func TestSubmitRedemptionProof(t *testing.T) { bytesFromHex := func(str string) []byte { value, err := hex.DecodeString(str) @@ -72,13 +162,15 @@ func TestSubmitRedemptionProof(t *testing.T) { return nil, nil, fmt.Errorf("error while assembling spv proof") } + recorder := newFakeMetricsRecorder() + err = submitRedemptionProof( redemptionTransaction.Hash(), requiredConfirmations, btcChain, spvChain, mockSpvProofAssembler, - getGlobalMetricsRecorder(), + recorder, ) if err != nil { t.Fatal(err) @@ -110,6 +202,27 @@ func TestSubmitRedemptionProof(t *testing.T) { } testutils.AssertBytesEqual(t, bytesFromHex("03b74d6893ad46dfdd01b9e0e3b3385f4fce2d1e"), submittedProof.walletPublicKeyHash[:]) + + // A successful submission must record one attempt and one success, and + // must not record a failure. + testutils.AssertIntsEqual( + t, + "total submissions counter", + 1, + int(recorder.counters[clientinfo.MetricRedemptionProofSubmissionsTotal]), + ) + testutils.AssertIntsEqual( + t, + "success counter", + 1, + int(recorder.counters[clientinfo.MetricRedemptionProofSubmissionsSuccessTotal]), + ) + testutils.AssertIntsEqual( + t, + "failed counter", + 0, + int(recorder.counters[clientinfo.MetricRedemptionProofSubmissionsFailedTotal]), + ) } func TestGetUnprovenRedemptionTransactions(t *testing.T) { diff --git a/pkg/maintainer/spv/spv.go b/pkg/maintainer/spv/spv.go index f842821c84..1a07476485 100644 --- a/pkg/maintainer/spv/spv.go +++ b/pkg/maintainer/spv/spv.go @@ -6,7 +6,6 @@ import ( "encoding/hex" "fmt" "math/big" - "sync" "time" "github.com/keep-network/keep-core/pkg/tbtc" @@ -22,51 +21,33 @@ var logger = log.Logger("keep-maintainer-spv") // The length of the Bitcoin difficulty epoch in blocks. const difficultyEpochLength = 2016 +// MetricsRecorder records counter metrics for SPV proof submissions. It is +// satisfied by *clientinfo.PerformanceMetrics. A nil MetricsRecorder is a valid +// argument that disables metrics recording: callers must treat nil as "metrics +// off" and guard every invocation against it. +type MetricsRecorder interface { + IncrementCounter(name string, value float64) +} + func Initialize( ctx context.Context, config Config, spvChain Chain, btcDiffChain btcdiff.Chain, btcChain bitcoin.Chain, + metricsRecorder MetricsRecorder, ) { spvMaintainer := &spvMaintainer{ - config: config, - spvChain: spvChain, - btcDiffChain: btcDiffChain, - btcChain: btcChain, + config: config, + spvChain: spvChain, + btcDiffChain: btcDiffChain, + btcChain: btcChain, + metricsRecorder: metricsRecorder, } go spvMaintainer.startControlLoop(ctx) } -// globalMetricsRecorder is a package-level variable to access metrics recorder -// from proof submission functions. -var ( - globalMetricsRecorderMu sync.RWMutex - globalMetricsRecorder interface { - IncrementCounter(name string, value float64) - } -) - -// SetMetricsRecorder sets the metrics recorder for the SPV maintainer. -// This allows recording metrics for proof submissions. -func SetMetricsRecorder(recorder interface { - IncrementCounter(name string, value float64) -}) { - globalMetricsRecorderMu.Lock() - defer globalMetricsRecorderMu.Unlock() - globalMetricsRecorder = recorder -} - -// getMetricsRecorder safely retrieves the metrics recorder. -func getMetricsRecorder() interface { - IncrementCounter(name string, value float64) -} { - globalMetricsRecorderMu.RLock() - defer globalMetricsRecorderMu.RUnlock() - return globalMetricsRecorder -} - // proofTypes holds the information about proof types supported by the // SPV maintainer. var proofTypes = map[tbtc.WalletActionType]struct { @@ -92,10 +73,11 @@ var proofTypes = map[tbtc.WalletActionType]struct { } type spvMaintainer struct { - config Config - spvChain Chain - btcDiffChain btcdiff.Chain - btcChain bitcoin.Chain + config Config + spvChain Chain + btcDiffChain btcdiff.Chain + btcChain bitcoin.Chain + metricsRecorder MetricsRecorder } func (sm *spvMaintainer) startControlLoop(ctx context.Context) { @@ -173,6 +155,7 @@ type transactionProofSubmitter func( requiredConfirmations uint, btcChain bitcoin.Chain, spvChain Chain, + metricsRecorder MetricsRecorder, ) error // proveTransactions gets unproven Bitcoin transactions using the provided @@ -244,6 +227,7 @@ func (sm *spvMaintainer) proveTransactions( requiredConfirmations, sm.btcChain, sm.spvChain, + sm.metricsRecorder, ) if err != nil { return err @@ -495,6 +479,79 @@ func uniqueWalletPublicKeyHashes[T walletEvent](events []T) [][20]byte { return publicKeyHashes } +// unprovenSearchStartBlock returns the starting block of the range in which +// the events used to find unproven transactions are searched for. It is +// derived from the current chain tip and the configured history depth. +func unprovenSearchStartBlock( + historyDepth uint64, + spvChain Chain, +) (uint64, error) { + blockCounter, err := spvChain.BlockCounter() + if err != nil { + return 0, fmt.Errorf("failed to get block counter: [%v]", err) + } + + currentBlock, err := blockCounter.CurrentBlock() + if err != nil { + return 0, fmt.Errorf("failed to get current block: [%v]", err) + } + + // Guard against unsigned underflow on short chains (e.g. early test + // networks) where the configured history depth can exceed the current + // tip; clamp the search start to the genesis block instead of wrapping + // around to a near-maximum block number. + if historyDepth > currentBlock { + return 0, nil + } + + return currentBlock - historyDepth, nil +} + +// collectUnprovenWalletTransactions returns the recent transactions of the +// wallet identified by lookupPublicKeyHash that satisfy the isUnproven +// predicate. When stopAtFirstMatch is true it returns as soon as the first +// matching transaction is found, which is sufficient for wallet operations +// that can have at most one unproven transaction at a time. +func collectUnprovenWalletTransactions( + lookupPublicKeyHash [20]byte, + transactionLimit int, + btcChain bitcoin.Chain, + isUnproven func(transaction *bitcoin.Transaction) (bool, error), + stopAtFirstMatch bool, +) ([]*bitcoin.Transaction, error) { + walletTransactions, err := btcChain.GetTransactionsForPublicKeyHash( + lookupPublicKeyHash, + transactionLimit, + ) + if err != nil { + return nil, fmt.Errorf( + "failed to get transactions for wallet: [%v]", + err, + ) + } + + var unprovenTransactions []*bitcoin.Transaction + + for _, transaction := range walletTransactions { + matched, err := isUnproven(transaction) + if err != nil { + return nil, fmt.Errorf( + "failed to check if transaction is unproven: [%v]", + err, + ) + } + + if matched { + unprovenTransactions = append(unprovenTransactions, transaction) + if stopAtFirstMatch { + break + } + } + } + + return unprovenTransactions, nil +} + // spvProofAssembler is a type representing a function that is used // to assemble an SPV proof for the given transaction hash and confirmations // count. diff --git a/pkg/maintainer/spv/spv_test.go b/pkg/maintainer/spv/spv_test.go index 6f11fd6e2b..c1b7b69b39 100644 --- a/pkg/maintainer/spv/spv_test.go +++ b/pkg/maintainer/spv/spv_test.go @@ -2,6 +2,7 @@ package spv import ( "encoding/hex" + "fmt" "math/big" "reflect" "strings" @@ -374,3 +375,148 @@ func TestIsInputCurrentWalletsMainUTXO(t *testing.T) { }) } } + +// stubTransactionChain overrides GetTransactionsForPublicKeyHash on the local +// Bitcoin chain so that collectUnprovenWalletTransactions can be exercised with +// a controlled set of transactions and error, independently of how the local +// chain filters by public key hash. +type stubTransactionChain struct { + *localBitcoinChain + transactions []*bitcoin.Transaction + err error +} + +func (s *stubTransactionChain) GetTransactionsForPublicKeyHash( + _ [20]byte, + _ int, +) ([]*bitcoin.Transaction, error) { + return s.transactions, s.err +} + +func TestCollectUnprovenWalletTransactions(t *testing.T) { + // Distinct transactions identified by pointer. Empty Transaction values + // compare equal under reflect.DeepEqual, so assertions below rely on + // pointer identity, not value equality. + tx1 := &bitcoin.Transaction{} + tx2 := &bitcoin.Transaction{} + tx3 := &bitcoin.Transaction{} + + // matches returns a predicate reporting a transaction as unproven when it + // is one of the given transactions (compared by pointer). + matches := func(unproven ...*bitcoin.Transaction) func(*bitcoin.Transaction) (bool, error) { + return func(transaction *bitcoin.Transaction) (bool, error) { + for _, u := range unproven { + if transaction == u { + return true, nil + } + } + return false, nil + } + } + + predicateErr := fmt.Errorf("predicate failure") + chainErr := fmt.Errorf("chain failure") + + tests := map[string]struct { + transactions []*bitcoin.Transaction + isUnproven func(*bitcoin.Transaction) (bool, error) + stopAtFirstMatch bool + chainErr error + expectedResult []*bitcoin.Transaction + expectedErr string + }{ + "returns all matches when not stopping at first match": { + transactions: []*bitcoin.Transaction{tx1, tx2, tx3}, + isUnproven: matches(tx1, tx3), + stopAtFirstMatch: false, + expectedResult: []*bitcoin.Transaction{tx1, tx3}, + }, + "returns only the first match when stopping at first match": { + transactions: []*bitcoin.Transaction{tx1, tx2, tx3}, + isUnproven: matches(tx2, tx3), + stopAtFirstMatch: true, + expectedResult: []*bitcoin.Transaction{tx2}, + }, + "returns nothing when no transaction matches": { + transactions: []*bitcoin.Transaction{tx1, tx2, tx3}, + isUnproven: matches(), + stopAtFirstMatch: false, + expectedResult: nil, + }, + "propagates the chain error": { + transactions: []*bitcoin.Transaction{tx1}, + isUnproven: matches(tx1), + chainErr: chainErr, + expectedErr: "failed to get transactions for wallet", + }, + "propagates the predicate error": { + transactions: []*bitcoin.Transaction{tx1}, + isUnproven: func(*bitcoin.Transaction) (bool, error) { + return false, predicateErr + }, + expectedErr: "failed to check if transaction is unproven", + }, + } + + for testName, test := range tests { + t.Run(testName, func(t *testing.T) { + btcChain := &stubTransactionChain{ + localBitcoinChain: newLocalBitcoinChain(), + transactions: test.transactions, + err: test.chainErr, + } + + result, err := collectUnprovenWalletTransactions( + [20]byte{}, + len(test.transactions), + btcChain, + test.isUnproven, + test.stopAtFirstMatch, + ) + + if test.expectedErr != "" { + if result != nil { + t.Errorf( + "expected nil result on error, got [%v]", + result, + ) + } + if err == nil { + t.Fatalf( + "expected error containing [%s], got nil", + test.expectedErr, + ) + } + if !strings.Contains(err.Error(), test.expectedErr) { + t.Errorf( + "unexpected error\nexpected to contain: [%s]\nactual: [%v]", + test.expectedErr, + err, + ) + } + return + } + + if err != nil { + t.Fatalf("unexpected error: [%v]", err) + } + + testutils.AssertIntsEqual( + t, + "number of unproven transactions", + len(test.expectedResult), + len(result), + ) + + for i, expected := range test.expectedResult { + if result[i] != expected { + t.Errorf( + "unexpected transaction at index [%d]; "+ + "pointer identity mismatch", + i, + ) + } + } + }) + } +} diff --git a/pkg/net/libp2p/channel.go b/pkg/net/libp2p/channel.go index 852c40dc60..e935b5f357 100644 --- a/pkg/net/libp2p/channel.go +++ b/pkg/net/libp2p/channel.go @@ -84,11 +84,7 @@ type channel struct { retransmissionTicker *retransmission.Ticker // metricsRecorder is optional and used for recording performance metrics - metricsRecorder interface { - IncrementCounter(name string, value float64) - SetGauge(name string, value float64) - RecordDuration(name string, duration time.Duration) - } + metricsRecorder fullMetricsRecorder // monitorQueueSizesOnce ensures the monitoring goroutine is started only once monitorQueueSizesOnce sync.Once @@ -453,11 +449,7 @@ func extractPublicKey(peer peer.ID) (*operator.PublicKey, error) { // setMetricsRecorder sets the metrics recorder for the channel and starts // periodic queue size monitoring. -func (c *channel) setMetricsRecorder(recorder interface { - IncrementCounter(name string, value float64) - SetGauge(name string, value float64) - RecordDuration(name string, duration time.Duration) -}) { +func (c *channel) setMetricsRecorder(recorder fullMetricsRecorder) { c.metricsRecorder = recorder // Start periodic queue size monitoring (only once) if recorder != nil { diff --git a/pkg/net/libp2p/channel_manager.go b/pkg/net/libp2p/channel_manager.go index bcb10f7ffb..42ae17818e 100644 --- a/pkg/net/libp2p/channel_manager.go +++ b/pkg/net/libp2p/channel_manager.go @@ -50,11 +50,7 @@ type channelManager struct { topics map[string]*pubsub.Topic // metricsRecorder is optional and used for recording performance metrics - metricsRecorder interface { - IncrementCounter(name string, value float64) - SetGauge(name string, value float64) - RecordDuration(name string, duration time.Duration) - } + metricsRecorder fullMetricsRecorder } func newChannelManager( @@ -126,11 +122,7 @@ func (cm *channelManager) getChannel(name string) (*channel, error) { // setMetricsRecorder sets the metrics recorder for the channel manager // and wires it into existing channels. -func (cm *channelManager) setMetricsRecorder(recorder interface { - IncrementCounter(name string, value float64) - SetGauge(name string, value float64) - RecordDuration(name string, duration time.Duration) -}) { +func (cm *channelManager) setMetricsRecorder(recorder fullMetricsRecorder) { // Wire metrics into existing channels cm.channelsMutex.Lock() defer cm.channelsMutex.Unlock() diff --git a/pkg/net/libp2p/libp2p.go b/pkg/net/libp2p/libp2p.go index 0d8339df86..1a9f5ca860 100644 --- a/pkg/net/libp2p/libp2p.go +++ b/pkg/net/libp2p/libp2p.go @@ -394,6 +394,12 @@ func Connect( // SetMetricsRecorder sets the metrics recorder for the provider and wires it // into network components. +// +// The parameter is an anonymous interface rather than the named +// fullMetricsRecorder on purpose: callers (e.g. cmd.start) wire metrics via a +// structural type assertion against this exact signature to avoid importing the +// unexported provider type. Changing it to a named type would silently break +// that assertion. func (p *provider) SetMetricsRecorder(recorder interface { IncrementCounter(name string, value float64) SetGauge(name string, value float64) @@ -583,18 +589,10 @@ func buildNotifiee(libp2pHost host.Host, p *provider) libp2pnet.Notifiee { logger.Infof("established connection to [%v]", peerMultiaddress) - var recorder interface { - IncrementCounter(name string, value float64) - SetGauge(name string, value float64) - RecordDuration(name string, duration time.Duration) - } + var recorder fullMetricsRecorder if p.metricsRecorder != nil { if metricsRecorderValue := p.metricsRecorder.Load(); metricsRecorderValue != nil { - recorder = metricsRecorderValue.(interface { - IncrementCounter(name string, value float64) - SetGauge(name string, value float64) - RecordDuration(name string, duration time.Duration) - }) + recorder = metricsRecorderValue.(fullMetricsRecorder) recorder.IncrementCounter(clientinfo.MetricPeerConnectionsTotal, 1) } } @@ -625,11 +623,7 @@ func buildNotifiee(libp2pHost host.Host, p *provider) libp2pnet.Notifiee { if p.metricsRecorder != nil { if metricsRecorderValue := p.metricsRecorder.Load(); metricsRecorderValue != nil { - recorder := metricsRecorderValue.(interface { - IncrementCounter(name string, value float64) - SetGauge(name string, value float64) - RecordDuration(name string, duration time.Duration) - }) + recorder := metricsRecorderValue.(fullMetricsRecorder) recorder.IncrementCounter(clientinfo.MetricPeerDisconnectionsTotal, 1) } } @@ -650,11 +644,7 @@ func executePingTest( libp2pHost host.Host, peerID peer.ID, peerMultiaddress string, - metricsRecorder interface { - IncrementCounter(name string, value float64) - SetGauge(name string, value float64) - RecordDuration(name string, duration time.Duration) - }, + metricsRecorder fullMetricsRecorder, ) { logger.Infof("starting ping test for [%v]", peerMultiaddress) diff --git a/pkg/net/libp2p/metrics_wiring_test.go b/pkg/net/libp2p/metrics_wiring_test.go new file mode 100644 index 0000000000..e7df04c602 --- /dev/null +++ b/pkg/net/libp2p/metrics_wiring_test.go @@ -0,0 +1,29 @@ +package libp2p + +import ( + "testing" + "time" +) + +// TestProviderSatisfiesMetricsSetterAssertion pins the wiring contract used by +// cmd.start to inject the performance metrics recorder into the provider. That +// wiring is a structural type assertion against an anonymous interface, so the +// provider's SetMetricsRecorder parameter must remain that exact anonymous +// interface. If it is changed to a named type, the assertion silently fails at +// runtime (ok == false) and provider metrics stop being recorded, which no +// build error or other test would catch. +func TestProviderSatisfiesMetricsSetterAssertion(t *testing.T) { + var np interface{} = &provider{} + if _, ok := np.(interface { + SetMetricsRecorder(recorder interface { + IncrementCounter(name string, value float64) + SetGauge(name string, value float64) + RecordDuration(name string, duration time.Duration) + }) + }); !ok { + t.Fatal( + "provider no longer satisfies the metrics-setter assertion used " + + "by cmd.start; provider metrics wiring is broken", + ) + } +} diff --git a/pkg/net/libp2p/transport.go b/pkg/net/libp2p/transport.go index ccaa51f523..e0b761a7b1 100644 --- a/pkg/net/libp2p/transport.go +++ b/pkg/net/libp2p/transport.go @@ -46,6 +46,15 @@ type MetricsRecorder interface { RecordDuration(name string, duration time.Duration) } +// fullMetricsRecorder is a MetricsRecorder that also supports gauge metrics. +// It is the contract for components that record counters, durations, and +// gauges (e.g. message queue sizes), unlike the transport which records only +// counters and durations. +type fullMetricsRecorder interface { + MetricsRecorder + SetGauge(name string, value float64) +} + // transport constructs an encrypted and authenticated connection for a peer. type transport struct { protocolID protocol.ID diff --git a/pkg/protocol/inactivity/marshalling.go b/pkg/protocol/inactivity/marshaling.go similarity index 100% rename from pkg/protocol/inactivity/marshalling.go rename to pkg/protocol/inactivity/marshaling.go diff --git a/pkg/protocol/inactivity/marshalling_test.go b/pkg/protocol/inactivity/marshaling_test.go similarity index 100% rename from pkg/protocol/inactivity/marshalling_test.go rename to pkg/protocol/inactivity/marshaling_test.go diff --git a/pkg/protocol/state/sync_machine.go b/pkg/protocol/state/sync_machine.go index a6f2ec20ff..7d23fcc8ce 100644 --- a/pkg/protocol/state/sync_machine.go +++ b/pkg/protocol/state/sync_machine.go @@ -72,7 +72,7 @@ func (sm *SyncMachine) Execute(startBlockHeight uint64) (SyncState, uint64, erro err := sm.blockCounter.WaitForBlockHeight(startBlockHeight) if err != nil { cancelCtx() - return nil, 0, fmt.Errorf("failed to wait for the execution start block") + return nil, 0, fmt.Errorf("failed to wait for the execution start block: [%w]", err) } lastStateEndBlockHeight := startBlockHeight diff --git a/pkg/tbtc/coordination.go b/pkg/tbtc/coordination.go index 2dd75e9614..43e0b2d79f 100644 --- a/pkg/tbtc/coordination.go +++ b/pkg/tbtc/coordination.go @@ -380,9 +380,6 @@ func (ce *coordinationExecutor) coordinate( startTime := time.Now() - // Record duration metric once at the end using defer - var coordinationFailed bool - seed, err := ce.getSeed(window.coordinationBlock) if err != nil { return nil, fmt.Errorf("failed to compute coordination seed: [%v]", err) @@ -431,7 +428,6 @@ func (ce *coordinationExecutor) coordinate( // no point to keep the context active as retransmissions do not // occur anyway. cancelCtx() - coordinationFailed = true if ce.metricsRecorder != nil { ce.metricsRecorder.IncrementCounter(clientinfo.MetricCoordinationFailedTotal, 1) } @@ -455,7 +451,6 @@ func (ce *coordinationExecutor) coordinate( append(actionsChecklist, ActionNoop), ) if err != nil { - coordinationFailed = true // Record as leader timeout observation, not as a failure of this node. // The actual failure is on the leader's side. if ce.metricsRecorder != nil { @@ -498,7 +493,7 @@ func (ce *coordinationExecutor) coordinate( execLogger.Infof("coordination completed with result: [%s]", result) // Record successful coordination counter - if ce.metricsRecorder != nil && !coordinationFailed { + if ce.metricsRecorder != nil { ce.metricsRecorder.IncrementCounter(clientinfo.MetricCoordinationProceduresExecutedTotal, 1) ce.metricsRecorder.RecordDuration(clientinfo.MetricCoordinationDurationSeconds, time.Since(startTime)) } @@ -608,15 +603,12 @@ func (ce *coordinationExecutor) getActionsChecklist( // proposal generator performs a full-history chain scan. if coordinationBlock < DepositSweepEveryWindowActivationBlock { if windowIndex%frequencyWindows == 0 { - actions = append(actions, ActionDepositSweep) - } - - if windowIndex%frequencyWindows == 0 { - actions = append(actions, ActionMovedFundsSweep) - } - - if windowIndex%frequencyWindows == 0 { - actions = append(actions, ActionMovingFunds) + actions = append( + actions, + ActionDepositSweep, + ActionMovedFundsSweep, + ActionMovingFunds, + ) } } else { actions = append(actions, ActionDepositSweep) diff --git a/pkg/tbtc/coordination_test.go b/pkg/tbtc/coordination_test.go index c597048fb0..ae461216db 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -7,6 +7,7 @@ import ( "fmt" "math/big" "reflect" + "sync" "testing" "time" @@ -1245,6 +1246,14 @@ func TestCoordinationExecutor_ExecuteFollowerRoutine(t *testing.T) { localChain.Signing(), ) + // Wrap the follower's channel so the sender can wait until the follower + // routine has registered its message handler, rather than guessing with a + // fixed sleep. + followerChannel := &recvSignalingChannel{ + BroadcastChannel: follower1.channel, + recvRegistered: make(chan struct{}), + } + // Set up the executor for follower 1. executor := &coordinationExecutor{ // Set only relevant fields. @@ -1252,7 +1261,7 @@ func TestCoordinationExecutor_ExecuteFollowerRoutine(t *testing.T) { coordinatedWallet: coordinatedWallet, membersIndexes: coordinatedWallet.membersByOperator(follower1.address), operatorAddress: follower1.address, - broadcastChannel: follower1.channel, + broadcastChannel: followerChannel, membershipValidator: membershipValidator, } @@ -1260,9 +1269,13 @@ func TestCoordinationExecutor_ExecuteFollowerRoutine(t *testing.T) { defer cancelCtx() go func() { - // Give the follower routine some time to start and set up the - // broadcast channel handler. - time.Sleep(1 * time.Second) + // Wait until the follower routine has registered its broadcast channel + // handler; otherwise messages sent before registration are dropped. + select { + case <-followerChannel.recvRegistered: + case <-ctx.Done(): + return + } // Send message of wrong type. err := leader.channel.Send(ctx, &signingDoneMessage{ @@ -1557,3 +1570,17 @@ func (mcpg *mockCoordinationProposalGenerator) Generate( mcpg.calls, ) } + +// recvSignalingChannel wraps a broadcast channel and closes recvRegistered the +// first time a handler is installed via Recv. It lets a test deterministically +// wait for the receiver to be ready instead of relying on a fixed sleep. +type recvSignalingChannel struct { + net.BroadcastChannel + recvRegistered chan struct{} + once sync.Once +} + +func (c *recvSignalingChannel) Recv(ctx context.Context, handler func(m net.Message)) { + c.BroadcastChannel.Recv(ctx, handler) + c.once.Do(func() { close(c.recvRegistered) }) +} diff --git a/pkg/tbtc/coordination_window_metrics.go b/pkg/tbtc/coordination_window_metrics.go index 2b57fc4c52..cedbf028fb 100644 --- a/pkg/tbtc/coordination_window_metrics.go +++ b/pkg/tbtc/coordination_window_metrics.go @@ -218,16 +218,13 @@ func (cwm *coordinationWindowMetrics) recordWalletCoordination( wm.WalletsFailed++ } - // Track leader leaderStr := leader.String() wm.Leaders[leaderStr]++ - // Track action type if actionType != "" { wm.ActionTypes[actionType]++ } - // Track faults faultDetails := make([]faultDetail, 0, len(faults)) for _, fault := range faults { faultTypeStr := fault.faultType.String() diff --git a/pkg/tbtc/deposit_sweep.go b/pkg/tbtc/deposit_sweep.go index 824ce29d28..8dad10af1f 100644 --- a/pkg/tbtc/deposit_sweep.go +++ b/pkg/tbtc/deposit_sweep.go @@ -51,13 +51,16 @@ const ( depositSweepBroadcastCheckDelay = 1 * time.Minute ) +// DepositKey identifies a deposit by the outpoint of its funding transaction. +type DepositKey struct { + FundingTxHash bitcoin.Hash + FundingOutputIndex uint32 +} + // DepositSweepProposal represents a deposit sweep proposal issued by a // wallet's coordination leader. type DepositSweepProposal struct { - DepositsKeys []struct { - FundingTxHash bitcoin.Hash - FundingOutputIndex uint32 - } + DepositsKeys []DepositKey SweepTxFee *big.Int DepositsRevealBlocks []*big.Int } diff --git a/pkg/tbtc/deposit_sweep_test.go b/pkg/tbtc/deposit_sweep_test.go index c98f75a3c0..3c87bce393 100644 --- a/pkg/tbtc/deposit_sweep_test.go +++ b/pkg/tbtc/deposit_sweep_test.go @@ -42,10 +42,7 @@ func TestDepositSweepAction_Execute(t *testing.T) { } // depositsKeys will be needed to build the proposal instance. - depositsKeys := make([]struct { - FundingTxHash bitcoin.Hash - FundingOutputIndex uint32 - }, len(scenario.Deposits)) + depositsKeys := make([]DepositKey, len(scenario.Deposits)) // depositsExtraInfo will be needed to perform on-chain proposal // validation. @@ -66,10 +63,7 @@ func TestDepositSweepAction_Execute(t *testing.T) { t.Fatal(err) } - depositsKeys[i] = struct { - FundingTxHash bitcoin.Hash - FundingOutputIndex uint32 - }{ + depositsKeys[i] = DepositKey{ FundingTxHash: fundingTxHash, FundingOutputIndex: fundingOutputIndex, } diff --git a/pkg/tbtc/dkg.go b/pkg/tbtc/dkg.go index 177e225a18..997fe08093 100644 --- a/pkg/tbtc/dkg.go +++ b/pkg/tbtc/dkg.go @@ -506,7 +506,7 @@ func (de *dkgExecutor) registerSigner( de.groupParameters, ) if err != nil { - return nil, fmt.Errorf("failed to resolve final signing group members") + return nil, fmt.Errorf("failed to resolve final signing group members: [%w]", err) } // Just like the final and original group may differ, the diff --git a/pkg/tbtc/marshaling.go b/pkg/tbtc/marshaling.go index 02b5195e45..5483b43d0d 100644 --- a/pkg/tbtc/marshaling.go +++ b/pkg/tbtc/marshaling.go @@ -326,10 +326,7 @@ func (dsp *DepositSweepProposal) Unmarshal(bytes []byte) error { } depositsKeys := make( - []struct { - FundingTxHash bitcoin.Hash - FundingOutputIndex uint32 - }, + []DepositKey, len(pbMsg.DepositsKeys), ) for i, depositKey := range pbMsg.DepositsKeys { @@ -344,10 +341,7 @@ func (dsp *DepositSweepProposal) Unmarshal(bytes []byte) error { ) } - depositsKeys[i] = struct { - FundingTxHash bitcoin.Hash - FundingOutputIndex uint32 - }{ + depositsKeys[i] = DepositKey{ FundingTxHash: hash, FundingOutputIndex: depositKey.FundingOutputIndex, } diff --git a/pkg/tbtc/marshaling_test.go b/pkg/tbtc/marshaling_test.go index 32b6977f0a..6fcbdcf831 100644 --- a/pkg/tbtc/marshaling_test.go +++ b/pkg/tbtc/marshaling_test.go @@ -186,10 +186,7 @@ func TestCoordinationMessage_MarshalingRoundtrip(t *testing.T) { }, "with deposit sweep proposal": { proposal: &DepositSweepProposal{ - DepositsKeys: []struct { - FundingTxHash bitcoin.Hash - FundingOutputIndex uint32 - }{ + DepositsKeys: []DepositKey{ { FundingTxHash: parseHash("709b55bd3da0f5a838125bd0ee20c5bfdd7caba173912d4281cae816b79a201b"), FundingOutputIndex: 0, diff --git a/pkg/tbtc/moving_funds.go b/pkg/tbtc/moving_funds.go index 1e9c01b0a3..156a1977b9 100644 --- a/pkg/tbtc/moving_funds.go +++ b/pkg/tbtc/moving_funds.go @@ -363,6 +363,34 @@ func ValidateMovingFundsProposal( return nil } +// movingFundsSafetyMarginChain is the chain interface required to evaluate the +// moving funds safety margin and to determine whether a wallet is a pending +// moving funds target. +type movingFundsSafetyMarginChain interface { + BlockCounter() (chain.BlockCounter, error) + + GetWallet(walletPublicKeyHash [20]byte) (*WalletChainData, error) + + GetMovingFundsParameters() ( + txMaxTotalFee uint64, + dustThreshold uint64, + timeoutResetDelay uint32, + timeout uint32, + timeoutSlashingAmount *big.Int, + timeoutNotifierRewardMultiplier uint32, + commitmentGasOffset uint16, + sweepTxMaxTotalFee uint64, + sweepTimeout uint32, + sweepTimeoutSlashingAmount *big.Int, + sweepTimeoutNotifierRewardMultiplier uint32, + err error, + ) + + PastMovingFundsCommitmentSubmittedEvents( + filter *MovingFundsCommitmentSubmittedEventFilter, + ) ([]*MovingFundsCommitmentSubmittedEvent, error) +} + // ValidateMovingFundsSafetyMargin checks if the moving funds safety margin // is in force. // @@ -380,30 +408,7 @@ func ValidateMovingFundsProposal( // wallets. In this case a longer safety margin should be used. func ValidateMovingFundsSafetyMargin( walletPublicKeyHash [20]byte, - chain interface { - BlockCounter() (chain.BlockCounter, error) - - GetWallet(walletPublicKeyHash [20]byte) (*WalletChainData, error) - - GetMovingFundsParameters() ( - txMaxTotalFee uint64, - dustThreshold uint64, - timeoutResetDelay uint32, - timeout uint32, - timeoutSlashingAmount *big.Int, - timeoutNotifierRewardMultiplier uint32, - commitmentGasOffset uint16, - sweepTxMaxTotalFee uint64, - sweepTimeout uint32, - sweepTimeoutSlashingAmount *big.Int, - sweepTimeoutNotifierRewardMultiplier uint32, - err error, - ) - - PastMovingFundsCommitmentSubmittedEvents( - filter *MovingFundsCommitmentSubmittedEventFilter, - ) ([]*MovingFundsCommitmentSubmittedEvent, error) - }, + chain movingFundsSafetyMarginChain, ) error { // In most cases the safety margin of 24 hours should be enough. It will // allow the wallet to sweep the last deposits that were made before the @@ -471,30 +476,7 @@ func (mfa *movingFundsAction) actionType() WalletActionType { func isWalletPendingMovingFundsTarget( walletPublicKeyHash [20]byte, - chain interface { - BlockCounter() (chain.BlockCounter, error) - - GetWallet(walletPublicKeyHash [20]byte) (*WalletChainData, error) - - GetMovingFundsParameters() ( - txMaxTotalFee uint64, - dustThreshold uint64, - timeoutResetDelay uint32, - timeout uint32, - timeoutSlashingAmount *big.Int, - timeoutNotifierRewardMultiplier uint32, - commitmentGasOffset uint16, - sweepTxMaxTotalFee uint64, - sweepTimeout uint32, - sweepTimeoutSlashingAmount *big.Int, - sweepTimeoutNotifierRewardMultiplier uint32, - err error, - ) - - PastMovingFundsCommitmentSubmittedEvents( - filter *MovingFundsCommitmentSubmittedEventFilter, - ) ([]*MovingFundsCommitmentSubmittedEvent, error) - }, + chain movingFundsSafetyMarginChain, ) (bool, error) { blockCounter, err := chain.BlockCounter() if err != nil { diff --git a/pkg/tbtc/wallet.go b/pkg/tbtc/wallet.go index ca346dec69..82719eca30 100644 --- a/pkg/tbtc/wallet.go +++ b/pkg/tbtc/wallet.go @@ -36,18 +36,18 @@ const ( // ParseWalletActionType parses the given value into a WalletActionType. func ParseWalletActionType(value uint8) (WalletActionType, error) { - switch value { - case 0: + switch WalletActionType(value) { + case ActionNoop: return ActionNoop, nil - case 1: + case ActionHeartbeat: return ActionHeartbeat, nil - case 2: + case ActionDepositSweep: return ActionDepositSweep, nil - case 3: + case ActionRedemption: return ActionRedemption, nil - case 4: + case ActionMovingFunds: return ActionMovingFunds, nil - case 5: + case ActionMovedFundsSweep: return ActionMovedFundsSweep, nil default: return 0, fmt.Errorf("unknown wallet action type [%v]", value) diff --git a/pkg/tbtcpg/chain.go b/pkg/tbtcpg/chain.go index 01e519c462..1ff35b53dd 100644 --- a/pkg/tbtcpg/chain.go +++ b/pkg/tbtcpg/chain.go @@ -157,7 +157,8 @@ type Chain interface { proposal *tbtc.MovingFundsProposal, ) error - // Submits the moving funds target wallets commitment. + // SubmitMovingFundsCommitment submits the moving funds target wallets + // commitment. SubmitMovingFundsCommitment( walletPublicKeyHash [20]byte, walletMainUTXO bitcoin.UnspentTransactionOutput, @@ -174,8 +175,8 @@ type Chain interface { proposal *tbtc.MovedFundsSweepProposal, ) error - // Computes the moving funds commitment hash from the provided public key - // hashes of target wallets. + // ComputeMovingFundsCommitmentHash computes the moving funds commitment hash + // from the provided public key hashes of target wallets. ComputeMovingFundsCommitmentHash(targetWallets [][20]byte) [32]byte // GetRedemptionDelay returns the processing delay for the given redemption. diff --git a/pkg/tbtcpg/deposit_sweep.go b/pkg/tbtcpg/deposit_sweep.go index b5c2346a25..d52c9b1727 100644 --- a/pkg/tbtcpg/deposit_sweep.go +++ b/pkg/tbtcpg/deposit_sweep.go @@ -166,7 +166,7 @@ func FindDeposits( // The filterStartBlock parameter controls the earliest block from which // deposit-revealed events are queried. func findDeposits( - fnLogger log.StandardLogger, + taskLogger log.StandardLogger, chain Chain, btcChain bitcoin.Chain, walletPublicKeyHash [20]byte, @@ -175,7 +175,7 @@ func findDeposits( skipUnconfirmed bool, filterStartBlock uint64, ) ([]*Deposit, error) { - fnLogger.Infof("reading revealed deposits from chain") + taskLogger.Infof("reading revealed deposits from chain") depositMinAgeSeconds, err := chain.GetDepositMinAge() if err != nil { @@ -201,14 +201,14 @@ func findDeposits( ) } - fnLogger.Infof("found [%d] DepositRevealed events", len(depositRevealedEvents)) + taskLogger.Infof("found [%d] DepositRevealed events", len(depositRevealedEvents)) // Take the oldest first sort.SliceStable(depositRevealedEvents, func(i, j int) bool { return depositRevealedEvents[i].BlockNumber < depositRevealedEvents[j].BlockNumber }) - fnLogger.Infof("getting deposits details") + taskLogger.Infof("getting deposits details") resultSliceCapacity := len(depositRevealedEvents) if maxNumberOfDeposits > 0 { @@ -227,7 +227,7 @@ func findDeposits( depositKey := chain.BuildDepositKey(event.FundingTxHash, event.FundingOutputIndex) depositKeyStr := depositKey.Text(16) - fnLogger.Debugf("getting details of deposit [%s]", depositKeyStr) + taskLogger.Debugf("getting details of deposit [%s]", depositKeyStr) depositRequest, found, err := chain.GetDepositRequest( event.FundingTxHash, @@ -249,26 +249,26 @@ func findDeposits( matureAt := depositRequest.RevealedAt.Add(depositMinAge) if !timeNow.After(matureAt) { - fnLogger.Infof("deposit [%s] is not old enough", depositKeyStr) + taskLogger.Infof("deposit [%s] is not old enough", depositKeyStr) continue } isSwept := depositRequest.SweptAt.Unix() != 0 if skipSwept && isSwept { - fnLogger.Debugf("deposit [%s] is already swept", depositKeyStr) + taskLogger.Debugf("deposit [%s] is already swept", depositKeyStr) continue } confirmations, err := btcChain.GetTransactionConfirmations(event.FundingTxHash) if err != nil { - fnLogger.Errorf( + taskLogger.Errorf( "failed to get bitcoin transaction confirmations: [%v]", err, ) } if skipUnconfirmed && confirmations < tbtc.DepositSweepRequiredFundingTxConfirmations { - fnLogger.Debugf( + taskLogger.Debugf( "deposit [%s] funding transaction doesn't have enough confirmations: [%d/%d]", depositKeyStr, confirmations, @@ -522,18 +522,12 @@ func (dst *DepositSweepTask) ProposeDepositsSweep( taskLogger.Infof("sweep transaction fee: [%d]", fee) - depositsKeys := make([]struct { - FundingTxHash bitcoin.Hash - FundingOutputIndex uint32 - }, len(deposits)) + depositsKeys := make([]tbtc.DepositKey, len(deposits)) depositsRevealBlocks := make([]*big.Int, len(deposits)) for i, deposit := range deposits { - depositsKeys[i] = struct { - FundingTxHash bitcoin.Hash - FundingOutputIndex uint32 - }{ + depositsKeys[i] = tbtc.DepositKey{ FundingTxHash: deposit.FundingTxHash, FundingOutputIndex: deposit.FundingOutputIndex, } @@ -610,7 +604,7 @@ func EstimateDepositsSweepFee( } else { sweepMaxSize, err := chain.GetDepositSweepMaxSize() if err != nil { - return nil, fmt.Errorf("cannot get sweep max size: [%v]", sweepMaxSize) + return nil, fmt.Errorf("cannot get sweep max size: [%v]", err) } for i := 1; i <= int(sweepMaxSize); i++ { diff --git a/pkg/tbtcpg/internal/test/marshaling.go b/pkg/tbtcpg/internal/test/marshaling.go index 91c390df6e..8a3b9e86ce 100644 --- a/pkg/tbtcpg/internal/test/marshaling.go +++ b/pkg/tbtcpg/internal/test/marshaling.go @@ -170,10 +170,7 @@ func (dsp *depositSweepProposal) convert() ( copy(walletPublicKeyHash[:], hexToSlice(dsp.WalletPublicKeyHash)) } - result.DepositsKeys = make([]struct { - FundingTxHash bitcoin.Hash - FundingOutputIndex uint32 - }, len(dsp.DepositsKeys)) + result.DepositsKeys = make([]tbtc.DepositKey, len(dsp.DepositsKeys)) for i, depositKey := range dsp.DepositsKeys { fundingTxHash, err := bitcoin.NewHashFromString(depositKey.FundingTxHash, bitcoin.ReversedByteOrder) if err != nil { diff --git a/pkg/tbtcpg/moved_funds_sweep.go b/pkg/tbtcpg/moved_funds_sweep.go index 628669ed2a..429c0b39ab 100644 --- a/pkg/tbtcpg/moved_funds_sweep.go +++ b/pkg/tbtcpg/moved_funds_sweep.go @@ -392,24 +392,10 @@ func EstimateMovedFundsSweepFee( AddPublicKeyHashInputs(inputCount, true). AddPublicKeyHashOutputs(1, true) - transactionSize, err := sizeEstimator.VirtualSize() - if err != nil { - return 0, fmt.Errorf( - "cannot estimate transaction virtual size: [%v]", - err, - ) - } - - feeEstimator := bitcoin.NewTransactionFeeEstimator(btcChain) - - totalFee, err := feeEstimator.EstimateFee(transactionSize) - if err != nil { - return 0, fmt.Errorf("cannot estimate transaction fee: [%v]", err) - } - - if uint64(totalFee) > sweepTxMaxTotalFee { - return 0, ErrSweepTxFeeTooHigh - } - - return totalFee, nil + return estimateCappedFee( + btcChain, + sizeEstimator, + sweepTxMaxTotalFee, + ErrSweepTxFeeTooHigh, + ) } diff --git a/pkg/tbtcpg/moving_funds.go b/pkg/tbtcpg/moving_funds.go index 22a842abc8..93eab148c5 100644 --- a/pkg/tbtcpg/moving_funds.go +++ b/pkg/tbtcpg/moving_funds.go @@ -626,17 +626,15 @@ func (mft *MovingFundsTask) ActionType() tbtc.WalletActionType { return tbtc.ActionMovingFunds } -// EstimateMovingFundsFee estimates fee for the moving funds transaction that -// moves funds from the source wallet to target wallets. -func EstimateMovingFundsFee( +// estimateCappedFee estimates the transaction fee for a transaction of the +// virtual size produced by the given size estimator. It returns feeTooHighErr +// if the estimated fee exceeds maxTotalFee. +func estimateCappedFee( btcChain bitcoin.Chain, - targetWalletsCount int, - txMaxTotalFee uint64, + sizeEstimator *bitcoin.TransactionSizeEstimator, + maxTotalFee uint64, + feeTooHighErr error, ) (int64, error) { - sizeEstimator := bitcoin.NewTransactionSizeEstimator(). - AddPublicKeyHashInputs(1, true). - AddPublicKeyHashOutputs(targetWalletsCount, true) - transactionSize, err := sizeEstimator.VirtualSize() if err != nil { return 0, fmt.Errorf( @@ -652,9 +650,23 @@ func EstimateMovingFundsFee( return 0, fmt.Errorf("cannot estimate transaction fee: [%v]", err) } - if uint64(totalFee) > txMaxTotalFee { - return 0, ErrFeeTooHigh + if uint64(totalFee) > maxTotalFee { + return 0, feeTooHighErr } return totalFee, nil } + +// EstimateMovingFundsFee estimates fee for the moving funds transaction that +// moves funds from the source wallet to target wallets. +func EstimateMovingFundsFee( + btcChain bitcoin.Chain, + targetWalletsCount int, + txMaxTotalFee uint64, +) (int64, error) { + sizeEstimator := bitcoin.NewTransactionSizeEstimator(). + AddPublicKeyHashInputs(1, true). + AddPublicKeyHashOutputs(targetWalletsCount, true) + + return estimateCappedFee(btcChain, sizeEstimator, txMaxTotalFee, ErrFeeTooHigh) +} diff --git a/pkg/tbtcpg/redemptions.go b/pkg/tbtcpg/redemptions.go index 981e9a8eb7..ff7724ad83 100644 --- a/pkg/tbtcpg/redemptions.go +++ b/pkg/tbtcpg/redemptions.go @@ -253,7 +253,7 @@ func (rt *RedemptionTask) ProposeRedemption( } func findPendingRedemptions( - fnLogger log.StandardLogger, + taskLogger log.StandardLogger, chain Chain, walletPublicKeyHash [20]byte, currentBlockNumber uint64, @@ -323,9 +323,9 @@ func findPendingRedemptions( eventsSet[hexutils.Encode(redemptionKey.Bytes())] = event } - fnLogger.Infof("found [%d] RedemptionRequested events", len(eventsSet)) + taskLogger.Infof("found [%d] RedemptionRequested events", len(eventsSet)) - fnLogger.Infof("checking pending redemptions details") + taskLogger.Infof("checking pending redemptions details") pendingRedemptions := make([]*RedemptionRequest, 0) @@ -334,7 +334,7 @@ redemptionRequestedLoop: for redemptionKey, event := range eventsSet { eventIndex++ - fnLogger.Debugf( + taskLogger.Debugf( "getting pending redemption details [%s]", redemptionKey, ) @@ -352,7 +352,7 @@ redemptionRequestedLoop: ) } if !found { - fnLogger.Infof( + taskLogger.Infof( "redemption request [%s] is no longer pending", redemptionKey, ) @@ -411,7 +411,7 @@ redemptionRequestedLoop: minAge = delay } - fnLogger.Infof( + taskLogger.Infof( "minimum age for redemption request [%s] is [%v]", redemption.RedemptionKey, minAge, @@ -428,7 +428,7 @@ redemptionRequestedLoop: // Check if timeout passed for the redemption request. if pendingRedemption.RequestedAt.Before(redemptionRequestsRangeStartTimestamp) { - fnLogger.Infof( + taskLogger.Infof( "redemption request [%s] has already timed out", pendingRedemption.RedemptionKey, ) @@ -448,7 +448,7 @@ redemptionRequestedLoop: // Check if enough time elapsed since the redemption request. if pendingRedemption.RequestedAt.After(rangeEndTimestamp) { - fnLogger.Infof( + taskLogger.Infof( "redemption request [%s] is not old enough", pendingRedemption.RedemptionKey, ) diff --git a/tools.go b/tools.go index e0dacdde1c..40225ad7d2 100644 --- a/tools.go +++ b/tools.go @@ -1,8 +1,9 @@ //go:build tools -// tools.go: Build-time dependencies required for Ethereum bindings generation -// These are imported to ensure they remain in go.mod and go.sum even though -// they're not directly used in the runtime code. +// tools.go pins build-time-only dependencies that would otherwise be dropped +// by `go mod tidy`. They are anchored here with blank imports so they remain in +// go.mod and go.sum for reproducible builds, even though they are not +// referenced directly by runtime or generated code. package tools import (