diff --git a/.github/workflows/pr.yml b/.github/workflows/pr.yml index c073e5b..575c85c 100644 --- a/.github/workflows/pr.yml +++ b/.github/workflows/pr.yml @@ -17,12 +17,10 @@ jobs: # Clone the logos-delivery checkout OUTSIDE the module tree: a directory # named `vendor/` at the module root would put Go into vendor mode. LOGOS_DELIVERY_DIR: ${{ github.workspace }}/.logos-delivery - # Both kernel (libwaku) and messaging (liblogosdelivery) headers are on - # the include path so every package compiles. `-l` is deliberately NOT - # set here: each internal/ffi subpackage selects its own library via a - # `#cgo LDFLAGS` directive, so no binary links both libs (they export - # overlapping symbols until logos-delivery#3851). - CGO_CFLAGS: -I${{ github.workspace }}/.logos-delivery/library/ -I${{ github.workspace }}/.logos-delivery/liblogosdelivery/ + # The single liblogosdelivery library exposes the full API; its headers + # live in library/. The bridge self-links -llogosdelivery via a #cgo + # directive, so no -l is set here. + CGO_CFLAGS: -I${{ github.workspace }}/.logos-delivery/library/ CGO_LDFLAGS: -L${{ github.workspace }}/.logos-delivery/build/ -Wl,-rpath,${{ github.workspace }}/.logos-delivery/build/ # Build in module mode; never use a vendor/ dir. GOFLAGS: -mod=mod @@ -39,9 +37,8 @@ jobs: go-version: "1.24" - name: Resolve logos-delivery commit - # Cache the built libraries keyed on the exact upstream commit, so the - # expensive clone + build is skipped while logos-delivery's HEAD is - # unchanged. ls-remote gives us the SHA before we clone. + # Cache the built library keyed on the exact upstream commit, so the + # expensive clone + build is skipped while master's HEAD is unchanged. id: logos-delivery-rev run: | rev=$(git ls-remote https://github.com/logos-messaging/logos-delivery.git HEAD | cut -f1) @@ -58,9 +55,9 @@ jobs: if: steps.logos-delivery-cache.outputs.cache-hit != 'true' run: git clone --depth 1 https://github.com/logos-messaging/logos-delivery.git "$LOGOS_DELIVERY_DIR" - - name: Build libwaku + liblogosdelivery + - name: Build liblogosdelivery if: steps.logos-delivery-cache.outputs.cache-hit != 'true' - run: make -C "$LOGOS_DELIVERY_DIR" libwaku liblogosdelivery -j + run: make -C "$LOGOS_DELIVERY_DIR" liblogosdelivery -j - name: go build run: go build ./... @@ -88,9 +85,9 @@ jobs: - name: go test messaging (run) # Fast, network-free unit tests for the Messaging API + its ffi bridge. - run: go test ./pkg/messaging/... ./internal/ffi/liblogosdelivery/... + run: go test ./pkg/messaging/... ./internal/ffi/... - name: go test kernel (compile) # The kernel suite is heavy integration (runs nightly in CI.yml), so # only compile its test binaries here. - run: go test -run '^$' ./pkg/kernel/... ./internal/ffi/libwaku/... + run: go test -run '^$' ./pkg/kernel/... diff --git a/internal/ffi/doc.go b/internal/ffi/doc.go deleted file mode 100644 index de4ccfe..0000000 --- a/internal/ffi/doc.go +++ /dev/null @@ -1,7 +0,0 @@ -// Package ffi groups the cgo bridges over the logos-delivery C libraries. -// -// Each C library gets its own subpackage (libwaku now; liblogosdelivery in a -// follow-up) so that a binary links exactly the libraries it imports — the -// two .so files carry overlapping symbols and must never be linked together -// (until logos-delivery#3851 consolidates them). -package ffi diff --git a/internal/ffi/liblogosdelivery/liblogosdelivery.go b/internal/ffi/liblogosdelivery.go similarity index 91% rename from internal/ffi/liblogosdelivery/liblogosdelivery.go rename to internal/ffi/liblogosdelivery.go index 5f91d8a..430b887 100644 --- a/internal/ffi/liblogosdelivery/liblogosdelivery.go +++ b/internal/ffi/liblogosdelivery.go @@ -1,13 +1,12 @@ -// Package liblogosdelivery is the cgo bridge over liblogosdelivery (the -// logos-delivery Messaging API C library). It owns the synchronous -// request/response callback plumbing, the shared async event callback, and the -// handle->handler registry, and exposes Go-typed primitives so the public -// messaging package stays pure Go. -// -// It links liblogosdelivery via a #cgo directive; it must never be linked into -// the same binary as the libwaku bridge (overlapping symbols) until -// logos-delivery#3851 consolidates the two libraries. -package liblogosdelivery +// Package ffi is the cgo bridge over the single liblogosdelivery C library, +// which exposes the full logos-delivery API. It is split across two files +// mirroring the two headers: this file (liblogosdelivery.go) owns the shared +// plumbing — the synchronous request/response callback, the async event +// callback, the handle->handler registry and the node lifecycle — plus the +// stable Messaging API (liblogosdelivery.h); libwaku.go adds the low-level +// Kernel API (waku_*, liblogosdelivery_kernel.h). It exposes Go-typed +// primitives so pkg/kernel and pkg/messaging stay pure Go. +package ffi /* #cgo LDFLAGS: -llogosdelivery diff --git a/internal/ffi/libwaku/libwaku.go b/internal/ffi/libwaku.go similarity index 59% rename from internal/ffi/libwaku/libwaku.go rename to internal/ffi/libwaku.go index 2f5279e..0f7d102 100644 --- a/internal/ffi/libwaku/libwaku.go +++ b/internal/ffi/libwaku.go @@ -1,266 +1,107 @@ -// Package libwaku is the cgo bridge over libwaku (the legacy Kernel API -// library): the synchronous request/callback plumbing, the global event -// callback, and the handle registry. It exposes Go-typed primitives so -// pkg/kernel stays pure Go. -package libwaku +// This file holds the low-level Kernel (waku_*) tier of the single +// liblogosdelivery library (declared in liblogosdelivery_kernel.h). The shared +// plumbing — Handle, RetOK, EventHandler, the node lifecycle, the response/event +// callbacks and the call helper — lives in liblogosdelivery.go (same package). +package ffi /* -#cgo LDFLAGS: -lwaku -#include +#include #include -// wakuGoCallback (sync request/response) and wakuEventCallback (async events) -// are implemented in Go and exported below. -extern void wakuGoCallback(int ret, char* msg, size_t len, void* resp); -extern void wakuEventCallback(int ret, char* msg, size_t len, void* userData); - -// wakuResp carries a single synchronous call's result back from the callback, -// plus a pointer to the Go sync.WaitGroup the caller blocks on. -typedef struct { - int ret; - char* msg; - size_t len; - void* wg; -} wakuResp; - -static void* allocWakuResp(void* wg) { - wakuResp* r = (wakuResp*) calloc(1, sizeof(wakuResp)); - r->wg = wg; - return r; -} -static void freeWakuResp(void* resp) { if (resp != NULL) free(resp); } -static char* wakuRespMsg(void* resp) { return resp ? ((wakuResp*)resp)->msg : NULL; } -static size_t wakuRespLen(void* resp) { return resp ? ((wakuResp*)resp)->len : 0; } -static int wakuRespRet(void* resp) { return resp ? ((wakuResp*)resp)->ret : RET_ERR; } - -// Thin wrappers binding the shared Go callback to each libwaku entry point. -static void* cGoWakuNew(const char* configJson, void* resp) { - return waku_new(configJson, (FFICallBack) wakuGoCallback, resp); -} -static void cGoWakuStart(void* ctx, void* resp) { - waku_start(ctx, (FFICallBack) wakuGoCallback, resp); -} -static void cGoWakuStop(void* ctx, void* resp) { - waku_stop(ctx, (FFICallBack) wakuGoCallback, resp); -} -static void cGoWakuDestroy(void* ctx, void* resp) { - waku_destroy(ctx, (FFICallBack) wakuGoCallback, resp); -} +// logosGoCallback (the synchronous response callback) is defined in +// liblogosdelivery.go; the kernel wrappers below reuse it. +extern void logosGoCallback(int ret, char* msg, size_t len, void* resp); static void cGoWakuStartDiscV5(void* ctx, void* resp) { - waku_start_discv5(ctx, (FFICallBack) wakuGoCallback, resp); + waku_start_discv5(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuStopDiscV5(void* ctx, void* resp) { - waku_stop_discv5(ctx, (FFICallBack) wakuGoCallback, resp); + waku_stop_discv5(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuVersion(void* ctx, void* resp) { - waku_version(ctx, (FFICallBack) wakuGoCallback, resp); -} -static void cGoWakuSetEventCallback(void* ctx) { - // The ctx doubles as userData so the shared event callback can route the - // event to the right registered handler. - set_event_callback(ctx, (FFICallBack) wakuEventCallback, ctx); + waku_version(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuRelayPublish(void* ctx, const char* pubSubTopic, const char* jsonWakuMessage, int timeoutMs, void* resp) { - waku_relay_publish(ctx, (FFICallBack) wakuGoCallback, resp, pubSubTopic, jsonWakuMessage, timeoutMs); + waku_relay_publish(ctx, (FFICallBack) logosGoCallback, resp, pubSubTopic, jsonWakuMessage, timeoutMs); } static void cGoWakuRelaySubscribe(void* ctx, const char* pubSubTopic, void* resp) { - waku_relay_subscribe(ctx, (FFICallBack) wakuGoCallback, resp, pubSubTopic); + waku_relay_subscribe(ctx, (FFICallBack) logosGoCallback, resp, pubSubTopic); } static void cGoWakuRelayAddProtectedShard(void* ctx, int clusterId, int shardId, char* publicKey, void* resp) { - waku_relay_add_protected_shard(ctx, (FFICallBack) wakuGoCallback, resp, clusterId, shardId, publicKey); + waku_relay_add_protected_shard(ctx, (FFICallBack) logosGoCallback, resp, clusterId, shardId, publicKey); } static void cGoWakuRelayUnsubscribe(void* ctx, const char* pubSubTopic, void* resp) { - waku_relay_unsubscribe(ctx, (FFICallBack) wakuGoCallback, resp, pubSubTopic); + waku_relay_unsubscribe(ctx, (FFICallBack) logosGoCallback, resp, pubSubTopic); } static void cGoWakuConnect(void* ctx, const char* peerMultiAddr, int timeoutMs, void* resp) { - waku_connect(ctx, (FFICallBack) wakuGoCallback, resp, peerMultiAddr, timeoutMs); + waku_connect(ctx, (FFICallBack) logosGoCallback, resp, peerMultiAddr, timeoutMs); } static void cGoWakuDialPeer(void* ctx, const char* peerMultiAddr, const char* protocol, int timeoutMs, void* resp) { - waku_dial_peer(ctx, (FFICallBack) wakuGoCallback, resp, peerMultiAddr, protocol, timeoutMs); + waku_dial_peer(ctx, (FFICallBack) logosGoCallback, resp, peerMultiAddr, protocol, timeoutMs); } static void cGoWakuDialPeerById(void* ctx, const char* peerId, const char* protocol, int timeoutMs, void* resp) { - waku_dial_peer_by_id(ctx, (FFICallBack) wakuGoCallback, resp, peerId, protocol, timeoutMs); + waku_dial_peer_by_id(ctx, (FFICallBack) logosGoCallback, resp, peerId, protocol, timeoutMs); } static void cGoWakuDisconnectPeerById(void* ctx, const char* peerId, void* resp) { - waku_disconnect_peer_by_id(ctx, (FFICallBack) wakuGoCallback, resp, peerId); + waku_disconnect_peer_by_id(ctx, (FFICallBack) logosGoCallback, resp, peerId); } static void cGoWakuDisconnectAllPeers(void* ctx, void* resp) { - waku_disconnect_all_peers(ctx, (FFICallBack) wakuGoCallback, resp); + waku_disconnect_all_peers(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuListenAddresses(void* ctx, void* resp) { - waku_listen_addresses(ctx, (FFICallBack) wakuGoCallback, resp); + waku_listen_addresses(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuGetMyENR(void* ctx, void* resp) { - waku_get_my_enr(ctx, (FFICallBack) wakuGoCallback, resp); + waku_get_my_enr(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuGetMyPeerId(void* ctx, void* resp) { - waku_get_my_peerid(ctx, (FFICallBack) wakuGoCallback, resp); + waku_get_my_peerid(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuPingPeer(void* ctx, const char* peerAddr, int timeoutMs, void* resp) { - waku_ping_peer(ctx, (FFICallBack) wakuGoCallback, resp, peerAddr, timeoutMs); + waku_ping_peer(ctx, (FFICallBack) logosGoCallback, resp, peerAddr, timeoutMs); } static void cGoWakuGetPeersInMesh(void* ctx, const char* pubSubTopic, void* resp) { - waku_relay_get_peers_in_mesh(ctx, (FFICallBack) wakuGoCallback, resp, pubSubTopic); + waku_relay_get_peers_in_mesh(ctx, (FFICallBack) logosGoCallback, resp, pubSubTopic); } static void cGoWakuGetNumPeersInMesh(void* ctx, const char* pubSubTopic, void* resp) { - waku_relay_get_num_peers_in_mesh(ctx, (FFICallBack) wakuGoCallback, resp, pubSubTopic); + waku_relay_get_num_peers_in_mesh(ctx, (FFICallBack) logosGoCallback, resp, pubSubTopic); } static void cGoWakuGetNumConnectedRelayPeers(void* ctx, const char* pubSubTopic, void* resp) { - waku_relay_get_num_connected_peers(ctx, (FFICallBack) wakuGoCallback, resp, pubSubTopic); + waku_relay_get_num_connected_peers(ctx, (FFICallBack) logosGoCallback, resp, pubSubTopic); } static void cGoWakuGetConnectedRelayPeers(void* ctx, const char* pubSubTopic, void* resp) { - waku_relay_get_connected_peers(ctx, (FFICallBack) wakuGoCallback, resp, pubSubTopic); + waku_relay_get_connected_peers(ctx, (FFICallBack) logosGoCallback, resp, pubSubTopic); } static void cGoWakuGetConnectedPeers(void* ctx, void* resp) { - waku_get_connected_peers(ctx, (FFICallBack) wakuGoCallback, resp); + waku_get_connected_peers(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuGetPeerIdsFromPeerStore(void* ctx, void* resp) { - waku_get_peerids_from_peerstore(ctx, (FFICallBack) wakuGoCallback, resp); + waku_get_peerids_from_peerstore(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuGetConnectedPeersInfo(void* ctx, void* resp) { - waku_get_connected_peers_info(ctx, (FFICallBack) wakuGoCallback, resp); + waku_get_connected_peers_info(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuStoreQuery(void* ctx, const char* jsonQuery, const char* peerAddr, int timeoutMs, void* resp) { - waku_store_query(ctx, (FFICallBack) wakuGoCallback, resp, jsonQuery, peerAddr, timeoutMs); + waku_store_query(ctx, (FFICallBack) logosGoCallback, resp, jsonQuery, peerAddr, timeoutMs); } static void cGoWakuPeerExchangeQuery(void* ctx, uint64_t numPeers, void* resp) { - waku_peer_exchange_request(ctx, (FFICallBack) wakuGoCallback, resp, numPeers); + waku_peer_exchange_request(ctx, (FFICallBack) logosGoCallback, resp, numPeers); } static void cGoWakuGetPeerIdsByProtocol(void* ctx, const char* protocol, void* resp) { - waku_get_peerids_by_protocol(ctx, (FFICallBack) wakuGoCallback, resp, protocol); + waku_get_peerids_by_protocol(ctx, (FFICallBack) logosGoCallback, resp, protocol); } static void cGoWakuDnsDiscovery(void* ctx, const char* entTreeUrl, const char* nameDnsServer, int timeoutMs, void* resp) { - waku_dns_discovery(ctx, (FFICallBack) wakuGoCallback, resp, entTreeUrl, nameDnsServer, timeoutMs); + waku_dns_discovery(ctx, (FFICallBack) logosGoCallback, resp, entTreeUrl, nameDnsServer, timeoutMs); } static void cGoWakuIsOnline(void* ctx, void* resp) { - waku_is_online(ctx, (FFICallBack) wakuGoCallback, resp); + waku_is_online(ctx, (FFICallBack) logosGoCallback, resp); } static void cGoWakuGetMetrics(void* ctx, void* resp) { - waku_get_metrics(ctx, (FFICallBack) wakuGoCallback, resp); + waku_get_metrics(ctx, (FFICallBack) logosGoCallback, resp); } */ import "C" -import ( - "errors" - "sync" - "unsafe" -) - -// Handle is an opaque pointer to a node context owned by the C library. -type Handle = unsafe.Pointer - -// RetOK is the return code callbacks report on success. -const RetOK = C.RET_OK - -// EventHandler receives every event libwaku emits for a node: the raw -// event JSON when ret == RetOK, an error message otherwise. -type EventHandler func(ret int, msg string) - -// eventHandlers maps a node handle to the Go function that receives its -// events. The shared C event callback looks the handler up by handle. -var ( - eventHandlersMu sync.RWMutex - eventHandlers = make(map[Handle]EventHandler) -) - -//export wakuGoCallback -func wakuGoCallback(ret C.int, msg *C.char, length C.size_t, resp unsafe.Pointer) { - if resp == nil { - return - } - r := (*C.wakuResp)(resp) - r.ret = ret - r.msg = msg - r.len = length - wg := (*sync.WaitGroup)(r.wg) - wg.Done() -} - -//export wakuEventCallback -func wakuEventCallback(ret C.int, msg *C.char, length C.size_t, userData unsafe.Pointer) { - eventHandlersMu.RLock() - fn := eventHandlers[userData] // userData carries the node's handle - eventHandlersMu.RUnlock() - if fn != nil { - fn(int(ret), C.GoStringN(msg, C.int(length))) - } -} - -// call runs a synchronous libwaku entry point that reports its result -// through the response callback, blocks until it completes, and returns the -// callback message (on RetOK) or an error built from it. -func call(invoke func(resp unsafe.Pointer)) (string, error) { - var wg sync.WaitGroup - wg.Add(1) - resp := C.allocWakuResp(unsafe.Pointer(&wg)) - defer C.freeWakuResp(resp) - - invoke(resp) - wg.Wait() - - msg := C.GoStringN(C.wakuRespMsg(resp), C.int(C.wakuRespLen(resp))) - if C.wakuRespRet(resp) != C.RET_OK { - return "", errors.New(msg) - } - return msg, nil -} - -// New builds a node from a WakuConfig JSON string and returns its handle. -// The handle must be released with Destroy. -func New(configJSON string) (Handle, error) { - cCfg := C.CString(configJSON) - defer C.free(unsafe.Pointer(cCfg)) - - var wg sync.WaitGroup - wg.Add(1) - resp := C.allocWakuResp(unsafe.Pointer(&wg)) - defer C.freeWakuResp(resp) - - ctx := C.cGoWakuNew(cCfg, resp) - wg.Wait() - - if C.wakuRespRet(resp) != C.RET_OK || ctx == nil { - msg := C.GoStringN(C.wakuRespMsg(resp), C.int(C.wakuRespLen(resp))) - if msg == "" { - msg = "waku_new returned no context" - } - return nil, errors.New(msg) - } - return Handle(ctx), nil -} - -// SetEventHandler registers fn to receive events for the node and wires up -// the underlying C event callback. -func SetEventHandler(h Handle, fn EventHandler) { - eventHandlersMu.Lock() - eventHandlers[h] = fn - eventHandlersMu.Unlock() - C.cGoWakuSetEventCallback(h) -} - -// Start starts the node. -func Start(h Handle) error { - _, err := call(func(resp unsafe.Pointer) { C.cGoWakuStart(h, resp) }) - return err -} - -// Stop stops the node. -func Stop(h Handle) error { - _, err := call(func(resp unsafe.Pointer) { C.cGoWakuStop(h, resp) }) - return err -} - -// Destroy releases the node context and unregisters its event handler. -func Destroy(h Handle) error { - _, err := call(func(resp unsafe.Pointer) { C.cGoWakuDestroy(h, resp) }) - if err == nil { - eventHandlersMu.Lock() - delete(eventHandlers, h) - eventHandlersMu.Unlock() - } - return err -} +import "unsafe" // StartDiscV5 starts DiscV5 peer discovery. func StartDiscV5(h Handle) error { diff --git a/pkg/kernel/Makefile b/pkg/kernel/Makefile index 68b9104..0385c84 100644 --- a/pkg/kernel/Makefile +++ b/pkg/kernel/Makefile @@ -14,11 +14,11 @@ export LIBWAKU_HEADER_PATH ?= $(LOGOS_DELIVERY_DIR)/library export LIBWAKU_LIB_PATH ?= $(LOGOS_DELIVERY_DIR)/build export CGO_CFLAGS := -I$(LIBWAKU_HEADER_PATH)/ -export CGO_LDFLAGS := -L$(LIBWAKU_LIB_PATH)/ -lwaku -Wl,-rpath,$(LIBWAKU_LIB_PATH)/ +export CGO_LDFLAGS := -L$(LIBWAKU_LIB_PATH)/ -llogosdelivery -Wl,-rpath,$(LIBWAKU_LIB_PATH)/ # Expected files -HEADER_FILE := $(LIBWAKU_HEADER_PATH)/libwaku.h -LIB_FILES := $(wildcard $(LIBWAKU_LIB_PATH)/libwaku.*) +HEADER_FILE := $(LIBWAKU_HEADER_PATH)/liblogosdelivery.h +LIB_FILES := $(wildcard $(LIBWAKU_LIB_PATH)/liblogosdelivery.*) .PHONY: all clean prepare build test build-auto test-auto build-libwaku @@ -40,9 +40,9 @@ check-folders: echo "ERROR: Library path does not exist: $(LIBWAKU_LIB_PATH)"; exit 1; \ fi - @echo Checking for libwaku.h ... + @echo Checking for liblogosdelivery.h ... @if [ ! -f "$(HEADER_FILE)" ]; then \ - echo "ERROR: libwaku.h not found at: $(HEADER_FILE)"; exit 1; \ + echo "ERROR: liblogosdelivery.h not found at: $(HEADER_FILE)"; exit 1; \ fi @echo Checking for libwaku library file ... diff --git a/pkg/kernel/common/config.go b/pkg/kernel/common/config.go index a40fb74..66cf917 100644 --- a/pkg/kernel/common/config.go +++ b/pkg/kernel/common/config.go @@ -1,11 +1,12 @@ package common type WakuConfig struct { - Host string `json:"host,omitempty"` + // Host is the LibP2P listening address; the consolidated WakuNodeConf calls + // it listenAddress. This expects an IPv4. + Host string `json:"listenAddress,omitempty"` Nodekey string `json:"nodekey,omitempty"` Relay bool `json:"relay"` Store bool `json:"store,omitempty"` - LegacyStore bool `json:"legacyStore"` Storenode string `json:"storenode,omitempty"` StoreMessageRetentionPolicy string `json:"storeMessageRetentionPolicy,omitempty"` StoreMessageDbUrl string `json:"storeMessageDbUrl,omitempty"` diff --git a/pkg/kernel/doc.go b/pkg/kernel/doc.go index a2bd914..a789efc 100644 --- a/pkg/kernel/doc.go +++ b/pkg/kernel/doc.go @@ -1,3 +1,4 @@ // Package kernel is the low-level Go wrapper over the logos-delivery Kernel API -// (libwaku): relay, store, lightpush, filter, peer management, discovery. +// (the waku_* tier of the single liblogosdelivery library): relay, store, +// lightpush, filter, peer management, discovery. package kernel diff --git a/pkg/kernel/nwaku.go b/pkg/kernel/nwaku.go index 42baeba..78458ff 100644 --- a/pkg/kernel/nwaku.go +++ b/pkg/kernel/nwaku.go @@ -12,7 +12,7 @@ import ( "strings" "time" - "github.com/logos-messaging/logos-delivery-go-bindings/internal/ffi/libwaku" + "github.com/logos-messaging/logos-delivery-go-bindings/internal/ffi" "github.com/logos-messaging/logos-delivery-go-bindings/pkg/kernel/timesource" "github.com/ethereum/go-ethereum/crypto" @@ -33,7 +33,7 @@ const ConnectionChangeChanBufferSize = 1024 // WakuNode represents an instance of an nwaku node type WakuNode struct { - wakuCtx libwaku.Handle + wakuCtx ffi.Handle config *common.WakuConfig MsgChan chan common.Envelope TopicHealthChan chan topicHealth @@ -54,7 +54,7 @@ func NewWakuNode(config *common.WakuConfig, nodeName string) (*WakuNode, error) return nil, err } - n.wakuCtx, err = libwaku.New(string(jsonConfig)) + n.wakuCtx, err = ffi.New(string(jsonConfig)) if err != nil { Error("error wakuNew for %s: %v", nodeName, err) return nil, err @@ -64,15 +64,15 @@ func NewWakuNode(config *common.WakuConfig, nodeName string) (*WakuNode, error) n.TopicHealthChan = make(chan topicHealth, TopicHealthChanBufferSize) n.ConnectionChangeChan = make(chan connectionChange, ConnectionChangeChanBufferSize) - libwaku.SetEventHandler(n.wakuCtx, n.onRawEvent) + ffi.SetEventHandler(n.wakuCtx, n.onRawEvent) Debug("Successfully created WakuNode: %s", nodeName) return n, nil } -// onRawEvent receives every libwaku event for this node from the ffi bridge. +// onRawEvent receives every kernel event for this node from the ffi bridge. func (n *WakuNode) onRawEvent(ret int, msg string) { - if ret == libwaku.RetOK { + if ret == ffi.RetOK { n.OnEvent(msg) return } @@ -168,7 +168,7 @@ func (n *WakuNode) GetNumConnectedRelayPeers(optPubsubTopic ...string) (int, err pubsubTopic = optPubsubTopic[0] } - numPeersStr, err := libwaku.GetNumConnectedRelayPeers(n.wakuCtx, pubsubTopic) + numPeersStr, err := ffi.GetNumConnectedRelayPeers(n.wakuCtx, pubsubTopic) if err != nil { errMsg := "error GetNumConnectedRelayPeers: " + err.Error() Error("Failed to get number of connected relay peers for %s: %s", n.nodeName, errMsg) @@ -199,7 +199,7 @@ func (n *WakuNode) GetConnectedRelayPeers(optPubsubTopic ...string) (peer.IDSlic Debug("Fetching connected relay peers for pubsubTopic: %v, node: %v", pubsubTopic, n.nodeName) - peersStr, err := libwaku.GetConnectedRelayPeers(n.wakuCtx, pubsubTopic) + peersStr, err := ffi.GetConnectedRelayPeers(n.wakuCtx, pubsubTopic) if err != nil { errMsg := "error GetConnectedRelayPeers: " + err.Error() Error("Failed to get connected relay peers for pubsubTopic: %v:, node: %v. %v", pubsubTopic, n.nodeName, errMsg) @@ -227,14 +227,14 @@ func (n *WakuNode) GetConnectedRelayPeers(optPubsubTopic ...string) (peer.IDSlic } func (n *WakuNode) DisconnectPeerByID(peerID peer.ID) error { - if err := libwaku.DisconnectPeerByID(n.wakuCtx, peerID.String()); err != nil { + if err := ffi.DisconnectPeerByID(n.wakuCtx, peerID.String()); err != nil { return fmt.Errorf("error DisconnectPeerById: %w", err) } return nil } func (n *WakuNode) DisconnectAllPeers() error { - if err := libwaku.DisconnectAllPeers(n.wakuCtx); err != nil { + if err := ffi.DisconnectAllPeers(n.wakuCtx); err != nil { return fmt.Errorf("error DisconnectAllPeers: %w", err) } return nil @@ -249,7 +249,7 @@ func (n *WakuNode) GetConnectedPeers() (peer.IDSlice, error) { Debug("Fetching connected peers for %v", n.nodeName) - peersStr, err := libwaku.GetConnectedPeers(n.wakuCtx) + peersStr, err := ffi.GetConnectedPeers(n.wakuCtx) if err != nil { errMsg := "error GetConnectedPeers: " + err.Error() Error("Failed to get connected peers for %v: %v", n.nodeName, errMsg) @@ -285,7 +285,7 @@ func (n *WakuNode) GetPeersInMesh(pubsubTopic string) (peer.IDSlice, error) { Debug("Fetching peers in mesh peers for pubsubTopic: %v, node: %v", pubsubTopic, n.nodeName) - peersStr, err := libwaku.GetPeersInMesh(n.wakuCtx, pubsubTopic) + peersStr, err := ffi.GetPeersInMesh(n.wakuCtx, pubsubTopic) if err != nil { errMsg := "error GetPeersInMesh: " + err.Error() Error("Failed to get peers in mesh for pubsubTopic: %v:, node: %v. %v", pubsubTopic, n.nodeName, errMsg) @@ -321,7 +321,7 @@ func (n *WakuNode) RelaySubscribe(pubsubTopic string) error { return errors.New("wakuCtx is nil") } - if err := libwaku.RelaySubscribe(n.wakuCtx, pubsubTopic); err != nil { + if err := ffi.RelaySubscribe(n.wakuCtx, pubsubTopic); err != nil { Error("Failed to subscribe to relay on node %s, pubsubTopic: %s, error: %v", n.nodeName, pubsubTopic, err) return fmt.Errorf("error WakuRelaySubscribe: %w", err) } @@ -341,7 +341,7 @@ func (n *WakuNode) RelayAddProtectedShard(clusterId uint16, shardId uint16, pubk keyHexStr := hex.EncodeToString(crypto.FromECDSAPub(pubkey)) - if err := libwaku.RelayAddProtectedShard(n.wakuCtx, int(clusterId), int(shardId), keyHexStr); err != nil { + if err := ffi.RelayAddProtectedShard(n.wakuCtx, int(clusterId), int(shardId), keyHexStr); err != nil { return fmt.Errorf("error WakuRelayAddProtectedShard: %w", err) } return nil @@ -359,7 +359,7 @@ func (n *WakuNode) RelayUnsubscribe(pubsubTopic string) error { } Debug("Attempting to unsubscribe from relay on node %s, pubsubTopic: %s", n.nodeName, pubsubTopic) - if err := libwaku.RelayUnsubscribe(n.wakuCtx, pubsubTopic); err != nil { + if err := ffi.RelayUnsubscribe(n.wakuCtx, pubsubTopic); err != nil { Error("Failed to unsubscribe from relay on node %s, pubsubTopic: %s, error: %v", n.nodeName, pubsubTopic, err) return fmt.Errorf("error WakuRelayUnsubscribe: %w", err) } @@ -369,7 +369,7 @@ func (n *WakuNode) RelayUnsubscribe(pubsubTopic string) error { } func (n *WakuNode) PeerExchangeRequest(numPeers uint64) (uint64, error) { - numRecvPeersStr, err := libwaku.PeerExchangeRequest(n.wakuCtx, numPeers) + numRecvPeersStr, err := ffi.PeerExchangeRequest(n.wakuCtx, numPeers) if err != nil { Error("PeerExchangeRequest failed: %v", err) return 0, err @@ -386,7 +386,7 @@ func (n *WakuNode) PeerExchangeRequest(numPeers uint64) (uint64, error) { func (n *WakuNode) StartDiscV5() error { Debug("Starting DiscV5 for node: %s", n.nodeName) - if err := libwaku.StartDiscV5(n.wakuCtx); err != nil { + if err := ffi.StartDiscV5(n.wakuCtx); err != nil { errMsg := "error WakuStartDiscV5: " + err.Error() Error("Failed to start DiscV5 for node %s: %v", n.nodeName, errMsg) return errors.New(errMsg) @@ -396,7 +396,7 @@ func (n *WakuNode) StartDiscV5() error { } func (n *WakuNode) StopDiscV5() error { - if err := libwaku.StopDiscV5(n.wakuCtx); err != nil { + if err := ffi.StopDiscV5(n.wakuCtx); err != nil { errMsg := "error WakuStopDiscV5: " + err.Error() Error("Failed to stop DiscV5 for node %s: %v", n.nodeName, errMsg) return errors.New(errMsg) @@ -406,7 +406,7 @@ func (n *WakuNode) StopDiscV5() error { } func (n *WakuNode) Version() (string, error) { - version, err := libwaku.Version(n.wakuCtx) + version, err := ffi.Version(n.wakuCtx) if err != nil { errMsg := "error WakuVersion: " + err.Error() Error("Failed to fetch Waku version for node %s: %v", n.nodeName, errMsg) @@ -430,7 +430,7 @@ func (n *WakuNode) StoreQuery(ctx context.Context, storeRequest *common.StoreQue addrs[i] = addr.String() } - jsonResponseStr, err := libwaku.StoreQuery(n.wakuCtx, string(b), strings.Join(addrs, ","), timeoutMs) + jsonResponseStr, err := ffi.StoreQuery(n.wakuCtx, string(b), strings.Join(addrs, ","), timeoutMs) if err != nil { return nil, fmt.Errorf("error WakuStoreQuery: %w", err) } @@ -446,12 +446,28 @@ func (n *WakuNode) StoreQuery(ctx context.Context, storeRequest *common.StoreQue func (n *WakuNode) RelayPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string) (common.MessageHash, error) { timeoutMs := getContextTimeoutMilliseconds(ctx) - jsonMsg, err := json.Marshal(message) + // The library expects the WakuMessage wire format (camelCase keys); the + // generated protobuf struct marshals content_topic, so marshal explicitly. + jsonMsg, err := json.Marshal(struct { + Payload []byte `json:"payload,omitempty"` + ContentTopic string `json:"contentTopic"` + Version *uint32 `json:"version,omitempty"` + Timestamp *int64 `json:"timestamp,omitempty"` + Meta []byte `json:"meta,omitempty"` + Ephemeral *bool `json:"ephemeral,omitempty"` + }{ + Payload: message.Payload, + ContentTopic: message.ContentTopic, + Version: message.Version, + Timestamp: message.Timestamp, + Meta: message.Meta, + Ephemeral: message.Ephemeral, + }) if err != nil { return common.MessageHash(""), err } - msgHash, err := libwaku.RelayPublish(n.wakuCtx, pubsubTopic, string(jsonMsg), timeoutMs) + msgHash, err := ffi.RelayPublish(n.wakuCtx, pubsubTopic, string(jsonMsg), timeoutMs) if err != nil { return common.MessageHash(""), fmt.Errorf("WakuRelayPublish: %w", err) } @@ -489,7 +505,7 @@ func (n *WakuNode) RelayPublishNoCTX(pubsubTopic string, message *pb.WakuMessage func (n *WakuNode) DnsDiscovery(ctx context.Context, enrTreeUrl string, nameDnsServer string) ([]multiaddr.Multiaddr, error) { timeoutMs := getContextTimeoutMilliseconds(ctx) - nodeAddresses, err := libwaku.DnsDiscovery(n.wakuCtx, enrTreeUrl, nameDnsServer, timeoutMs) + nodeAddresses, err := ffi.DnsDiscovery(n.wakuCtx, enrTreeUrl, nameDnsServer, timeoutMs) if err != nil { return nil, fmt.Errorf("error WakuDnsDiscovery: %w", err) } @@ -514,7 +530,7 @@ func (n *WakuNode) PingPeer(ctx context.Context, peerInfo peer.AddrInfo) (time.D timeoutMs := getContextTimeoutMilliseconds(ctx) - rttStr, err := libwaku.PingPeer(n.wakuCtx, strings.Join(addrs, ","), timeoutMs) + rttStr, err := ffi.PingPeer(n.wakuCtx, strings.Join(addrs, ","), timeoutMs) if err != nil { return 0, fmt.Errorf("PingPeer: %w", err) } @@ -529,7 +545,7 @@ func (n *WakuNode) PingPeer(ctx context.Context, peerInfo peer.AddrInfo) (time.D func (n *WakuNode) Start() error { Debug("Starting %s", n.nodeName) - if err := libwaku.Start(n.wakuCtx); err != nil { + if err := ffi.Start(n.wakuCtx); err != nil { errMsg := "error WakuStart: " + err.Error() Error("Failed to start %s: %s", n.nodeName, errMsg) return errors.New(errMsg) @@ -542,7 +558,7 @@ func (n *WakuNode) Start() error { func (n *WakuNode) Stop() error { Debug("Stopping %s", n.nodeName) - if err := libwaku.Stop(n.wakuCtx); err != nil { + if err := ffi.Stop(n.wakuCtx); err != nil { errMsg := "error WakuStop: " + err.Error() Error("Failed to stop %s: %s", n.nodeName, errMsg) return errors.New(errMsg) @@ -561,7 +577,7 @@ func (n *WakuNode) Destroy() error { Debug("Destroying %v", n.nodeName) - if err := libwaku.Destroy(n.wakuCtx); err != nil { + if err := ffi.Destroy(n.wakuCtx); err != nil { errMsg := "error WakuDestroy: " + err.Error() Error("Failed to destroy %v: %v", n.nodeName, errMsg) return errors.New(errMsg) @@ -572,7 +588,7 @@ func (n *WakuNode) Destroy() error { } func (n *WakuNode) PeerID() (peer.ID, error) { - peerIdStr, err := libwaku.GetMyPeerID(n.wakuCtx) + peerIdStr, err := ffi.GetMyPeerID(n.wakuCtx) if err != nil { return "", err } @@ -588,7 +604,7 @@ func (n *WakuNode) PeerID() (peer.ID, error) { func (n *WakuNode) Connect(ctx context.Context, addr multiaddr.Multiaddr) error { timeoutMs := getContextTimeoutMilliseconds(ctx) - if err := libwaku.Connect(n.wakuCtx, addr.String(), timeoutMs); err != nil { + if err := ffi.Connect(n.wakuCtx, addr.String(), timeoutMs); err != nil { return fmt.Errorf("error WakuConnect: %w", err) } return nil @@ -597,14 +613,14 @@ func (n *WakuNode) Connect(ctx context.Context, addr multiaddr.Multiaddr) error func (n *WakuNode) DialPeerByID(ctx context.Context, peerID peer.ID, protocol libp2pproto.ID) error { timeoutMs := getContextTimeoutMilliseconds(ctx) - if err := libwaku.DialPeerByID(n.wakuCtx, peerID.String(), string(protocol), timeoutMs); err != nil { + if err := ffi.DialPeerByID(n.wakuCtx, peerID.String(), string(protocol), timeoutMs); err != nil { return fmt.Errorf("error DialPeerById: %w", err) } return nil } func (n *WakuNode) ListenAddresses() ([]multiaddr.Multiaddr, error) { - listenAddresses, err := libwaku.ListenAddresses(n.wakuCtx) + listenAddresses, err := ffi.ListenAddresses(n.wakuCtx) if err != nil { return nil, fmt.Errorf("error WakuListenAddresses: %w", err) } @@ -622,7 +638,7 @@ func (n *WakuNode) ListenAddresses() ([]multiaddr.Multiaddr, error) { } func (n *WakuNode) ENR() (*enode.Node, error) { - enrStr, err := libwaku.GetMyENR(n.wakuCtx) + enrStr, err := ffi.GetMyENR(n.wakuCtx) if err != nil { return nil, fmt.Errorf("error WakuGetMyENR: %w", err) } @@ -635,7 +651,7 @@ func (n *WakuNode) ENR() (*enode.Node, error) { } func (n *WakuNode) GetNumPeersInMesh(pubsubTopic string) (int, error) { - numPeersStr, err := libwaku.GetNumPeersInMesh(n.wakuCtx, pubsubTopic) + numPeersStr, err := ffi.GetNumPeersInMesh(n.wakuCtx, pubsubTopic) if err != nil { return 0, fmt.Errorf("error GetNumPeersInMesh: %w", err) } @@ -649,7 +665,7 @@ func (n *WakuNode) GetNumPeersInMesh(pubsubTopic string) (int, error) { } func (n *WakuNode) GetPeerIDsFromPeerStore() (peer.IDSlice, error) { - peersStr, err := libwaku.GetPeerIDsFromPeerStore(n.wakuCtx) + peersStr, err := ffi.GetPeerIDsFromPeerStore(n.wakuCtx) if err != nil { return nil, fmt.Errorf("GetPeerIdsFromPeerStore: %s", err.Error()) } @@ -672,7 +688,7 @@ func (n *WakuNode) GetPeerIDsFromPeerStore() (peer.IDSlice, error) { } func (n *WakuNode) GetConnectedPeersInfo() (common.PeersData, error) { - jsonStr, err := libwaku.GetConnectedPeersInfo(n.wakuCtx) + jsonStr, err := ffi.GetConnectedPeersInfo(n.wakuCtx) if err != nil { return nil, fmt.Errorf("GetConnectedPeersInfo: %s", err.Error()) } @@ -691,7 +707,7 @@ func (n *WakuNode) GetConnectedPeersInfo() (common.PeersData, error) { } func (n *WakuNode) GetPeerIDsByProtocol(protocol libp2pproto.ID) (peer.IDSlice, error) { - peersStr, err := libwaku.GetPeerIDsByProtocol(n.wakuCtx, string(protocol)) + peersStr, err := ffi.GetPeerIDsByProtocol(n.wakuCtx, string(protocol)) if err != nil { return nil, fmt.Errorf("GetPeerIdsByProtocol: error GetPeerIdsByProtocol: %s", err.Error()) } @@ -716,7 +732,7 @@ func (n *WakuNode) GetPeerIDsByProtocol(protocol libp2pproto.ID) (peer.IDSlice, func (n *WakuNode) DialPeer(ctx context.Context, peerAddr multiaddr.Multiaddr, protocol libp2pproto.ID) error { timeoutMs := getContextTimeoutMilliseconds(ctx) - if err := libwaku.DialPeer(n.wakuCtx, peerAddr.String(), string(protocol), timeoutMs); err != nil { + if err := ffi.DialPeer(n.wakuCtx, peerAddr.String(), string(protocol), timeoutMs); err != nil { return fmt.Errorf("error DialPeer: %w", err) } return nil @@ -934,7 +950,7 @@ func (n *WakuNode) IsOnline() (bool, error) { Debug("Querying online state for %v", n.nodeName) - onlineStr, err := libwaku.IsOnline(n.wakuCtx) + onlineStr, err := ffi.IsOnline(n.wakuCtx) if err != nil { errMsg := "error IsOnline: " + err.Error() Error("Failed to query online state for %v: %v", n.nodeName, errMsg) @@ -953,7 +969,7 @@ func (n *WakuNode) GetMetrics() (string, error) { Debug("Querying metrics for %v", n.nodeName) - metricsStr, err := libwaku.GetMetrics(n.wakuCtx) + metricsStr, err := ffi.GetMetrics(n.wakuCtx) if err != nil { errMsg := "error GetMetrics: " + err.Error() Error("Failed to query metrics for %v: %v", n.nodeName, errMsg) diff --git a/pkg/kernel/nwaku_test.go b/pkg/kernel/nwaku_test.go index 4a73ed3..f3f74cc 100644 --- a/pkg/kernel/nwaku_test.go +++ b/pkg/kernel/nwaku_test.go @@ -672,7 +672,6 @@ func TestStore(t *testing.T) { Shards: []uint16{64}, Discv5UdpPort: udpPort, TcpPort: tcpPort, - LegacyStore: false, } senderNode, err := NewWakuNode(&senderNodeWakuConfig, "senderNode") @@ -692,7 +691,6 @@ func TestStore(t *testing.T) { Shards: []uint16{64}, Discv5UdpPort: udpPort, TcpPort: tcpPort, - LegacyStore: false, } receiverNode, err := NewWakuNode(&receiverNodeWakuConfig, "receiverNode") require.NoError(t, err) diff --git a/pkg/kernel/relay_send_receive_test.go b/pkg/kernel/relay_send_receive_test.go new file mode 100644 index 0000000..f951af9 --- /dev/null +++ b/pkg/kernel/relay_send_receive_test.go @@ -0,0 +1,90 @@ +package kernel + +import ( + "context" + "crypto/rand" + "testing" + "time" + + "github.com/logos-messaging/logos-delivery-go-bindings/pkg/kernel/common" + "github.com/logos-messaging/logos-delivery-go-bindings/pkg/kernel/pb" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +// TestRelaySendReceive is an end-to-end check against the single liblogosdelivery +// library: two relay nodes are connected, one publishes a random payload and the +// other must receive it. It exercises the unified node lifecycle and the Kernel +// relay ops (waku_relay_subscribe/publish) over the one library. +func TestRelaySendReceive(t *testing.T) { + const clusterID, shardID = 16, 64 + + newNode := func(name string) *WakuNode { + node, err := StartWakuNode(name, &common.WakuConfig{ + Relay: true, + LogLevel: "ERROR", + Discv5Discovery: false, + ClusterID: clusterID, + Shards: []uint16{shardID}, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = node.StopAndDestroy() }) + return node + } + + sender := newNode("sender") + receiver := newNode("receiver") + + topic := FormatWakuRelayTopic(clusterID, shardID) + require.NoError(t, sender.RelaySubscribe(topic)) + require.NoError(t, receiver.RelaySubscribe(topic)) + + // Dial the receiver from the sender using the receiver's listen multiaddr + // (it already embeds the peer id). + addrs, err := receiver.ListenAddresses() + require.NoError(t, err) + require.NotEmpty(t, addrs) + + connCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + require.NoError(t, sender.Connect(connCtx, addrs[0])) + require.Eventually(t, func() bool { + n, _ := sender.GetNumConnectedPeers() + return n >= 1 + }, 15*time.Second, time.Second, "sender never connected to the receiver") + + payload := make([]byte, 16) + _, err = rand.Read(payload) + require.NoError(t, err) + + publish := func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + // May fail with NoPeersToPublish until gossipsub grafts the mesh; retried. + _, _ = sender.RelayPublish(ctx, &pb.WakuMessage{ + Payload: payload, + ContentTopic: "/kernel-test/1/relay/proto", + Version: proto.Uint32(0), + Timestamp: proto.Int64(time.Now().UnixNano()), + }, topic) + } + + // Publish immediately, then retry each second while waiting for delivery — + // the relay mesh takes a moment to form after the connection is established. + publish() + ticker := time.NewTicker(1 * time.Second) + defer ticker.Stop() + deadline := time.After(10 * time.Second) + for { + select { + case env := <-receiver.MsgChan: + if string(env.Message().GetPayload()) == string(payload) { + return // received our exact message — success + } + case <-ticker.C: + publish() + case <-deadline: + t.Fatal("timed out waiting for the receiver to get the message") + } + } +} diff --git a/pkg/kernel/store_test.go b/pkg/kernel/store_test.go index 5b01f72..27c07a8 100644 --- a/pkg/kernel/store_test.go +++ b/pkg/kernel/store_test.go @@ -837,7 +837,6 @@ func TestCheckLegacyStore(t *testing.T) { node2Config := DefaultWakuConfig node2Config.Relay = true node2Config.Store = true - node2Config.LegacyStore = true Debug("Creating Node2") node2, err := StartWakuNode("Node2", &node2Config)