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

Add mesh churn observability and reaction list light-poll cache.

Browse Source 
Emit guild.mesh.churn spans on peer join/leave, push slot rotation, and gossip outbox flush; refresh discovery after offline recovery; cache reactions during light view polls to cut HyperDB load.

Co-authored-by: Cursor <[email protected]>
This commit is contained in:
Raven Scott
2026-06-01 20:48:45 -04:00
co-authored by Cursor
parent aa62cd6562
commit bf14d15b68
1 changed files with 116 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
index.js
+116 -27
View File
@@ -454,6 +454,30 @@ class PearcordPlatform extends EventEmitter {
this._contactsPresenceDebounce = null
this._lastGuildSyncIngestStart = 0
this._guildSyncBandwidthPeerIndex = 0
this._reactionListCacheKey = ''
this._reactionListCache = null
this._reactionListCacheGen = 0
}
_invalidateReactionListCache () {
this._reactionListCacheGen = (this._reactionListCacheGen || 0) + 1
this._reactionListCacheKey = ''
this._reactionListCache = null
}
_logMeshChurn (action, extra = {}) {
const guildId = extra.guildId || this.guild?.guild?.id
if (!guildId) return
this.log.info('guild.mesh.churn', {
spanKind: 'guild.mesh.churn',
action,
guildId,
meshPeerCount: this.guild?.peers?.size ?? 0,
pushPeerSlot: this._guildSyncPushPeerSlot ?? null,
pushHalted: this._isGuildSyncPushHalted(guildId),
pendingGossipCount: this._guildGossipOutbox?.size ?? 0,
...extra
})
}
_reactionTombstoneKey (key, guildId) {
@@ -3461,12 +3485,30 @@ class PearcordPlatform extends EventEmitter {
return { queued: true }
}
async _flushGuildGossipOutbox () {
async _flushGuildGossipOutbox (opts = {}) {
const guildId = this.guild?.guild?.id
const before = this._guildGossipOutbox.size
const out = await this._guildGossipOutbox.flush()
this._patchGuildSyncHealth({
guildId: this.guild?.guild?.id,
guildId,
pendingGossipCount: this._guildGossipOutbox.size
})
const flushed = Number(out?.flushed) || 0
if (guildId && flushed > 0) {
this._logMeshChurn('gossip-outbox-flush', {
guildId,
flushed,
remaining: Number(out?.remaining) || this._guildGossipOutbox.size,
reason: opts.reason || null
})
} else if (guildId && before > 0 && opts.reason === 'peer-join') {
this._logMeshChurn('gossip-outbox-flush', {
guildId,
flushed: 0,
remaining: before,
reason: 'peer-join-empty'
})
}
return out
}
@@ -4256,6 +4298,7 @@ class PearcordPlatform extends EventEmitter {
createdAt
})
if (inTs > tombTs) this._reactionTombstones.delete(tombKey)
this._invalidateReactionListCache()
return 1
}
@@ -4270,7 +4313,10 @@ class PearcordPlatform extends EventEmitter {
changed += await this._mergeReactionGossip(r)
}
span.end({ changed, rows: reactions?.length || 0 })
if (changed > 0) this.emit('reaction')
if (changed > 0) {
this._invalidateReactionListCache()
this.emit('reaction')
}
return changed
}
@@ -4689,7 +4735,7 @@ class PearcordPlatform extends EventEmitter {
return {
peers: this._settingsHealthChannels?.size ?? 0,
topic: this.identity?.user?.id
? require('pearcord-shared').settingsTopic(this.identity.user.id)
? require('pearcord-swarm-manager').settingsMeshTopicLabel(this.identity.user.id)
: null,
lastGossipAt: this._settingsMeshHealthLastAt || null,
lastRemoteDeviceId: lastRemote?.deviceId || null,
@@ -5444,10 +5490,18 @@ class PearcordPlatform extends EventEmitter {
action: 'offline-recovery',
guildId
})
const flush = await this._flushGuildGossipOutbox().catch(() => ({ flushed: 0 }))
this._logMeshChurn('offline-recovery-start', { guildId })
const flush = await this._flushGuildGossipOutbox({ reason: 'offline-recovery' }).catch(
() => ({ flushed: 0 })
)
this.clearGuildSyncPushHalt(guildId)
const req = this.requestGuildSyncFromHost()
const burst = this.requestGuildSyncBurst({ clearPushHalt: false, forcePush: true })
await this.refreshDiscoveryListingsFromMesh().catch(() => {})
this._logMeshChurn('offline-recovery-complete', {
guildId,
flushed: Number(flush?.flushed ?? flush?.count) || 0
})
span.end({
guildId,
flushed: Number(flush?.flushed ?? flush?.count) || 0,
@@ -5970,10 +6024,18 @@ class PearcordPlatform extends EventEmitter {
}
await this.guild.gossipGuildSync(wireBundle)
const peerCount = this.guild.peers?.size ?? 0
const prevSlot = this._guildSyncPushPeerSlot
this._guildSyncPushPeerSlot = nextRoundRobinIndex(
Math.max(1, peerCount),
this._guildSyncPushPeerSlot
)
if (peerCount > 1 && this._guildSyncPushPeerSlot !== prevSlot) {
this._logMeshChurn('push-peer-slot', {
guildId,
pushPeerSlot: this._guildSyncPushPeerSlot,
meshPeerCount: peerCount
})
}
const prevHealth = this.getGuildSyncHealth(guildId)
await this._refreshGuildSyncTailWatermark(guildId, opts.priorityChannelId || opts.channelId)
this._patchGuildSyncHealth({
@@ -6636,6 +6698,11 @@ class PearcordPlatform extends EventEmitter {
const gid = this.guild?.guild?.id
const peers = this.guild?.peers?.size ?? 0
if (gid) {
this._logMeshChurn('peer-leave', {
guildId: gid,
peerId: peerId || null,
meshPeerCount: peers
})
this._patchGuildSyncHealth({
guildId: gid,
peers,
@@ -6662,8 +6729,15 @@ class PearcordPlatform extends EventEmitter {
40,
Math.min(800, Date.now() - joinedAt)
)
void this._flushGuildGossipOutbox().catch(() => {})
void this._flushGuildGossipOutbox({ reason: 'peer-join' }).catch(() => {})
const gid = this.guild?.guild?.id
if (gid) {
this._logMeshChurn('peer-join', {
guildId: gid,
peerId: peerId || null,
meshPeerCount: this.guild?.peers?.size ?? 0
})
}
const conn = peerId ? this.guild?.peers?.get(peerId) : null
if (gid && conn) {
void this._getGuildReplicator(gid)
@@ -17207,6 +17281,7 @@ class PearcordPlatform extends EventEmitter {
})
gossipSent = true
}
this._invalidateReactionListCache()
this.emit('reaction', result)
const reactionRows = await this.messages.listReactions()
const reactionCount = reactionRows.filter(
@@ -17530,31 +17605,45 @@ class PearcordPlatform extends EventEmitter {
if (this.messages) {
const reactionGuildId =
this.mode === 'dm' ? DM_GUILD_ID : this.guild?.guild?.id || null
const listSpan = this.log.time('reaction.list', {
spanKind: 'reaction.list',
channelId: this.activeChannelId,
guildId: reactionGuildId
})
try {
const rows = await this.messages.listReactions()
reactions = this._groupReactions(rows)
listSpan.end({
const cacheKey = `${reactionGuildId || ''}:${this.activeChannelId || ''}:${this._reactionListCacheGen}`
if (
light &&
this._reactionListCacheKey === cacheKey &&
this._reactionListCache?.reactions
) {
reactions = this._reactionListCache.reactions
} else {
const listSpan = this.log.time('reaction.list', {
spanKind: 'reaction.list',
channelId: this.activeChannelId,
guildId: reactionGuildId,
rowCount: rows.length,
messageCount: Object.keys(reactions).length,
activeChannelId: this.activeChannelId,
guildCount: (this.guilds || []).length
light
})
} catch (err) {
this.log.error('reaction.list error', {
channelId: this.activeChannelId,
guildId: reactionGuildId,
error: err?.message || String(err)
})
listSpan.fail(err)
reactions = {}
try {
const rows = await this.messages.listReactions()
reactions = this._groupReactions(rows)
this._reactionListCacheKey = cacheKey
this._reactionListCache = { rows, reactions }
listSpan.end({
spanKind: 'reaction.list',
channelId: this.activeChannelId,
guildId: reactionGuildId,
rowCount: rows.length,
messageCount: Object.keys(reactions).length,
activeChannelId: this.activeChannelId,
guildCount: (this.guilds || []).length,
cached: false,
light
})
} catch (err) {
this.log.error('reaction.list error', {
channelId: this.activeChannelId,
guildId: reactionGuildId,
error: err?.message || String(err)
})
listSpan.fail(err)
reactions = {}
}
}
}
this.guilds = await this.listGuilds()