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

fix: deliver friend requests reliably on shared contacts mesh

Browse Source 
Hold pending_out topics with redelivery retries, and broadcast with
sendIfNotReady so CONTACT_REQUEST lands on shared-swarm sockets without
multi-second fullyOpened stalls.
This commit is contained in:
Raven Scott
2026-07-13 15:39:05 -04:00
parent 858ff8d953
commit 3fe8075299
2 changed files with 226 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
index.js
+219 -11
View File
@@ -49,6 +49,8 @@ class PearcordContacts extends EventEmitter {
this._channels = new Map()
this._meshJoined = false
this._ephemeralTopics = new Set()
/** Topics held while we have pending_out (or other durable 1:1 contact RPCs). */
this._pendingTopics = new Set()
this._friendTopics = new Set()
this._dmVoiceTopics = new Set()
this._globalPresenceTopics = new Set()
@@ -56,6 +58,10 @@ class PearcordContacts extends EventEmitter {
this._blockedPeerIds = new Set()
this._friendPresence = new Map()
this._lastSelfPresence = null
/** peerUserId → last CONTACT_REQUEST payload for reconnect redelivery */
this._pendingOutPayloads = new Map()
this._pendingRetryTimer = null
this._pendingRetryInFlight = false
}
async ready () {
@@ -112,7 +118,17 @@ class PearcordContacts extends EventEmitter {
if (existing?.status === CONTACT_STATUS.BLOCKED) {
throw new Error('contact blocked')
}
// Re-gossip even when already pending_out — first attempt often soft-delivers 0
// peers before DHT/wire is ready; early-return made retries a no-op.
if (existing?.status === CONTACT_STATUS.PENDING_OUT) {
const payload = {
...this._profilePayload(),
toUserId: peerUserId,
peerDisplayName: existing.peerDisplayName || peerDisplayName || peerUserId.slice(0, 8)
}
this._pendingOutPayloads.set(peerUserId, payload)
await this._gossipToPeer(peerUserId, RPC.CONTACT_REQUEST, payload)
this.emit('contact', existing)
return existing
}
const row = {
@@ -130,6 +146,7 @@ class PearcordContacts extends EventEmitter {
toUserId: peerUserId,
peerDisplayName: row.peerDisplayName
}
this._pendingOutPayloads.set(peerUserId, payload)
await this._gossipToPeer(peerUserId, RPC.CONTACT_REQUEST, payload)
this.emit('contact', row)
return row
@@ -149,7 +166,9 @@ class PearcordContacts extends EventEmitter {
row.status = CONTACT_STATUS.ACCEPTED
row.acceptedAt = now()
await this.store.insert(CONTACTS_COLLECTION, row)
this._pendingOutPayloads.delete(peerUserId)
await this._syncFriendTopicLinks()
await this._syncPendingTopicLinks()
await this._gossipToPeer(peerUserId, RPC.CONTACT_ACCEPT, {
...this._profilePayload(),
toUserId: peerUserId,
@@ -165,6 +184,8 @@ class PearcordContacts extends EventEmitter {
throw new Error('no pending request')
}
await this.store.delete(CONTACTS_COLLECTION, contactKey(this.userId, peerUserId))
this._pendingOutPayloads.delete(peerUserId)
await this._syncPendingTopicLinks()
await this._gossipToPeer(peerUserId, RPC.CONTACT_DECLINE, {
...this._profilePayload(),
toUserId: peerUserId
@@ -177,6 +198,7 @@ class PearcordContacts extends EventEmitter {
const row = await this.get(peerUserId)
if (!row) return null
await this.store.delete(CONTACTS_COLLECTION, contactKey(this.userId, peerUserId))
this._pendingOutPayloads.delete(peerUserId)
if (row.status === CONTACT_STATUS.ACCEPTED || row.status === CONTACT_STATUS.PENDING_OUT) {
await this._gossipToPeer(peerUserId, RPC.CONTACT_REMOVE, {
...this._profilePayload(),
@@ -184,6 +206,7 @@ class PearcordContacts extends EventEmitter {
}).catch(() => {})
}
await this._syncFriendTopicLinks()
await this._syncPendingTopicLinks()
this.emit('contact', null)
return row
}
@@ -259,9 +282,21 @@ class PearcordContacts extends EventEmitter {
async ingestRequest (payload) {
const from = payload?.fromUserId
if (!from || from === this.userId) return null
// Drop mis-addressed frames (broadcast noise on shared swarm sockets).
if (payload?.toUserId && payload.toUserId !== this.userId) return null
const blocked = await this.get(from)
if (blocked?.status === CONTACT_STATUS.BLOCKED) return null
if (blocked?.status === CONTACT_STATUS.ACCEPTED) return blocked
// Idempotent: already pending_in — refresh display fields, re-emit so UI refreshes.
if (blocked?.status === CONTACT_STATUS.PENDING_IN) {
blocked.peerUsername = payload.fromUsername || blocked.peerUsername || null
blocked.peerDisplayName =
payload.fromDisplayName || payload.fromUsername || blocked.peerDisplayName
await this.store.insert(CONTACTS_COLLECTION, blocked)
this.emit('contact-request', blocked)
this.emit('contact', blocked)
return blocked
}
const mutual = blocked?.status === CONTACT_STATUS.PENDING_OUT
const row = {
id: blocked?.id || id(),
@@ -274,6 +309,7 @@ class PearcordContacts extends EventEmitter {
acceptedAt: mutual ? now() : null
}
if (mutual) {
this._pendingOutPayloads.delete(from)
await this._gossipToPeer(from, RPC.CONTACT_ACCEPT, {
...this._profilePayload(),
toUserId: from,
@@ -281,7 +317,10 @@ class PearcordContacts extends EventEmitter {
}).catch(() => {})
}
await this.store.insert(CONTACTS_COLLECTION, row)
if (mutual) await this._syncFriendTopicLinks()
if (mutual) {
await this._syncFriendTopicLinks()
await this._syncPendingTopicLinks()
}
this.emit('contact-request', row)
this.emit('contact', row)
return row
@@ -290,6 +329,7 @@ class PearcordContacts extends EventEmitter {
async ingestAccept (payload) {
const from = payload?.fromUserId
if (!from) return null
if (payload?.toUserId && payload.toUserId !== this.userId) return null
let row = await this.get(from)
if (!row) {
row = {
@@ -305,7 +345,9 @@ class PearcordContacts extends EventEmitter {
row.peerUsername = payload.fromUsername || row.peerUsername || null
row.peerDisplayName = payload.fromDisplayName || payload.fromUsername || row.peerDisplayName
await this.store.insert(CONTACTS_COLLECTION, row)
this._pendingOutPayloads.delete(from)
await this._syncFriendTopicLinks()
await this._syncPendingTopicLinks()
this.emit('contact', row)
return row
}
@@ -316,6 +358,8 @@ class PearcordContacts extends EventEmitter {
const row = await this.get(from)
if (row?.status === CONTACT_STATUS.PENDING_OUT) {
await this.store.delete(CONTACTS_COLLECTION, contactKey(this.userId, from))
this._pendingOutPayloads.delete(from)
await this._syncPendingTopicLinks()
this.emit('contact', null)
}
return null
@@ -325,6 +369,8 @@ class PearcordContacts extends EventEmitter {
const from = payload?.fromUserId
if (!from) return null
await this.store.delete(CONTACTS_COLLECTION, contactKey(this.userId, from))
this._pendingOutPayloads.delete(from)
await this._syncPendingTopicLinks()
this.emit('contact', null)
return null
}
@@ -607,28 +653,175 @@ class PearcordContacts extends EventEmitter {
this._dmVoiceTopics.delete(topic)
}
/**
* Hold the peer's contacts topic while we still have pending_out rows so a late
* online peer (or slow DHT) can still receive CONTACT_REQUEST without a re-send.
*/
async _syncPendingTopicLinks () {
if (!this.swarm || !this._meshJoined) return
const pending = await this.listPendingOutgoing().catch(() => [])
const want = new Set()
for (const row of pending) {
if (row?.peerUserId) want.add(contactsTopic(row.peerUserId))
}
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(() => {})
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(() => {})
this._pendingTopics.delete(topic)
this._ephemeralTopics.delete(topic)
}
if (want.size) this._armPendingRetryTimer()
else this._clearPendingRetryTimer()
}
_armPendingRetryTimer () {
if (this._pendingRetryTimer || !this._meshJoined) return
const everyMs = Math.max(
4000,
Number(process.env.PEARCORD_CONTACTS_PENDING_RETRY_MS) || 12000
)
this._pendingRetryTimer = setInterval(() => {
void this._retryPendingOutgoingGossip().catch(() => {})
}, everyMs)
this._pendingRetryTimer.unref?.()
}
_clearPendingRetryTimer () {
if (!this._pendingRetryTimer) return
clearInterval(this._pendingRetryTimer)
this._pendingRetryTimer = null
}
async _retryPendingOutgoingGossip () {
if (!this._meshJoined || this._pendingRetryInFlight) return { delivered: 0, tried: 0 }
this._pendingRetryInFlight = true
try {
const pending = await this.listPendingOutgoing().catch(() => [])
if (!pending.length) {
this._clearPendingRetryTimer()
return { delivered: 0, tried: 0 }
}
let delivered = 0
let tried = 0
for (const row of pending) {
if (!row?.peerUserId) continue
const payload =
this._pendingOutPayloads.get(row.peerUserId) || {
...this._profilePayload(),
toUserId: row.peerUserId,
peerDisplayName: row.peerDisplayName
}
this._pendingOutPayloads.set(row.peerUserId, payload)
tried++
// 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(() => {})
this._pendingTopics.add(topic)
this._ephemeralTopics.add(topic)
}
const n = await broadcastContactsGossip(this, RPC.CONTACT_REQUEST, payload).catch(
() => 0
)
delivered += Number(n) || 0
}
return { delivered, tried }
} finally {
this._pendingRetryInFlight = false
}
}
async _waitForContactsChannels (timeoutMs = 5000) {
const deadline = Date.now() + Math.max(200, Number(timeoutMs) || 5000)
while (Date.now() < deadline) {
if (this._channels.size > 0) {
// Prefer at least one session that looks open when possible.
for (const session of this._channels.values()) {
if (session?.opened || session?.v2?.opened) return true
if (typeof session?.fullyOpened === 'function') {
const ok = await Promise.race([
session.fullyOpened().catch(() => false),
new Promise((r) => setTimeout(() => r(false), 120))
])
if (ok) return true
}
}
return true
}
await new Promise((r) => setTimeout(r, 100))
}
return this._channels.size > 0
}
async _gossipToPeer (peerUserId, method, payload) {
if (!this.swarm) {
await this.gossipLocal(method, payload)
return
return 0
}
const topic = contactsTopic(peerUserId)
const buf = topicToBuffer(topic)
const persistent = this._friendTopics.has(topic)
const wasEphemeral = this._ephemeralTopics.has(topic)
if (!persistent && !wasEphemeral) {
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 })
this._ephemeralTopics.add(topic)
if (keepPending || method === RPC.CONTACT_REQUEST) {
this._pendingTopics.add(topic)
}
} else if (method === RPC.CONTACT_REQUEST) {
this._pendingTopics.add(topic)
}
await this._flushSwarm(6000)
await broadcastContactsGossip(this, method, payload)
if (!persistent && !wasEphemeral) {
await new Promise((r) => setTimeout(r, 1200))
await broadcastContactsGossip(this, method, payload).catch(() => {})
// Keep the send path snappy: short flush + brief channel wait, then background retries.
const flushMs = Number(process.env.PEARCORD_CONTACTS_GOSSIP_FLUSH_MS) || 2500
await this._flushSwarm(flushMs)
await this._waitForContactsChannels(
Number(process.env.PEARCORD_CONTACTS_GOSSIP_WAIT_MS) || 1500
)
let delivered = 0
const attempts = method === RPC.CONTACT_REQUEST ? 2 : 1
for (let i = 0; i < attempts; i++) {
const n = await broadcastContactsGossip(this, method, payload).catch(() => 0)
delivered = Math.max(delivered, Number(n) || 0)
if (delivered > 0) break
if (i + 1 < attempts) {
await new Promise((r) => setTimeout(r, 350))
await this._flushSwarm(800)
}
}
// CONTACT_REQUEST stays on the peer topic until accept/decline/cancel so late
// peers still receive it. Other one-shots may leave after a short rebroadcast.
if (method === RPC.CONTACT_REQUEST) {
this._armPendingRetryTimer()
// Background waves — do not block UI / sendFriendRequest.
for (const delay of [800, 2500, 6000]) {
setTimeout(() => {
void broadcastContactsGossip(this, method, payload).catch(() => {})
}, delay).unref?.()
}
return delivered
}
if (!persistent && !this._pendingTopics.has(topic)) {
await new Promise((r) => setTimeout(r, 400))
await this.swarm.leave(buf).catch(() => {})
this._ephemeralTopics.delete(topic)
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(() => {})
this._ephemeralTopics.delete(topic)
}
}
return delivered
}
async joinMesh (swarm) {
@@ -651,6 +844,10 @@ class PearcordContacts extends EventEmitter {
const wire = attachContactsMesh(this, conn)
if (wire) this._channels.set(peerId, wire)
this.emit('peer', { peerId, type: 'join', peerInfo: peerInfo || null })
// Redeliver pending friend requests once the contacts wire is up.
setTimeout(() => {
void this._retryPendingOutgoingGossip().catch(() => {})
}, 200).unref?.()
if (!conn._pearcordContactsClose) {
conn._pearcordContactsClose = true
conn.on('close', () => {
@@ -692,16 +889,25 @@ class PearcordContacts extends EventEmitter {
this._syncFriendTopicLinks(),
new Promise((resolve) => setTimeout(resolve, linkMs))
])
// Restore pending_out topic subscriptions after restart so unacked requests
// can still reach a peer who comes online later.
await Promise.race([
this._syncPendingTopicLinks(),
new Promise((resolve) => setTimeout(resolve, linkMs))
])
if (this._lastSelfPresence) {
await broadcastContactsGossip(this, RPC.PRESENCE_UPDATE, this._lastSelfPresence)
}
void this._retryPendingOutgoingGossip().catch(() => {})
return this
}
async leaveMesh () {
this._clearPendingRetryTimer()
this.peers.clear()
this._channels.clear()
this._ephemeralTopics.clear()
this._pendingTopics.clear()
this._friendTopics.clear()
this._globalPresenceTopics.clear()
this._meshJoined = false
@@ -725,6 +931,8 @@ class PearcordContacts extends EventEmitter {
return {
peers,
friendTopics: this._friendTopics.size,
pendingTopics: this._pendingTopics.size,
pendingOutTracked: this._pendingOutPayloads.size,
globalPresenceTopics: this._globalPresenceTopics.size,
globalPresenceEnabled: this._globalPresenceEnabled,
meshLive: peers > 0
mesh.js
+7 -1
View File
@@ -20,7 +20,13 @@ function attachContactsMesh (contacts, conn) {
}
async function broadcastContactsGossip (contacts, method, payload) {
await broadcastGossipToSessions(contacts._channels, method, payload, { wireReadyMs: 6000 })
// Soft delivery (return count). CONTACT_REQUEST path retries when 0.
// sendIfNotReady: shared-swarm peers often have the Noise socket up before
// protomux-rpc fullyOpened flips; waiting multi-seconds blocked friend requests.
return broadcastGossipToSessions(contacts._channels, method, payload, {
wireReadyMs: Number(process.env.PEARCORD_CONTACTS_WIRE_READY_MS) || 800,
sendIfNotReady: true
})
}
module.exports = { attachContactsMesh, broadcastContactsGossip, CONTACTS_PROTOCOL }