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

refactor(platform): Phase 882 gossip RPC pull config + mesh mixin (v0.8.849)

Browse Source 
Non-breaking: centralize guild gossip RPC pull timeout env defaults in
gossip-rpc-pull-config.js; extract guild sync/member page RPC pull into
platform-mesh-rpc-pull-mixin.js spread by platform-mesh-view-mixin. Default
gossip v1 unchanged; fleet v2 cutover deferred.

Tests: test:ci-phase882 (boot gate, unit, view snapshot, method maps, live audit RPC).
Docs: PLATFORM_ROADMAP, PROTOMUX_RPC_MIGRATION, AUTOMATED_TESTING, release notes.
This commit is contained in:
Raven Scott
2026-06-04 18:19:47 -04:00
parent 2f423d0a7c
commit 2758c73d6b
8 changed files with 186 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+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 882 (v0.8.849):** Shared `gossip-rpc-pull-config.js` for pull timeout env defaults; `platform-mesh-rpc-pull-mixin.js` for guild sync/member page RPC. Bundle: `npm run test:ci-phase882`.
**Phase 881 (v0.8.848):** Split `gossip-rpc-pull.js` into peer/session/validators/orchestrators/view-snapshot modules; barrel export unchanged. Bundle: `npm run test:ci-phase881`.
**Phase 880 (v0.8.847):** Host-prefer RPC pull (`fetchGuildGossipRpcFromHostHit`) for guild sync/member page; search merge via `fetchGuildGossipRpcPullMerged`. Bundle: `npm run test:ci-phase880`.
gossip-rpc-pull-config.js
+52
View File
@@ -0,0 +1,52 @@
'use strict'
const {
isValidAuditExportArchiveRpcResponse,
isValidAutomationDigestSnapshotRpcResponse,
isValidGuildSyncBundleRpcResponse,
isValidMemberPageRpcResponse,
mergeGuildMessageSearchRpcHits
} = require('./gossip-rpc-validators')
/** Shared timeout env + defaults for guild gossip RPC pull paths (Phase 882). */
const gossipRpcPullConfigs = {
auditExportArchive: {
rpcMsEnv: 'PEARCORD_AUDIT_EXPORT_ARCHIVE_RPC_MS',
wireMsEnv: 'PEARCORD_AUDIT_EXPORT_ARCHIVE_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 12000, wireMs: 8000 },
pickPredicate: isValidAuditExportArchiveRpcResponse
},
automationDigestSnapshot: {
rpcMsEnv: 'PEARCORD_AUTOMATION_DIGEST_SNAPSHOT_RPC_MS',
wireMsEnv: 'PEARCORD_AUTOMATION_DIGEST_SNAPSHOT_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 12000, wireMs: 8000 },
pickPredicate: isValidAutomationDigestSnapshotRpcResponse
},
guildSyncHost: {
rpcMsEnv: 'PEARCORD_GUILD_SYNC_RPC_MS',
wireMsEnv: 'PEARCORD_GUILD_SYNC_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 30000, wireMs: 12000 },
validateResponse: isValidGuildSyncBundleRpcResponse
},
memberPageHost: {
rpcMsEnv: 'PEARCORD_GUILD_MEMBER_PAGE_RPC_MS',
wireMsEnv: 'PEARCORD_GUILD_MEMBER_PAGE_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 25000, wireMs: 12000 },
validateResponse: isValidMemberPageRpcResponse
},
messageSearchMerge: {
strategy: 'all',
rpcMsEnv: 'PEARCORD_MESSAGE_SEARCH_RPC_MS',
wireMsEnv: 'PEARCORD_MESSAGE_SEARCH_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 8000, wireMs: 6000 },
mergeResponses: mergeGuildMessageSearchRpcHits
}
}
function gossipRpcPullConfig (name) {
const cfg = gossipRpcPullConfigs[name]
if (!cfg) return null
return { ...cfg }
}
module.exports = { gossipRpcPullConfigs, gossipRpcPullConfig }
gossip-rpc-pull.js
+3 -1
View File
@@ -6,11 +6,13 @@ const session = require('./gossip-rpc-session')
const validators = require('./gossip-rpc-validators')
const orchestrators = require('./gossip-rpc-orchestrators')
const viewSnapshot = require('./gossip-rpc-view-snapshot')
const pullConfig = require('./gossip-rpc-pull-config')
module.exports = {
...peer,
...session,
...validators,
...orchestrators,
...viewSnapshot
...viewSnapshot,
...pullConfig
}
mixins/runtime/features/platform-audit-archive-rpc-mixin.js
+2 -5
View File
@@ -6,7 +6,7 @@ const {
matchesGuildRpcTargetMember,
markRpcResponse,
fetchGuildGossipRpcPullHit,
isValidAuditExportArchiveRpcResponse
gossipRpcPullConfig
} = require('../../../gossip-rpc-pull')
const platformAuditArchiveRpcMixin = {
@@ -37,12 +37,9 @@ const platformAuditArchiveRpcMixin = {
async _tryFetchAuditExportArchiveFromMeshRpc (archiveId, opts = {}) {
return fetchGuildGossipRpcPullHit(this, {
rpcMsEnv: 'PEARCORD_AUDIT_EXPORT_ARCHIVE_RPC_MS',
wireMsEnv: 'PEARCORD_AUDIT_EXPORT_ARCHIVE_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 12000, wireMs: 8000 },
...gossipRpcPullConfig('auditExportArchive'),
targetMemberId: opts.targetMemberId || null,
rpcMethod: sharedScope.RPC.AUDIT_EXPORT_ARCHIVE_REQUEST,
pickPredicate: isValidAuditExportArchiveRpcResponse,
buildPayload: async ({ guildId }) => ({
guildId,
archiveId,
mixins/runtime/features/platform-automation-digest-snapshot-rpc-mixin.js
+2 -5
View File
@@ -6,7 +6,7 @@ const {
matchesGuildRpcTargetMember,
markRpcResponse,
fetchGuildGossipRpcPullHit,
isValidAutomationDigestSnapshotRpcResponse
gossipRpcPullConfig
} = require('../../../gossip-rpc-pull')
const platformAutomationDigestSnapshotRpcMixin = {
@@ -39,12 +39,9 @@ const platformAutomationDigestSnapshotRpcMixin = {
async _tryFetchAutomationDigestSnapshotFromMeshRpc (snapshotId, opts = {}) {
return fetchGuildGossipRpcPullHit(this, {
rpcMsEnv: 'PEARCORD_AUTOMATION_DIGEST_SNAPSHOT_RPC_MS',
wireMsEnv: 'PEARCORD_AUTOMATION_DIGEST_SNAPSHOT_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 12000, wireMs: 8000 },
...gossipRpcPullConfig('automationDigestSnapshot'),
targetMemberId: opts.targetMemberId || null,
rpcMethod: sharedScope.RPC.AUTOMATION_DIGEST_EXPORT_SNAPSHOT_REQUEST,
pickPredicate: isValidAutomationDigestSnapshotRpcResponse,
buildPayload: async ({ guildId }) => ({
guildId,
snapshotId,
mixins/runtime/mesh/platform-mesh-rpc-pull-mixin.js
+120
View File
@@ -0,0 +1,120 @@
'use strict'
const sharedScope = require('../../../platform-pearcord-shared-imports')
const {
fetchGuildGossipRpcFromHostHit,
resolveGuildGossipRpcHostPeerId,
markRpcResponse,
gossipRpcPullConfig
} = require('../../../gossip-rpc-pull')
/** Guild sync + member page protomux-rpc pull helpers (Phase 882 extract). */
const platformMeshRpcPullMixin = {
async _tryMemberPageRequestRpc (offset = 0) {
const guildId = this.guild?.guild?.id
const userId = this.identity?.user?.id
if (!guildId || !userId) return null
const off = Math.max(0, Number(offset) || 0)
const page = await fetchGuildGossipRpcFromHostHit(this, {
...gossipRpcPullConfig('memberPageHost'),
rpcMethod: sharedScope.RPC.MEMBER_PAGE_REQUEST,
buildPayload: async ({ guildId: gid }) => ({
guildId: gid,
requesterId: userId,
offset: off
})
})
if (!page) return null
const added = await this._ingestMemberPage(page)
return markRpcResponse({
requested: true,
offset: page.offset,
memberCount: page.members?.length || 0,
added
})
},
async _tryGuildSyncRequestRpc (channelId) {
const guildId = this.guild?.guild?.id
const userId = this.identity?.user?.id
if (!guildId || !userId) return null
const wm = this._getGuildSyncWatermark(guildId)
const ch = channelId || this.activeChannelId || null
const hostPublicKey = await this._lookupGuildHostPublicKey(this.guild.guild)
const bundle = await fetchGuildGossipRpcFromHostHit(this, {
...gossipRpcPullConfig('guildSyncHost'),
hostPublicKey,
rpcMethod: sharedScope.RPC.GUILD_SYNC_REQUEST,
beforeRequest: async (platform) =>
platform.guild.guild.ownerId === platform.identity?.user?.id,
buildPayload: async ({ guildId: gid }) => ({
guildId: gid,
userId,
channelId: ch,
sinceTimestamp: wm.sinceTimestamp || 0,
sinceMessageId: wm.sinceMessageId || null
})
})
if (!bundle) return null
await this._ingestGuildSync(bundle)
const hostPk = resolveGuildGossipRpcHostPeerId(this.guild, hostPublicKey)
return markRpcResponse({
requested: true,
peerCount: 1,
hostPublicKey: hostPk
})
},
_fanoutGuildSyncRequestBroadcast (channelId) {
const guildId = this.guild?.guild?.id
const userId = this.identity?.user?.id
if (!guildId || !userId || !this.guild?.gossipGuildSyncRequest) {
return { requested: false, peerCount: 0 }
}
const wm = this._getGuildSyncWatermark(guildId)
const ch = channelId || this.activeChannelId || null
const peerIds = [...(this.guild.peers?.keys() || [])]
void this._flushGuildGossipOutbox().catch(() => {})
void this._guildGossipOrQueue('guild-sync-request', () =>
this.guild.gossipGuildSyncRequest({
guildId,
userId,
channelId: ch,
sinceTimestamp: wm.sinceTimestamp || 0,
sinceMessageId: wm.sinceMessageId || null,
fanoutPeerCount: peerIds.length
})
).catch(() => {})
const ackMap = this._guildSyncAcks.get(guildId)
return {
requested: true,
peerCount: peerIds.length,
watermark: wm,
ackCount: ackMap?.size ?? 0,
acks: ackMap
? [...ackMap.entries()].map(([uid, row]) => ({ userId: uid, ...row }))
: []
}
},
_fanoutGuildSyncRequestWithAckWatermarks (channelId) {
if (sharedScope.isRpcGossipRequestEnabled?.()) {
void this._tryGuildSyncRequestRpc(channelId)
.then((rpc) => {
if (!rpc?.viaRpc) this._fanoutGuildSyncRequestBroadcast(channelId)
})
.catch(() => {
this._fanoutGuildSyncRequestBroadcast(channelId)
})
const peerIds = [...(this.guild.peers?.keys() || [])]
return {
requested: true,
peerCount: peerIds.length,
pendingRpc: true
}
}
return this._fanoutGuildSyncRequestBroadcast(channelId)
}
}
module.exports = { platformMeshRpcPullMixin }
mixins/runtime/mesh/platform-mesh-view-mixin.js
+2 -119
View File
@@ -1,127 +1,10 @@
'use strict'
const localScope = require('../../../platform-index-local-imports')
const sharedScope = require('../../../platform-pearcord-shared-imports')
const {
fetchGuildGossipRpcFromHostHit,
resolveGuildGossipRpcHostPeerId,
markRpcResponse,
isValidGuildSyncBundleRpcResponse,
isValidMemberPageRpcResponse
} = require('../../../gossip-rpc-pull')
const { platformMeshRpcPullMixin } = require('./platform-mesh-rpc-pull-mixin')
const platformMeshViewMixin = {
async _tryMemberPageRequestRpc (offset = 0) {
const guildId = this.guild?.guild?.id
const userId = this.identity?.user?.id
if (!guildId || !userId) return null
const off = Math.max(0, Number(offset) || 0)
const page = await fetchGuildGossipRpcFromHostHit(this, {
rpcMsEnv: 'PEARCORD_GUILD_MEMBER_PAGE_RPC_MS',
wireMsEnv: 'PEARCORD_GUILD_MEMBER_PAGE_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 25000, wireMs: 12000 },
rpcMethod: sharedScope.RPC.MEMBER_PAGE_REQUEST,
validateResponse: isValidMemberPageRpcResponse,
buildPayload: async ({ guildId: gid }) => ({
guildId: gid,
requesterId: userId,
offset: off
})
})
if (!page) return null
const added = await this._ingestMemberPage(page)
return markRpcResponse({
requested: true,
offset: page.offset,
memberCount: page.members?.length || 0,
added
})
},
async _tryGuildSyncRequestRpc (channelId) {
const guildId = this.guild?.guild?.id
const userId = this.identity?.user?.id
if (!guildId || !userId) return null
const wm = this._getGuildSyncWatermark(guildId)
const ch = channelId || this.activeChannelId || null
const hostPublicKey = await this._lookupGuildHostPublicKey(this.guild.guild)
const bundle = await fetchGuildGossipRpcFromHostHit(this, {
hostPublicKey,
rpcMsEnv: 'PEARCORD_GUILD_SYNC_RPC_MS',
wireMsEnv: 'PEARCORD_GUILD_SYNC_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 30000, wireMs: 12000 },
rpcMethod: sharedScope.RPC.GUILD_SYNC_REQUEST,
validateResponse: isValidGuildSyncBundleRpcResponse,
beforeRequest: async (platform) =>
platform.guild.guild.ownerId === platform.identity?.user?.id,
buildPayload: async ({ guildId: gid }) => ({
guildId: gid,
userId,
channelId: ch,
sinceTimestamp: wm.sinceTimestamp || 0,
sinceMessageId: wm.sinceMessageId || null
})
})
if (!bundle) return null
await this._ingestGuildSync(bundle)
const hostPk = resolveGuildGossipRpcHostPeerId(this.guild, hostPublicKey)
return markRpcResponse({
requested: true,
peerCount: 1,
hostPublicKey: hostPk
})
},
_fanoutGuildSyncRequestBroadcast (channelId) {
const guildId = this.guild?.guild?.id
const userId = this.identity?.user?.id
if (!guildId || !userId || !this.guild?.gossipGuildSyncRequest) {
return { requested: false, peerCount: 0 }
}
const wm = this._getGuildSyncWatermark(guildId)
const ch = channelId || this.activeChannelId || null
const peerIds = [...(this.guild.peers?.keys() || [])]
void this._flushGuildGossipOutbox().catch(() => {})
void this._guildGossipOrQueue('guild-sync-request', () =>
this.guild.gossipGuildSyncRequest({
guildId,
userId,
channelId: ch,
sinceTimestamp: wm.sinceTimestamp || 0,
sinceMessageId: wm.sinceMessageId || null,
fanoutPeerCount: peerIds.length
})
).catch(() => {})
const ackMap = this._guildSyncAcks.get(guildId)
return {
requested: true,
peerCount: peerIds.length,
watermark: wm,
ackCount: ackMap?.size ?? 0,
acks: ackMap
? [...ackMap.entries()].map(([uid, row]) => ({ userId: uid, ...row }))
: []
}
},
_fanoutGuildSyncRequestWithAckWatermarks (channelId) {
if (sharedScope.isRpcGossipRequestEnabled?.()) {
void this._tryGuildSyncRequestRpc(channelId)
.then((rpc) => {
if (!rpc?.viaRpc) this._fanoutGuildSyncRequestBroadcast(channelId)
})
.catch(() => {
this._fanoutGuildSyncRequestBroadcast(channelId)
})
const peerIds = [...(this.guild.peers?.keys() || [])]
return {
requested: true,
peerCount: peerIds.length,
pendingRpc: true
}
}
return this._fanoutGuildSyncRequestBroadcast(channelId)
},
...platformMeshRpcPullMixin,
_meshPeerQualityForView (guildId) {
const gid = guildId || this.guild?.guild?.id
mixins/runtime/search-discovery/platform-search-mesh-gossip-mixin.js
+3 -7
View File
@@ -6,7 +6,7 @@ const { id } = require('../../../platform-class-imports')
const {
markRpcResponse,
fetchGuildGossipRpcPullMerged,
mergeGuildMessageSearchRpcHits
gossipRpcPullConfig
} = require('../../../gossip-rpc-pull')
const platformSearchMeshGossipMixin = {
@@ -14,10 +14,7 @@ const platformSearchMeshGossipMixin = {
if (!this.guild?.guild || this.guild.guild.id !== guildId) return null
const requestId = id()
return fetchGuildGossipRpcPullMerged(this, {
strategy: 'all',
rpcMsEnv: 'PEARCORD_MESSAGE_SEARCH_RPC_MS',
wireMsEnv: 'PEARCORD_MESSAGE_SEARCH_RPC_WIRE_READY_MS',
timeoutDefaults: { rpcMs: 8000, wireMs: 6000 },
...gossipRpcPullConfig('messageSearchMerge'),
rpcMethod: sharedScope.RPC.MESSAGE_SEARCH_REQUEST,
buildPayload: async ({ guildId: gid }) => ({
guildId: gid,
@@ -26,8 +23,7 @@ const platformSearchMeshGossipMixin = {
limit: opts.limit || 50,
requestedBy: this.identity.user?.id || null,
at: Date.now()
}),
mergeResponses: mergeGuildMessageSearchRpcHits
})
})
},