Add stream latency calculation from neighbour RTTs

src/components/Stats.tsx

@@ -29,6 +29,7 @@ import { Interval } from './Graphs/Graphs'
 import { Intervals } from './Graphs/Intervals'
 import { TimeSeries } from './Graphs/TimeSeries'
 import { useStore } from '../Store'
+import { getNeighbors } from '../getters'
 
 type StatProps = {
   id: string
@@ -264,8 +265,24 @@ function useStreamStatsQuery(chainId: number, streamId: string) {
 
       const { messagesPerSecond, peerCount } = stream
 
+      const neighbors = await getNeighbors({
+        streamId: stream.id,
+        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: undefined as undefined | number,
+        latency,
         messagesPerSecond,
         peerCount,
       }
@@ -297,7 +314,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)}
       />
     </Stats>
   )

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'];
 };
 
@@ -171,7 +172,7 @@ export type GetNeighborsQueryVariables = Exact<{
 }>;
 
 
-export type GetNeighborsQuery = { __typename?: 'Query', neighbors: { __typename?: 'Neighbors', cursor?: string | null, items: Array<{ __typename?: 'Neighbor', streamPartId: string, nodeId1: string, nodeId2: string }> } };
+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 GetNodesDocument = gql`
@@ -229,6 +230,7 @@ export const GetNeighborsDocument = gql`
       streamPartId
       nodeId1
       nodeId2
+      rtt
     }
     cursor
   }

src/getters.tsx

@@ -77,7 +77,7 @@ interface GetNeighborsParams {
 export async function getNeighbors(params: GetNeighborsParams): Promise<Neighbour[]> {
   const pageSize = 1000
 
-  const { node, streamPartitionId, chainId } = params
+  const { node, streamId, streamPartitionId, chainId } = params
 
   const items: Neighbour[] = []
 
@@ -95,6 +95,7 @@ export async function getNeighbors(params: GetNeighborsParams): Promise<Neighbou
         cursor,
         node,
         pageSize,
+        streamId,
         streamPart: streamPartitionId,
       },
     })
@@ -103,6 +104,7 @@ export async function getNeighbors(params: GetNeighborsParams): Promise<Neighbou
       nodeId1: a,
       nodeId2: b,
       streamPartId: finalStreamPartitionId,
+      rtt,
     } of neighbors.items) {
       const pair = [a, b].sort() as [string, string]
 
@@ -120,6 +122,7 @@ export async function getNeighbors(params: GetNeighborsParams): Promise<Neighbou
         nodeId0,
         nodeId1,
         streamPartitionId: finalStreamPartitionId,
+        rtt: rtt ?? undefined,
       })
     }
 

src/queries/indexer.ts

@@ -65,6 +65,7 @@ gql`
         streamPartId
         nodeId1
         nodeId2
+        rtt
       }
       cursor
     }

src/types/index.ts

@@ -24,6 +24,7 @@ export interface Neighbour {
   nodeId0: string
   nodeId1: string
   streamPartitionId: string
+  rtt?: number
 }
 
 interface Stream {