more efficient cleanup, lessons learned

forwarding_history.go

@@ -36,19 +36,25 @@ func (cfe *copyFromEvents) Err() error {
 	return cfe.err
 }
 
-func channelsSynchronize(client *LndClient) {
+func channelsSynchronize(ctx context.Context, client *LndClient) {
 	lastSync := time.Now().Add(-6 * time.Minute)
 	for {
-		cancellableCtx, cancel := context.WithCancel(context.Background())
+		if ctx.Err() != nil {
-		stream, err := client.chainNotifierClient.RegisterBlockEpochNtfn(cancellableCtx, &chainrpc.BlockEpoch{})
+			return
+		}
+
+		stream, err := client.chainNotifierClient.RegisterBlockEpochNtfn(ctx, &chainrpc.BlockEpoch{})
 		if err != nil {
 			log.Printf("chainNotifierClient.RegisterBlockEpochNtfn(): %v", err)
-			cancel()
 			<-time.After(time.Second)
 			continue
 		}
 
 		for {
+			if ctx.Err() != nil {
+				return
+			}
+
 			_, err := stream.Recv()
 			if err != nil {
 				log.Printf("stream.Recv: %v", err)
@@ -56,13 +62,16 @@ func channelsSynchronize(client *LndClient) {
 				break
 			}
 			if lastSync.Add(5 * time.Minute).Before(time.Now()) {
-				<-time.After(30 * time.Second)
+				select {
+				case <-ctx.Done():
+					return
+				case <-time.After(1 * time.Minute):
+				}
 				err = channelsSynchronizeOnce(client)
 				lastSync = time.Now()
 				log.Printf("channelsSynchronizeOnce() err: %v", err)
 			}
 		}
-		cancel()
 	}
 }
 
@@ -99,11 +108,18 @@ func channelsSynchronizeOnce(client *LndClient) error {
 	return nil
 }
 
-func forwardingHistorySynchronize(client *LndClient) {
+func forwardingHistorySynchronize(ctx context.Context, client *LndClient) {
 	for {
+		if ctx.Err() != nil {
+			return
+		}
+
 		err := forwardingHistorySynchronizeOnce(client)
 		log.Printf("forwardingHistorySynchronizeOnce() err: %v", err)
-		<-time.After(1 * time.Minute)
+		select {
+		case <-time.After(1 * time.Minute):
+		case <-ctx.Done():
+		}
 	}
 }
 

itest/lnd_lspd_node.go

@@ -6,7 +6,6 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
-	"runtime"
 	"strings"
 	"sync"
 
@@ -153,12 +152,20 @@ func (c *LndLspNode) Start() {
 		Name: fmt.Sprintf("%s: cmd", c.lspBase.name),
 		Fn: func() error {
 			proc := cmd.Process
-			if proc != nil {
+			if proc == nil {
-				if runtime.GOOS == "windows" {
+				return nil
-					return proc.Signal(os.Kill)
 			}
 
-				return proc.Signal(os.Interrupt)
+			proc.Kill()
+
+			log.Printf("About to wait for lspd to exit")
+			status, err := proc.Wait()
+			if err != nil {
+				log.Printf("waiting for lspd process error: %v, status: %v", err, status)
+			}
+			err = cmd.Wait()
+			if err != nil {
+				log.Printf("waiting for lspd cmd error: %v", err)
 			}
 
 			return nil

lnd_interceptor.go

@@ -15,9 +15,10 @@ import (
 
 type LndHtlcInterceptor struct {
 	client *LndClient
-	stopRequested bool
 	initWg sync.WaitGroup
 	doneWg sync.WaitGroup
+	ctx    context.Context
+	cancel context.CancelFunc
 }
 
 func NewLndHtlcInterceptor() *LndHtlcInterceptor {
@@ -31,14 +32,18 @@ func NewLndHtlcInterceptor() *LndHtlcInterceptor {
 }
 
 func (i *LndHtlcInterceptor) Start() error {
-	go forwardingHistorySynchronize(i.client)
+	ctx, cancel := context.WithCancel(context.Background())
-	go channelsSynchronize(i.client)
+	i.ctx = ctx
-	i.initWg.Done()
+	i.cancel = cancel
+	go forwardingHistorySynchronize(ctx, i.client)
+	go channelsSynchronize(ctx, i.client)
+
 	return i.intercept()
 }
 
 func (i *LndHtlcInterceptor) Stop() error {
-	i.stopRequested = true
+	i.cancel()
+	i.doneWg.Wait()
 	return nil
 }
 
@@ -48,44 +53,53 @@ func (i *LndHtlcInterceptor) WaitStarted() LightningClient {
 }
 
 func (i *LndHtlcInterceptor) intercept() error {
+	inited := false
 	defer func() {
+		if !inited {
+			i.initWg.Done()
+		}
 		log.Printf("LND intercept(): stopping. Waiting for in-progress interceptions to complete.")
 		i.doneWg.Wait()
 	}()
 
 	for {
-		if i.stopRequested {
+		if i.ctx.Err() != nil {
-			return nil
+			return i.ctx.Err()
 		}
 
 		log.Printf("Connecting LND HTLC interceptor.")
-		cancellableCtx, cancel := context.WithCancel(context.Background())
+		interceptorClient, err := i.client.routerClient.HtlcInterceptor(i.ctx)
-		interceptorClient, err := i.client.routerClient.HtlcInterceptor(cancellableCtx)
 		if err != nil {
 			log.Printf("routerClient.HtlcInterceptor(): %v", err)
-			cancel()
 			<-time.After(time.Second)
 			continue
 		}
 
 		for {
-			if i.stopRequested {
+			if i.ctx.Err() != nil {
-				cancel()
+				return i.ctx.Err()
-				return nil
 			}
+
+			if !inited {
+				inited = true
+				i.initWg.Done()
+			}
+
 			request, err := interceptorClient.Recv()
 			if err != nil {
 				// If it is  just the error result of the context cancellation
 				// the we exit silently.
 				status, ok := status.FromError(err)
 				if ok && status.Code() == codes.Canceled {
+					log.Printf("Got code canceled. Break.")
 					break
 				}
+
 				// Otherwise it an unexpected error, we fail the test.
 				log.Printf("unexpected error in interceptor.Recv() %v", err)
-				cancel()
 				break
 			}
+
 			fmt.Printf("htlc: %v\nchanID: %v\nincoming amount: %v\noutgoing amount: %v\nincomin expiry: %v\noutgoing expiry: %v\npaymentHash: %x\nonionBlob: %x\n\n",
 				request.IncomingCircuitKey.HtlcId,
 				request.IncomingCircuitKey.ChanId,
@@ -131,7 +145,6 @@ func (i *LndHtlcInterceptor) intercept() error {
 			}()
 		}
 
-		cancel()
 		<-time.After(time.Second)
 	}
 }

main.go

@@ -4,7 +4,9 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"os/signal"
 	"sync"
+	"syscall"
 
 	"github.com/btcsuite/btcd/btcec/v2"
 )
@@ -86,6 +88,15 @@ func main() {
 		wg.Done()
 	}()
 
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+	go func() {
+		sig := <-c
+		log.Printf("Received stop signal %v. Stopping.", sig)
+		s.Stop()
+		interceptor.Stop()
+	}()
+
 	wg.Wait()
 	log.Printf("lspd exited")
 }