refactor(nwaku): ping

status-im · Oct 24, 2024 · ff3ebe6 · ff3ebe6
1 parent 275c1f3
commit ff3ebe6
Show file tree

Hide file tree

Showing 9 changed files with 93 additions and 34 deletions.
diff --git a/eth-node/bridge/geth/wakuv2.go b/eth-node/bridge/geth/wakuv2.go
@@ -331,7 +331,9 @@ func (w *gethWakuV2Wrapper) OnStorenodeAvailable() <-chan peer.ID {
 }
 
 func (w *gethWakuV2Wrapper) WaitForAvailableStoreNode(timeout time.Duration) bool {
-	return w.waku.StorenodeCycle.WaitForAvailableStoreNode(context.TODO(), timeout)
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	return w.waku.StorenodeCycle.WaitForAvailableStoreNode(ctx)
 }
 
 func (w *gethWakuV2Wrapper) SetStorenodeConfigProvider(c history.StorenodeConfigProvider) {

diff --git a/go.mod b/go.mod
@@ -95,7 +95,7 @@ require (
 	github.com/schollz/peerdiscovery v1.7.0
 	github.com/siphiuel/lc-proxy-wrapper v0.0.0-20230516150924-246507cee8c7
 	github.com/urfave/cli/v2 v2.27.2
-	github.com/waku-org/go-waku v0.8.1-0.20241022133615-5dc634be1056
+	github.com/waku-org/go-waku v0.8.1-0.20241024184757-fdb3c3d0b35a
 	github.com/wk8/go-ordered-map/v2 v2.1.7
 	github.com/yeqown/go-qrcode/v2 v2.2.1
 	github.com/yeqown/go-qrcode/writer/standard v1.2.1

diff --git a/go.sum b/go.sum
@@ -2136,8 +2136,8 @@ github.com/waku-org/go-libp2p-pubsub v0.12.0-gowaku.0.20240823143342-b0f2429ca27
 github.com/waku-org/go-libp2p-pubsub v0.12.0-gowaku.0.20240823143342-b0f2429ca27f/go.mod h1:Oi0zw9aw8/Y5GC99zt+Ef2gYAl+0nZlwdJonDyOz/sE=
 github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0 h1:R4YYx2QamhBRl/moIxkDCNW+OP7AHbyWLBygDc/xIMo=
 github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0/go.mod h1:EhZP9fee0DYjKH/IOQvoNSy1tSHp2iZadsHGphcAJgY=
-github.com/waku-org/go-waku v0.8.1-0.20241022133615-5dc634be1056 h1:R2LscQHxKdVVdRIz7zcZWOkjcZDz753fflW5TPunJN0=
-github.com/waku-org/go-waku v0.8.1-0.20241022133615-5dc634be1056/go.mod h1:1BRnyg2mQ2aBNLTBaPq6vEvobzywGykPOhGQFbHGf74=
+github.com/waku-org/go-waku v0.8.1-0.20241024184757-fdb3c3d0b35a h1:epN2bp1mPzdg3S7S2iR72GsUPix7irc3UgM4W9NZJpU=
+github.com/waku-org/go-waku v0.8.1-0.20241024184757-fdb3c3d0b35a/go.mod h1:1BRnyg2mQ2aBNLTBaPq6vEvobzywGykPOhGQFbHGf74=
 github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59 h1:jisj+OCI6QydLtFq3Pyhu49wl9ytPN7oAHjMfepHDrA=
 github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59/go.mod h1:1PdBdPzyTaKt3VnpAHk3zj+r9dXPFOr3IHZP9nFle6E=
 github.com/waku-org/go-zerokit-rln-apple v0.0.0-20230916172309-ee0ee61dde2b h1:KgZVhsLkxsj5gb/FfndSCQu6VYwALrCOgYI3poR95yE=

diff --git a/vendor/github.com/waku-org/go-waku/waku/v2/api/common/pinger.go b/vendor/github.com/waku-org/go-waku/waku/v2/api/common/pinger.go
diff --git a/.../waku/v2/api/common/storenodeRequestor.go → ...waku/v2/api/common/storenode_requestor.go b/.../waku/v2/api/common/storenodeRequestor.go → ...waku/v2/api/common/storenode_requestor.go
diff --git a/vendor/github.com/waku-org/go-waku/waku/v2/api/history/cycle.go b/vendor/github.com/waku-org/go-waku/waku/v2/api/history/cycle.go
diff --git a/vendor/modules.txt b/vendor/modules.txt
@@ -1015,7 +1015,7 @@ github.com/waku-org/go-discover/discover/v5wire
 github.com/waku-org/go-libp2p-rendezvous
 github.com/waku-org/go-libp2p-rendezvous/db
 github.com/waku-org/go-libp2p-rendezvous/pb
-# github.com/waku-org/go-waku v0.8.1-0.20241022133615-5dc634be1056
+# github.com/waku-org/go-waku v0.8.1-0.20241024184757-fdb3c3d0b35a
 ## explicit; go 1.21
 github.com/waku-org/go-waku/logging
 github.com/waku-org/go-waku/tests

diff --git a/wakuv2/gowaku.go b/wakuv2/gowaku.go
@@ -75,6 +75,8 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/protocol/store"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 
+	commonapi "github.com/waku-org/go-waku/waku/v2/api/common"
+
 	gocommon "github.com/status-im/status-go/common"
 	"github.com/status-im/status-go/connection"
 	"github.com/status-im/status-go/eth-node/types"
@@ -1055,10 +1057,10 @@ func (w *Waku) Start() error {
 		return fmt.Errorf("failed to start go-waku node: %v", err)
 	}
 
-	w.StorenodeCycle = history.NewStorenodeCycle(w.logger)
-	w.HistoryRetriever = history.NewHistoryRetriever(w.node.Store(), NewHistoryProcessorWrapper(w), w.logger)
+	w.StorenodeCycle = history.NewStorenodeCycle(w.logger, commonapi.NewDefaultPinger(w.node.Host()))
+	w.HistoryRetriever = history.NewHistoryRetriever(missing.NewDefaultStorenodeRequestor(w.node.Store()), NewHistoryProcessorWrapper(w), w.logger)
 
-	w.StorenodeCycle.Start(w.ctx, w.node.Host())
+	w.StorenodeCycle.Start(w.ctx)
 
 	w.logger.Info("WakuV2 PeerID", zap.Stringer("id", w.node.Host().ID()))
 

diff --git a/wakuv2/nwaku.go b/wakuv2/nwaku.go
@@ -208,6 +208,10 @@ package wakuv2
 		WAKU_CALL (waku_get_my_enr(ctx, (WakuCallBack) callback, resp) );
 	}
 
+	static void cGoWakuPingPeer(void* ctx, char* peerAddr, int timeoutMs, void* resp) {
+		WAKU_CALL (waku_ping_peer(ctx, peerAddr, timeoutMs, (WakuCallBack) callback, resp) );
+	}
+
 	static void cGoWakuListPeersInMesh(void* ctx, char* pubSubTopic, void* resp) {
 		WAKU_CALL (waku_relay_get_num_peers_in_mesh(ctx, pubSubTopic, (WakuCallBack) callback, resp) );
 	}
@@ -1360,10 +1364,10 @@ func (w *Waku) Start() error {
 		return fmt.Errorf("failed to start go-waku node: %v", err)
 	}
 	*/
-	w.StorenodeCycle = history.NewStorenodeCycle(w.logger)
+	w.StorenodeCycle = history.NewStorenodeCycle(w.logger, newPinger(w.wakuCtx))
 
 	w.HistoryRetriever = history.NewHistoryRetriever(newStorenodeRequestor(w.wakuCtx, w.logger), NewHistoryProcessorWrapper(w), w.logger)
-	w.StorenodeCycle.Start(w.ctx, newPinger(w.wakuCtx))
+	w.StorenodeCycle.Start(w.ctx)
 
 	w.logger.Info("WakuV2 PeerID", zap.Stringer("id", w.PeerID()))
 
@@ -3007,18 +3011,48 @@ func (d *storenodeMessageVerifier) MessageHashesExist(ctx context.Context, reque
 	return result, nil
 }
 
-func newStorenodeRequestor(wakuCtx unsafe.Pointer, logger *zap.Logger) commonapi.StorenodeRequestor {
-	return &storenodeRequestor{
+type pinger struct {
+	wakuCtx unsafe.Pointer
+}
+
+func newPinger(wakuCtx unsafe.Pointer) commonapi.Pinger {
+	return &pinger{
 		wakuCtx: wakuCtx,
-		logger:  logger.Named("storenodeRequestor"),
 	}
 }
 
+func (p *pinger) PingPeer(ctx context.Context, peerID peer.ID) (time.Duration, error) {
+	var resp = C.allocResp()
+	var cPeerId = C.CString(peerID.String())
+	defer C.freeResp(resp)
+	defer C.free(unsafe.Pointer(cPeerId))
+
+	C.cGoWakuPingPeer(p.wakuCtx, cPeerId, C.int(time.Minute.Milliseconds()), resp)
+	if C.getRet(resp) == C.RET_OK {
+		rttStr := C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
+		rttInt, err := strconv.ParseInt(rttStr, 10, 64)
+		if err != nil {
+			return 0, err
+		}
+		return time.Duration(rttInt), nil
+	}
+
+	errMsg := C.GoStringN(C.getMyCharPtr(resp), C.int(C.getMyCharLen(resp)))
+	return 0, fmt.Errorf("PingPeer: %s", errMsg)
+}
+
 type storenodeRequestor struct {
 	wakuCtx unsafe.Pointer
 	logger  *zap.Logger
 }
 
+func newStorenodeRequestor(wakuCtx unsafe.Pointer, logger *zap.Logger) commonapi.StorenodeRequestor {
+	return &storenodeRequestor{
+		wakuCtx: wakuCtx,
+		logger:  logger.Named("storenodeRequestor"),
+	}
+}
+
 func (s *storenodeRequestor) GetMessagesByHash(ctx context.Context, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) (commonapi.StoreRequestResult, error) {
 	requestIDStr := hex.EncodeToString(protocol.GenerateRequestID())