fix(tbtc): remove frequency gate on reservation checklist actions

pkg/net/local/broadcast_channel_manager.go

@@ -16,6 +16,7 @@ const RetransmissionTick = 50 * time.Millisecond
 
 var broadcastChannelsMutex sync.Mutex
 var broadcastChannels map[string][]*localChannel
+var broadcastChannelCancels map[string][]context.CancelFunc
 
 // getBroadcastChannel returns a BroadcastChannel designed to mediate between local
 // participants. It delivers all messages sent to the channel through its
@@ -31,12 +32,19 @@ func getBroadcastChannel(
 	if broadcastChannels == nil {
 		broadcastChannels = make(map[string][]*localChannel)
 	}
+	if broadcastChannelCancels == nil {
+		broadcastChannelCancels = make(map[string][]context.CancelFunc)
+	}
 
 	_, exists := broadcastChannels[name]
 	if !exists {
 		broadcastChannels[name] = make([]*localChannel, 0)
+		broadcastChannelCancels[name] = make([]context.CancelFunc, 0)
 	}
 
+	tickerCtx, cancelTicker := context.WithCancel(context.Background())
+	broadcastChannelCancels[name] = append(broadcastChannelCancels[name], cancelTicker)
+
 	identifier := randomLocalIdentifier()
 	channel := &localChannel{
 		name:                 name,
@@ -47,7 +55,7 @@ func getBroadcastChannel(
 		unmarshalersMutex:    sync.Mutex{},
 		unmarshalersByType:   make(map[string]func() net.TaggedUnmarshaler, 0),
 		retransmissionTicker: retransmission.NewTimeTicker(
-			context.Background(), RetransmissionTick,
+			tickerCtx, RetransmissionTick,
 		),
 	}
 	broadcastChannels[name] = append(broadcastChannels[name], channel)
@@ -66,3 +74,21 @@ func broadcastMessage(name string, message net.Message) error {
 
 	return nil
 }
+
+// ReleaseBroadcastChannel cancels every outstanding retransmission ticker
+// registered under name and removes name's entry from the registry, so a
+// later invocation reusing name starts from an empty registry regardless of
+// whether an earlier invocation's leader was still retransmitting. Callers
+// that create broadcast channels in tests should call this from t.Cleanup,
+// passing the same name they created the channel(s) under.
+func ReleaseBroadcastChannel(name string) {
+	broadcastChannelsMutex.Lock()
+	defer broadcastChannelsMutex.Unlock()
+
+	for _, cancel := range broadcastChannelCancels[name] {
+		cancel()
+	}
+
+	delete(broadcastChannels, name)
+	delete(broadcastChannelCancels, name)
+}

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")
+}

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()) {

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)
 	}