Updates

hyper-p2p-distributed-event-bus/index.js

@@ -50,6 +50,7 @@ class HyperP2PDistributedEventBus extends EventEmitter {
     this.expiryMs = opts.expiry || DEFAULT_EXPIRY
     this.metadata = opts.metadata || { agent: 'hyper-p2p-distributed-event-bus' }
     this.enableSigning = opts.enableSigning !== false
+    this._enableBackgroundTimers = opts.enableBackgroundTimers === true
     this._metrics = { published: 0, received: 0, signed: 0, verified: 0, pruned: 0, errors: 0 }
 
     // Internal structures
@@ -81,8 +82,10 @@ class HyperP2PDistributedEventBus extends EventEmitter {
     if (this._joined) return this
     await this._initStorage()
     await this._initSwarm()
-    this._startAnnounceTimer()
+    if (this._enableBackgroundTimers) {
-    this._startCleanupTimer()
+      this._startAnnounceTimer()
+      this._startCleanupTimer()
+    }
     this._joined = true
     this.emit('ready')
     return this
@@ -144,12 +147,12 @@ class HyperP2PDistributedEventBus extends EventEmitter {
     const peerPub = peerInfo.publicKey || connection.remotePublicKey
     const peerHex = peerPub ? b4a.toString(peerPub, 'hex') : b4a.toString(crypto.randomBytes(32), 'hex')
     const self = this
+    this.emit('peer-connected', { peer: peerHex })
 
-    const { channel, msg } = protocolChannel(mux, {
+    protocolChannel(mux, {
       protocol: EVENT_BUS_PROTOCOL,
-      onopen () {
+      onopen (channel, msg) {
         self.peers.set(peerHex, { lastSeen: Date.now(), metadata: {}, channel, msg })
-        self.emit('peer-connected', { peer: peerHex })
       },
       onclose () {
         self.peers.delete(peerHex)

hyper-p2p-distributed-lock/index.js

@@ -46,6 +46,8 @@ class HyperP2PDistributedLock extends EventEmitter {
     this._protocol = null
     this._peerMsgs = new Map()
     this._cleanupTimer = null
+    this._leaseTimers = new Set()
+    this._enableBackgroundTimers = opts.enableBackgroundTimers === true
     this._joined = false
     this.metrics = {
       acquiresAttempted: 0,
@@ -93,7 +95,7 @@ class HyperP2PDistributedLock extends EventEmitter {
     if (this._joined) return
     await this._initStorage()
     await this._initP2P()
-    this._startCleanup()
+    if (this._enableBackgroundTimers) this._startCleanup()
     this._joined = true
     this.emit('ready')
   }
@@ -108,6 +110,7 @@ class HyperP2PDistributedLock extends EventEmitter {
 
   async _initP2P () {
     this._peerMsgs = new Map()
+    this.peers = this._peerMsgs
     if (!this.topic) {
       this._protocol = null
       return
@@ -116,8 +119,15 @@ class HyperP2PDistributedLock extends EventEmitter {
     this.swarm = swarm
     const self = this
     wireConnection(swarm, (socket, peerInfo, mux) => {
-      const { msg } = protocolChannel(mux, {
+      const peerHex = peerInfo.publicKey ? b4a.toString(peerInfo.publicKey, 'hex') : null
+      protocolChannel(mux, {
         protocol: LOCK_PROTOCOL,
+        onopen (channel, msg) {
+          if (peerHex) {
+            self._peerMsgs.set(peerHex, { msg })
+            self.emit('peer-connected', { peer: peerHex })
+          }
+        },
         onmessage (data) {
           if (!data || !data.type) return
           if (data.type === 'claim' && data.resourceId && data.claim) {
@@ -128,10 +138,11 @@ class HyperP2PDistributedLock extends EventEmitter {
               self._forceRelease(data.resourceId, lock.lockId, 'remote-release')
             }
           }
+        },
+        onclose () {
+          if (peerHex) self._peerMsgs.delete(peerHex)
         }
       })
-      const peerHex = peerInfo.publicKey ? b4a.toString(peerInfo.publicKey, 'hex') : null
-      if (peerHex) self._peerMsgs.set(peerHex, msg)
       socket.on('close', () => {
         if (peerHex) self._peerMsgs.delete(peerHex)
       })
@@ -208,12 +219,13 @@ class HyperP2PDistributedLock extends EventEmitter {
     this.metrics.acquiresSucceeded++
     this.emit('acquired', { resourceId, lockId, fencingToken, expiresAt, owner: this.publicKeyHex })
 
-    // Auto-release on timeout for owner (safety)
+    const leaseTimer = setTimeout(() => {
-    setTimeout(() => {
+      this._leaseTimers.delete(leaseTimer)
       if (this.locks.get(resourceId)?.lockId === lockId && Date.now() >= expiresAt) {
         this._forceRelease(resourceId, lockId, 'lease-expired')
       }
     }, leaseMs + 1000)
+    this._leaseTimers.add(leaseTimer)
 
     return { lockId, fencingToken, acquiredAt: lockInfo.acquiredAt, expiresAt, owner: this.publicKeyHex, resourceId }
   }
@@ -379,6 +391,8 @@ class HyperP2PDistributedLock extends EventEmitter {
   }
 
   async close () {
+    for (const t of this._leaseTimers) clearTimeout(t)
+    this._leaseTimers.clear()
     if (this._cleanupTimer) {
       clearInterval(this._cleanupTimer)
       this._cleanupTimer = null

hyper-p2p-intent-router/index.js

@@ -35,6 +35,7 @@ class HyperP2PIntentRouter extends EventEmitter {
     this.maxIntentsPerPeer = opts.maxIntentsPerPeer || 64
     this.matchThreshold = opts.matchThreshold || 0.3 // Jaccard-like similarity
     this.options = opts
+    this._enableBackgroundTimers = opts.enableBackgroundTimers === true
 
     this.localIntents = new Map() // intentId -> intent
     this.peerIntents = new Map() // peerPubHex -> { intents: Map, lastSeen: ts, connections: Set }
@@ -46,6 +47,7 @@ class HyperP2PIntentRouter extends EventEmitter {
     this._joined = false
     this._protocol = null
     this._connections = new Map() // peerPubHex -> { msg, channel }
+    this.peers = this._connections
     this._topics = new Map() // topicHex -> swarm topic handle
   }
 
@@ -54,8 +56,10 @@ class HyperP2PIntentRouter extends EventEmitter {
     await this._initStorage()
     await this._initSwarm()
     await this._loadPersistedIntents()
-    this._startAnnounceTimer()
+    if (this._enableBackgroundTimers) {
-    this._startCleanupTimer()
+      this._startAnnounceTimer()
+      this._startCleanupTimer()
+    }
     this._joined = true
     this.emit('ready')
   }
@@ -302,7 +306,8 @@ class HyperP2PIntentRouter extends EventEmitter {
 
     const { channel, msg } = protocolChannel(mux, {
       protocol: INTENT_PROTOCOL,
-      onopen () {
+      onopen (channel, msg) {
+        self._connections.set(peerHex, { msg, channel })
         self.emit('peer:connected', { peerPublicKey: peerHex })
         self._exchangeIntents(peerHex, msg)
       },
@@ -319,8 +324,6 @@ class HyperP2PIntentRouter extends EventEmitter {
       }
     })
 
-    this._connections.set(peerHex, { msg, channel })
-
     if (!this.peerIntents.has(peerHex)) {
       this.peerIntents.set(peerHex, { intents: new Map(), lastSeen: Date.now(), connections: new Set() })
     }

hyper-p2p-presence/index.js

@@ -27,6 +27,7 @@ class HyperP2PPresence extends EventEmitter {
     this.announceIntervalMs = opts.announceInterval || DEFAULT_ANNOUNCE_INTERVAL
     this.expiryMs = opts.expiry || DEFAULT_EXPIRY
     this.metadata = opts.metadata || {}
+    this._enableBackgroundTimers = opts.enableBackgroundTimers === true
     this.peers = new Map() // publicKeyHex -> presence info
     this.swarm = null
     this.corestore = null
@@ -56,7 +57,10 @@ class HyperP2PPresence extends EventEmitter {
     await this._initStorage()
     await this._initSwarm()
     this._ensureSelfRegistered()
-    this._startCleanupTimer()
+    if (this._enableBackgroundTimers) {
+      this._startAnnounceTimer()
+      this._startCleanupTimer()
+    }
     this._joined = true
     this.emit('ready')
   }
@@ -109,8 +113,6 @@ class HyperP2PPresence extends EventEmitter {
     wireConnection(this.swarm, (socket, info, mux) => {
       this._handleConnection(socket, info, mux)
     })
-
-    this._startAnnounceTimer()
   }
 
   _handleConnection (socket, info, mux) {

hyper-p2p-reactive-state/index.js

@@ -45,6 +45,8 @@ class HyperP2PReactiveState extends EventEmitter {
     this.syncIntervalMs = opts.syncInterval || DEFAULT_SYNC_INTERVAL
     this.expiryMs = opts.expiry || DEFAULT_EXPIRY
     this.metadata = opts.metadata || { agent: 'hyper-p2p-reactive-state' }
+    this._enableBackgroundTimers = opts.enableBackgroundTimers === true
+    this._memoryOnly = opts.memoryOnly === true
 
     // Internal state: key -> { value, timestamp, peerId, signature? }
     this.state = new Map()
@@ -73,14 +75,17 @@ class HyperP2PReactiveState extends EventEmitter {
     if (this._joined) return this
     await this._initStorage()
     await this._initSwarm()
-    this._startSyncTimer()
+    if (this._enableBackgroundTimers) {
-    this._startCleanupTimer()
+      this._startSyncTimer()
+      this._startCleanupTimer()
+    }
     this._joined = true
     this.emit('ready')
     return this
   }
 
   async _initStorage () {
+    if (this._memoryOnly) return
     try {
       await fs.mkdir(this.storageDir, { recursive: true })
     } catch (err) {
@@ -112,9 +117,7 @@ class HyperP2PReactiveState extends EventEmitter {
   }
 
   async _initSwarm () {
-    if (!this.topic) {
+    if (!this.topic) return
-      throw new Error('topic is required for P2P discovery')
-    }
 
     const { swarm } = await createSwarm({ keyPair: this.keyPair, topic: this.topic })
     this.swarm = swarm
@@ -215,6 +218,7 @@ class HyperP2PReactiveState extends EventEmitter {
   }
 
   async _persistState () {
+    if (!this.bee) return
     for (const [key, data] of this.state) {
       try {
         await this.bee.put(LWW_PREFIX + key, data)
@@ -278,8 +282,7 @@ class HyperP2PReactiveState extends EventEmitter {
       this.emit('change', { key, value, peer: this.publicKeyHex, type: 'set' })
       this.emit('set', { key, value, peer: this.publicKeyHex })
 
-      // Persist
+      if (this.bee) await this.bee.put(LWW_PREFIX + key, data)
-      await this.bee.put(LWW_PREFIX + key, data)
 
       // Broadcast update to peers (simplified - in prod would use open channels)
       this._broadcastUpdate(key, data)
@@ -321,7 +324,7 @@ class HyperP2PReactiveState extends EventEmitter {
     this.emit('change', { key, value: null, peer: this.publicKeyHex, type: 'delete' })
     this.emit('delete', { key, peer: this.publicKeyHex })
 
-    await this.bee.put(LWW_PREFIX + key, data)
+    if (this.bee) await this.bee.put(LWW_PREFIX + key, data)
     this._broadcastUpdate(key, data)
     return true
   }
@@ -388,7 +391,9 @@ class HyperP2PReactiveState extends EventEmitter {
       await this.swarm.destroy().catch(() => {})
     }
     if (this.bee) {
+      const core = this.bee.core
       await this.bee.close().catch(() => {})
+      if (core && !core.closed) await core.close().catch(() => {})
     }
     this._joined = false
     this.emit('closed')

hyper-p2p-reputation-system/test-reputation-attest-1779313150685/state.json

@@ -0,0 +1,23 @@
+{
+  "scores": {
+    "9a7447a62939fed8972a21584358ae841176d9bea5a778ee595eb0dcc974faef": {
+      "score": 75,
+      "lastUpdated": 1779313150686,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "9a7447a62939fed8972a21584358ae841176d9bea5a778ee595eb0dcc974faef": [
+      {
+        "ts": 1779313150686,
+        "delta": 75,
+        "attester": "b553633830fad5f0ae72dfb4777ad1c30a1f631be1b7f9c8b69a57d82d99f1d2",
+        "signature": "/S3yN5f7BiMDQ4Ia/3tyS84NisrHn1YQqBkQ6bz0wqqselX6szZ4u+xWr1Snwp4yZ3NHKhIU4/CNLARSUk9eBw==",
+        "metadata": {
+          "context": "test"
+        }
+      }
+    ]
+  },
+  "lastPersist": 1779313150746
+}

hyper-p2p-reputation-system/test-reputation-attest-1779314988623/state.json

@@ -0,0 +1,23 @@
+{
+  "scores": {
+    "7aad1180ce02770a762e8ade23298a72b7fb69a410aaf96967cf4cc8bbe263f7": {
+      "score": 75,
+      "lastUpdated": 1779314988625,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "7aad1180ce02770a762e8ade23298a72b7fb69a410aaf96967cf4cc8bbe263f7": [
+      {
+        "ts": 1779314988625,
+        "delta": 75,
+        "attester": "54c78e088fa0763e649487d92cc567e4438dea71b62e4a798899ea4ee3581f08",
+        "signature": "fKEF7p7sK8biri3SKW2HLLWHKbmtPbFTmfj+3kol9mPwYcs1Nw9KS9gx+HJCuU93rYLmvVqtsZAoR154py0SCg==",
+        "metadata": {
+          "context": "test"
+        }
+      }
+    ]
+  },
+  "lastPersist": 1779314988683
+}

hyper-p2p-reputation-system/test-reputation-decay-1779313150749/state.json

@@ -0,0 +1,21 @@
+{
+  "scores": {
+    "decay-peer": {
+      "score": 90,
+      "lastUpdated": 1779313150749,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "decay-peer": [
+      {
+        "ts": 1779313150749,
+        "delta": 100,
+        "attester": "7acd28b48a42bdd04a0f3998666195a59f6985159678fea2664a991325924629",
+        "signature": "NwMPLq/7XFlrlTWvvm7WgKHTGIR0e2qwfbDGCgE+wLSYzTiul4KFZrh0G9H8rNzW93Qofq91qtZLSlkRi66MDQ==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779313150750
+}

hyper-p2p-reputation-system/test-reputation-decay-1779314988686/state.json

@@ -0,0 +1,21 @@
+{
+  "scores": {
+    "decay-peer": {
+      "score": 90,
+      "lastUpdated": 1779314988686,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "decay-peer": [
+      {
+        "ts": 1779314988686,
+        "delta": 100,
+        "attester": "a25f07e97664f33d97a123e00bba1b6f445b581dc762f10c799dfd9037505a2e",
+        "signature": "JY0WvtXqMvEugNw3Ju5t4VoNMDMNTe/wcwNd/SHXSWZOs09wkrw91uGQlN7M15+PLk2fQfMve+r3XfOC8ijAAg==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779314988687
+}

hyper-p2p-reputation-system/test-reputation-lifecycle-1779313150674/state.json

@@ -0,0 +1,5 @@
+{
+  "scores": {},
+  "history": {},
+  "lastPersist": 1779313150684
+}

hyper-p2p-reputation-system/test-reputation-lifecycle-1779314988613/state.json

@@ -0,0 +1,5 @@
+{
+  "scores": {},
+  "history": {},
+  "lastPersist": 1779314988623
+}

hyper-p2p-reputation-system/test-reputation-persist-1779313150750/state.json

@@ -0,0 +1,21 @@
+{
+  "scores": {
+    "persist-peer": {
+      "score": 55,
+      "lastUpdated": 1779313150750,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "persist-peer": [
+      {
+        "ts": 1779313150750,
+        "delta": 55,
+        "attester": "0fdb685f0288995edeab58e47aa89293b53ed542256729cc28f8a4d36a0c09f6",
+        "signature": "jN/+Z+WZu8D/HNCUfap6i1kXmunAx/ja9y5ZwNKY3Tt30+p8Fu2dD1Cw3ypLPwMqtsTSB0TrZkJCjYxoQiH2Cg==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779313150751
+}

hyper-p2p-reputation-system/test-reputation-persist-1779314988687/state.json

@@ -0,0 +1,21 @@
+{
+  "scores": {
+    "persist-peer": {
+      "score": 55,
+      "lastUpdated": 1779314988687,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "persist-peer": [
+      {
+        "ts": 1779314988687,
+        "delta": 55,
+        "attester": "dbd358d15af6021800b1e2d0be35d467946677f1191035775688a47c451232af",
+        "signature": "ZT+iAPrcUpo5vKLVJavv8QhZevzw1oWPpue5Q0dA2TejxqZmzO3ZDYr349gEskosg9nx9QphR3xQCzGmEeGFDQ==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779314988688
+}

hyper-p2p-reputation-system/test-reputation-queries-1779313150747/state.json

@@ -0,0 +1,35 @@
+{
+  "scores": {
+    "peer1": {
+      "score": 100,
+      "lastUpdated": 1779313150747,
+      "attestationsCount": 1
+    },
+    "peer2": {
+      "score": 60,
+      "lastUpdated": 1779313150748,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "peer1": [
+      {
+        "ts": 1779313150747,
+        "delta": 100,
+        "attester": "89e865aac9e474c3ddeed73a45c1e4ef73f5d8d5c3e0194056b0ad86893b78d8",
+        "signature": "btVkpLdN5N44isEU2u5i8aF9Ge8YZ42GHqofOwZxaj+YePsg4C0BHP2ojg/NmoddjKWZuaPq5ifby5zNi8vVCQ==",
+        "metadata": {}
+      }
+    ],
+    "peer2": [
+      {
+        "ts": 1779313150748,
+        "delta": 60,
+        "attester": "89e865aac9e474c3ddeed73a45c1e4ef73f5d8d5c3e0194056b0ad86893b78d8",
+        "signature": "WkfnZOAW7j3LbxA6voXlGFednXObrfgBjHEiS78+iLlrdvPFJE9qI/uA5MM82BsIi4/wil0fkiG1vsSAYRgdCg==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779313150748
+}

hyper-p2p-reputation-system/test-reputation-queries-1779314988684/state.json

@@ -0,0 +1,35 @@
+{
+  "scores": {
+    "peer1": {
+      "score": 100,
+      "lastUpdated": 1779314988685,
+      "attestationsCount": 1
+    },
+    "peer2": {
+      "score": 60,
+      "lastUpdated": 1779314988685,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "peer1": [
+      {
+        "ts": 1779314988685,
+        "delta": 100,
+        "attester": "ccc9bea7120a8d19a79a68dfe0db251882f52eae3b83dbfd5cfbd5a7439da21a",
+        "signature": "K5E220bP18H4c7oQem+h2voSl3nEOvI0sV2dHSwlgtnuig0UOqh9joxbGUaof+m/Mwulf8GrWwZ46BUjS10JAw==",
+        "metadata": {}
+      }
+    ],
+    "peer2": [
+      {
+        "ts": 1779314988685,
+        "delta": 60,
+        "attester": "ccc9bea7120a8d19a79a68dfe0db251882f52eae3b83dbfd5cfbd5a7439da21a",
+        "signature": "RE+TyQw1wiqS93togQpXRuXrpAXV4kiCuErXjfTJEjNFo3qeEE6s5HsX0Mb5yBLca8rL5SWifwh1a3PHzB4bBg==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779314988686
+}

hyper-p2p-reputation-system/test-reputation-receive-1779313150746/state.json

@@ -0,0 +1,5 @@
+{
+  "scores": {},
+  "history": {},
+  "lastPersist": 1779313150747
+}

hyper-p2p-reputation-system/test-reputation-receive-1779314988683/state.json

@@ -0,0 +1,5 @@
+{
+  "scores": {},
+  "history": {},
+  "lastPersist": 1779314988684
+}

hyper-p2p-reputation-system/test-reputation-snapshot-1779313150751/state.json

@@ -0,0 +1,37 @@
+{
+  "scores": {
+    "snapshot-peer-1": {
+      "score": 120,
+      "lastUpdated": 1779313150752,
+      "attestationsCount": 1
+    },
+    "snapshot-peer-2": {
+      "score": 80,
+      "lastUpdated": 1779313150752,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "snapshot-peer-1": [
+      {
+        "ts": 1779313150752,
+        "delta": 120,
+        "attester": "8550888040dc7099af6bd90144a43c6e0ba78c95551696152b18f34c46d53904",
+        "signature": "LK7bflK96+Jh+M/ZzPMLjTJO5J39kVtsaD4WplQOuXiBbcIz+RdBmDAZ1eez9XFw4Vt5RBtbmiEe1I72c9XfDw==",
+        "metadata": {
+          "category": "reliability"
+        }
+      }
+    ],
+    "snapshot-peer-2": [
+      {
+        "ts": 1779313150752,
+        "delta": 80,
+        "attester": "8550888040dc7099af6bd90144a43c6e0ba78c95551696152b18f34c46d53904",
+        "signature": "5wYmYD8JQ52mgH+zJTECClxS6qKKqiPQ2VrPQasw9wYjVgHnyjUIouANAP067ZQsW0dbXRSXnCiLd7GTUF2IAw==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779313150753
+}

hyper-p2p-reputation-system/test-reputation-snapshot-1779314988688/state.json

@@ -0,0 +1,37 @@
+{
+  "scores": {
+    "snapshot-peer-1": {
+      "score": 120,
+      "lastUpdated": 1779314988689,
+      "attestationsCount": 1
+    },
+    "snapshot-peer-2": {
+      "score": 80,
+      "lastUpdated": 1779314988689,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "snapshot-peer-1": [
+      {
+        "ts": 1779314988689,
+        "delta": 120,
+        "attester": "0294f3f95780a29da1bba2808769869655894c041f1c8f206f5816cfce561354",
+        "signature": "AYp8BMiEZwGuM9xLFW8aFtwaLukXEgYTAV+uG8VEi/ZX2QsYnqY4u4NcV4zGm5Bbdt9AGg8MBcbC0gWiAPcYAQ==",
+        "metadata": {
+          "category": "reliability"
+        }
+      }
+    ],
+    "snapshot-peer-2": [
+      {
+        "ts": 1779314988689,
+        "delta": 80,
+        "attester": "0294f3f95780a29da1bba2808769869655894c041f1c8f206f5816cfce561354",
+        "signature": "pufS+vOSUwtby+rH2R1tdvs/+SWO8xa0Jx/+1DzS9HNkIUkcK6kZOeU1CUOT4ndvSkrVI53bGEaMWtxJ+yNqDA==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779314988690
+}

hyper-p2p-reputation-system/test-reputation-snapshot2-1779313150752/state.json

@@ -0,0 +1,37 @@
+{
+  "scores": {
+    "snapshot-peer-1": {
+      "score": 120,
+      "lastUpdated": 1779313150752,
+      "attestationsCount": 1
+    },
+    "snapshot-peer-2": {
+      "score": 80,
+      "lastUpdated": 1779313150752,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "snapshot-peer-1": [
+      {
+        "ts": 1779313150752,
+        "delta": 120,
+        "attester": "8550888040dc7099af6bd90144a43c6e0ba78c95551696152b18f34c46d53904",
+        "signature": "LK7bflK96+Jh+M/ZzPMLjTJO5J39kVtsaD4WplQOuXiBbcIz+RdBmDAZ1eez9XFw4Vt5RBtbmiEe1I72c9XfDw==",
+        "metadata": {
+          "category": "reliability"
+        }
+      }
+    ],
+    "snapshot-peer-2": [
+      {
+        "ts": 1779313150752,
+        "delta": 80,
+        "attester": "8550888040dc7099af6bd90144a43c6e0ba78c95551696152b18f34c46d53904",
+        "signature": "5wYmYD8JQ52mgH+zJTECClxS6qKKqiPQ2VrPQasw9wYjVgHnyjUIouANAP067ZQsW0dbXRSXnCiLd7GTUF2IAw==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779313150753
+}

hyper-p2p-reputation-system/test-reputation-snapshot2-1779314988689/state.json

@@ -0,0 +1,37 @@
+{
+  "scores": {
+    "snapshot-peer-1": {
+      "score": 120,
+      "lastUpdated": 1779314988689,
+      "attestationsCount": 1
+    },
+    "snapshot-peer-2": {
+      "score": 80,
+      "lastUpdated": 1779314988689,
+      "attestationsCount": 1
+    }
+  },
+  "history": {
+    "snapshot-peer-1": [
+      {
+        "ts": 1779314988689,
+        "delta": 120,
+        "attester": "0294f3f95780a29da1bba2808769869655894c041f1c8f206f5816cfce561354",
+        "signature": "AYp8BMiEZwGuM9xLFW8aFtwaLukXEgYTAV+uG8VEi/ZX2QsYnqY4u4NcV4zGm5Bbdt9AGg8MBcbC0gWiAPcYAQ==",
+        "metadata": {
+          "category": "reliability"
+        }
+      }
+    ],
+    "snapshot-peer-2": [
+      {
+        "ts": 1779314988689,
+        "delta": 80,
+        "attester": "0294f3f95780a29da1bba2808769869655894c041f1c8f206f5816cfce561354",
+        "signature": "pufS+vOSUwtby+rH2R1tdvs/+SWO8xa0Jx/+1DzS9HNkIUkcK6kZOeU1CUOT4ndvSkrVI53bGEaMWtxJ+yNqDA==",
+        "metadata": {}
+      }
+    ]
+  },
+  "lastPersist": 1779314988690
+}

hyper-p2p-rpc/docs/API.md

@@ -108,4 +108,12 @@ npm install
 npx brittle-bare test/test.js
 ```
 
-Integration (2-node): [`../../real_tests/integration/`](../../real_tests/integration/) — see [DEVELOPMENT.md](../../DEVELOPMENT.md).
+### SecretStream pairing (in-process)
+
+```js
+const { pairSecretStreams, waitSecretStreamsConnected } = require('../lib/pair.js')
+pairSecretStreams(initiatorStream, responderStream)
+await waitSecretStreamsConnected(initiatorStream, responderStream)
+```
+
+Integration: `npx brittle-bare test/integration-two-node.js` (from this package). See [DEVELOPMENT.md](../../../DEVELOPMENT.md).

hyper-p2p-rpc/examples/basic.js

@@ -1,17 +1,13 @@
 require('bare-process/global')
 const { RPCServer, RPCClient } = require('../index.js')
 const SecretStream = require('@hyperswarm/secret-stream')
-const hypercoreCrypto = require('hypercore-crypto')
+const { pairSecretStreams, waitSecretStreamsConnected } = require('../lib/pair.js')
 
 async function demo () {
   console.log('=== hyper-p2p-rpc Basic Demo (SecretStream) ===')
 
-  const kp = hypercoreCrypto.keyPair()
+  const a = new SecretStream(true)
-  const a = new SecretStream(true, null, { keyPair: kp })
+  const b = new SecretStream(false)
-  const b = new SecretStream(false, null, { keyPair: hypercoreCrypto.keyPair() })
-
-  a.isInitiator = true
-  b.isInitiator = false
 
   const server = new RPCServer()
   server.register('add', async (params) => ({ result: (params.a || 0) + (params.b || 0) }))
@@ -20,12 +16,8 @@ async function demo () {
   server.handleConnection(a)
   const client = new RPCClient(b)
 
-  await new Promise((resolve, reject) => {
+  pairSecretStreams(a, b)
-    a.on('connect', resolve)
+  await waitSecretStreamsConnected(a, b)
-    b.on('connect', resolve)
-    a.connect(b)
-    setTimeout(() => reject(new Error('connect timeout')), 5000)
-  })
 
   const sum = await client.call('add', { a: 42, b: 58 })
   console.log('42 + 58 =', sum.result)

hyper-p2p-rpc/examples/streaming.js

@@ -1,15 +1,13 @@
 require('bare-process/global')
 const { RPCServer, RPCClient } = require('../index.js')
 const SecretStream = require('@hyperswarm/secret-stream')
-const hypercoreCrypto = require('hypercore-crypto')
+const { pairSecretStreams, waitSecretStreamsConnected } = require('../lib/pair.js')
-const { setInterval, clearInterval } = require('bare-timers')
 
 async function streamingDemo () {
   console.log('=== hyper-p2p-rpc Streaming Demo (SecretStream) ===')
 
-  const kp = hypercoreCrypto.keyPair()
+  const a = new SecretStream(true)
-  const a = new SecretStream(true, null, { keyPair: kp })
+  const b = new SecretStream(false)
-  const b = new SecretStream(false, null, { keyPair: hypercoreCrypto.keyPair() })
 
   const server = new RPCServer()
   server.register('count-stream', async function * (params) {
@@ -23,13 +21,8 @@ async function streamingDemo () {
   server.handleConnection(a)
   const client = new RPCClient(b)
 
-  await new Promise((resolve, reject) => {
+  pairSecretStreams(a, b)
-    const timer = setTimeout(() => reject(new Error('connect timeout')), 5000)
+  await waitSecretStreamsConnected(a, b)
-    const done = () => { clearTimeout(timer); resolve() }
-    a.on('connect', done)
-    b.on('connect', done)
-    a.connect(b)
-  })
 
   const stream = await client.callStream('count-stream', { max: 4 })
   for await (const chunk of stream) {

hyper-p2p-rpc/index.js

@@ -72,7 +72,7 @@ class RPCServer extends EventEmitter {
 
       if (result && typeof result[Symbol.asyncIterator] === 'function') {
         rpcMsg.send({ id: msg.id, stream: true })
-        await this._handleStreamingResponse(socket, msg.id, result)
+        await this._streamOnRpcChannel(rpcMsg, msg.id, result)
       } else {
         rpcMsg.send({ id: msg.id, result })
       }
@@ -81,22 +81,14 @@ class RPCServer extends EventEmitter {
     }
   }
 
-  async _handleStreamingResponse (socket, callId, sourceStream) {
+  async _streamOnRpcChannel (rpcMsg, callId, sourceStream) {
-    const mux = Protomux.from(socket)
-    const streamChannel = mux.createChannel({
-      protocol: STREAM_PROTOCOL + '/' + callId
-    })
-
-    const dataMsg = streamChannel.addMessage({ encoding: c.json })
-    streamChannel.open()
-
     try {
       for await (const chunk of sourceStream) {
-        dataMsg.send({ id: callId, chunk })
+        rpcMsg.send({ id: callId, chunk })
       }
-      dataMsg.send({ id: callId, done: true })
+      rpcMsg.send({ id: callId, done: true })
     } catch (err) {
-      dataMsg.send({ id: callId, error: err.message })
+      rpcMsg.send({ id: callId, error: err.message })
     }
   }
 
@@ -153,7 +145,18 @@ class RPCClient extends EventEmitter {
       onmessage (reply) {
         const pending = self.pending.get(reply.id)
         if (!pending) return
-        if (reply.stream) return
+        if (reply.stream) {
+          if (pending.streamReady) {
+            clearTimeout(pending.timer)
+            pending.streamReady()
+          }
+          return
+        }
+        if (reply.chunk !== undefined || reply.done || reply.error) {
+          if (pending.onStreamMsg) pending.onStreamMsg(reply)
+          if (reply.done || reply.error) self.pending.delete(reply.id)
+          return
+        }
 
         clearTimeout(pending.timer)
         self.pending.delete(reply.id)
@@ -188,48 +191,48 @@ class RPCClient extends EventEmitter {
 
   async callStream (method, params = {}, timeoutMs = this.defaultTimeout) {
     const id = generateId()
+    const queue = []
+    const waiters = []
 
-    return new Promise((resolve, reject) => {
+    const push = (msg) => {
+      if (waiters.length) waiters.shift()(msg)
+      else queue.push(msg)
+    }
+
+    await new Promise((resolve, reject) => {
       const timer = setTimeout(() => {
         this.pending.delete(id)
         reject(new Error('STREAM_TIMEOUT'))
       }, timeoutMs)
 
-      const mux = this.mux
+      this.pending.set(id, {
-      const streamChannel = mux.createChannel({
+        reject,
-        protocol: STREAM_PROTOCOL + '/' + id
+        timer,
+        streamReady: resolve,
+        onStreamMsg: push
       })
+      this.rpcMsg.send({ id, method, params })
+    })
 
-      const dataMsg = streamChannel.addMessage({ encoding: c.json })
+    const next = () => new Promise((resolve, reject) => {
-      streamChannel.open()
+      if (queue.length) return resolve(queue.shift())
+      const t = setTimeout(() => reject(new Error('stream read timeout')), timeoutMs)
+      waiters.push((msg) => {
+        clearTimeout(t)
+        resolve(msg)
+      })
+    })
 
-      let ended = false
+    return {
-
+      [Symbol.asyncIterator]: async function * () {
-      const readable = {
+        while (true) {
-        [Symbol.asyncIterator]: async function * () {
+          const data = await next()
-          while (!ended) {
+          if (data.error) throw new Error(data.error)
-            const data = await new Promise((res, rej) => {
+          if (data.done) return
-              const t = setTimeout(() => rej(new Error('stream read timeout')), 5000)
+          if (data.chunk !== undefined) yield data.chunk
-              const handler = (msg) => {
-                clearTimeout(t)
-                res(msg)
-              }
-              dataMsg.onmessage = handler
-            })
-
-            if (data.done) {
-              ended = true
-              return
-            }
-            if (data.error) throw new Error(data.error)
-            if (data.chunk !== undefined) yield data.chunk
-          }
         }
       }
-
+    }
-      this.rpcMsg.send({ id, method, params, stream: true })
-      resolve(readable)
-    })
   }
 
   close () {

hyper-p2p-rpc/lib/pair.js

@@ -0,0 +1,10 @@
+/** Pair two SecretStream instances in-process (Protomux-style loopback). */
+function pairSecretStreams (initiator, responder) {
+  initiator.rawStream.pipe(responder.rawStream).pipe(initiator.rawStream)
+}
+
+async function waitSecretStreamsConnected (...streams) {
+  await Promise.all(streams.map((s) => s.opened))
+}
+
+module.exports = { pairSecretStreams, waitSecretStreamsConnected }

hyper-p2p-rpc/test/integration-two-node.js

@@ -0,0 +1,25 @@
+require('bare-process/global')
+const test = require('brittle')
+const SecretStream = require('@hyperswarm/secret-stream')
+const { RPCServer, RPCClient } = require('../index.js')
+const { pairSecretStreams, waitSecretStreamsConnected } = require('../lib/pair.js')
+
+test('rpc: call over SecretStream pair', async function (t) {
+  const a = new SecretStream(true)
+  const b = new SecretStream(false)
+
+  const server = new RPCServer()
+  server.register('echo', async (params) => ({ echo: params.msg }))
+  server.handleConnection(a)
+  const client = new RPCClient(b)
+
+  pairSecretStreams(a, b)
+  await waitSecretStreamsConnected(a, b)
+
+  const res = await client.call('echo', { msg: 'p2p' })
+  t.is(res.echo, 'p2p')
+
+  server.close()
+  a.destroy()
+  b.destroy()
+})

hyper-p2p-rpc/test/streaming-test.js

@@ -0,0 +1,29 @@
+require('bare-process/global')
+const test = require('brittle')
+const SecretStream = require('@hyperswarm/secret-stream')
+const { pairSecretStreams, waitSecretStreamsConnected } = require('../lib/pair.js')
+const { RPCServer, RPCClient } = require('../index.js')
+
+test('rpc: streaming call', async function (t) {
+  const a = new SecretStream(true)
+  const b = new SecretStream(false)
+  pairSecretStreams(a, b)
+  await waitSecretStreamsConnected(a, b)
+
+  const server = new RPCServer()
+  server.register('nums', async function * () {
+    yield { n: 1 }
+    yield { n: 2 }
+  })
+  server.handleConnection(a)
+  const client = new RPCClient(b)
+
+  const stream = await client.callStream('nums', {})
+  const out = []
+  for await (const chunk of stream) out.push(chunk.n)
+  t.alike(out, [1, 2])
+
+  server.close()
+  a.destroy()
+  b.destroy()
+})