Phase 871: protomux-rpc MEMBER_PAGE_REQUEST pull (v0.8.838)

Opt-in member page RPC pull mirrors guild sync request wire: _buildMemberPagePayload,
_handleMemberPageRequestRpc, _tryMemberPageRequestRpc, RPC-first requestGuildMemberPage
with gossip fanout fallback. Non-breaking; default event path unchanged.
This commit is contained in:
Raven Scott
2026-06-04 16:52:55 -04:00
parent 727ef8c764
commit 89639389d5
4 changed files with 102 additions and 24 deletions
+2
View File
@@ -92,6 +92,8 @@ Application facade: one `PearcordPlatform` class that wires identity, database,
**Phase 822 (v0.8.789):** Triple-peer topology via `guild-mesh-topology` + `getMeshStabilityStats`. Bundle: `npm run test:ci-phase822`.
**Phase 871 (v0.8.838):** Opt-in protomux-rpc `MEMBER_PAGE_REQUEST` pull via `_tryMemberPageRequestRpc` (`PEARCORD_RPC_REQUEST_WIRE=1`). Bundle: `npm run test:ci-phase871`.
**Phase 869 (v0.8.836):** Opt-in protomux-rpc `GUILD_SYNC_REQUEST` pull via `_tryGuildSyncRequestRpc` (`PEARCORD_RPC_REQUEST_WIRE=1`). Bundle: `npm run test:ci-phase869`.
**Phase 821 (v0.8.788):** Triple-peer ACK watermark via `_fanoutGuildSyncRequestWithAckWatermarks`. Bundle: `npm run test:ci-phase821`.
@@ -295,13 +295,30 @@ requestGuildMemberPage (offset = 0, opts = {}) {
}
const off = Math.max(0, Number(offset) || 0)
if (!opts.debounced) this._recordMemberPageBurst()
this.guild
.gossipMemberPageRequest({
const fanoutMemberPageRequest = () => {
this.guild
.gossipMemberPageRequest({
guildId,
requesterId: userId,
offset: off
})
.catch(() => {})
}
if (sharedScope.isRpcGossipRequestEnabled?.()) {
void this._tryMemberPageRequestRpc(off)
.then((rpc) => {
if (!rpc?.viaRpc) fanoutMemberPageRequest()
})
.catch(() => fanoutMemberPageRequest())
return {
guildId,
requesterId: userId,
offset: off
})
.catch(() => {})
offset: off,
requested: true,
pendingRpc: true,
burstRemaining: this._memberPageBurstRemaining()
}
}
fanoutMemberPageRequest()
return {
guildId,
offset: off,
@@ -329,23 +346,12 @@ async _handleGuildSyncRequestRpc (payload) {
return bundle
},
async _handleMemberPageRequest (payload) {
const span = this.log.time('guild.sync.member-page', {
spanKind: 'guild.sync.member-page',
guildId: payload?.guildId,
offset: payload?.offset
})
async _buildMemberPagePayload (payload) {
const guildId = this.guild?.guild?.id
if (!guildId || payload?.guildId !== guildId) {
span.end({ sent: 0, reason: 'guild-mismatch' })
return 0
}
if (!guildId || payload?.guildId !== guildId) return null
const isOwner = this.guild.guild.ownerId === this.identity?.user?.id
const canMod = await this._canModeratorPushGuildSync()
if (!isOwner && !canMod) {
span.end({ sent: 0, reason: 'forbidden' })
return 0
}
if (!isOwner && !canMod) return null
const allMembers = await this.guild.listMembers()
const offset = Math.max(0, Number(payload.offset) || 0)
const pageSize = this._getGuildSyncMemberPageSize()
@@ -354,16 +360,40 @@ async _handleMemberPageRequest (payload) {
)
const page = sorted.slice(offset, offset + pageSize)
const users = await this._collectUsersForSync(page)
await this.guild.gossipMemberPage({
return {
guildId,
offset,
members: page.map((m) => hyperdbScope.sanitizeMemberForHyperDb(m)),
users,
memberTotal: sorted.length,
requesterId: payload.requesterId || null
}
},
async _handleMemberPageRequestRpc (payload) {
const page = await this._buildMemberPagePayload(payload)
if (page) page.viaRpc = true
return page
},
async _handleMemberPageRequest (payload) {
const span = this.log.time('guild.sync.member-page', {
spanKind: 'guild.sync.member-page',
guildId: payload?.guildId,
offset: payload?.offset
})
span.end({ sent: page.length, offset, memberTotal: sorted.length })
return page.length
const page = await this._buildMemberPagePayload(payload)
if (!page) {
span.end({ sent: 0, reason: 'forbidden-or-mismatch' })
return 0
}
await this.guild.gossipMemberPage(page)
span.end({
sent: page.members.length,
offset: page.offset,
memberTotal: page.memberTotal
})
return page.members.length
},
async _handleMemberJoinIdentifyRequest (payload) {
@@ -34,7 +34,10 @@ const platformGuildWireMixin = {
_bindGuildGossipRpcHandlers (guildInstance) {
const enabled = !!sharedScope.isRpcGossipRequestEnabled?.()
const next = enabled
? { GUILD_SYNC_REQUEST: (payload) => this._handleGuildSyncRequestRpc(payload) }
? {
GUILD_SYNC_REQUEST: (payload) => this._handleGuildSyncRequestRpc(payload),
MEMBER_PAGE_REQUEST: (payload) => this._handleMemberPageRequestRpc(payload)
}
: null
const hadHandlers = !!guildInstance._gossipRpcHandlers
guildInstance._gossipRpcHandlers = next
@@ -17,6 +17,49 @@ function meshPeerIdHex (pk) {
}
const platformMeshViewMixin = {
async _tryMemberPageRequestRpc (offset = 0) {
if (!sharedScope.isRpcGossipRequestEnabled?.()) return null
const guildId = this.guild?.guild?.id
const userId = this.identity?.user?.id
if (!guildId || !userId || !this.guild?.requestGossipFromPeer) return null
let hostPk = meshPeerIdHex(await this._lookupGuildHostPublicKey(this.guild.guild))
const peerKeys = [...(this.guild.peers?.keys() || [])]
if (hostPk && !this.guild.peers?.has(hostPk) && peerKeys.length === 1) {
hostPk = peerKeys[0]
}
if (!hostPk || !(this.guild.peers?.has(hostPk))) return null
await this._flushGuildGossipOutbox().catch(() => {})
try {
const { ensureGuildWireChannels } = require('pearcord-guild/mesh')
ensureGuildWireChannels(this.guild)
} catch {
// ignore
}
const page = await this.guild.requestGossipFromPeer(
hostPk,
sharedScope.RPC.MEMBER_PAGE_REQUEST,
{
guildId,
requesterId: userId,
offset: Math.max(0, Number(offset) || 0)
},
{
timeout: Number(process.env.PEARCORD_GUILD_MEMBER_PAGE_RPC_MS) || 25000,
wireReadyMs:
Number(process.env.PEARCORD_GUILD_MEMBER_PAGE_RPC_WIRE_READY_MS) || 12000
}
)
if (!page?.guildId) return null
const added = await this._ingestMemberPage(page)
return {
requested: true,
viaRpc: true,
offset: page.offset,
memberCount: page.members?.length || 0,
added
}
},
async _tryGuildSyncRequestRpc (channelId) {
if (!sharedScope.isRpcGossipRequestEnabled?.()) return null
const guildId = this.guild?.guild?.id