circuitbreaker/process_test.go

package main

import (
	"context"
	"testing"
	"time"

	"github.com/lightningnetwork/lnd/lnrpc/routerrpc"
	"github.com/stretchr/testify/require"
)

func TestProcess(t *testing.T) {
	defer Timeout()()

	t.Run("settle", func(t *testing.T) {
		testProcess(t, resolveEventSettle)
	})
	t.Run("forward fail", func(t *testing.T) {
		testProcess(t, resolveEventForwardFail)
	})
	t.Run("link fail", func(t *testing.T) {
		testProcess(t, resolveEventLinkFail)
	})
}

type resolveEvent int

const (
	resolveEventSettle resolveEvent = iota
	resolveEventForwardFail
	resolveEventLinkFail
)

func testProcess(t *testing.T, event resolveEvent) {
	cfg := &config{
		groupConfig: groupConfig{
			MaxPendingHtlcs: 2,
		},
	}

	client := newLndclientMock()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	p := newProcess(client, cfg)

	resolved := make(chan struct{})
	p.resolvedCallback = func() {
		close(resolved)
	}

	exit := make(chan error)
	go func() {
		exit <- p.run(ctx)
	}()

	key := &routerrpc.CircuitKey{
		ChanId: 2,
		HtlcId: 5,
	}
	client.htlcInterceptorRequests <- &routerrpc.ForwardHtlcInterceptRequest{
		IncomingCircuitKey: key,
	}

	resp := <-client.htlcInterceptorResponses
	require.Equal(t, routerrpc.ResolveHoldForwardAction_RESUME, resp.Action)

	htlcEvent := &routerrpc.HtlcEvent{
		EventType:         routerrpc.HtlcEvent_FORWARD,
		IncomingChannelId: key.ChanId,
		IncomingHtlcId:    key.HtlcId,
	}

	switch event {
	case resolveEventForwardFail:
		htlcEvent.Event = &routerrpc.HtlcEvent_ForwardFailEvent{}

	case resolveEventLinkFail:
		htlcEvent.Event = &routerrpc.HtlcEvent_LinkFailEvent{}

	case resolveEventSettle:
		htlcEvent.Event = &routerrpc.HtlcEvent_SettleEvent{}
	}

	client.htlcEvents <- htlcEvent

	<-resolved

	cancel()
	require.ErrorIs(t, <-exit, context.Canceled)
}

func TestLimits(t *testing.T) {
	for _, mode := range []Mode{ModeFail, ModeQueue, ModeQueuePeerInitiated} {
		t.Run(mode.String(), func(t *testing.T) {
			t.Run("rate limit", func(t *testing.T) { testRateLimit(t, mode) })
			t.Run("max pending", func(t *testing.T) { testMaxPending(t, mode) })
		})
	}
}

func testRateLimit(t *testing.T, mode Mode) {
	defer Timeout()()

	cfg := &config{
		groupConfig: groupConfig{
			HtlcMinInterval: 2 * time.Second,
			HtlcBurstSize:   2,
			Mode:            mode,
		},
	}

	client := newLndclientMock()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	p := newProcess(client, cfg)

	exit := make(chan error)
	go func() {
		exit <- p.run(ctx)
	}()

	var chanId uint64 = 2
	if mode == ModeQueuePeerInitiated {
		// We are the initiator of the channel. Not queueing is expected in this
		// mode.
		chanId = 3
	}

	key := &routerrpc.CircuitKey{
		ChanId: chanId,
		HtlcId: 5,
	}
	interceptReq := &routerrpc.ForwardHtlcInterceptRequest{
		IncomingCircuitKey: key,
	}

	// First htlc accepted.
	client.htlcInterceptorRequests <- interceptReq
	resp := <-client.htlcInterceptorResponses
	require.Equal(t, routerrpc.ResolveHoldForwardAction_RESUME, resp.Action)

	// Second htlc right after is also accepted because of burst size 2.
	interceptReq.IncomingCircuitKey.HtlcId++
	client.htlcInterceptorRequests <- interceptReq
	resp = <-client.htlcInterceptorResponses
	require.Equal(t, routerrpc.ResolveHoldForwardAction_RESUME, resp.Action)

	// Third htlc again right after should hit the rate limit.
	interceptReq.IncomingCircuitKey.HtlcId++
	client.htlcInterceptorRequests <- interceptReq

	interceptStart := time.Now()

	resp = <-client.htlcInterceptorResponses

	if mode == ModeQueue {
		require.Equal(t, routerrpc.ResolveHoldForwardAction_RESUME, resp.Action)
		require.GreaterOrEqual(t, time.Since(interceptStart), time.Second)
	} else {
		require.Equal(t, routerrpc.ResolveHoldForwardAction_FAIL, resp.Action)

		htlcEvent := &routerrpc.HtlcEvent{
			EventType:         routerrpc.HtlcEvent_FORWARD,
			IncomingChannelId: key.ChanId,
			IncomingHtlcId:    key.HtlcId,
			Event:             &routerrpc.HtlcEvent_ForwardFailEvent{},
		}

		client.htlcEvents <- htlcEvent

		// Allow some time for the peer controller to process the failed forward
		// event.
		time.Sleep(time.Second)
	}

	cancel()
	require.ErrorIs(t, <-exit, context.Canceled)
}

func testMaxPending(t *testing.T, mode Mode) {
	defer Timeout()()

	cfg := &config{
		groupConfig: groupConfig{
			HtlcMinInterval: time.Minute,
			HtlcBurstSize:   2,
			MaxPendingHtlcs: 1,
			Mode:            mode,
		},
	}

	client := newLndclientMock()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	p := newProcess(client, cfg)

	exit := make(chan error)
	go func() {
		exit <- p.run(ctx)
	}()

	var chanId uint64 = 2
	if mode == ModeQueuePeerInitiated {
		// We are the initiator of the channel. Not queueing is expected in this
		// mode.
		chanId = 3
	}

	key := &routerrpc.CircuitKey{
		ChanId: chanId,
		HtlcId: 5,
	}
	interceptReq := &routerrpc.ForwardHtlcInterceptRequest{
		IncomingCircuitKey: key,
	}

	// First htlc accepted.
	client.htlcInterceptorRequests <- interceptReq
	resp := <-client.htlcInterceptorResponses
	require.Equal(t, routerrpc.ResolveHoldForwardAction_RESUME, resp.Action)

	// Second htlc should be hitting the max pending htlcs limit.
	interceptReq.IncomingCircuitKey.HtlcId++
	client.htlcInterceptorRequests <- interceptReq

	if mode == ModeQueue {
		select {
		case <-client.htlcInterceptorResponses:
			require.Fail(t, "unexpected response")

		case <-time.After(time.Second):
		}
	} else {
		resp = <-client.htlcInterceptorResponses
		require.Equal(t, routerrpc.ResolveHoldForwardAction_FAIL, resp.Action)
	}

	cancel()
	require.ErrorIs(t, <-exit, context.Canceled)
}