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/clientinfo/performance_test.go b/pkg/clientinfo/performance_test.go index 5ebf253288..6d2d0403de 100644 --- a/pkg/clientinfo/performance_test.go +++ b/pkg/clientinfo/performance_test.go @@ -430,3 +430,39 @@ func TestJoinFailureAndOnChainCountersRegistered(t *testing.T) { } } } + +// TestDepositSweepProofSubmissionCountersRegistered tests that the +// deposit-sweep proof-submission counters are registered upfront so they +// appear in the metrics endpoint before any increment. +func TestDepositSweepProofSubmissionCountersRegistered(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + registry := &Registry{keepclientinfo.NewRegistry(), ctx} + pm := NewPerformanceMetrics(ctx, registry) + + expectedCounters := []string{ + MetricDepositSweepProofSubmissionsTotal, + MetricDepositSweepProofSubmissionsSuccessTotal, + MetricDepositSweepProofSubmissionsFailedTotal, + } + + for _, counterName := range expectedCounters { + pm.countersMutex.RLock() + _, exists := pm.counters[counterName] + pm.countersMutex.RUnlock() + if !exists { + t.Errorf("counter %s should be registered upfront", counterName) + continue + } + + if value := pm.GetCounterValue(counterName); value != 0 { + t.Errorf("counter %s should start at 0, got %v", counterName, value) + } + + pm.IncrementCounter(counterName, 1) + if value := pm.GetCounterValue(counterName); value != 1 { + t.Errorf("counter %s should increment to 1, got %v", counterName, value) + } + } +} diff --git a/pkg/maintainer/spv/deposit_sweep.go b/pkg/maintainer/spv/deposit_sweep.go index f8405e1576..d2414c9bdc 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 @@ -26,7 +27,7 @@ func SubmitDepositSweepProof( btcChain, spvChain, bitcoin.AssembleSpvProof, - getGlobalMetricsRecorder(), + nil, ) } @@ -42,12 +43,12 @@ func submitDepositSweepProof( ) 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 +62,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 +77,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 +92,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 +102,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 +119,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 +252,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 +297,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..8c2f85b8b5 100644 --- a/pkg/maintainer/spv/deposit_sweep_test.go +++ b/pkg/maintainer/spv/deposit_sweep_test.go @@ -96,7 +96,7 @@ func TestSubmitDepositSweepProof(t *testing.T) { btcChain, spvChain, mockSpvProofAssembler, - getGlobalMetricsRecorder(), + nil, ) if err != nil { t.Fatal(err) diff --git a/pkg/maintainer/spv/moved_funds_sweep.go b/pkg/maintainer/spv/moved_funds_sweep.go index 417a1f5347..4a93d83c67 100644 --- a/pkg/maintainer/spv/moved_funds_sweep.go +++ b/pkg/maintainer/spv/moved_funds_sweep.go @@ -137,20 +137,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 +202,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..a102a1581f 100644 --- a/pkg/maintainer/spv/moving_funds.go +++ b/pkg/maintainer/spv/moving_funds.go @@ -133,20 +133,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 +188,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..690edc538c 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. @@ -31,7 +24,7 @@ func SubmitRedemptionProof( btcChain, spvChain, bitcoin.AssembleSpvProof, - getGlobalMetricsRecorder(), + nil, ) } @@ -167,20 +160,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 +205,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..b16d70b474 100644 --- a/pkg/maintainer/spv/redemptions_test.go +++ b/pkg/maintainer/spv/redemptions_test.go @@ -78,7 +78,7 @@ func TestSubmitRedemptionProof(t *testing.T) { btcChain, spvChain, mockSpvProofAssembler, - getGlobalMetricsRecorder(), + nil, ) if err != nil { t.Fatal(err) diff --git a/pkg/maintainer/spv/spv.go b/pkg/maintainer/spv/spv.go index 133e2f48c8..4ba7440396 100644 --- a/pkg/maintainer/spv/spv.go +++ b/pkg/maintainer/spv/spv.go @@ -16,7 +16,6 @@ import ( "encoding/hex" "fmt" "math/big" - "sync" "time" "github.com/keep-network/keep-core/pkg/tbtc" @@ -54,34 +53,6 @@ func Initialize( 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 { @@ -467,6 +438,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 088c619883..ae3341c604 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" @@ -471,3 +472,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 87f0094d48..184042d46f 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..88ec3413fa 100644 --- a/pkg/tbtc/deposit_sweep.go +++ b/pkg/tbtc/deposit_sweep.go @@ -51,13 +51,24 @@ const ( depositSweepBroadcastCheckDelay = 1 * time.Minute ) +// DepositKey identifies a deposit by the outpoint of its funding transaction. +// +// Note: DepositKey is a named type; it replaced the anonymous struct +// previously used inline as the element type of +// DepositSweepProposal.DepositsKeys. Go does not allow assigning an +// anonymous-struct-typed slice literal to a named-struct-typed slice field, +// so code outside this module that builds a DepositSweepProposal from the +// old anonymous struct literal must switch to constructing []DepositKey +// values instead. +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 4385f25fd1..a073cd47b8 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 8cfdbc8072..1c3fe17255 100644 --- a/pkg/tbtc/moving_funds.go +++ b/pkg/tbtc/moving_funds.go @@ -350,6 +350,21 @@ 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() (MovingFundsParameters, error) + + PastMovingFundsCommitmentSubmittedEvents( + filter *MovingFundsCommitmentSubmittedEventFilter, + ) ([]*MovingFundsCommitmentSubmittedEvent, error) +} + // ValidateMovingFundsSafetyMargin checks if the moving funds safety margin // is in force. // @@ -367,17 +382,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() (MovingFundsParameters, 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 @@ -444,17 +449,7 @@ func (mfa *movingFundsAction) actionType() WalletActionType { func isWalletPendingMovingFundsTarget( walletPublicKeyHash [20]byte, - chain interface { - BlockCounter() (chain.BlockCounter, error) - - GetWallet(walletPublicKeyHash [20]byte) (*WalletChainData, error) - - GetMovingFundsParameters() (MovingFundsParameters, 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 af939852e5..a1bead42c5 100644 --- a/pkg/tbtcpg/chain.go +++ b/pkg/tbtcpg/chain.go @@ -133,7 +133,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, @@ -150,8 +151,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/chain_test.go b/pkg/tbtcpg/chain_test.go index cdff0f01e3..48754b3972 100644 --- a/pkg/tbtcpg/chain_test.go +++ b/pkg/tbtcpg/chain_test.go @@ -57,6 +57,7 @@ type LocalChain struct { operatorIDs map[chain.Address]uint32 redemptionDelays map[[32]byte]time.Duration depositMinAge uint32 + depositSweepMaxSizeErr error } func NewLocalChain() *LocalChain { @@ -870,9 +871,26 @@ func (lc *LocalChain) SetRedemptionRequestMinAge(redemptionRequestMinAge uint32) } func (lc *LocalChain) GetDepositSweepMaxSize() (uint16, error) { + lc.mutex.Lock() + defer lc.mutex.Unlock() + + if lc.depositSweepMaxSizeErr != nil { + return 0, lc.depositSweepMaxSizeErr + } + panic("unsupported") } +// SetDepositSweepMaxSizeError configures the error GetDepositSweepMaxSize +// returns, allowing tests to exercise the max-size-lookup failure path +// without a real chain implementation. +func (lc *LocalChain) SetDepositSweepMaxSizeError(err error) { + lc.mutex.Lock() + defer lc.mutex.Unlock() + + lc.depositSweepMaxSizeErr = err +} + func (lc *LocalChain) BlockCounter() (chain.BlockCounter, error) { lc.mutex.Lock() defer lc.mutex.Unlock() diff --git a/pkg/tbtcpg/deposit_sweep.go b/pkg/tbtcpg/deposit_sweep.go index 9c14863a79..ffac245ab6 100644 --- a/pkg/tbtcpg/deposit_sweep.go +++ b/pkg/tbtcpg/deposit_sweep.go @@ -146,7 +146,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, @@ -155,7 +155,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 { @@ -181,14 +181,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 { @@ -207,7 +207,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, @@ -229,26 +229,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, @@ -503,18 +503,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, } @@ -592,7 +586,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/deposit_sweep_fee_test.go b/pkg/tbtcpg/deposit_sweep_fee_test.go index 185da44781..6e7e40144c 100644 --- a/pkg/tbtcpg/deposit_sweep_fee_test.go +++ b/pkg/tbtcpg/deposit_sweep_fee_test.go @@ -1,6 +1,7 @@ package tbtcpg_test import ( + "errors" "strings" "testing" @@ -34,7 +35,9 @@ func sweepVirtualSize(t *testing.T, depositsCount int) int64 { // the informational SatPerVByteFee and the TotalFee actually broadcast on-chain // are asserted, and multi-deposit sweeps (where transactionSize grows // sub-linearly while totalMaxFee grows linearly) are exercised on both the happy -// path and the floor-exceeds-cap error branch. +// path and the floor-exceeds-cap error branch. A depositsCount of 0 is also +// covered, verifying the max-size-lookup failure is reported with its real +// underlying cause rather than the zero-value size. func TestEstimateDepositsSweepFee_MinimumFloorAndBuffer(t *testing.T) { // Virtual sizes used to pin the cap and the expected total fee (the on-chain // value) relative to the fee rate. The cap and expected-total expectations @@ -48,6 +51,7 @@ func TestEstimateDepositsSweepFee_MinimumFloorAndBuffer(t *testing.T) { depositsCount int estimateSatPerVByte int64 perDepositMaxFee uint64 + sweepMaxSizeErr error expectedSatPerVByteFee int64 expectedTotalFee int64 expectErrorContains string @@ -113,12 +117,25 @@ func TestEstimateDepositsSweepFee_MinimumFloorAndBuffer(t *testing.T) { perDepositMaxFee: uint64(size3), expectErrorContains: "minimum safe transaction fee", }, + "depositsCount of 0 with a failing max size lookup returns the wrapped error": { + depositsCount: 0, + // A depositsCount of 0 takes the "estimate for every count" branch, + // which looks up the max sweep size first. Inject a distinctive + // underlying error so the assertion below fails if that cause is + // ever dropped again (the bug this guards against formatted the + // zero-value max size instead of the real error). + sweepMaxSizeErr: errors.New("boom"), + expectErrorContains: "cannot get sweep max size: [boom]", + }, } for name, test := range tests { t.Run(name, func(t *testing.T) { tbtcChain := tbtcpg.NewLocalChain() tbtcChain.SetDepositParameters(0, 0, test.perDepositMaxFee, 0) + if test.sweepMaxSizeErr != nil { + tbtcChain.SetDepositSweepMaxSizeError(test.sweepMaxSizeErr) + } btcChain := tbtcpg.NewLocalBitcoinChain() btcChain.SetEstimateSatPerVByteFee(1, test.estimateSatPerVByte) 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 47b188f587..40fc2b2bdb 100644 --- a/pkg/tbtcpg/moved_funds_sweep.go +++ b/pkg/tbtcpg/moved_funds_sweep.go @@ -393,32 +393,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 - } - - // Enforce the safe minimum fee rate and buffer so a non-RBF moved funds - // sweep transaction is never broadcast below the floor where it could get - // stuck and jam the wallet. - totalFee, err = applyWalletTxFeeFloor(totalFee, transactionSize, sweepTxMaxTotalFee) - if err != nil { - return 0, err - } - - return totalFee, nil + return estimateCappedFee( + btcChain, + sizeEstimator, + sweepTxMaxTotalFee, + ErrSweepTxFeeTooHigh, + ) } diff --git a/pkg/tbtcpg/moving_funds.go b/pkg/tbtcpg/moving_funds.go index 9f78e4a871..ebacd8493c 100644 --- a/pkg/tbtcpg/moving_funds.go +++ b/pkg/tbtcpg/moving_funds.go @@ -628,17 +628,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( @@ -654,17 +652,31 @@ 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 } // Enforce the safe minimum fee rate and buffer so a non-RBF moving funds // transaction is never broadcast below the floor where it could get stuck // and jam the wallet. - totalFee, err = applyWalletTxFeeFloor(totalFee, transactionSize, txMaxTotalFee) + totalFee, err = applyWalletTxFeeFloor(totalFee, transactionSize, maxTotalFee) if err != nil { return 0, err } 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 a52d00eeb9..a9ede0bef8 100644 --- a/pkg/tbtcpg/redemptions.go +++ b/pkg/tbtcpg/redemptions.go @@ -292,7 +292,7 @@ func (rt *RedemptionTask) ProposeRedemption( } func findPendingRedemptions( - fnLogger log.StandardLogger, + taskLogger log.StandardLogger, chain Chain, walletPublicKeyHash [20]byte, currentBlockNumber uint64, @@ -362,9 +362,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) @@ -373,7 +373,7 @@ redemptionRequestedLoop: for redemptionKey, event := range eventsSet { eventIndex++ - fnLogger.Debugf( + taskLogger.Debugf( "getting pending redemption details [%s]", redemptionKey, ) @@ -391,7 +391,7 @@ redemptionRequestedLoop: ) } if !found { - fnLogger.Infof( + taskLogger.Infof( "redemption request [%s] is no longer pending", redemptionKey, ) @@ -450,7 +450,7 @@ redemptionRequestedLoop: minAge = delay } - fnLogger.Infof( + taskLogger.Infof( "minimum age for redemption request [%s] is [%v]", redemption.RedemptionKey, minAge, @@ -467,7 +467,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, ) @@ -487,7 +487,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 (