Implement peer synchronization and management features

- Added a new Store implementation for managing peer states with tests.
- Introduced a Supervisor to manage the lifecycle of the peer sync loop, including restart logic and error handling.
- Created tests for the Supervisor to ensure it respects maximum restart limits and stops on cancellation.
- Removed outdated PollMessage and related service code, consolidating functionality into the new Store and Supervisor.
- Updated tests to reflect the removal of the old Poll service and ensure new functionality is covered.

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 {
@@ -520,13 +520,23 @@ func (cl *ClightningClient) isPeerConnected(nodeId string) bool {
 }
 
 // peerRunsPeerSwap returns true if the peer with peerId is listed in the
-// pollService.
+// peerRunsPeerSwap returns true if the peersync store has a compatible capability for the peer.
 func (cl *ClightningClient) peerRunsPeerSwap(peerId string) bool {
-	pollInfo, err := cl.pollService.GetPollFrom(peerId)
-	if err == nil && pollInfo != nil {
-		return true
+	if cl.peerStore == nil {
+		return false
+	}
+	id, err := peersync.NewPeerID(peerId)
+	if err != nil {
+		return false
+	}
+	peer, err := cl.peerStore.GetPeerState(id)
+	if err != nil || peer == nil {
+		return false
+	}
+	if peer.Capability() == nil {
+		return false
 	}
-	return false
+	return peer.IsCompatibleWith(peersync.NewVersion(swap.PEERSWAP_PROTOCOL_VERSION))
 }
 
 // This is called after the Plugin starts up successfully
@@ -549,10 +559,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,7 @@ import (
 
 	"github.com/elementsproject/peerswap/log"
 	"github.com/elementsproject/peerswap/peerswaprpc"
+	"github.com/elementsproject/peerswap/peersync"
 	"github.com/elementsproject/peerswap/premium"
 
 	"github.com/elementsproject/glightning/glightning"
@@ -559,15 +561,28 @@ func (l *ListPeers) Call() (jrpc2.Result, error) {
 		fundingChannels[channel.ShortChannelId] = channel
 	}
 
-	// get polls
-	polls, err := l.cl.pollService.GetCompatiblePolls()
+	peerStates, err := l.cl.peerStore.GetAllPeerStates()
 	if err != nil {
 		return nil, err
 	}
 
+	compatible := make(map[string]*peersync.Peer)
+	targetVersion := peersync.NewVersion(swap.PEERSWAP_PROTOCOL_VERSION)
+	for _, state := range peerStates {
+		if state == nil || state.Capability() == nil {
+			continue
+		}
+		if !state.IsCompatibleWith(targetVersion) {
+			continue
+		}
+		compatible[state.ID().String()] = state
+	}
+
 	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
@@ -603,8 +618,8 @@ func (l *ListPeers) Call() (jrpc2.Result, error) {
 
 			peerSwapPeer := &peerswaprpc.PeerSwapPeer{
 				NodeId:          peer.Id,
-				SwapsAllowed:    p.PeerAllowed,
-				SupportedAssets: p.Assets,
+				SwapsAllowed:    capability.IsAllowed(),
+				SupportedAssets: assetsFromCapability(capability),
 				AsSender: &peerswaprpc.SwapStats{
 					SwapsOut: SenderSwapsOut,
 					SwapsIn:  SenderSwapsIn,
@@ -624,22 +639,22 @@ func (l *ListPeers) Call() (jrpc2.Result, error) {
 						{
 							Asset:          peerswaprpc.AssetType_BTC,
 							Operation:      peerswaprpc.OperationType_SWAP_IN,
-							PremiumRatePpm: p.BTCSwapInPremiumRatePPM,
+							PremiumRatePpm: ppmFromCapability(capability, premium.BTC, premium.SwapIn),
 						},
 						{
 							Asset:          peerswaprpc.AssetType_BTC,
 							Operation:      peerswaprpc.OperationType_SWAP_OUT,
-							PremiumRatePpm: p.BTCSwapOutPremiumRatePPM,
+							PremiumRatePpm: ppmFromCapability(capability, premium.BTC, premium.SwapOut),
 						},
 						{
 							Asset:          peerswaprpc.AssetType_LBTC,
 							Operation:      peerswaprpc.OperationType_SWAP_IN,
-							PremiumRatePpm: p.LBTCSwapInPremiumRatePPM,
+							PremiumRatePpm: ppmFromCapability(capability, premium.LBTC, premium.SwapIn),
 						},
 						{
 							Asset:          peerswaprpc.AssetType_LBTC,
 							Operation:      peerswaprpc.OperationType_SWAP_OUT,
-							PremiumRatePpm: p.LBTCSwapOutPremiumRatePPM,
+							PremiumRatePpm: ppmFromCapability(capability, premium.LBTC, premium.SwapOut),
 						},
 					},
 				},
@@ -672,6 +687,27 @@ func (l *ListPeers) Call() (jrpc2.Result, error) {
 	return peerSwappers, nil
 }
 
+func assetsFromCapability(capability *peersync.PeerCapability) []string {
+	assets := capability.SupportedAssets()
+	res := make([]string, 0, len(assets))
+	for _, asset := range assets {
+		res = append(res, asset.String())
+	}
+	return res
+}
+
+func ppmFromCapability(
+	capability *peersync.PeerCapability,
+	asset premium.AssetType,
+	operation premium.OperationType,
+) int64 {
+	rate := capability.GetPremiumRate(asset, operation)
+	if rate == nil {
+		return 0
+	}
+	return rate.Value()
+}
+
 func channelActive(state string) bool {
 	return state == "CHANNELD_NORMAL"
 }
@@ -754,8 +790,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
 }
 

cmd/peerswap-plugin/main.go

@@ -26,8 +26,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 +372,47 @@ 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)
+
+	if err := peerSync.Start(ctx); err != nil {
+		return err
+	}
+	defer func() {
+		if stopErr := peerSync.Stop(); stopErr != nil {
+			log.Infof("peersync stop failed: %v", stopErr)
+		}
+	}()
 
 	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