fix(kad-dht): sort reprovide CIDs in bounded batches to avoid OOM

Replaces the unbounded toReprovide array with a fixed-size batch
approach. CIDs are accumulated into a batch of at most sortBatchSize
(default 512) entries; each full batch is sorted by Kademlia key and
queued before the next batch begins. A final flush handles the
remainder. This bounds peak memory to O(sortBatchSize) regardless of
the total number of stored CIDs, while still delivering the connection-
reuse benefit for XOR-adjacent CIDs within each batch.

Also adds sortBatchSize to ReproviderInit and a test that verifies
within-batch Kademlia ordering when batches are smaller than the CID
count.

Author: paschal533

packages/kad-dht/src/constants.ts

@@ -28,6 +28,9 @@ export const REPROVIDE_INTERVAL = hour
 // How long to reprovide for
 export const REPROVIDE_TIMEOUT = hour
 
+// How many CIDs to sort at once during reprovide (bounds memory use)
+export const REPROVIDE_SORT_BATCH_SIZE = 512
+
 export const READ_MESSAGE_TIMEOUT = 10 * second
 
 // How long to process newly connected peers for

packages/kad-dht/src/reprovider.ts

@@ -1,7 +1,7 @@
 import { AdaptiveTimeout, Queue } from '@libp2p/utils'
 import drain from 'it-drain'
 import { TypedEventEmitter, setMaxListeners } from 'main-event'
-import { PROVIDERS_VALIDITY, REPROVIDE_CONCURRENCY, REPROVIDE_INTERVAL, REPROVIDE_MAX_QUEUE_SIZE, REPROVIDE_THRESHOLD, REPROVIDE_TIMEOUT } from './constants.ts'
+import { PROVIDERS_VALIDITY, REPROVIDE_CONCURRENCY, REPROVIDE_INTERVAL, REPROVIDE_MAX_QUEUE_SIZE, REPROVIDE_SORT_BATCH_SIZE, REPROVIDE_THRESHOLD, REPROVIDE_TIMEOUT } from './constants.ts'
 import { convertBuffer, parseProviderKey, readProviderTime, timeOperationMethod } from './utils.ts'
 import type { ContentRouting } from './content-routing/index.ts'
 import type { OperationMetrics } from './kad-dht.ts'
