From 1dc86ee52412a8ff4f6af1f98b1848fb51fedf7c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Wojciech=20Pi=C4=85tkowski?= Date: Sat, 29 Nov 2025 23:25:38 +0100 Subject: [PATCH 1/2] chore: enrich logging about pubsub topic state --- be/src/start-libp2p-relay.ts | 82 +++++++++++++++++++++++------------- 1 file changed, 52 insertions(+), 30 deletions(-) diff --git a/be/src/start-libp2p-relay.ts b/be/src/start-libp2p-relay.ts index 5b45849..7d89bc0 100644 --- a/be/src/start-libp2p-relay.ts +++ b/be/src/start-libp2p-relay.ts @@ -10,7 +10,7 @@ import { initRoomRegistrationProtocol, shortenPeerId } from "@icod2/protocols"; import { autoNAT } from "@libp2p/autonat"; import { circuitRelayServer } from "@libp2p/circuit-relay-v2"; import { identify } from "@libp2p/identify"; -import type { Libp2p, PeerId } from "@libp2p/interface"; +import type { Libp2p, PeerId, SubscriptionChangeData } from "@libp2p/interface"; import { tcp } from "@libp2p/tcp"; import { webSockets } from "@libp2p/websockets"; import { createLibp2p } from "libp2p"; @@ -202,18 +202,41 @@ export async function startLibp2pRelay({ debouncedLogPeerUpdated(peerIdStr); }); - libp2p.services.pubsub.addEventListener("subscription-change", () => { - debouncedSubscriptionChange(); + libp2p.services.pubsub.addEventListener("subscription-change", (e) => { + debouncedSubscriptionChange(e); }); - function subscriptionChanged() { + function subscriptionChanged(e: CustomEvent) { + const topic = e.detail.subscriptions[0].topic; + logger.info(e.detail, "Subscription changed event detail"); + logger.info( + libp2p.services.pubsub + .getSubscribers(topic) + .map((peerId) => shortenPeerId(peerId.toString())), + `Subscribers to ${topic} topic `, + ); const roomStats = getRoomStats(); logger.info({ roomStats }, "Subscription changed"); + debouncedPrintInfoAboutPubSubTopics(); } const debouncedLogPeerUpdated = debounce(peerUpdated, 2000); const debouncedSubscriptionChange = debounce(subscriptionChanged, 2000); + function getTopics() { + return libp2p.services.pubsub.getTopics(); + } + + const debouncedPrintInfoAboutPubSubTopics = debounce( + printInfoAboutPubSubTopics, + 2000, + ); + + function printInfoAboutPubSubTopics() { + const topics = getTopics(); + logger.info({ topics }, "All Topics"); + } + function peerUpdated(peerIdStr: string) { logger.info( { @@ -224,6 +247,7 @@ export async function startLibp2pRelay({ }, "Peer updated", ); + debouncedPrintInfoAboutPubSubTopics(); } function getRoomStats() { @@ -239,33 +263,31 @@ export async function startLibp2pRelay({ return acc; }, {}); } +} - function formatConnectedPeers(peers: Set): string[] { - return Array.from(peers.values()).map((peerId) => shortenPeerId(peerId)); - } +function formatConnectedPeers(peers: Set): string[] { + return Array.from(peers.values()).map((peerId) => shortenPeerId(peerId)); +} - function toPeerIdString(peerId: string | PeerId): string { - return typeof peerId === "string" ? peerId : peerId.toString(); - } +function toPeerIdString(peerId: string | PeerId): string { + return typeof peerId === "string" ? peerId : peerId.toString(); +} - function getConnectionStats(libp2p: Libp2p, peerIdStr: string) { - return libp2p.getConnections().reduce( - (acc, conn) => { - const object = acc.find( - (x) => x.peerId === shortenPeerId(peerIdStr), - ) ?? { - peerId: shortenPeerId(conn.remotePeer.toString()), - connectionCount: 0, - streamsCount: 0, - }; - if (object.connectionCount === 0) { - acc.push(object); - } - object.connectionCount += 1; - object.streamsCount += conn.streams.length; - return acc; - }, - [] as { peerId: string; connectionCount: number; streamsCount: number }[], - ); - } +function getConnectionStats(libp2p: Libp2p, peerIdStr: string) { + return libp2p.getConnections().reduce( + (acc, conn) => { + const object = acc.find((x) => x.peerId === shortenPeerId(peerIdStr)) ?? { + peerId: shortenPeerId(conn.remotePeer.toString()), + connectionCount: 0, + streamsCount: 0, + }; + if (object.connectionCount === 0) { + acc.push(object); + } + object.connectionCount += 1; + object.streamsCount += conn.streams.length; + return acc; + }, + [] as { peerId: string; connectionCount: number; streamsCount: number }[], + ); } From b0631b1e4165cc5ab3033532f945b49a8e42b585 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Wojciech=20Pi=C4=85tkowski?= Date: Sun, 30 Nov 2025 20:25:16 +0100 Subject: [PATCH 2/2] feat: expose relay versoin to fe node --- .github/workflows/be-ci.yaml | 6 ++++ be/Dockerfile | 4 +++ be/config.yaml | 6 ++-- be/src/index.ts | 5 +++ be/src/start-libp2p-relay.ts | 17 +++++++++- .../useFollowerConnection.ts | 5 ++- .../connection-setups/useLeaderConnection.ts | 5 ++- .../libp2p/useRelayPeerMessageProto/index.ts | 1 + .../useRelayPeerMessageProto.ts | 24 +++++++++++++ .../src/peers-message-exchange-protocol.ts | 34 ++++++++++++------- 10 files changed, 88 insertions(+), 19 deletions(-) create mode 100644 fe/src/services/libp2p/useRelayPeerMessageProto/index.ts create mode 100644 fe/src/services/libp2p/useRelayPeerMessageProto/useRelayPeerMessageProto.ts diff --git a/.github/workflows/be-ci.yaml b/.github/workflows/be-ci.yaml index 5b90a57..88f214c 100644 --- a/.github/workflows/be-ci.yaml +++ b/.github/workflows/be-ci.yaml @@ -67,6 +67,10 @@ jobs: type=raw,value=latest type=sha + - name: Get commit date + id: commit-date + run: echo "date=$(git show -s --format=%cI ${{ github.sha }} | cut -c1-19 | sed 's/[:]/-/g')" >> $GITHUB_OUTPUT + - name: Build and push Docker image uses: docker/build-push-action@v5 with: @@ -78,3 +82,5 @@ jobs: labels: ${{ steps.meta.outputs.labels }} cache-from: type=gha cache-to: type=gha,mode=max + build-args: | + VERSION_TAG=${{ steps.commit-date.outputs.date }} diff --git a/be/Dockerfile b/be/Dockerfile index 4d680ee..e237dc0 100644 --- a/be/Dockerfile +++ b/be/Dockerfile @@ -1,5 +1,9 @@ FROM node:lts-slim AS base +ARG VERSION_TAG=dev +ENV VERSION_TAG=${VERSION_TAG} +LABEL version="${VERSION_TAG}" + WORKDIR /app # Copy relevant files early for better caching diff --git a/be/config.yaml b/be/config.yaml index acdcb36..1e2b515 100644 --- a/be/config.yaml +++ b/be/config.yaml @@ -9,8 +9,8 @@ libp2p: logging: level: info # axiom: - # enabled: false - # dataset: your-dataset - # orgId: your-org-id + # enabled: true + # dataset: stashcrate-logs + # orgId: coding-nimbus-ovls # token: your-token # Prefer environment variables AXIOM_TOKEN/AXIOM_ORG_ID/AXIOM_DATASET # url: https://api.axiom.co diff --git a/be/src/index.ts b/be/src/index.ts index a8a282a..eb3a3da 100644 --- a/be/src/index.ts +++ b/be/src/index.ts @@ -10,6 +10,10 @@ const { libp2p, logging } = config; const logger = initializeLogger(logging); startToListenToErrors(logger); +// Log version information on startup +const versionTag = process.env.VERSION_TAG || "dev"; +logger.info(`Version: ${versionTag}`); + const stdoutEmitter = new StdoutEmitter(); stdoutEmitter.on("data", (data) => { @@ -29,6 +33,7 @@ try { startLibp2pRelay({ announceMultiaddrs: libp2p.announceMultiaddrs, listenMultiaddrs: libp2p.listenMultiaddrs, + versionTag, }); } catch (error) { logger.error({ error }, "Failed to start libp2p relay"); diff --git a/be/src/start-libp2p-relay.ts b/be/src/start-libp2p-relay.ts index 7d89bc0..7a71a34 100644 --- a/be/src/start-libp2p-relay.ts +++ b/be/src/start-libp2p-relay.ts @@ -6,7 +6,11 @@ import { import { noise } from "@chainsafe/libp2p-noise"; import { yamux } from "@chainsafe/libp2p-yamux"; import type { RoomRegistrationProtocolResponses } from "@icod2/protocols"; -import { initRoomRegistrationProtocol, shortenPeerId } from "@icod2/protocols"; +import { + initRoomRegistrationProtocol, + PeerMessageExchangeProtocol, + shortenPeerId, +} from "@icod2/protocols"; import { autoNAT } from "@libp2p/autonat"; import { circuitRelayServer } from "@libp2p/circuit-relay-v2"; import { identify } from "@libp2p/identify"; @@ -22,11 +26,13 @@ import { getPeerIdFromEnv } from "./utils/get-or-create-peer-id.js"; export type Args = { listenMultiaddrs: string[]; announceMultiaddrs?: string[]; + versionTag?: string; }; export async function startLibp2pRelay({ listenMultiaddrs, announceMultiaddrs, + versionTag = "unknown", }: Args) { const logger = getLogger(); const privateKey = await getPeerIdFromEnv(); @@ -74,6 +80,11 @@ export async function startLibp2pRelay({ const peerId = libp2p.peerId.toString(); const multiaddrs = libp2p.getMultiaddrs(); + const peerMessageExchange = new PeerMessageExchangeProtocol({ + protocolId: "/icod2/relay-peer-message-exchange/1.0.0", + }); + peerMessageExchange.initialize(libp2p); + logger.info( { peerId, shortPeerId: shortenPeerId(peerId) }, "Relay peer ready", @@ -169,6 +180,10 @@ export async function startLibp2pRelay({ libp2p.addEventListener("peer:connect", (event) => { const peerIdStr = event.detail.toString(); connectedPeers.add(peerIdStr); + peerMessageExchange.sendMessageToPeer(peerIdStr, { + type: "hello-from-relay", + message: `Hello my I'm a relay. My version is ${versionTag}`, + }); logger.info( { peerId: shortenPeerId(peerIdStr), diff --git a/fe/src/services/libp2p/connection-setups/useFollowerConnection.ts b/fe/src/services/libp2p/connection-setups/useFollowerConnection.ts index e642981..1897c75 100644 --- a/fe/src/services/libp2p/connection-setups/useFollowerConnection.ts +++ b/fe/src/services/libp2p/connection-setups/useFollowerConnection.ts @@ -9,6 +9,7 @@ import { type PeerMessageExchangeProtocol, usePeerMessageProto, } from "../usePeerMessageProto"; +import { useRelayPeerMessageProto } from "../useRelayPeerMessageProto"; import { type IgnoredErrors, ignoredErrors } from "./ignoredErrors"; export const useFollowerConnection = ({ roomToken }: { roomToken: string }) => { @@ -21,6 +22,8 @@ export const useFollowerConnection = ({ roomToken }: { roomToken: string }) => { onMessageListener: routerMng.currentCombinedRouter, }); + const relayMessageProto = useRelayPeerMessageProto(); + const [error, setError] = useState< Libp2pServiceErrors | ConnectionErrors | undefined >(undefined); @@ -51,7 +54,7 @@ export const useFollowerConnection = ({ roomToken }: { roomToken: string }) => { onFailedToConnect: (error) => { setFilteredError(error); }, - protos: [messageProto], + protos: [messageProto, relayMessageProto], }); return { diff --git a/fe/src/services/libp2p/connection-setups/useLeaderConnection.ts b/fe/src/services/libp2p/connection-setups/useLeaderConnection.ts index 19e9d1d..8f7759f 100644 --- a/fe/src/services/libp2p/connection-setups/useLeaderConnection.ts +++ b/fe/src/services/libp2p/connection-setups/useLeaderConnection.ts @@ -9,6 +9,7 @@ import { type PeerMessageExchangeProtocol, usePeerMessageProto, } from "../usePeerMessageProto"; +import { useRelayPeerMessageProto } from "../useRelayPeerMessageProto"; import { useRoomRegistration } from "../useRoomRegistrationProto"; import { type IgnoredErrors, ignoredErrors } from "./ignoredErrors"; @@ -56,6 +57,8 @@ export const useLeaderConnection = ({ onMessageListener: routerMng.currentCombinedRouter, }); + const relayMessageProto = useRelayPeerMessageProto(); + const { isRelayReconnecting } = useLibp2p({ roomTokenProvider: roomTokenProvider, connectedPeersStorage: connectedPeersStorage.current, @@ -65,7 +68,7 @@ export const useLeaderConnection = ({ onFailedToConnect: (error) => { setFilteredError(error); }, - protos: [roomRegistrationObject, messageProto], + protos: [roomRegistrationObject, messageProto, relayMessageProto], dontDialDiscoveredPeers: true, }); diff --git a/fe/src/services/libp2p/useRelayPeerMessageProto/index.ts b/fe/src/services/libp2p/useRelayPeerMessageProto/index.ts new file mode 100644 index 0000000..f6adf90 --- /dev/null +++ b/fe/src/services/libp2p/useRelayPeerMessageProto/index.ts @@ -0,0 +1 @@ +export { useRelayPeerMessageProto } from "./useRelayPeerMessageProto"; diff --git a/fe/src/services/libp2p/useRelayPeerMessageProto/useRelayPeerMessageProto.ts b/fe/src/services/libp2p/useRelayPeerMessageProto/useRelayPeerMessageProto.ts new file mode 100644 index 0000000..9366f0c --- /dev/null +++ b/fe/src/services/libp2p/useRelayPeerMessageProto/useRelayPeerMessageProto.ts @@ -0,0 +1,24 @@ +import { PeerMessageExchangeProtocol } from "@icod2/protocols"; +import type { Libp2p } from "@libp2p/interface"; +import { useEffect, useRef } from "react"; + +export const useRelayPeerMessageProto = () => { + const peerMessageProtoRef = useRef( + undefined, + ); + + useEffect(() => { + const peerMessageProto = new PeerMessageExchangeProtocol({ + protocolId: "/icod2/relay-peer-message-exchange/1.0.0", + }); + + peerMessageProtoRef.current = peerMessageProto; + }, []); + + return { + initialize: (libp2p: Libp2p) => { + peerMessageProtoRef.current?.initialize(libp2p); + }, + peerMessageProtoRef, + }; +}; diff --git a/libs/protocols/src/peers-message-exchange-protocol.ts b/libs/protocols/src/peers-message-exchange-protocol.ts index d983f31..2b62432 100644 --- a/libs/protocols/src/peers-message-exchange-protocol.ts +++ b/libs/protocols/src/peers-message-exchange-protocol.ts @@ -45,7 +45,12 @@ export class PeerMessageExchangeProtocol< this.listener = onMessage; } - initialize(libp2p: Libp2p) { + initialize( + libp2p: Libp2p, + { + skipHandleInitialization = false, + }: { skipHandleInitialization?: boolean } = {}, + ) { console.log( "initialize()", `protocol: ${this.protocolId}`, @@ -54,18 +59,21 @@ export class PeerMessageExchangeProtocol< if (this.initialized) return; this.libp2p = libp2p; - registerProtoHandle(this.protocolId, libp2p, (message, peerId) => { - const jsonMessage = parseJsonSafely(message); - if (!jsonMessage) return; - loggerGate.canLog && - console.log( - "Received message from peer:", - shortenPeerId(peerId), - "message:", - jsonMessage, - ); - this.listener?.(peerId, jsonMessage as TPeerMessagePayload, this); - }); + + if (skipHandleInitialization) { + registerProtoHandle(this.protocolId, libp2p, (message, peerId) => { + const jsonMessage = parseJsonSafely(message); + if (!jsonMessage) return; + loggerGate.canLog && + console.log( + "Received message from peer:", + shortenPeerId(peerId), + "message:", + jsonMessage, + ); + this.listener?.(peerId, jsonMessage as TPeerMessagePayload, this); + }); + } this.initialized = true; }