11import { setMaxListeners } from 'events'
22import take from 'it-take'
33import length from 'it-length'
4- import { QUERY_SELF_INTERVAL , QUERY_SELF_TIMEOUT , K } from './constants.js'
4+ import { QUERY_SELF_INTERVAL , QUERY_SELF_TIMEOUT , K , QUERY_SELF_INITIAL_INTERVAL } from './constants.js'
55import { anySignal } from 'any-signal'
66import { logger , Logger } from '@libp2p/logger'
77import type { PeerRouting } from './peer-routing/index.js'
88import type { Startable } from '@libp2p/interfaces/startable'
99import { pipe } from 'it-pipe'
1010import type { KadDHTComponents } from './index.js'
11+ import type { DeferredPromise } from 'p-defer'
12+ import type { RoutingTable } from './routing-table/index.js'
1113
1214export interface QuerySelfInit {
1315 lan : boolean
1416 peerRouting : PeerRouting
17+ routingTable : RoutingTable
1518 count ?: number
1619 interval ?: number
20+ initialInterval ?: number
1721 queryTimeout ?: number
22+ initialQuerySelfHasRun : DeferredPromise < void >
23+ }
24+
25+ function debounce ( func : ( ) => void , wait : number ) : ( ) => void {
26+ let timeout : ReturnType < typeof setTimeout > | undefined
27+
28+ return function ( ) {
29+ const later = function ( ) : void {
30+ timeout = undefined
31+ func ( )
32+ }
33+
34+ clearTimeout ( timeout )
35+ timeout = setTimeout ( later , wait )
36+ }
1837}
1938
2039/**
@@ -24,40 +43,51 @@ export class QuerySelf implements Startable {
2443 private readonly log : Logger
2544 private readonly components : KadDHTComponents
2645 private readonly peerRouting : PeerRouting
46+ private readonly routingTable : RoutingTable
2747 private readonly count : number
2848 private readonly interval : number
49+ private readonly initialInterval : number
2950 private readonly queryTimeout : number
51+ private started : boolean
3052 private running : boolean
3153 private timeoutId ?: NodeJS . Timer
3254 private controller ?: AbortController
55+ private initialQuerySelfHasRun ?: DeferredPromise < void >
3356
3457 constructor ( components : KadDHTComponents , init : QuerySelfInit ) {
35- const { peerRouting, lan, count, interval, queryTimeout } = init
58+ const { peerRouting, lan, count, interval, queryTimeout, routingTable } = init
3659
3760 this . components = components
3861 this . log = logger ( `libp2p:kad-dht:${ lan ? 'lan' : 'wan' } )
3962 this . running = false
63+ this . started = false
4064 this . peerRouting = peerRouting
65+ this . routingTable = routingTable
4166 this . count = count ?? K
4267 this . interval = interval ?? QUERY_SELF_INTERVAL
68+ this . initialInterval = init . initialInterval ?? QUERY_SELF_INITIAL_INTERVAL
4369 this . queryTimeout = queryTimeout ?? QUERY_SELF_TIMEOUT
70+ this . initialQuerySelfHasRun = init . initialQuerySelfHasRun
71+
72+ this . querySelf = debounce ( this . querySelf . bind ( this ) , 100 )
4473 }
4574
4675 isStarted ( ) : boolean {
47- return this . running
76+ return this . started
4877 }
4978
5079 async start ( ) : Promise < void > {
51- if ( this . running ) {
80+ if ( this . started ) {
5281 return
5382 }
5483
55- this . running = true
56- this . _querySelf ( )
84+ this . started = true
85+ clearTimeout ( this . timeoutId )
86+ this . timeoutId = setTimeout ( this . querySelf . bind ( this ) , this . initialInterval )
5787 }
5888
5989 async stop ( ) : Promise < void > {
60- this . running = false
90+ this . started = false
6191
6292 if ( this . timeoutId != null ) {
6393 clearTimeout ( this . timeoutId )
@@ -68,36 +98,76 @@ export class QuerySelf implements Startable {
6898 }
6999 }
70100
71- _querySelf ( ) : void {
72- Promise . resolve ( ) . then ( async ( ) => {
73- this . controller = new AbortController ( )
74- const signal = anySignal ( [ this . controller . signal , AbortSignal . timeout ( this . queryTimeout ) ] )
101+ querySelf ( ) : void {
102+ if ( ! this . started ) {
103+ this . log ( 'skip self-query because we are not started' )
104+ return
105+ }
75106
76- // this controller will get used for lots of dial attempts so make sure we don't cause warnings to be logged
77- try {
78- if ( setMaxListeners != null ) {
79- setMaxListeners ( Infinity , signal )
80- }
81- } catch { } // fails on node < 15.4
82-
83- try {
84- const found = await pipe (
85- this . peerRouting . getClosestPeers ( this . components . peerId . toBytes ( ) , {
86- signal
87- } ) ,
88- ( source ) => take ( source , this . count ) ,
89- async ( source ) => await length ( source )
90- )
91-
92- this . log ( 'query ran successfully - found %d peers' , found )
93- } catch ( err : any ) {
94- this . log ( 'query error' , err )
95- } finally {
96- this . timeoutId = setTimeout ( this . _querySelf . bind ( this ) , this . interval )
97- signal . clear ( )
107+ if ( this . running ) {
108+ this . log ( 'skip self-query because we are already running, will run again in %dms' , this . interval )
109+ return
110+ }
111+
112+ if ( this . routingTable . size === 0 ) {
113+ let nextInterval = this . interval
114+
115+ if ( this . initialQuerySelfHasRun != null ) {
116+ // if we've not yet run the first self query, shorten the interval until we try again
117+ nextInterval = this . initialInterval
98118 }
99- } ) . catch ( err => {
100- this . log ( 'query error' , err )
101- } )
119+
120+ this . log ( 'skip self-query because routing table is empty, will run again in %dms' , nextInterval )
121+ clearTimeout ( this . timeoutId )
122+ this . timeoutId = setTimeout ( this . querySelf . bind ( this ) , nextInterval )
123+ return
124+ }
125+
126+ this . running = true
127+
128+ Promise . resolve ( )
129+ . then ( async ( ) => {
130+ this . controller = new AbortController ( )
131+ const signal = anySignal ( [ this . controller . signal , AbortSignal . timeout ( this . queryTimeout ) ] )
132+
133+ // this controller will get used for lots of dial attempts so make sure we don't cause warnings to be logged
134+ try {
135+ if ( setMaxListeners != null ) {
136+ setMaxListeners ( Infinity , signal )
137+ }
138+ } catch { } // fails on node < 15.4
139+
140+ try {
141+ this . log ( 'run self-query, look for %d peers timing out after %dms' , this . count , this . queryTimeout )
142+
143+ const found = await pipe (
144+ this . peerRouting . getClosestPeers ( this . components . peerId . toBytes ( ) , {
145+ signal,
146+ isSelfQuery : true
147+ } ) ,
148+ ( source ) => take ( source , this . count ) ,
149+ async ( source ) => await length ( source )
150+ )
151+
152+ this . log ( 'self-query ran successfully - found %d peers' , found )
153+
154+ if ( this . initialQuerySelfHasRun != null ) {
155+ this . initialQuerySelfHasRun . resolve ( )
156+ this . initialQuerySelfHasRun = undefined
157+ }
158+ } catch ( err : any ) {
159+ this . log . error ( 'self-query error' , err )
160+ } finally {
161+ signal . clear ( )
162+ }
163+ } ) . catch ( err => {
164+ this . log ( 'self-query error' , err )
165+ } ) . finally ( ( ) => {
166+ this . running = false
167+
168+ this . log ( 'running self-query again in %dms' , this . interval )
169+ clearTimeout ( this . timeoutId )
170+ this . timeoutId = setTimeout ( this . querySelf . bind ( this ) , this . interval )
171+ } )
102172 }
103173}
0 commit comments