@@ -27,6 +27,7 @@ export interface ReproviderInit {
   operationMetrics: OperationMetrics
   concurrency?: number
   maxQueueSize?: number
+  sortBatchSize?: number
   threshold?: number
   validity?: number
   interval?: number
@@ -52,6 +53,7 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
   private running: boolean
   private shutdownController?: AbortController
   private readonly reprovideThreshold: number
+  private readonly sortBatchSize: number
   private readonly contentRouting: ContentRouting
   private readonly datastorePrefix: string
   private readonly addressManager: AddressManager
@@ -78,6 +80,7 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
     this.addressManager = components.addressManager
     this.datastorePrefix = `${init.datastorePrefix}/provider`
     this.reprovideThreshold = init.threshold ?? REPROVIDE_THRESHOLD
+    this.sortBatchSize = init.sortBatchSize ?? REPROVIDE_SORT_BATCH_SIZE
     this.maxQueueSize = init.maxQueueSize ?? REPROVIDE_MAX_QUEUE_SIZE
     this.validity = init.validity ?? PROVIDERS_VALIDITY
     this.interval = init.interval ?? REPROVIDE_INTERVAL
@@ -121,14 +124,18 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
    * reprovided consecutively. Since nearby CIDs in the keyspace share the
    * same K closest peers, connections opened for one CID are likely to be
    * reused for the next, reducing the number of new dials per reprovide run.
+   *
+   * To avoid unbounded memory growth, CIDs are sorted and queued in batches
+   * of at most `sortBatchSize` entries.
    */
   private async processRecords (options?: AbortOptions): Promise<void> {
     try {
       this.safeDispatchEvent('reprovide:start')
       this.log('starting reprovide/cleanup')
 
-      // collect CIDs that need reproviding so we can sort them before queueing
-      const toReprovide: CID[] = []
+      // Accumulate CIDs for batched Kademlia-key sorting. Flushed every
+      // sortBatchSize entries so the array never grows without bound.
+      const batch: CID[] = []
 
       // Get all provider entries from the datastore
       for await (const entry of this.datastore.query({
@@ -155,40 +162,19 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
           // collect for reproviding
           if (this.shouldReprovide(isSelf, expires)) {
             this.log('scheduling reprovide of %c', cid)
-            toReprovide.push(cid)
+            batch.push(cid)
+
+            if (batch.length >= this.sortBatchSize) {
+              await this.sortAndQueueBatch(batch.splice(0), options)
+            }
           }
         } catch (err: any) {
           this.log.error('error processing datastore key %s - %s', entry.key, err.message)
         }
       }
 
-      // sort collected CIDs by their Kademlia key so XOR-adjacent CIDs are
-      // queued consecutively — peers responsible for one CID are likely to
-      // also be responsible for adjacent CIDs, so connections can be reused
-      if (toReprovide.length > 1) {
-        const kadKeys = await Promise.all(
-          toReprovide.map(cid => convertBuffer(cid.multihash.bytes, options))
-        )
-
-        const sortable = toReprovide.map((cid, i) => ({ cid, kadKey: kadKeys[i] }))
-        sortable.sort((a, b) => {
-          for (let i = 0; i < a.kadKey.length; i++) {
-            if (a.kadKey[i] !== b.kadKey[i]) {
-              return a.kadKey[i] - b.kadKey[i]
-            }
-          }
-          return 0
-        })
-
-        toReprovide.splice(0, toReprovide.length, ...sortable.map(({ cid }) => cid))
-      }
-
-      // queue reprovides in Kademlia key order
-      for (const cid of toReprovide) {
-        this.queueReprovide(cid)
-          .catch(err => {
-            this.log.error('could not reprovide %c - %e', cid, err)
-          })
+      if (batch.length > 0) {
+        await this.sortAndQueueBatch(batch, options)
       }
 
       this.log('reprovide/cleanup successful')
@@ -208,6 +194,33 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
     }
   }
 
+  private async sortAndQueueBatch (batch: CID[], options?: AbortOptions): Promise<void> {
+    if (batch.length > 1) {
+      const kadKeys = await Promise.all(
+        batch.map(cid => convertBuffer(cid.multihash.bytes, options))
+      )
+
+      const sortable = batch.map((cid, i) => ({ cid, kadKey: kadKeys[i] }))
+      sortable.sort((a, b) => {
+        for (let i = 0; i < a.kadKey.length; i++) {
+          if (a.kadKey[i] !== b.kadKey[i]) {
+            return a.kadKey[i] - b.kadKey[i]
+          }
+        }
+        return 0
+      })
+
+      batch = sortable.map(({ cid }) => cid)
+    }
+
+    for (const cid of batch) {
+      this.queueReprovide(cid)
+        .catch(err => {
+          this.log.error('could not reprovide %c - %e', cid, err)
+        })
+    }
+  }
+
   /**
    * Determines if a record should be reprovided
    */

packages/kad-dht/test/reprovider.spec.ts

@@ -197,6 +197,7 @@ describe('reprovider', () => {
     }
 
     // recreate reprovider with concurrency=1 so provides are strictly sequential
+    // sortBatchSize >= cids.length so all CIDs are sorted in one batch
     reprovider = new Reprovider(components, {
       logPrefix: 'libp2p',
       datastorePrefix: '/dht',
@@ -206,6 +207,7 @@ describe('reprovider', () => {
       validity: 200,
       interval: 200,
       concurrency: 1,
+      sortBatchSize: cids.length,
       operationMetrics: {}
     })
 
@@ -254,6 +256,81 @@ describe('reprovider', () => {
     }
   })
 
+  it('should sort within each batch when CID count exceeds sortBatchSize', async function () {
+    this.timeout(5000)
+
+    const cids = [
+      CID.parse('QmZ8eiDPqQqWR17EPxiwCDgrKPVhCHLcyn6xSCNpFAdAZb'),
+      CID.parse('QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1n'),
+      CID.parse('QmRgutAxd8t7oGkSm4wmeuByG6M51wcTso6cubDdQtuEfL'),
+      CID.parse('QmPZ9gcCEpqKTo6aq61g2nXGUhM4iCL3ewB6LDXZCtioEB'),
+      CID.parse('QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN')
+    ]
+
+    for (const cid of cids) {
+      await providers.addProvider(cid, components.peerId)
+    }
+
+    // sortBatchSize=2: CIDs are sorted in batches of 2. Ordering within each
+    // batch is verified; cross-batch ordering is intentionally not guaranteed.
+    reprovider = new Reprovider(components, {
+      logPrefix: 'libp2p',
+      datastorePrefix: '/dht',
+      metricsPrefix: '',
+      contentRouting,
+      threshold: 100,
+      validity: 200,
+      interval: 200,
+      concurrency: 1,
+      sortBatchSize: 2,
+      operationMetrics: {}
+    })
+
+    const provisionMultihashes: Uint8Array[] = []
+
+    let resolveWhenDone!: () => void
+    const whenAllDone = new Promise<void>(resolve => { resolveWhenDone = resolve })
+    let provided = 0
+
+    contentRouting.provide.callsFake(async function * (cid: CID) {
+      provisionMultihashes.push(cid.multihash.bytes)
+      provided++
+      if (provided === cids.length) {
+        resolveWhenDone()
+      }
+      yield * []
+    })
+
+    await start(reprovider)
+    await pEvent(reprovider, 'reprovide:start')
+    await pEvent(reprovider, 'reprovide:end')
+    await whenAllDone
+
+    expect(provisionMultihashes).to.have.lengthOf(cids.length)
+
+    // verify ordering within each batch of sortBatchSize
+    const batchSize = 2
+    for (let b = 0; b < provisionMultihashes.length; b += batchSize) {
+      const batchEnd = Math.min(b + batchSize, provisionMultihashes.length)
+      for (let i = b + 1; i < batchEnd; i++) {
+        const prevKey = await convertBuffer(provisionMultihashes[i - 1])
+        const currKey = await convertBuffer(provisionMultihashes[i])
+
+        let comparison = 0
+        for (let j = 0; j < prevKey.length; j++) {
+          if (prevKey[j] !== currKey[j]) {
+            comparison = prevKey[j] - currKey[j]
+            break
+          }
+        }
+
+        expect(comparison).to.be.lessThanOrEqual(0,
+          `within batch: CID at position ${i - 1} should have a smaller or equal Kademlia key than position ${i}`
+        )
+      }
+    }
+  })
+
   describe('shouldReprovide', () => {
     it('should return false for non-self providers', () => {
       const expires = Date.now() + 50