diff --git a/pkg/net/local/broadcast_channel_manager.go b/pkg/net/local/broadcast_channel_manager.go index 2d9a90b5df..20b70addfe 100644 --- a/pkg/net/local/broadcast_channel_manager.go +++ b/pkg/net/local/broadcast_channel_manager.go @@ -16,6 +16,7 @@ const RetransmissionTick = 50 * time.Millisecond var broadcastChannelsMutex sync.Mutex var broadcastChannels map[string][]*localChannel +var broadcastChannelCancels map[string][]context.CancelFunc // getBroadcastChannel returns a BroadcastChannel designed to mediate between local // participants. It delivers all messages sent to the channel through its @@ -31,12 +32,19 @@ func getBroadcastChannel( if broadcastChannels == nil { broadcastChannels = make(map[string][]*localChannel) } + if broadcastChannelCancels == nil { + broadcastChannelCancels = make(map[string][]context.CancelFunc) + } _, exists := broadcastChannels[name] if !exists { broadcastChannels[name] = make([]*localChannel, 0) + broadcastChannelCancels[name] = make([]context.CancelFunc, 0) } + tickerCtx, cancelTicker := context.WithCancel(context.Background()) + broadcastChannelCancels[name] = append(broadcastChannelCancels[name], cancelTicker) + identifier := randomLocalIdentifier() channel := &localChannel{ name: name, @@ -47,7 +55,7 @@ func getBroadcastChannel( unmarshalersMutex: sync.Mutex{}, unmarshalersByType: make(map[string]func() net.TaggedUnmarshaler, 0), retransmissionTicker: retransmission.NewTimeTicker( - context.Background(), RetransmissionTick, + tickerCtx, RetransmissionTick, ), } broadcastChannels[name] = append(broadcastChannels[name], channel) @@ -66,3 +74,21 @@ func broadcastMessage(name string, message net.Message) error { return nil } + +// ReleaseBroadcastChannel cancels every outstanding retransmission ticker +// registered under name and removes name's entry from the registry, so a +// later invocation reusing name starts from an empty registry regardless of +// whether an earlier invocation's leader was still retransmitting. Callers +// that create broadcast channels in tests should call this from t.Cleanup, +// passing the same name they created the channel(s) under. +func ReleaseBroadcastChannel(name string) { + broadcastChannelsMutex.Lock() + defer broadcastChannelsMutex.Unlock() + + for _, cancel := range broadcastChannelCancels[name] { + cancel() + } + + delete(broadcastChannels, name) + delete(broadcastChannelCancels, name) +} diff --git a/pkg/net/local/broadcast_channel_manager_test.go b/pkg/net/local/broadcast_channel_manager_test.go new file mode 100644 index 0000000000..173a449759 --- /dev/null +++ b/pkg/net/local/broadcast_channel_manager_test.go @@ -0,0 +1,121 @@ +package local + +import ( + "context" + "testing" + "time" + + "github.com/keep-network/keep-core/pkg/net" + "github.com/keep-network/keep-core/pkg/operator" +) + +// TestReleaseBroadcastChannel verifies ReleaseBroadcastChannel's actual +// effect, not just that it can be called: a released channel's +// retransmission ticker stops firing, and a name reused after release only +// delivers to the newly-registered channel, not any stale one left over +// from before the release. +// +// Delivery is observed via a raw messageHandler registered directly +// (bypassing Recv's retransmission.WithRetransmissionSupport dedup wrapper), +// because the standard retransmission strategy resends the same message +// with the same sequence number on every tick, and the dedup layer collapses +// those to a single callback invocation - counting through it would make +// "the ticker kept firing" indistinguishable from "the ticker fired once". +func TestReleaseBroadcastChannel(t *testing.T) { + // Use a name unique to this test (not a shared literal like + // "channel name", which broadcast_channel_test.go also uses) so a + // channel this test forgets to release can never cross-contaminate + // another test file's assertions in the same test binary. + name := t.Name() + t.Cleanup(func() { ReleaseBroadcastChannel(name) }) + + _, pubKey, err := operator.GenerateKeyPair(DefaultCurve) + if err != nil { + t.Fatal(err) + } + + createChannel := func(name string) *localChannel { + ch := getBroadcastChannel(name, pubKey) + lc := ch.(*localChannel) + lc.SetUnmarshaler(func() net.TaggedUnmarshaler { + return &mockNetMessage{} + }) + return lc + } + + registerRawHandler := func(lc *localChannel) <-chan net.Message { + handler := &messageHandler{ + ctx: context.Background(), + channel: make(chan net.Message, 64), + } + lc.messageHandlersMutex.Lock() + lc.messageHandlers = append(lc.messageHandlers, handler) + lc.messageHandlersMutex.Unlock() + return handler.channel + } + + // drain counts every message received on ch during window; used to + // count raw delivery attempts (one per ticker firing), not distinct + // messages. + drain := func(ch <-chan net.Message, window time.Duration) int { + deadline := time.After(window) + count := 0 + for { + select { + case <-ch: + count++ + case <-deadline: + return count + } + } + } + + // 1. Open a channel, let its retransmission ticker fire a few times, + // release it, then assert no further deliveries occur. + ch1 := createChannel(name) + ch1Deliveries := registerRawHandler(ch1) + + if err := ch1.Send(context.Background(), &mockNetMessage{}); err != nil { + t.Fatal(err) + } + + if got := drain(ch1Deliveries, RetransmissionTick*3); got <= 1 { + t.Fatalf( + "expected repeated ticker deliveries before release, got %d", + got, + ) + } + + ReleaseBroadcastChannel(name) + + if got := drain(ch1Deliveries, RetransmissionTick*3); got != 0 { + t.Errorf("expected no deliveries after release, got %d", got) + } + + // 2. Open a new channel under the same, just-released name, send a + // message on it, and assert only the new channel's handler receives + // it - proving the old channel's registration was actually dropped + // by the release, not merely shadowed by a map pointer swap. + ch2 := createChannel(name) + ch2Deliveries := registerRawHandler(ch2) + + if err := ch2.Send(context.Background(), &mockNetMessage{}); err != nil { + t.Fatal(err) + } + + if got := drain(ch2Deliveries, RetransmissionTick*3); got <= 1 { + t.Errorf( + "expected repeated ticker deliveries from the new channel, got %d", + got, + ) + } + if got := drain(ch1Deliveries, RetransmissionTick*2); got != 0 { + t.Errorf( + "expected the released channel to receive nothing further, got %d", + got, + ) + } + + // 3. Releasing a name with zero registered channels is a safe no-op. + ReleaseBroadcastChannel("nonexistent") +} diff --git a/pkg/net/retransmission/ticker.go b/pkg/net/retransmission/ticker.go index a9e3e8e802..b794179e49 100644 --- a/pkg/net/retransmission/ticker.go +++ b/pkg/net/retransmission/ticker.go @@ -75,9 +75,10 @@ func (t *Ticker) start() { t.handlersMutex.Unlock() } - for ctx := range t.handlers { - delete(t.handlers, ctx) - } + t.handlersMutex.Lock() + defer t.handlersMutex.Unlock() + + clear(t.handlers) } func (t *Ticker) onTick(ctx context.Context, fn func()) { diff --git a/pkg/tbtc/coordination.go b/pkg/tbtc/coordination.go index 1840b54042..5bc3e02fe3 100644 --- a/pkg/tbtc/coordination.go +++ b/pkg/tbtc/coordination.go @@ -600,8 +600,11 @@ func (ce *coordinationExecutor) getActionsChecklist( var actions []WalletActionType - // Redemption action is a priority action and should be checked on every - // coordination window. + // Redemption is a priority action and should be checked on every + // coordination window: unlike MovingFunds (and, pre-activation, the + // sweep actions) which remain frequency-gated below for throughput + // reasons, an unredeemed request risks user funds being stuck, not + // just throughput. actions = append(actions, ActionRedemption) // Other actions should be checked with a lower frequency. The default @@ -640,23 +643,28 @@ func (ce *coordinationExecutor) getActionsChecklist( } } - // Reservation actions (acceptance, re-anchor) checklist gate is deliberately - // config-independent and height-only so every operator computes an - // identical checklist once the network-wide activation block passes. - // If the checklist depended on each operator's local config flag, - // operators with different local settings would compute different checklists - // and fault each other's proposals via FaultLeaderMistake. Checklist - // agreement is achieved because the gate ignores local config and uses only - // globally-observable chain height. Config.Reservations.Enabled controls - // only whether THIS operator originates (leader-proposes) new reservation - // actions and whether its reservation watchers run - it does NOT prevent - // this operator from evaluating/countersigning another leader's reservation - // proposal as a follower once the activation height passes, regardless of - // this operator's own local flag setting. Frequency-gated like - // DepositSweep/MovingFunds below the activation block: reservation - // acceptance/re-anchor windows are not as time-critical as redemption. - if coordinationBlock >= ReservationsActivationBlock && - windowIndex%frequencyWindows == 0 { + // Reservation actions (acceptance, re-anchor) are custody-critical like + // Redemption and are checked on every coordination window once the + // activation block is reached, not frequency-gated like the + // throughput-driven DepositSweep/MovedFundsSweep/MovingFunds actions + // above: a delayed reservation acceptance or re-anchor risks the + // on-chain ReservationActionTimeout backstop firing before the wallet + // subsystem gets a chance to act. There is no per-operator enable + // flag here; the activation block is a network-wide constant, which + // keeps leader and follower checklists in agreement without relying + // on local config (a follower whose local config diverged from the + // leader's would otherwise fault an honest leader's reservation + // proposal as FaultLeaderMistake because the action would not appear + // in its own checklist). + // + // Note: because getActionsChecklist appends reservation actions after + // Redemption/DepositSweep/MovedFundsSweep/MovingFunds and + // ProposalGenerator.Generate returns on the first checklist action + // that yields a proposal, a wallet with steady redemption/sweep + // traffic can still delay reservation acceptance/re-anchor even + // though the checklist entry itself is unconditional. This is an + // accepted tradeoff bounded by ReservationActionTimeout, not a bug. + if coordinationBlock >= ReservationsActivationBlock { actions = append(actions, ActionReservationAnchor) actions = append(actions, ActionReservationReanchor) } diff --git a/pkg/tbtc/coordination_test.go b/pkg/tbtc/coordination_test.go index 4623485107..365e72c508 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -173,116 +173,226 @@ func TestWatchCoordinationWindows(t *testing.T) { expectWindow(1800) } -func TestCoordinationExecutor_Coordinate(t *testing.T) { - // Uncompressed public key corresponding to the 20-byte public key hash: - // aa768412ceed10bd423c025542ca90071f9fb62d. - publicKeyHex, err := hex.DecodeString( - "0471e30bca60f6548d7b42582a478ea37ada63b402af7b3ddd57f0c95bb6843175" + - "aa0d2053a91a050a6797d85c38f2909cb7027f2344a01986aa2f9f8ca7a0c289", +// coordinationOperatorFixture bundles the per-operator state +// needed to run coordinationExecutor.coordinate as an independent +// in-process simulated node, its own local chain fake plus a broadcast +// channel shared with its peers the same way pkg/tbtc/node wires a real operator. +type coordinationOperatorFixture struct { + chain Chain + address chain.Address + channel net.BroadcastChannel + waitForBlockHeight func(ctx context.Context, blockHeight uint64) error +} + +// newCoordinationOperator builds one simulated operator shared by +// every coordinationExecutor.coordinate integration test in this file: a +// deterministic keypair (so leader election is reproducible across runs), a +// local chain fake wired to that keypair, and a broadcast channel joined to a +// local network shared by every operator in the same test so they exchange +// real coordinationMessage wire traffic. channelName need not be unique +// across test invocations: this registers a t.Cleanup that calls +// netlocal.ReleaseBroadcastChannel(channelName), which cancels that +// specific channel's retransmission ticker and clears the registry, so a +// later invocation reusing the same name starts from an empty registry regardless +// of whether an earlier invocation's leader was still retransmitting. +// channelName is passed as t.Name() purely so a leaked broadcast is easy to +// attribute to its source test. +func newCoordinationOperator( + t *testing.T, + privateKey int64, + coordinationBlock uint64, + channelName string, +) *coordinationOperatorFixture { + t.Helper() + + privateKeyBigInt := big.NewInt(privateKey) + x, y := local_v1.DefaultCurve.ScalarBaseMult(privateKeyBigInt.Bytes()) + + localChain := ConnectWithKey( + &operator.PrivateKey{ + PublicKey: operator.PublicKey{ + Curve: operator.Secp256k1, + X: x, + Y: y, + }, + D: privateKeyBigInt, + }, + 100*time.Millisecond, ) + + localChain.setBlockHashByNumber( + coordinationBlock-32, + "1422996cbcbc38fc924a46f4df5f9064279d3ab43396e58386dac9b87440d64f", + ) + + operatorAddress, err := localChain.operatorAddress() if err != nil { t.Fatal(err) } - // 20-byte public key hash corresponding to the public key above. - buffer, err := hex.DecodeString("aa768412ceed10bd423c025542ca90071f9fb62d") + _, operatorPublicKey, err := localChain.OperatorKeyPair() if err != nil { t.Fatal(err) } - var publicKeyHash [20]byte - copy(publicKeyHash[:], buffer) - - parseScript := func(script string) bitcoin.Script { - parsed, err := hex.DecodeString(script) - if err != nil { - t.Fatal(err) - } - return parsed + broadcastChannel, err := netlocal.ConnectWithKey(operatorPublicKey). + BroadcastChannelFor(channelName) + if err != nil { + t.Fatal(err) } + t.Cleanup(func() { netlocal.ReleaseBroadcastChannel(channelName) }) - coordinationBlock := uint64(900) - - type operatorFixture struct { - chain Chain - address chain.Address - channel net.BroadcastChannel - waitForBlockHeight func(ctx context.Context, blockHeight uint64) error - } - - generateOperator := func(privateKey int64) *operatorFixture { - // Generate operators with deterministic addresses that don't change - // between test runs. This is required to assert the leader selection. - privateKeyBigInt := big.NewInt(privateKey) - x, y := local_v1.DefaultCurve.ScalarBaseMult(privateKeyBigInt.Bytes()) - - localChain := ConnectWithKey( - &operator.PrivateKey{ - PublicKey: operator.PublicKey{ - Curve: operator.Secp256k1, - X: x, - Y: y, - }, - D: privateKeyBigInt, - }, - 100*time.Millisecond, - ) - - localChain.setBlockHashByNumber( - coordinationBlock-32, - "1422996cbcbc38fc924a46f4df5f9064279d3ab43396e58386dac9b87440d64f", - ) + broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { + return &coordinationMessage{} + }) - operatorAddress, err := localChain.operatorAddress() + waitForBlockHeight := func(ctx context.Context, blockHeight uint64) error { + blockCounter, err := localChain.BlockCounter() if err != nil { - t.Fatal(err) + return err } - _, operatorPublicKey, err := localChain.OperatorKeyPair() + // The local chain fake's block counter always starts at 0 + // regardless of coordinationBlock (see local_v1.BlockCounter), + // but every caller here only ever asks to wait for a height + // derived as coordinationBlock + a fixed small offset (e.g. + // window.activePhaseEndBlock()). Waiting on the raw absolute + // height would take the fake's block-rate multiplied by + // coordinationBlock itself - days of wall-clock time for the + // mainnet-scale coordinationBlock values these tests use - + // leaking the leader's goroutine and this waiter registration + // for the life of the test binary, since coordinate() only + // cancels this context on failure, not on success. Translating + // to the counter's own relative frame makes the wait actually + // reachable in a few seconds instead. + wait, err := blockCounter.BlockHeightWaiter(blockHeight - coordinationBlock) if err != nil { - t.Fatal(err) + return err } - broadcastChannel, err := netlocal.ConnectWithKey(operatorPublicKey). - BroadcastChannelFor("test") - if err != nil { - t.Fatal(err) + select { + case <-wait: + case <-ctx.Done(): } - broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { - return &coordinationMessage{} - }) + return nil + } - waitForBlockHeight := func(ctx context.Context, blockHeight uint64) error { - blockCounter, err := localChain.BlockCounter() - if err != nil { - return err - } + return &coordinationOperatorFixture{ + chain: localChain, + address: operatorAddress, + channel: broadcastChannel, + waitForBlockHeight: waitForBlockHeight, + } +} - wait, err := blockCounter.BlockHeightWaiter(blockHeight) - if err != nil { - return err - } +// coordinationReport captures one simulated operator's outcome +// from a single coordination round. +type coordinationReport struct { + operatorIndex int + result *coordinationResult + err error +} + +// runCoordinationRound runs coordinationExecutor.coordinate +// concurrently for every given operator against the same window - one +// goroutine per operator, sharing one proposalGenerator, membershipValidator, +// and protocolLatch across all three (the leader is the only goroutine that +// calls Generate; the latch is a shared, mutex-guarded execution counter, +// safe to share precisely because it does not serialize or order the +// goroutines -- the trailing protocolLatch.IsExecuting() == false assertion +// in each caller verifies all three balanced their Lock/Unlock) the same +// way a real node would have each operator drive its own executor in a +// separate process. Fails the test if not every operator reports within the +func runCoordinationRound( + t *testing.T, + operators []*coordinationOperatorFixture, + coordinatedWallet wallet, + proposalGenerator CoordinationProposalGenerator, + membershipValidator *group.MembershipValidator, + protocolLatch *generator.ProtocolLatch, + window *coordinationWindow, +) []*coordinationReport { + t.Helper() + + reportChan := make(chan *coordinationReport, len(operators)) + + for i, currentOperator := range operators { + go func(operatorIndex int, op *coordinationOperatorFixture) { + executor := newCoordinationExecutor( + op.chain, + coordinatedWallet, + coordinatedWallet.membersByOperator(op.address), + op.address, + proposalGenerator, + op.channel, + membershipValidator, + protocolLatch, + op.waitForBlockHeight, + ) - select { - case <-wait: - case <-ctx.Done(): + result, err := executor.coordinate(window) + + reportChan <- &coordinationReport{ + operatorIndex: operatorIndex, + result: result, + err: err, } + }(i+1, currentOperator) + } - return nil + deadline := time.After(30 * time.Second) + reports := make([]*coordinationReport, 0, len(operators)) + for len(reports) < len(operators) { + select { + case report := <-reportChan: + reports = append(reports, report) + case <-deadline: + t.Fatalf( + "timed out waiting for coordination reports; got %d of %d", + len(reports), + len(operators), + ) } + } - return &operatorFixture{ - chain: localChain, - address: operatorAddress, - channel: broadcastChannel, - waitForBlockHeight: waitForBlockHeight, - } + return reports +} + +// newCoordinationWallet returns the 3-operator wallet fixture +// shared by every coordinationExecutor.coordinate integration test in this +// file: same wallet public key hash and operator-to-member-index layout, so +// leader election (operator2 wins) is identical across all of them - the +// seed depends only on the wallet public key hash and the safe-block hash +// newCoordinationOperator injects at coordinationBlock-32 (both +// identical across every caller here), not on the raw coordinationBlock +// value itself, so this holds regardless of which block a given caller +// passes. +func newCoordinationWallet( + t *testing.T, + operators []*coordinationOperatorFixture, +) (wallet, [20]byte) { + t.Helper() + + // Uncompressed public key corresponding to the 20-byte public key hash: + // aa768412ceed10bd423c025542ca90071f9fb62d. + publicKeyHex, err := hex.DecodeString( + "0471e30bca60f6548d7b42582a478ea37ada63b402af7b3ddd57f0c95bb6843175" + + "aa0d2053a91a050a6797d85c38f2909cb7027f2344a01986aa2f9f8ca7a0c289", + ) + if err != nil { + t.Fatal(err) + } + + // 20-byte public key hash corresponding to the public key above. + buffer, err := hex.DecodeString("aa768412ceed10bd423c025542ca90071f9fb62d") + if err != nil { + t.Fatal(err) } + var publicKeyHash [20]byte + copy(publicKeyHash[:], buffer) - operator1 := generateOperator(1) - operator2 := generateOperator(2) - operator3 := generateOperator(3) + operator1, operator2, operator3 := operators[0], operators[1], operators[2] coordinatedWallet := wallet{ publicKey: mustUnmarshalPublicKey(t, publicKeyHex), @@ -300,6 +410,32 @@ func TestCoordinationExecutor_Coordinate(t *testing.T) { }, } + return coordinatedWallet, publicKeyHash +} + +func TestCoordinationExecutor_Coordinate(t *testing.T) { + coordinationBlock := uint64(900) + + parseScript := func(script string) bitcoin.Script { + parsed, err := hex.DecodeString(script) + if err != nil { + t.Fatal(err) + } + + return parsed + } + + channelName := t.Name() + + operator1 := newCoordinationOperator(t, 1, coordinationBlock, channelName) + operator2 := newCoordinationOperator(t, 2, coordinationBlock, channelName) + operator3 := newCoordinationOperator(t, 3, coordinationBlock, channelName) + operators := []*coordinationOperatorFixture{ + operator1, operator2, operator3, + } + + coordinatedWallet, publicKeyHash := newCoordinationWallet(t, operators) + proposalGenerator := newMockCoordinationProposalGenerator( func( walletPublicKeyHash [20]byte, @@ -330,65 +466,17 @@ func TestCoordinationExecutor_Coordinate(t *testing.T) { protocolLatch := generator.NewProtocolLatch() - generateExecutor := func(operator *operatorFixture) *coordinationExecutor { - return newCoordinationExecutor( - operator.chain, - coordinatedWallet, - coordinatedWallet.membersByOperator(operator.address), - operator.address, - proposalGenerator, - operator.channel, - membershipValidator, - protocolLatch, - operator.waitForBlockHeight, - ) - } - window := newCoordinationWindow(coordinationBlock) - type report struct { - operatorIndex int - result *coordinationResult - err error - } - - reportChan := make(chan *report, 3) - - for i, currentOperator := range []*operatorFixture{ - operator1, - operator2, - operator3, - } { - go func(operatorIndex int, operator *operatorFixture) { - result, err := generateExecutor(operator).coordinate(window) - - reportChan <- &report{ - operatorIndex: operatorIndex, - result: result, - err: err, - } - }(i+1, currentOperator) - } - - reports := make([]*report, 0) -loop: - //lint:ignore S1000 for-select is used as the channel is not closed by senders. - for { - select { - case r := <-reportChan: - reports = append(reports, r) - - if len(reports) == 3 { - break loop - } - } - } - - slices.SortFunc(reports, func(i, j *report) int { - return i.operatorIndex - j.operatorIndex - }) - - testutils.AssertIntsEqual(t, "reports count", 3, len(reports)) + reports := runCoordinationRound( + t, + operators, + coordinatedWallet, + proposalGenerator, + membershipValidator, + protocolLatch, + window, + ) expectedResult := &coordinationResult{ wallet: coordinatedWallet, @@ -404,32 +492,22 @@ loop: faults: nil, } - expectedReports := []*report{ - { - operatorIndex: 1, - result: expectedResult, - err: nil, - }, - { - operatorIndex: 2, - result: expectedResult, - err: nil, - }, - { - operatorIndex: 3, - result: expectedResult, - err: nil, - }, - } - if !reflect.DeepEqual(expectedReports, reports) { - t.Errorf( - "unexpected reports:\n"+ - "expected: %v\n"+ - "actual: %v", - expectedReports, - reports, - ) - + for _, report := range reports { + if report.err != nil { + t.Fatalf( + "operator %d: unexpected error: %v", + report.operatorIndex, + report.err, + ) + } + if !reflect.DeepEqual(expectedResult, report.result) { + t.Errorf( + "operator %d: unexpected result:\nexpected: %+v\nactual: %+v", + report.operatorIndex, + expectedResult, + report.result, + ) + } } testutils.AssertBoolsEqual( @@ -440,6 +518,148 @@ loop: ) } +// TestCoordinationExecutor_Coordinate_ReservationProposals is the M1 +// multi-signer simulated integration test for Milestone 3: it scales +// TestCoordinationExecutor_Coordinate's 3-operator, real-broadcast-channel, +// real-leader-election harness to the two reservation proposal types, +// proving the leader/follower coordination round-trip (checklist generation +// -> leader election -> broadcast -> follower validation -> convergence) +// that no mocked unit test in pkg/tbtcpg can cover, since those call +// task.Run(request) directly and never go through +// coordinationExecutor.coordinate. The protobuf wire format for both +// proposal types and the checklist activation gate each already have their +// own dedicated coverage elsewhere in this file and in marshaling_test.go; +// this test's unduplicated value is proving the two compose correctly +// through a real coordinate() round-trip. +// +// This test requires ActionReservationAnchor/ActionReservationReanchor to +// actually appear in getActionsChecklist's output; without it, every +// operator's checklist search below falls through to NoopProposal. +func TestCoordinationExecutor_Coordinate_ReservationProposals(t *testing.T) { + coordinationBlock := uint64(26500500) + + tests := map[string]struct { + matchingAction WalletActionType + generatedProposal CoordinationProposal + expectedProposal CoordinationProposal + }{ + "anchor": { + matchingAction: ActionReservationAnchor, + generatedProposal: &ReservationAnchorProposal{ + DepositFundingTxHash: bitcoin.Hash{0x01, 0x02, 0x03}, + DepositFundingOutputIndex: 1, + RequestNonce: 7, + AnchorTxFee: big.NewInt(1500), + }, + expectedProposal: &ReservationAnchorProposal{ + DepositFundingTxHash: bitcoin.Hash{0x01, 0x02, 0x03}, + DepositFundingOutputIndex: 1, + RequestNonce: 7, + AnchorTxFee: big.NewInt(1500), + }, + }, + "reanchor": { + matchingAction: ActionReservationReanchor, + generatedProposal: &ReservationReanchorProposal{ + ReservationKey: big.NewInt(424242), + RequestNonce: 4, + TargetWalletPublicKeyHash: [20]byte{0xf8, 0x7e, 0xb7}, + ReanchorTxFee: big.NewInt(1200), + }, + expectedProposal: &ReservationReanchorProposal{ + ReservationKey: big.NewInt(424242), + RequestNonce: 4, + TargetWalletPublicKeyHash: [20]byte{0xf8, 0x7e, 0xb7}, + ReanchorTxFee: big.NewInt(1200), + }, + }, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + channelName := t.Name() + + operator1 := newCoordinationOperator(t, 1, coordinationBlock, channelName) + operator2 := newCoordinationOperator(t, 2, coordinationBlock, channelName) + operator3 := newCoordinationOperator(t, 3, coordinationBlock, channelName) + operators := []*coordinationOperatorFixture{ + operator1, operator2, operator3, + } + + coordinatedWallet, publicKeyHash := newCoordinationWallet(t, operators) + + proposalGenerator := newMockCoordinationProposalGenerator( + func( + walletPublicKeyHash [20]byte, + actionsChecklist []WalletActionType, + _ uint, + ) (CoordinationProposal, error) { + for _, action := range actionsChecklist { + if walletPublicKeyHash == publicKeyHash && action == test.matchingAction { + return test.generatedProposal, nil + } + } + + return &NoopProposal{}, nil + }, + ) + + membershipValidator := group.NewMembershipValidator( + &testutils.MockLogger{}, + coordinatedWallet.signingGroupOperators, + Connect().Signing(), + ) + + protocolLatch := generator.NewProtocolLatch() + + window := newCoordinationWindow(coordinationBlock) + + reports := runCoordinationRound( + t, + operators, + coordinatedWallet, + proposalGenerator, + membershipValidator, + protocolLatch, + window, + ) + + expectedResult := &coordinationResult{ + wallet: coordinatedWallet, + window: window, + leader: operator2.address, + proposal: test.expectedProposal, + faults: nil, + } + + for _, report := range reports { + if report.err != nil { + t.Fatalf( + "operator %d: unexpected error: %v", + report.operatorIndex, + report.err, + ) + } + if !reflect.DeepEqual(expectedResult, report.result) { + t.Errorf( + "operator %d: unexpected result:\nexpected: %+v\nactual: %+v", + report.operatorIndex, + expectedResult, + report.result, + ) + } + } + + testutils.AssertBoolsEqual( + t, + "protocol latch state", + false, + protocolLatch.IsExecuting(), + ) + }) + } +} + func TestCoordinationExecutor_GetSeed(t *testing.T) { coordinationBlock := uint64(900) @@ -531,10 +751,13 @@ func TestCoordinationExecutor_GetLeader(t *testing.T) { func TestCoordinationExecutor_GetActionsChecklist(t *testing.T) { // All test cases below exercise the pre-activation code path because - // their coordination blocks are below - // DepositSweepEveryWindowActivationBlock. In this mode, all three - // actions (DepositSweep, MovedFundsSweep, MovingFunds) are gated to - // every 4th coordination window. + // their coordination blocks are below both + // DepositSweepEveryWindowActivationBlock and + // ReservationsActivationBlock. In this mode, DepositSweep, + // MovedFundsSweep, and MovingFunds are all gated to every 4th + // coordination window, and reservation actions never appear at all + // (see TestCoordinationExecutor_GetActionsChecklist_Reservations for + // the activation-block gate itself). tests := map[string]struct { coordinationBlock uint64 expectedChecklist []WalletActionType @@ -563,8 +786,8 @@ func TestCoordinationExecutor_GetActionsChecklist(t *testing.T) { coordinationBlock: 2700, expectedChecklist: []WalletActionType{ActionRedemption}, }, - // 4th-window (window 4): all actions present. Heartbeat randomly - // selected for this specific seed. + // 4th-window (window 4): sweep/moving-funds actions present. + // Heartbeat randomly selected for this specific seed. "block 3600": { coordinationBlock: 3600, expectedChecklist: []WalletActionType{ @@ -587,7 +810,8 @@ func TestCoordinationExecutor_GetActionsChecklist(t *testing.T) { coordinationBlock: 6300, expectedChecklist: []WalletActionType{ActionRedemption}, }, - // 4th-window (window 8): all actions present except heartbeat. + // 4th-window (window 8): sweep/moving-funds actions present, + // no heartbeat for this seed. "block 7200": { coordinationBlock: 7200, expectedChecklist: []WalletActionType{ @@ -609,7 +833,8 @@ func TestCoordinationExecutor_GetActionsChecklist(t *testing.T) { coordinationBlock: 9900, expectedChecklist: []WalletActionType{ActionRedemption}, }, - // 4th-window (window 12): all actions present except heartbeat. + // 4th-window (window 12): sweep/moving-funds actions present, + // no heartbeat for this seed. "block 10800": { coordinationBlock: 10800, expectedChecklist: []WalletActionType{ @@ -625,15 +850,14 @@ func TestCoordinationExecutor_GetActionsChecklist(t *testing.T) { }, "block 12600": { coordinationBlock: 12600, - expectedChecklist: []WalletActionType{ - ActionRedemption, - }, + expectedChecklist: []WalletActionType{ActionRedemption}, }, "block 13500": { coordinationBlock: 13500, expectedChecklist: []WalletActionType{ActionRedemption}, }, - // 4th-window (window 16): all actions present except heartbeat. + // 4th-window (window 16): sweep/moving-funds actions present, + // no heartbeat for this seed. "block 14400": { coordinationBlock: 14400, expectedChecklist: []WalletActionType{ @@ -697,7 +921,7 @@ func TestCoordinationExecutor_GetActionsChecklist_PostActivation(t *testing.T) { is4thWindow bool }{ // Non-4th window (window 27289): DepositSweep and - // MovedFundsSweep present, MovingFunds absent. + // MovedFundsSweep present, MovingFunds absent (frequency-gated). "post-activation non-4th window 27289": { coordinationBlock: 24560100, expectedChecklist: []WalletActionType{ @@ -726,7 +950,8 @@ func TestCoordinationExecutor_GetActionsChecklist_PostActivation(t *testing.T) { is4thWindow: false, }, // 4th window (window 27292, divisible by 4): MovingFunds - // appears. Heartbeat is NOT triggered for this seed. + // appears (frequency-gated). Heartbeat is NOT triggered for + // this seed. "post-activation 4th window 27292 no heartbeat": { coordinationBlock: 24562800, expectedChecklist: []WalletActionType{ @@ -760,7 +985,8 @@ func TestCoordinationExecutor_GetActionsChecklist_PostActivation(t *testing.T) { is4thWindow: false, }, // 4th window (window 27320, divisible by 4): MovingFunds - // appears. Heartbeat is also triggered for this seed. + // appears (frequency-gated). Heartbeat is also triggered for + // this seed. "post-activation 4th window 27320 with heartbeat": { coordinationBlock: 24588000, expectedChecklist: []WalletActionType{ @@ -773,8 +999,9 @@ func TestCoordinationExecutor_GetActionsChecklist_PostActivation(t *testing.T) { is4thWindow: true, }, // 4th window (window 27296, divisible by 4): MovingFunds - // appears. Heartbeat is NOT triggered, verifying that - // 4th-window behavior works independently of heartbeat. + // appears (frequency-gated). Heartbeat is NOT triggered, + // verifying that 4th-window behavior works independently of + // heartbeat. "post-activation 4th window 27296 no heartbeat": { coordinationBlock: 24566400, expectedChecklist: []WalletActionType{ @@ -841,31 +1068,31 @@ func TestCoordinationExecutor_GetActionsChecklist_PostActivation(t *testing.T) { // TestCoordinationExecutor_GetActionsChecklist_Reservations verifies the // reservation actions checklist gate depends solely on the activation -// block and the frequency window, never on a local per-operator +// block, never on the frequency window or a local per-operator // configuration flag - see coordinationExecutor.getActionsChecklist's // comment for why: a follower gating checklist validation on its own // local flag would wrongly fault an honest leader whenever the two // operators' local configs diverge. func TestCoordinationExecutor_GetActionsChecklist_Reservations(t *testing.T) { tests := map[string]struct { - coordinationBlock uint64 - windowIndex uint64 - expectedActions []WalletActionType + coordinationBlock uint64 + windowIndex uint64 + expectedReservationActions []WalletActionType }{ "below activation": { - coordinationBlock: ReservationsActivationBlock - 1, - windowIndex: 4, - expectedActions: []WalletActionType{ActionRedemption}, + coordinationBlock: ReservationsActivationBlock - 1, + windowIndex: 4, + expectedReservationActions: nil, }, "at activation, non-4th window": { - coordinationBlock: ReservationsActivationBlock, - windowIndex: 5, - expectedActions: []WalletActionType{ActionRedemption}, + coordinationBlock: ReservationsActivationBlock, + windowIndex: 5, + expectedReservationActions: []WalletActionType{ActionReservationAnchor, ActionReservationReanchor}, }, "at activation, 4th window": { - coordinationBlock: ReservationsActivationBlock, - windowIndex: 4, - expectedActions: []WalletActionType{ActionRedemption, ActionReservationAnchor, ActionReservationReanchor}, + coordinationBlock: ReservationsActivationBlock, + windowIndex: 4, + expectedReservationActions: []WalletActionType{ActionReservationAnchor, ActionReservationReanchor}, }, } @@ -891,14 +1118,7 @@ func TestCoordinationExecutor_GetActionsChecklist_Reservations(t *testing.T) { } } - var expectedReservationActions []WalletActionType - for _, action := range test.expectedActions { - if action == ActionReservationAnchor || action == ActionReservationReanchor { - expectedReservationActions = append(expectedReservationActions, action) - } - } - - if diff := deep.Equal(actualReservationActions, expectedReservationActions); diff != nil { + if diff := deep.Equal(actualReservationActions, test.expectedReservationActions); diff != nil { t.Errorf("reservation actions mismatch: %v", diff) } }) @@ -967,8 +1187,8 @@ func assertPostActivationSafety( // assertChecklistOrdering verifies that actions appear in canonical priority // order: Redemption < DepositSweep < MovedFundsSweep < MovingFunds < -// Heartbeat. Each consecutive pair of actions must have strictly increasing -// priority values. +// ReservationAnchor < ReservationReanchor < Heartbeat. Each consecutive pair +// of actions must have strictly increasing priority values. func assertChecklistOrdering( t *testing.T, checklist []WalletActionType, @@ -1072,6 +1292,7 @@ func TestCoordinationExecutor_ExecuteLeaderRoutine(t *testing.T) { if err != nil { t.Fatal(err) } + t.Cleanup(func() { netlocal.ReleaseBroadcastChannel("test") }) broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { return &coordinationMessage{} @@ -1281,6 +1502,7 @@ func TestCoordinationExecutor_ExecuteFollowerRoutine(t *testing.T) { if err != nil { t.Fatal(err) } + t.Cleanup(func() { netlocal.ReleaseBroadcastChannel("test") }) broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { return &coordinationMessage{} @@ -1566,6 +1788,7 @@ func TestCoordinationExecutor_ExecuteFollowerRoutine_WithIdleLeader(t *testing.T if err != nil { t.Fatal(err) } + t.Cleanup(func() { netlocal.ReleaseBroadcastChannel("test-idle") }) executor := &coordinationExecutor{ // Set only relevant fields. diff --git a/pkg/tbtc/gen/pb/message.pb.go b/pkg/tbtc/gen/pb/message.pb.go index 10ffa49071..3d5037d7a4 100644 --- a/pkg/tbtc/gen/pb/message.pb.go +++ b/pkg/tbtc/gen/pb/message.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.28.0 -// protoc v3.19.4 +// protoc v3.21.12 // source: pkg/tbtc/gen/pb/message.proto package pb diff --git a/pkg/tbtc/marshaling.go b/pkg/tbtc/marshaling.go index 3874e51ea6..ffddebf4b2 100644 --- a/pkg/tbtc/marshaling.go +++ b/pkg/tbtc/marshaling.go @@ -493,6 +493,7 @@ func validateMemberIndex(protoIndex uint32) error { return nil } +// Marshal converts the reservationAnchorProposal to a byte array. func (rap *ReservationAnchorProposal) Marshal() ([]byte, error) { return proto.Marshal( &pb.ReservationAnchorProposal{ @@ -503,6 +504,7 @@ func (rap *ReservationAnchorProposal) Marshal() ([]byte, error) { }) } +// Unmarshal converts a byte array back to the reservationAnchorProposal. func (rap *ReservationAnchorProposal) Unmarshal(data []byte) error { pbMsg := pb.ReservationAnchorProposal{} if err := proto.Unmarshal(data, &pbMsg); err != nil { @@ -536,6 +538,7 @@ func (rap *ReservationAnchorProposal) Unmarshal(data []byte) error { return nil } +// Marshal converts the reservationReanchorProposal to a byte array. func (rrp *ReservationReanchorProposal) Marshal() ([]byte, error) { return proto.Marshal( &pb.ReservationReanchorProposal{ @@ -546,6 +549,7 @@ func (rrp *ReservationReanchorProposal) Marshal() ([]byte, error) { }) } +// Unmarshal converts a byte array back to the reservationReanchorProposal. func (rrp *ReservationReanchorProposal) Unmarshal(data []byte) error { pbMsg := pb.ReservationReanchorProposal{} if err := proto.Unmarshal(data, &pbMsg); err != nil { diff --git a/pkg/tbtc/reservation.go b/pkg/tbtc/reservation.go index e761e50e28..2a36f08787 100644 --- a/pkg/tbtc/reservation.go +++ b/pkg/tbtc/reservation.go @@ -261,9 +261,6 @@ func (rrp *ReservationReanchorProposal) ValidityBlocks() uint64 { return reservationReanchorProposalValidityBlocks } -// Marshal/Unmarshal for ReservationReanchorProposal live in marshaling.go, -// alongside every other coordination proposal type's wire-format methods. - // AssembleReservationAnchorTransaction constructs an unsigned reservation // anchor transaction: a 1-input-1-output spend of the given reserved deposit // into a fresh output controlled by the given wallet. The anchor mirrors the diff --git a/pkg/tbtc/reservation_test.go b/pkg/tbtc/reservation_test.go index d0649d9329..21bdb5fe41 100644 --- a/pkg/tbtc/reservation_test.go +++ b/pkg/tbtc/reservation_test.go @@ -137,13 +137,18 @@ func TestReservationProposals_MarshalingRoundtrip(t *testing.T) { roundtrip(reanchorProposal, &ReservationReanchorProposal{}) } -func TestReservationProposals_UnmarshalRejectsInvalidFields(t *testing.T) { +func TestReservationProposals_UnmarshalRejectsInvalidPayloads(t *testing.T) { tests := map[string]struct { actionType WalletActionType payload []byte expectedError string }{ - "anchor empty object": { + // Proto3 scalar fields have no wire presence, so an entirely + // empty payload and one with every field explicitly zeroed are + // indistinguishable - a single "empty payload" case per type + // covers what the old JSON test split into "empty object" and + // "null payload" cases. + "anchor empty payload": { actionType: ActionReservationAnchor, payload: marshalPb(t, &pb.ReservationAnchorProposal{}), expectedError: "cannot unmarshal proposal payload: [anchor transaction fee is required]", @@ -160,7 +165,15 @@ func TestReservationProposals_UnmarshalRejectsInvalidFields(t *testing.T) { }), expectedError: "cannot unmarshal proposal payload: [request nonce is required]", }, - "re-anchor null payload": { + "anchor invalid deposit funding tx hash length": { + actionType: ActionReservationAnchor, + payload: marshalPb(t, &pb.ReservationAnchorProposal{ + RequestNonce: 1, + AnchorTxFee: big.NewInt(1500).Bytes(), + }), + expectedError: "cannot unmarshal proposal payload: [invalid deposit funding tx hash length: [0]]", + }, + "re-anchor empty payload": { actionType: ActionReservationReanchor, payload: nil, expectedError: "cannot unmarshal proposal payload: [reservation key is required]", @@ -231,6 +244,25 @@ func TestReservationProposals_UnmarshalRejectsInvalidFields(t *testing.T) { }), expectedError: "cannot unmarshal proposal payload: [re-anchor transaction fee is required]", }, + "re-anchor invalid target wallet hash length": { + actionType: ActionReservationReanchor, + payload: marshalPb(t, &pb.ReservationReanchorProposal{ + ReservationKey: big.NewInt(54321).Bytes(), + RequestNonce: 3, + ReanchorTxFee: big.NewInt(1700).Bytes(), + }), + expectedError: "cannot unmarshal proposal payload: [invalid target wallet public key hash length: [0]]", + }, + "re-anchor zero-value target wallet hash": { + actionType: ActionReservationReanchor, + payload: marshalPb(t, &pb.ReservationReanchorProposal{ + ReservationKey: big.NewInt(54321).Bytes(), + RequestNonce: 3, + TargetWalletPublicKeyHash: make([]byte, 20), + ReanchorTxFee: big.NewInt(1700).Bytes(), + }), + expectedError: "cannot unmarshal proposal payload: [target wallet public key hash is required]", + }, } for testName, test := range tests {