fix: use keep-alive as a tag prefix

achingbrain · achingbrain · commit c3ff22ab2aec · 2024-10-09T12:34:56.000+01:00
To prevent components removing each other's keep-alive tag, treat
it as a prefix.
diff --git a/packages/interface/src/peer-store/tags.ts b/packages/interface/src/peer-store/tags.ts
@@ -1 +1,9 @@
+/**
+ * When a peer that is tagged with this prefix disconnects, we will attempt to
+ * redial it, up to a limit.
+ *
+ * To allow multiple components to add/remove their own keep-alive tags without
+ * accidentally overwriting those of other components, attach a unique suffix to
+ * the tag, e.g. `keep-alive-circuit-relay` or `keep-alive-kad-dht`, etc.
+ */
 export const KEEP_ALIVE = 'keep-alive'
diff --git a/packages/kad-dht/src/constants.ts b/packages/kad-dht/src/constants.ts
@@ -5,6 +5,8 @@
 // until the year 2020 (a great time in the future). For that record to stick around
 // it must be rebroadcasted more frequently than once every 'MaxRecordAge'
 
+import { KEEP_ALIVE } from '@libp2p/interface'
+
 export const second = 1000
 export const minute = 60 * second
 export const hour = 60 * minute
@@ -51,3 +53,6 @@ export const TABLE_REFRESH_QUERY_TIMEOUT = 30 * second
 
 // When a timeout is not specified, run a query for this long
 export const DEFAULT_QUERY_TIMEOUT = 180 * second
+
+// used to ensure connections to our closest peers remain open
+export const KEEP_ALIVE_TAG = `${KEEP_ALIVE}-kad-dht`
diff --git a/packages/kad-dht/src/routing-table/closest-peers.ts b/packages/kad-dht/src/routing-table/closest-peers.ts
@@ -1,5 +1,5 @@
-import { KEEP_ALIVE } from '@libp2p/interface'
 import { PeerSet } from '@libp2p/peer-collections'
+import { KEEP_ALIVE_TAG } from '../constants.js'
 import { PeerDistanceList } from '../peer-distance-list.js'
 import { convertPeerId } from '../utils.js'
 import type { RoutingTable } from './index.js'
@@ -94,7 +94,7 @@ export class ClosestPeers implements Startable {
             [this.closeTagName]: {
               value: this.closeTagValue
             },
-            [KEEP_ALIVE]: {
+            [KEEP_ALIVE_TAG]: {
               value: 1
             }
           }
@@ -104,7 +104,7 @@ export class ClosestPeers implements Startable {
         await this.components.peerStore.merge(peerId, {
           tags: {
             [this.closeTagName]: undefined,
-            [KEEP_ALIVE]: undefined
+            [KEEP_ALIVE_TAG]: undefined
           }
         })
       })
diff --git a/packages/kad-dht/test/closest-peers.spec.ts b/packages/kad-dht/test/closest-peers.spec.ts
@@ -12,6 +12,7 @@ import { convertPeerId } from '../src/utils.js'
 import type { RoutingTable } from '../src/routing-table/index.js'
 import type { ComponentLogger, PeerId, PeerStore } from '@libp2p/interface'
 import type { StubbedInstance } from 'sinon-ts'
+import { KEEP_ALIVE_TAG } from '../src/constants.js'
 
 interface ClosestPeersComponents {
   peerId: PeerId
@@ -112,7 +113,7 @@ function assertTagged (peerId: PeerId, peerStore: StubbedInstance<PeerStore>): v
       'kad-close': {
         value: 50
       },
-      [KEEP_ALIVE]: {
+      [KEEP_ALIVE_TAG]: {
         value: 1
       }
     }
@@ -123,7 +124,7 @@ function assertUnTagged (peerId: PeerId, peerStore: StubbedInstance<PeerStore>):
   expect(peerStore.merge.calledWith(peerId, {
     tags: {
       'kad-close': undefined,
-      [KEEP_ALIVE]: undefined
+      [KEEP_ALIVE_TAG]: undefined
     }
   })).to.be.true()
 }
