2021-06-16 20:19:45 +00:00
// Copyright 2019 The Waku Library Authors.
//
// The Waku library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The Waku library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty off
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the Waku library. If not, see <http://www.gnu.org/licenses/>.
//
// This software uses the go-ethereum library, which is licensed
// under the GNU Lesser General Public Library, version 3 or any later.
package wakuv2
import (
"context"
"crypto/ecdsa"
"crypto/sha256"
2021-10-12 12:39:28 +00:00
"database/sql"
2021-06-16 20:19:45 +00:00
"errors"
"fmt"
2022-12-09 16:16:21 +00:00
"math"
2021-06-16 20:19:45 +00:00
"net"
"runtime"
2023-06-07 09:02:19 +00:00
"sort"
2021-11-09 12:22:34 +00:00
"strings"
2021-06-16 20:19:45 +00:00
"sync"
"time"
2022-11-04 13:56:45 +00:00
"github.com/libp2p/go-libp2p/core/peer"
2021-11-22 15:27:05 +00:00
"github.com/libp2p/go-libp2p/p2p/protocol/identify"
2021-10-06 16:08:54 +00:00
"github.com/multiformats/go-multiaddr"
2021-06-16 20:19:45 +00:00
"go.uber.org/zap"
mapset "github.com/deckarep/golang-set"
"golang.org/x/crypto/pbkdf2"
gethcommon "github.com/ethereum/go-ethereum/common"
2021-08-30 14:57:28 +00:00
"github.com/ethereum/go-ethereum/common/hexutil"
2021-06-16 20:19:45 +00:00
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/p2p"
2021-11-22 13:40:14 +00:00
"github.com/ethereum/go-ethereum/p2p/enode"
2021-06-16 20:19:45 +00:00
"github.com/ethereum/go-ethereum/rpc"
2021-08-30 21:35:37 +00:00
"github.com/libp2p/go-libp2p"
2021-10-19 13:43:41 +00:00
pubsub "github.com/libp2p/go-libp2p-pubsub"
2022-11-04 13:56:45 +00:00
"github.com/libp2p/go-libp2p/core/metrics"
2021-10-19 13:43:41 +00:00
2022-11-22 22:05:54 +00:00
"github.com/waku-org/go-waku/waku/v2/dnsdisc"
2023-06-30 13:41:32 +00:00
"github.com/waku-org/go-waku/waku/v2/peers"
2022-11-22 22:05:54 +00:00
"github.com/waku-org/go-waku/waku/v2/protocol"
2023-06-07 09:02:19 +00:00
"github.com/waku-org/go-waku/waku/v2/protocol/filter"
2022-11-29 12:43:11 +00:00
"github.com/waku-org/go-waku/waku/v2/protocol/peer_exchange"
2022-11-22 22:05:54 +00:00
"github.com/waku-org/go-waku/waku/v2/protocol/relay"
2021-06-16 20:19:45 +00:00
2022-12-09 16:16:21 +00:00
"github.com/status-im/status-go/connection"
2021-06-16 20:19:45 +00:00
"github.com/status-im/status-go/eth-node/types"
2022-12-06 15:24:01 +00:00
"github.com/status-im/status-go/timesource"
2021-06-16 20:19:45 +00:00
"github.com/status-im/status-go/wakuv2/common"
2021-10-12 12:39:28 +00:00
"github.com/status-im/status-go/wakuv2/persistence"
2021-06-16 20:19:45 +00:00
2022-11-22 22:05:54 +00:00
node "github.com/waku-org/go-waku/waku/v2/node"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/go-waku/waku/v2/protocol/store"
2023-02-22 21:58:17 +00:00
storepb "github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
2021-06-16 20:19:45 +00:00
)
const messageQueueLimit = 1024
2022-11-04 13:56:45 +00:00
const requestTimeout = 30 * time . Second
2022-12-09 16:16:21 +00:00
const bootnodesQueryBackoffMs = 200
const bootnodesMaxRetries = 7
2022-03-17 18:06:02 +00:00
2021-06-16 20:19:45 +00:00
type settings struct {
2022-01-12 20:04:43 +00:00
LightClient bool // Indicates if the node is a light client
2022-11-29 12:43:11 +00:00
MinPeersForRelay int // Indicates the minimum number of peers required for using Relay Protocol
2023-06-07 09:02:19 +00:00
MinPeersForFilter int // Indicates the minimum number of peers required for using Filter Protocol
2022-01-12 20:04:43 +00:00
MaxMsgSize uint32 // Maximal message length allowed by the waku node
EnableConfirmations bool // Enable sending message confirmations
2022-11-29 12:43:11 +00:00
PeerExchange bool // Enable peer exchange
DiscoveryLimit int // Indicates the number of nodes to discover
2023-01-03 14:14:59 +00:00
Nameserver string // Optional nameserver to use for dns discovery
2023-02-24 13:29:58 +00:00
EnableDiscV5 bool // Indicates whether discv5 is enabled or not
2023-04-14 16:08:06 +00:00
Options [ ] node . WakuNodeOption
2021-06-16 20:19:45 +00:00
}
// Waku represents a dark communication interface through the Ethereum
// network, using its very own P2P communication layer.
type Waku struct {
2021-11-22 15:27:05 +00:00
node * node . WakuNode // reference to a libp2p waku node
identifyService identify . IDService
appDB * sql . DB
2021-06-16 20:19:45 +00:00
2022-09-15 14:32:54 +00:00
dnsAddressCache map [ string ] [ ] dnsdisc . DiscoveredNode // Map to store the multiaddresses returned by dns discovery
dnsAddressCacheLock * sync . RWMutex // lock to handle access to the map
2021-11-09 12:22:34 +00:00
2023-06-07 09:02:19 +00:00
// Filter-related
filters * common . Filters // Message filters installed with Subscribe function
filterSubscriptions map [ * common . Filter ] map [ string ] * filter . SubscriptionDetails // wakuv2 filter subscription details
filterPeerDisconnectMap map [ peer . ID ] int64
isFilterSubAlive func ( sub * filter . SubscriptionDetails ) error
2021-06-16 20:19:45 +00:00
privateKeys map [ string ] * ecdsa . PrivateKey // Private key storage
symKeys map [ string ] [ ] byte // Symmetric key storage
keyMu sync . RWMutex // Mutex associated with key stores
envelopes map [ gethcommon . Hash ] * common . ReceivedMessage // Pool of envelopes currently tracked by this node
expirations map [ uint32 ] mapset . Set // Message expiration pool
poolMu sync . RWMutex // Mutex to sync the message and expiration pools
2021-08-30 21:35:37 +00:00
bandwidthCounter * metrics . BandwidthCounter
2021-08-03 19:27:15 +00:00
2023-05-22 21:38:02 +00:00
protectedTopicStore * persistence . ProtectedTopicsStore
sendQueue chan * protocol . Envelope
msgQueue chan * common . ReceivedMessage // Message queue for waku messages that havent been decoded
2023-07-17 17:20:55 +00:00
ctx context . Context
cancel context . CancelFunc
wg sync . WaitGroup
2021-06-16 20:19:45 +00:00
2023-04-14 16:08:06 +00:00
cfg * Config
2021-06-16 20:19:45 +00:00
settings settings // Holds configuration settings that can be dynamically changed
settingsMu sync . RWMutex // Mutex to sync the settings access
envelopeFeed event . Feed
2021-12-01 15:15:18 +00:00
storeMsgIDs map [ gethcommon . Hash ] bool // Map of the currently processing ids
storeMsgIDsMu sync . RWMutex
2023-04-14 16:08:06 +00:00
connStatusChan chan node . ConnStatus
2022-01-12 16:02:01 +00:00
connStatusSubscriptions map [ string ] * types . ConnStatusSubscription
connStatusMu sync . Mutex
2021-06-16 20:19:45 +00:00
logger * zap . Logger
2022-12-06 15:24:01 +00:00
// NTP Synced timesource
timesource * timesource . NTPTimeSource
2022-12-09 16:16:21 +00:00
// seededBootnodesForDiscV5 indicates whether we manage to retrieve discovery
// bootnodes successfully
seededBootnodesForDiscV5 bool
// offline indicates whether we have detected connectivity
offline bool
// connectionChanged is channel that notifies when connectivity has changed
connectionChanged chan struct { }
// discV5BootstrapNodes is the ENR to be used to fetch bootstrap nodes for discovery
discV5BootstrapNodes [ ] string
2023-07-05 15:56:34 +00:00
onHistoricMessagesRequestFailed func ( [ ] byte , peer . ID , error )
onPeerStats func ( types . ConnStatus )
2021-06-16 20:19:45 +00:00
}
2023-01-17 04:03:19 +00:00
func getUsableUDPPort ( ) ( int , error ) {
conn , err := net . ListenUDP ( "udp" , & net . UDPAddr {
IP : net . IPv4zero ,
Port : 0 ,
} )
if err != nil {
return 0 , err
}
defer conn . Close ( )
return conn . LocalAddr ( ) . ( * net . UDPAddr ) . Port , nil
}
2021-06-16 20:19:45 +00:00
// New creates a WakuV2 client ready to communicate through the LibP2P network.
2023-07-05 15:56:34 +00:00
func New ( nodeKey string , fleet string , cfg * Config , logger * zap . Logger , appDB * sql . DB , ts * timesource . NTPTimeSource , onHistoricMessagesRequestFailed func ( [ ] byte , peer . ID , error ) , onPeerStats func ( types . ConnStatus ) ) ( * Waku , error ) {
2022-12-09 16:16:21 +00:00
var err error
2021-06-16 20:19:45 +00:00
if logger == nil {
2022-12-09 16:16:21 +00:00
logger , err = zap . NewDevelopment ( )
if err != nil {
return nil , err
}
2021-06-16 20:19:45 +00:00
}
2023-03-24 12:05:42 +00:00
if ts == nil {
ts = timesource . Default ( )
}
2021-11-18 18:31:28 +00:00
cfg = setDefaults ( cfg )
2023-01-17 04:03:19 +00:00
if cfg . UDPPort == 0 {
cfg . UDPPort , err = getUsableUDPPort ( )
if err != nil {
return nil , err
}
}
2021-06-16 20:19:45 +00:00
logger . Debug ( "starting wakuv2 with config" , zap . Any ( "config" , cfg ) )
waku := & Waku {
2023-07-05 15:56:34 +00:00
appDB : appDB ,
cfg : cfg ,
privateKeys : make ( map [ string ] * ecdsa . PrivateKey ) ,
symKeys : make ( map [ string ] [ ] byte ) ,
envelopes : make ( map [ gethcommon . Hash ] * common . ReceivedMessage ) ,
expirations : make ( map [ uint32 ] mapset . Set ) ,
msgQueue : make ( chan * common . ReceivedMessage , messageQueueLimit ) ,
sendQueue : make ( chan * protocol . Envelope , 1000 ) ,
connStatusChan : make ( chan node . ConnStatus , 100 ) ,
connStatusSubscriptions : make ( map [ string ] * types . ConnStatusSubscription ) ,
wg : sync . WaitGroup { } ,
dnsAddressCache : make ( map [ string ] [ ] dnsdisc . DiscoveredNode ) ,
dnsAddressCacheLock : & sync . RWMutex { } ,
storeMsgIDs : make ( map [ gethcommon . Hash ] bool ) ,
filterPeerDisconnectMap : make ( map [ peer . ID ] int64 ) ,
filterSubscriptions : make ( map [ * common . Filter ] map [ string ] * filter . SubscriptionDetails ) ,
timesource : ts ,
storeMsgIDsMu : sync . RWMutex { } ,
logger : logger ,
discV5BootstrapNodes : cfg . DiscV5BootstrapNodes ,
onHistoricMessagesRequestFailed : onHistoricMessagesRequestFailed ,
onPeerStats : onPeerStats ,
2021-06-16 20:19:45 +00:00
}
2023-07-17 17:20:55 +00:00
2023-06-07 09:02:19 +00:00
// This fn is being mocked in test
waku . isFilterSubAlive = func ( sub * filter . SubscriptionDetails ) error {
2023-07-17 17:20:55 +00:00
return waku . node . FilterLightnode ( ) . IsSubscriptionAlive ( waku . ctx , sub )
2023-06-07 09:02:19 +00:00
}
2021-06-16 20:19:45 +00:00
waku . settings = settings {
2023-06-07 09:02:19 +00:00
MaxMsgSize : cfg . MaxMessageSize ,
LightClient : cfg . LightClient ,
MinPeersForRelay : cfg . MinPeersForRelay ,
MinPeersForFilter : cfg . MinPeersForFilter ,
PeerExchange : cfg . PeerExchange ,
DiscoveryLimit : cfg . DiscoveryLimit ,
Nameserver : cfg . Nameserver ,
EnableDiscV5 : cfg . EnableDiscV5 ,
2021-06-16 20:19:45 +00:00
}
waku . filters = common . NewFilters ( )
2021-08-30 21:35:37 +00:00
waku . bandwidthCounter = metrics . NewBandwidthCounter ( )
2021-06-16 20:19:45 +00:00
var privateKey * ecdsa . PrivateKey
if nodeKey != "" {
privateKey , err = crypto . HexToECDSA ( nodeKey )
} else {
// If no nodekey is provided, create an ephemeral key
privateKey , err = crypto . GenerateKey ( )
}
if err != nil {
return nil , fmt . Errorf ( "failed to setup the go-waku private key: %v" , err )
}
hostAddr , err := net . ResolveTCPAddr ( "tcp" , fmt . Sprint ( cfg . Host , ":" , cfg . Port ) )
if err != nil {
return nil , fmt . Errorf ( "failed to setup the network interface: %v" , err )
}
2021-09-23 16:17:57 +00:00
if cfg . KeepAliveInterval == 0 {
cfg . KeepAliveInterval = DefaultConfig . KeepAliveInterval
}
2021-10-12 12:39:28 +00:00
libp2pOpts := node . DefaultLibP2POptions
libp2pOpts = append ( libp2pOpts , libp2p . BandwidthReporter ( waku . bandwidthCounter ) )
2022-03-30 00:51:51 +00:00
libp2pOpts = append ( libp2pOpts , libp2p . NATPortMap ( ) )
2021-10-12 12:39:28 +00:00
2021-09-23 16:17:57 +00:00
opts := [ ] node . WakuNodeOption {
2021-11-09 12:22:34 +00:00
node . WithLibP2POptions ( libp2pOpts ... ) ,
2021-06-16 20:19:45 +00:00
node . WithPrivateKey ( privateKey ) ,
2021-11-22 13:40:14 +00:00
node . WithHostAddress ( hostAddr ) ,
2023-04-14 16:08:06 +00:00
node . WithConnectionStatusChannel ( waku . connStatusChan ) ,
2021-09-23 16:17:57 +00:00
node . WithKeepAlive ( time . Duration ( cfg . KeepAliveInterval ) * time . Second ) ,
2023-01-16 13:00:07 +00:00
node . WithDiscoverParams ( cfg . DiscoveryLimit ) ,
2022-10-20 16:14:11 +00:00
node . WithLogger ( logger ) ,
2021-09-23 16:17:57 +00:00
}
2021-11-22 13:40:14 +00:00
if cfg . EnableDiscV5 {
2023-04-14 16:08:06 +00:00
bootnodes , err := waku . getDiscV5BootstrapNodes ( context . Background ( ) , cfg . DiscV5BootstrapNodes )
2022-09-15 14:32:54 +00:00
if err != nil {
2022-12-09 16:16:21 +00:00
logger . Error ( "failed to get bootstrap nodes" , zap . Error ( err ) )
2022-09-15 14:32:54 +00:00
return nil , err
2021-11-22 13:40:14 +00:00
}
2022-12-09 16:16:21 +00:00
2023-01-16 13:00:07 +00:00
opts = append ( opts , node . WithDiscoveryV5 ( uint ( cfg . UDPPort ) , bootnodes , cfg . AutoUpdate ) )
2021-10-06 16:08:54 +00:00
}
2021-09-23 16:17:57 +00:00
if cfg . LightClient {
2023-06-07 09:02:19 +00:00
opts = append ( opts , node . WithWakuFilterLightNode ( ) )
2021-09-23 16:17:57 +00:00
} else {
2021-10-19 13:43:41 +00:00
relayOpts := [ ] pubsub . Option {
pubsub . WithMaxMessageSize ( int ( waku . settings . MaxMsgSize ) ) ,
2021-09-28 14:03:32 +00:00
}
2022-10-04 17:54:13 +00:00
opts = append ( opts , node . WithWakuRelayAndMinPeers ( waku . settings . MinPeersForRelay , relayOpts ... ) )
2021-09-23 16:17:57 +00:00
}
2022-08-19 16:34:07 +00:00
if cfg . EnableStore {
2023-01-07 15:25:55 +00:00
opts = append ( opts , node . WithWakuStore ( ) )
2022-08-19 16:34:07 +00:00
dbStore , err := persistence . NewDBStore ( logger , persistence . WithDB ( appDB ) , persistence . WithRetentionPolicy ( cfg . StoreCapacity , time . Duration ( cfg . StoreSeconds ) * time . Second ) )
if err != nil {
return nil , err
}
opts = append ( opts , node . WithMessageProvider ( dbStore ) )
}
2023-05-22 21:38:02 +00:00
if appDB != nil {
waku . protectedTopicStore , err = persistence . NewProtectedTopicsStore ( logger , appDB )
if err != nil {
return nil , err
}
}
2023-04-14 16:08:06 +00:00
waku . settings . Options = opts
2022-11-04 13:56:45 +00:00
waku . logger . Info ( "setup the go-waku node successfully" )
2021-09-23 16:17:57 +00:00
return waku , nil
}
2022-01-12 16:02:01 +00:00
func ( w * Waku ) SubscribeToConnStatusChanges ( ) * types . ConnStatusSubscription {
w . connStatusMu . Lock ( )
defer w . connStatusMu . Unlock ( )
subscription := types . NewConnStatusSubscription ( )
w . connStatusSubscriptions [ subscription . ID ] = subscription
return subscription
}
2021-11-22 15:27:05 +00:00
func ( w * Waku ) getDiscV5BootstrapNodes ( ctx context . Context , addresses [ ] string ) ( [ ] * enode . Node , error ) {
2022-09-15 14:32:54 +00:00
wg := sync . WaitGroup { }
mu := sync . Mutex { }
var result [ ] * enode . Node
2021-11-22 15:27:05 +00:00
retrieveENR := func ( d dnsdisc . DiscoveredNode , wg * sync . WaitGroup ) {
2022-09-15 14:32:54 +00:00
mu . Lock ( )
defer mu . Unlock ( )
2021-11-22 15:27:05 +00:00
defer wg . Done ( )
2022-09-16 13:09:13 +00:00
if d . ENR != nil {
result = append ( result , d . ENR )
}
2022-09-15 14:32:54 +00:00
}
for _ , addrString := range addresses {
if addrString == "" {
continue
}
if strings . HasPrefix ( addrString , "enrtree://" ) {
// Use DNS Discovery
wg . Add ( 1 )
go func ( addr string ) {
defer wg . Done ( )
2021-11-22 15:27:05 +00:00
w . dnsDiscover ( ctx , addr , retrieveENR )
2022-09-15 14:32:54 +00:00
} ( addrString )
} else {
// It's a normal enr
bootnode , err := enode . Parse ( enode . ValidSchemes , addrString )
if err != nil {
return nil , err
}
result = append ( result , bootnode )
}
}
wg . Wait ( )
2022-12-09 16:16:21 +00:00
w . seededBootnodesForDiscV5 = len ( result ) > 0
2022-09-15 14:32:54 +00:00
return result , nil
}
2021-11-22 15:27:05 +00:00
type fnApplyToEachPeer func ( d dnsdisc . DiscoveredNode , wg * sync . WaitGroup )
func ( w * Waku ) dnsDiscover ( ctx context . Context , enrtreeAddress string , apply fnApplyToEachPeer ) {
2022-12-09 16:16:21 +00:00
w . logger . Info ( "retrieving nodes" , zap . String ( "enr" , enrtreeAddress ) )
2021-11-22 15:27:05 +00:00
ctx , cancel := context . WithTimeout ( ctx , requestTimeout )
2021-11-09 12:22:34 +00:00
defer cancel ( )
2022-09-16 13:09:13 +00:00
w . dnsAddressCacheLock . Lock ( )
defer w . dnsAddressCacheLock . Unlock ( )
2021-10-06 16:08:54 +00:00
2022-09-16 13:09:13 +00:00
discNodes , ok := w . dnsAddressCache [ enrtreeAddress ]
2021-11-09 12:22:34 +00:00
if ! ok {
2023-01-03 14:14:59 +00:00
w . settingsMu . RLock ( )
nameserver := w . settings . Nameserver
w . settingsMu . RUnlock ( )
var opts [ ] dnsdisc . DnsDiscoveryOption
if nameserver != "" {
opts = append ( opts , dnsdisc . WithNameserver ( nameserver ) )
}
discoveredNodes , err := dnsdisc . RetrieveNodes ( ctx , enrtreeAddress , opts ... )
2021-10-06 16:08:54 +00:00
if err != nil {
2022-11-04 13:56:45 +00:00
w . logger . Warn ( "dns discovery error " , zap . Error ( err ) )
2021-11-09 12:22:34 +00:00
return
2021-10-06 16:08:54 +00:00
}
2022-09-16 13:09:13 +00:00
2022-12-09 16:16:21 +00:00
if len ( discoveredNodes ) != 0 {
w . dnsAddressCache [ enrtreeAddress ] = append ( w . dnsAddressCache [ enrtreeAddress ] , discoveredNodes ... )
discNodes = w . dnsAddressCache [ enrtreeAddress ]
}
2021-11-09 12:22:34 +00:00
}
2021-10-06 16:08:54 +00:00
2021-11-22 15:27:05 +00:00
wg := & sync . WaitGroup { }
wg . Add ( len ( discNodes ) )
2022-09-15 14:32:54 +00:00
for _ , d := range discNodes {
2021-11-22 15:27:05 +00:00
apply ( d , wg )
}
wg . Wait ( )
}
func ( w * Waku ) addWakuV2Peers ( ctx context . Context , cfg * Config ) error {
fnApply := func ( d dnsdisc . DiscoveredNode , wg * sync . WaitGroup ) {
2023-05-16 15:57:47 +00:00
if len ( d . PeerInfo . Addrs ) != 0 {
2021-11-22 15:27:05 +00:00
go func ( ma multiaddr . Multiaddr ) {
w . identifyAndConnect ( ctx , w . settings . LightClient , ma )
wg . Done ( )
2023-05-16 15:57:47 +00:00
} ( d . PeerInfo . Addrs [ 0 ] )
2021-11-22 15:27:05 +00:00
}
}
identifyWg := & sync . WaitGroup { }
identifyWg . Add ( len ( cfg . WakuNodes ) )
for _ , addrString := range cfg . WakuNodes {
2023-01-03 14:14:59 +00:00
addrString := addrString
2021-11-22 15:27:05 +00:00
if strings . HasPrefix ( addrString , "enrtree://" ) {
// Use DNS Discovery
go func ( ) {
w . dnsDiscover ( ctx , addrString , fnApply )
identifyWg . Done ( )
} ( )
} else {
// It is a normal multiaddress
addr , err := multiaddr . NewMultiaddr ( addrString )
if err != nil {
w . logger . Warn ( "invalid peer multiaddress" , zap . String ( "ma" , addrString ) , zap . Error ( err ) )
continue
}
go func ( ma multiaddr . Multiaddr ) {
w . identifyAndConnect ( ctx , cfg . LightClient , ma )
identifyWg . Done ( )
} ( addr )
}
2021-10-06 16:08:54 +00:00
}
2021-11-22 15:27:05 +00:00
identifyWg . Wait ( )
return nil
2021-10-06 16:08:54 +00:00
}
2021-11-22 15:27:05 +00:00
func ( w * Waku ) identifyAndConnect ( ctx context . Context , isLightClient bool , ma multiaddr . Multiaddr ) {
peerInfo , err := peer . AddrInfoFromP2pAddr ( ma )
2021-11-09 12:22:34 +00:00
if err != nil {
2021-11-22 15:27:05 +00:00
w . logger . Warn ( "invalid peer multiaddress" , zap . String ( "addr" , ma . String ( ) ) , zap . Error ( err ) )
2021-11-09 12:22:34 +00:00
return
}
2021-11-22 15:27:05 +00:00
ctx , cancel := context . WithTimeout ( ctx , 3 * time . Second )
defer cancel ( )
err = w . node . Host ( ) . Connect ( ctx , * peerInfo )
if err != nil {
w . logger . Error ( "could not extract peerinfo" , zap . String ( "ma" , ma . String ( ) ) , zap . Error ( err ) )
return
2022-09-15 14:32:54 +00:00
}
2021-11-09 12:22:34 +00:00
2021-11-22 15:27:05 +00:00
conns := w . node . Host ( ) . Network ( ) . ConnsToPeer ( peerInfo . ID )
if len ( conns ) == 0 {
return // No connection
2021-11-09 12:22:34 +00:00
}
2021-11-22 15:27:05 +00:00
w . identifyService . IdentifyConn ( conns [ 0 ] )
if isLightClient {
err = w . node . Host ( ) . Network ( ) . ClosePeer ( peerInfo . ID )
if err != nil {
w . logger . Error ( "could not close connections to peer" , zap . Any ( "peer" , peerInfo . ID ) , zap . Error ( err ) )
2021-09-23 16:17:57 +00:00
}
2021-11-22 15:27:05 +00:00
return
2021-06-16 20:19:45 +00:00
}
2023-02-22 21:58:17 +00:00
supportedProtocols , err := w . node . Host ( ) . Peerstore ( ) . SupportsProtocols ( peerInfo . ID , relay . WakuRelayID_v200 )
2021-11-22 15:27:05 +00:00
if err != nil {
w . logger . Error ( "could not obtain protocols" , zap . Any ( "peer" , peerInfo . ID ) , zap . Error ( err ) )
return
}
if len ( supportedProtocols ) == 0 {
err = w . node . Host ( ) . Network ( ) . ClosePeer ( peerInfo . ID )
if err != nil {
w . logger . Error ( "could not close connections to peer" , zap . Any ( "peer" , peerInfo . ID ) , zap . Error ( err ) )
}
}
2021-06-16 20:19:45 +00:00
}
2023-01-31 22:46:47 +00:00
func ( w * Waku ) telemetryBandwidthStats ( telemetryServerURL string ) {
if telemetryServerURL == "" {
return
}
2023-02-17 13:15:28 +00:00
telemetry := NewBandwidthTelemetryClient ( w . logger , telemetryServerURL )
2023-01-31 22:46:47 +00:00
ticker := time . NewTicker ( time . Second * 20 )
defer ticker . Stop ( )
2023-02-01 15:46:01 +00:00
today := time . Now ( )
2023-01-31 22:46:47 +00:00
for {
select {
2023-07-17 17:20:55 +00:00
case <- w . ctx . Done ( ) :
2023-01-31 22:46:47 +00:00
return
2023-02-01 15:46:01 +00:00
case now := <- ticker . C :
// Reset totals when day changes
if now . Day ( ) != today . Day ( ) {
today = now
w . bandwidthCounter . Reset ( )
}
2023-01-31 22:46:47 +00:00
storeStats := w . bandwidthCounter . GetBandwidthForProtocol ( store . StoreID_v20beta4 )
relayStats := w . bandwidthCounter . GetBandwidthForProtocol ( relay . WakuRelayID_v200 )
go telemetry . PushProtocolStats ( relayStats , storeStats )
}
}
}
2021-08-03 19:27:15 +00:00
func ( w * Waku ) GetStats ( ) types . StatsSummary {
2021-08-30 21:35:37 +00:00
stats := w . bandwidthCounter . GetBandwidthTotals ( )
return types . StatsSummary {
UploadRate : uint64 ( stats . RateOut ) ,
DownloadRate : uint64 ( stats . RateIn ) ,
}
2021-08-03 19:27:15 +00:00
}
2022-11-29 12:43:11 +00:00
func ( w * Waku ) runPeerExchangeLoop ( ) {
defer w . wg . Done ( )
2023-05-08 21:43:27 +00:00
if ! w . settings . PeerExchange || ! w . settings . LightClient {
2022-11-29 12:43:11 +00:00
// Currently peer exchange is only used for full nodes
// TODO: should it be used for lightpush? or lightpush nodes
// are only going to be selected from a specific set of peers?
return
}
ticker := time . NewTicker ( time . Second * 5 )
defer ticker . Stop ( )
for {
select {
2023-07-17 17:20:55 +00:00
case <- w . ctx . Done ( ) :
2022-11-29 12:43:11 +00:00
return
case <- ticker . C :
w . logger . Debug ( "Running peer exchange loop" )
connectedPeers := w . node . Host ( ) . Network ( ) . Peers ( )
peersWithRelay := 0
for _ , p := range connectedPeers {
2023-02-22 21:58:17 +00:00
supportedProtocols , err := w . node . Host ( ) . Peerstore ( ) . SupportsProtocols ( p , relay . WakuRelayID_v200 )
2022-11-29 12:43:11 +00:00
if err != nil {
continue
}
if len ( supportedProtocols ) != 0 {
peersWithRelay ++
}
}
peersToDiscover := w . settings . DiscoveryLimit - peersWithRelay
if peersToDiscover <= 0 {
continue
}
// We select only the nodes discovered via DNS Discovery that support peer exchange
w . dnsAddressCacheLock . RLock ( )
var withThesePeers [ ] peer . ID
for _ , record := range w . dnsAddressCache {
for _ , discoveredNode := range record {
2023-05-16 15:57:47 +00:00
if len ( discoveredNode . PeerInfo . Addrs ) == 0 {
2022-11-29 12:43:11 +00:00
continue
}
// Obtaining peer ID
2023-05-16 15:57:47 +00:00
peerIDString , err := discoveredNode . PeerInfo . Addrs [ 0 ] . ValueForProtocol ( multiaddr . P_P2P )
2022-11-29 12:43:11 +00:00
if err != nil {
2023-05-16 15:57:47 +00:00
w . logger . Warn ( "multiaddress does not contain peerID" , zap . String ( "multiaddr" , discoveredNode . PeerInfo . Addrs [ 0 ] . String ( ) ) )
2022-11-29 12:43:11 +00:00
continue // No peer ID available somehow
}
peerID , err := peer . Decode ( peerIDString )
if err != nil {
w . logger . Warn ( "couldnt decode peerID" , zap . String ( "peerIDString" , peerIDString ) )
continue // Couldnt decode the peerID for some reason?
}
2023-02-22 21:58:17 +00:00
supportsProtocol , _ := w . node . Host ( ) . Peerstore ( ) . SupportsProtocols ( peerID , peer_exchange . PeerExchangeID_v20alpha1 )
2022-11-29 12:43:11 +00:00
if len ( supportsProtocol ) != 0 {
withThesePeers = append ( withThesePeers , peerID )
}
}
}
w . dnsAddressCacheLock . RUnlock ( )
if len ( withThesePeers ) == 0 {
continue // No peers with peer exchange have been discovered via DNS Discovery so far, skip this iteration
}
2023-07-17 17:20:55 +00:00
err := w . node . PeerExchange ( ) . Request ( w . ctx , peersToDiscover , peer_exchange . WithAutomaticPeerSelection ( withThesePeers ... ) )
2022-11-29 12:43:11 +00:00
if err != nil {
w . logger . Error ( "couldnt request peers via peer exchange" , zap . Error ( err ) )
}
}
}
}
2023-05-22 21:38:02 +00:00
func ( w * Waku ) subscribeToPubsubTopicWithWakuRelay ( topic string , pubkey * ecdsa . PublicKey ) error {
2021-09-23 16:17:57 +00:00
if w . settings . LightClient {
2023-05-22 21:38:02 +00:00
return errors . New ( "only available for full nodes" )
}
if w . node . Relay ( ) . IsSubscribed ( topic ) {
return nil
2021-09-23 16:17:57 +00:00
}
2023-05-22 21:38:02 +00:00
if pubkey != nil {
err := w . node . Relay ( ) . AddSignedTopicValidator ( topic , pubkey )
if err != nil {
return err
}
}
sub , err := w . node . Relay ( ) . SubscribeToTopic ( context . Background ( ) , topic )
2021-06-16 20:19:45 +00:00
if err != nil {
2023-05-22 21:38:02 +00:00
return err
2021-06-16 20:19:45 +00:00
}
2023-05-22 21:38:02 +00:00
w . wg . Add ( 1 )
go func ( ) {
defer w . wg . Done ( )
for {
select {
case <- w . ctx . Done ( ) :
sub . Unsubscribe ( )
return
case env := <- sub . Ch :
envelopeErrors , err := w . OnNewEnvelopes ( env , common . RelayedMessageType )
if err != nil {
w . logger . Error ( "onNewEnvelope error" , zap . Error ( err ) )
}
// TODO: should these be handled?
_ = envelopeErrors
_ = err
2022-11-11 14:34:56 +00:00
}
2022-11-04 13:56:45 +00:00
}
2023-05-22 21:38:02 +00:00
} ( )
return nil
2021-06-16 20:19:45 +00:00
}
2023-06-07 09:02:19 +00:00
func ( w * Waku ) runFilterSubscriptionLoop ( sub * filter . SubscriptionDetails ) {
2021-09-23 16:17:57 +00:00
for {
select {
2023-07-17 17:20:55 +00:00
case <- w . ctx . Done ( ) :
2021-09-23 16:17:57 +00:00
return
2023-06-07 09:02:19 +00:00
case env , ok := <- sub . C :
2023-04-17 15:45:45 +00:00
if ok {
envelopeErrors , err := w . OnNewEnvelopes ( env , common . RelayedMessageType )
// TODO: should these be handled?
_ = envelopeErrors
_ = err
2023-06-07 09:02:19 +00:00
} else {
return
2023-04-17 15:45:45 +00:00
}
2021-09-23 16:17:57 +00:00
}
}
}
2023-06-07 09:02:19 +00:00
func ( w * Waku ) runFilterMsgLoop ( ) {
defer w . wg . Done ( )
2021-10-19 13:43:41 +00:00
2023-06-07 09:02:19 +00:00
if ! w . settings . LightClient {
return
2021-10-19 13:43:41 +00:00
}
2021-11-09 12:22:34 +00:00
2023-06-07 09:02:19 +00:00
// Use it to ping filter peer(s) periodically
ticker := time . NewTicker ( time . Duration ( w . cfg . KeepAliveInterval ) * time . Second )
defer ticker . Stop ( )
for {
select {
2023-07-17 17:20:55 +00:00
case <- w . ctx . Done ( ) :
2023-06-07 09:02:19 +00:00
return
case <- ticker . C :
for f , subMap := range w . filterSubscriptions {
if len ( subMap ) == 0 {
// All peers have disconnected on previous iteration,
// attempt full reconnect
err := w . subscribeToFilter ( f )
if err != nil {
w . logger . Error ( "Failed to subscribe to filter" )
}
2023-06-15 14:42:54 +00:00
break
2023-06-07 09:02:19 +00:00
}
for id , sub := range subMap {
err := w . isFilterSubAlive ( sub )
if err != nil {
2023-06-15 14:42:54 +00:00
// Unsubscribe on light node
2023-05-22 21:38:02 +00:00
contentFilter := w . buildContentFilter ( f . PubsubTopic , f . Topics )
2023-06-15 14:42:54 +00:00
// TODO Better return value handling for WakuFilterPushResult
2023-07-17 17:20:55 +00:00
_ , err := w . node . FilterLightnode ( ) . Unsubscribe ( w . ctx , contentFilter , filter . Peer ( sub . PeerID ) )
2023-06-15 14:42:54 +00:00
if err != nil {
w . logger . Warn ( "could not unsubscribe wakuv2 filter for peer" , zap . Any ( "peer" , sub . PeerID ) )
continue
}
// Remove entry from maps
2023-06-07 09:02:19 +00:00
w . filterPeerDisconnectMap [ sub . PeerID ] = time . Now ( ) . Unix ( )
delete ( subMap , id )
// Re-subscribe
peers := w . findFilterPeers ( )
if len ( peers ) > 0 && len ( subMap ) < w . settings . MinPeersForFilter {
2023-07-17 17:20:55 +00:00
subDetails , err := w . node . FilterLightnode ( ) . Subscribe ( w . ctx , contentFilter , filter . WithPeer ( peers [ 0 ] ) )
2023-06-07 09:02:19 +00:00
if err != nil {
w . logger . Warn ( "could not add wakuv2 filter for peer" , zap . Any ( "peer" , peers [ 0 ] ) )
2023-06-15 14:42:54 +00:00
break
2023-06-07 09:02:19 +00:00
}
subMap [ subDetails . ID ] = subDetails
go w . runFilterSubscriptionLoop ( subDetails )
break
}
}
}
}
}
}
}
2023-05-22 21:38:02 +00:00
func ( w * Waku ) buildContentFilter ( pubsubTopic string , topics [ ] [ ] byte ) filter . ContentFilter {
2023-06-07 09:02:19 +00:00
contentFilter := filter . ContentFilter {
2023-05-22 21:38:02 +00:00
Topic : pubsubTopic ,
2023-06-07 09:02:19 +00:00
}
for _ , topic := range topics {
contentFilter . ContentTopics = append ( contentFilter . ContentTopics , common . BytesToTopic ( topic ) . ContentTopic ( ) )
2021-09-23 16:17:57 +00:00
}
2021-11-09 12:22:34 +00:00
2023-06-07 09:02:19 +00:00
return contentFilter
2021-09-23 16:17:57 +00:00
}
2021-06-16 20:19:45 +00:00
// MaxMessageSize returns the maximum accepted message size.
func ( w * Waku ) MaxMessageSize ( ) uint32 {
w . settingsMu . RLock ( )
defer w . settingsMu . RUnlock ( )
return w . settings . MaxMsgSize
}
// ConfirmationsEnabled returns true if message confirmations are enabled.
func ( w * Waku ) ConfirmationsEnabled ( ) bool {
w . settingsMu . RLock ( )
defer w . settingsMu . RUnlock ( )
return w . settings . EnableConfirmations
}
// CurrentTime returns current time.
func ( w * Waku ) CurrentTime ( ) time . Time {
2023-03-24 12:05:42 +00:00
return w . timesource . Now ( )
2021-06-16 20:19:45 +00:00
}
// APIs returns the RPC descriptors the Waku implementation offers
func ( w * Waku ) APIs ( ) [ ] rpc . API {
return [ ] rpc . API {
{
Namespace : Name ,
Version : VersionStr ,
Service : NewPublicWakuAPI ( w ) ,
Public : false ,
} ,
}
}
// Protocols returns the waku sub-protocols ran by this particular client.
func ( w * Waku ) Protocols ( ) [ ] p2p . Protocol {
return [ ] p2p . Protocol { }
}
func ( w * Waku ) SendEnvelopeEvent ( event common . EnvelopeEvent ) int {
return w . envelopeFeed . Send ( event )
}
// SubscribeEnvelopeEvents subscribes to envelopes feed.
// In order to prevent blocking waku producers events must be amply buffered.
func ( w * Waku ) SubscribeEnvelopeEvents ( events chan <- common . EnvelopeEvent ) event . Subscription {
return w . envelopeFeed . Subscribe ( events )
}
// NewKeyPair generates a new cryptographic identity for the client, and injects
// it into the known identities for message decryption. Returns ID of the new key pair.
func ( w * Waku ) NewKeyPair ( ) ( string , error ) {
key , err := crypto . GenerateKey ( )
if err != nil || ! validatePrivateKey ( key ) {
key , err = crypto . GenerateKey ( ) // retry once
}
if err != nil {
return "" , err
}
if ! validatePrivateKey ( key ) {
return "" , fmt . Errorf ( "failed to generate valid key" )
}
id , err := toDeterministicID ( hexutil . Encode ( crypto . FromECDSAPub ( & key . PublicKey ) ) , common . KeyIDSize )
if err != nil {
return "" , err
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . privateKeys [ id ] != nil {
return "" , fmt . Errorf ( "failed to generate unique ID" )
}
w . privateKeys [ id ] = key
return id , nil
}
// DeleteKeyPair deletes the specified key if it exists.
func ( w * Waku ) DeleteKeyPair ( key string ) bool {
deterministicID , err := toDeterministicID ( key , common . KeyIDSize )
if err != nil {
return false
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . privateKeys [ deterministicID ] != nil {
delete ( w . privateKeys , deterministicID )
return true
}
return false
}
// AddKeyPair imports a asymmetric private key and returns it identifier.
func ( w * Waku ) AddKeyPair ( key * ecdsa . PrivateKey ) ( string , error ) {
id , err := makeDeterministicID ( hexutil . Encode ( crypto . FromECDSAPub ( & key . PublicKey ) ) , common . KeyIDSize )
if err != nil {
return "" , err
}
if w . HasKeyPair ( id ) {
return id , nil // no need to re-inject
}
w . keyMu . Lock ( )
w . privateKeys [ id ] = key
w . keyMu . Unlock ( )
return id , nil
}
// SelectKeyPair adds cryptographic identity, and makes sure
// that it is the only private key known to the node.
func ( w * Waku ) SelectKeyPair ( key * ecdsa . PrivateKey ) error {
id , err := makeDeterministicID ( hexutil . Encode ( crypto . FromECDSAPub ( & key . PublicKey ) ) , common . KeyIDSize )
if err != nil {
return err
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
w . privateKeys = make ( map [ string ] * ecdsa . PrivateKey ) // reset key store
w . privateKeys [ id ] = key
return nil
}
// DeleteKeyPairs removes all cryptographic identities known to the node
func ( w * Waku ) DeleteKeyPairs ( ) error {
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
w . privateKeys = make ( map [ string ] * ecdsa . PrivateKey )
return nil
}
// HasKeyPair checks if the waku node is configured with the private key
// of the specified public pair.
func ( w * Waku ) HasKeyPair ( id string ) bool {
deterministicID , err := toDeterministicID ( id , common . KeyIDSize )
if err != nil {
return false
}
w . keyMu . RLock ( )
defer w . keyMu . RUnlock ( )
return w . privateKeys [ deterministicID ] != nil
}
// GetPrivateKey retrieves the private key of the specified identity.
func ( w * Waku ) GetPrivateKey ( id string ) ( * ecdsa . PrivateKey , error ) {
deterministicID , err := toDeterministicID ( id , common . KeyIDSize )
if err != nil {
return nil , err
}
w . keyMu . RLock ( )
defer w . keyMu . RUnlock ( )
key := w . privateKeys [ deterministicID ]
if key == nil {
return nil , fmt . Errorf ( "invalid id" )
}
return key , nil
}
// GenerateSymKey generates a random symmetric key and stores it under id,
// which is then returned. Will be used in the future for session key exchange.
func ( w * Waku ) GenerateSymKey ( ) ( string , error ) {
key , err := common . GenerateSecureRandomData ( common . AESKeyLength )
if err != nil {
return "" , err
} else if ! common . ValidateDataIntegrity ( key , common . AESKeyLength ) {
return "" , fmt . Errorf ( "error in GenerateSymKey: crypto/rand failed to generate random data" )
}
id , err := common . GenerateRandomID ( )
if err != nil {
return "" , fmt . Errorf ( "failed to generate ID: %s" , err )
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . symKeys [ id ] != nil {
return "" , fmt . Errorf ( "failed to generate unique ID" )
}
w . symKeys [ id ] = key
return id , nil
}
// AddSymKey stores the key with a given id.
func ( w * Waku ) AddSymKey ( id string , key [ ] byte ) ( string , error ) {
deterministicID , err := toDeterministicID ( id , common . KeyIDSize )
if err != nil {
return "" , err
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . symKeys [ deterministicID ] != nil {
return "" , fmt . Errorf ( "key already exists: %v" , id )
}
w . symKeys [ deterministicID ] = key
return deterministicID , nil
}
// AddSymKeyDirect stores the key, and returns its id.
func ( w * Waku ) AddSymKeyDirect ( key [ ] byte ) ( string , error ) {
if len ( key ) != common . AESKeyLength {
return "" , fmt . Errorf ( "wrong key size: %d" , len ( key ) )
}
id , err := common . GenerateRandomID ( )
if err != nil {
return "" , fmt . Errorf ( "failed to generate ID: %s" , err )
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . symKeys [ id ] != nil {
return "" , fmt . Errorf ( "failed to generate unique ID" )
}
w . symKeys [ id ] = key
return id , nil
}
// AddSymKeyFromPassword generates the key from password, stores it, and returns its id.
func ( w * Waku ) AddSymKeyFromPassword ( password string ) ( string , error ) {
id , err := common . GenerateRandomID ( )
if err != nil {
return "" , fmt . Errorf ( "failed to generate ID: %s" , err )
}
if w . HasSymKey ( id ) {
return "" , fmt . Errorf ( "failed to generate unique ID" )
}
// kdf should run no less than 0.1 seconds on an average computer,
// because it's an once in a session experience
derived := pbkdf2 . Key ( [ ] byte ( password ) , nil , 65356 , common . AESKeyLength , sha256 . New )
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
// double check is necessary, because deriveKeyMaterial() is very slow
if w . symKeys [ id ] != nil {
return "" , fmt . Errorf ( "critical error: failed to generate unique ID" )
}
w . symKeys [ id ] = derived
return id , nil
}
// HasSymKey returns true if there is a key associated with the given id.
// Otherwise returns false.
func ( w * Waku ) HasSymKey ( id string ) bool {
w . keyMu . RLock ( )
defer w . keyMu . RUnlock ( )
return w . symKeys [ id ] != nil
}
// DeleteSymKey deletes the key associated with the name string if it exists.
func ( w * Waku ) DeleteSymKey ( id string ) bool {
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . symKeys [ id ] != nil {
delete ( w . symKeys , id )
return true
}
return false
}
// GetSymKey returns the symmetric key associated with the given id.
func ( w * Waku ) GetSymKey ( id string ) ( [ ] byte , error ) {
w . keyMu . RLock ( )
defer w . keyMu . RUnlock ( )
if w . symKeys [ id ] != nil {
return w . symKeys [ id ] , nil
}
return nil , fmt . Errorf ( "non-existent key ID" )
}
// Subscribe installs a new message handler used for filtering, decrypting
// and subsequent storing of incoming messages.
func ( w * Waku ) Subscribe ( f * common . Filter ) ( string , error ) {
2023-05-22 21:38:02 +00:00
if f . PubsubTopic == "" {
f . PubsubTopic = relay . DefaultWakuTopic
}
2023-06-07 09:02:19 +00:00
2021-06-16 20:19:45 +00:00
s , err := w . filters . Install ( f )
if err != nil {
return s , err
}
2021-09-23 16:17:57 +00:00
if w . settings . LightClient {
2023-06-15 14:42:54 +00:00
go func ( ) {
ticker := time . NewTicker ( 1 * time . Second )
for range ticker . C {
err := w . subscribeToFilter ( f )
if err == nil {
break
}
}
} ( )
2021-09-23 16:17:57 +00:00
}
2021-06-16 20:19:45 +00:00
return s , nil
}
// Unsubscribe removes an installed message handler.
2023-07-14 13:42:02 +00:00
func ( w * Waku ) Unsubscribe ( ctx context . Context , id string ) error {
2021-10-19 13:43:41 +00:00
f := w . filters . Get ( id )
if f != nil && w . settings . LightClient {
2023-05-22 21:38:02 +00:00
contentFilter := w . buildContentFilter ( f . PubsubTopic , f . Topics )
2023-07-14 13:42:02 +00:00
if _ , err := w . node . FilterLightnode ( ) . Unsubscribe ( ctx , contentFilter ) ; err != nil {
2021-10-19 13:43:41 +00:00
return fmt . Errorf ( "failed to unsubscribe: %w" , err )
2021-09-23 16:17:57 +00:00
}
}
2021-06-16 20:19:45 +00:00
ok := w . filters . Uninstall ( id )
if ! ok {
return fmt . Errorf ( "failed to unsubscribe: invalid ID '%s'" , id )
}
return nil
}
2023-06-07 09:02:19 +00:00
// GetFilter returns the filter by id.
func ( w * Waku ) GetFilter ( id string ) * common . Filter {
return w . filters . Get ( id )
}
2021-06-16 20:19:45 +00:00
// Unsubscribe removes an installed message handler.
func ( w * Waku ) UnsubscribeMany ( ids [ ] string ) error {
for _ , id := range ids {
w . logger . Debug ( "cleaning up filter" , zap . String ( "id" , id ) )
ok := w . filters . Uninstall ( id )
if ! ok {
w . logger . Warn ( "could not remove filter with id" , zap . String ( "id" , id ) )
}
}
return nil
}
2021-11-18 18:31:28 +00:00
func ( w * Waku ) broadcast ( ) {
for {
select {
2023-03-08 15:05:46 +00:00
case envelope := <- w . sendQueue :
var err error
2022-10-04 17:54:13 +00:00
if w . settings . LightClient {
2023-05-22 21:38:02 +00:00
w . logger . Info ( "publishing message via lightpush" , zap . String ( "envelopeHash" , hexutil . Encode ( envelope . Hash ( ) ) ) , zap . String ( "pubsubTopic" , envelope . PubsubTopic ( ) ) )
_ , err = w . node . Lightpush ( ) . PublishToTopic ( context . Background ( ) , envelope . Message ( ) , envelope . PubsubTopic ( ) )
2021-11-18 18:31:28 +00:00
} else {
2023-05-22 21:38:02 +00:00
w . logger . Info ( "publishing message via relay" , zap . String ( "envelopeHash" , hexutil . Encode ( envelope . Hash ( ) ) ) , zap . String ( "pubsubTopic" , envelope . PubsubTopic ( ) ) )
_ , err = w . node . Relay ( ) . PublishToTopic ( context . Background ( ) , envelope . Message ( ) , envelope . PubsubTopic ( ) )
2021-11-18 18:31:28 +00:00
}
if err != nil {
2023-05-22 21:38:02 +00:00
w . logger . Error ( "could not send message" , zap . String ( "envelopeHash" , hexutil . Encode ( envelope . Hash ( ) ) ) , zap . String ( "pubsubTopic" , envelope . PubsubTopic ( ) ) , zap . Error ( err ) )
2021-11-18 18:31:28 +00:00
w . envelopeFeed . Send ( common . EnvelopeEvent {
2023-03-08 15:05:46 +00:00
Hash : gethcommon . BytesToHash ( envelope . Hash ( ) ) ,
2021-11-18 18:31:28 +00:00
Event : common . EventEnvelopeExpired ,
} )
continue
}
event := common . EnvelopeEvent {
Event : common . EventEnvelopeSent ,
2023-03-08 15:05:46 +00:00
Hash : gethcommon . BytesToHash ( envelope . Hash ( ) ) ,
2021-11-18 18:31:28 +00:00
}
w . SendEnvelopeEvent ( event )
2023-07-17 17:20:55 +00:00
case <- w . ctx . Done ( ) :
2021-11-18 18:31:28 +00:00
return
}
2021-11-02 18:27:37 +00:00
}
2021-11-18 18:31:28 +00:00
}
2021-11-02 18:27:37 +00:00
2021-11-18 18:31:28 +00:00
// Send injects a message into the waku send queue, to be distributed in the
// network in the coming cycles.
2023-05-22 21:38:02 +00:00
func ( w * Waku ) Send ( pubsubTopic string , msg * pb . WakuMessage ) ( [ ] byte , error ) {
if pubsubTopic == "" {
pubsubTopic = relay . DefaultWakuTopic
}
if w . protectedTopicStore != nil {
privKey , err := w . protectedTopicStore . FetchPrivateKey ( pubsubTopic )
if err != nil {
return nil , err
}
if privKey != nil {
err = relay . SignMessage ( privKey , msg , pubsubTopic )
if err != nil {
return nil , err
}
}
}
envelope := protocol . NewEnvelope ( msg , msg . Timestamp , pubsubTopic )
2021-11-02 18:27:37 +00:00
2023-03-08 15:05:46 +00:00
w . sendQueue <- envelope
2021-11-18 18:31:28 +00:00
2021-11-02 18:27:37 +00:00
w . poolMu . Lock ( )
2023-03-08 15:05:46 +00:00
_ , alreadyCached := w . envelopes [ gethcommon . BytesToHash ( envelope . Hash ( ) ) ]
2021-11-02 18:27:37 +00:00
w . poolMu . Unlock ( )
if ! alreadyCached {
2021-12-01 15:15:18 +00:00
recvMessage := common . NewReceivedMessage ( envelope , common . RelayedMessageType )
2021-11-18 18:31:28 +00:00
w . postEvent ( recvMessage ) // notify the local node about the new message
2021-11-02 18:27:37 +00:00
w . addEnvelope ( recvMessage )
}
2023-03-08 15:05:46 +00:00
return envelope . Hash ( ) , nil
2021-06-16 20:19:45 +00:00
}
2023-05-22 21:38:02 +00:00
func ( w * Waku ) query ( ctx context . Context , peerID peer . ID , pubsubTopic string , topics [ ] common . TopicType , from uint64 , to uint64 , opts [ ] store . HistoryRequestOption ) ( * store . Result , error ) {
2021-06-16 20:19:45 +00:00
strTopics := make ( [ ] string , len ( topics ) )
for i , t := range topics {
2021-09-01 21:25:31 +00:00
strTopics [ i ] = t . ContentTopic ( )
2021-06-16 20:19:45 +00:00
}
2022-11-24 21:09:17 +00:00
opts = append ( opts , store . WithPeer ( peerID ) )
2021-11-02 18:27:37 +00:00
query := store . Query {
2022-03-02 14:48:51 +00:00
StartTime : int64 ( from ) * int64 ( time . Second ) ,
EndTime : int64 ( to ) * int64 ( time . Second ) ,
2021-11-02 18:27:37 +00:00
ContentTopics : strTopics ,
2023-05-22 21:38:02 +00:00
Topic : pubsubTopic ,
2021-11-02 18:27:37 +00:00
}
2023-02-22 21:58:17 +00:00
return w . node . Store ( ) . Query ( ctx , query , opts ... )
}
2023-05-22 21:38:02 +00:00
func ( w * Waku ) Query ( ctx context . Context , peerID peer . ID , pubsubTopic string , topics [ ] common . TopicType , from uint64 , to uint64 , opts [ ] store . HistoryRequestOption ) ( cursor * storepb . Index , err error ) {
2023-02-22 21:58:17 +00:00
requestID := protocol . GenerateRequestId ( )
opts = append ( opts , store . WithRequestId ( requestID ) )
2023-05-22 21:38:02 +00:00
result , err := w . query ( ctx , peerID , pubsubTopic , topics , from , to , opts )
2023-01-26 20:31:27 +00:00
if err != nil {
2023-01-26 16:29:00 +00:00
w . logger . Error ( "error querying storenode" , zap . String ( "requestID" , hexutil . Encode ( requestID ) ) , zap . String ( "peerID" , peerID . String ( ) ) , zap . Error ( err ) )
2023-07-05 15:56:34 +00:00
if w . onHistoricMessagesRequestFailed != nil {
w . onHistoricMessagesRequestFailed ( requestID , peerID , err )
}
2022-11-24 21:09:17 +00:00
return nil , err
2021-11-02 18:27:37 +00:00
}
2021-06-16 20:19:45 +00:00
for _ , msg := range result . Messages {
2022-12-20 17:59:59 +00:00
// Temporarily setting RateLimitProof to nil so it matches the WakuMessage protobuffer we are sending
// See https://github.com/vacp2p/rfc/issues/563
msg . RateLimitProof = nil
2023-05-22 21:38:02 +00:00
envelope := protocol . NewEnvelope ( msg , msg . Timestamp , pubsubTopic )
w . logger . Info ( "received waku2 store message" , zap . Any ( "envelopeHash" , hexutil . Encode ( envelope . Hash ( ) ) ) , zap . String ( "pubsubTopic" , pubsubTopic ) )
2021-12-01 15:15:18 +00:00
_ , err = w . OnNewEnvelopes ( envelope , common . StoreMessageType )
2021-06-16 20:19:45 +00:00
if err != nil {
2021-07-21 19:02:50 +00:00
return nil , err
2021-06-16 20:19:45 +00:00
}
}
2022-10-03 21:57:43 +00:00
if ! result . IsComplete ( ) {
2021-11-02 18:27:37 +00:00
cursor = result . Cursor ( )
2021-07-21 19:02:50 +00:00
}
return
2021-06-16 20:19:45 +00:00
}
// Start implements node.Service, starting the background data propagation thread
// of the Waku protocol.
2021-06-30 11:40:54 +00:00
func ( w * Waku ) Start ( ) error {
2023-04-14 16:08:06 +00:00
var err error
if w . node , err = node . New ( w . settings . Options ... ) ; err != nil {
return fmt . Errorf ( "failed to create a go-waku node: %v" , err )
}
2023-04-17 15:45:45 +00:00
w . connectionChanged = make ( chan struct { } )
2023-07-17 17:20:55 +00:00
ctx , cancel := context . WithCancel ( context . Background ( ) )
w . ctx = ctx
w . cancel = cancel
2023-04-14 16:08:06 +00:00
if err = w . node . Start ( ctx ) ; err != nil {
return fmt . Errorf ( "failed to start go-waku node: %v" , err )
}
2023-08-25 17:02:13 +00:00
w . logger . Info ( "WakuV2 PeerID" , zap . Stringer ( "id" , w . node . Host ( ) . ID ( ) ) )
2023-05-08 21:43:27 +00:00
idService , err := identify . NewIDService ( w . node . Host ( ) )
if err != nil {
return err
}
w . identifyService = idService
2023-04-14 16:08:06 +00:00
if err = w . addWakuV2Peers ( ctx , w . cfg ) ; err != nil {
return fmt . Errorf ( "failed to add wakuv2 peers: %v" , err )
}
if w . cfg . EnableDiscV5 {
err := w . node . DiscV5 ( ) . Start ( ctx )
if err != nil {
return err
}
}
2023-05-22 21:38:02 +00:00
w . wg . Add ( 3 )
2023-04-14 16:08:06 +00:00
go func ( ) {
defer w . wg . Done ( )
isConnected := false
for {
select {
2023-07-17 17:20:55 +00:00
case <- w . ctx . Done ( ) :
2023-04-14 16:08:06 +00:00
return
case c := <- w . connStatusChan :
w . connStatusMu . Lock ( )
latestConnStatus := formatConnStatus ( w . node , c )
for k , subs := range w . connStatusSubscriptions {
if subs . Active ( ) {
subs . C <- latestConnStatus
} else {
delete ( w . connStatusSubscriptions , k )
}
}
w . connStatusMu . Unlock ( )
2023-07-05 15:56:34 +00:00
if w . onPeerStats != nil {
w . onPeerStats ( latestConnStatus )
}
2023-04-14 16:08:06 +00:00
if w . cfg . EnableDiscV5 {
// Restarting DiscV5
if ! latestConnStatus . IsOnline && isConnected {
w . logger . Debug ( "Restarting DiscV5: offline and is connected" )
isConnected = false
w . node . DiscV5 ( ) . Stop ( )
} else if latestConnStatus . IsOnline && ! isConnected {
w . logger . Debug ( "Restarting DiscV5: online and is not connected" )
isConnected = true
if ! w . node . DiscV5 ( ) . IsStarted ( ) {
err := w . node . DiscV5 ( ) . Start ( ctx )
if err != nil {
w . logger . Error ( "Could not start DiscV5" , zap . Error ( err ) )
}
}
}
}
}
}
} ( )
go w . telemetryBandwidthStats ( w . cfg . TelemetryServerURL )
go w . runPeerExchangeLoop ( )
2023-05-22 21:38:02 +00:00
go w . runFilterMsgLoop ( )
err = w . setupRelaySubscriptions ( )
if err != nil {
return err
}
2023-04-14 16:08:06 +00:00
2021-06-16 20:19:45 +00:00
numCPU := runtime . NumCPU ( )
for i := 0 ; i < numCPU ; i ++ {
go w . processQueue ( )
}
2021-11-18 18:31:28 +00:00
go w . broadcast ( )
2022-12-09 16:16:21 +00:00
go w . seedBootnodesForDiscV5 ( )
2021-11-18 18:31:28 +00:00
2021-06-16 20:19:45 +00:00
return nil
}
2023-05-22 21:38:02 +00:00
func ( w * Waku ) setupRelaySubscriptions ( ) error {
if w . settings . LightClient {
return nil
}
if w . protectedTopicStore != nil {
protectedTopics , err := w . protectedTopicStore . ProtectedTopics ( )
if err != nil {
return err
}
for _ , pt := range protectedTopics {
// Adding subscription to protected topics
err = w . subscribeToPubsubTopicWithWakuRelay ( pt . Topic , pt . PubKey )
if err != nil {
return err
}
}
}
// Default Waku Topic (TODO: remove once sharding is added)
err := w . subscribeToPubsubTopicWithWakuRelay ( relay . DefaultWakuTopic , nil )
if err != nil {
return err
}
return nil
}
2021-06-16 20:19:45 +00:00
// Stop implements node.Service, stopping the background data propagation thread
// of the Waku protocol.
func ( w * Waku ) Stop ( ) error {
2023-07-17 17:20:55 +00:00
w . cancel ( )
2023-05-22 21:38:02 +00:00
err := w . identifyService . Close ( )
if err != nil {
return err
}
2021-06-16 20:19:45 +00:00
w . node . Stop ( )
2023-05-22 21:38:02 +00:00
if w . protectedTopicStore != nil {
err = w . protectedTopicStore . Close ( )
if err != nil {
return err
}
}
2022-12-09 16:16:21 +00:00
close ( w . connectionChanged )
2021-11-22 15:27:05 +00:00
w . wg . Wait ( )
2021-06-16 20:19:45 +00:00
return nil
}
2021-11-22 15:27:05 +00:00
func ( w * Waku ) OnNewEnvelopes ( envelope * protocol . Envelope , msgType common . MessageType ) ( [ ] common . EnvelopeError , error ) {
2023-02-08 14:58:10 +00:00
if envelope == nil {
2023-05-08 21:43:27 +00:00
return nil , nil
2023-02-08 14:58:10 +00:00
}
2023-02-08 12:10:07 +00:00
2021-12-01 15:15:18 +00:00
recvMessage := common . NewReceivedMessage ( envelope , msgType )
2023-02-24 13:32:13 +00:00
if recvMessage == nil {
return nil , nil
}
2021-06-16 20:19:45 +00:00
envelopeErrors := make ( [ ] common . EnvelopeError , 0 )
2022-11-07 15:25:50 +00:00
logger := w . logger . With ( zap . String ( "hash" , recvMessage . Hash ( ) . Hex ( ) ) )
logger . Debug ( "received new envelope" )
2021-06-16 20:19:45 +00:00
trouble := false
_ , err := w . add ( recvMessage )
if err != nil {
2022-11-07 15:25:50 +00:00
logger . Info ( "invalid envelope received" , zap . Error ( err ) )
2022-11-04 13:56:45 +00:00
trouble = true
2021-06-16 20:19:45 +00:00
}
common . EnvelopesValidatedCounter . Inc ( )
if trouble {
return envelopeErrors , errors . New ( "received invalid envelope" )
}
return envelopeErrors , nil
}
// addEnvelope adds an envelope to the envelope map, used for sending
func ( w * Waku ) addEnvelope ( envelope * common . ReceivedMessage ) {
hash := envelope . Hash ( )
w . poolMu . Lock ( )
w . envelopes [ hash ] = envelope
w . poolMu . Unlock ( )
}
func ( w * Waku ) add ( recvMessage * common . ReceivedMessage ) ( bool , error ) {
common . EnvelopesReceivedCounter . Inc ( )
hash := recvMessage . Hash ( )
w . poolMu . Lock ( )
_ , alreadyCached := w . envelopes [ hash ]
w . poolMu . Unlock ( )
if ! alreadyCached {
w . addEnvelope ( recvMessage )
}
if alreadyCached {
2022-11-04 13:56:45 +00:00
w . logger . Debug ( "w envelope already cached" , zap . String ( "envelopeHash" , recvMessage . Hash ( ) . Hex ( ) ) )
2021-06-16 20:19:45 +00:00
common . EnvelopesCachedCounter . WithLabelValues ( "hit" ) . Inc ( )
} else {
2022-11-04 13:56:45 +00:00
w . logger . Debug ( "cached w envelope" , zap . String ( "envelopeHash" , recvMessage . Hash ( ) . Hex ( ) ) )
2021-06-16 20:19:45 +00:00
common . EnvelopesCachedCounter . WithLabelValues ( "miss" ) . Inc ( )
2023-03-08 15:05:46 +00:00
common . EnvelopesSizeMeter . Observe ( float64 ( len ( recvMessage . Envelope . Message ( ) . Payload ) ) )
2021-06-16 20:19:45 +00:00
w . postEvent ( recvMessage ) // notify the local node about the new message
}
return true , nil
}
// postEvent queues the message for further processing.
func ( w * Waku ) postEvent ( envelope * common . ReceivedMessage ) {
w . msgQueue <- envelope
}
// processQueue delivers the messages to the watchers during the lifetime of the waku node.
func ( w * Waku ) processQueue ( ) {
for {
select {
2023-07-17 17:20:55 +00:00
case <- w . ctx . Done ( ) :
2021-06-16 20:19:45 +00:00
return
case e := <- w . msgQueue :
2021-12-01 15:15:18 +00:00
if e . MsgType == common . StoreMessageType {
// We need to insert it first, and then remove it if not matched,
// as messages are processed asynchronously
w . storeMsgIDsMu . Lock ( )
w . storeMsgIDs [ e . Hash ( ) ] = true
w . storeMsgIDsMu . Unlock ( )
}
matched := w . filters . NotifyWatchers ( e )
// If not matched we remove it
if ! matched {
2023-05-22 21:38:02 +00:00
w . logger . Debug ( "filters did not match" , zap . String ( "hash" , e . Hash ( ) . String ( ) ) , zap . String ( "contentTopic" , e . ContentTopic . ContentTopic ( ) ) )
2021-12-01 15:15:18 +00:00
w . storeMsgIDsMu . Lock ( )
delete ( w . storeMsgIDs , e . Hash ( ) )
w . storeMsgIDsMu . Unlock ( )
}
2021-06-16 20:19:45 +00:00
w . envelopeFeed . Send ( common . EnvelopeEvent {
2023-05-22 21:38:02 +00:00
Topic : e . ContentTopic ,
2021-06-16 20:19:45 +00:00
Hash : e . Hash ( ) ,
Event : common . EventEnvelopeAvailable ,
} )
}
}
}
// Envelopes retrieves all the messages currently pooled by the node.
func ( w * Waku ) Envelopes ( ) [ ] * common . ReceivedMessage {
w . poolMu . RLock ( )
defer w . poolMu . RUnlock ( )
all := make ( [ ] * common . ReceivedMessage , 0 , len ( w . envelopes ) )
for _ , envelope := range w . envelopes {
all = append ( all , envelope )
}
return all
}
// GetEnvelope retrieves an envelope from the message queue by its hash.
// It returns nil if the envelope can not be found.
func ( w * Waku ) GetEnvelope ( hash gethcommon . Hash ) * common . ReceivedMessage {
w . poolMu . RLock ( )
defer w . poolMu . RUnlock ( )
return w . envelopes [ hash ]
}
// isEnvelopeCached checks if envelope with specific hash has already been received and cached.
func ( w * Waku ) IsEnvelopeCached ( hash gethcommon . Hash ) bool {
w . poolMu . Lock ( )
defer w . poolMu . Unlock ( )
_ , exist := w . envelopes [ hash ]
return exist
}
2021-08-30 14:57:28 +00:00
func ( w * Waku ) PeerCount ( ) int {
return w . node . PeerCount ( )
}
2022-11-24 15:00:44 +00:00
func ( w * Waku ) Peers ( ) map [ string ] types . WakuV2Peer {
return FormatPeerStats ( w . node , w . node . PeerStats ( ) )
2021-11-22 13:40:14 +00:00
}
2022-11-24 21:27:46 +00:00
func ( w * Waku ) ListenAddresses ( ) [ ] string {
addrs := w . node . ListenAddresses ( )
var result [ ] string
for _ , addr := range addrs {
result = append ( result , addr . String ( ) )
}
return result
}
2023-05-22 21:38:02 +00:00
func ( w * Waku ) SubscribeToPubsubTopic ( topic string , pubkey * ecdsa . PublicKey ) error {
if ! w . settings . LightClient {
err := w . subscribeToPubsubTopicWithWakuRelay ( topic , pubkey )
if err != nil {
return err
}
}
return nil
}
func ( w * Waku ) StorePubsubTopicKey ( topic string , privKey * ecdsa . PrivateKey ) error {
return w . protectedTopicStore . Insert ( topic , privKey , & privKey . PublicKey )
}
2021-11-22 13:40:14 +00:00
func ( w * Waku ) StartDiscV5 ( ) error {
if w . node . DiscV5 ( ) == nil {
return errors . New ( "discv5 is not setup" )
}
2023-07-17 17:20:55 +00:00
return w . node . DiscV5 ( ) . Start ( w . ctx )
2021-11-22 13:40:14 +00:00
}
func ( w * Waku ) StopDiscV5 ( ) error {
if w . node . DiscV5 ( ) == nil {
return errors . New ( "discv5 is not setup" )
}
w . node . DiscV5 ( ) . Stop ( )
return nil
2021-08-30 14:57:28 +00:00
}
2022-12-09 16:16:21 +00:00
func ( w * Waku ) ConnectionChanged ( state connection . State ) {
if ! state . Offline && w . offline {
select {
case w . connectionChanged <- struct { } { } :
2023-01-13 17:12:46 +00:00
default :
w . logger . Warn ( "could not write on connection changed channel" )
2022-12-09 16:16:21 +00:00
}
}
w . offline = ! state . Offline
}
// seedBootnodesForDiscV5 tries to fetch bootnodes
// from an ENR periodically.
// It backs off exponentially until maxRetries, at which point it restarts from 0
// It also restarts if there's a connection change signalled from the client
func ( w * Waku ) seedBootnodesForDiscV5 ( ) {
2023-02-24 13:29:58 +00:00
if ! w . settings . EnableDiscV5 || w . node . DiscV5 ( ) == nil {
return
}
2023-07-03 12:48:21 +00:00
ticker := time . NewTicker ( 500 * time . Millisecond )
2022-12-09 16:16:21 +00:00
defer ticker . Stop ( )
var retries = 0
2023-07-03 12:48:21 +00:00
now := func ( ) int64 {
return time . Now ( ) . UnixNano ( ) / int64 ( time . Millisecond )
}
var lastTry = now ( )
canQuery := func ( ) bool {
backoff := bootnodesQueryBackoffMs * int64 ( math . Exp2 ( float64 ( retries ) ) )
return lastTry + backoff < now ( )
}
2022-12-09 16:16:21 +00:00
for {
select {
case <- ticker . C :
2023-07-03 12:48:21 +00:00
if w . seededBootnodesForDiscV5 && len ( w . node . Host ( ) . Network ( ) . Peers ( ) ) > 3 {
w . logger . Info ( "not querying bootnodes" , zap . Bool ( "seeded" , w . seededBootnodesForDiscV5 ) , zap . Int ( "peer-count" , len ( w . node . Host ( ) . Network ( ) . Peers ( ) ) ) )
continue
2022-12-09 16:16:21 +00:00
}
2023-07-03 12:48:21 +00:00
if canQuery ( ) {
w . logger . Info ( "querying bootnodes" , zap . Int ( "peer-count" , len ( w . node . Host ( ) . Network ( ) . Peers ( ) ) ) )
w . logger . Info ( "querying bootnodes to restore connectivity" )
2022-12-09 16:16:21 +00:00
err := w . restartDiscV5 ( )
if err != nil {
w . logger . Warn ( "failed to restart discv5" , zap . Error ( err ) )
}
2023-07-03 12:48:21 +00:00
lastTry = now ( )
2022-12-09 16:16:21 +00:00
retries ++
// We reset the retries after a while and restart
if retries > bootnodesMaxRetries {
retries = 0
}
2023-07-03 12:48:21 +00:00
} else {
w . logger . Info ( "can't query bootnodes" , zap . Int ( "peer-count" , len ( w . node . Host ( ) . Network ( ) . Peers ( ) ) ) )
w . logger . Info ( "can't query" , zap . Int64 ( "lastTry" , lastTry ) , zap . Int64 ( "now" , now ( ) ) , zap . Int64 ( "backoff" , bootnodesQueryBackoffMs * int64 ( math . Exp2 ( float64 ( retries ) ) ) ) , zap . Int ( "retries" , retries ) )
2022-12-09 16:16:21 +00:00
}
// If we go online, trigger immediately
case <- w . connectionChanged :
2023-07-03 12:48:21 +00:00
if canQuery ( ) {
2022-12-09 16:16:21 +00:00
err := w . restartDiscV5 ( )
if err != nil {
w . logger . Warn ( "failed to restart discv5" , zap . Error ( err ) )
}
}
retries = 0
2023-07-03 12:48:21 +00:00
lastTry = now ( )
2022-12-09 16:16:21 +00:00
2023-07-17 17:20:55 +00:00
case <- w . ctx . Done ( ) :
2022-12-09 16:16:21 +00:00
return
}
}
}
// Restart discv5, re-retrieving bootstrap nodes
func ( w * Waku ) restartDiscV5 ( ) error {
2023-07-17 17:20:55 +00:00
ctx , cancel := context . WithTimeout ( w . ctx , 30 * time . Second )
2022-12-09 16:16:21 +00:00
defer cancel ( )
bootnodes , err := w . getDiscV5BootstrapNodes ( ctx , w . discV5BootstrapNodes )
if err != nil {
return err
}
if len ( bootnodes ) == 0 {
return errors . New ( "failed to fetch bootnodes" )
}
2023-07-03 12:48:21 +00:00
if ! w . node . DiscV5 ( ) . IsStarted ( ) {
w . logger . Info ( "is not started restarting" )
err := w . node . DiscV5 ( ) . Start ( ctx )
if err != nil {
w . logger . Error ( "Could not start DiscV5" , zap . Error ( err ) )
}
} else {
w . node . DiscV5 ( ) . Stop ( )
w . logger . Info ( "is started restarting" )
2023-07-17 17:20:55 +00:00
select {
case <- w . ctx . Done ( ) : // Don't start discv5 if we are stopping waku
return nil
default :
}
2023-07-03 12:48:21 +00:00
err := w . node . DiscV5 ( ) . Start ( ctx )
if err != nil {
w . logger . Error ( "Could not start DiscV5" , zap . Error ( err ) )
}
}
2022-12-09 16:16:21 +00:00
w . logger . Info ( "restarting discv5 with nodes" , zap . Any ( "nodes" , bootnodes ) )
return w . node . SetDiscV5Bootnodes ( bootnodes )
}
2023-02-22 21:58:17 +00:00
func ( w * Waku ) AddStorePeer ( address string ) ( peer . ID , error ) {
2021-10-06 16:08:54 +00:00
addr , err := multiaddr . NewMultiaddr ( address )
if err != nil {
return "" , err
}
2023-06-30 13:41:32 +00:00
peerID , err := w . node . AddPeer ( addr , peers . Static , store . StoreID_v20beta4 )
2021-09-10 17:06:06 +00:00
if err != nil {
return "" , err
}
2023-02-22 21:58:17 +00:00
return peerID , nil
2021-08-30 14:57:28 +00:00
}
2022-12-06 15:24:01 +00:00
func ( w * Waku ) timestamp ( ) int64 {
2023-03-24 12:05:42 +00:00
return w . timesource . Now ( ) . UnixNano ( )
2022-12-06 15:24:01 +00:00
}
2023-02-22 21:58:17 +00:00
func ( w * Waku ) AddRelayPeer ( address string ) ( peer . ID , error ) {
2021-10-06 16:08:54 +00:00
addr , err := multiaddr . NewMultiaddr ( address )
if err != nil {
return "" , err
}
2023-06-30 13:41:32 +00:00
peerID , err := w . node . AddPeer ( addr , peers . Static , relay . WakuRelayID_v200 )
2021-09-10 17:06:06 +00:00
if err != nil {
return "" , err
}
2023-02-22 21:58:17 +00:00
return peerID , nil
2021-09-10 17:06:06 +00:00
}
func ( w * Waku ) DialPeer ( address string ) error {
2023-07-17 17:20:55 +00:00
ctx , cancel := context . WithTimeout ( w . ctx , requestTimeout )
2021-09-30 23:17:17 +00:00
defer cancel ( )
return w . node . DialPeer ( ctx , address )
2021-09-10 17:06:06 +00:00
}
func ( w * Waku ) DialPeerByID ( peerID string ) error {
2023-07-17 17:20:55 +00:00
ctx , cancel := context . WithTimeout ( w . ctx , requestTimeout )
2021-09-30 23:17:17 +00:00
defer cancel ( )
2023-01-14 13:49:20 +00:00
pid , err := peer . Decode ( peerID )
if err != nil {
return err
}
return w . node . DialPeerByID ( ctx , pid )
2021-08-30 14:57:28 +00:00
}
func ( w * Waku ) DropPeer ( peerID string ) error {
2023-01-14 13:49:20 +00:00
pid , err := peer . Decode ( peerID )
if err != nil {
return err
}
return w . node . ClosePeerById ( pid )
2021-08-30 14:57:28 +00:00
}
2021-12-01 15:15:18 +00:00
func ( w * Waku ) ProcessingP2PMessages ( ) bool {
w . storeMsgIDsMu . Lock ( )
defer w . storeMsgIDsMu . Unlock ( )
return len ( w . storeMsgIDs ) != 0
}
func ( w * Waku ) MarkP2PMessageAsProcessed ( hash gethcommon . Hash ) {
w . storeMsgIDsMu . Lock ( )
defer w . storeMsgIDsMu . Unlock ( )
delete ( w . storeMsgIDs , hash )
}
2021-06-16 20:19:45 +00:00
// validatePrivateKey checks the format of the given private key.
func validatePrivateKey ( k * ecdsa . PrivateKey ) bool {
if k == nil || k . D == nil || k . D . Sign ( ) == 0 {
return false
}
return common . ValidatePublicKey ( & k . PublicKey )
}
// makeDeterministicID generates a deterministic ID, based on a given input
func makeDeterministicID ( input string , keyLen int ) ( id string , err error ) {
buf := pbkdf2 . Key ( [ ] byte ( input ) , nil , 4096 , keyLen , sha256 . New )
if ! common . ValidateDataIntegrity ( buf , common . KeyIDSize ) {
return "" , fmt . Errorf ( "error in GenerateDeterministicID: failed to generate key" )
}
id = gethcommon . Bytes2Hex ( buf )
return id , err
}
// toDeterministicID reviews incoming id, and transforms it to format
// expected internally be private key store. Originally, public keys
// were used as keys, now random keys are being used. And in order to
// make it easier to consume, we now allow both random IDs and public
// keys to be passed.
func toDeterministicID ( id string , expectedLen int ) ( string , error ) {
if len ( id ) != ( expectedLen * 2 ) { // we received hex key, so number of chars in id is doubled
var err error
id , err = makeDeterministicID ( id , expectedLen )
if err != nil {
return "" , err
}
}
return id , nil
}
2021-08-30 14:57:28 +00:00
2022-11-24 15:00:44 +00:00
func FormatPeerStats ( wakuNode * node . WakuNode , peers node . PeerStats ) map [ string ] types . WakuV2Peer {
p := make ( map [ string ] types . WakuV2Peer )
2021-08-30 14:57:28 +00:00
for k , v := range peers {
2022-11-24 15:00:44 +00:00
peerInfo := wakuNode . Host ( ) . Peerstore ( ) . PeerInfo ( k )
wakuV2Peer := types . WakuV2Peer { }
wakuV2Peer . Protocols = v
hostInfo , _ := multiaddr . NewMultiaddr ( fmt . Sprintf ( "/p2p/%s" , k . Pretty ( ) ) )
for _ , addr := range peerInfo . Addrs {
wakuV2Peer . Addresses = append ( wakuV2Peer . Addresses , addr . Encapsulate ( hostInfo ) . String ( ) )
}
p [ k . Pretty ( ) ] = wakuV2Peer
2021-08-30 14:57:28 +00:00
}
return p
}
2022-11-24 15:00:44 +00:00
func formatConnStatus ( wakuNode * node . WakuNode , c node . ConnStatus ) types . ConnStatus {
2022-01-12 16:02:01 +00:00
return types . ConnStatus {
2021-08-30 14:57:28 +00:00
IsOnline : c . IsOnline ,
HasHistory : c . HasHistory ,
2022-11-24 15:00:44 +00:00
Peers : FormatPeerStats ( wakuNode , c . Peers ) ,
2021-08-30 14:57:28 +00:00
}
}
2023-06-07 09:02:19 +00:00
// Find suitable peer(s). For this we use a peerDisconnectMap, it works so that
// peers that have been recently disconnected from have lower priority
func ( w * Waku ) findFilterPeers ( ) [ ] peer . ID {
allPeers := w . node . Host ( ) . Peerstore ( ) . Peers ( )
var peers peer . IDSlice
for _ , peer := range allPeers {
protocols , err := w . node . Host ( ) . Peerstore ( ) . SupportsProtocols ( peer , filter . FilterSubscribeID_v20beta1 , relay . WakuRelayID_v200 )
if err != nil {
continue
}
if len ( protocols ) == 2 {
peers = append ( peers , peer )
}
}
if len ( peers ) > 0 {
sort . Slice ( peers , func ( i , j int ) bool {
// If element not found in map, [] operator will return 0
return w . filterPeerDisconnectMap [ peers [ i ] ] < w . filterPeerDisconnectMap [ peers [ j ] ]
} )
}
var peerLen = len ( peers )
if w . settings . MinPeersForFilter < peerLen {
peerLen = w . settings . MinPeersForFilter
}
peers = peers [ 0 : peerLen ]
return peers
}
func ( w * Waku ) subscribeToFilter ( f * common . Filter ) error {
peers := w . findFilterPeers ( )
if len ( peers ) > 0 {
2023-05-22 21:38:02 +00:00
contentFilter := w . buildContentFilter ( f . PubsubTopic , f . Topics )
2023-06-07 09:02:19 +00:00
for i := 0 ; i < len ( peers ) && i < w . settings . MinPeersForFilter ; i ++ {
2023-07-17 17:20:55 +00:00
subDetails , err := w . node . FilterLightnode ( ) . Subscribe ( w . ctx , contentFilter , filter . WithPeer ( peers [ i ] ) )
2023-06-07 09:02:19 +00:00
if err != nil {
2023-08-25 17:02:13 +00:00
w . logger . Warn ( "could not add wakuv2 filter for peer" , zap . Stringer ( "peer" , peers [ i ] ) )
2023-06-07 09:02:19 +00:00
continue
}
subMap := w . filterSubscriptions [ f ]
if subMap == nil {
subMap = make ( map [ string ] * filter . SubscriptionDetails )
w . filterSubscriptions [ f ] = subMap
}
subMap [ subDetails . ID ] = subDetails
go w . runFilterSubscriptionLoop ( subDetails )
2023-08-25 17:02:13 +00:00
w . logger . Info ( "wakuv2 filter subscription success" , zap . Stringer ( "peer" , peers [ i ] ) , zap . String ( "pubsubTopic" , contentFilter . Topic ) , zap . Strings ( "contentTopics" , contentFilter . ContentTopics ) )
2023-06-07 09:02:19 +00:00
}
} else {
return errors . New ( "could not select a suitable peer for filter" )
}
return nil
}