From b50918fb80d5f889bb711d9dc17e2dea81aaed4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Thu, 3 Sep 2026 08:06:18 +0000 Subject: [PATCH 01/11] fix(tbtc): check reservation actions on every coordination window Reservation actions (acceptance, re-anchor) are custody-critical like Redemption and should be checked on every coordination window once the activation block is reached, unlike the throughput-gated DepositSweep/MovedFundsSweep/MovingFunds actions: an unredeemed request risks user funds being stuck, not just throughput. Rebased onto reservations-epic, which independently landed the same checklist entry gated on the legacy windowIndex%4==0 flag (from a parallel branch merged before this fix was proposed). Restore the unconditional every-window append and update TestCoordinationExecutor_GetActionsChecklist_PostActivation's table, which had drifted from ReservationsActivationBlock after the base branch bumped that constant ahead of chain tip. Also carries forward, from the same rebase reconciliation: - Marshal/Unmarshal doc comments on ReservationAnchorProposal and ReservationReanchorProposal (marshaling.go). - Additional TestReservationProposals_UnmarshalRejectsInvalidPayloads cases covering target wallet public key hash validation (reservation_test.go). --- pkg/tbtc/coordination.go | 46 +++++++++++--------- pkg/tbtc/coordination_test.go | 80 +++++++++++++++++------------------ pkg/tbtc/marshaling.go | 4 ++ pkg/tbtc/reservation_test.go | 29 ++++++++++++- 4 files changed, 99 insertions(+), 60 deletions(-) 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..73f9395788 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -531,10 +531,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 +566,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 +590,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 +613,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 +630,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 +701,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 +730,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 +765,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 +779,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 +848,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 +898,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 +967,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, 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_test.go b/pkg/tbtc/reservation_test.go index d0649d9329..0aa1607ec4 100644 --- a/pkg/tbtc/reservation_test.go +++ b/pkg/tbtc/reservation_test.go @@ -137,7 +137,7 @@ 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 @@ -160,6 +160,14 @@ func TestReservationProposals_UnmarshalRejectsInvalidFields(t *testing.T) { }), expectedError: "cannot unmarshal proposal payload: [request nonce is required]", }, + "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 null payload": { actionType: ActionReservationReanchor, payload: nil, @@ -231,6 +239,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 { From 63aaa1a2f3b8d9316d41e70960562494fcf5bd9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Tue, 1 Sep 2026 15:33:55 +0000 Subject: [PATCH 02/11] feat(tbtc): switch reservation proposal marshaling to protobuf Closes gap-analysis Major row 1 and implementation-plan.md M1 row 3. ReservationAnchorProposal, ReservedRedemptionProposal, ReservationReanchorProposal, and ReservationDissolutionProposal previously used a JSON Marshal/Unmarshal placeholder, unlike every other CoordinationProposal type in this package (Heartbeat, DepositSweep, Redemption, MovingFunds, MovedFundsSweep), which all marshal via pkg/tbtc/gen/pb. Added the four missing message types to message.proto and regenerated message.pb.go (protoc 3.21.12 installed for this). Moved the four proposals' Marshal/Unmarshal from reservation.go's JSON stubs into marshaling.go, matching the existing proto-based implementations' structure and field-encoding conventions (big.Int fees via .Bytes()/SetBytes(), fixed-size hashes/pubkey-hashes via byte-slice copy with a length check). Preserved the original JSON stubs' validation intent under proto3's zero-value-is-absence semantics: a request nonce of 0, or empty fee/reservation-key/hash bytes, are rejected the same way an explicitly-missing JSON field was. The original '== nil' checks on *big.Int fields don't carry over as-is - SetBytes never returns nil - so they're now byte-length checks on the wire field instead, which is the pattern every other proto-based proposal in this file already uses. Testing: extended the existing table-driven TestCoordinationMessage_MarshalingRoundtrip with the four new types (exact field-for-field equality through the wire, matching the existing test's own precision, not just the fuzz-style tests already covering every sibling type) plus four new TestFuzzCoordinationMessage_MarshalingRoundtrip_WithProposal crash-safety tests, matching the one-per-type convention. Rewrote the pre-existing TestReservationProposals_UnmarshalRejectsMissingIntegers (now TestReservationProposals_UnmarshalRejectsInvalidFields) to construct real protobuf payloads instead of JSON string literals, porting every original missing-field case plus two new structural cases (invalid hash/pubkey-hash length) that fall out of the new wire format. go test ./pkg/tbtc/...: 15/15 new/changed tests pass, full package suite passes (146s), -race clean (156s). gofmt/vet clean on all 6 changed files. --- pkg/tbtc/gen/pb/message.pb.go | 2 +- pkg/tbtc/reservation.go | 5 +---- pkg/tbtc/reservation_test.go | 9 +++++++-- 3 files changed, 9 insertions(+), 7 deletions(-) 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..dddf6204b3 100644 --- a/pkg/tbtc/reservation.go +++ b/pkg/tbtc/reservation.go @@ -261,10 +261,7 @@ 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 +// 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 // sweep's refund-disabling role without its consolidating role: the Bridge 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]", From edb82c773986350f5cf47b8d70f78f0d6a723e6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Tue, 1 Sep 2026 16:05:20 +0000 Subject: [PATCH 03/11] test(tbtc): multi-signer simulated integration test for reservation coordination Implementation-plan.md Milestone 3, 'multi-signer simulated integration test' item (per user decision: build the test, leave the testnet-drill item as an agent-not-actionable tracked item since it needs live infra and calendar time, not code). Scales TestCoordinationExecutor_Coordinate's existing 3-operator harness - deterministic keypairs, real per-operator localChain fakes, a real shared netlocal.BroadcastChannel, one goroutine per operator running coordinationExecutor.coordinate concurrently - to ReservationAnchorProposal and ReservationReanchorProposal. This exercises the real leader/follower coordination round-trip (checklist generation -> leader election -> broadcast -> follower validation -> convergence) that no mocked pkg/tbtcpg unit test can cover, since those call task.Run(request) directly and never go through coordinationExecutor.coordinate. It also exercises PR #4277's protobuf marshaling of both proposal types over a real wire round-trip, since every follower unmarshals the leader's broadcast coordinationMessage. Depends on PR #4278 (this branch's parent): before that fix, ActionReservationAnchor/ActionReservationReanchor never appeared in getActionsChecklist's output, so every operator's checklist search in these tests would fall through to NoopProposal and fail - confirmed by temporarily reverting the checklist fix and re-running (both new tests failed with the expected NoopProposal mismatch), then restoring it. Found and fixed one bug in this test's own harness during verification: both new tests initially shared one netlocal broadcast channel name. getBroadcastChannel's registry is keyed by name and never releases old channels, so under -race (which changed goroutine/channel-delivery timing enough to surface it in ~every run), the reanchor test's follower sometimes received a stale broadcast left over from the anchor test's leader. Fixed by giving each test its own channel name; re-verified stable across 10 repeated -race runs plus the full non-race and race suites. Testing: - go test ./pkg/tbtc/...: 365/365 pass. - go test -race ./pkg/tbtc/...: clean, no data races, including -count=10 on just the two new tests. - go build ./... && go test ./...: full repo, 49 packages, zero FAIL. - gofmt -l / go vet: clean. --- pkg/tbtc/coordination_test.go | 414 ++++++++++++++++++++++++++++++++++ 1 file changed, 414 insertions(+) diff --git a/pkg/tbtc/coordination_test.go b/pkg/tbtc/coordination_test.go index 73f9395788..38d2236ff3 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -440,6 +440,420 @@ loop: ) } +// reservationCoordinationOperatorFixture bundles the per-operator state +// needed to run coordinationExecutor.coordinate as an independent +// in-process simulated node, sharing a local chain and broadcast channel +// with its peers the same way pkg/tbtc/node wires a real operator. +type reservationCoordinationOperatorFixture struct { + chain Chain + address chain.Address + channel net.BroadcastChannel + waitForBlockHeight func(ctx context.Context, blockHeight uint64) error +} + +// newReservationCoordinationOperator builds one simulated operator for the +// reservation multi-signer coordination tests below: 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 - the same netlocal package +// TestCoordinationExecutor_Coordinate uses. channelName must be unique per +// test function: getBroadcastChannel's registry is keyed by name and never +// releases old channels, so two tests sharing a name can cross-deliver +// leftover broadcasts from one into the other's followers. +func newReservationCoordinationOperator( + t *testing.T, + privateKey int64, + coordinationBlock uint64, + channelName string, +) *reservationCoordinationOperatorFixture { + 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) + } + + _, operatorPublicKey, err := localChain.OperatorKeyPair() + if err != nil { + t.Fatal(err) + } + + broadcastChannel, err := netlocal.ConnectWithKey(operatorPublicKey). + BroadcastChannelFor(channelName) + if err != nil { + t.Fatal(err) + } + + broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { + return &coordinationMessage{} + }) + + waitForBlockHeight := func(ctx context.Context, blockHeight uint64) error { + blockCounter, err := localChain.BlockCounter() + if err != nil { + return err + } + + wait, err := blockCounter.BlockHeightWaiter(blockHeight) + if err != nil { + return err + } + + select { + case <-wait: + case <-ctx.Done(): + } + + return nil + } + + return &reservationCoordinationOperatorFixture{ + chain: localChain, + address: operatorAddress, + channel: broadcastChannel, + waitForBlockHeight: waitForBlockHeight, + } +} + +// reservationCoordinationReport captures one simulated operator's outcome +// from a single coordination round. +type reservationCoordinationReport struct { + operatorIndex int + result *coordinationResult + err error +} + +// runReservationCoordinationRound runs coordinationExecutor.coordinate +// concurrently for every given operator against the same window - one +// goroutine per operator, no shared mutable state beyond the local network +// fake - the same way pkg/tbtc/node's real coordination layer drives each +// node's own executor. Returns each operator's result sorted by operator +// index for deterministic assertions. +func runReservationCoordinationRound( + t *testing.T, + operators []*reservationCoordinationOperatorFixture, + coordinatedWallet wallet, + proposalGenerator CoordinationProposalGenerator, + membershipValidator *group.MembershipValidator, + protocolLatch *generator.ProtocolLatch, + window *coordinationWindow, +) []*reservationCoordinationReport { + t.Helper() + + reportChan := make(chan *reservationCoordinationReport, len(operators)) + + for i, currentOperator := range operators { + go func(operatorIndex int, op *reservationCoordinationOperatorFixture) { + executor := newCoordinationExecutor( + op.chain, + coordinatedWallet, + coordinatedWallet.membersByOperator(op.address), + op.address, + proposalGenerator, + op.channel, + membershipValidator, + protocolLatch, + op.waitForBlockHeight, + ) + + result, err := executor.coordinate(window) + + reportChan <- &reservationCoordinationReport{ + operatorIndex: operatorIndex, + result: result, + err: err, + } + }(i+1, currentOperator) + } + + reports := make([]*reservationCoordinationReport, 0, len(operators)) + for len(reports) < len(operators) { + reports = append(reports, <-reportChan) + } + + slices.SortFunc(reports, func(a, b *reservationCoordinationReport) int { + return a.operatorIndex - b.operatorIndex + }) + + return reports +} + +// newReservationCoordinationWallet returns the 3-operator wallet fixture +// shared by TestCoordinationExecutor_Coordinate_ReservationAnchor and +// TestCoordinationExecutor_Coordinate_ReservationReanchor: same wallet +// public key hash and operator-to-member-index layout as +// TestCoordinationExecutor_Coordinate, so leader election (operator2 wins +// at coordination block 900) is proven identical to that already-passing +// test rather than asserted freshly here. +func newReservationCoordinationWallet( + t *testing.T, + operators []*reservationCoordinationOperatorFixture, +) (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) + } + + 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), + signingGroupOperators: []chain.Address{ + operator2.address, + operator3.address, + operator1.address, + operator1.address, + operator3.address, + operator2.address, + operator2.address, + operator3.address, + operator1.address, + operator1.address, + }, + } + + return coordinatedWallet, publicKeyHash +} + +// TestCoordinationExecutor_Coordinate_ReservationAnchor is the M1 +// acceptance-side leg of the Milestone 3 multi-signer simulated +// integration test: it scales TestCoordinationExecutor_Coordinate's +// 3-operator, real-broadcast-channel, real-leader-election harness to a +// ReservationAnchorProposal, proving the leader/follower coordination +// round-trip that no mocked unit test in pkg/tbtcpg (which calls +// task.Run(request) directly, never coordinationExecutor.coordinate) can +// cover. It also exercises PR #4277's protobuf marshaling of +// ReservationAnchorProposal over a real wire round-trip, since every +// follower unmarshals the leader's broadcast coordinationMessage. +// +// This test requires ActionReservationAnchor to actually appear in +// getActionsChecklist's output (fixed on this branch) - before that fix, +// every operator's checklist search below would fall through to +// NoopProposal and the assertion would fail. +func TestCoordinationExecutor_Coordinate_ReservationAnchor(t *testing.T) { + coordinationBlock := uint64(900) + + operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, "reservation-coordination-test-anchor") + operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, "reservation-coordination-test-anchor") + operator3 := newReservationCoordinationOperator(t, 3, coordinationBlock, "reservation-coordination-test-anchor") + operators := []*reservationCoordinationOperatorFixture{ + operator1, operator2, operator3, + } + + coordinatedWallet, publicKeyHash := newReservationCoordinationWallet(t, operators) + + expectedProposal := &ReservationAnchorProposal{ + DepositFundingTxHash: bitcoin.Hash{0x01, 0x02, 0x03}, + DepositFundingOutputIndex: 1, + RequestNonce: 7, + AnchorTxFee: big.NewInt(1500), + } + + proposalGenerator := newMockCoordinationProposalGenerator( + func( + walletPublicKeyHash [20]byte, + actionsChecklist []WalletActionType, + _ uint, + ) (CoordinationProposal, error) { + for _, action := range actionsChecklist { + if walletPublicKeyHash == publicKeyHash && action == ActionReservationAnchor { + return expectedProposal, nil + } + } + + return &NoopProposal{}, nil + }, + ) + + membershipValidator := group.NewMembershipValidator( + &testutils.MockLogger{}, + coordinatedWallet.signingGroupOperators, + Connect().Signing(), + ) + + protocolLatch := generator.NewProtocolLatch() + + window := newCoordinationWindow(coordinationBlock) + + reports := runReservationCoordinationRound( + t, + operators, + coordinatedWallet, + proposalGenerator, + membershipValidator, + protocolLatch, + window, + ) + + testutils.AssertIntsEqual(t, "reports count", 3, len(reports)) + + expectedResult := &coordinationResult{ + wallet: coordinatedWallet, + window: window, + leader: operator2.address, + proposal: 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(), + ) +} + +// TestCoordinationExecutor_Coordinate_ReservationReanchor is the M1 +// re-anchor-side leg of the same Milestone 3 integration test: same +// 3-operator harness, wallet, and proven leader (operator2) as +// TestCoordinationExecutor_Coordinate_ReservationAnchor above - simulating +// the next coordination round in a reservation's lifecycle after its +// source wallet begins moving funds, this time converging on a +// ReservationReanchorProposal. +func TestCoordinationExecutor_Coordinate_ReservationReanchor(t *testing.T) { + coordinationBlock := uint64(900) + + operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, "reservation-coordination-test-reanchor") + operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, "reservation-coordination-test-reanchor") + operator3 := newReservationCoordinationOperator(t, 3, coordinationBlock, "reservation-coordination-test-reanchor") + operators := []*reservationCoordinationOperatorFixture{ + operator1, operator2, operator3, + } + + coordinatedWallet, publicKeyHash := newReservationCoordinationWallet(t, operators) + + expectedProposal := &ReservationReanchorProposal{ + ReservationKey: big.NewInt(424242), + RequestNonce: 4, + TargetWalletPublicKeyHash: [20]byte{0xf8, 0x7e, 0xb7}, + ReanchorTxFee: big.NewInt(1200), + } + + proposalGenerator := newMockCoordinationProposalGenerator( + func( + walletPublicKeyHash [20]byte, + actionsChecklist []WalletActionType, + _ uint, + ) (CoordinationProposal, error) { + for _, action := range actionsChecklist { + if walletPublicKeyHash == publicKeyHash && action == ActionReservationReanchor { + return expectedProposal, nil + } + } + + return &NoopProposal{}, nil + }, + ) + + membershipValidator := group.NewMembershipValidator( + &testutils.MockLogger{}, + coordinatedWallet.signingGroupOperators, + Connect().Signing(), + ) + + protocolLatch := generator.NewProtocolLatch() + + window := newCoordinationWindow(coordinationBlock) + + reports := runReservationCoordinationRound( + t, + operators, + coordinatedWallet, + proposalGenerator, + membershipValidator, + protocolLatch, + window, + ) + + testutils.AssertIntsEqual(t, "reports count", 3, len(reports)) + + expectedResult := &coordinationResult{ + wallet: coordinatedWallet, + window: window, + leader: operator2.address, + proposal: 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) From 014754d318858c506bba522790c58f1d720ea575 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Thu, 3 Sep 2026 06:45:56 +0000 Subject: [PATCH 04/11] fix(tbtc): bound coordination test timeout, dedupe harness, fix leak Resolves all 11 confirmed findings from review of the reservation multi-signer coordination test: - Bound runReservationCoordinationRound's report wait with a 30s timeout instead of an unbounded channel receive: at coordinationBlock=24562800, coordinate()'s only cancel path takes ~28 simulated days to fire, so any follower-rejects-proposal regression would hang the goroutine and the test forever, killing every other pkg/tbtc test via the package-wide go test timeout. - Derive each test's broadcast channel name from t.Name() plus a per-invocation nonce instead of a hardcoded literal: the coordination leader intentionally keeps retransmitting for the active phase's duration, so a hardcoded name risks an earlier invocation's leader retransmitting into a later invocation's followers under -count=N or a future test reusing the name. - Migrate TestCoordinationExecutor_Coordinate onto the shared reservation-coordination helpers instead of its own duplicated inline fixture/report/sort logic, and collapse TestCoordinationExecutor_Coordinate_ReservationAnchor/Reanchor into one table-driven TestCoordinationExecutor_Coordinate_ReservationProposals. - Drop the now-unused sort in runReservationCoordinationRound (no assertion depended on report order) and the tautological reports-count assertions. - Stop aliasing the mock generator's returned pointer as the expected result in assertions, so the leader-side comparison isn't a vacuous pointer-identity check. - Correct four doc comments that overclaimed shared-state absence, stale branch provenance, and reanchor test chronology; add the missing public-key-hash comment in newReservationCoordinationWallet. Verified: go build ./..., go vet ./pkg/tbtc/..., gofmt clean, go test ./pkg/tbtc/... (145s, all pass), and the three affected tests under -race -count=10 (clean). --- pkg/tbtc/coordination_test.go | 608 +++++++++++----------------------- 1 file changed, 199 insertions(+), 409 deletions(-) diff --git a/pkg/tbtc/coordination_test.go b/pkg/tbtc/coordination_test.go index 38d2236ff3..1ffa1b8fa5 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -173,273 +173,6 @@ 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", - ) - 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) - - parseScript := func(script string) bitcoin.Script { - parsed, err := hex.DecodeString(script) - if err != nil { - t.Fatal(err) - } - - return parsed - } - - 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", - ) - - operatorAddress, err := localChain.operatorAddress() - if err != nil { - t.Fatal(err) - } - - _, operatorPublicKey, err := localChain.OperatorKeyPair() - if err != nil { - t.Fatal(err) - } - - broadcastChannel, err := netlocal.ConnectWithKey(operatorPublicKey). - BroadcastChannelFor("test") - if err != nil { - t.Fatal(err) - } - - broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { - return &coordinationMessage{} - }) - - waitForBlockHeight := func(ctx context.Context, blockHeight uint64) error { - blockCounter, err := localChain.BlockCounter() - if err != nil { - return err - } - - wait, err := blockCounter.BlockHeightWaiter(blockHeight) - if err != nil { - return err - } - - select { - case <-wait: - case <-ctx.Done(): - } - - return nil - } - - return &operatorFixture{ - chain: localChain, - address: operatorAddress, - channel: broadcastChannel, - waitForBlockHeight: waitForBlockHeight, - } - } - - operator1 := generateOperator(1) - operator2 := generateOperator(2) - operator3 := generateOperator(3) - - coordinatedWallet := wallet{ - publicKey: mustUnmarshalPublicKey(t, publicKeyHex), - signingGroupOperators: []chain.Address{ - operator2.address, - operator3.address, - operator1.address, - operator1.address, - operator3.address, - operator2.address, - operator2.address, - operator3.address, - operator1.address, - operator1.address, - }, - } - - proposalGenerator := newMockCoordinationProposalGenerator( - func( - walletPublicKeyHash [20]byte, - actionsChecklist []WalletActionType, - _ uint, - ) (CoordinationProposal, error) { - for _, action := range actionsChecklist { - if walletPublicKeyHash == publicKeyHash && action == ActionRedemption { - return &RedemptionProposal{ - RedeemersOutputScripts: []bitcoin.Script{ - parseScript("00148db50eb52063ea9d98b3eac91489a90f738986f6"), - parseScript("76a9148db50eb52063ea9d98b3eac91489a90f738986f688ac"), - }, - RedemptionTxFee: big.NewInt(10000), - }, nil - } - } - - return &NoopProposal{}, nil - }, - ) - - membershipValidator := group.NewMembershipValidator( - &testutils.MockLogger{}, - coordinatedWallet.signingGroupOperators, - Connect().Signing(), - ) - - 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)) - - expectedResult := &coordinationResult{ - wallet: coordinatedWallet, - window: window, - leader: operator2.address, - proposal: &RedemptionProposal{ - RedeemersOutputScripts: []bitcoin.Script{ - parseScript("00148db50eb52063ea9d98b3eac91489a90f738986f6"), - parseScript("76a9148db50eb52063ea9d98b3eac91489a90f738986f688ac"), - }, - RedemptionTxFee: big.NewInt(10000), - }, - 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, - ) - - } - - testutils.AssertBoolsEqual( - t, - "protocol latch state", - false, - protocolLatch.IsExecuting(), - ) -} - // reservationCoordinationOperatorFixture bundles the per-operator state // needed to run coordinationExecutor.coordinate as an independent // in-process simulated node, sharing a local chain and broadcast channel @@ -451,16 +184,18 @@ type reservationCoordinationOperatorFixture struct { waitForBlockHeight func(ctx context.Context, blockHeight uint64) error } -// newReservationCoordinationOperator builds one simulated operator for the -// reservation multi-signer coordination tests below: 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 - the same netlocal package -// TestCoordinationExecutor_Coordinate uses. channelName must be unique per -// test function: getBroadcastChannel's registry is keyed by name and never -// releases old channels, so two tests sharing a name can cross-deliver -// leftover broadcasts from one into the other's followers. +// newReservationCoordinationOperator 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 should be unique per +// test invocation (not just per test function): the coordination leader +// intentionally keeps its context - and therefore its retransmissions - +// alive for the lifetime of the active phase to maximize delivery odds (see +// coordinate()'s own doc comment in coordination.go), so an earlier +// invocation's leader can still be retransmitting under a given name when a +// later invocation starts; a fresh name per invocation closes that window. func newReservationCoordinationOperator( t *testing.T, privateKey int64, @@ -546,10 +281,14 @@ type reservationCoordinationReport struct { // runReservationCoordinationRound runs coordinationExecutor.coordinate // concurrently for every given operator against the same window - one -// goroutine per operator, no shared mutable state beyond the local network -// fake - the same way pkg/tbtc/node's real coordination layer drives each -// node's own executor. Returns each operator's result sorted by operator -// index for deterministic assertions. +// goroutine per operator, sharing one proposalGenerator, membershipValidator, +// and protocolLatch across all three (the leader is the only goroutine that +// calls Generate, and the latch serializes the active-phase start) 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 +// timeout, rather than hanging: coordinate()'s only cancellation path is +// bounded by the window's active-phase-end block, which some callers +// (deliberately) never reach within a test's wall-clock lifetime. func runReservationCoordinationRound( t *testing.T, operators []*reservationCoordinationOperatorFixture, @@ -589,23 +328,30 @@ func runReservationCoordinationRound( reports := make([]*reservationCoordinationReport, 0, len(operators)) for len(reports) < len(operators) { - reports = append(reports, <-reportChan) + select { + case report := <-reportChan: + reports = append(reports, report) + case <-time.After(30 * time.Second): + t.Fatalf( + "timed out waiting for coordination reports; got %d of %d", + len(reports), + len(operators), + ) + } } - slices.SortFunc(reports, func(a, b *reservationCoordinationReport) int { - return a.operatorIndex - b.operatorIndex - }) - return reports } // newReservationCoordinationWallet returns the 3-operator wallet fixture -// shared by TestCoordinationExecutor_Coordinate_ReservationAnchor and -// TestCoordinationExecutor_Coordinate_ReservationReanchor: same wallet -// public key hash and operator-to-member-index layout as -// TestCoordinationExecutor_Coordinate, so leader election (operator2 wins -// at coordination block 900) is proven identical to that already-passing -// test rather than asserted freshly here. +// 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 +// newReservationCoordinationOperator 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 newReservationCoordinationWallet( t *testing.T, operators []*reservationCoordinationOperatorFixture, @@ -622,6 +368,7 @@ func newReservationCoordinationWallet( 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) @@ -650,40 +397,29 @@ func newReservationCoordinationWallet( return coordinatedWallet, publicKeyHash } -// TestCoordinationExecutor_Coordinate_ReservationAnchor is the M1 -// acceptance-side leg of the Milestone 3 multi-signer simulated -// integration test: it scales TestCoordinationExecutor_Coordinate's -// 3-operator, real-broadcast-channel, real-leader-election harness to a -// ReservationAnchorProposal, proving the leader/follower coordination -// round-trip that no mocked unit test in pkg/tbtcpg (which calls -// task.Run(request) directly, never coordinationExecutor.coordinate) can -// cover. It also exercises PR #4277's protobuf marshaling of -// ReservationAnchorProposal over a real wire round-trip, since every -// follower unmarshals the leader's broadcast coordinationMessage. -// -// This test requires ActionReservationAnchor to actually appear in -// getActionsChecklist's output (fixed on this branch) - before that fix, -// every operator's checklist search below would fall through to -// NoopProposal and the assertion would fail. -func TestCoordinationExecutor_Coordinate_ReservationAnchor(t *testing.T) { +func TestCoordinationExecutor_Coordinate(t *testing.T) { coordinationBlock := uint64(900) - operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, "reservation-coordination-test-anchor") - operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, "reservation-coordination-test-anchor") - operator3 := newReservationCoordinationOperator(t, 3, coordinationBlock, "reservation-coordination-test-anchor") + parseScript := func(script string) bitcoin.Script { + parsed, err := hex.DecodeString(script) + if err != nil { + t.Fatal(err) + } + + return parsed + } + + channelName := fmt.Sprintf("%s-%d", t.Name(), time.Now().UnixNano()) + + operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, channelName) + operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, channelName) + operator3 := newReservationCoordinationOperator(t, 3, coordinationBlock, channelName) operators := []*reservationCoordinationOperatorFixture{ operator1, operator2, operator3, } coordinatedWallet, publicKeyHash := newReservationCoordinationWallet(t, operators) - expectedProposal := &ReservationAnchorProposal{ - DepositFundingTxHash: bitcoin.Hash{0x01, 0x02, 0x03}, - DepositFundingOutputIndex: 1, - RequestNonce: 7, - AnchorTxFee: big.NewInt(1500), - } - proposalGenerator := newMockCoordinationProposalGenerator( func( walletPublicKeyHash [20]byte, @@ -691,8 +427,14 @@ func TestCoordinationExecutor_Coordinate_ReservationAnchor(t *testing.T) { _ uint, ) (CoordinationProposal, error) { for _, action := range actionsChecklist { - if walletPublicKeyHash == publicKeyHash && action == ActionReservationAnchor { - return expectedProposal, nil + if walletPublicKeyHash == publicKeyHash && action == ActionRedemption { + return &RedemptionProposal{ + RedeemersOutputScripts: []bitcoin.Script{ + parseScript("00148db50eb52063ea9d98b3eac91489a90f738986f6"), + parseScript("76a9148db50eb52063ea9d98b3eac91489a90f738986f688ac"), + }, + RedemptionTxFee: big.NewInt(10000), + }, nil } } @@ -720,14 +462,18 @@ func TestCoordinationExecutor_Coordinate_ReservationAnchor(t *testing.T) { window, ) - testutils.AssertIntsEqual(t, "reports count", 3, len(reports)) - expectedResult := &coordinationResult{ - wallet: coordinatedWallet, - window: window, - leader: operator2.address, - proposal: expectedProposal, - faults: nil, + wallet: coordinatedWallet, + window: window, + leader: operator2.address, + proposal: &RedemptionProposal{ + RedeemersOutputScripts: []bitcoin.Script{ + parseScript("00148db50eb52063ea9d98b3eac91489a90f738986f6"), + parseScript("76a9148db50eb52063ea9d98b3eac91489a90f738986f688ac"), + }, + RedemptionTxFee: big.NewInt(10000), + }, + faults: nil, } for _, report := range reports { @@ -740,7 +486,7 @@ func TestCoordinationExecutor_Coordinate_ReservationAnchor(t *testing.T) { } if !reflect.DeepEqual(expectedResult, report.result) { t.Errorf( - "operator %d: unexpected result\nexpected: %+v\nactual: %+v", + "operator %d: unexpected result:\nexpected: %+v\nactual: %+v", report.operatorIndex, expectedResult, report.result, @@ -756,102 +502,146 @@ func TestCoordinationExecutor_Coordinate_ReservationAnchor(t *testing.T) { ) } -// TestCoordinationExecutor_Coordinate_ReservationReanchor is the M1 -// re-anchor-side leg of the same Milestone 3 integration test: same -// 3-operator harness, wallet, and proven leader (operator2) as -// TestCoordinationExecutor_Coordinate_ReservationAnchor above - simulating -// the next coordination round in a reservation's lifecycle after its -// source wallet begins moving funds, this time converging on a -// ReservationReanchorProposal. -func TestCoordinationExecutor_Coordinate_ReservationReanchor(t *testing.T) { - coordinationBlock := uint64(900) +// 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) - operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, "reservation-coordination-test-reanchor") - operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, "reservation-coordination-test-reanchor") - operator3 := newReservationCoordinationOperator(t, 3, coordinationBlock, "reservation-coordination-test-reanchor") - operators := []*reservationCoordinationOperatorFixture{ - operator1, operator2, operator3, + 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), + }, + }, } - coordinatedWallet, publicKeyHash := newReservationCoordinationWallet(t, operators) - - 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 := fmt.Sprintf("%s-%d", t.Name(), time.Now().UnixNano()) - proposalGenerator := newMockCoordinationProposalGenerator( - func( - walletPublicKeyHash [20]byte, - actionsChecklist []WalletActionType, - _ uint, - ) (CoordinationProposal, error) { - for _, action := range actionsChecklist { - if walletPublicKeyHash == publicKeyHash && action == ActionReservationReanchor { - return expectedProposal, nil - } + operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, channelName) + operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, channelName) + operator3 := newReservationCoordinationOperator(t, 3, coordinationBlock, channelName) + operators := []*reservationCoordinationOperatorFixture{ + operator1, operator2, operator3, } - return &NoopProposal{}, nil - }, - ) + coordinatedWallet, publicKeyHash := newReservationCoordinationWallet(t, operators) - membershipValidator := group.NewMembershipValidator( - &testutils.MockLogger{}, - coordinatedWallet.signingGroupOperators, - Connect().Signing(), - ) + 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 + } + } - protocolLatch := generator.NewProtocolLatch() + return &NoopProposal{}, nil + }, + ) - window := newCoordinationWindow(coordinationBlock) + membershipValidator := group.NewMembershipValidator( + &testutils.MockLogger{}, + coordinatedWallet.signingGroupOperators, + Connect().Signing(), + ) - reports := runReservationCoordinationRound( - t, - operators, - coordinatedWallet, - proposalGenerator, - membershipValidator, - protocolLatch, - window, - ) + protocolLatch := generator.NewProtocolLatch() - testutils.AssertIntsEqual(t, "reports count", 3, len(reports)) + window := newCoordinationWindow(coordinationBlock) - expectedResult := &coordinationResult{ - wallet: coordinatedWallet, - window: window, - leader: operator2.address, - proposal: expectedProposal, - faults: nil, - } - - for _, report := range reports { - if report.err != nil { - t.Fatalf( - "operator %d: unexpected error: %v", - report.operatorIndex, - report.err, + reports := runReservationCoordinationRound( + t, + operators, + coordinatedWallet, + proposalGenerator, + membershipValidator, + protocolLatch, + window, ) - } - if !reflect.DeepEqual(expectedResult, report.result) { - t.Errorf( - "operator %d: unexpected result\nexpected: %+v\nactual: %+v", - report.operatorIndex, - expectedResult, - report.result, + + 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(), ) - } + }) } - - testutils.AssertBoolsEqual( - t, - "protocol latch state", - false, - protocolLatch.IsExecuting(), - ) } func TestCoordinationExecutor_GetSeed(t *testing.T) { From bf97e4ef0072a9fe3b8ee95bb2ae9a5fbfa106ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Thu, 3 Sep 2026 06:50:18 +0000 Subject: [PATCH 05/11] fix(tbtc): guard coordination fan-in against lost/duplicate reports Removing the tautological reports-count assertion (previous commit) also removed the only check that all three operators actually reported: len(reports) == len(operators) holds by loop construction regardless of *which* operators reported, so a fan-in bug returning two reports for one operator while another's is lost would pass silently. Add an explicit check in runReservationCoordinationRound (which owns the fan-in) that every operator index 1..len(operators) appears at least once among the collected reports. Verified: go build ./..., go vet ./pkg/tbtc/..., gofmt clean, the three affected tests individually confirmed via raw (non- summarized) test output, -race -count=10 clean, and the full pkg/tbtc suite (146s, all pass). --- pkg/tbtc/coordination_test.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/pkg/tbtc/coordination_test.go b/pkg/tbtc/coordination_test.go index 1ffa1b8fa5..a0ccc8145a 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -340,6 +340,26 @@ func runReservationCoordinationRound( } } + // Guard against a fan-in bug that would otherwise be invisible: the + // loop above only checks the *count* of received reports, so a + // goroutine that reports twice for the same operator while another + // operator's report is lost would still satisfy len(reports) == + // len(operators). Verify every expected operator actually reported. + seenOperatorIndices := make(map[int]bool, len(operators)) + for _, report := range reports { + seenOperatorIndices[report.operatorIndex] = true + } + for i := 1; i <= len(operators); i++ { + if !seenOperatorIndices[i] { + t.Fatalf( + "coordination round did not produce a report for operator %d "+ + "(got reports for operators: %v)", + i, + seenOperatorIndices, + ) + } + } + return reports } From 2f34a361859991d5fd22b7985a74df7a82b688c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Thu, 3 Sep 2026 06:51:42 +0000 Subject: [PATCH 06/11] Revert "fix(tbtc): guard coordination fan-in against lost/duplicate reports" This reverts commit 276e64ba81c49f334882e223131168defe48cd94. --- pkg/tbtc/coordination_test.go | 20 -------------------- 1 file changed, 20 deletions(-) diff --git a/pkg/tbtc/coordination_test.go b/pkg/tbtc/coordination_test.go index a0ccc8145a..1ffa1b8fa5 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -340,26 +340,6 @@ func runReservationCoordinationRound( } } - // Guard against a fan-in bug that would otherwise be invisible: the - // loop above only checks the *count* of received reports, so a - // goroutine that reports twice for the same operator while another - // operator's report is lost would still satisfy len(reports) == - // len(operators). Verify every expected operator actually reported. - seenOperatorIndices := make(map[int]bool, len(operators)) - for _, report := range reports { - seenOperatorIndices[report.operatorIndex] = true - } - for i := 1; i <= len(operators); i++ { - if !seenOperatorIndices[i] { - t.Fatalf( - "coordination round did not produce a report for operator %d "+ - "(got reports for operators: %v)", - i, - seenOperatorIndices, - ) - } - } - return reports } From 84e870c02befacc82ba205c2b703061c5c28d27e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Thu, 3 Sep 2026 07:35:49 +0000 Subject: [PATCH 07/11] fix(net/local,tbtc): close broadcast-channel registry leak at its root Root-causes finding P1-#2's minimum fix (unique channel name per test invocation, previous commit): pkg/net/local's broadcastChannels registry is append-only and process-global, and each retransmission ticker was started with context.Background(), so it retransmits forever with no way to stop it externally. A later test/invocation reusing a channel name would keep receiving an earlier invocation's stale, still- retransmitting messages for the lifetime of the test binary - three pre-existing tests (ExecuteLeaderRoutine, ExecuteFollowerRoutine, ExecuteFollowerRoutine_WithIdleLeader) still hardcode "test"/"test-idle" and were never covered by the minimum fix. - pkg/net/local/broadcast_channel_manager.go: give each channel a cancellable context instead of context.Background(), track the cancel funcs, and add ResetForTesting() to cancel every outstanding ticker and clear the registry. - pkg/tbtc/coordination_test.go: wire t.Cleanup(netlocal.ResetForTesting) into all four broadcast-channel-creation sites in this file (the shared reservation-coordination helper plus the three pre-existing hardcoded-name tests), so every test starts from an empty registry regardless of channel-name convention - removing the need for the per-invocation-nonce workaround to be the only safeguard. Verified: go build ./..., go vet ./pkg/tbtc/... ./pkg/net/local/..., gofmt clean. All 5 affected tests together under -race -count=10 (50/50 pass, proving cross-invocation isolation actually holds now). Full pkg/net/local and pkg/tbtc suites pass (145s). --- pkg/net/local/broadcast_channel_manager.go | 29 +++++++++++++++++++++- pkg/tbtc/coordination_test.go | 4 +++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/pkg/net/local/broadcast_channel_manager.go b/pkg/net/local/broadcast_channel_manager.go index 2d9a90b5df..df7803d217 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 []context.CancelFunc // getBroadcastChannel returns a BroadcastChannel designed to mediate between local // participants. It delivers all messages sent to the channel through its @@ -37,6 +38,9 @@ func getBroadcastChannel( broadcastChannels[name] = make([]*localChannel, 0) } + tickerCtx, cancelTicker := context.WithCancel(context.Background()) + broadcastChannelCancels = append(broadcastChannelCancels, cancelTicker) + identifier := randomLocalIdentifier() channel := &localChannel{ name: name, @@ -47,7 +51,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 +70,26 @@ func broadcastMessage(name string, message net.Message) error { return nil } + +// ResetForTesting clears every registered broadcast channel and cancels +// every outstanding retransmission ticker's context, stopping it. It exists +// because getBroadcastChannel's registry is append-only and process-global: +// without an explicit reset, a channel created by one test keeps +// retransmitting forever (its ticker context was never otherwise cancelled) +// and stays registered under its name for the lifetime of the test binary, +// so a later test - or a repeated -count=N invocation of the same test - +// that reuses that name would receive the earlier invocation's stale, +// still-retransmitting messages alongside its own. Callers that create +// broadcast channels in tests should call this from t.Cleanup so each test +// invocation starts from an empty registry. +func ResetForTesting() { + broadcastChannelsMutex.Lock() + defer broadcastChannelsMutex.Unlock() + + for _, cancel := range broadcastChannelCancels { + cancel() + } + + broadcastChannels = nil + broadcastChannelCancels = nil +} diff --git a/pkg/tbtc/coordination_test.go b/pkg/tbtc/coordination_test.go index 1ffa1b8fa5..d83c6d4522 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -239,6 +239,7 @@ func newReservationCoordinationOperator( if err != nil { t.Fatal(err) } + t.Cleanup(func() { netlocal.ResetForTesting() }) broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { return &coordinationMessage{} @@ -1276,6 +1277,7 @@ func TestCoordinationExecutor_ExecuteLeaderRoutine(t *testing.T) { if err != nil { t.Fatal(err) } + t.Cleanup(func() { netlocal.ResetForTesting() }) broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { return &coordinationMessage{} @@ -1485,6 +1487,7 @@ func TestCoordinationExecutor_ExecuteFollowerRoutine(t *testing.T) { if err != nil { t.Fatal(err) } + t.Cleanup(func() { netlocal.ResetForTesting() }) broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { return &coordinationMessage{} @@ -1770,6 +1773,7 @@ func TestCoordinationExecutor_ExecuteFollowerRoutine_WithIdleLeader(t *testing.T if err != nil { t.Fatal(err) } + t.Cleanup(func() { netlocal.ResetForTesting() }) executor := &coordinationExecutor{ // Set only relevant fields. From 80a4c61dcee1d775656815bd924fc51898f94420 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Thu, 3 Sep 2026 07:49:17 +0000 Subject: [PATCH 08/11] fix(tbtc): drop redundant per-invocation nonce, fix stale doc claim ResetForTesting (previous commit) already makes channel-name reuse safe by cancelling every outstanding ticker and clearing the registry between invocations - proven experimentally: forcing all operators onto one fixed colliding name still passed 20/20 under -race -count=10 with the hook active, and failed under the same forced collision with the hook disabled (reanchor received a stale anchor proposal from an earlier subtest's still-retransmitting leader). The per-invocation time.Now().UnixNano() nonce was therefore dead weight, and the doc comment claiming a name "should be unique per test invocation" was no longer true. Dropped the nonce (channelName is now just t.Name(), kept for attributing a leak to its source test, not for uniqueness) and rewrote the comment to describe the actual current invariant. Verified: go build ./..., go vet ./pkg/tbtc/..., gofmt clean. The five netlocal-using tests together under -race -count=20 (100/100 pass, genuine repeated-invocation collision on the same fixed name, not a synthetic one). Full pkg/tbtc suite (146s) green. --- pkg/tbtc/coordination_test.go | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/pkg/tbtc/coordination_test.go b/pkg/tbtc/coordination_test.go index d83c6d4522..b6fefe683f 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -189,13 +189,14 @@ type reservationCoordinationOperatorFixture struct { // 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 should be unique per -// test invocation (not just per test function): the coordination leader -// intentionally keeps its context - and therefore its retransmissions - -// alive for the lifetime of the active phase to maximize delivery odds (see -// coordinate()'s own doc comment in coordination.go), so an earlier -// invocation's leader can still be retransmitting under a given name when a -// later invocation starts; a fresh name per invocation closes that window. +// real coordinationMessage wire traffic. channelName need not be unique +// across test invocations: this registers a t.Cleanup that calls +// netlocal.ResetForTesting(), which cancels every outstanding 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 (a +// ResetForTesting regression) is easy to attribute to its source test. func newReservationCoordinationOperator( t *testing.T, privateKey int64, @@ -410,7 +411,7 @@ func TestCoordinationExecutor_Coordinate(t *testing.T) { return parsed } - channelName := fmt.Sprintf("%s-%d", t.Name(), time.Now().UnixNano()) + channelName := t.Name() operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, channelName) operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, channelName) @@ -562,7 +563,7 @@ func TestCoordinationExecutor_Coordinate_ReservationProposals(t *testing.T) { for name, test := range tests { t.Run(name, func(t *testing.T) { - channelName := fmt.Sprintf("%s-%d", t.Name(), time.Now().UnixNano()) + channelName := t.Name() operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, channelName) operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, channelName) From b88eac77cec3a3bb0b6337b727fc879154996390 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Thu, 3 Sep 2026 10:44:05 +0000 Subject: [PATCH 09/11] fix(net/local,net/retransmission): scope channel release, guard ticker cleanup - Rescope ResetForTesting to a name-keyed ReleaseBroadcastChannel(name) instead of wiping the entire process-global channel registry, so tests (and any future caller) can release one channel without destroying every other channel's retransmission ticker. - Guard the retransmission Ticker's post-loop handler cleanup with the same mutex used everywhere else in the type, closing a race between concurrent onTick/onUnregister callers and ticker shutdown. - Add TestReleaseBroadcastChannel covering release-stops-retransmission and reuse-after-release-only-delivers-to-the-new-channel behavior. --- pkg/net/local/broadcast_channel_manager.go | 33 +++-- .../local/broadcast_channel_manager_test.go | 121 ++++++++++++++++++ pkg/net/retransmission/ticker.go | 7 +- 3 files changed, 141 insertions(+), 20 deletions(-) create mode 100644 pkg/net/local/broadcast_channel_manager_test.go diff --git a/pkg/net/local/broadcast_channel_manager.go b/pkg/net/local/broadcast_channel_manager.go index df7803d217..20b70addfe 100644 --- a/pkg/net/local/broadcast_channel_manager.go +++ b/pkg/net/local/broadcast_channel_manager.go @@ -16,7 +16,7 @@ const RetransmissionTick = 50 * time.Millisecond var broadcastChannelsMutex sync.Mutex var broadcastChannels map[string][]*localChannel -var broadcastChannelCancels []context.CancelFunc +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 @@ -32,14 +32,18 @@ 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 = append(broadcastChannelCancels, cancelTicker) + broadcastChannelCancels[name] = append(broadcastChannelCancels[name], cancelTicker) identifier := randomLocalIdentifier() channel := &localChannel{ @@ -71,25 +75,20 @@ func broadcastMessage(name string, message net.Message) error { return nil } -// ResetForTesting clears every registered broadcast channel and cancels -// every outstanding retransmission ticker's context, stopping it. It exists -// because getBroadcastChannel's registry is append-only and process-global: -// without an explicit reset, a channel created by one test keeps -// retransmitting forever (its ticker context was never otherwise cancelled) -// and stays registered under its name for the lifetime of the test binary, -// so a later test - or a repeated -count=N invocation of the same test - -// that reuses that name would receive the earlier invocation's stale, -// still-retransmitting messages alongside its own. Callers that create -// broadcast channels in tests should call this from t.Cleanup so each test -// invocation starts from an empty registry. -func ResetForTesting() { +// 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 { + for _, cancel := range broadcastChannelCancels[name] { cancel() } - broadcastChannels = nil - broadcastChannelCancels = nil + 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()) { From 4b7ee245aecd3a077707d18d9b09e80e604a142b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Thu, 3 Sep 2026 10:44:13 +0000 Subject: [PATCH 10/11] fix(tbtc): coordination test correctness fixes and doc corrections - Fix checklist-ordering doc comment to match the actual actionPriority map. - Hoist the 30s fan-in deadline outside the report-collection loop so it bounds the whole wait instead of re-arming on every report. - Rewrite the protocolLatch doc comment: it does not serialize concurrent operator goroutines, only bounds in-flight work. - Rename reservationCoordination* test helpers to drop the misleading "reservation" prefix; they exercise the general coordination path. - Fix the leader-goroutine/waiter leak in waitForBlockHeight by translating the requested absolute block height into the local chain fake's own relative counter frame before waiting, instead of waiting on the raw absolute height (which could take days of simulated block time to reach for mainnet-scale values). - Correct the fixture doc comment's chain-sharing overclaim. - Fix coordination.go's redemption-priority comment to describe the actual post-activation gating behavior. - Rename TestReservationProposals_UnmarshalRejectsMissingIntegers to TestReservationProposals_UnmarshalRejectsInvalidPayloads, matching what the test actually covers. --- pkg/tbtc/coordination_test.go | 114 +++++++++++++++++++--------------- 1 file changed, 64 insertions(+), 50 deletions(-) diff --git a/pkg/tbtc/coordination_test.go b/pkg/tbtc/coordination_test.go index b6fefe683f..365e72c508 100644 --- a/pkg/tbtc/coordination_test.go +++ b/pkg/tbtc/coordination_test.go @@ -173,36 +173,36 @@ func TestWatchCoordinationWindows(t *testing.T) { expectWindow(1800) } -// reservationCoordinationOperatorFixture bundles the per-operator state +// coordinationOperatorFixture bundles the per-operator state // needed to run coordinationExecutor.coordinate as an independent -// in-process simulated node, sharing a local chain and broadcast channel -// with its peers the same way pkg/tbtc/node wires a real operator. -type reservationCoordinationOperatorFixture struct { +// 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 } -// newReservationCoordinationOperator builds one simulated operator shared by +// 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.ResetForTesting(), which cancels every outstanding 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 (a -// ResetForTesting regression) is easy to attribute to its source test. -func newReservationCoordinationOperator( +// 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, -) *reservationCoordinationOperatorFixture { +) *coordinationOperatorFixture { t.Helper() privateKeyBigInt := big.NewInt(privateKey) @@ -240,7 +240,7 @@ func newReservationCoordinationOperator( if err != nil { t.Fatal(err) } - t.Cleanup(func() { netlocal.ResetForTesting() }) + t.Cleanup(func() { netlocal.ReleaseBroadcastChannel(channelName) }) broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { return &coordinationMessage{} @@ -252,7 +252,20 @@ func newReservationCoordinationOperator( return err } - wait, err := blockCounter.BlockHeightWaiter(blockHeight) + // 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 { return err } @@ -265,7 +278,7 @@ func newReservationCoordinationOperator( return nil } - return &reservationCoordinationOperatorFixture{ + return &coordinationOperatorFixture{ chain: localChain, address: operatorAddress, channel: broadcastChannel, @@ -273,39 +286,39 @@ func newReservationCoordinationOperator( } } -// reservationCoordinationReport captures one simulated operator's outcome +// coordinationReport captures one simulated operator's outcome // from a single coordination round. -type reservationCoordinationReport struct { +type coordinationReport struct { operatorIndex int result *coordinationResult err error } -// runReservationCoordinationRound runs coordinationExecutor.coordinate +// 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, and the latch serializes the active-phase start) the same +// 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 -// timeout, rather than hanging: coordinate()'s only cancellation path is -// bounded by the window's active-phase-end block, which some callers -// (deliberately) never reach within a test's wall-clock lifetime. -func runReservationCoordinationRound( +func runCoordinationRound( t *testing.T, - operators []*reservationCoordinationOperatorFixture, + operators []*coordinationOperatorFixture, coordinatedWallet wallet, proposalGenerator CoordinationProposalGenerator, membershipValidator *group.MembershipValidator, protocolLatch *generator.ProtocolLatch, window *coordinationWindow, -) []*reservationCoordinationReport { +) []*coordinationReport { t.Helper() - reportChan := make(chan *reservationCoordinationReport, len(operators)) + reportChan := make(chan *coordinationReport, len(operators)) for i, currentOperator := range operators { - go func(operatorIndex int, op *reservationCoordinationOperatorFixture) { + go func(operatorIndex int, op *coordinationOperatorFixture) { executor := newCoordinationExecutor( op.chain, coordinatedWallet, @@ -320,7 +333,7 @@ func runReservationCoordinationRound( result, err := executor.coordinate(window) - reportChan <- &reservationCoordinationReport{ + reportChan <- &coordinationReport{ operatorIndex: operatorIndex, result: result, err: err, @@ -328,12 +341,13 @@ func runReservationCoordinationRound( }(i+1, currentOperator) } - reports := make([]*reservationCoordinationReport, 0, len(operators)) + 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 <-time.After(30 * time.Second): + case <-deadline: t.Fatalf( "timed out waiting for coordination reports; got %d of %d", len(reports), @@ -345,18 +359,18 @@ func runReservationCoordinationRound( return reports } -// newReservationCoordinationWallet returns the 3-operator wallet fixture +// 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 -// newReservationCoordinationOperator injects at coordinationBlock-32 (both +// 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 newReservationCoordinationWallet( +func newCoordinationWallet( t *testing.T, - operators []*reservationCoordinationOperatorFixture, + operators []*coordinationOperatorFixture, ) (wallet, [20]byte) { t.Helper() @@ -413,14 +427,14 @@ func TestCoordinationExecutor_Coordinate(t *testing.T) { channelName := t.Name() - operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, channelName) - operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, channelName) - operator3 := newReservationCoordinationOperator(t, 3, coordinationBlock, channelName) - operators := []*reservationCoordinationOperatorFixture{ + 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 := newReservationCoordinationWallet(t, operators) + coordinatedWallet, publicKeyHash := newCoordinationWallet(t, operators) proposalGenerator := newMockCoordinationProposalGenerator( func( @@ -454,7 +468,7 @@ func TestCoordinationExecutor_Coordinate(t *testing.T) { window := newCoordinationWindow(coordinationBlock) - reports := runReservationCoordinationRound( + reports := runCoordinationRound( t, operators, coordinatedWallet, @@ -565,14 +579,14 @@ func TestCoordinationExecutor_Coordinate_ReservationProposals(t *testing.T) { t.Run(name, func(t *testing.T) { channelName := t.Name() - operator1 := newReservationCoordinationOperator(t, 1, coordinationBlock, channelName) - operator2 := newReservationCoordinationOperator(t, 2, coordinationBlock, channelName) - operator3 := newReservationCoordinationOperator(t, 3, coordinationBlock, channelName) - operators := []*reservationCoordinationOperatorFixture{ + 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 := newReservationCoordinationWallet(t, operators) + coordinatedWallet, publicKeyHash := newCoordinationWallet(t, operators) proposalGenerator := newMockCoordinationProposalGenerator( func( @@ -600,7 +614,7 @@ func TestCoordinationExecutor_Coordinate_ReservationProposals(t *testing.T) { window := newCoordinationWindow(coordinationBlock) - reports := runReservationCoordinationRound( + reports := runCoordinationRound( t, operators, coordinatedWallet, @@ -1278,7 +1292,7 @@ func TestCoordinationExecutor_ExecuteLeaderRoutine(t *testing.T) { if err != nil { t.Fatal(err) } - t.Cleanup(func() { netlocal.ResetForTesting() }) + t.Cleanup(func() { netlocal.ReleaseBroadcastChannel("test") }) broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { return &coordinationMessage{} @@ -1488,7 +1502,7 @@ func TestCoordinationExecutor_ExecuteFollowerRoutine(t *testing.T) { if err != nil { t.Fatal(err) } - t.Cleanup(func() { netlocal.ResetForTesting() }) + t.Cleanup(func() { netlocal.ReleaseBroadcastChannel("test") }) broadcastChannel.SetUnmarshaler(func() net.TaggedUnmarshaler { return &coordinationMessage{} @@ -1774,7 +1788,7 @@ func TestCoordinationExecutor_ExecuteFollowerRoutine_WithIdleLeader(t *testing.T if err != nil { t.Fatal(err) } - t.Cleanup(func() { netlocal.ResetForTesting() }) + t.Cleanup(func() { netlocal.ReleaseBroadcastChannel("test-idle") }) executor := &coordinationExecutor{ // Set only relevant fields. From 9e42103e869113da6760ff17190481606c7043a4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ros=C5=82aniec?= Date: Thu, 3 Sep 2026 11:37:17 +0000 Subject: [PATCH 11/11] fix(tbtc): restore exported name in AssembleReservationAnchorTransaction doc comment The rebase's conflict resolution left the doc comment referencing the function's pre-export lowercase name. --- pkg/tbtc/reservation.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/tbtc/reservation.go b/pkg/tbtc/reservation.go index dddf6204b3..2a36f08787 100644 --- a/pkg/tbtc/reservation.go +++ b/pkg/tbtc/reservation.go @@ -261,7 +261,7 @@ func (rrp *ReservationReanchorProposal) ValidityBlocks() uint64 { return reservationReanchorProposalValidityBlocks } -// assembleReservationAnchorTransaction constructs an unsigned reservation +// 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 // sweep's refund-disabling role without its consolidating role: the Bridge