peersync: replace poll service with new package

This change prepares support for 2-hop peerswap.
It removes the old poll service.
It adds a new peersync package.
Both the CLN and LND plugins start one shared PeerSync instance.
The external CLI and RPC interfaces stay the same.
The existing swap and premium logic also stay the same.
No data migration is required.
Version compatibility requirements do not change.

The refactor makes the peer sync logic easier to understand and test.
The peersync package is split into smaller parts with unit tests.
CLN and LND now share the same code path.

PeerSync can resend updates at once on
premium changes or policy reloads.
This reduces configuration delay and makes
 operations and monitoring easier.

Author: YusukeShimizu

clightning/clightning.go

@@ -10,18 +10,16 @@ import (
 	"os"
 	"time"
 
-	"github.com/elementsproject/peerswap/log"
-	"github.com/elementsproject/peerswap/premium"
-
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/elementsproject/glightning/gbitcoin"
-	"github.com/elementsproject/peerswap/onchain"
-
 	"github.com/elementsproject/glightning/glightning"
 	"github.com/elementsproject/glightning/jrpc2"
 	"github.com/elementsproject/peerswap/lightning"
+	"github.com/elementsproject/peerswap/log"
 	"github.com/elementsproject/peerswap/messages"
-	"github.com/elementsproject/peerswap/poll"
+	"github.com/elementsproject/peerswap/onchain"
+	"github.com/elementsproject/peerswap/peersync"
+	"github.com/elementsproject/peerswap/premium"
 	"github.com/elementsproject/peerswap/swap"
 	"github.com/elementsproject/peerswap/wallet"
 	goerrors "github.com/go-errors/errors"
@@ -75,7 +73,8 @@ type ClightningClient struct {
 	swaps          *swap.SwapService
 	requestedSwaps *swap.RequestedSwapsPrinter
 	policy         PolicyReloader
-	pollService    *poll.Service
+	peerSync       *peersync.PeerSync
+	peerStore      *peersync.Store
 	ps             *premium.Setting
 
 	gbitcoin       *gbitcoin.Bitcoin
@@ -326,14 +325,15 @@ func (cl *ClightningClient) GetPreimage() (lightning.Preimage, error) {
 func (cl *ClightningClient) SetupClients(liquidWallet wallet.Wallet,
 	swaps *swap.SwapService,
 	policy PolicyReloader, requestedSwaps *swap.RequestedSwapsPrinter,
-	bitcoin *gbitcoin.Bitcoin, bitcoinChain *onchain.BitcoinOnChain, pollService *poll.Service,
+	bitcoin *gbitcoin.Bitcoin, bitcoinChain *onchain.BitcoinOnChain, peerSync *peersync.PeerSync, peerStore *peersync.Store,
 	ps *premium.Setting) {
 	cl.liquidWallet = liquidWallet
 	cl.requestedSwaps = requestedSwaps
 	cl.swaps = swaps
 	cl.policy = policy
 	cl.gbitcoin = bitcoin
-	cl.pollService = pollService
+	cl.peerSync = peerSync
+	cl.peerStore = peerStore
 	cl.bitcoinChain = bitcoinChain
 	cl.ps = ps
 	if cl.bitcoinChain != nil {
@@ -519,16 +519,6 @@ func (cl *ClightningClient) isPeerConnected(nodeId string) bool {
 	return peer.Connected
 }
 
-// peerRunsPeerSwap returns true if the peer with peerId is listed in the
-// pollService.
-func (cl *ClightningClient) peerRunsPeerSwap(peerId string) bool {
-	pollInfo, err := cl.pollService.GetPollFrom(peerId)
-	if err == nil && pollInfo != nil {
-		return true
-	}
-	return false
-}
-
 // This is called after the Plugin starts up successfully
 func (cl *ClightningClient) onInit(plugin *glightning.Plugin, options map[string]glightning.Option, config *glightning.Config) {
 	cl.glightning.StartUp(config.RpcFile, config.LightningDir)
@@ -549,10 +539,20 @@ func (cl *ClightningClient) OnConnect(connectEvent *glightning.ConnectEvent) {
 	go func() {
 		for {
 			time.Sleep(10 * time.Second)
-			if cl.pollService != nil {
-				cl.pollService.RequestPoll(
-					lo.Ternary(connectEvent.PeerId != "",
-						connectEvent.PeerId, connectEvent.Conn.PeerId))
+			if cl.peerSync != nil {
+				peerIDValue := lo.Ternary(connectEvent.PeerId != "",
+					connectEvent.PeerId, connectEvent.Conn.PeerId)
+				if peerIDValue == "" {
+					return
+				}
+				peerID, err := peersync.NewPeerID(peerIDValue)
+				if err != nil {
+					log.Debugf("invalid peer id for request poll: %v", err)
+					return
+				}
+				if err := cl.peerSync.RequestPoll(cl.ctx, peerID); err != nil {
+					log.Debugf("failed to request poll for %s: %v", peerIDValue, err)
+				}
 				return
 			}
 		}

clightning/clightning_commands.go

@@ -1,6 +1,7 @@
 package clightning
 
 import (
+	"context"
 	"crypto/rand"
 	"encoding/hex"
 	"errors"
@@ -12,6 +13,8 @@ import (
 
 	"github.com/elementsproject/peerswap/log"
 	"github.com/elementsproject/peerswap/peerswaprpc"
+	"github.com/elementsproject/peerswap/peersync"
+	"github.com/elementsproject/peerswap/peersync/format"
 	"github.com/elementsproject/peerswap/premium"
 
 	"github.com/elementsproject/glightning/glightning"
@@ -225,8 +228,10 @@ func (l *SwapOut) Call() (jrpc2.Result, error) {
 	}
 
 	// Skip this check when `force` is set.
-	if !l.Force && !l.cl.peerRunsPeerSwap(fundingChannels.Id) {
-		return nil, fmt.Errorf("peer does not run peerswap")
+	if !l.Force {
+		if l.cl.peerSync == nil || !l.cl.peerSync.HasCompatiblePeer(fundingChannels.Id) {
+			return nil, fmt.Errorf("peer does not run peerswap")
+		}
 	}
 
 	if !l.cl.isPeerConnected(fundingChannels.Id) {
@@ -345,8 +350,10 @@ func (l *SwapIn) Call() (jrpc2.Result, error) {
 	}
 
 	// Skip this check when `force` is set.
-	if !l.Force && !l.cl.peerRunsPeerSwap(fundingChannels.Id) {
-		return nil, fmt.Errorf("peer does not run peerswap")
+	if !l.Force {
+		if l.cl.peerSync == nil || !l.cl.peerSync.HasCompatiblePeer(fundingChannels.Id) {
+			return nil, fmt.Errorf("peer does not run peerswap")
+		}
 	}
 
 	if !l.cl.isPeerConnected(fundingChannels.Id) {
@@ -559,91 +566,26 @@ func (l *ListPeers) Call() (jrpc2.Result, error) {
 		fundingChannels[channel.ShortChannelId] = channel
 	}
 
-	// get polls
-	polls, err := l.cl.pollService.GetCompatiblePolls()
-	if err != nil {
-		return nil, err
+	compatible := make(map[string]*peersync.Peer)
+	if l.cl.peerSync != nil {
+		compatible, err = l.cl.peerSync.CompatiblePeers()
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	peerSwappers := []*peerswaprpc.PeerSwapPeer{}
 	for _, peer := range peers {
-		if p, ok := polls[peer.Id]; ok {
+		peerState, ok := compatible[peer.Id]
+		if ok {
+			capability := peerState.Capability()
 			swaps, err := l.cl.swaps.ListSwapsByPeer(peer.Id)
 			if err != nil {
 				return nil, err
 			}
 
-			var paidFees uint64
-			var ReceiverSwapsOut, ReceiverSwapsIn, ReceiverSatsOut, ReceiverSatsIn uint64
-			var SenderSwapsOut, SenderSwapsIn, SenderSatsOut, SenderSatsIn uint64
-			for _, s := range swaps {
-				// We only list successful swaps. They all end in an
-				// State_ClaimedPreimage state.
-				if s.Current == swap.State_ClaimedPreimage {
-					if s.Role == swap.SWAPROLE_SENDER {
-						paidFees += s.Data.OpeningTxFee
-						if s.Type == swap.SWAPTYPE_OUT {
-							SenderSwapsOut++
-							SenderSatsOut += s.Data.GetAmount()
-						} else {
-							SenderSwapsIn++
-							SenderSatsIn += s.Data.GetAmount()
-						}
-					} else {
-						if s.Type == swap.SWAPTYPE_OUT {
-							ReceiverSwapsOut++
-							ReceiverSatsOut += s.Data.GetAmount()
-						} else {
-							ReceiverSwapsIn++
-							ReceiverSatsIn += s.Data.GetAmount()
-						}
-					}
-				}
-			}
-
-			peerSwapPeer := &peerswaprpc.PeerSwapPeer{
-				NodeId:          peer.Id,
-				SwapsAllowed:    p.PeerAllowed,
-				SupportedAssets: p.Assets,
-				AsSender: &peerswaprpc.SwapStats{
-					SwapsOut: SenderSwapsOut,
-					SwapsIn:  SenderSwapsIn,
-					SatsOut:  SenderSatsOut,
-					SatsIn:   SenderSatsIn,
-				},
-				AsReceiver: &peerswaprpc.SwapStats{
-					SwapsOut: ReceiverSwapsOut,
-					SwapsIn:  ReceiverSwapsIn,
-					SatsOut:  ReceiverSatsOut,
-					SatsIn:   ReceiverSatsIn,
-				},
-				PaidFee: paidFees,
-				PeerPremium: &peerswaprpc.PeerPremium{
-					NodeId: peer.Id,
-					Rates: []*peerswaprpc.PremiumRate{
-						{
-							Asset:          peerswaprpc.AssetType_BTC,
-							Operation:      peerswaprpc.OperationType_SWAP_IN,
-							PremiumRatePpm: p.BTCSwapInPremiumRatePPM,
-						},
-						{
-							Asset:          peerswaprpc.AssetType_BTC,
-							Operation:      peerswaprpc.OperationType_SWAP_OUT,
-							PremiumRatePpm: p.BTCSwapOutPremiumRatePPM,
-						},
-						{
-							Asset:          peerswaprpc.AssetType_LBTC,
-							Operation:      peerswaprpc.OperationType_SWAP_IN,
-							PremiumRatePpm: p.LBTCSwapInPremiumRatePPM,
-						},
-						{
-							Asset:          peerswaprpc.AssetType_LBTC,
-							Operation:      peerswaprpc.OperationType_SWAP_OUT,
-							PremiumRatePpm: p.LBTCSwapOutPremiumRatePPM,
-						},
-					},
-				},
-			}
+			view := format.BuildPeerView(peer.Id, capability, swaps)
+			peerSwapPeer := peerswaprpc.NewPeerSwapPeerFromView(view)
 			channels, err := l.cl.glightning.ListChannelsBySource(peer.Id)
 			if err != nil {
 				return nil, err
@@ -754,8 +696,10 @@ func (c ReloadPolicyFile) Call() (jrpc2.Result, error) {
 	if err != nil {
 		return nil, err
 	}
-	// Resend poll
-	c.cl.pollService.PollAllPeers()
+	// Resend capability updates
+	if c.cl.peerSync != nil {
+		c.cl.peerSync.ForcePollAllPeers(context.Background())
+	}
 	return c.cl.policy.Get(), nil
 }
 
@@ -1217,10 +1161,13 @@ func (c *UpdatePremiumRate) Call() (jrpc2.Result, error) {
 	if err != nil {
 		return nil, fmt.Errorf("error creating premium rate: %v", err)
 	}
-	err = c.cl.ps.SetRate(c.PeerID, rate)
+	err = c.cl.ps.SetRate(context.Background(), c.PeerID, rate)
 	if err != nil {
 		return nil, fmt.Errorf("error setting premium rate: %v", err)
 	}
+	if c.cl.peerSync != nil {
+		c.cl.peerSync.ForcePollAllPeers(context.Background())
+	}
 	return &peerswaprpc.PremiumRate{
 		Asset:          peerswaprpc.ToAssetType(rate.Asset()),
 		Operation:      peerswaprpc.ToOperationType(rate.Operation()),
@@ -1263,11 +1210,14 @@ func (c *DeletePremiumRate) Call() (jrpc2.Result, error) {
 	if !c.cl.isReady {
 		return nil, ErrWaitingForReady
 	}
-	err := c.cl.ps.DeleteRate(c.PeerID, toPremiumAssetType(c.Asset),
+	err := c.cl.ps.DeleteRate(context.Background(), c.PeerID, toPremiumAssetType(c.Asset),
 		toPremiumOperationType(c.Operation))
 	if err != nil {
 		return nil, fmt.Errorf("error deleting premium rate: %v", err)
 	}
+	if c.cl.peerSync != nil {
+		c.cl.peerSync.ForcePollAllPeers(context.Background())
+	}
 	return &peerswaprpc.PremiumRate{
 		Asset:          peerswaprpc.ToAssetType(toPremiumAssetType(c.Asset)),
 		Operation:      peerswaprpc.ToOperationType(toPremiumOperationType(c.Operation)),
@@ -1315,10 +1265,13 @@ func (c *UpdateGlobalPremiumRate) Call() (jrpc2.Result, error) {
 	if err != nil {
 		return nil, fmt.Errorf("error creating premium rate: %v", err)
 	}
-	err = c.cl.ps.SetDefaultRate(rate)
+	err = c.cl.ps.SetDefaultRate(context.Background(), rate)
 	if err != nil {
 		return nil, fmt.Errorf("error setting default premium rate: %v", err)
 	}
+	if c.cl.peerSync != nil {
+		c.cl.peerSync.ForcePollAllPeers(context.Background())
+	}
 	return &peerswaprpc.PremiumRate{
 		Asset:          peerswaprpc.ToAssetType(rate.Asset()),
 		Operation:      peerswaprpc.ToOperationType(rate.Operation()),

cmd/peerswap-plugin/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/elementsproject/peerswap/log"
 	"github.com/elementsproject/peerswap/lwk"
 	"github.com/elementsproject/peerswap/version"
+	"github.com/thejerf/suture/v4"
 	"golang.org/x/sys/unix"
 
 	"github.com/vulpemventures/go-elements/network"
@@ -26,8 +27,8 @@ import (
 	"github.com/elementsproject/peerswap/clightning"
 	"github.com/elementsproject/peerswap/messages"
 	"github.com/elementsproject/peerswap/onchain"
+	"github.com/elementsproject/peerswap/peersync"
 	"github.com/elementsproject/peerswap/policy"
-	"github.com/elementsproject/peerswap/poll"
 	"github.com/elementsproject/peerswap/premium"
 	"github.com/elementsproject/peerswap/swap"
 	"github.com/elementsproject/peerswap/txwatcher"
@@ -372,16 +373,54 @@ func run(ctx context.Context, lightningPlugin *clightning.ClightningClient) erro
 		return err
 	}
 
-	pollStore, err := poll.NewStore(swapDb)
+	peerSyncDBPath := filepath.Join(config.PeerswapDir, "peersync.db")
+	peerStore, err := peersync.NewStore(peerSyncDBPath)
 	if err != nil {
 		return err
 	}
-	pollService := poll.NewService(1*time.Hour, 2*time.Hour, pollStore, lightningPlugin, pol, lightningPlugin, supportedAssets, ps)
-	pollService.Start()
-	defer pollService.Stop()
+	defer func() {
+		if closeErr := peerStore.Close(); closeErr != nil {
+			log.Infof("peersync store close failed: %v", closeErr)
+		}
+	}()
+
+	nodeID, err := peersync.NewPeerID(lightningPlugin.GetNodeId())
+	if err != nil {
+		return err
+	}
+
+	// Use the peersync CLN lightning adapter; it self-registers message handler
+	clnAdapter := peersync.NewClnLightningAdapter(lightningPlugin)
+	peerSync := peersync.NewPeerSync(nodeID, peerStore, clnAdapter, pol, supportedAssets, ps)
+
+	psSupervisor := suture.New("peersync", suture.Spec{
+		EventHook: func(e suture.Event) {
+			log.Infof("peersync supervisor event: %s", e)
+		},
+	})
+	psSupervisor.Add(peerSync)
+
+	peerSyncCtx, peerSyncCancel := context.WithCancel(ctx)
+	peerSyncErrCh := psSupervisor.ServeBackground(peerSyncCtx)
+	defer func() {
+		peerSyncCancel()
+		if err := <-peerSyncErrCh; err != nil && !errors.Is(err, context.Canceled) {
+			log.Infof("peersync supervisor exited: %v", err)
+		}
+	}()
 
 	sp := swap.NewRequestedSwapsPrinter(requestedSwapStore)
-	lightningPlugin.SetupClients(liquidRpcWallet, swapService, pol, sp, bitcoinCli, bitcoinOnChainService, pollService, ps)
+	lightningPlugin.SetupClients(
+		liquidRpcWallet,
+		swapService,
+		pol,
+		sp,
+		bitcoinCli,
+		bitcoinOnChainService,
+		peerSync,
+		peerStore,
+		ps,
+	)
 
 	// We are ready to accept and handle requests.
 	// FIXME: Once we reworked the recovery service (non-blocking) we want to