multi: add ability to autoloop in

liquidity/autoloop_test.go

@@ -50,13 +50,22 @@ func TestAutoLoopDisabled(t *testing.T) {
 	// loop in/out swaps. We expect a swap for our channel to be suggested,
 	// but do not expect any swaps to be executed, since autoloop is
 	// disabled by default.
-	c.autoloop(1, chan1Rec.Amount+1, nil, quotes, nil)
+	step := &autoloopStep{
+		minAmt:    1,
+		maxAmt:    chan1Rec.Amount + 1,
+		quotesOut: quotes,
+	}
+	c.autoloop(step)
 
 	// Trigger another autoloop, this time setting our server restrictions
 	// to have a minimum swap amount greater than the amount that we need
 	// to swap. In this case we don't even expect to get a quote, because
 	// our suggested swap is beneath the minimum swap size.
-	c.autoloop(chan1Rec.Amount+1, chan1Rec.Amount+2, nil, nil, nil)
+	step = &autoloopStep{
+		minAmt: chan1Rec.Amount + 1,
+		maxAmt: chan1Rec.Amount + 2,
+	}
+	c.autoloop(step)
 
 	c.stop()
 }
@@ -192,7 +201,13 @@ func TestAutoLoopEnabled(t *testing.T) {
 
 	// Tick our autolooper with no existing swaps, we expect a loop out
 	// swap to be dispatched for each channel.
-	c.autoloop(1, amt+1, nil, quotes, loopOuts)
+	step := &autoloopStep{
+		minAmt:      1,
+		maxAmt:      amt + 1,
+		quotesOut:   quotes,
+		expectedOut: loopOuts,
+	}
+	c.autoloop(step)
 
 	// Tick again with both of our swaps in progress. We haven't shifted our
 	// channel balances at all, so swaps should still be suggested, but we
@@ -202,7 +217,12 @@ func TestAutoLoopEnabled(t *testing.T) {
 		existingSwapFromRequest(chan2Swap, testTime, nil),
 	}
 
-	c.autoloop(1, amt+1, existing, nil, nil)
+	step = &autoloopStep{
+		minAmt:      1,
+		maxAmt:      amt + 1,
+		existingOut: existing,
+	}
+	c.autoloop(step)
 
 	// Now, we update our channel 2 swap to have failed due to off chain
 	// failure and our first swap to have succeeded.
@@ -255,7 +275,14 @@ func TestAutoLoopEnabled(t *testing.T) {
 	// We tick again, this time we expect another swap on channel 1 (which
 	// still has balances which reflect that we need to swap), but nothing
 	// for channel 2, since it has had a failure.
-	c.autoloop(1, amt+1, existing, quotes, loopOuts)
+	step = &autoloopStep{
+		minAmt:      1,
+		maxAmt:      amt + 1,
+		existingOut: existing,
+		quotesOut:   quotes,
+		expectedOut: loopOuts,
+	}
+	c.autoloop(step)
 
 	// Now, we progress our time so that we have sufficiently backed off
 	// for channel 2, and could perform another swap.
@@ -269,7 +296,13 @@ func TestAutoLoopEnabled(t *testing.T) {
 		existingSwapFromRequest(chan2Swap, testTime, failedOffChain),
 	}
 
-	c.autoloop(1, amt+1, existing, quotes, nil)
+	step = &autoloopStep{
+		minAmt:      1,
+		maxAmt:      amt + 1,
+		existingOut: existing,
+		quotesOut:   quotes,
+	}
+	c.autoloop(step)
 
 	c.stop()
 }
@@ -427,7 +460,13 @@ func TestCompositeRules(t *testing.T) {
 	// swap to be dispatched for each of our rules. We set our server side
 	// maximum to be greater than the swap amount for our peer swap (which
 	// is the larger of the two swaps).
-	c.autoloop(1, peerAmount+1, nil, quotes, loopOuts)
+	step := &autoloopStep{
+		minAmt:      1,
+		maxAmt:      peerAmount + 1,
+		quotesOut:   quotes,
+		expectedOut: loopOuts,
+	}
+	c.autoloop(step)
 
 	c.stop()
 }

liquidity/autoloop_testcontext_test.go

@@ -27,10 +27,21 @@ type autoloopTestCtx struct {
 	// quotes is a channel that we get loop out quote requests on.
 	quotes chan *loop.LoopOutQuote
 
+	// quoteRequestIn is a channel that requests for loop in quotes are
+	// pushed into.
+	quoteRequestIn chan *loop.LoopInQuoteRequest
+
+	// quotesIn is a channel that we get loop in quote responses on.
+	quotesIn chan *loop.LoopInQuote
+
 	// loopOutRestrictions is a channel that we get the server's
 	// restrictions on.
 	loopOutRestrictions chan *Restrictions
 
+	// loopInRestrictions is a channel that we get the server's
+	// loop in restrictions on.
+	loopInRestrictions chan *Restrictions
+
 	// loopOuts is a channel that we get existing loop out swaps on.
 	loopOuts chan []*loopdb.LoopOut
 
@@ -47,6 +58,13 @@ type autoloopTestCtx struct {
 	// loopOut is a channel that we return loop out responses on.
 	loopOut chan *loop.LoopOutSwapInfo
 
+	// inRequest is a channel that requests to dispatch loop in swaps are
+	// pushed into.
+	inRequest chan *loop.LoopInRequest
+
+	// loopIn is a channel that we return loop in responses on.
+	loopIn chan *loop.LoopInSwapInfo
+
 	// errChan is a channel that we send run errors into.
 	errChan chan error
 
@@ -80,14 +98,18 @@ func newAutoloopTestCtx(t *testing.T, parameters Parameters,
 
 		quoteRequest:        make(chan *loop.LoopOutQuoteRequest),
 		quotes:              make(chan *loop.LoopOutQuote),
+		quoteRequestIn:      make(chan *loop.LoopInQuoteRequest),
+		quotesIn:            make(chan *loop.LoopInQuote),
 		loopOutRestrictions: make(chan *Restrictions),
+		loopInRestrictions:  make(chan *Restrictions),
 		loopOuts:            make(chan []*loopdb.LoopOut),
 		loopIns:             make(chan []*loopdb.LoopIn),
 		restrictions:        make(chan *Restrictions),
 		outRequest:          make(chan *loop.OutRequest),
 		loopOut:             make(chan *loop.LoopOutSwapInfo),
-
-		errChan: make(chan error, 1),
+		inRequest:           make(chan *loop.LoopInRequest),
+		loopIn:              make(chan *loop.LoopInSwapInfo),
+		errChan:             make(chan error, 1),
 	}
 
 	// Set lnd's channels to equal the set of channels we want for our
@@ -96,10 +118,14 @@ func newAutoloopTestCtx(t *testing.T, parameters Parameters,
 
 	cfg := &Config{
 		AutoloopTicker: ticker.NewForce(DefaultAutoloopTicker),
-		Restrictions: func(context.Context, swap.Type) (*Restrictions,
+		Restrictions: func(_ context.Context, swapType swap.Type) (*Restrictions,
 			error) {
 
-			return <-testCtx.loopOutRestrictions, nil
+			if swapType == swap.TypeOut {
+				return <-testCtx.loopOutRestrictions, nil
+			}
+
+			return <-testCtx.loopInRestrictions, nil
 		},
 		ListLoopOut: func() ([]*loopdb.LoopOut, error) {
 			return <-testCtx.loopOuts, nil
@@ -123,6 +149,20 @@ func newAutoloopTestCtx(t *testing.T, parameters Parameters,
 
 			return <-testCtx.loopOut, nil
 		},
+		LoopInQuote: func(_ context.Context,
+			req *loop.LoopInQuoteRequest) (*loop.LoopInQuote, error) {
+
+			testCtx.quoteRequestIn <- req
+
+			return <-testCtx.quotesIn, nil
+		},
+		LoopIn: func(_ context.Context,
+			req *loop.LoopInRequest) (*loop.LoopInSwapInfo, error) {
+
+			testCtx.inRequest <- req
+
+			return <-testCtx.loopIn, nil
+		},
 		MinimumConfirmations: loop.DefaultSweepConfTarget,
 		Lnd:                  &testCtx.lnd.LndServices,
 		Clock:                testCtx.testClock,
@@ -177,31 +217,70 @@ type loopOutRequestResp struct {
 	response *loop.LoopOutSwapInfo
 }
 
+// quoteInRequestResp pairs an expected loop in quote request with the response
+// we would like to provide the manager with.
+type quoteInRequestResp struct {
+	request *loop.LoopInQuoteRequest
+	quote   *loop.LoopInQuote
+}
+
+// loopInRequestResp pairs and expected loop in request with the response we
+// would like the mocked server to respond with.
+type loopInRequestResp struct {
+	request  *loop.LoopInRequest
+	response *loop.LoopInSwapInfo
+}
+
+// autoloopStep contains all of the information to required to step
+// through an autoloop tick.
+type autoloopStep struct {
+	minAmt      btcutil.Amount
+	maxAmt      btcutil.Amount
+	existingOut []*loopdb.LoopOut
+	existingIn  []*loopdb.LoopIn
+	quotesOut   []quoteRequestResp
+	quotesIn    []quoteInRequestResp
+	expectedOut []loopOutRequestResp
+	expectedIn  []loopInRequestResp
+}
+
 // autoloop walks our test context through the process of triggering our
 // autoloop functionality, providing mocked values as required. The set of
 // quotes provided indicates that we expect swap suggestions to be made (since
 // we will query for a quote for each suggested swap). The set of expected
 // swaps indicates whether we expect any of these swap suggestions to actually
 // be dispatched by the autolooper.
-func (c *autoloopTestCtx) autoloop(minAmt, maxAmt btcutil.Amount,
-	existingOut []*loopdb.LoopOut, quotes []quoteRequestResp,
-	expectedSwaps []loopOutRequestResp) {
-
+func (c *autoloopTestCtx) autoloop(step *autoloopStep) {
 	// Tick our autoloop ticker to force assessing whether we want to loop.
 	c.manager.cfg.AutoloopTicker.Force <- testTime
 
 	// Send a mocked response from the server with the swap size limits.
-	c.loopOutRestrictions <- NewRestrictions(minAmt, maxAmt)
+	c.loopOutRestrictions <- NewRestrictions(step.minAmt, step.maxAmt)
+	c.loopInRestrictions <- NewRestrictions(step.minAmt, step.maxAmt)
 
 	// Provide the liquidity manager with our desired existing set of swaps.
-	c.loopOuts <- existingOut
-	c.loopIns <- nil
+	c.loopOuts <- step.existingOut
+	c.loopIns <- step.existingIn
 
 	// Assert that we query the server for a quote for each of our
 	// recommended swaps. Note that this differs from our set of expected
 	// swaps because we may get quotes for suggested swaps but then just
 	// log them.
-	for _, expected := range quotes {
+	for _, expected := range step.quotesIn {
+		request := <-c.quoteRequestIn
+		assert.Equal(
+			c.t, expected.request.Amount, request.Amount,
+		)
+
+		assert.Equal(
+			c.t, expected.request.HtlcConfTarget,
+			request.HtlcConfTarget,
+		)
+
+		c.quotesIn <- expected.quote
+	}
+
+	for _, expected := range step.quotesOut {
 		request := <-c.quoteRequest
 		assert.Equal(
 			c.t, expected.request.Amount, request.Amount,
@@ -214,7 +293,7 @@ func (c *autoloopTestCtx) autoloop(minAmt, maxAmt btcutil.Amount,
 	}
 
 	// Assert that we dispatch the expected set of swaps.
-	for _, expected := range expectedSwaps {
+	for _, expected := range step.expectedOut {
 		actual := <-c.outRequest
 
 		// Set our destination address to nil so that we do not need to
@@ -224,4 +303,12 @@ func (c *autoloopTestCtx) autoloop(minAmt, maxAmt btcutil.Amount,
 		assert.Equal(c.t, expected.request, actual)
 		c.loopOut <- expected.response
 	}
+
+	for _, expected := range step.expectedIn {
+		actual := <-c.inRequest
+
+		assert.Equal(c.t, expected.request, actual)
+
+		c.loopIn <- expected.response
+	}
 }

liquidity/liquidity.go

@@ -183,6 +183,10 @@ type Config struct {
 	LoopOut func(ctx context.Context, request *loop.OutRequest) (
 		*loop.LoopOutSwapInfo, error)
 
+	// LoopIn dispatches a loop in swap.
+	LoopIn func(ctx context.Context,
+		request *loop.LoopInRequest) (*loop.LoopInSwapInfo, error)
+
 	// Clock allows easy mocking of time in unit tests.
 	Clock clock.Clock
 
@@ -324,6 +328,11 @@ func (p Parameters) validate(minConfs int32, openChans []lndclient.ChannelInfo,
 			return ErrZeroChannelID
 		}
 
+		if rule.Type == swap.TypeIn {
+			return errors.New("channel level rules not supported for " +
+				"loop in swaps, only peer-level rules allowed")
+		}
+
 		if err := rule.validate(); err != nil {
 			return fmt.Errorf("channel: %v has invalid rule: %v",
 				channel.ToUint64(), err)
@@ -526,7 +535,7 @@ func (m *Manager) autoloop(ctx context.Context) error {
 		// If we don't actually have dispatch of swaps enabled, log
 		// suggestions.
 		if !m.params.Autoloop {
-			log.Debugf("recommended autoloop: %v sats over "+
+			log.Debugf("recommended autoloop out: %v sats over "+
 				"%v", swap.Amount, swap.OutgoingChanSet)
 
 			continue
@@ -544,6 +553,27 @@ func (m *Manager) autoloop(ctx context.Context) error {
 			loopOut.HtlcAddressP2WSH)
 	}
 
+	for _, in := range suggestion.InSwaps {
+		// If we don't actually have dispatch of swaps enabled, log
+		// suggestions.
+		if !m.params.Autoloop {
+			log.Debugf("recommended autoloop in: %v sats over "+
+				"%v", in.Amount, in.LastHop)
+
+			continue
+		}
+
+		in := in
+		loopIn, err := m.cfg.LoopIn(ctx, &in)
+		if err != nil {
+			return err
+		}
+
+		log.Infof("loop in automatically dispatched: hash: %v, "+
+			"address: %v", loopIn.SwapHash,
+			loopIn.HtlcAddressNP2WSH)
+	}
+
 	return nil
 }
 
@@ -564,6 +594,9 @@ type Suggestions struct {
 	// OutSwaps is the set of loop out swaps that we suggest executing.
 	OutSwaps []loop.OutRequest
 
+	// InSwaps is the set of loop in swaps that we suggest executing.
+	InSwaps []loop.LoopInRequest
+
 	// DisqualifiedChans maps the set of channels that we do not recommend
 	// swaps on to the reason that we did not recommend a swap.
 	DisqualifiedChans map[lnwire.ShortChannelID]Reason
@@ -581,13 +614,17 @@ func newSuggestions() *Suggestions {
 }
 
 func (s *Suggestions) addSwap(swap swapSuggestion) error {
-	out, ok := swap.(*loopOutSwapSuggestion)
-	if !ok {
+	switch t := swap.(type) {
+	case *loopOutSwapSuggestion:
+		s.OutSwaps = append(s.OutSwaps, t.OutRequest)
+
+	case *loopInSwapSuggestion:
+		s.InSwaps = append(s.InSwaps, t.LoopInRequest)
+
+	default:
 		return fmt.Errorf("unexpected swap type: %T", swap)
 	}
 
-	s.OutSwaps = append(s.OutSwaps, out.OutRequest)
-
 	return nil
 }
 
@@ -642,6 +679,11 @@ func (m *Manager) SuggestSwaps(ctx context.Context, autoloop bool) (
 		return nil, err
 	}
 
+	inRestrictions, err := m.getSwapRestrictions(ctx, swap.TypeIn)
+	if err != nil {
+		return nil, err
+	}
+
 	// List our current set of swaps so that we can determine which channels
 	// are already being utilized by swaps. Note that these calls may race
 	// with manual initiation of swaps.
@@ -726,7 +768,7 @@ func (m *Manager) SuggestSwaps(ctx context.Context, autoloop bool) (
 
 		suggestion, err := m.suggestSwap(
 			ctx, traffic, balances, rule, outRestrictions,
-			autoloop,
+			inRestrictions, autoloop,
 		)
 		var reasonErr *reasonError
 		if errors.As(err, &reasonErr) {
@@ -752,7 +794,7 @@ func (m *Manager) SuggestSwaps(ctx context.Context, autoloop bool) (
 
 		suggestion, err := m.suggestSwap(
 			ctx, traffic, balance, rule, outRestrictions,
-			autoloop,
+			inRestrictions, autoloop,
 		)
 
 		var reasonErr *reasonError
@@ -848,18 +890,24 @@ func (m *Manager) SuggestSwaps(ctx context.Context, autoloop bool) (
 // swap request for the rule provided.
 func (m *Manager) suggestSwap(ctx context.Context, traffic *swapTraffic,
 	balance *balances, rule *SwapRule, outRestrictions *Restrictions,
-	autoloop bool) (swapSuggestion, error) {
+	inRestrictions *Restrictions, autoloop bool) (swapSuggestion, error) {
 
 	var (
 		builder      swapBuilder
 		restrictions *Restrictions
 	)
 
+	// Get an appropriate builder and set of restrictions based on our swap
+	// type.
 	switch rule.Type {
 	case swap.TypeOut:
 		builder = newLoopOutBuilder(m.cfg)
 		restrictions = outRestrictions
 
+	case swap.TypeIn:
+		builder = newLoopInBuilder(m.cfg)
+		restrictions = inRestrictions
+
 	default:
 		return nil, fmt.Errorf("unsupported swap type: %v", rule.Type)
 	}
@@ -881,7 +929,7 @@ func (m *Manager) suggestSwap(ctx context.Context, traffic *swapTraffic,
 
 	// Next, get the amount that we need to swap for this entity, skipping
 	// over it if no change in liquidity is required.
-	amount := rule.swapAmount(balance, restrictions)
+	amount := rule.swapAmount(balance, restrictions, rule.Type)
 	if amount == 0 {
 		return nil, newReasonError(ReasonLiquidityOk)
 	}

liquidity/liquidity_test.go

@@ -1446,7 +1446,7 @@ func TestSizeRestrictions(t *testing.T) {
 
 				mockServer.On(
 					"Restrictions", mock.Anything,
-					swap.TypeOut,
+					mock.Anything,
 				).Return(&restrictions, nil)
 			}
 

liquidity/threshold_rule.go

@@ -72,7 +72,7 @@ func (r *ThresholdRule) validate() error {
 // swapAmount suggests a swap based on the liquidity thresholds configured,
 // returning zero if no swap is recommended.
 func (r *ThresholdRule) swapAmount(channel *balances,
-	restrictions *Restrictions) btcutil.Amount {
+	restrictions *Restrictions, swapType swap.Type) btcutil.Amount {
 
 	var (
 		// For loop out swaps, we want to adjust our incoming liquidity
@@ -95,6 +95,14 @@ func (r *ThresholdRule) swapAmount(channel *balances,
 		reservePercentage = uint64(r.MinimumOutgoing)
 	)
 
+	// For loop in swaps, we reverse our target and reserve values.
+	if swapType == swap.TypeIn {
+		targetBalance = channel.outgoing
+		targetPercentage = uint64(r.MinimumOutgoing)
+		reserveBalance = channel.incoming
+		reservePercentage = uint64(r.MinimumIncoming)
+	}
+
 	// Examine our total balance and required ratios to decide whether we
 	// need to swap.
 	amount := calculateSwapAmount(

liquidity/threshold_rule_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/btcsuite/btcutil"
+	"github.com/lightninglabs/loop/swap"
 	"github.com/stretchr/testify/require"
 )
 
@@ -249,6 +250,7 @@ func TestSuggestSwap(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			swap := test.rule.swapAmount(
 				test.channel, test.outRestrictions,
+				swap.TypeOut,
 			)
 			require.Equal(t, test.swap, swap)
 		})

loopd/utils.go

@@ -39,6 +39,7 @@ func getLiquidityManager(client *loop.Client) *liquidity.Manager {
 	mngrCfg := &liquidity.Config{
 		AutoloopTicker: ticker.NewForce(liquidity.DefaultAutoloopTicker),
 		LoopOut:        client.LoopOut,
+		LoopIn:         client.LoopIn,
 		Restrictions: func(ctx context.Context,
 			swapType swap.Type) (*liquidity.Restrictions, error) {