Merge pull request #1973 from streamr-dev/stream-latency

Add stream latency calculation

Author: tumppi

src/components/Stats.tsx

@@ -104,7 +104,7 @@ export function StreamStats({ streamId }: StreamStatsProps) {
                 <Stat
                     id="latency"
                     label="Latency ms"
-                    value={latency == null ? undefined : latency.toFixed(2)}
+                    value={latency == null ? undefined : latency.toFixed(0)}
                 />
             </ButtonGrid>
         </StreamStatsRoot>

src/generated/gql/indexer.ts

@@ -35,6 +35,7 @@ export type Neighbor = {
   __typename?: 'Neighbor';
   nodeId1: Scalars['String']['output'];
   nodeId2: Scalars['String']['output'];
+  rtt?: Maybe<Scalars['Int']['output']>;
   streamPartId: Scalars['String']['output'];
 };
 
@@ -157,6 +158,17 @@ export type GetGlobalStreamsStatsQueryVariables = Exact<{ [key: string]: never;
 
 export type GetGlobalStreamsStatsQuery = { __typename?: 'Query', summary: { __typename?: 'Summary', bytesPerSecond: number, messagesPerSecond: number, streamCount: number } };
 
+export type GetNeighborsQueryVariables = Exact<{
+  cursor?: InputMaybe<Scalars['String']['input']>;
+  pageSize?: InputMaybe<Scalars['Int']['input']>;
+  streamPart?: InputMaybe<Scalars['String']['input']>;
+  node?: InputMaybe<Scalars['String']['input']>;
+  streamId?: InputMaybe<Scalars['String']['input']>;
+}>;
+
+
+export type GetNeighborsQuery = { __typename?: 'Query', neighbors: { __typename?: 'Neighbors', cursor?: string | null, items: Array<{ __typename?: 'Neighbor', streamPartId: string, nodeId1: string, nodeId2: string, rtt?: number | null }> } };
+
 
 export const GetStreamsDocument = gql`
     query getStreams($streamIds: [String!], $first: Int, $orderBy: StreamOrderBy, $orderDirection: OrderDirection, $search: String, $owner: String, $cursor: String) {
@@ -192,4 +204,24 @@ export const GetGlobalStreamsStatsDocument = gql`
   }
 }
     `;
-export type GetGlobalStreamsStatsQueryResult = Apollo.QueryResult<GetGlobalStreamsStatsQuery, GetGlobalStreamsStatsQueryVariables>;
+export type GetGlobalStreamsStatsQueryResult = Apollo.QueryResult<GetGlobalStreamsStatsQuery, GetGlobalStreamsStatsQueryVariables>;
+export const GetNeighborsDocument = gql`
+    query getNeighbors($cursor: String, $pageSize: Int, $streamPart: String, $node: String, $streamId: String) {
+  neighbors(
+    cursor: $cursor
+    pageSize: $pageSize
+    streamPart: $streamPart
+    node: $node
+    stream: $streamId
+  ) {
+    items {
+      streamPartId
+      nodeId1
+      nodeId2
+      rtt
+    }
+    cursor
+  }
+}
+    `;
+export type GetNeighborsQueryResult = Apollo.QueryResult<GetNeighborsQuery, GetNeighborsQueryVariables>;

src/generated/gql/network.ts

@@ -4307,7 +4307,7 @@ export type StreamPermission = {
   /** subscribeExpires timestamp tells until what time this address may subscribe to the stream */
   subscribeExpiration?: Maybe<Scalars['BigInt']['output']>;
   /** [DEPRECATED] Ethereum address, owner of this permission; only if permission granting didn't use *forUserId functions */
-  userAddress?: Maybe<Scalars['Bytes']['output']>;
+  userAddress: Scalars['Bytes']['output'];
   /** Ethereum address or other ID, owner of this permission */
   userId: Scalars['Bytes']['output'];
 };

src/getters/getNeighbors.ts

@@ -0,0 +1,89 @@
+import { getIndexerClient } from '~/getters/getGraphClient'
+import {
+    GetNeighborsDocument,
+    GetNeighborsQuery,
+    GetNeighborsQueryVariables,
+} from '../generated/gql/indexer'
+
+interface GetNeighborsParams {
+    node?: string
+    streamId?: string
+    streamPartitionId?: string
+    chainId: number
+}
+
+interface Neighbour {
+    nodeId0: string
+    nodeId1: string
+    streamPartitionId: string
+    rtt?: number
+}
+
+export async function getNeighbors(params: GetNeighborsParams): Promise<Neighbour[]> {
+    const pageSize = 1000
+
+    const { node, streamId, streamPartitionId, chainId } = params
+
+    const items: Neighbour[] = []
+
+    const uniquenessGate: Record<string, true> = {}
+
+    let cursor = '0'
+
+    for (;;) {
+        const client = getIndexerClient(chainId)
+
+        if (!client) {
+            console.error('Could not get indexer client for chainId', chainId)
+            break
+        }
+
+        const {
+            data: { neighbors },
+        } = await client.query<GetNeighborsQuery, GetNeighborsQueryVariables>({
+            fetchPolicy: 'network-only',
+            query: GetNeighborsDocument,
+            variables: {
+                cursor,
+                node,
+                pageSize,
+                streamId,
+                streamPart: streamPartitionId,
+            },
+        })
+
+        for (const {
+            nodeId1: a,
+            nodeId2: b,
+            streamPartId: finalStreamPartitionId,
+            rtt,
+        } of neighbors.items) {
+            const pair = [a, b].sort() as [string, string]
+
+            const key = pair.join('-')
+
+            if (uniquenessGate[key]) {
+                continue
+            }
+
+            uniquenessGate[key] = true
+
+            const [nodeId0, nodeId1] = pair
+
+            items.push({
+                nodeId0,
+                nodeId1,
+                streamPartitionId: finalStreamPartitionId,
+                rtt: rtt ?? undefined,
+            })
+        }
+
+        if (!neighbors.cursor || neighbors.cursor === cursor) {
+            break
+        }
+
+        cursor = neighbors.cursor
+    }
+
+    return items
+}

src/getters/getStreamStats.ts

@@ -4,6 +4,7 @@ import {
     GetStreamsQuery,
     GetStreamsQueryVariables,
 } from '../generated/gql/indexer'
+import { getNeighbors } from './getNeighbors'
 
 export const defaultStreamStats = {
     latency: undefined,
@@ -36,9 +37,31 @@ export const getStreamStats = async (streamId: string) => {
 
     const { messagesPerSecond, peerCount } = stream
 
+    const latency = await calculateLatencyForStream(streamId, 137)
+
     return {
-        latency: undefined as undefined | number,
+        latency,
         messagesPerSecond,
         peerCount,
     }
 }
+
+export async function calculateLatencyForStream(streamId: string, chainId: number) {
+    const neighbors = await getNeighbors({
+        streamId,
+        chainId,
+    })
+
+    const validRTTs = neighbors
+        .map((n) => n.rtt)
+        .filter((rtt): rtt is number => typeof rtt === 'number' && rtt > 0)
+
+    // Calculate average one-way latency from neighbors with valid RTT.
+    // Latency is the average RTT of neighbors in the stream, divided by 2.
+    const latency =
+        validRTTs.length > 0
+            ? validRTTs.reduce((sum, rtt) => sum + rtt, 0) / validRTTs.length / 2
+            : undefined
+
+    return latency
+}

src/hooks/streams.tsx

@@ -269,7 +269,10 @@ async function getStreamsFromGraph(
 
     const streams = result.map((s) => {
         const { publisherCount, subscriberCount } = getStatsFromPermissions(
-            s.permissions || [],
+            s.permissions?.map((p) => ({
+                ...p,
+                userAddress: p.userId, // We need to add this deprecated field for now
+            })) || [],
         )
 
         return {

src/hooks/useStreamStats.tsx

@@ -25,14 +25,14 @@ export function useMultipleStreamStatsQuery(streamIds: string[]) {
             )) as StreamStats[]
             return stats.reduce(
                 (acc: StreamStats, curr: StreamStats) => ({
-                    // For latency, we can take the average of non-undefined values
-                    latency:
-                        acc.latency === undefined && curr.latency === undefined
-                            ? undefined
-                            : ((acc.latency || 0) + (curr.latency || 0)) /
-                              (acc.latency !== undefined && curr.latency !== undefined
-                                  ? 2
-                                  : 1),
+                    // Take the maximum latency among all streams
+                    latency: Math.max(
+                        acc.latency ?? -Infinity,
+                        curr.latency ?? -Infinity
+                    ) === -Infinity ? undefined : Math.max(
+                        acc.latency ?? -Infinity,
+                        curr.latency ?? -Infinity
+                    ),
                     messagesPerSecond: acc.messagesPerSecond + curr.messagesPerSecond,
                     peerCount: acc.peerCount + curr.peerCount,
                 }),

src/queries/indexer.ts

@@ -39,4 +39,28 @@ gql`
             streamCount
         }
     }
+
+    query getNeighbors(
+        $cursor: String
+        $pageSize: Int
+        $streamPart: String
+        $node: String
+        $streamId: String
+    ) {
+        neighbors(
+            cursor: $cursor
+            pageSize: $pageSize
+            streamPart: $streamPart
+            node: $node
+            stream: $streamId
+        ) {
+            items {
+                streamPartId
+                nodeId1
+                nodeId2
+                rtt
+            }
+            cursor
+        }
+    }
 `

src/services/streams.ts

@@ -170,7 +170,16 @@ export const getPagedStreams = async (
     })
 
     if (streams && streams.length > 0) {
-        return prepareStreamResult(streams, first)
+        return prepareStreamResult(
+            streams.map((s) => ({
+                ...s,
+                permissions: s.permissions?.map((p) => ({
+                    ...p,
+                    userAddress: p.userId, // We need to add this deprecated field for now
+                })),
+            })),
+            first,
+        )
     }
 
     return {