feat(observability,encoding,network,pubsub): expand support category modules

- stats-exporter: snapshot history, compare, filter
- health-probe: unhealthyPeers, averageRtt, pruneStale
- schema-validator: assertValid, validateBatch, maxLength rules
- compact-codec-bridge: encodeBatch, listCodecs
- discovery-health: topPeers, pruneStale, averageRtt
- peer-bootstrap-store: removeBootstrap, findByHint
- retained-messages: retainBatch, totalRetained, channelCount
- ci: restore flat-repo production test workflow

Co-authored-by: Cursor <[email protected]>

.github/workflows/test.yml

@@ -0,0 +1,19 @@
+name: Hyper-P2P Module Tests
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  unit-all:
+    runs-on: ubuntu-latest
+    timeout-minutes: 90
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: '20'
+      - name: Run production unit tests
+        run: |
+          chmod +x scripts/run-module.sh scripts/run-all-production.sh
+          ./scripts/run-all-production.sh --tier=production

encoding-wire/hyper-p2p-compact-codec-bridge/index.js

@@ -9,10 +9,11 @@ class HyperP2PCompactCodecBridge extends EventEmitter {
   constructor (opts = {}) {
     super()
     this.registry = opts.registry || new HyperP2PWireRegistry()
-    this._stats = { encode: 0, decode: 0 }
+    this._stats = { encode: 0, decode: 0, frames: 0 }
     if (!opts.registry) {
       this.registry.registerCodec('json', c.json)
       this.registry.registerCodec('string', c.string)
+      this.registry.registerCodec('uint32', c.uint32)
     }
   }
 
@@ -30,19 +31,35 @@ class HyperP2PCompactCodecBridge extends EventEmitter {
 
   encodeFrame (protocolId, codecId, value) {
     const body = this.encode(codecId, value)
-    return { protocolId, codecId, body }
+    const frame = { protocolId, codecId, body, at: Date.now() }
+    this._stats.frames++
+    return frame
   }
 
   decodeFrame (frame) {
     if (!frame || !frame.codecId || !frame.body) throw new Error('invalid frame')
     return {
       protocolId: frame.protocolId || null,
-      value: this.decode(frame.codecId, frame.body)
+      value: this.decode(frame.codecId, frame.body),
+      at: frame.at
     }
   }
 
+  encodeBatch (codecId, values) {
+    if (!Array.isArray(values)) throw new Error('values must be an array')
+    return values.map((v) => this.encode(codecId, v))
+  }
+
+  listCodecs () {
+    return this.registry.listCodecs()
+  }
+
   getStats () {
-    return { ...this._stats, protocol: PROTOCOL }
+    return {
+      ...this._stats,
+      codecs: this.registry.listCodecs().length,
+      protocol: PROTOCOL
+    }
   }
 
   async ready () {
@@ -52,6 +69,7 @@ class HyperP2PCompactCodecBridge extends EventEmitter {
 
   async close () {
     await this.registry.close()
+    this.emit('closed')
   }
 }
 

encoding-wire/hyper-p2p-schema-validator/index.js

@@ -9,18 +9,36 @@ class HyperP2PSchemaValidator extends EventEmitter {
     super()
     this._schemas = new Map()
     this._stats = { validated: 0, failed: 0 }
+    this.strict = opts.strict !== false
   }
 
   register (name, schema) {
     assertNonEmpty(name, 'name')
     if (!schema || typeof schema !== 'object') throw new Error('schema object required')
     this._schemas.set(name, schema)
+    this.emit('register', { name })
     return true
   }
 
-  listSchemas () { return [...this._schemas.keys()] }
+  unregister (name) {
+    return this._schemas.delete(name)
+  }
 
-  hasSchema (name) { return this._schemas.has(name) }
+  listSchemas () {
+    return [...this._schemas.keys()]
+  }
+
+  hasSchema (name) {
+    return this._schemas.has(name)
+  }
+
+  _checkValue (key, expected, value, errors) {
+    if (expected === 'array') {
+      if (!Array.isArray(value)) errors.push(`${key} must be array`)
+      return
+    }
+    if (typeof value !== expected) errors.push(`${key} must be ${expected}`)
+  }
 
   validate (name, value) {
     const schema = this._schemas.get(name)
@@ -33,25 +51,50 @@ class HyperP2PSchemaValidator extends EventEmitter {
     }
     if (schema.types) {
       for (const [key, type] of Object.entries(schema.types)) {
-        if (value && value[key] !== undefined && typeof value[key] !== type) {
+        if (value && value[key] !== undefined) {
-          errors.push(`${key} must be ${type}`)
+          this._checkValue(key, type, value[key], errors)
         }
       }
     }
+    if (schema.maxLength && value) {
+      for (const [key, max] of Object.entries(schema.maxLength)) {
+        const v = value[key]
+        if (typeof v === 'string' && v.length > max) errors.push(`${key} too long`)
+      }
+    }
     if (errors.length) {
       this._stats.failed++
+      this.emit('fail', { name, errors })
       return { ok: false, errors }
     }
     this._stats.validated++
+    this.emit('pass', { name })
     return { ok: true }
   }
 
+  assertValid (name, value) {
+    const r = this.validate(name, value)
+    if (!r.ok) throw new Error(r.errors.join('; '))
+    return value
+  }
+
+  validateBatch (name, values) {
+    if (!Array.isArray(values)) throw new Error('values must be an array')
+    return values.map((v) => this.validate(name, v))
+  }
+
   getStats () {
     return { ...this._stats, schemas: this._schemas.size, protocol: PROTOCOL }
   }
 
-  async ready () { return this }
+  async ready () {
-  async close () { this._schemas.clear() }
+    return this
+  }
+
+  async close () {
+    this._schemas.clear()
+    this.emit('closed')
+  }
 }
 
 module.exports = { HyperP2PSchemaValidator, PROTOCOL }

messaging-pubsub/hyper-p2p-retained-messages/index.js

@@ -24,6 +24,11 @@ class HyperP2PRetainedMessages extends EventEmitter {
     return entry
   }
 
+  retainBatch (channel, items) {
+    if (!Array.isArray(items)) throw new Error('items must be an array')
+    return items.map((item) => this.retain(channel, item.payload, item.meta || {}))
+  }
+
   latest (channel) {
     const list = this._store.get(channel)
     if (!list || !list.length) return null
@@ -33,34 +38,56 @@ class HyperP2PRetainedMessages extends EventEmitter {
 
   list (channel, limit = 10) {
     const list = this._store.get(channel) || []
-    return list.slice(-limit)
+    return list.slice(-Math.max(0, limit | 0))
   }
 
   clear (channel) {
     if (channel == null) {
+      const n = this._store.size
       this._store.clear()
-      this._stats.cleared++
+      this._stats.cleared += n
-      return true
+      return n
     }
     const ok = this._store.delete(channel)
     if (ok) this._stats.cleared++
     return ok
   }
 
-  listChannels () { return [...this._store.keys()] }
+  channelCount () {
+    return this._store.size
+  }
 
-  hasChannel (channel) { return this._store.has(channel) }
+  totalRetained () {
+    let n = 0
+    for (const list of this._store.values()) n += list.length
+    return n
+  }
+
+  listChannels () {
+    return [...this._store.keys()]
+  }
+
+  hasChannel (channel) {
+    return this._store.has(channel)
+  }
 
   getStats () {
     return {
       ...this._stats,
       channels: this._store.size,
+      total: this.totalRetained(),
       protocol: PROTOCOL
     }
   }
 
-  async ready () { return this }
+  async ready () {
-  async close () { this._store.clear() }
+    return this
+  }
+
+  async close () {
+    this._store.clear()
+    this.emit('closed')
+  }
 }
 
 module.exports = { HyperP2PRetainedMessages, PROTOCOL }

network-discovery/hyper-p2p-discovery-health/index.js

@@ -10,8 +10,9 @@ class HyperP2PDiscoveryHealth extends EventEmitter {
     super()
     this.topic = opts.topic || null
     this.keyPair = opts.keyPair || require('hypercore-crypto').keyPair()
+    this.minScore = opts.minScore ?? 0.5
     this._peers = new Map()
-    this._stats = { reports: 0, gossipIn: 0, gossipOut: 0 }
+    this._stats = { reports: 0, gossipIn: 0, gossipOut: 0, pruned: 0 }
     this.swarm = null
     this._peerMsgs = null
   }
@@ -38,7 +39,38 @@ class HyperP2PDiscoveryHealth extends EventEmitter {
   }
 
   healthyPeers () {
-    return [...this._peers.values()].filter((e) => e.ok)
+    return [...this._peers.values()].filter((e) => e.ok && e.score >= this.minScore)
+  }
+
+  unhealthyPeers () {
+    return [...this._peers.values()].filter((e) => !e.ok || e.score < this.minScore)
+  }
+
+  topPeers (n = 5) {
+    return [...this._peers.values()]
+      .filter((e) => e.ok)
+      .sort((a, b) => b.score - a.score || a.rttMs - b.rttMs)
+      .slice(0, Math.max(0, n | 0))
+  }
+
+  pruneStale (maxAgeMs = 300000) {
+    const cutoff = Date.now() - maxAgeMs
+    let n = 0
+    for (const [id, e] of this._peers) {
+      if ((e.at || 0) < cutoff) {
+        this._peers.delete(id)
+        n++
+      }
+    }
+    this._stats.pruned += n
+    return n
+  }
+
+  averageRtt () {
+    const peers = this.healthyPeers()
+    if (!peers.length) return null
+    const sum = peers.reduce((a, p) => a + (p.rttMs || 0), 0)
+    return sum / peers.length
   }
 
   _gossip (data) {

network-discovery/hyper-p2p-peer-bootstrap-store/index.js

@@ -41,6 +41,23 @@ class HyperP2PPeerBootstrapStore extends EventEmitter {
     return [...this._bootstraps.values()]
   }
 
+  removeBootstrap (peerId) {
+    const ok = this._bootstraps.delete(peerId)
+    if (ok) {
+      this._gossip({ type: 'bootstrap-remove', peerId, at: Date.now() })
+      this.emit('remove', { peerId })
+    }
+    return ok
+  }
+
+  findByHint (key, value) {
+    return this.allBootstraps().filter((e) => e.hints && e.hints[key] === value)
+  }
+
+  bootstrapCount () {
+    return this._bootstraps.size
+  }
+
   _gossip (data) {
     if (!this._peerMsgs) return
     gossipSend(this, data)
@@ -48,6 +65,10 @@ class HyperP2PPeerBootstrapStore extends EventEmitter {
   }
 
   _onGossip (data, peerInfo) {
+    if (data.type === 'bootstrap-remove' && data.peerId) {
+      this._bootstraps.delete(data.peerId)
+      return
+    }
     if (!data || data.type !== 'bootstrap-add' || !data.peerId) return
     this._stats.gossipIn++
     const from = data.from || (peerInfo && peerInfo.publicKey

observability/hyper-p2p-health-probe/index.js

@@ -35,6 +35,28 @@ class HyperP2PHealthProbe extends EventEmitter {
     return [...this._reports.values()].filter((r) => r.ok)
   }
 
+  unhealthyPeers () {
+    return [...this._reports.values()].filter((r) => !r.ok)
+  }
+
+  averageRtt () {
+    const ok = this.healthyPeers()
+    if (!ok.length) return null
+    return ok.reduce((a, p) => a + (p.rttMs || 0), 0) / ok.length
+  }
+
+  pruneStale (maxAgeMs = 300000) {
+    const cutoff = Date.now() - maxAgeMs
+    let n = 0
+    for (const [id, r] of this._reports) {
+      if ((r.at || 0) < cutoff) {
+        this._reports.delete(id)
+        n++
+      }
+    }
+    return n
+  }
+
   _onGossip (data) {
     if (!data || data.type !== 'health' || !data.entry) return
     this._stats.gossipIn++

observability/hyper-p2p-stats-exporter/index.js

@@ -7,6 +7,8 @@ class HyperP2PStatsExporter extends EventEmitter {
   constructor (opts = {}) {
     super()
     this._sources = new Map()
+    this._history = []
+    this._maxHistory = opts.maxHistory ?? 32
     this._stats = { exports: 0, sources: 0 }
   }
 
@@ -15,16 +17,20 @@ class HyperP2PStatsExporter extends EventEmitter {
     if (typeof getStatsFn !== 'function') throw new Error('getStatsFn required')
     this._sources.set(name, getStatsFn)
     this._stats.sources++
+    this.emit('register', { name })
     return true
   }
 
   unregister (name) {
-    return this._sources.delete(name)
+    const ok = this._sources.delete(name)
+    if (ok) this.emit('unregister', { name })
+    return ok
   }
 
-  snapshot () {
+  snapshot (filter = null) {
     const out = { at: Date.now(), modules: {} }
     for (const [name, fn] of this._sources) {
+      if (filter && !filter(name)) continue
       try {
         out.modules[name] = fn()
       } catch (err) {
@@ -32,24 +38,57 @@ class HyperP2PStatsExporter extends EventEmitter {
       }
     }
     this._stats.exports++
+    this._history.push(out)
+    if (this._history.length > this._maxHistory) this._history.shift()
     this.emit('snapshot', out)
     return out
   }
 
-  toJSON () {
+  compare (prevAt) {
-    return JSON.stringify(this.snapshot(), null, 2)
+    const prev = this._history.find((h) => h.at === prevAt)
+    const cur = this.snapshot()
+    if (!prev) return { cur, delta: null }
+    const delta = { modules: {} }
+    for (const [name, stats] of Object.entries(cur.modules)) {
+      delta.modules[name] = { before: prev.modules[name], after: stats }
+    }
+    return { cur, delta }
   }
 
-  listSources () { return [...this._sources.keys()] }
+  history (limit = 10) {
+    return this._history.slice(-Math.max(0, limit | 0))
+  }
 
-  hasSource (name) { return this._sources.has(name) }
+  toJSON (filter = null) {
+    return JSON.stringify(this.snapshot(filter), null, 2)
+  }
+
+  listSources () {
+    return [...this._sources.keys()]
+  }
+
+  hasSource (name) {
+    return this._sources.has(name)
+  }
 
   getStats () {
-    return { ...this._stats, registered: this._sources.size, protocol: PROTOCOL }
+    return {
+      ...this._stats,
+      registered: this._sources.size,
+      history: this._history.length,
+      protocol: PROTOCOL
+    }
   }
 
-  async ready () { return this }
+  async ready () {
-  async close () { this._sources.clear() }
+    return this
+  }
+
+  async close () {
+    this._sources.clear()
+    this._history = []
+    this.emit('closed')
+  }
 }
 
 module.exports = { HyperP2PStatsExporter, PROTOCOL }