reusable open channel logic and share macaroon

cln_client.go

@@ -43,25 +43,23 @@ func (c *ClnClient) GetInfo() (*GetInfoResult, error) {
 	}, nil
 }
 
-func (c *ClnClient) IsConnected(destination []byte) (*bool, error) {
+func (c *ClnClient) IsConnected(destination []byte) (bool, error) {
 	pubKey := hex.EncodeToString(destination)
 	peers, err := c.client.ListPeers()
 	if err != nil {
 		log.Printf("CLN: client.ListPeers() error: %v", err)
-		return nil, fmt.Errorf("CLN: client.ListPeers() error: %w", err)
+		return false, fmt.Errorf("CLN: client.ListPeers() error: %w", err)
 	}
 
 	for _, peer := range peers {
 		if pubKey == peer.Id {
 			log.Printf("destination online: %x", destination)
-			result := true
-			return &result, nil
+			return true, nil
 		}
 	}
 
 	log.Printf("CLN: destination offline: %x", destination)
-	result := false
-	return &result, nil
+	return false, nil
 }
 
 func (c *ClnClient) OpenChannel(req *OpenChannelRequest) (*wire.OutPoint, error) {

cln_interceptor.go

@@ -8,7 +8,6 @@ import (
 	"log"
 	"os"
 	"sync"
-	"time"
 
 	sphinx "github.com/lightningnetwork/lightning-onion"
 	"github.com/lightningnetwork/lnd/record"
@@ -100,8 +99,8 @@ func (i *ClnHtlcInterceptor) OnHtlcAccepted(event *glightning.HtlcAcceptedEvent)
 
 	interceptResult := intercept(paymentHashBytes, onion.ForwardAmount, uint32(event.Htlc.CltvExpiry))
 	switch interceptResult.action {
-	case INTERCEPT_RESUME_OR_CANCEL:
-		return i.resumeOrCancel(event, interceptResult), nil
+	case INTERCEPT_RESUME_WITH_ONION:
+		return i.resumeWithOnion(event, interceptResult), nil
 	case INTERCEPT_FAIL_HTLC:
 		return event.Fail(uint16(FAILURE_TEMPORARY_CHANNEL_FAILURE)), nil
 	case INTERCEPT_FAIL_HTLC_WITH_CODE:
@@ -113,51 +112,16 @@ func (i *ClnHtlcInterceptor) OnHtlcAccepted(event *glightning.HtlcAcceptedEvent)
 	}
 }
 
-func (i *ClnHtlcInterceptor) resumeOrCancel(event *glightning.HtlcAcceptedEvent, interceptResult interceptResult) *glightning.HtlcAcceptedResponse {
-	deadline := time.Now().Add(60 * time.Second)
-
-	for {
-		chanResult, _ := i.client.GetChannel(interceptResult.destination, *interceptResult.channelPoint)
-		if chanResult != nil {
-			log.Printf("channel opended successfully alias: %v, confirmed: %v", chanResult.InitialChannelID.ToString(), chanResult.ConfirmedChannelID.ToString())
-
-			err := insertChannel(
-				uint64(chanResult.InitialChannelID),
-				uint64(chanResult.ConfirmedChannelID),
-				interceptResult.channelPoint.String(),
-				interceptResult.destination,
-				time.Now(),
-			)
-
-			if err != nil {
-				log.Printf("insertChannel error: %v", err)
-				return event.Fail(uint16(FAILURE_TEMPORARY_CHANNEL_FAILURE))
-			}
-
-			channelID := uint64(chanResult.ConfirmedChannelID)
-			if channelID == 0 {
-				channelID = uint64(chanResult.InitialChannelID)
-			}
-			//decoding and encoding onion with alias in type 6 record.
-			newPayload, err := encodePayloadWithNextHop(event.Onion.Payload, channelID)
-			if err != nil {
-				log.Printf("encodePayloadWithNextHop error: %v", err)
-				return event.Fail(uint16(FAILURE_TEMPORARY_CHANNEL_FAILURE))
-			}
-
-			log.Printf("forwarding htlc to the destination node and a new private channel was opened")
-			return event.ContinueWithPayload(newPayload)
-		}
-
-		log.Printf("waiting for channel to get opened.... %v\n", interceptResult.destination)
-		if time.Now().After(deadline) {
-			log.Printf("Stop retrying getChannel(%v, %v)", interceptResult.destination, interceptResult.channelPoint.String())
-			break
-		}
-		time.Sleep(1 * time.Second)
+func (i *ClnHtlcInterceptor) resumeWithOnion(event *glightning.HtlcAcceptedEvent, interceptResult interceptResult) *glightning.HtlcAcceptedResponse {
+	//decoding and encoding onion with alias in type 6 record.
+	newPayload, err := encodePayloadWithNextHop(event.Onion.Payload, interceptResult.channelId)
+	if err != nil {
+		log.Printf("encodePayloadWithNextHop error: %v", err)
+		return event.Fail(uint16(FAILURE_TEMPORARY_CHANNEL_FAILURE))
 	}
-	log.Printf("Error: Channel failed to opened... timed out. ")
-	return event.Fail(uint16(FAILURE_TEMPORARY_CHANNEL_FAILURE))
+
+	log.Printf("forwarding htlc to the destination node and a new private channel was opened")
+	return event.ContinueWithPayload(newPayload)
 }
 
 func encodePayloadWithNextHop(payloadHex string, channelId uint64) (string, error) {

forwarding_history.go

@@ -5,13 +5,11 @@ import (
 	"encoding/hex"
 	"fmt"
 	"log"
-	"os"
 	"time"
 
 	"github.com/lightningnetwork/lnd/htlcswitch/hop"
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnrpc/chainrpc"
-	"google.golang.org/grpc/metadata"
 )
 
 type copyFromEvents struct {
@@ -42,8 +40,7 @@ func channelsSynchronize(client *LndClient) {
 	lastSync := time.Now().Add(-6 * time.Minute)
 	for {
 		cancellableCtx, cancel := context.WithCancel(context.Background())
-		clientCtx := metadata.AppendToOutgoingContext(cancellableCtx, "macaroon", os.Getenv("LND_MACAROON_HEX"))
-		stream, err := client.chainNotifierClient.RegisterBlockEpochNtfn(clientCtx, &chainrpc.BlockEpoch{})
+		stream, err := client.chainNotifierClient.RegisterBlockEpochNtfn(cancellableCtx, &chainrpc.BlockEpoch{})
 		if err != nil {
 			log.Printf("chainNotifierClient.RegisterBlockEpochNtfn(): %v", err)
 			cancel()
@@ -70,8 +67,7 @@ func channelsSynchronize(client *LndClient) {
 
 func channelsSynchronizeOnce(client *LndClient) error {
 	log.Printf("channelsSynchronizeOnce - begin")
-	clientCtx := metadata.AppendToOutgoingContext(context.Background(), "macaroon", os.Getenv("LND_MACAROON_HEX"))
-	channels, err := client.client.ListChannels(clientCtx, &lnrpc.ListChannelsRequest{PrivateOnly: true})
+	channels, err := client.client.ListChannels(context.Background(), &lnrpc.ListChannelsRequest{PrivateOnly: true})
 	if err != nil {
 		log.Printf("ListChannels error: %v", err)
 		return fmt.Errorf("client.ListChannels() error: %w", err)
@@ -123,10 +119,9 @@ func forwardingHistorySynchronizeOnce(client *LndClient) error {
 	log.Printf("last2: %v", last)
 	now := time.Now()
 	endTime := uint64(now.Add(time.Hour * 24).Unix())
-	clientCtx := metadata.AppendToOutgoingContext(context.Background(), "macaroon", os.Getenv("LND_MACAROON_HEX"))
 	indexOffset := uint32(0)
 	for {
-		forwardHistory, err := client.client.ForwardingHistory(clientCtx, &lnrpc.ForwardingHistoryRequest{
+		forwardHistory, err := client.client.ForwardingHistory(context.Background(), &lnrpc.ForwardingHistoryRequest{
 			StartTime:    uint64(last),
 			EndTime:      endTime,
 			NumMaxEvents: 10000,

intercept.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log"
 	"math/big"
+	"time"
 
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/wire"
@@ -20,7 +21,7 @@ type interceptAction int
 
 const (
 	INTERCEPT_RESUME              interceptAction = 0
-	INTERCEPT_RESUME_OR_CANCEL    interceptAction = 1
+	INTERCEPT_RESUME_WITH_ONION   interceptAction = 1
 	INTERCEPT_FAIL_HTLC           interceptAction = 2
 	INTERCEPT_FAIL_HTLC_WITH_CODE interceptAction = 3
 )
@@ -40,6 +41,7 @@ type interceptResult struct {
 	destination  []byte
 	amountMsat   uint64
 	channelPoint *wire.OutPoint
+	channelId    uint64
 	onionBlob    []byte
 }
 
@@ -53,114 +55,157 @@ func intercept(reqPaymentHash []byte, reqOutgoingAmountMsat uint64, reqOutgoingE
 				action: INTERCEPT_FAIL_HTLC,
 			}, nil
 		}
-		log.Printf("paymentHash:%x\npaymentSecret:%x\ndestination:%x\nincomingAmountMsat:%v\noutgoingAmountMsat:%v\n\n",
+		log.Printf("paymentHash:%x\npaymentSecret:%x\ndestination:%x\nincomingAmountMsat:%v\noutgoingAmountMsat:%v",
 			paymentHash, paymentSecret, destination, incomingAmountMsat, outgoingAmountMsat)
-		if paymentSecret != nil {
-
-			if channelPoint == nil {
-				if bytes.Equal(paymentHash, reqPaymentHash) {
-					channelPoint, err = openChannel(client, reqPaymentHash, destination, incomingAmountMsat)
-					log.Printf("openChannel(%x, %v) err: %v", destination, incomingAmountMsat, err)
-					if err != nil {
-						return interceptResult{
-							action: INTERCEPT_FAIL_HTLC,
-						}, nil
-					}
-				} else { //probing
-					failureCode := FAILURE_TEMPORARY_CHANNEL_FAILURE
-					isConnected, _ := client.IsConnected(destination)
-					if err != nil || !*isConnected {
-						failureCode = FAILURE_INCORRECT_OR_UNKNOWN_PAYMENT_DETAILS
-					}
-
-					return interceptResult{
-						action:      INTERCEPT_FAIL_HTLC_WITH_CODE,
-						failureCode: failureCode,
-					}, nil
-				}
-			}
-
-			pubKey, err := btcec.ParsePubKey(destination)
-			if err != nil {
-				log.Printf("btcec.ParsePubKey(%x): %v", destination, err)
-				return interceptResult{
-					action: INTERCEPT_FAIL_HTLC,
-				}, nil
-			}
-
-			sessionKey, err := btcec.NewPrivateKey()
-			if err != nil {
-				log.Printf("btcec.NewPrivateKey(): %v", err)
-				return interceptResult{
-					action: INTERCEPT_FAIL_HTLC,
-				}, nil
-			}
-
-			var bigProd, bigAmt big.Int
-			amt := (bigAmt.Div(bigProd.Mul(big.NewInt(outgoingAmountMsat), big.NewInt(int64(reqOutgoingAmountMsat))), big.NewInt(incomingAmountMsat))).Int64()
-
-			var addr [32]byte
-			copy(addr[:], paymentSecret)
-			hop := route.Hop{
-				AmtToForward:     lnwire.MilliSatoshi(amt),
-				OutgoingTimeLock: reqOutgoingExpiry,
-				MPP:              record.NewMPP(lnwire.MilliSatoshi(outgoingAmountMsat), addr),
-				CustomRecords:    make(record.CustomSet),
-			}
-
-			var b bytes.Buffer
-			err = hop.PackHopPayload(&b, uint64(0))
-			if err != nil {
-				log.Printf("hop.PackHopPayload(): %v", err)
-				return interceptResult{
-					action: INTERCEPT_FAIL_HTLC,
-				}, nil
-			}
-
-			payload, err := sphinx.NewHopPayload(nil, b.Bytes())
-			if err != nil {
-				log.Printf("sphinx.NewHopPayload(): %v", err)
-				return interceptResult{
-					action: INTERCEPT_FAIL_HTLC,
-				}, nil
-			}
-
-			var sphinxPath sphinx.PaymentPath
-			sphinxPath[0] = sphinx.OnionHop{
-				NodePub:    *pubKey,
-				HopPayload: payload,
-			}
-			sphinxPacket, err := sphinx.NewOnionPacket(
-				&sphinxPath, sessionKey, reqPaymentHash,
-				sphinx.DeterministicPacketFiller,
-			)
-			if err != nil {
-				log.Printf("sphinx.NewOnionPacket(): %v", err)
-				return interceptResult{
-					action: INTERCEPT_FAIL_HTLC,
-				}, nil
-			}
-			var onionBlob bytes.Buffer
-			err = sphinxPacket.Encode(&onionBlob)
-			if err != nil {
-				log.Printf("sphinxPacket.Encode(): %v", err)
-				return interceptResult{
-					action: INTERCEPT_FAIL_HTLC,
-				}, nil
-			}
-
-			return interceptResult{
-				action:       INTERCEPT_RESUME_OR_CANCEL,
-				destination:  destination,
-				channelPoint: channelPoint,
-				amountMsat:   uint64(amt),
-				onionBlob:    onionBlob.Bytes(),
-			}, nil
-		} else {
+		if paymentSecret == nil {
 			return interceptResult{
 				action: INTERCEPT_RESUME,
 			}, nil
 		}
+
+		if channelPoint == nil {
+			if bytes.Equal(paymentHash, reqPaymentHash) {
+				channelPoint, err = openChannel(client, reqPaymentHash, destination, incomingAmountMsat)
+				if err != nil {
+					log.Printf("openChannel(%x, %v) err: %v", destination, incomingAmountMsat, err)
+					return interceptResult{
+						action: INTERCEPT_FAIL_HTLC,
+					}, nil
+				}
+			} else { //probing
+				failureCode := FAILURE_TEMPORARY_CHANNEL_FAILURE
+				isConnected, _ := client.IsConnected(destination)
+				if err != nil || !isConnected {
+					failureCode = FAILURE_INCORRECT_OR_UNKNOWN_PAYMENT_DETAILS
+				}
+
+				return interceptResult{
+					action:      INTERCEPT_FAIL_HTLC_WITH_CODE,
+					failureCode: failureCode,
+				}, nil
+			}
+		}
+
+		pubKey, err := btcec.ParsePubKey(destination)
+		if err != nil {
+			log.Printf("btcec.ParsePubKey(%x): %v", destination, err)
+			return interceptResult{
+				action: INTERCEPT_FAIL_HTLC,
+			}, nil
+		}
+
+		sessionKey, err := btcec.NewPrivateKey()
+		if err != nil {
+			log.Printf("btcec.NewPrivateKey(): %v", err)
+			return interceptResult{
+				action: INTERCEPT_FAIL_HTLC,
+			}, nil
+		}
+
+		var bigProd, bigAmt big.Int
+		amt := (bigAmt.Div(bigProd.Mul(big.NewInt(outgoingAmountMsat), big.NewInt(int64(reqOutgoingAmountMsat))), big.NewInt(incomingAmountMsat))).Int64()
+
+		var addr [32]byte
+		copy(addr[:], paymentSecret)
+		hop := route.Hop{
+			AmtToForward:     lnwire.MilliSatoshi(amt),
+			OutgoingTimeLock: reqOutgoingExpiry,
+			MPP:              record.NewMPP(lnwire.MilliSatoshi(outgoingAmountMsat), addr),
+			CustomRecords:    make(record.CustomSet),
+		}
+
+		var b bytes.Buffer
+		err = hop.PackHopPayload(&b, uint64(0))
+		if err != nil {
+			log.Printf("hop.PackHopPayload(): %v", err)
+			return interceptResult{
+				action: INTERCEPT_FAIL_HTLC,
+			}, nil
+		}
+
+		payload, err := sphinx.NewHopPayload(nil, b.Bytes())
+		if err != nil {
+			log.Printf("sphinx.NewHopPayload(): %v", err)
+			return interceptResult{
+				action: INTERCEPT_FAIL_HTLC,
+			}, nil
+		}
+
+		var sphinxPath sphinx.PaymentPath
+		sphinxPath[0] = sphinx.OnionHop{
+			NodePub:    *pubKey,
+			HopPayload: payload,
+		}
+		sphinxPacket, err := sphinx.NewOnionPacket(
+			&sphinxPath, sessionKey, reqPaymentHash,
+			sphinx.DeterministicPacketFiller,
+		)
+		if err != nil {
+			log.Printf("sphinx.NewOnionPacket(): %v", err)
+			return interceptResult{
+				action: INTERCEPT_FAIL_HTLC,
+			}, nil
+		}
+		var onionBlob bytes.Buffer
+		err = sphinxPacket.Encode(&onionBlob)
+		if err != nil {
+			log.Printf("sphinxPacket.Encode(): %v", err)
+			return interceptResult{
+				action: INTERCEPT_FAIL_HTLC,
+			}, nil
+		}
+
+		deadline := time.Now().Add(60 * time.Second)
+
+		for {
+			chanResult, _ := client.GetChannel(destination, *channelPoint)
+			if chanResult != nil {
+				log.Printf("channel opended successfully alias: %v, confirmed: %v", chanResult.InitialChannelID.ToString(), chanResult.ConfirmedChannelID.ToString())
+
+				err := insertChannel(
+					uint64(chanResult.InitialChannelID),
+					uint64(chanResult.ConfirmedChannelID),
+					channelPoint.String(),
+					destination,
+					time.Now(),
+				)
+
+				if err != nil {
+					log.Printf("insertChannel error: %v", err)
+					return interceptResult{
+						action:      INTERCEPT_FAIL_HTLC_WITH_CODE,
+						failureCode: FAILURE_TEMPORARY_CHANNEL_FAILURE,
+					}, nil
+				}
+
+				channelID := uint64(chanResult.ConfirmedChannelID)
+				if channelID == 0 {
+					channelID = uint64(chanResult.InitialChannelID)
+				}
+
+				return interceptResult{
+					action:       INTERCEPT_RESUME_WITH_ONION,
+					destination:  destination,
+					channelPoint: channelPoint,
+					channelId:    channelID,
+					amountMsat:   uint64(amt),
+					onionBlob:    onionBlob.Bytes(),
+				}, nil
+			}
+
+			log.Printf("waiting for channel to get opened.... %v\n", destination)
+			if time.Now().After(deadline) {
+				log.Printf("Stop retrying getChannel(%v, %v)", destination, channelPoint.String())
+				break
+			}
+			time.Sleep(1 * time.Second)
+		}
+
+		log.Printf("Error: Channel failed to opened... timed out. ")
+		return interceptResult{
+			action:      INTERCEPT_FAIL_HTLC_WITH_CODE,
+			failureCode: FAILURE_TEMPORARY_CHANNEL_FAILURE,
+		}, nil
 	})
 
 	return resp.(interceptResult)

lightning_client.go

@@ -23,7 +23,7 @@ type OpenChannelRequest struct {
 
 type LightningClient interface {
 	GetInfo() (*GetInfoResult, error)
-	IsConnected(destination []byte) (*bool, error)
+	IsConnected(destination []byte) (bool, error)
 	OpenChannel(req *OpenChannelRequest) (*wire.OutPoint, error)
 	GetChannel(peerID []byte, channelPoint wire.OutPoint) (*GetChannelResult, error)
 	GetNodeChannelCount(nodeID []byte) (int, error)

lnd_client.go

@@ -33,9 +33,14 @@ func NewLndClient() *LndClient {
 		log.Fatalf("credentials: failed to append certificates")
 	}
 	creds := credentials.NewClientTLSFromCert(cp, "")
+	macCred := NewMacaroonCredential(os.Getenv("LND_MACAROON_HEX"))
 
 	// Address of an LND instance
-	conn, err := grpc.Dial(os.Getenv("LND_ADDRESS"), grpc.WithTransportCredentials(creds))
+	conn, err := grpc.Dial(
+		os.Getenv("LND_ADDRESS"),
+		grpc.WithTransportCredentials(creds),
+		grpc.WithPerRPCCredentials(macCred),
+	)
 	if err != nil {
 		log.Fatalf("Failed to connect to LND gRPC: %v", err)
 	}
@@ -56,7 +61,7 @@ func (c *LndClient) Close() {
 }
 
 func (c *LndClient) GetInfo() (*GetInfoResult, error) {
-	info, err := c.client.GetInfo(macaroonContext(), &lnrpc.GetInfoRequest{})
+	info, err := c.client.GetInfo(context.Background(), &lnrpc.GetInfoRequest{})
 	if err != nil {
 		log.Printf("LND: client.GetInfo() error: %v", err)
 		return nil, err
@@ -68,29 +73,27 @@ func (c *LndClient) GetInfo() (*GetInfoResult, error) {
 	}, nil
 }
 
-func (c *LndClient) IsConnected(destination []byte) (*bool, error) {
+func (c *LndClient) IsConnected(destination []byte) (bool, error) {
 	pubKey := hex.EncodeToString(destination)
 
-	r, err := c.client.ListPeers(macaroonContext(), &lnrpc.ListPeersRequest{LatestError: true})
+	r, err := c.client.ListPeers(context.Background(), &lnrpc.ListPeersRequest{LatestError: true})
 	if err != nil {
 		log.Printf("LND: client.ListPeers() error: %v", err)
-		return nil, fmt.Errorf("LND: client.ListPeers() error: %w", err)
+		return false, fmt.Errorf("LND: client.ListPeers() error: %w", err)
 	}
 	for _, peer := range r.Peers {
 		if pubKey == peer.PubKey {
 			log.Printf("destination online: %x", destination)
-			result := true
-			return &result, nil
+			return true, nil
 		}
 	}
 
 	log.Printf("LND: destination offline: %x", destination)
-	result := false
-	return &result, nil
+	return false, nil
 }
 
 func (c *LndClient) OpenChannel(req *OpenChannelRequest) (*wire.OutPoint, error) {
-	channelPoint, err := c.client.OpenChannelSync(macaroonContext(), &lnrpc.OpenChannelRequest{
+	channelPoint, err := c.client.OpenChannelSync(context.Background(), &lnrpc.OpenChannelRequest{
 		NodePubkey:         req.Destination,
 		LocalFundingAmount: int64(req.CapacitySat),
 		TargetConf:         int32(req.TargetConf),
@@ -115,7 +118,7 @@ func (c *LndClient) OpenChannel(req *OpenChannelRequest) (*wire.OutPoint, error)
 }
 
 func (c *LndClient) GetChannel(peerID []byte, channelPoint wire.OutPoint) (*GetChannelResult, error) {
-	r, err := c.client.ListChannels(macaroonContext(), &lnrpc.ListChannelsRequest{Peer: peerID})
+	r, err := c.client.ListChannels(context.Background(), &lnrpc.ListChannelsRequest{Peer: peerID})
 	if err != nil {
 		log.Printf("client.ListChannels(%x) error: %v", peerID, err)
 		return nil, err
@@ -197,7 +200,7 @@ func (c *LndClient) GetClosedChannels(nodeID string, channelPoints map[string]ui
 }
 
 func (c *LndClient) getWaitingCloseChannels(nodeID string) ([]*lnrpc.PendingChannelsResponse_WaitingCloseChannel, error) {
-	pendingResponse, err := c.client.PendingChannels(macaroonContext(), &lnrpc.PendingChannelsRequest{})
+	pendingResponse, err := c.client.PendingChannels(context.Background(), &lnrpc.PendingChannelsRequest{})
 	if err != nil {
 		return nil, err
 	}
@@ -209,7 +212,3 @@ func (c *LndClient) getWaitingCloseChannels(nodeID string) ([]*lnrpc.PendingChan
 	}
 	return waitingCloseChannels, nil
 }
-
-func macaroonContext() context.Context {
-	return metadata.AppendToOutgoingContext(context.Background(), "macaroon", os.Getenv("LND_MACAROON_HEX"))
-}

lnd_interceptor.go

@@ -4,14 +4,12 @@ import (
 	"context"
 	"fmt"
 	"log"
-	"os"
 	"sync"
 	"time"
 
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnrpc/routerrpc"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 )
 
@@ -19,6 +17,7 @@ type LndHtlcInterceptor struct {
 	client        *LndClient
 	stopRequested bool
 	initWg        sync.WaitGroup
+	doneWg        sync.WaitGroup
 }
 
 func NewLndHtlcInterceptor() *LndHtlcInterceptor {
@@ -49,14 +48,18 @@ func (i *LndHtlcInterceptor) WaitStarted() LightningClient {
 }
 
 func (i *LndHtlcInterceptor) intercept() error {
+	defer func() {
+		log.Printf("LND intercept(): stopping. Waiting for in-progress interceptions to complete.")
+		i.doneWg.Wait()
+	}()
+
 	for {
 		if i.stopRequested {
 			return nil
 		}
 
 		cancellableCtx, cancel := context.WithCancel(context.Background())
-		clientCtx := metadata.AppendToOutgoingContext(cancellableCtx, "macaroon", os.Getenv("LND_MACAROON_HEX"))
-		interceptorClient, err := i.client.routerClient.HtlcInterceptor(clientCtx)
+		interceptorClient, err := i.client.routerClient.HtlcInterceptor(cancellableCtx)
 		if err != nil {
 			log.Printf("routerClient.HtlcInterceptor(): %v", err)
 			cancel()
@@ -93,30 +96,40 @@ func (i *LndHtlcInterceptor) intercept() error {
 				request.OnionBlob,
 			)
 
-			interceptResult := intercept(request.PaymentHash, request.OutgoingAmountMsat, request.OutgoingExpiry)
-			switch interceptResult.action {
-			case INTERCEPT_RESUME_OR_CANCEL:
-				go resumeOrCancel(clientCtx, interceptorClient, request.IncomingCircuitKey,
-					interceptResult)
-			case INTERCEPT_FAIL_HTLC:
-				failForwardSend(interceptorClient, request.IncomingCircuitKey)
-			case INTERCEPT_FAIL_HTLC_WITH_CODE:
-				interceptorClient.Send(&routerrpc.ForwardHtlcInterceptResponse{
-					IncomingCircuitKey: request.IncomingCircuitKey,
-					Action:             routerrpc.ResolveHoldForwardAction_FAIL,
-					FailureCode:        mapFailureCode(interceptResult.failureCode),
-				})
-			case INTERCEPT_RESUME:
-				fallthrough
-			default:
-				interceptorClient.Send(&routerrpc.ForwardHtlcInterceptResponse{
-					IncomingCircuitKey:      request.IncomingCircuitKey,
-					Action:                  routerrpc.ResolveHoldForwardAction_RESUME,
-					OutgoingAmountMsat:      request.OutgoingAmountMsat,
-					OutgoingRequestedChanId: request.OutgoingRequestedChanId,
-					OnionBlob:               request.OnionBlob,
-				})
-			}
+			i.doneWg.Add(1)
+			go func() {
+				interceptResult := intercept(request.PaymentHash, request.OutgoingAmountMsat, request.OutgoingExpiry)
+				switch interceptResult.action {
+				case INTERCEPT_RESUME_WITH_ONION:
+					interceptorClient.Send(&routerrpc.ForwardHtlcInterceptResponse{
+						IncomingCircuitKey:      request.IncomingCircuitKey,
+						Action:                  routerrpc.ResolveHoldForwardAction_RESUME,
+						OutgoingAmountMsat:      interceptResult.amountMsat,
+						OutgoingRequestedChanId: uint64(interceptResult.channelId),
+						OnionBlob:               interceptResult.onionBlob,
+					})
+				case INTERCEPT_FAIL_HTLC:
+					failForwardSend(interceptorClient, request.IncomingCircuitKey)
+				case INTERCEPT_FAIL_HTLC_WITH_CODE:
+					interceptorClient.Send(&routerrpc.ForwardHtlcInterceptResponse{
+						IncomingCircuitKey: request.IncomingCircuitKey,
+						Action:             routerrpc.ResolveHoldForwardAction_FAIL,
+						FailureCode:        mapFailureCode(interceptResult.failureCode),
+					})
+				case INTERCEPT_RESUME:
+					fallthrough
+				default:
+					interceptorClient.Send(&routerrpc.ForwardHtlcInterceptResponse{
+						IncomingCircuitKey:      request.IncomingCircuitKey,
+						Action:                  routerrpc.ResolveHoldForwardAction_RESUME,
+						OutgoingAmountMsat:      request.OutgoingAmountMsat,
+						OutgoingRequestedChanId: request.OutgoingRequestedChanId,
+						OnionBlob:               request.OnionBlob,
+					})
+				}
+
+				i.doneWg.Done()
+			}()
 		}
 
 		cancel()
@@ -140,41 +153,3 @@ func failForwardSend(interceptorClient routerrpc.Router_HtlcInterceptorClient, i
 		Action:             routerrpc.ResolveHoldForwardAction_FAIL,
 	})
 }
-
-func resumeOrCancel(
-	ctx context.Context,
-	interceptorClient routerrpc.Router_HtlcInterceptorClient,
-	incomingCircuitKey *routerrpc.CircuitKey,
-	interceptResult interceptResult,
-) {
-	deadline := time.Now().Add(10 * time.Second)
-	for {
-		ch, err := client.GetChannel(interceptResult.destination, *interceptResult.channelPoint)
-		if err != nil {
-			failForwardSend(interceptorClient, incomingCircuitKey)
-			return
-		}
-
-		if uint64(ch.InitialChannelID) != 0 {
-			interceptorClient.Send(&routerrpc.ForwardHtlcInterceptResponse{
-				IncomingCircuitKey:      incomingCircuitKey,
-				Action:                  routerrpc.ResolveHoldForwardAction_RESUME,
-				OutgoingAmountMsat:      interceptResult.amountMsat,
-				OutgoingRequestedChanId: uint64(ch.InitialChannelID),
-				OnionBlob:               interceptResult.onionBlob,
-			})
-			err := insertChannel(uint64(ch.InitialChannelID), uint64(ch.ConfirmedChannelID), interceptResult.channelPoint.String(), interceptResult.destination, time.Now())
-			if err != nil {
-				log.Printf("insertChannel error: %v", err)
-			}
-			return
-		}
-		log.Printf("getChannel(%x, %v) returns 0", interceptResult.destination, interceptResult.channelPoint.String())
-		if time.Now().After(deadline) {
-			log.Printf("Stop retrying getChannel(%x, %v)", interceptResult.destination, interceptResult.channelPoint.String())
-			break
-		}
-		time.Sleep(1 * time.Second)
-	}
-	failForwardSend(interceptorClient, incomingCircuitKey)
-}

lnd_macaroon_credential.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"context"
+)
+
+type MacaroonCredential struct {
+	MacaroonHex string
+}
+
+func NewMacaroonCredential(hex string) *MacaroonCredential {
+	return &MacaroonCredential{
+		MacaroonHex: hex,
+	}
+}
+
+func (m *MacaroonCredential) RequireTransportSecurity() bool {
+	return true
+}
+
+func (m *MacaroonCredential) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
+	md := make(map[string]string)
+	md["macaroon"] = m.MacaroonHex
+	return md, nil
+}