htlcswitch: implement flush and commit lifecycle hooks for channelLink

htlcswitch/link.go

@@ -375,10 +375,72 @@ type channelLink struct {
 	// UpdateAddHTLC.
 	isIncomingAddBlocked atomic.Bool
 
+	// flushHooks is a hookMap that is triggered when we reach a channel
+	// state with no live HTLCs.
+	flushHooks hookMap
+
+	// outgoingCommitHooks is a hookMap that is triggered after we send our
+	// next CommitSig.
+	outgoingCommitHooks hookMap
+
+	// incomingCommitHooks is a hookMap that is triggered after we receive
+	// our next CommitSig.
+	incomingCommitHooks hookMap
+
 	wg   sync.WaitGroup
 	quit chan struct{}
 }
 
+// hookMap is a data structure that is used to track the hooks that need to be
+// called in various parts of the channelLink's lifecycle.
+//
+// WARNING: NOT thread-safe.
+type hookMap struct {
+	// allocIdx keeps track of the next id we haven't yet allocated.
+	allocIdx atomic.Uint64
+
+	// transient is a map of hooks that are only called the next time invoke
+	// is called. These hooks are deleted during invoke.
+	transient map[uint64]func()
+
+	// newTransients is a channel that we use to accept new hooks into the
+	// hookMap.
+	newTransients chan func()
+}
+
+// newHookMap initializes a new empty hookMap.
+func newHookMap() hookMap {
+	return hookMap{
+		allocIdx:      atomic.Uint64{},
+		transient:     make(map[uint64]func()),
+		newTransients: make(chan func()),
+	}
+}
+
+// alloc allocates space in the hook map for the supplied hook, the second
+// argument determines whether it goes into the transient or persistent part
+// of the hookMap.
+func (m *hookMap) alloc(hook func()) uint64 {
+	// We assume we never overflow a uint64. Seems OK.
+	hookID := m.allocIdx.Add(1)
+	if hookID == 0 {
+		panic("hookMap allocIdx overflow")
+	}
+	m.transient[hookID] = hook
+
+	return hookID
+}
+
+// invoke is used on a hook map to call all the registered hooks and then clear
+// out the transient hooks so they are not called again.
+func (m *hookMap) invoke() {
+	for _, hook := range m.transient {
+		hook()
+	}
+
+	m.transient = make(map[uint64]func())
+}
+
 // hodlHtlc contains htlc data that is required for resolution.
 type hodlHtlc struct {
 	pd         *lnwallet.PaymentDescriptor
@@ -393,14 +455,17 @@ func NewChannelLink(cfg ChannelLinkConfig,
 	logPrefix := fmt.Sprintf("ChannelLink(%v):", channel.ChannelPoint())
 
 	return &channelLink{
-		cfg:             cfg,
+		cfg:                 cfg,
-		channel:         channel,
+		channel:             channel,
-		shortChanID:     channel.ShortChanID(),
+		shortChanID:         channel.ShortChanID(),
-		shutdownRequest: make(chan *shutdownReq),
+		shutdownRequest:     make(chan *shutdownReq),
-		hodlMap:         make(map[models.CircuitKey]hodlHtlc),
+		hodlMap:             make(map[models.CircuitKey]hodlHtlc),
-		hodlQueue:       queue.NewConcurrentQueue(10),
+		hodlQueue:           queue.NewConcurrentQueue(10),
-		log:             build.NewPrefixLog(logPrefix, log),
+		log:                 build.NewPrefixLog(logPrefix, log),
-		quit:            make(chan struct{}),
+		flushHooks:          newHookMap(),
+		outgoingCommitHooks: newHookMap(),
+		incomingCommitHooks: newHookMap(),
+		quit:                make(chan struct{}),
 	}
 }
 
@@ -612,20 +677,33 @@ func (l *channelLink) IsFlushing(linkDirection LinkDirection) bool {
 	return l.isIncomingAddBlocked.Load()
 }
 
-// OnFlushedOnce adds a hook that will be called the next time the
+// OnFlushedOnce adds a hook that will be called the next time the channel
-// channel state reaches zero htlcs. This hook will only ever be called
+// state reaches zero htlcs. This hook will only ever be called once. If the
-// once. If the channel state already has zero htlcs, then this will be
+// channel state already has zero htlcs, then this will be called immediately.
-// called immediately.
+func (l *channelLink) OnFlushedOnce(hook func()) {
-func (l *channelLink) OnFlushedOnce(func()) {
+	select {
-	// TODO(proofofkeags): Implement
+	case l.flushHooks.newTransients <- hook:
+	case <-l.quit:
+	}
 }
 
 // OnCommitOnce adds a hook that will be called the next time a CommitSig
 // message is sent in the argument's LinkDirection. This hook will only ever be
 // called once. If no CommitSig is owed in the argument's LinkDirection, then
-// we will call this hook immediately.
+// we will call this hook be run immediately.
-func (l *channelLink) OnCommitOnce(LinkDirection, func()) {
+func (l *channelLink) OnCommitOnce(direction LinkDirection, hook func()) {
-	// TODO(proofofkeags): Implement
+	var queue chan func()
+
+	if direction == Outgoing {
+		queue = l.outgoingCommitHooks.newTransients
+	} else {
+		queue = l.incomingCommitHooks.newTransients
+	}
+
+	select {
+	case queue <- hook:
+	case <-l.quit:
+	}
 }
 
 // isReestablished returns true if the link has successfully completed the
@@ -1216,6 +1294,33 @@ func (l *channelLink) htlcManager() {
 		}
 
 		select {
+		// We have a new hook that needs to be run when we reach a clean
+		// channel state.
+		case hook := <-l.flushHooks.newTransients:
+			if l.channel.IsChannelClean() {
+				hook()
+			} else {
+				l.flushHooks.alloc(hook)
+			}
+
+		// We have a new hook that needs to be run when we have
+		// committed all of our updates.
+		case hook := <-l.outgoingCommitHooks.newTransients:
+			if !l.channel.OweCommitment() {
+				hook()
+			} else {
+				l.outgoingCommitHooks.alloc(hook)
+			}
+
+		// We have a new hook that needs to be run when our peer has
+		// committed all of their updates.
+		case hook := <-l.incomingCommitHooks.newTransients:
+			if !l.channel.NeedCommitment() {
+				hook()
+			} else {
+				l.incomingCommitHooks.alloc(hook)
+			}
+
 		// Our update fee timer has fired, so we'll check the network
 		// fee to see if we should adjust our commitment fee.
 		case <-l.updateFeeTimer.C:
@@ -2101,6 +2206,13 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 			)
 			return
 		}
+
+		// As soon as we are ready to send our next revocation, we can
+		// invoke the incoming commit hooks.
+		l.RWMutex.Lock()
+		l.incomingCommitHooks.invoke()
+		l.RWMutex.Unlock()
+
 		l.cfg.Peer.SendMessage(false, nextRevocation)
 
 		// Notify the incoming htlcs of which the resolutions were
@@ -2138,19 +2250,26 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 		default:
 		}
 
-		// If both commitment chains are fully synced from our PoV,
+		// If the remote party initiated the state transition,
-		// then we don't need to reply with a signature as both sides
+		// we'll reply with a signature to provide them with their
-		// already have a commitment with the latest accepted.
+		// version of the latest commitment. Otherwise, both commitment
-		if !l.channel.OweCommitment() {
+		// chains are fully synced from our PoV, then we don't need to
-			return
+		// reply with a signature as both sides already have a
+		// commitment with the latest accepted.
+		if l.channel.OweCommitment() {
+			if !l.updateCommitTxOrFail() {
+				return
+			}
 		}
 
-		// Otherwise, the remote party initiated the state transition,
+		// Now that we have finished processing the incoming CommitSig
-		// so we'll reply with a signature to provide them with their
+		// and sent out our RevokeAndAck, we invoke the flushHooks if
-		// version of the latest commitment.
+		// the channel state is clean.
-		if !l.updateCommitTxOrFail() {
+		l.RWMutex.Lock()
-			return
+		if l.channel.IsChannelClean() {
+			l.flushHooks.invoke()
 		}
+		l.RWMutex.Unlock()
 
 	case *lnwire.RevokeAndAck:
 		// We've received a revocation from the remote chain, if valid,
@@ -2233,6 +2352,14 @@ func (l *channelLink) handleUpstreamMsg(msg lnwire.Message) {
 			}
 		}
 
+		// Now that we have finished processing the RevokeAndAck, we
+		// can invoke the flushHooks if the channel state is clean.
+		l.RWMutex.Lock()
+		if l.channel.IsChannelClean() {
+			l.flushHooks.invoke()
+		}
+		l.RWMutex.Unlock()
+
 	case *lnwire.UpdateFee:
 		// We received fee update from peer. If we are the initiator we
 		// will fail the channel, if not we will apply the update.
@@ -2441,6 +2568,12 @@ func (l *channelLink) updateCommitTx() error {
 	}
 	l.cfg.Peer.SendMessage(false, commitSig)
 
+	// Now that we have sent out a new CommitSig, we invoke the outgoing set
+	// of commit hooks.
+	l.RWMutex.Lock()
+	l.outgoingCommitHooks.invoke()
+	l.RWMutex.Unlock()
+
 	return nil
 }
 

htlcswitch/link_test.go

@@ -7113,7 +7113,7 @@ func TestLinkFlushApiGateStateIdempotence(t *testing.T) {
 }
 
 func TestLinkOutgoingCommitHooksCalled(t *testing.T) {
-	aliceLink, _, batchTicker, start, _, err :=
+	aliceLink, bobChannel, batchTicker, start, _, err :=
 		newSingleLinkTestHarness(
 			t, 5*btcutil.SatoshiPerBitcoin,
 			btcutil.SatoshiPerBitcoin,
@@ -7123,6 +7123,23 @@ func TestLinkOutgoingCommitHooksCalled(t *testing.T) {
 	require.NoError(t, start(), "could not start link")
 
 	hookCalled := make(chan struct{})
+
+	//nolint:forcetypeassert
+	aliceMsgs := aliceLink.(*channelLink).cfg.Peer.(*mockPeer).sentMsgs
+
+	ctx := linkTestContext{
+		t:          t,
+		aliceLink:  aliceLink,
+		bobChannel: bobChannel,
+		aliceMsgs:  aliceMsgs,
+	}
+
+	// Set up a pending HTLC on the link.
+	//nolint:forcetypeassert
+	htlc := generateHtlc(t, aliceLink.(*channelLink), 0)
+	ctx.sendHtlcAliceToBob(0, htlc)
+	ctx.receiveHtlcAliceToBob()
+
 	aliceLink.OnCommitOnce(Outgoing, func() {
 		close(hookCalled)
 	})

lnwallet/channel.go

@@ -5466,6 +5466,17 @@ func (lc *LightningChannel) OweCommitment() bool {
 	return lc.oweCommitment(true)
 }
 
+// NeedCommitment returns a boolean value reflecting whether we are waiting on
+// a commitment signature because there are outstanding remote updates and/or
+// updates in the remote commit tx that aren't reflected in the local commit tx
+// yet.
+func (lc *LightningChannel) NeedCommitment() bool {
+	lc.RLock()
+	defer lc.RUnlock()
+
+	return lc.oweCommitment(false)
+}
+
 // oweCommitment is the internal version of OweCommitment. This function expects
 // to be executed with a lock held.
 func (lc *LightningChannel) oweCommitment(local bool) bool {

peer/test_utils.go

@@ -4,9 +4,11 @@ import (
 	"bytes"
 	crand "crypto/rand"
 	"encoding/binary"
+	"fmt"
 	"io"
 	"math/rand"
 	"net"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -456,7 +458,9 @@ func (m *mockMessageSwitch) GetLinksByInterface(pub [33]byte) (
 // mockUpdateHandler is a mock implementation of the ChannelUpdateHandler
 // interface. It is used in mockMessageSwitch's GetLinksByInterface method.
 type mockUpdateHandler struct {
-	cid lnwire.ChannelID
+	cid                  lnwire.ChannelID
+	isOutgoingAddBlocked atomic.Bool
+	isIncomingAddBlocked atomic.Bool
 }
 
 // newMockUpdateHandler creates a new mockUpdateHandler.
@@ -509,25 +513,55 @@ type mockMessageConn struct {
 	readRaceDetectingCounter int
 }
 
-func (m *mockUpdateHandler) EnableAdds(
+func (m *mockUpdateHandler) EnableAdds(dir htlcswitch.LinkDirection) error {
-	htlcswitch.LinkDirection,
+	switch dir {
-) error {
+	case htlcswitch.Outgoing:
-	// TODO(proofofkeags): Implement
+		if !m.isOutgoingAddBlocked.Swap(false) {
+			return fmt.Errorf("%v adds already enabled", dir)
+		}
+	case htlcswitch.Incoming:
+		if !m.isIncomingAddBlocked.Swap(false) {
+			return fmt.Errorf("%v adds already enabled", dir)
+		}
+	}
+
 	return nil
 }
-func (m *mockUpdateHandler) DisableAdds(htlcswitch.LinkDirection) error {
+
-	// TODO(proofofkeags): Implement
+func (m *mockUpdateHandler) DisableAdds(dir htlcswitch.LinkDirection) error {
+	switch dir {
+	case htlcswitch.Outgoing:
+		if m.isOutgoingAddBlocked.Swap(true) {
+			return fmt.Errorf("%v adds already disabled", dir)
+		}
+	case htlcswitch.Incoming:
+		if m.isIncomingAddBlocked.Swap(true) {
+			return fmt.Errorf("%v adds already disabled", dir)
+		}
+	}
+
 	return nil
 }
-func (m *mockUpdateHandler) IsFlushing(htlcswitch.LinkDirection) bool {
+
-	// TODO(proofofkeags): Implement
+func (m *mockUpdateHandler) IsFlushing(dir htlcswitch.LinkDirection) bool {
+	switch dir {
+	case htlcswitch.Outgoing:
+		return m.isOutgoingAddBlocked.Load()
+	case htlcswitch.Incoming:
+		return m.isIncomingAddBlocked.Load()
+	}
+
 	return false
 }
-func (m *mockUpdateHandler) OnFlushedOnce(func()) {
+
-	// TODO(proofofkeags): Implement
+func (m *mockUpdateHandler) OnFlushedOnce(hook func()) {
+	hook()
 }
-func (m *mockUpdateHandler) OnCommitOnce(htlcswitch.LinkDirection, func()) {
+func (m *mockUpdateHandler) OnCommitOnce(
-	// TODO(proofofkeags): Implement
+	_ htlcswitch.LinkDirection, hook func(),
+) {
+
+	hook()
 }
 
 func newMockConn(t *testing.T, expectedMessages int) *mockMessageConn {