diff --git a/packages/libp2p/src/connection-manager/reconnect-queue.ts b/packages/libp2p/src/connection-manager/reconnect-queue.ts
@@ -63,7 +63,7 @@ export class ReconnectQueue implements Startable {
 
     const peer = await this.peerStore.get(peerId)
 
-    if (!peer.tags.has(KEEP_ALIVE)) {
+    if (!hasKeepAliveTag(peer)) {
       return
     }
 
@@ -94,8 +94,23 @@ export class ReconnectQueue implements Startable {
     }, {
       peerId
     })
-      .catch(err => {
-        this.log.error('failed to reconnect to %p', peerId, err)
+      .catch(async err => {
+        this.log.error('failed to reconnect to %p - %e', peerId, err)
+
+        const tags: Record<string, undefined> = {}
+
+        ;[...peer.tags.keys()].forEach(key => {
+          if (key.startsWith(KEEP_ALIVE)) {
+            tags[key] = undefined
+          }
+        })
+
+        await this.peerStore.merge(peerId, {
+          tags
+        })
+      })
+      .catch(async err => {
+        this.log.error('failed to remove keep-alive tag from %p - %e', peerId, err)
       })
   }
 
@@ -108,9 +123,9 @@ export class ReconnectQueue implements Startable {
     void Promise.resolve()
       .then(async () => {
         const keepAlivePeers: Peer[] = await this.peerStore.all({
-          filters: [(peer) => {
-            return peer.tags.has(KEEP_ALIVE)
-          }]
+          filters: [
+            (peer) => hasKeepAliveTag(peer)
+          ]
         })
 
         await Promise.all(
@@ -132,3 +147,13 @@ export class ReconnectQueue implements Startable {
     this.queue.abort()
   }
 }
+
+function hasKeepAliveTag (peer: Peer): boolean {
+  for (const tag of peer.tags.keys()) {
+    if (tag.startsWith(KEEP_ALIVE)) {
+      return true
+    }
+  }
+
+  return false
+}
diff --git a/packages/libp2p/test/connection-manager/reconnect-queue.spec.ts b/packages/libp2p/test/connection-manager/reconnect-queue.spec.ts
@@ -115,4 +115,43 @@ describe('reconnect queue', () => {
 
     expect(components.connectionManager.openConnection.calledWith(nonKeepAlivePeer)).to.be.false()
   })
+
+  it('should remove KEEP_ALIVE tags when reconnecting fails', async () => {
+    queue = new ReconnectQueue(components, {
+      retries: 1,
+      retryInterval: 10,
+      backoffFactor: 1
+    })
+
+    const keepAlivePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519'))
+
+    components.peerStore.all.resolves([])
+    components.peerStore.get.withArgs(keepAlivePeer).resolves(
+      stubInterface<Peer>({
+        id: keepAlivePeer,
+        tags: new Map([[KEEP_ALIVE, {
+          value: 1
+        }]])
+      })
+    )
+
+    await start(queue)
+
+    components.connectionManager.openConnection.withArgs(keepAlivePeer).rejects(new Error('Dial failed'))
+
+    components.events.safeDispatchEvent('peer:disconnect', new CustomEvent('peer:disconnect', {
+      detail: keepAlivePeer
+    }))
+
+    await pRetry(() => {
+      expect(components.peerStore.merge.calledWith(keepAlivePeer, {
+        tags: {
+          [KEEP_ALIVE]: undefined
+        }
+      })).to.be.true()
+    }, {
+      retries: 5,
+      factor: 1
+    })
+  })
 })
diff --git a/packages/transport-circuit-relay-v2/src/constants.ts b/packages/transport-circuit-relay-v2/src/constants.ts
@@ -1,3 +1,5 @@
+import { KEEP_ALIVE } from '@libp2p/interface'
+
 const second = 1000
 const minute = 60 * second
 
@@ -40,6 +42,8 @@ export const RELAY_SOURCE_TAG = 'circuit-relay-source'
 
 export const RELAY_TAG = 'circuit-relay-relay'
 
+export const KEEP_ALIVE_TAG = `${KEEP_ALIVE}-circuit-relay`
+
 // circuit v2 connection limits
 // https://github.com/libp2p/go-libp2p/blob/master/p2p/protocol/circuitv2/relay/resources.go#L61-L66
 
diff --git a/packages/transport-circuit-relay-v2/src/transport/reservation-store.ts b/packages/transport-circuit-relay-v2/src/transport/reservation-store.ts
@@ -1,11 +1,11 @@
-import { KEEP_ALIVE, TypedEventEmitter, setMaxListeners } from '@libp2p/interface'
+import { TypedEventEmitter, setMaxListeners } from '@libp2p/interface'
 import { PeerMap } from '@libp2p/peer-collections'
 import { createBloomFilter } from '@libp2p/utils/filters'
 import { PeerQueue } from '@libp2p/utils/peer-queue'
 import { multiaddr } from '@multiformats/multiaddr'
 import { pbStream } from 'it-protobuf-stream'
 import { equals as uint8ArrayEquals } from 'uint8arrays/equals'
-import { DEFAULT_MAX_RESERVATION_QUEUE_LENGTH, DEFAULT_RESERVATION_COMPLETION_TIMEOUT, DEFAULT_RESERVATION_CONCURRENCY, RELAY_TAG, RELAY_V2_HOP_CODEC } from '../constants.js'
+import { DEFAULT_MAX_RESERVATION_QUEUE_LENGTH, DEFAULT_RESERVATION_COMPLETION_TIMEOUT, DEFAULT_RESERVATION_CONCURRENCY, KEEP_ALIVE_TAG, RELAY_TAG, RELAY_V2_HOP_CODEC } from '../constants.js'
 import { HopMessage, Status } from '../pb/index.js'
 import { getExpirationMilliseconds } from '../utils.js'
 import type { Reservation } from '../pb/index.js'

Original file line number	Diff line number	Diff line change
`@@ -12,6 +12,7 @@ import { convertPeerId } from '../src/utils.js'`
`12`	`12`	`import type { RoutingTable } from '../src/routing-table/index.js'`
`13`	`13`	`import type { ComponentLogger, PeerId, PeerStore } from '@libp2p/interface'`
`14`	`14`	`import type { StubbedInstance } from 'sinon-ts'`
	`15`	`+import { KEEP_ALIVE_TAG } from '../src/constants.js'`
`15`	`16`
`16`	`17`	`interface ClosestPeersComponents {`
`17`	`18`	`peerId: PeerId`
`@@ -112,7 +113,7 @@ function assertTagged (peerId: PeerId, peerStore: StubbedInstance<PeerStore>): v`
`112`	`113`	`'kad-close': {`
`113`	`114`	`value: 50`
`114`	`115`	`},`
`115`		`- [KEEP_ALIVE]: {`
	`116`	`+ [KEEP_ALIVE_TAG]: {`
`116`	`117`	`value: 1`
`117`	`118`	`}`
`118`	`119`	`}`
`@@ -123,7 +124,7 @@ function assertUnTagged (peerId: PeerId, peerStore: StubbedInstance<PeerStore>):`
`123`	`124`	`expect(peerStore.merge.calledWith(peerId, {`
`124`	`125`	`tags: {`
`125`	`126`	`'kad-close': undefined,`
`126`		`- [KEEP_ALIVE]: undefined`
	`127`	`+ [KEEP_ALIVE_TAG]: undefined`
`127`	`128`	`}`
`128`	`129`	`})).to.be.true()`
`129`	`130`	`}`