Phase 872: protomux-rpc MESSAGE_SEARCH_REQUEST multi-peer pull (v0.8.839)

Opt-in mesh search RPC to all connected peers with deduped merge and gossip
fanout fallback. Non-breaking; default event path unchanged.
This commit is contained in:
Raven Scott
2026-06-04 16:58:47 -04:00
parent 89639389d5
commit b956ef4764
3 changed files with 87 additions and 12 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 822 (v0.8.789):** Triple-peer topology via `guild-mesh-topology` + `getMeshStabilityStats`. Bundle: `npm run test:ci-phase822`.
**Phase 872 (v0.8.839):** Opt-in protomux-rpc `MESSAGE_SEARCH_REQUEST` multi-peer pull via `_fetchGuildSearchFromMeshRpc` (`PEARCORD_RPC_REQUEST_WIRE=1`). Bundle: `npm run test:ci-phase872`.
**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 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 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`.
@@ -36,7 +36,8 @@ _bindGuildGossipRpcHandlers (guildInstance) {
const next = enabled const next = enabled
? { ? {
GUILD_SYNC_REQUEST: (payload) => this._handleGuildSyncRequestRpc(payload), GUILD_SYNC_REQUEST: (payload) => this._handleGuildSyncRequestRpc(payload),
MEMBER_PAGE_REQUEST: (payload) => this._handleMemberPageRequestRpc(payload) MEMBER_PAGE_REQUEST: (payload) => this._handleMemberPageRequestRpc(payload),
MESSAGE_SEARCH_REQUEST: (payload) => this._handleMessageSearchRequestRpc(payload)
} }
: null : null
const hadHandlers = !!guildInstance._gossipRpcHandlers const hadHandlers = !!guildInstance._gossipRpcHandlers
@@ -2,11 +2,71 @@
const searchScope = require('pearcord-search') const searchScope = require('pearcord-search')
const enrichmentScope = require('../../../platform-message-enrichment') const enrichmentScope = require('../../../platform-message-enrichment')
const sharedScope = require('../../../platform-pearcord-shared-imports')
const { id } = require('../../../platform-class-imports') const { id } = require('../../../platform-class-imports')
const platformSearchMeshGossipMixin = { const platformSearchMeshGossipMixin = {
async _fetchGuildSearchFromMeshRpc (guildId, query, opts = {}) {
if (!sharedScope.isRpcGossipRequestEnabled?.()) return null
if (!this.guild?.guild || this.guild.guild.id !== guildId) return null
if (!this.guild?.requestGossipFromPeer) return null
const peerIds = [...(this.guild.peers?.keys() || [])]
if (!peerIds.length) return null
const requestId = id()
const payload = {
guildId,
requestId,
query: String(query || '').trim(),
limit: opts.limit || 50,
requestedBy: this.identity.user?.id || null,
at: Date.now()
}
await this._flushGuildGossipOutbox().catch(() => {})
try {
const { ensureGuildWireChannels } = require('pearcord-guild/mesh')
ensureGuildWireChannels(this.guild)
} catch {
// ignore
}
const rpcMs = Number(process.env.PEARCORD_MESSAGE_SEARCH_RPC_MS) || 8000
const wireMs =
Number(process.env.PEARCORD_MESSAGE_SEARCH_RPC_WIRE_READY_MS) || 6000
const responses = await Promise.all(
peerIds.map((peerId) =>
this.guild
.requestGossipFromPeer(
peerId,
sharedScope.RPC.MESSAGE_SEARCH_REQUEST,
payload,
{ timeout: rpcMs, wireReadyMs: wireMs }
)
.catch(() => null)
)
)
const anyResponse = responses.some((r) => r && Array.isArray(r.hits))
if (!anyResponse) return null
const merged = []
const seen = new Set()
for (const res of responses) {
if (!res?.hits?.length) continue
for (const h of res.hits) {
const m = searchScope.meshHitToMessage(h, guildId)
if (!m) continue
const key = `${m.channelId}:${m.id}`
if (seen.has(key)) continue
seen.add(key)
merged.push(m)
}
}
return merged
},
async _fetchGuildSearchFromMeshOnce (guildId, query, opts = {}) { async _fetchGuildSearchFromMeshOnce (guildId, query, opts = {}) {
if (!this.guild?.guild || this.guild.guild.id !== guildId) return [] if (!this.guild?.guild || this.guild.guild.id !== guildId) return []
if (sharedScope.isRpcGossipRequestEnabled?.()) {
const rpcHits = await this._fetchGuildSearchFromMeshRpc(guildId, query, opts)
if (rpcHits !== null) return rpcHits
}
const timeoutMs = Math.min(8000, Math.max(600, Number(opts.timeoutMs) || 2800)) const timeoutMs = Math.min(8000, Math.max(600, Number(opts.timeoutMs) || 2800))
const requestId = id() const requestId = id()
const waitPromise = new Promise((resolve) => { const waitPromise = new Promise((resolve) => {
@@ -39,24 +99,36 @@ const platformSearchMeshGossipMixin = {
fn(hits) fn(hits)
}, },
async _onMessageSearchRequestGossip (payload) { async _buildMessageSearchResponsePayload (payload) {
if (!payload?.guildId || !payload?.requestId || !payload?.query) return null if (!payload?.guildId || !payload?.requestId || !payload?.query) return null
if (this.guild?.guild?.id !== payload.guildId) return null if (this.guild?.guild?.id !== payload.guildId) return null
const local = await this.searchGuildMessages(payload.guildId, payload.query, { const local = await this.searchGuildMessages(payload.guildId, payload.query, {
limit: Math.min(30, Number(payload.limit) || 30) limit: Math.min(30, Number(payload.limit) || 30)
}) })
const compact = local.map((m) => searchScope.compactMeshSearchHit(m)) const compact = local.map((m) => searchScope.compactMeshSearchHit(m))
if (this.guild.gossipMessageSearchResponse) { return {
this.guild.gossipMessageSearchResponse({ guildId: payload.guildId,
guildId: payload.guildId, requestId: payload.requestId,
requestId: payload.requestId, query: payload.query,
query: payload.query, hits: compact,
hits: compact, respondedBy: this.identity.user?.id || null,
respondedBy: this.identity.user?.id || null, at: Date.now()
at: Date.now()
})
} }
return compact },
async _handleMessageSearchRequestRpc (payload) {
const res = await this._buildMessageSearchResponsePayload(payload)
if (res) res.viaRpc = true
return res
},
async _onMessageSearchRequestGossip (payload) {
const res = await this._buildMessageSearchResponsePayload(payload)
if (!res) return null
if (this.guild.gossipMessageSearchResponse) {
this.guild.gossipMessageSearchResponse(res)
}
return res.hits
}, },
async _onMessageSearchResponseGossip (payload) { async _onMessageSearchResponseGossip (payload) {