pearcord/pearcord-contacts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: use awaitSwarmJoin and shorten friend-request DHT flush

Browse Source 
Stop calling .catch on Hyperswarm join (not a Promise). Cap contacts
mesh flush so send-friend-request does not block multi-second on DHT.
This commit is contained in:
Raven Scott
2026-07-13 15:56:00 -04:00
parent 3fe8075299
commit 48e14fb0b5
1 changed files with 64 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
index.js
+64 -35
View File
@@ -5,7 +5,18 @@ const path = require('bare-path')
const EventEmitter = require('bare-events')
const b4a = require('b4a')
const { JsonStore } = require('pearcord-db/store-json')
const { RPC, contactsTopic, globalPresenceTopic, dmVoiceTopic, topicToBuffer, id, now, USER_STATUS } = require('pearcord-shared')
const {
RPC,
contactsTopic,
globalPresenceTopic,
dmVoiceTopic,
topicToBuffer,
id,
now,
USER_STATUS,
awaitSwarmJoin,
awaitSwarmLeave
} = require('pearcord-shared')
const { attachContactsMesh, broadcastContactsGossip } = require('./mesh')
const { normalizeActivity } = require('pearcord-activity')
@@ -418,7 +429,7 @@ class PearcordContacts extends EventEmitter {
const topic = globalPresenceTopic(this.userId)
const buf = topicToBuffer(topic)
if (!this._globalPresenceTopics.has(topic)) {
await this.swarm.join(buf, { server: true, client: true })
await awaitSwarmJoin(this.swarm, buf, { server: true, client: true })
this._globalPresenceTopics.add(topic)
await this._flushSwarm(2500)
}
@@ -588,12 +599,15 @@ class PearcordContacts extends EventEmitter {
const want = new Set(accepted.map((r) => contactsTopic(r.peerUserId)))
for (const topic of want) {
if (this._friendTopics.has(topic)) continue
await this.swarm.join(topicToBuffer(topic), { server: false, client: true })
await awaitSwarmJoin(this.swarm, topicToBuffer(topic), {
server: false,
client: true
})
this._friendTopics.add(topic)
}
for (const topic of [...this._friendTopics]) {
if (want.has(topic)) continue
await this.swarm.leave(topicToBuffer(topic)).catch(() => {})
await awaitSwarmLeave(this.swarm, topicToBuffer(topic))
this._friendTopics.delete(topic)
}
await this._syncGlobalPresenceTopicLinks(accepted)
@@ -604,7 +618,7 @@ class PearcordContacts extends EventEmitter {
if (!this.swarm || !this._meshJoined || !this._globalPresenceEnabled) {
for (const topic of [...this._globalPresenceTopics]) {
if (topic === globalPresenceTopic(this.userId)) continue
await this.swarm.leave(topicToBuffer(topic)).catch(() => {})
await awaitSwarmLeave(this.swarm, topicToBuffer(topic))
this._globalPresenceTopics.delete(topic)
}
return
@@ -617,12 +631,15 @@ class PearcordContacts extends EventEmitter {
for (const topic of want) {
if (this._globalPresenceTopics.has(topic)) continue
const own = topic === globalPresenceTopic(this.userId)
await this.swarm.join(topicToBuffer(topic), { server: own, client: true })
await awaitSwarmJoin(this.swarm, topicToBuffer(topic), {
server: own,
client: true
})
this._globalPresenceTopics.add(topic)
}
for (const topic of [...this._globalPresenceTopics]) {
if (want.has(topic)) continue
await this.swarm.leave(topicToBuffer(topic)).catch(() => {})
await awaitSwarmLeave(this.swarm, topicToBuffer(topic))
this._globalPresenceTopics.delete(topic)
}
}
@@ -639,7 +656,10 @@ class PearcordContacts extends EventEmitter {
if (!this.swarm || !channelId) return null
const topic = dmVoiceTopic(channelId)
if (this._dmVoiceTopics.has(topic)) return topic
await this.swarm.join(topicToBuffer(topic), { server: true, client: true })
await awaitSwarmJoin(this.swarm, topicToBuffer(topic), {
server: true,
client: true
})
this._dmVoiceTopics.add(topic)
await this._flushSwarm(2500)
return topic
@@ -649,7 +669,7 @@ class PearcordContacts extends EventEmitter {
if (!this.swarm || !channelId) return
const topic = dmVoiceTopic(channelId)
if (!this._dmVoiceTopics.has(topic)) return
await this.swarm.leave(topicToBuffer(topic)).catch(() => {})
await awaitSwarmLeave(this.swarm, topicToBuffer(topic))
this._dmVoiceTopics.delete(topic)
}
@@ -666,13 +686,16 @@ class PearcordContacts extends EventEmitter {
}
for (const topic of want) {
if (this._friendTopics.has(topic) || this._pendingTopics.has(topic)) continue
await this.swarm.join(topicToBuffer(topic), { server: true, client: true }).catch(() => {})
await awaitSwarmJoin(this.swarm, topicToBuffer(topic), {
server: true,
client: true
})
this._pendingTopics.add(topic)
this._ephemeralTopics.add(topic)
}
for (const topic of [...this._pendingTopics]) {
if (want.has(topic) || this._friendTopics.has(topic)) continue
await this.swarm.leave(topicToBuffer(topic)).catch(() => {})
await awaitSwarmLeave(this.swarm, topicToBuffer(topic))
this._pendingTopics.delete(topic)
this._ephemeralTopics.delete(topic)
}
@@ -722,9 +745,10 @@ class PearcordContacts extends EventEmitter {
// Ensure topic still joined (peer may have come online since last leave path).
const topic = contactsTopic(row.peerUserId)
if (!this._friendTopics.has(topic) && !this._pendingTopics.has(topic)) {
await this.swarm
?.join(topicToBuffer(topic), { server: true, client: true })
.catch(() => {})
await awaitSwarmJoin(this.swarm, topicToBuffer(topic), {
server: true,
client: true
})
this._pendingTopics.add(topic)
this._ephemeralTopics.add(topic)
}
@@ -772,7 +796,7 @@ class PearcordContacts extends EventEmitter {
const keepPending =
method === RPC.CONTACT_REQUEST || this._pendingTopics.has(topic)
if (!persistent && !this._pendingTopics.has(topic) && !this._ephemeralTopics.has(topic)) {
await this.swarm.join(buf, { server: true, client: true })
await awaitSwarmJoin(this.swarm, buf, { server: true, client: true })
this._ephemeralTopics.add(topic)
if (keepPending || method === RPC.CONTACT_REQUEST) {
this._pendingTopics.add(topic)
@@ -781,10 +805,11 @@ class PearcordContacts extends EventEmitter {
this._pendingTopics.add(topic)
}
// Keep the send path snappy: short flush + brief channel wait, then background retries.
const flushMs = Number(process.env.PEARCORD_CONTACTS_GOSSIP_FLUSH_MS) || 2500
// 800ms default (was 2500) — friend-request IPC should not block ~3s on DHT flush.
const flushMs = Number(process.env.PEARCORD_CONTACTS_GOSSIP_FLUSH_MS) || 800
await this._flushSwarm(flushMs)
await this._waitForContactsChannels(
Number(process.env.PEARCORD_CONTACTS_GOSSIP_WAIT_MS) || 1500
Number(process.env.PEARCORD_CONTACTS_GOSSIP_WAIT_MS) || 900
)
let delivered = 0
@@ -817,7 +842,7 @@ class PearcordContacts extends EventEmitter {
await broadcastContactsGossip(this, method, payload).catch(() => {})
// Only leave if still not held as pending_out / friend.
if (!this._friendTopics.has(topic) && !this._pendingTopics.has(topic)) {
await this.swarm.leave(buf).catch(() => {})
await awaitSwarmLeave(this.swarm, buf)
this._ephemeralTopics.delete(topic)
}
}
@@ -865,24 +890,25 @@ class PearcordContacts extends EventEmitter {
}
}
const joinMs = Number(process.env.PEARCORD_MESH_JOIN_MS) || 8000
await Promise.race([
this.swarm.join(topicToBuffer(contactsTopic(this.userId)), {
server: true,
client: true
}),
new Promise((resolve) => setTimeout(resolve, joinMs))
])
// join() returns PeerDiscoverySession (not a Promise) — do not race raw join.
await awaitSwarmJoin(this.swarm, topicToBuffer(contactsTopic(this.userId)), {
server: true,
client: true
})
if (this._globalPresenceEnabled) {
await Promise.race([
this.swarm.join(topicToBuffer(globalPresenceTopic(this.userId)), {
server: true,
client: true
}),
new Promise((resolve) => setTimeout(resolve, joinMs))
])
await awaitSwarmJoin(
this.swarm,
topicToBuffer(globalPresenceTopic(this.userId)),
{ server: true, client: true }
)
this._globalPresenceTopics.add(globalPresenceTopic(this.userId))
}
await this._flushSwarm(3500)
// Cap initial DHT settle so session startup meshes cannot hang forever.
const flushCap = Math.min(3500, joinMs)
await Promise.race([
this._flushSwarm(flushCap),
new Promise((resolve) => setTimeout(resolve, flushCap))
])
this._meshJoined = true
const linkMs = Number(process.env.PEARCORD_CONTACTS_FRIEND_LINK_MS) || 8000
await Promise.race([
@@ -912,8 +938,11 @@ class PearcordContacts extends EventEmitter {
this._globalPresenceTopics.clear()
this._meshJoined = false
if (this.swarm) {
await this.swarm.leave(topicToBuffer(contactsTopic(this.userId))).catch(() => {})
await this.swarm.leave(topicToBuffer(globalPresenceTopic(this.userId))).catch(() => {})
await awaitSwarmLeave(this.swarm, topicToBuffer(contactsTopic(this.userId)))
await awaitSwarmLeave(
this.swarm,
topicToBuffer(globalPresenceTopic(this.userId))
)
if (this._connectionHandler) {
try {
this.swarm.removeListener('connection', this._connectionHandler)