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_test.go b/pkg/tbtc/coordination_test.go index 73f9395788..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, + ) + + result, err := executor.coordinate(window) - select { - case <-wait: - case <-ctx.Done(): + 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) } - operator1 := generateOperator(1) - operator2 := generateOperator(2) - operator3 := generateOperator(3) + // 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, 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) @@ -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/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 0aa1607ec4..21bdb5fe41 100644 --- a/pkg/tbtc/reservation_test.go +++ b/pkg/tbtc/reservation_test.go @@ -143,7 +143,12 @@ func TestReservationProposals_UnmarshalRejectsInvalidPayloads(t *testing.T) { 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]", @@ -168,7 +173,7 @@ func TestReservationProposals_UnmarshalRejectsInvalidPayloads(t *testing.T) { }), expectedError: "cannot unmarshal proposal payload: [invalid deposit funding tx hash length: [0]]", }, - "re-anchor null payload": { + "re-anchor empty payload": { actionType: ActionReservationReanchor, payload: nil, expectedError: "cannot unmarshal proposal payload: [reservation key is required]",