Merge pull request #4683 from carlaKC/4471-acceptorupfrontshutdown

channelacceptor: add new fields to acceptor response

chanacceptor/acceptor_test.go

@@ -1,156 +1,316 @@
 package chanacceptor
 
 import (
-	"bytes"
-	"sync/atomic"
+	"errors"
+	"math/big"
 	"testing"
 	"time"
 
-	"github.com/lightningnetwork/lnd/lnrpc"
-
 	"github.com/btcsuite/btcd/btcec"
+	"github.com/btcsuite/btcd/chaincfg"
+	"github.com/btcsuite/btcutil"
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"github.com/lightningnetwork/lnd/lnwallet/chancloser"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
-func randKey(t *testing.T) *btcec.PublicKey {
-	t.Helper()
+const testTimeout = time.Second
 
-	priv, err := btcec.NewPrivateKey(btcec.S256())
-	if err != nil {
-		t.Fatalf("unable to generate new public key")
-	}
+type channelAcceptorCtx struct {
+	t *testing.T
 
-	return priv.PubKey()
+	// extRequests is the channel that we send our channel accept requests
+	// into, this channel mocks sending of a request to the rpc acceptor.
+	// This channel should be buffered with the number of requests we want
+	// to send so that it does not block (like a rpc stream).
+	extRequests chan []byte
+
+	// responses is a map of pending channel IDs to the response which we
+	// wish to mock the remote channel acceptor sending.
+	responses map[[32]byte]*lnrpc.ChannelAcceptResponse
+
+	// acceptor is the channel acceptor we create for the test.
+	acceptor *RPCAcceptor
+
+	// errChan is a channel that the error the channel acceptor exits with
+	// is sent into.
+	errChan chan error
+
+	// quit is a channel that can be used to shutdown the channel acceptor
+	// and return errShuttingDown.
+	quit chan struct{}
 }
 
-// requestInfo encapsulates the information sent from the RPCAcceptor to the
-// receiver on the other end of the stream.
-type requestInfo struct {
-	chanReq      *ChannelAcceptRequest
-	responseChan chan lnrpc.ChannelAcceptResponse
-}
+func newChanAcceptorCtx(t *testing.T, acceptCallCount int,
+	responses map[[32]byte]*lnrpc.ChannelAcceptResponse) *channelAcceptorCtx {
 
-var defaultAcceptTimeout = 5 * time.Second
-
-func acceptAndIncrementCtr(rpc ChannelAcceptor, req *ChannelAcceptRequest,
-	ctr *uint32, success chan struct{}) {
-
-	result := rpc.Accept(req)
-	if !result {
-		return
+	testCtx := &channelAcceptorCtx{
+		t:           t,
+		extRequests: make(chan []byte, acceptCallCount),
+		responses:   responses,
+		errChan:     make(chan error),
+		quit:        make(chan struct{}),
 	}
 
-	val := atomic.AddUint32(ctr, 1)
-	if val == 3 {
-		success <- struct{}{}
-	}
-}
-
-// TestMultipleRPCClients tests that the RPCAcceptor is able to handle multiple
-// callers to its Accept method and respond to them correctly.
-func TestRPCMultipleAcceptClients(t *testing.T) {
-
-	var (
-		node = randKey(t)
-
-		firstOpenReq = &ChannelAcceptRequest{
-			Node: node,
-			OpenChanMsg: &lnwire.OpenChannel{
-				PendingChannelID: [32]byte{0},
-			},
-		}
-
-		secondOpenReq = &ChannelAcceptRequest{
-			Node: node,
-			OpenChanMsg: &lnwire.OpenChannel{
-				PendingChannelID: [32]byte{1},
-			},
-		}
-
-		thirdOpenReq = &ChannelAcceptRequest{
-			Node: node,
-			OpenChanMsg: &lnwire.OpenChannel{
-				PendingChannelID: [32]byte{2},
-			},
-		}
-
-		counter uint32
+	testCtx.acceptor = NewRPCAcceptor(
+		testCtx.receiveResponse, testCtx.sendRequest, testTimeout*5,
+		&chaincfg.TestNet3Params, testCtx.quit,
 	)
 
-	quit := make(chan struct{})
-	defer close(quit)
+	return testCtx
+}
 
-	// Create channels to handle requests and successes.
-	requests := make(chan *requestInfo)
-	successChan := make(chan struct{})
-	errChan := make(chan struct{}, 4)
+// sendRequest mocks sending a request to the channel acceptor.
+func (c *channelAcceptorCtx) sendRequest(request *lnrpc.ChannelAcceptRequest) error {
+	select {
+	case c.extRequests <- request.PendingChanId:
 
-	// demultiplexReq is a closure used to abstract the RPCAcceptor's request
-	// and response logic.
-	demultiplexReq := func(req *ChannelAcceptRequest) bool {
-		respChan := make(chan lnrpc.ChannelAcceptResponse, 1)
-
-		newRequest := &requestInfo{
-			chanReq:      req,
-			responseChan: respChan,
-		}
-
-		// Send the newRequest to the requests channel.
-		select {
-		case requests <- newRequest:
-		case <-quit:
-			return false
-		}
-
-		// Receive the response and verify that the PendingChanId matches
-		// the ID found in the ChannelAcceptRequest. If no response has been
-		// received in defaultAcceptTimeout, then return false.
-		select {
-		case resp := <-respChan:
-			pendingID := req.OpenChanMsg.PendingChannelID
-			if !bytes.Equal(pendingID[:], resp.PendingChanId) {
-				errChan <- struct{}{}
-				return false
-			}
-
-			return resp.Accept
-		case <-time.After(defaultAcceptTimeout):
-			errChan <- struct{}{}
-			return false
-		case <-quit:
-			return false
-		}
+	case <-time.After(testTimeout):
+		c.t.Fatalf("timeout sending request: %v", request.PendingChanId)
 	}
 
-	rpcAcceptor := NewRPCAcceptor(demultiplexReq)
+	return nil
+}
 
-	// Now we call the Accept method for each request.
+// receiveResponse mocks sending of a response from the channel acceptor.
+func (c *channelAcceptorCtx) receiveResponse() (*lnrpc.ChannelAcceptResponse,
+	error) {
+
+	select {
+	case id := <-c.extRequests:
+		scratch := [32]byte{}
+		copy(scratch[:], id)
+
+		resp, ok := c.responses[scratch]
+		assert.True(c.t, ok)
+
+		return resp, nil
+
+	case <-time.After(testTimeout):
+		c.t.Fatalf("timeout receiving request")
+		return nil, errors.New("receiveResponse timeout")
+
+	// Exit if our test acceptor closes the done channel, which indicates
+	// that the acceptor is shutting down.
+	case <-c.acceptor.done:
+		return nil, errors.New("acceptor shutting down")
+	}
+}
+
+// start runs our channel acceptor in a goroutine which sends its exit error
+// into our test error channel.
+func (c *channelAcceptorCtx) start() {
 	go func() {
-		acceptAndIncrementCtr(rpcAcceptor, firstOpenReq, &counter, successChan)
+		c.errChan <- c.acceptor.Run()
 	}()
+}
 
-	go func() {
-		acceptAndIncrementCtr(rpcAcceptor, secondOpenReq, &counter, successChan)
-	}()
+// stop shuts down the test's channel acceptor and asserts that it exits with
+// our expected error.
+func (c *channelAcceptorCtx) stop() {
+	close(c.quit)
 
-	go func() {
-		acceptAndIncrementCtr(rpcAcceptor, thirdOpenReq, &counter, successChan)
-	}()
+	select {
+	case actual := <-c.errChan:
+		assert.Equal(c.t, errShuttingDown, actual)
 
-	for {
+	case <-time.After(testTimeout):
+		c.t.Fatal("timeout waiting for acceptor to exit")
+	}
+}
+
+// queryAndAssert takes a map of open channel requests which we want to call
+// Accept for to the outcome we expect from the acceptor, dispatches each
+// request in a goroutine and then asserts that we get the outcome we expect.
+func (c *channelAcceptorCtx) queryAndAssert(queries map[*lnwire.OpenChannel]*ChannelAcceptResponse) {
+	var (
+		node = &btcec.PublicKey{
+			X: big.NewInt(1),
+			Y: big.NewInt(1),
+		}
+
+		responses = make(chan struct{})
+	)
+
+	for request, expected := range queries {
+		request := request
+		expected := expected
+
+		go func() {
+			resp := c.acceptor.Accept(&ChannelAcceptRequest{
+				Node:        node,
+				OpenChanMsg: request,
+			})
+			assert.Equal(c.t, expected, resp)
+			responses <- struct{}{}
+		}()
+	}
+
+	// Wait for each of our requests to return a response before we exit.
+	for i := 0; i < len(queries); i++ {
 		select {
-		case newRequest := <-requests:
-			newResponse := lnrpc.ChannelAcceptResponse{
-				Accept:        true,
-				PendingChanId: newRequest.chanReq.OpenChanMsg.PendingChannelID[:],
-			}
-
-			newRequest.responseChan <- newResponse
-		case <-errChan:
-			t.Fatalf("unable to accept ChannelAcceptRequest")
-		case <-successChan:
-			return
-		case <-quit:
+		case <-responses:
+		case <-time.After(testTimeout):
+			c.t.Fatalf("did not receive response")
 		}
 	}
 }
+
+// TestMultipleAcceptClients tests that the RPC acceptor is capable of handling
+// multiple requests to its Accept function and responding to them correctly.
+func TestMultipleAcceptClients(t *testing.T) {
+	testAddr := "bcrt1qwrmq9uca0t3dy9t9wtuq5tm4405r7tfzyqn9pp"
+	testUpfront, err := chancloser.ParseUpfrontShutdownAddress(
+		testAddr, &chaincfg.TestNet3Params,
+	)
+	require.NoError(t, err)
+
+	var (
+		chan1 = &lnwire.OpenChannel{
+			PendingChannelID: [32]byte{1},
+		}
+		chan2 = &lnwire.OpenChannel{
+			PendingChannelID: [32]byte{2},
+		}
+		chan3 = &lnwire.OpenChannel{
+			PendingChannelID: [32]byte{3},
+		}
+
+		customError = errors.New("go away")
+
+		// Queries is a map of the channel IDs we will query Accept
+		// with, and the set of outcomes we expect.
+		queries = map[*lnwire.OpenChannel]*ChannelAcceptResponse{
+			chan1: NewChannelAcceptResponse(
+				true, nil, testUpfront, 1, 2, 3, 4, 5, 6,
+			),
+			chan2: NewChannelAcceptResponse(
+				false, errChannelRejected, nil, 0, 0, 0,
+				0, 0, 0,
+			),
+			chan3: NewChannelAcceptResponse(
+				false, customError, nil, 0, 0, 0, 0, 0, 0,
+			),
+		}
+
+		// Responses is a mocked set of responses from the remote
+		// channel acceptor.
+		responses = map[[32]byte]*lnrpc.ChannelAcceptResponse{
+			chan1.PendingChannelID: {
+				PendingChanId:   chan1.PendingChannelID[:],
+				Accept:          true,
+				UpfrontShutdown: testAddr,
+				CsvDelay:        1,
+				MaxHtlcCount:    2,
+				MinAcceptDepth:  3,
+				ReserveSat:      4,
+				InFlightMaxMsat: 5,
+				MinHtlcIn:       6,
+			},
+			chan2.PendingChannelID: {
+				PendingChanId: chan2.PendingChannelID[:],
+				Accept:        false,
+			},
+			chan3.PendingChannelID: {
+				PendingChanId: chan3.PendingChannelID[:],
+				Accept:        false,
+				Error:         customError.Error(),
+			},
+		}
+	)
+
+	// Create and start our channel acceptor.
+	testCtx := newChanAcceptorCtx(t, len(queries), responses)
+	testCtx.start()
+
+	// Dispatch three queries and assert that we get our expected response.
+	// for each.
+	testCtx.queryAndAssert(queries)
+
+	// Shutdown our acceptor.
+	testCtx.stop()
+}
+
+// TestInvalidResponse tests the case where our remote channel acceptor sends us
+// an invalid response, so the channel acceptor stream terminates.
+func TestInvalidResponse(t *testing.T) {
+	var (
+		chan1 = [32]byte{1}
+
+		// We make a single query, and expect it to fail with our
+		// generic error because our response is invalid.
+		queries = map[*lnwire.OpenChannel]*ChannelAcceptResponse{
+			{
+				PendingChannelID: chan1,
+			}: NewChannelAcceptResponse(
+				false, errChannelRejected, nil, 0, 0,
+				0, 0, 0, 0,
+			),
+		}
+
+		// Create a single response which is invalid because it accepts
+		// the channel but also contains an error message.
+		responses = map[[32]byte]*lnrpc.ChannelAcceptResponse{
+			chan1: {
+				PendingChanId: chan1[:],
+				Accept:        true,
+				Error:         "has an error as well",
+			},
+		}
+	)
+
+	// Create and start our channel acceptor.
+	testCtx := newChanAcceptorCtx(t, len(queries), responses)
+	testCtx.start()
+
+	testCtx.queryAndAssert(queries)
+
+	// We do not expect our channel acceptor to exit because of one invalid
+	// response, so we shutdown and assert here.
+	testCtx.stop()
+}
+
+// TestInvalidReserve tests validation of the channel reserve proposed by the
+// acceptor against the dust limit that was proposed by the remote peer.
+func TestInvalidReserve(t *testing.T) {
+	var (
+		chan1 = [32]byte{1}
+
+		dustLimit = btcutil.Amount(1000)
+		reserve   = dustLimit / 2
+
+		// We make a single query, and expect it to fail with our
+		// generic error because channel reserve is too low.
+		queries = map[*lnwire.OpenChannel]*ChannelAcceptResponse{
+			{
+				PendingChannelID: chan1,
+				DustLimit:        dustLimit,
+			}: NewChannelAcceptResponse(
+				false, errChannelRejected, nil, 0, 0,
+				0, reserve, 0, 0,
+			),
+		}
+
+		// Create a single response which is invalid because the
+		// proposed reserve is below our dust limit.
+		responses = map[[32]byte]*lnrpc.ChannelAcceptResponse{
+			chan1: {
+				PendingChanId: chan1[:],
+				Accept:        true,
+				ReserveSat:    uint64(reserve),
+			},
+		}
+	)
+
+	// Create and start our channel acceptor.
+	testCtx := newChanAcceptorCtx(t, len(queries), responses)
+	testCtx.start()
+
+	testCtx.queryAndAssert(queries)
+
+	// We do not expect our channel acceptor to exit because of one invalid
+	// response, so we shutdown and assert here.
+	testCtx.stop()
+}

chanacceptor/chainedacceptor.go

@@ -46,18 +46,48 @@ func (c *ChainedAcceptor) RemoveAcceptor(id uint64) {
 // and returns the conjunction of all these predicates.
 //
 // NOTE: Part of the ChannelAcceptor interface.
-func (c *ChainedAcceptor) Accept(req *ChannelAcceptRequest) bool {
-	result := true
-
+func (c *ChainedAcceptor) Accept(req *ChannelAcceptRequest) *ChannelAcceptResponse {
 	c.acceptorsMtx.RLock()
-	for _, acceptor := range c.acceptors {
-		// We call Accept first in case any acceptor (perhaps an RPCAcceptor)
-		// wishes to be notified about ChannelAcceptRequest.
-		result = acceptor.Accept(req) && result
-	}
-	c.acceptorsMtx.RUnlock()
+	defer c.acceptorsMtx.RUnlock()
 
-	return result
+	var finalResp ChannelAcceptResponse
+
+	for _, acceptor := range c.acceptors {
+		// Call our acceptor to determine whether we want to accept this
+		// channel.
+		acceptorResponse := acceptor.Accept(req)
+
+		// If we should reject the channel, we can just exit early. This
+		// has the effect of returning the error belonging to our first
+		// failed acceptor.
+		if acceptorResponse.RejectChannel() {
+			return acceptorResponse
+		}
+
+		// If we have accepted the channel, we need to set the other
+		// fields that were set in the response. However, since we are
+		// dealing with multiple responses, we need to make sure that we
+		// have not received inconsistent values (eg a csv delay of 1
+		// from one acceptor, and a delay of 120 from another). We
+		// set each value on our final response if it has not been set
+		// yet, and allow duplicate sets if the value is the same. If
+		// we cannot set a field, we return an error response.
+		var err error
+		finalResp, err = mergeResponse(finalResp, *acceptorResponse)
+		if err != nil {
+			log.Errorf("response for: %x has inconsistent values: %v",
+				req.OpenChanMsg.PendingChannelID, err)
+
+			return NewChannelAcceptResponse(
+				false, errChannelRejected, nil, 0, 0,
+				0, 0, 0, 0,
+			)
+		}
+	}
+
+	// If we have gone through all of our acceptors with no objections, we
+	// can return an acceptor with a nil error.
+	return &finalResp
 }
 
 // A compile-time constraint to ensure ChainedAcceptor implements the

chanacceptor/errors.go

@@ -0,0 +1,8 @@
+package chanacceptor
+
+// ChanAcceptError is an error that it returned when an external channel
+// acceptor rejects a channel. Note that this error type is whitelisted and will
+// be delivered to the peer initiating a channel.
+type ChanAcceptError struct {
+	error
+}

chanacceptor/interface.go

@@ -1,10 +1,20 @@
 package chanacceptor
 
 import (
+	"errors"
+
 	"github.com/btcsuite/btcd/btcec"
+	"github.com/btcsuite/btcutil"
 	"github.com/lightningnetwork/lnd/lnwire"
 )
 
+var (
+	// errChannelRejected is returned when the rpc channel acceptor rejects
+	// a channel due to acceptor timeout, shutdown, or because no custom
+	// error value is available when the channel was rejected.
+	errChannelRejected = errors.New("channel rejected")
+)
+
 // ChannelAcceptRequest is a struct containing the requesting node's public key
 // along with the lnwire.OpenChannel message that they sent when requesting an
 // inbound channel. This information is provided to each acceptor so that they
@@ -18,8 +28,88 @@ type ChannelAcceptRequest struct {
 	OpenChanMsg *lnwire.OpenChannel
 }
 
-// ChannelAcceptor is an interface that represents a predicate on the data
+// ChannelAcceptResponse is a struct containing the response to a request to
+// open an inbound channel. Note that fields added to this struct must be added
+// to the mergeResponse function to allow combining of responses from different
+// acceptors.
+type ChannelAcceptResponse struct {
+	// ChanAcceptError the error returned by the channel acceptor. If the
+	// channel was accepted, this value will be nil.
+	ChanAcceptError
+
+	// UpfrontShutdown is the address that we will set as our upfront
+	// shutdown address.
+	UpfrontShutdown lnwire.DeliveryAddress
+
+	// CSVDelay is the csv delay we require for the remote peer.
+	CSVDelay uint16
+
+	// Reserve is the amount that require the remote peer hold in reserve
+	// on the channel.
+	Reserve btcutil.Amount
+
+	// InFlightTotal is the maximum amount that we allow the remote peer to
+	// hold in outstanding htlcs.
+	InFlightTotal lnwire.MilliSatoshi
+
+	// HtlcLimit is the maximum number of htlcs that we allow the remote
+	// peer to offer us.
+	HtlcLimit uint16
+
+	// MinHtlcIn is the minimum incoming htlc value allowed on the channel.
+	MinHtlcIn lnwire.MilliSatoshi
+
+	// MinAcceptDepth is the minimum depth that the initiator of the
+	// channel should wait before considering the channel open.
+	MinAcceptDepth uint16
+}
+
+// NewChannelAcceptResponse is a constructor for a channel accept response,
+// which creates a response with an appropriately wrapped error (in the case of
+// a rejection) so that the error will be whitelisted and delivered to the
+// initiating peer. Accepted channels simply return a response containing a nil
+// error.
+func NewChannelAcceptResponse(accept bool, acceptErr error,
+	upfrontShutdown lnwire.DeliveryAddress, csvDelay, htlcLimit,
+	minDepth uint16, reserve btcutil.Amount, inFlight,
+	minHtlcIn lnwire.MilliSatoshi) *ChannelAcceptResponse {
+
+	resp := &ChannelAcceptResponse{
+		UpfrontShutdown: upfrontShutdown,
+		CSVDelay:        csvDelay,
+		Reserve:         reserve,
+		InFlightTotal:   inFlight,
+		HtlcLimit:       htlcLimit,
+		MinHtlcIn:       minHtlcIn,
+		MinAcceptDepth:  minDepth,
+	}
+
+	// If we want to accept the channel, we return a response with a nil
+	// error.
+	if accept {
+		return resp
+	}
+
+	// Use a generic error when no custom error is provided.
+	if acceptErr == nil {
+		acceptErr = errChannelRejected
+	}
+
+	resp.ChanAcceptError = ChanAcceptError{
+		error: acceptErr,
+	}
+
+	return resp
+}
+
+// RejectChannel returns a boolean that indicates whether we should reject the
+// channel.
+func (c *ChannelAcceptResponse) RejectChannel() bool {
+	return c.error != nil
+}
+
+// ChannelAcceptor is an interface that represents  a predicate on the data
 // contained in ChannelAcceptRequest.
 type ChannelAcceptor interface {
-	Accept(req *ChannelAcceptRequest) bool
+	Accept(req *ChannelAcceptRequest) *ChannelAcceptResponse
 }

chanacceptor/log.go

@@ -0,0 +1,32 @@
+package chanacceptor
+
+import (
+	"github.com/btcsuite/btclog"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+// Subsystem defines the logging code for this subsystem.
+const Subsystem = "CHAC"
+
+// log is a logger that is initialized with no output filters.  This
+// means the package will not perform any logging by default until the caller
+// requests it.
+var log btclog.Logger
+
+// The default amount of logging is none.
+func init() {
+	UseLogger(build.NewSubLogger(Subsystem, nil))
+}
+
+// DisableLog disables all library log output.  Logging output is disabled
+// by default until UseLogger is called.
+func DisableLog() {
+	UseLogger(btclog.Disabled)
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

chanacceptor/merge.go

@@ -0,0 +1,152 @@
+package chanacceptor
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/btcsuite/btcutil"
+	"github.com/lightningnetwork/lnd/lnwire"
+)
+
+const (
+	// We use field names in our errors for more readable errors. Create
+	// consts for them here so that we can exactly match in our unit tests.
+	fieldCSV             = "csv delay"
+	fieldHtlcLimit       = "htlc limit"
+	fieldMinDep          = "min depth"
+	fieldReserve         = "reserve"
+	fieldMinIn           = "min htlc in"
+	fieldInFlightTotal   = "in flight total"
+	fieldUpfrontShutdown = "upfront shutdown"
+)
+
+// fieldMismatchError returns a merge error for a named field when we get two
+// channel acceptor responses which have different values set.
+func fieldMismatchError(name string, current, new interface{}) error {
+	return fmt.Errorf("multiple values set for: %v, %v and %v",
+		name, current, new)
+}
+
+// mergeInt64 merges two int64 values, failing if they have different non-zero
+// values.
+func mergeInt64(name string, current, new int64) (int64, error) {
+	switch {
+	case current == 0:
+		return new, nil
+
+	case new == 0:
+		return current, nil
+
+	case current != new:
+		return 0, fieldMismatchError(name, current, new)
+
+	default:
+		return new, nil
+	}
+}
+
+// mergeMillisatoshi merges two msat values, failing if they have different
+// non-zero values.
+func mergeMillisatoshi(name string, current,
+	new lnwire.MilliSatoshi) (lnwire.MilliSatoshi, error) {
+
+	switch {
+	case current == 0:
+		return new, nil
+
+	case new == 0:
+		return current, nil
+
+	case current != new:
+		return 0, fieldMismatchError(name, current, new)
+
+	default:
+		return new, nil
+	}
+}
+
+// mergeDeliveryAddress merges two delivery address values, failing if they have
+// different non-zero values.
+func mergeDeliveryAddress(name string, current,
+	new lnwire.DeliveryAddress) (lnwire.DeliveryAddress, error) {
+
+	switch {
+	case current == nil:
+		return new, nil
+
+	case new == nil:
+		return current, nil
+
+	case !bytes.Equal(current, new):
+		return nil, fieldMismatchError(name, current, new)
+
+	default:
+		return new, nil
+	}
+}
+
+// mergeResponse takes two channel accept responses, and attempts to merge their
+// fields, failing if any fields conflict (are non-zero and not equal). It
+// returns a new response that has all the merged fields in it.
+func mergeResponse(current, new ChannelAcceptResponse) (ChannelAcceptResponse,
+	error) {
+
+	csv, err := mergeInt64(
+		fieldCSV, int64(current.CSVDelay), int64(new.CSVDelay),
+	)
+	if err != nil {
+		return current, err
+	}
+	current.CSVDelay = uint16(csv)
+
+	htlcLimit, err := mergeInt64(
+		fieldHtlcLimit, int64(current.HtlcLimit),
+		int64(new.HtlcLimit),
+	)
+	if err != nil {
+		return current, err
+	}
+	current.HtlcLimit = uint16(htlcLimit)
+
+	minDepth, err := mergeInt64(
+		fieldMinDep, int64(current.MinAcceptDepth),
+		int64(new.MinAcceptDepth),
+	)
+	if err != nil {
+		return current, err
+	}
+	current.MinAcceptDepth = uint16(minDepth)
+
+	reserve, err := mergeInt64(
+		fieldReserve, int64(current.Reserve), int64(new.Reserve),
+	)
+	if err != nil {
+		return current, err
+	}
+	current.Reserve = btcutil.Amount(reserve)
+
+	current.MinHtlcIn, err = mergeMillisatoshi(
+		fieldMinIn, current.MinHtlcIn, new.MinHtlcIn,
+	)
+	if err != nil {
+		return current, err
+	}
+
+	current.InFlightTotal, err = mergeMillisatoshi(
+		fieldInFlightTotal, current.InFlightTotal,
+		new.InFlightTotal,
+	)
+	if err != nil {
+		return current, err
+	}
+
+	current.UpfrontShutdown, err = mergeDeliveryAddress(
+		fieldUpfrontShutdown, current.UpfrontShutdown,
+		new.UpfrontShutdown,
+	)
+	if err != nil {
+		return current, err
+	}
+
+	return current, nil
+}

chanacceptor/merge_test.go

@@ -0,0 +1,188 @@
+package chanacceptor
+
+import (
+	"testing"
+
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/stretchr/testify/require"
+)
+
+// TestMergeResponse tests merging of channel acceptor responses.
+func TestMergeResponse(t *testing.T) {
+	var (
+		addr1 = lnwire.DeliveryAddress{1}
+		addr2 = lnwire.DeliveryAddress{2}
+
+		populatedResp = ChannelAcceptResponse{
+			UpfrontShutdown: addr1,
+			CSVDelay:        2,
+			Reserve:         3,
+			InFlightTotal:   4,
+			HtlcLimit:       5,
+			MinHtlcIn:       6,
+			MinAcceptDepth:  7,
+		}
+	)
+
+	tests := []struct {
+		name    string
+		current ChannelAcceptResponse
+		new     ChannelAcceptResponse
+		merged  ChannelAcceptResponse
+		err     error
+	}{
+		{
+			name:    "same response",
+			current: populatedResp,
+			new:     populatedResp,
+			merged:  populatedResp,
+			err:     nil,
+		},
+		{
+			name: "different upfront",
+			current: ChannelAcceptResponse{
+				UpfrontShutdown: addr1,
+			},
+			new: ChannelAcceptResponse{
+				UpfrontShutdown: addr2,
+			},
+			err: fieldMismatchError(fieldUpfrontShutdown, addr1, addr2),
+		},
+		{
+			name: "different csv",
+			current: ChannelAcceptResponse{
+				CSVDelay: 1,
+			},
+			new: ChannelAcceptResponse{
+				CSVDelay: 2,
+			},
+			err: fieldMismatchError(fieldCSV, 1, 2),
+		},
+		{
+			name: "different reserve",
+			current: ChannelAcceptResponse{
+				Reserve: 1,
+			},
+			new: ChannelAcceptResponse{
+				Reserve: 2,
+			},
+			err: fieldMismatchError(fieldReserve, 1, 2),
+		},
+		{
+			name: "different in flight",
+			current: ChannelAcceptResponse{
+				InFlightTotal: 1,
+			},
+			new: ChannelAcceptResponse{
+				InFlightTotal: 2,
+			},
+			err: fieldMismatchError(
+				fieldInFlightTotal, lnwire.MilliSatoshi(1),
+				lnwire.MilliSatoshi(2),
+			),
+		},
+		{
+			name: "different htlc limit",
+			current: ChannelAcceptResponse{
+				HtlcLimit: 1,
+			},
+			new: ChannelAcceptResponse{
+				HtlcLimit: 2,
+			},
+			err: fieldMismatchError(fieldHtlcLimit, 1, 2),
+		},
+		{
+			name: "different min in",
+			current: ChannelAcceptResponse{
+				MinHtlcIn: 1,
+			},
+			new: ChannelAcceptResponse{
+				MinHtlcIn: 2,
+			},
+			err: fieldMismatchError(
+				fieldMinIn, lnwire.MilliSatoshi(1),
+				lnwire.MilliSatoshi(2),
+			),
+		},
+		{
+			name: "different depth",
+			current: ChannelAcceptResponse{
+				MinAcceptDepth: 1,
+			},
+			new: ChannelAcceptResponse{
+				MinAcceptDepth: 2,
+			},
+			err: fieldMismatchError(fieldMinDep, 1, 2),
+		},
+		{
+			name: "merge all values",
+			current: ChannelAcceptResponse{
+				UpfrontShutdown: lnwire.DeliveryAddress{1},
+				CSVDelay:        1,
+				Reserve:         0,
+				InFlightTotal:   3,
+				HtlcLimit:       0,
+				MinHtlcIn:       5,
+				MinAcceptDepth:  0,
+			},
+			new: ChannelAcceptResponse{
+				UpfrontShutdown: nil,
+				CSVDelay:        0,
+				Reserve:         2,
+				InFlightTotal:   0,
+				HtlcLimit:       4,
+				MinHtlcIn:       0,
+				MinAcceptDepth:  6,
+			},
+			merged: ChannelAcceptResponse{
+				UpfrontShutdown: lnwire.DeliveryAddress{1},
+				CSVDelay:        1,
+				Reserve:         2,
+				InFlightTotal:   3,
+				HtlcLimit:       4,
+				MinHtlcIn:       5,
+				MinAcceptDepth:  6,
+			},
+			err: nil,
+		},
+		{
+			// Test the case where fields have the same non-zero
+			// value, and the case where only response value is
+			// non-zero.
+			name: "empty and identical",
+			current: ChannelAcceptResponse{
+				CSVDelay:      1,
+				Reserve:       2,
+				InFlightTotal: 0,
+			},
+			new: ChannelAcceptResponse{
+				CSVDelay:      0,
+				Reserve:       2,
+				InFlightTotal: 3,
+			},
+			merged: ChannelAcceptResponse{
+				CSVDelay:      1,
+				Reserve:       2,
+				InFlightTotal: 3,
+			},
+			err: nil,
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+
+		t.Run(test.name, func(t *testing.T) {
+			resp, err := mergeResponse(test.current, test.new)
+			require.Equal(t, test.err, err)
+
+			// If we expect an error, exit early rather than compare
+			// our result.
+			if test.err != nil {
+				return
+			}
+
+			require.Equal(t, test.merged, resp)
+		})
+	}
+}

chanacceptor/rpcacceptor.go

@@ -1,24 +1,402 @@
 package chanacceptor
 
+import (
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/btcsuite/btcd/chaincfg"
+	"github.com/btcsuite/btcutil"
+	"github.com/lightningnetwork/lnd/input"
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"github.com/lightningnetwork/lnd/lnwallet/chancloser"
+	"github.com/lightningnetwork/lnd/lnwire"
+)
+
+var (
+	errShuttingDown = errors.New("server shutting down")
+
+	// errCustomLength is returned when our custom error's length exceeds
+	// our maximum.
+	errCustomLength = fmt.Errorf("custom error message exceeds length "+
+		"limit: %v", maxErrorLength)
+
+	// errInvalidUpfrontShutdown is returned when we cannot parse the
+	// upfront shutdown address returned.
+	errInvalidUpfrontShutdown = fmt.Errorf("could not parse upfront " +
+		"shutdown address")
+
+	// errInsufficientReserve is returned when the reserve proposed by for
+	// a channel is less than the dust limit originally supplied.
+	errInsufficientReserve = fmt.Errorf("reserve lower than proposed dust " +
+		"limit")
+
+	// errAcceptWithError is returned when we get a response which accepts
+	// a channel but ambiguously also sets a custom error message.
+	errAcceptWithError = errors.New("channel acceptor response accepts " +
+		"channel, but also includes custom error")
+
+	// errMaxHtlcTooHigh is returned if our htlc count exceeds the number
+	// hard-set by BOLT 2.
+	errMaxHtlcTooHigh = fmt.Errorf("htlc limit exceeds spec limit of: %v",
+		input.MaxHTLCNumber/2)
+
+	// maxErrorLength is the maximum error length we allow the error we
+	// send to our peer to be.
+	maxErrorLength = 500
+)
+
+// chanAcceptInfo contains a request for a channel acceptor decision, and a
+// channel that the response should be sent on.
+type chanAcceptInfo struct {
+	request  *ChannelAcceptRequest
+	response chan *ChannelAcceptResponse
+}
+
 // RPCAcceptor represents the RPC-controlled variant of the ChannelAcceptor.
 // One RPCAcceptor allows one RPC client.
 type RPCAcceptor struct {
-	acceptClosure func(req *ChannelAcceptRequest) bool
+	// receive is a function from which we receive channel acceptance
+	// decisions. Note that this function is expected to block.
+	receive func() (*lnrpc.ChannelAcceptResponse, error)
+
+	// send is a function which sends requests for channel acceptance
+	// decisions into our rpc stream.
+	send func(request *lnrpc.ChannelAcceptRequest) error
+
+	// requests is a channel that we send requests for a acceptor response
+	// into.
+	requests chan *chanAcceptInfo
+
+	// timeout is the amount of time we allow the channel acceptance
+	// decision to take. This time includes the time to send a query to the
+	// acceptor, and the time it takes to receive a response.
+	timeout time.Duration
+
+	// params are our current chain params.
+	params *chaincfg.Params
+
+	// done is closed when the rpc client terminates.
+	done chan struct{}
+
+	// quit is closed when lnd is shutting down.
+	quit chan struct{}
+
+	wg sync.WaitGroup
 }
 
 // Accept is a predicate on the ChannelAcceptRequest which is sent to the RPC
-// client who will respond with the ultimate decision. This assumes an accept
-// closure has been specified during creation.
+// client who will respond with the ultimate decision. This function passes the
+// request into the acceptor's requests channel, and returns the response it
+// receives, failing the request if the timeout elapses.
 //
 // NOTE: Part of the ChannelAcceptor interface.
-func (r *RPCAcceptor) Accept(req *ChannelAcceptRequest) bool {
-	return r.acceptClosure(req)
+func (r *RPCAcceptor) Accept(req *ChannelAcceptRequest) *ChannelAcceptResponse {
+	respChan := make(chan *ChannelAcceptResponse, 1)
+
+	newRequest := &chanAcceptInfo{
+		request:  req,
+		response: respChan,
+	}
+
+	// timeout is the time after which ChannelAcceptRequests expire.
+	timeout := time.After(r.timeout)
+
+	// Create a rejection response which we can use for the cases where we
+	// reject the channel.
+	rejectChannel := NewChannelAcceptResponse(
+		false, errChannelRejected, nil, 0, 0, 0, 0, 0, 0,
+	)
+
+	// Send the request to the newRequests channel.
+	select {
+	case r.requests <- newRequest:
+
+	case <-timeout:
+		log.Errorf("RPCAcceptor returned false - reached timeout of %v",
+			r.timeout)
+		return rejectChannel
+
+	case <-r.done:
+		return rejectChannel
+
+	case <-r.quit:
+		return rejectChannel
+	}
+
+	// Receive the response and return it. If no response has been received
+	// in AcceptorTimeout, then return false.
+	select {
+	case resp := <-respChan:
+		return resp
+
+	case <-timeout:
+		log.Errorf("RPCAcceptor returned false - reached timeout of %v",
+			r.timeout)
+		return rejectChannel
+
+	case <-r.done:
+		return rejectChannel
+
+	case <-r.quit:
+		return rejectChannel
+	}
 }
 
 // NewRPCAcceptor creates and returns an instance of the RPCAcceptor.
-func NewRPCAcceptor(closure func(*ChannelAcceptRequest) bool) *RPCAcceptor {
+func NewRPCAcceptor(receive func() (*lnrpc.ChannelAcceptResponse, error),
+	send func(*lnrpc.ChannelAcceptRequest) error, timeout time.Duration,
+	params *chaincfg.Params, quit chan struct{}) *RPCAcceptor {
+
 	return &RPCAcceptor{
-		acceptClosure: closure,
+		receive:  receive,
+		send:     send,
+		requests: make(chan *chanAcceptInfo),
+		timeout:  timeout,
+		params:   params,
+		done:     make(chan struct{}),
+		quit:     quit,
+	}
+}
+
+// Run is the main loop for the RPC Acceptor. This function will block until
+// it receives the signal that lnd is shutting down, or the rpc stream is
+// cancelled by the client.
+func (r *RPCAcceptor) Run() error {
+	// Wait for our goroutines to exit before we return.
+	defer r.wg.Wait()
+
+	// Create a channel that responses from acceptors are sent into.
+	responses := make(chan lnrpc.ChannelAcceptResponse)
+
+	// errChan is used by the receive loop to signal any errors that occur
+	// during reading from the stream. This is primarily used to shutdown
+	// the send loop in the case of an RPC client disconnecting.
+	errChan := make(chan error, 1)
+
+	// Start a goroutine to receive responses from the channel acceptor.
+	// We expect the receive function to block, so it must be run in a
+	// goroutine (otherwise we could not send more than one channel accept
+	// request to the client).
+	r.wg.Add(1)
+	go func() {
+		r.receiveResponses(errChan, responses)
+		r.wg.Done()
+	}()
+
+	return r.sendAcceptRequests(errChan, responses)
+}
+
+// receiveResponses receives responses for our channel accept requests and
+// dispatches them into the responses channel provided, sending any errors that
+// occur into the error channel provided.
+func (r *RPCAcceptor) receiveResponses(errChan chan error,
+	responses chan lnrpc.ChannelAcceptResponse) {
+
+	for {
+		resp, err := r.receive()
+		if err != nil {
+			errChan <- err
+			return
+		}
+
+		var pendingID [32]byte
+		copy(pendingID[:], resp.PendingChanId)
+
+		openChanResp := lnrpc.ChannelAcceptResponse{
+			Accept:          resp.Accept,
+			PendingChanId:   pendingID[:],
+			Error:           resp.Error,
+			UpfrontShutdown: resp.UpfrontShutdown,
+			CsvDelay:        resp.CsvDelay,
+			ReserveSat:      resp.ReserveSat,
+			InFlightMaxMsat: resp.InFlightMaxMsat,
+			MaxHtlcCount:    resp.MaxHtlcCount,
+			MinHtlcIn:       resp.MinHtlcIn,
+			MinAcceptDepth:  resp.MinAcceptDepth,
+		}
+
+		// We have received a decision for one of our channel
+		// acceptor requests.
+		select {
+		case responses <- openChanResp:
+
+		case <-r.done:
+			return
+
+		case <-r.quit:
+			return
+		}
+	}
+}
+
+// sendAcceptRequests handles channel acceptor requests sent to us by our
+// Accept() function, dispatching them to our acceptor stream and coordinating
+// return of responses to their callers.
+func (r *RPCAcceptor) sendAcceptRequests(errChan chan error,
+	responses chan lnrpc.ChannelAcceptResponse) error {
+
+	// Close the done channel to indicate that the acceptor is no longer
+	// listening and any in-progress requests should be terminated.
+	defer close(r.done)
+
+	// Create a map of pending channel IDs to our original open channel
+	// request and a response channel. We keep the original chanel open
+	// message so that we can validate our response against it.
+	acceptRequests := make(map[[32]byte]*chanAcceptInfo)
+
+	for {
+		select {
+		// Consume requests passed to us from our Accept() function and
+		// send them into our stream.
+		case newRequest := <-r.requests:
+
+			req := newRequest.request
+			pendingChanID := req.OpenChanMsg.PendingChannelID
+
+			acceptRequests[pendingChanID] = newRequest
+
+			// A ChannelAcceptRequest has been received, send it to the client.
+			chanAcceptReq := &lnrpc.ChannelAcceptRequest{
+				NodePubkey:       req.Node.SerializeCompressed(),
+				ChainHash:        req.OpenChanMsg.ChainHash[:],
+				PendingChanId:    req.OpenChanMsg.PendingChannelID[:],
+				FundingAmt:       uint64(req.OpenChanMsg.FundingAmount),
+				PushAmt:          uint64(req.OpenChanMsg.PushAmount),
+				DustLimit:        uint64(req.OpenChanMsg.DustLimit),
+				MaxValueInFlight: uint64(req.OpenChanMsg.MaxValueInFlight),
+				ChannelReserve:   uint64(req.OpenChanMsg.ChannelReserve),
+				MinHtlc:          uint64(req.OpenChanMsg.HtlcMinimum),
+				FeePerKw:         uint64(req.OpenChanMsg.FeePerKiloWeight),
+				CsvDelay:         uint32(req.OpenChanMsg.CsvDelay),
+				MaxAcceptedHtlcs: uint32(req.OpenChanMsg.MaxAcceptedHTLCs),
+				ChannelFlags:     uint32(req.OpenChanMsg.ChannelFlags),
+			}
+
+			if err := r.send(chanAcceptReq); err != nil {
+				return err
+			}
+
+		// Process newly received responses from our channel acceptor,
+		// looking the original request up in our map of requests and
+		// dispatching the response.
+		case resp := <-responses:
+			// Look up the appropriate channel to send on given the
+			// pending ID. If a channel is found, send the response
+			// over it.
+			var pendingID [32]byte
+			copy(pendingID[:], resp.PendingChanId)
+			requestInfo, ok := acceptRequests[pendingID]
+			if !ok {
+				continue
+			}
+
+			// Validate the response we have received. If it is not
+			// valid, we log our error and proceed to deliver the
+			// rejection.
+			accept, acceptErr, shutdown, err := r.validateAcceptorResponse(
+				requestInfo.request.OpenChanMsg.DustLimit, resp,
+			)
+			if err != nil {
+				log.Errorf("Invalid acceptor response: %v", err)
+			}
+
+			requestInfo.response <- NewChannelAcceptResponse(
+				accept, acceptErr, shutdown,
+				uint16(resp.CsvDelay),
+				uint16(resp.MaxHtlcCount),
+				uint16(resp.MinAcceptDepth),
+				btcutil.Amount(resp.ReserveSat),
+				lnwire.MilliSatoshi(resp.InFlightMaxMsat),
+				lnwire.MilliSatoshi(resp.MinHtlcIn),
+			)
+
+			// Delete the channel from the acceptRequests map.
+			delete(acceptRequests, pendingID)
+
+		// If we failed to receive from our acceptor, we exit.
+		case err := <-errChan:
+			log.Errorf("Received an error: %v, shutting down", err)
+			return err
+
+		// Exit if we are shutting down.
+		case <-r.quit:
+			return errShuttingDown
+		}
+	}
+}
+
+// validateAcceptorResponse validates the response we get from the channel
+// acceptor, returning a boolean indicating whether to accept the channel, an
+// error to send to the peer, and any validation errors that occurred.
+func (r *RPCAcceptor) validateAcceptorResponse(dustLimit btcutil.Amount,
+	req lnrpc.ChannelAcceptResponse) (bool, error, lnwire.DeliveryAddress,
+	error) {
+
+	channelStr := hex.EncodeToString(req.PendingChanId)
+
+	// Check that the max htlc count is within the BOLT 2 hard-limit of 483.
+	// The initiating side should fail values above this anyway, but we
+	// catch the invalid user input here.
+	if req.MaxHtlcCount > input.MaxHTLCNumber/2 {
+		log.Errorf("Max htlc count: %v for channel: %v is greater "+
+			"than limit of: %v", req.MaxHtlcCount, channelStr,
+			input.MaxHTLCNumber/2)
+
+		return false, errChannelRejected, nil, errMaxHtlcTooHigh
+	}
+
+	// Ensure that the reserve that has been proposed, if it is set, is at
+	// least the dust limit that was proposed by the remote peer. This is
+	// required by BOLT 2.
+	reserveSat := btcutil.Amount(req.ReserveSat)
+	if reserveSat != 0 && reserveSat < dustLimit {
+		log.Errorf("Remote reserve: %v sat for channel: %v must be "+
+			"at least equal to proposed dust limit: %v",
+			req.ReserveSat, channelStr, dustLimit)
+
+		return false, errChannelRejected, nil, errInsufficientReserve
+	}
+
+	// Attempt to parse the upfront shutdown address provided.
+	upfront, err := chancloser.ParseUpfrontShutdownAddress(
+		req.UpfrontShutdown, r.params,
+	)
+	if err != nil {
+		log.Errorf("Could not parse upfront shutdown for "+
+			"%v: %v", channelStr, err)
+
+		return false, errChannelRejected, nil, errInvalidUpfrontShutdown
+	}
+
+	// Check that the custom error provided is valid.
+	if len(req.Error) > maxErrorLength {
+		return false, errChannelRejected, nil, errCustomLength
+	}
+
+	var haveCustomError = len(req.Error) != 0
+
+	switch {
+	// If accept is true, but we also have an error specified, we fail
+	// because this result is ambiguous.
+	case req.Accept && haveCustomError:
+		return false, errChannelRejected, nil, errAcceptWithError
+
+	// If we accept without an error message, we can just return a nil
+	// error.
+	case req.Accept:
+		return true, nil, upfront, nil
+
+	// If we reject the channel, and have a custom error, then we use it.
+	case haveCustomError:
+		return false, fmt.Errorf(req.Error), nil, nil
+
+	// Otherwise, we have rejected the channel with no custom error, so we
+	// just use a generic error to fail the channel.
+	default:
+		return false, errChannelRejected, nil, nil
 	}
 }
 

chanacceptor/rpcacceptor_test.go

@@ -0,0 +1,139 @@
+package chanacceptor
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/btcsuite/btcd/chaincfg"
+	"github.com/btcsuite/btcutil"
+	"github.com/lightningnetwork/lnd/input"
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"github.com/lightningnetwork/lnd/lnwallet/chancloser"
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/stretchr/testify/require"
+)
+
+// TestValidateAcceptorResponse test validation of acceptor responses.
+func TestValidateAcceptorResponse(t *testing.T) {
+	var (
+		customError = errors.New("custom error")
+		validAddr   = "bcrt1qwrmq9uca0t3dy9t9wtuq5tm4405r7tfzyqn9pp"
+		addr, _     = chancloser.ParseUpfrontShutdownAddress(
+			validAddr, &chaincfg.TestNet3Params,
+		)
+	)
+
+	tests := []struct {
+		name        string
+		dustLimit   btcutil.Amount
+		response    lnrpc.ChannelAcceptResponse
+		accept      bool
+		acceptorErr error
+		error       error
+		shutdown    lnwire.DeliveryAddress
+	}{
+		{
+			name: "accepted with error",
+			response: lnrpc.ChannelAcceptResponse{
+				Accept: true,
+				Error:  customError.Error(),
+			},
+			accept:      false,
+			acceptorErr: errChannelRejected,
+			error:       errAcceptWithError,
+		},
+		{
+			name: "custom error too long",
+			response: lnrpc.ChannelAcceptResponse{
+				Accept: false,
+				Error:  strings.Repeat(" ", maxErrorLength+1),
+			},
+			accept:      false,
+			acceptorErr: errChannelRejected,
+			error:       errCustomLength,
+		},
+		{
+			name: "accepted",
+			response: lnrpc.ChannelAcceptResponse{
+				Accept:          true,
+				UpfrontShutdown: validAddr,
+			},
+			accept:      true,
+			acceptorErr: nil,
+			error:       nil,
+			shutdown:    addr,
+		},
+		{
+			name: "rejected with error",
+			response: lnrpc.ChannelAcceptResponse{
+				Accept: false,
+				Error:  customError.Error(),
+			},
+			accept:      false,
+			acceptorErr: customError,
+			error:       nil,
+		},
+		{
+			name: "rejected with no error",
+			response: lnrpc.ChannelAcceptResponse{
+				Accept: false,
+			},
+			accept:      false,
+			acceptorErr: errChannelRejected,
+			error:       nil,
+		},
+		{
+			name: "invalid upfront shutdown",
+			response: lnrpc.ChannelAcceptResponse{
+				Accept:          true,
+				UpfrontShutdown: "invalid addr",
+			},
+			accept:      false,
+			acceptorErr: errChannelRejected,
+			error:       errInvalidUpfrontShutdown,
+		},
+		{
+			name:      "reserve too low",
+			dustLimit: 100,
+			response: lnrpc.ChannelAcceptResponse{
+				Accept:     true,
+				ReserveSat: 10,
+			},
+			accept:      false,
+			acceptorErr: errChannelRejected,
+			error:       errInsufficientReserve,
+		},
+		{
+			name:      "max htlcs too high",
+			dustLimit: 100,
+			response: lnrpc.ChannelAcceptResponse{
+				Accept:       true,
+				MaxHtlcCount: 1 + input.MaxHTLCNumber/2,
+			},
+			accept:      false,
+			acceptorErr: errChannelRejected,
+			error:       errMaxHtlcTooHigh,
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+
+		t.Run(test.name, func(t *testing.T) {
+			// Create an acceptor, everything can be nil because
+			// we just need the params.
+			acceptor := NewRPCAcceptor(
+				nil, nil, 0, &chaincfg.TestNet3Params, nil,
+			)
+
+			accept, acceptErr, shutdown, err := acceptor.validateAcceptorResponse(
+				test.dustLimit, test.response,
+			)
+			require.Equal(t, test.accept, accept)
+			require.Equal(t, test.acceptorErr, acceptErr)
+			require.Equal(t, test.error, err)
+			require.Equal(t, test.shutdown, shutdown)
+		})
+	}
+}

fundingmanager.go

@@ -746,6 +746,8 @@ func (f *fundingManager) failFundingFlow(peer lnpeer.Peer, tempChanID [32]byte,
 		msg = lnwire.ErrorData(e.Error())
 	case lnwire.FundingError:
 		msg = lnwire.ErrorData(e.Error())
+	case chanacceptor.ChanAcceptError:
+		msg = lnwire.ErrorData(e.Error())
 
 	// For all other error types we just send a generic error.
 	default:
@@ -1282,10 +1284,13 @@ func (f *fundingManager) handleFundingOpen(peer lnpeer.Peer,
 		OpenChanMsg: msg,
 	}
 
-	if !f.cfg.OpenChannelPredicate.Accept(chanReq) {
+	// Query our channel acceptor to determine whether we should reject
+	// the channel.
+	acceptorResp := f.cfg.OpenChannelPredicate.Accept(chanReq)
+	if acceptorResp.RejectChannel() {
 		f.failFundingFlow(
 			peer, msg.PendingChannelID,
-			fmt.Errorf("open channel request rejected"),
+			acceptorResp.ChanAcceptError,
 		)
 		return
 	}
@@ -1335,8 +1340,12 @@ func (f *fundingManager) handleFundingOpen(peer lnpeer.Peer,
 	// that we require before both of us consider the channel open. We'll
 	// use our mapping to derive the proper number of confirmations based on
 	// the amount of the channel, and also if any funds are being pushed to
-	// us.
+	// us. If a depth value was set by our channel acceptor, we will use
+	// that value instead.
 	numConfsReq := f.cfg.NumRequiredConfs(msg.FundingAmount, msg.PushAmount)
+	if acceptorResp.MinAcceptDepth != 0 {
+		numConfsReq = acceptorResp.MinAcceptDepth
+	}
 	reservation.SetNumConfsRequired(numConfsReq)
 
 	// We'll also validate and apply all the constraints the initiating
@@ -1360,10 +1369,10 @@ func (f *fundingManager) handleFundingOpen(peer lnpeer.Peer,
 
 	// Check whether the peer supports upfront shutdown, and get a new wallet
 	// address if our node is configured to set shutdown addresses by default.
-	// A nil address is set in place of user input, because this channel open
-	// was not initiated by the user.
+	// We use the upfront shutdown script provided by our channel acceptor
+	// (if any) in lieu of user input.
 	shutdown, err := getUpfrontShutdownScript(
-		f.cfg.EnableUpfrontShutdown, peer, nil,
+		f.cfg.EnableUpfrontShutdown, peer, acceptorResp.UpfrontShutdown,
 		func() (lnwire.DeliveryAddress, error) {
 			addr, err := f.cfg.Wallet.NewAddress(lnwallet.WitnessPubKey, false)
 			if err != nil {
@@ -1386,12 +1395,34 @@ func (f *fundingManager) handleFundingOpen(peer lnpeer.Peer,
 		msg.PendingChannelID, amt, msg.PushAmount,
 		commitType, msg.UpfrontShutdownScript)
 
-	// Generate our required constraints for the remote party.
+	// Generate our required constraints for the remote party, using the
+	// values provided by the channel acceptor if they are non-zero.
 	remoteCsvDelay := f.cfg.RequiredRemoteDelay(amt)
+	if acceptorResp.CSVDelay != 0 {
+		remoteCsvDelay = acceptorResp.CSVDelay
+	}
+
 	chanReserve := f.cfg.RequiredRemoteChanReserve(amt, msg.DustLimit)
+	if acceptorResp.Reserve != 0 {
+		chanReserve = acceptorResp.Reserve
+	}
+
 	remoteMaxValue := f.cfg.RequiredRemoteMaxValue(amt)
+	if acceptorResp.InFlightTotal != 0 {
+		remoteMaxValue = acceptorResp.InFlightTotal
+	}
+
 	maxHtlcs := f.cfg.RequiredRemoteMaxHTLCs(amt)
+	if acceptorResp.HtlcLimit != 0 {
+		maxHtlcs = acceptorResp.HtlcLimit
+	}
+
+	// Default to our default minimum hltc value, replacing it with the
+	// channel acceptor's value if it is set.
 	minHtlc := f.cfg.DefaultMinHtlcIn
+	if acceptorResp.MinHtlcIn != 0 {
+		minHtlc = acceptorResp.MinHtlcIn
+	}
 
 	// Once the reservation has been created successfully, we add it to
 	// this peer's map of pending reservations to track this particular

lnrpc/rpc.proto

@@ -791,6 +791,58 @@ message ChannelAcceptResponse {
 
     // The pending channel id to which this response applies.
     bytes pending_chan_id = 2;
+
+    /*
+    An optional error to send the initiating party to indicate why the channel
+    was rejected. This field *should not* contain sensitive information, it will
+    be sent to the initiating party. This field should only be set if accept is
+    false, the channel will be rejected if an error is set with accept=true
+    because the meaning of this response is ambiguous. Limited to 500
+    characters.
+    */
+    string error = 3;
+
+    /*
+    The upfront shutdown address to use if the initiating peer supports option
+    upfront shutdown script (see ListPeers for the features supported). Note
+    that the channel open will fail if this value is set for a peer that does
+    not support this feature bit.
+    */
+    string upfront_shutdown = 4;
+
+    /*
+    The csv delay (in blocks) that we require for the remote party.
+    */
+    uint32 csv_delay = 5;
+
+    /*
+    The reserve amount in satoshis that we require the remote peer to adhere to.
+    We require that the remote peer always have some reserve amount allocated to
+    them so that there is always a disincentive to broadcast old state (if they
+    hold 0 sats on their side of the channel, there is nothing to lose).
+    */
+    uint64 reserve_sat = 6;
+
+    /*
+    The maximum amount of funds in millisatoshis that we allow the remote peer
+    to have in outstanding htlcs.
+    */
+    uint64 in_flight_max_msat = 7;
+
+    /*
+    The maximum number of htlcs that the remote peer can offer us.
+    */
+    uint32 max_htlc_count = 8;
+
+    /*
+    The minimum value in millisatoshis for incoming htlcs on the channel.
+    */
+    uint64 min_htlc_in = 9;
+
+    /*
+    The number of confirmations we require before we consider the channel open.
+    */
+    uint32 min_accept_depth = 10;
 }
 
 message ChannelPoint {

lnwallet/chancloser/chancloser.go

@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"fmt"
 
+	"github.com/btcsuite/btcd/chaincfg"
+	"github.com/btcsuite/btcd/txscript"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btcutil"
 	"github.com/davecgh/go-spew/spew"
@@ -717,3 +719,23 @@ func calcCompromiseFee(chanPoint wire.OutPoint, ourIdealFee, lastSentFee,
 		return remoteFee
 	}
 }
+
+// ParseUpfrontShutdownAddress attempts to parse an upfront shutdown address.
+// If the address is empty, it returns nil. If it successfully decoded the
+// address, it returns a script that pays out to the address.
+func ParseUpfrontShutdownAddress(address string,
+	params *chaincfg.Params) (lnwire.DeliveryAddress, error) {
+
+	if len(address) == 0 {
+		return nil, nil
+	}
+
+	addr, err := btcutil.DecodeAddress(
+		address, params,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("invalid address: %v", err)
+	}
+
+	return txscript.PayToAddrScript(addr)
+}

log.go

@@ -11,6 +11,7 @@ import (
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/chainreg"
+	"github.com/lightningnetwork/lnd/chanacceptor"
 	"github.com/lightningnetwork/lnd/chanbackup"
 	"github.com/lightningnetwork/lnd/chanfitness"
 	"github.com/lightningnetwork/lnd/channeldb"
@@ -133,6 +134,7 @@ func SetupLoggers(root *build.RotatingLogWriter) {
 	AddSubLogger(root, verrpc.Subsystem, verrpc.UseLogger)
 	AddSubLogger(root, healthcheck.Subsystem, healthcheck.UseLogger)
 	AddSubLogger(root, chainreg.Subsystem, chainreg.UseLogger)
+	AddSubLogger(root, chanacceptor.Subsystem, chanacceptor.UseLogger)
 }
 
 // AddSubLogger is a helper method to conveniently create and register the

rpcserver.go

@@ -56,6 +56,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwallet/btcwallet"
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
+	"github.com/lightningnetwork/lnd/lnwallet/chancloser"
 	"github.com/lightningnetwork/lnd/lnwallet/chanfunding"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/macaroons"
@@ -1878,7 +1879,7 @@ func (r *rpcServer) parseOpenChannelReq(in *lnrpc.OpenChannelRequest,
 	rpcsLog.Debugf("[openchannel]: using fee of %v sat/kw for funding tx",
 		int64(feeRate))
 
-	script, err := parseUpfrontShutdownAddress(
+	script, err := chancloser.ParseUpfrontShutdownAddress(
 		in.CloseAddress, r.cfg.ActiveNetParams.Params,
 	)
 	if err != nil {
@@ -2053,26 +2054,6 @@ func (r *rpcServer) OpenChannelSync(ctx context.Context,
 	}
 }
 
-// parseUpfrontShutdownScript attempts to parse an upfront shutdown address.
-// If the address is empty, it returns nil. If it successfully decoded the
-// address, it returns a script that pays out to the address.
-func parseUpfrontShutdownAddress(address string,
-	params *chaincfg.Params) (lnwire.DeliveryAddress, error) {
-
-	if len(address) == 0 {
-		return nil, nil
-	}
-
-	addr, err := btcutil.DecodeAddress(
-		address, params,
-	)
-	if err != nil {
-		return nil, fmt.Errorf("invalid address: %v", err)
-	}
-
-	return txscript.PayToAddrScript(addr)
-}
-
 // GetChanPointFundingTxid returns the given channel point's funding txid in
 // raw bytes.
 func GetChanPointFundingTxid(chanPoint *lnrpc.ChannelPoint) (*chainhash.Hash, error) {
@@ -6401,14 +6382,6 @@ func (r *rpcServer) SubscribeChannelBackups(req *lnrpc.ChannelBackupSubscription
 	}
 }
 
-// chanAcceptInfo is used in the ChannelAcceptor bidirectional stream and
-// encapsulates the request information sent from the RPCAcceptor to the
-// RPCServer.
-type chanAcceptInfo struct {
-	chanReq      *chanacceptor.ChannelAcceptRequest
-	responseChan chan bool
-}
-
 // ChannelAcceptor dispatches a bi-directional streaming RPC in which
 // OpenChannel requests are sent to the client and the client responds with
 // a boolean that tells LND whether or not to accept the channel. This allows
@@ -6417,153 +6390,23 @@ type chanAcceptInfo struct {
 func (r *rpcServer) ChannelAcceptor(stream lnrpc.Lightning_ChannelAcceptorServer) error {
 	chainedAcceptor := r.chanPredicate
 
-	// Create two channels to handle requests and responses respectively.
-	newRequests := make(chan *chanAcceptInfo)
-	responses := make(chan lnrpc.ChannelAcceptResponse)
-
-	// Define a quit channel that will be used to signal to the RPCAcceptor's
-	// closure whether the stream still exists.
-	quit := make(chan struct{})
-	defer close(quit)
-
-	// demultiplexReq is a closure that will be passed to the RPCAcceptor and
-	// acts as an intermediary between the RPCAcceptor and the RPCServer.
-	demultiplexReq := func(req *chanacceptor.ChannelAcceptRequest) bool {
-		respChan := make(chan bool, 1)
-
-		newRequest := &chanAcceptInfo{
-			chanReq:      req,
-			responseChan: respChan,
-		}
-
-		// timeout is the time after which ChannelAcceptRequests expire.
-		timeout := time.After(r.cfg.AcceptorTimeout)
-
-		// Send the request to the newRequests channel.
-		select {
-		case newRequests <- newRequest:
-		case <-timeout:
-			rpcsLog.Errorf("RPCAcceptor returned false - reached timeout of %d",
-				r.cfg.AcceptorTimeout)
-			return false
-		case <-quit:
-			return false
-		case <-r.quit:
-			return false
-		}
-
-		// Receive the response and return it. If no response has been received
-		// in AcceptorTimeout, then return false.
-		select {
-		case resp := <-respChan:
-			return resp
-		case <-timeout:
-			rpcsLog.Errorf("RPCAcceptor returned false - reached timeout of %d",
-				r.cfg.AcceptorTimeout)
-			return false
-		case <-quit:
-			return false
-		case <-r.quit:
-			return false
-		}
-	}
-
-	// Create a new RPCAcceptor via the NewRPCAcceptor method.
-	rpcAcceptor := chanacceptor.NewRPCAcceptor(demultiplexReq)
+	// Create a new RPCAcceptor which will send requests into the
+	// newRequests channel when it receives them.
+	rpcAcceptor := chanacceptor.NewRPCAcceptor(
+		stream.Recv, stream.Send, r.cfg.AcceptorTimeout,
+		r.cfg.ActiveNetParams.Params, r.quit,
+	)
 
 	// Add the RPCAcceptor to the ChainedAcceptor and defer its removal.
 	id := chainedAcceptor.AddAcceptor(rpcAcceptor)
 	defer chainedAcceptor.RemoveAcceptor(id)
 
-	// errChan is used by the receive loop to signal any errors that occur
-	// during reading from the stream. This is primarily used to shutdown the
-	// send loop in the case of an RPC client disconnecting.
-	errChan := make(chan error, 1)
-
-	// We need to have the stream.Recv() in a goroutine since the call is
-	// blocking and would prevent us from sending more ChannelAcceptRequests to
-	// the RPC client.
-	go func() {
-		for {
-			resp, err := stream.Recv()
-			if err != nil {
-				errChan <- err
-				return
-			}
-
-			var pendingID [32]byte
-			copy(pendingID[:], resp.PendingChanId)
-
-			openChanResp := lnrpc.ChannelAcceptResponse{
-				Accept:        resp.Accept,
-				PendingChanId: pendingID[:],
-			}
-
-			// Now that we have the response from the RPC client, send it to
-			// the responses chan.
-			select {
-			case responses <- openChanResp:
-			case <-quit:
-				return
-			case <-r.quit:
-				return
-			}
-		}
-	}()
-
-	acceptRequests := make(map[[32]byte]chan bool)
-
-	for {
-		select {
-		case newRequest := <-newRequests:
-
-			req := newRequest.chanReq
-			pendingChanID := req.OpenChanMsg.PendingChannelID
-
-			acceptRequests[pendingChanID] = newRequest.responseChan
-
-			// A ChannelAcceptRequest has been received, send it to the client.
-			chanAcceptReq := &lnrpc.ChannelAcceptRequest{
-				NodePubkey:       req.Node.SerializeCompressed(),
-				ChainHash:        req.OpenChanMsg.ChainHash[:],
-				PendingChanId:    req.OpenChanMsg.PendingChannelID[:],
-				FundingAmt:       uint64(req.OpenChanMsg.FundingAmount),
-				PushAmt:          uint64(req.OpenChanMsg.PushAmount),
-				DustLimit:        uint64(req.OpenChanMsg.DustLimit),
-				MaxValueInFlight: uint64(req.OpenChanMsg.MaxValueInFlight),
-				ChannelReserve:   uint64(req.OpenChanMsg.ChannelReserve),
-				MinHtlc:          uint64(req.OpenChanMsg.HtlcMinimum),
-				FeePerKw:         uint64(req.OpenChanMsg.FeePerKiloWeight),
-				CsvDelay:         uint32(req.OpenChanMsg.CsvDelay),
-				MaxAcceptedHtlcs: uint32(req.OpenChanMsg.MaxAcceptedHTLCs),
-				ChannelFlags:     uint32(req.OpenChanMsg.ChannelFlags),
-			}
-
-			if err := stream.Send(chanAcceptReq); err != nil {
-				return err
-			}
-		case resp := <-responses:
-			// Look up the appropriate channel to send on given the pending ID.
-			// If a channel is found, send the response over it.
-			var pendingID [32]byte
-			copy(pendingID[:], resp.PendingChanId)
-			respChan, ok := acceptRequests[pendingID]
-			if !ok {
-				continue
-			}
-
-			// Send the response boolean over the buffered response channel.
-			respChan <- resp.Accept
-
-			// Delete the channel from the acceptRequests map.
-			delete(acceptRequests, pendingID)
-		case err := <-errChan:
-			rpcsLog.Errorf("Received an error: %v, shutting down", err)
-			return err
-		case <-r.quit:
-			return fmt.Errorf("RPC server is shutting down")
-		}
-	}
+	// Run the rpc acceptor, which will accept requests for channel
+	// acceptance decisions from our chained acceptor, send them to the
+	// channel acceptor and listen for and report responses. This function
+	// blocks, and will exit if the rpcserver receives the instruction to
+	// shutdown, or the client cancels.
+	return rpcAcceptor.Run()
 }
 
 // BakeMacaroon allows the creation of a new macaroon with custom read and write