htlcswitch: auto-fail held htlcs

Make the interceptable switch aware of htlc expiry and fail back htlcs in-time
to prevent the counterparty from force-closing the channel.

htlcswitch/held_htlc_set.go

@@ -37,6 +37,20 @@ func (h *heldHtlcSet) popAll(cb func(InterceptedForward)) {
 	h.set = make(map[channeldb.CircuitKey]InterceptedForward)
 }
 
+// popAutoFails calls the callback for each forward that has an auto-fail height
+// equal or less then the specified pop height and removes them from the set.
+func (h *heldHtlcSet) popAutoFails(height uint32, cb func(InterceptedForward)) {
+	for key, fwd := range h.set {
+		if uint32(fwd.Packet().AutoFailHeight) > height {
+			continue
+		}
+
+		cb(fwd)
+
+		delete(h.set, key)
+	}
+}
+
 // pop returns the specified forward and removes it from the set.
 func (h *heldHtlcSet) pop(key channeldb.CircuitKey) (InterceptedForward, error) {
 	intercepted, ok := h.set[key]

htlcswitch/held_htlc_set_test.go

@@ -78,3 +78,49 @@ func TestHeldHtlcSet(t *testing.T) {
 	_, err = set.pop(key)
 	require.Error(t, err)
 }
+
+func TestHeldHtlcSetAutoFails(t *testing.T) {
+	set := newHeldHtlcSet()
+
+	key := channeldb.CircuitKey{
+		ChanID: lnwire.NewShortChanIDFromInt(1),
+		HtlcID: 2,
+	}
+
+	const autoFailHeight = 100
+	fwd := &interceptedForward{
+		packet:         &htlcPacket{},
+		htlc:           &lnwire.UpdateAddHTLC{},
+		autoFailHeight: autoFailHeight,
+	}
+	require.NoError(t, set.push(key, fwd))
+
+	// Test popping auto fails up to one block before the auto-fail height
+	// of our forward.
+	set.popAutoFails(
+		autoFailHeight-1,
+		func(_ InterceptedForward) {
+			require.Fail(t, "unexpected fwd")
+		},
+	)
+
+	// Popping succeeds at the auto-fail height.
+	cbCalled := false
+	set.popAutoFails(
+		autoFailHeight,
+		func(poppedFwd InterceptedForward) {
+			cbCalled = true
+
+			require.Equal(t, fwd, poppedFwd)
+		},
+	)
+	require.True(t, cbCalled)
+
+	// After this, there should be nothing more to pop.
+	set.popAutoFails(
+		autoFailHeight,
+		func(_ InterceptedForward) {
+			require.Fail(t, "unexpected fwd")
+		},
+	)
+}

htlcswitch/interceptable_switch.go

@@ -64,6 +64,13 @@ type InterceptableSwitch struct {
 	// htlc where we no longer intercept it and instead cancel it back.
 	cltvRejectDelta uint32
 
+	// cltvInterceptDelta defines the number of blocks before the expiry of
+	// the htlc where we don't intercept anymore. This value must be greater
+	// than CltvRejectDelta, because we don't want to offer htlcs to the
+	// interceptor client for which there is no time left to resolve them
+	// anymore.
+	cltvInterceptDelta uint32
+
 	// notifier is an instance of a chain notifier that we'll use to signal
 	// the switch when a new block has arrived.
 	notifier chainntnfs.ChainNotifier
@@ -137,9 +144,17 @@ type InterceptableSwitchConfig struct {
 	Notifier chainntnfs.ChainNotifier
 
 	// CltvRejectDelta defines the number of blocks before the expiry of the
-	// htlc where we no longer intercept it and instead cancel it back.
+	// htlc where we auto-fail an intercepted htlc to prevent channel
+	// force-closure.
 	CltvRejectDelta uint32
 
+	// CltvInterceptDelta defines the number of blocks before the expiry of
+	// the htlc where we don't intercept anymore. This value must be greater
+	// than CltvRejectDelta, because we don't want to offer htlcs to the
+	// interceptor client for which there is no time left to resolve them
+	// anymore.
+	CltvInterceptDelta uint32
+
 	// RequireInterceptor indicates whether processing should block if no
 	// interceptor is connected.
 	RequireInterceptor bool
@@ -149,6 +164,12 @@ type InterceptableSwitchConfig struct {
 func NewInterceptableSwitch(cfg *InterceptableSwitchConfig) (
 	*InterceptableSwitch, error) {
 
+	if cfg.CltvInterceptDelta <= cfg.CltvRejectDelta {
+		return nil, fmt.Errorf("cltv intercept delta %v not greater "+
+			"than cltv reject delta %v",
+			cfg.CltvInterceptDelta, cfg.CltvRejectDelta)
+	}
+
 	return &InterceptableSwitch{
 		htlcSwitch:              cfg.Switch,
 		intercepted:             make(chan *interceptedPackets),
@@ -158,6 +179,7 @@ func NewInterceptableSwitch(cfg *InterceptableSwitchConfig) (
 		resolutionChan:          make(chan *fwdResolution),
 		requireInterceptor:      cfg.RequireInterceptor,
 		cltvRejectDelta:         cfg.CltvRejectDelta,
+		cltvInterceptDelta:      cfg.CltvInterceptDelta,
 		notifier:                cfg.Notifier,
 
 		quit: make(chan struct{}),
@@ -275,11 +297,30 @@ func (s *InterceptableSwitch) run() error {
 
 			s.currentHeight = currentBlock.Height
 
+			// A new block is appended. Fail any held htlcs that
+			// expire at this height to prevent channel force-close.
+			s.failExpiredHtlcs()
+
 		case <-s.quit:
 			return nil
 		}
 	}
 }
+
+func (s *InterceptableSwitch) failExpiredHtlcs() {
+	s.heldHtlcSet.popAutoFails(
+		uint32(s.currentHeight),
+		func(fwd InterceptedForward) {
+			err := fwd.FailWithCode(
+				lnwire.CodeTemporaryChannelFailure,
+			)
+			if err != nil {
+				log.Errorf("Cannot fail packet: %v", err)
+			}
+		},
+	)
+}
+
 func (s *InterceptableSwitch) sendForward(fwd InterceptedForward) {
 	err := s.interceptor(fwd.Packet())
 	if err != nil {
@@ -426,6 +467,8 @@ func (s *InterceptableSwitch) interceptForward(packet *htlcPacket,
 			htlc:       htlc,
 			packet:     packet,
 			htlcSwitch: s.htlcSwitch,
+			autoFailHeight: int32(packet.incomingTimeout -
+				s.cltvRejectDelta),
 		}
 
 		// Handle forwards that are too close to expiry.
@@ -516,7 +559,7 @@ func (s *InterceptableSwitch) handleExpired(fwd *interceptedForward) (
 	bool, error) {
 
 	height := uint32(s.currentHeight)
-	if fwd.packet.incomingTimeout >= height+s.cltvRejectDelta {
+	if fwd.packet.incomingTimeout >= height+s.cltvInterceptDelta {
 		return false, nil
 	}
 
@@ -540,9 +583,10 @@ func (s *InterceptableSwitch) handleExpired(fwd *interceptedForward) (
 // It is passed from the switch to external interceptors that are interested
 // in holding forwards and resolve them manually.
 type interceptedForward struct {
-	htlc       *lnwire.UpdateAddHTLC
-	packet     *htlcPacket
-	htlcSwitch *Switch
+	htlc           *lnwire.UpdateAddHTLC
+	packet         *htlcPacket
+	htlcSwitch     *Switch
+	autoFailHeight int32
 }
 
 // Packet returns the intercepted htlc packet.
@@ -560,6 +604,7 @@ func (f *interceptedForward) Packet() InterceptedPacket {
 		IncomingExpiry: f.packet.incomingTimeout,
 		CustomRecords:  f.packet.customRecords,
 		OnionBlob:      f.htlc.OnionBlob,
+		AutoFailHeight: f.autoFailHeight,
 	}
 }
 

htlcswitch/interfaces.go

@@ -312,6 +312,10 @@ type InterceptedPacket struct {
 
 	// OnionBlob is the onion packet for the next hop
 	OnionBlob [lnwire.OnionPacketSize]byte
+
+	// AutoFailHeight is the block height at which this intercept will be
+	// failed back automatically.
+	AutoFailHeight int32
 }
 
 // InterceptedForward is passed to the ForwardInterceptor for every forwarded

htlcswitch/switch_test.go

@@ -3704,6 +3704,8 @@ func (m *mockForwardInterceptor) InterceptForwardHtlc(
 }
 
 func (m *mockForwardInterceptor) getIntercepted() InterceptedPacket {
+	m.t.Helper()
+
 	select {
 	case p := <-m.interceptedChan:
 		return p
@@ -3764,11 +3766,12 @@ func assertOutgoingLinkReceiveIntercepted(t *testing.T,
 type interceptableSwitchTestContext struct {
 	t *testing.T
 
-	preimage        [sha256.Size]byte
-	rhash           [32]byte
-	onionBlob       [1366]byte
-	incomingHtlcID  uint64
-	cltvRejectDelta uint32
+	preimage           [sha256.Size]byte
+	rhash              [32]byte
+	onionBlob          [1366]byte
+	incomingHtlcID     uint64
+	cltvRejectDelta    uint32
+	cltvInterceptDelta uint32
 
 	forwardInterceptor *mockForwardInterceptor
 	aliceChannelLink   *mockChannelLink
@@ -3820,12 +3823,13 @@ func newInterceptableSwitchTestContext(
 	preimage := [sha256.Size]byte{1}
 
 	ctx := &interceptableSwitchTestContext{
-		t:               t,
-		preimage:        preimage,
-		rhash:           sha256.Sum256(preimage[:]),
-		onionBlob:       [1366]byte{4, 5, 6},
-		incomingHtlcID:  uint64(0),
-		cltvRejectDelta: 13,
+		t:                  t,
+		preimage:           preimage,
+		rhash:              sha256.Sum256(preimage[:]),
+		onionBlob:          [1366]byte{4, 5, 6},
+		incomingHtlcID:     uint64(0),
+		cltvRejectDelta:    10,
+		cltvInterceptDelta: 13,
 		forwardInterceptor: &mockForwardInterceptor{
 			t:               t,
 			interceptedChan: make(chan InterceptedPacket),
@@ -3844,7 +3848,7 @@ func (c *interceptableSwitchTestContext) createTestPacket() *htlcPacket {
 	return &htlcPacket{
 		incomingChanID:  c.aliceChannelLink.ShortChanID(),
 		incomingHTLCID:  c.incomingHtlcID,
-		incomingTimeout: testStartingHeight + c.cltvRejectDelta + 1,
+		incomingTimeout: testStartingHeight + c.cltvInterceptDelta + 1,
 		outgoingChanID:  c.bobChannelLink.ShortChanID(),
 		obfuscator:      NewMockObfuscator(),
 		htlc: &lnwire.UpdateAddHTLC{
@@ -3887,9 +3891,10 @@ func TestSwitchHoldForward(t *testing.T) {
 
 	switchForwardInterceptor, err := NewInterceptableSwitch(
 		&InterceptableSwitchConfig{
-			Switch:          c.s,
-			CltvRejectDelta: c.cltvRejectDelta,
-			Notifier:        notifier,
+			Switch:             c.s,
+			CltvRejectDelta:    c.cltvRejectDelta,
+			CltvInterceptDelta: c.cltvInterceptDelta,
+			Notifier:           notifier,
 		},
 	)
 	require.NoError(t, err)
@@ -4091,6 +4096,7 @@ func TestSwitchHoldForward(t *testing.T) {
 		&InterceptableSwitchConfig{
 			Switch:             c.s,
 			CltvRejectDelta:    c.cltvRejectDelta,
+			CltvInterceptDelta: c.cltvInterceptDelta,
 			RequireInterceptor: true,
 			Notifier:           notifier,
 		},
@@ -4162,6 +4168,67 @@ func TestSwitchHoldForward(t *testing.T) {
 	}
 }
 
+func TestInterceptableSwitchWatchDog(t *testing.T) {
+	t.Parallel()
+
+	c := newInterceptableSwitchTestContext(t)
+	defer c.finish()
+
+	// Start interceptable switch.
+	notifier := &mock.ChainNotifier{
+		EpochChan: make(chan *chainntnfs.BlockEpoch, 1),
+	}
+	notifier.EpochChan <- &chainntnfs.BlockEpoch{Height: testStartingHeight}
+
+	switchForwardInterceptor, err := NewInterceptableSwitch(
+		&InterceptableSwitchConfig{
+			Switch:             c.s,
+			CltvRejectDelta:    c.cltvRejectDelta,
+			CltvInterceptDelta: c.cltvInterceptDelta,
+			Notifier:           notifier,
+		},
+	)
+	require.NoError(t, err)
+	require.NoError(t, switchForwardInterceptor.Start())
+
+	// Set interceptor.
+	switchForwardInterceptor.SetInterceptor(
+		c.forwardInterceptor.InterceptForwardHtlc,
+	)
+
+	// Receive a packet.
+	linkQuit := make(chan struct{})
+
+	packet := c.createTestPacket()
+
+	err = switchForwardInterceptor.ForwardPackets(linkQuit, false, packet)
+	require.NoError(t, err, "can't forward htlc packet")
+
+	// Intercept the packet.
+	intercepted := c.forwardInterceptor.getIntercepted()
+
+	require.Equal(t,
+		int32(packet.incomingTimeout-c.cltvRejectDelta),
+		intercepted.AutoFailHeight,
+	)
+
+	// Htlc expires before a resolution from the interceptor.
+	notifier.EpochChan <- &chainntnfs.BlockEpoch{
+		Height: int32(packet.incomingTimeout) -
+			int32(c.cltvRejectDelta),
+	}
+
+	// Expect the htlc to be failed back.
+	assertOutgoingLinkReceive(t, c.aliceChannelLink, true)
+
+	// It is too late now to resolve. Expect an error.
+	require.Error(t, switchForwardInterceptor.Resolve(&FwdResolution{
+		Action:   FwdActionSettle,
+		Key:      intercepted.IncomingCircuit,
+		Preimage: c.preimage,
+	}))
+}
+
 // TestSwitchDustForwarding tests that the switch properly fails HTLC's which
 // have incoming or outgoing links that breach their dust thresholds.
 func TestSwitchDustForwarding(t *testing.T) {
@@ -5395,8 +5462,10 @@ func testSwitchAliasInterceptFail(t *testing.T, zeroConf bool) {
 
 	interceptSwitch, err := NewInterceptableSwitch(
 		&InterceptableSwitchConfig{
-			Switch:   s,
-			Notifier: notifier,
+			Switch:             s,
+			Notifier:           notifier,
+			CltvRejectDelta:    10,
+			CltvInterceptDelta: 13,
 		},
 	)
 	require.NoError(t, err)