Phase 744: extract guild sync watermark, gossip, bandwidth mixins
Non-breaking refactor: move sync watermark/ack/tail helpers, gossip outbox queue and flush, and bandwidth/member-page slice helpers out of platform-pearcord-platform-class.js. Manifest rows 73→76; runtime registry 13→16. Mesh sync behavior and observability unchanged. Co-authored-by: Cursor <[email protected]>
This commit is contained in:
Raven Scott
@@ -54,6 +54,8 @@ Application facade: one `PearcordPlatform` class that wires identity, database,
|
||||
|
||||
**Phase 723 (v0.8.699):** Dedicated slash registry & invoke audit export JSON — `slash-registry-json-mixin.js`, `slash-invoke-audit-json-mixin.js`, spans `slash-registry.json` / `slash-invoke-audit.json`, `getSlashInvokeAuditExportJsonExport`, `clearSlashInvokeAuditExportJsonExports`, deep links `openSlashRegistryJson` / `openSlashInvokeAuditJson`. Bundle: `npm run test:ci-phase723`.
|
||||
|
||||
**Phase 744 (v0.8.720):** `platform-guild-sync-watermark-mixin.js`, `platform-guild-gossip-outbox-mixin.js`, `platform-guild-sync-bandwidth-mixin.js` (76 mixin rows, 16 runtime mixins); sync watermarks/acks, gossip outbox queue/flush, bandwidth caps and member roster paging. Bundle: `npm run test:ci-phase744`.
|
||||
|
||||
**Phase 743 (v0.8.719):** `platform-mesh-view-mixin.js`, `platform-mesh-logging-mixin.js`, `platform-member-page-burst-mixin.js` (73 mixin rows, 13 runtime mixins); sync fanout watermarks, mesh status summary, mesh span logging, topic pool peer tracking, member page burst limits. Bundle: `npm run test:ci-phase743`.
|
||||
|
||||
**Phase 742 (v0.8.718):** `platform-guild-topic-replication-mixin.js`, `platform-mesh-auto-heal-mixin.js`, `platform-guild-mesh-peer-retry-mixin.js` (70 mixin rows, 10 runtime mixins); topic pool cap, channel replication diag, mesh auto-heal watchdog, peer retry scheduling. Bundle: `npm run test:ci-phase742`.
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
'use strict'
|
||||
|
||||
const swarmScope = require('./platform-swarm-manager-imports')
|
||||
|
||||
const platformGuildGossipOutboxMixin = {
|
||||
async _guildGossipOrQueue (label, sendFn) {
|
||||
if (!this.guild || typeof sendFn !== 'function') return null
|
||||
const peers = this.guild.peers?.size ?? 0
|
||||
if (peers > 0) {
|
||||
await sendFn()
|
||||
return { sent: true }
|
||||
}
|
||||
this._guildGossipOutbox.enqueue({
|
||||
label,
|
||||
send: sendFn,
|
||||
priority: swarmScope.gossipLaneForLabel(label)
|
||||
})
|
||||
const gossipStats = this._guildGossipOutbox.stats?.()
|
||||
if (gossipStats?.lastDrop?.at && Date.now() - gossipStats.lastDrop.at < 50) {
|
||||
this.log.info('guild.sync.sparse-qos', {
|
||||
spanKind: 'guild.sync.sparse-qos',
|
||||
action: 'gossip-lane-drop',
|
||||
label: gossipStats.lastDrop.label,
|
||||
dropCount: gossipStats.dropCount
|
||||
})
|
||||
}
|
||||
this._patchGuildSyncHealth({
|
||||
guildId: this.guild.guild?.id,
|
||||
pendingGossipCount: this._guildGossipOutbox.size,
|
||||
gossipOutboxDropCount: gossipStats?.dropCount || 0
|
||||
})
|
||||
return { queued: true }
|
||||
},
|
||||
|
||||
async _flushGuildGossipOutbox (opts = {}) {
|
||||
const guildId = this.guild?.guild?.id
|
||||
const before = this._guildGossipOutbox.size
|
||||
const out = await this._guildGossipOutbox.flush()
|
||||
this._patchGuildSyncHealth({
|
||||
guildId,
|
||||
pendingGossipCount: this._guildGossipOutbox.size
|
||||
})
|
||||
const flushed = Number(out?.flushed) || 0
|
||||
if (guildId && flushed > 0) {
|
||||
void this._persistGossipOutboxReplayWatermark(guildId).catch(() => {})
|
||||
const laneCounts = this._guildGossipOutbox.laneCounts?.() || null
|
||||
this._logMeshResilience('gossip-outbox-flush', {
|
||||
guildId,
|
||||
flushed,
|
||||
remaining: Number(out?.remaining) || this._guildGossipOutbox.size,
|
||||
reason: opts.reason || null,
|
||||
laneCounts
|
||||
})
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { platformGuildGossipOutboxMixin }
|
||||
@@ -0,0 +1,134 @@
|
||||
'use strict'
|
||||
|
||||
const swarmScope = require('./platform-swarm-manager-imports')
|
||||
const hyperdbScope = require('./platform-hyperdb-sync')
|
||||
|
||||
const platformGuildSyncBandwidthMixin = {
|
||||
_getGuildSyncBandwidthCap () {
|
||||
const cap = Number(process.env.PEARCORD_GUILD_SYNC_BYTES_PER_SEC)
|
||||
return Number.isFinite(cap) && cap > 0 ? Math.floor(cap) : 0
|
||||
},
|
||||
|
||||
_getGuildSyncBandwidthCapEffective () {
|
||||
const cap = this._getGuildSyncBandwidthCap()
|
||||
if (!cap) return 0
|
||||
let effective = cap
|
||||
if (process.env.PEARCORD_GUILD_SYNC_FAIR_SHARE !== '0') {
|
||||
const peers = Math.max(1, this.guild?.peers?.size ?? 1)
|
||||
effective = Math.max(4096, Math.floor(cap / peers))
|
||||
}
|
||||
if (process.env.PEARCORD_GUILD_SYNC_MULTI_GUILD_QOS !== '0') {
|
||||
const guildLoad = Math.max(1, (this.guilds || []).length)
|
||||
if (guildLoad > 2) {
|
||||
effective = Math.max(4096, Math.floor(effective / Math.min(guildLoad, 4)))
|
||||
}
|
||||
}
|
||||
if (process.env.PEARCORD_GUILD_SYNC_RTT_ADAPTIVE !== '0') {
|
||||
const rtt =
|
||||
Number(this._guildSyncBurstRttMs) ||
|
||||
Number(process.env.PEARCORD_GUILD_SYNC_RTT_MS) ||
|
||||
0
|
||||
if (rtt > 80) {
|
||||
const scale = rtt < 200 ? 0.85 : rtt < 400 ? 0.7 : 0.55
|
||||
effective = Math.max(4096, Math.floor(effective * scale))
|
||||
}
|
||||
}
|
||||
if (process.env.PEARCORD_GUILD_SYNC_ROSTER_BANDWIDTH_SCALE !== '0') {
|
||||
const health = this.getGuildSyncHealth()
|
||||
const rosterTotal = Math.max(
|
||||
0,
|
||||
Number(health?.memberTotal) || Number(health?.memberCount) || 0
|
||||
)
|
||||
if (rosterTotal > 100) {
|
||||
const rosterScale =
|
||||
rosterTotal < 200 ? 0.92 : rosterTotal < 500 ? 0.78 : 0.62
|
||||
effective = Math.max(4096, Math.floor(effective * rosterScale))
|
||||
}
|
||||
}
|
||||
return effective
|
||||
},
|
||||
|
||||
_reserveGuildSyncBandwidth (bytes) {
|
||||
const cap = this._getGuildSyncBandwidthCapEffective()
|
||||
if (!cap) return true
|
||||
const n = Math.max(0, Number(bytes) || 0)
|
||||
const now = Date.now()
|
||||
const win = this._guildSyncBandwidthWindow
|
||||
if (!win || now - win.at >= 1000) {
|
||||
this._guildSyncBandwidthWindow = { at: now, used: 0 }
|
||||
}
|
||||
const w = this._guildSyncBandwidthWindow
|
||||
if (w.used + n > cap) return false
|
||||
w.used += n
|
||||
return true
|
||||
},
|
||||
|
||||
_getGuildSyncMemberRosterCap () {
|
||||
return Math.max(8, Number(process.env.PEARCORD_GUILD_SYNC_MEMBER_CAP) || 100)
|
||||
},
|
||||
|
||||
_getGuildSyncMemberPageSize () {
|
||||
return Math.max(8, Number(process.env.PEARCORD_GUILD_SYNC_MEMBER_PAGE_SIZE) || 50)
|
||||
},
|
||||
|
||||
_sliceMembersForGuildSync (allMembers = [], opts = {}) {
|
||||
const cap = this._getGuildSyncMemberRosterCap()
|
||||
const pageSize = this._getGuildSyncMemberPageSize()
|
||||
const offset = Math.max(0, Number(opts.memberPageOffset) || 0)
|
||||
const sinceTs = Number(opts.sinceTimestamp) || 0
|
||||
if (opts.syncDelta && sinceTs > 0) {
|
||||
const churned = allMembers.filter((m) => {
|
||||
const j = Number(m.joinedAt) || 0
|
||||
const u = Number(m.updatedAt) || Number(m.createdAt) || 0
|
||||
return j > sinceTs || u > sinceTs
|
||||
})
|
||||
if (churned.length > 0 && churned.length < allMembers.length) {
|
||||
return {
|
||||
members: churned.map((m) => hyperdbScope.sanitizeMemberForHyperDb(m)),
|
||||
membersPartial: false,
|
||||
memberChurnOnly: true,
|
||||
memberTotal: allMembers.length,
|
||||
memberPageOffset: 0,
|
||||
memberPageSize: pageSize,
|
||||
memberPageNextOffset: null
|
||||
}
|
||||
}
|
||||
}
|
||||
const sorted = [...allMembers].sort(
|
||||
(a, b) => (Number(a.joinedAt) || 0) - (Number(b.joinedAt) || 0)
|
||||
)
|
||||
if (sorted.length <= cap && offset === 0) {
|
||||
return {
|
||||
members: sorted.map((m) => hyperdbScope.sanitizeMemberForHyperDb(m)),
|
||||
membersPartial: false,
|
||||
memberTotal: sorted.length,
|
||||
memberPageOffset: 0,
|
||||
memberPageSize: pageSize,
|
||||
memberPageNextOffset: null
|
||||
}
|
||||
}
|
||||
const page = sorted.slice(offset, offset + pageSize)
|
||||
const next = offset + page.length
|
||||
return {
|
||||
members: page.map((m) => hyperdbScope.sanitizeMemberForHyperDb(m)),
|
||||
membersPartial: true,
|
||||
memberTotal: sorted.length,
|
||||
memberPageOffset: offset,
|
||||
memberPageSize: pageSize,
|
||||
memberPageNextOffset: next < sorted.length ? next : null
|
||||
}
|
||||
},
|
||||
|
||||
_guildSyncBurstDelaysFromRtt () {
|
||||
const envRtt = Number(process.env.PEARCORD_GUILD_SYNC_RTT_MS)
|
||||
const rtt =
|
||||
Number(this._guildSyncBurstRttMs) > 0
|
||||
? Number(this._guildSyncBurstRttMs)
|
||||
: Number.isFinite(envRtt) && envRtt > 0
|
||||
? envRtt
|
||||
: 0
|
||||
return swarmScope.adaptiveBurstDelays(rtt)
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { platformGuildSyncBandwidthMixin }
|
||||
@@ -0,0 +1,142 @@
|
||||
'use strict'
|
||||
|
||||
const sharedScope = require('pearcord-shared')
|
||||
|
||||
const platformGuildSyncWatermarkMixin = {
|
||||
_getGuildSyncLimits () {
|
||||
const perChannel = Number(process.env.PEARCORD_GUILD_SYNC_PER_CHANNEL)
|
||||
const total = Number(process.env.PEARCORD_GUILD_SYNC_MAX_MESSAGES)
|
||||
return {
|
||||
perChannel: Number.isFinite(perChannel) && perChannel > 0 ? perChannel : 50,
|
||||
total: Number.isFinite(total) && total > 0 ? total : 200
|
||||
}
|
||||
},
|
||||
|
||||
_getGuildSyncWatermark (guildId) {
|
||||
const id = guildId || this.guild?.guild?.id
|
||||
if (!id) return { sinceTimestamp: 0, sinceMessageId: null }
|
||||
const row = this._guildSyncHealthByGuild.get(id)
|
||||
return row?.syncWatermark || { sinceTimestamp: 0, sinceMessageId: null }
|
||||
},
|
||||
|
||||
_updateGuildSyncWatermark (guildId, messages = []) {
|
||||
if (!guildId || !messages.length) return null
|
||||
let sinceTimestamp = 0
|
||||
let sinceMessageId = null
|
||||
for (const m of messages) {
|
||||
const ts = Number(m.createdAt) || 0
|
||||
if (ts > sinceTimestamp) {
|
||||
sinceTimestamp = ts
|
||||
sinceMessageId = m.id
|
||||
} else if (ts === sinceTimestamp && sinceMessageId && String(m.id) > String(sinceMessageId)) {
|
||||
sinceMessageId = m.id
|
||||
} else if (ts === sinceTimestamp && !sinceMessageId) {
|
||||
sinceMessageId = m.id
|
||||
}
|
||||
}
|
||||
if (!sinceTimestamp && !sinceMessageId) return null
|
||||
return this._patchGuildSyncHealth({
|
||||
guildId,
|
||||
syncWatermark: { sinceTimestamp, sinceMessageId }
|
||||
})
|
||||
},
|
||||
|
||||
_filterMessagesForDeltaSync (messages, sinceTimestamp, sinceMessageId) {
|
||||
const sinceTs = Number(sinceTimestamp) || 0
|
||||
const sinceId = sinceMessageId || null
|
||||
if (!sinceTs && !sinceId) return messages
|
||||
return messages.filter((m) => {
|
||||
const ts = Number(m.createdAt) || 0
|
||||
if (sinceTs && ts > sinceTs) return true
|
||||
if (sinceTs && ts === sinceTs && sinceId && String(m.id) > String(sinceId)) return true
|
||||
if (!sinceTs && sinceId && String(m.id) > String(sinceId)) return true
|
||||
return false
|
||||
})
|
||||
},
|
||||
|
||||
_ackMeetsWatermark (ack, watermark) {
|
||||
if (!ack || !watermark) return false
|
||||
const ackTs = Number(ack.sinceTimestamp) || 0
|
||||
const wmTs = Number(watermark.sinceTimestamp) || 0
|
||||
if (ackTs > wmTs) return true
|
||||
if (ackTs === wmTs && ack.sinceMessageId && watermark.sinceMessageId) {
|
||||
return String(ack.sinceMessageId) >= String(watermark.sinceMessageId)
|
||||
}
|
||||
return ackTs === wmTs && !!ack.sinceMessageId
|
||||
},
|
||||
|
||||
_recordGuildSyncAck (guildId, payload) {
|
||||
if (!guildId || !payload?.userId) return null
|
||||
if (!this._guildSyncAcks.has(guildId)) this._guildSyncAcks.set(guildId, new Map())
|
||||
const row = {
|
||||
userId: payload.userId,
|
||||
sinceTimestamp: Number(payload.sinceTimestamp) || 0,
|
||||
sinceMessageId: payload.sinceMessageId || null,
|
||||
channelId: payload.channelId || null,
|
||||
at: Date.now()
|
||||
}
|
||||
this._guildSyncAcks.get(guildId).set(payload.userId, row)
|
||||
const cachedTail = this._guildSyncHealthByGuild.get(guildId)?.syncTailWatermark
|
||||
if (cachedTail && this._ackMeetsWatermark(row, cachedTail)) {
|
||||
this._guildSyncPushHalted.add(guildId)
|
||||
}
|
||||
void this._refreshGuildSyncTailWatermark(guildId, row.channelId)
|
||||
.then((tail) => {
|
||||
if (tail && this._ackMeetsWatermark(row, tail)) {
|
||||
this._guildSyncPushHalted.add(guildId)
|
||||
}
|
||||
})
|
||||
.catch(() => {})
|
||||
return row
|
||||
},
|
||||
|
||||
_getGuildSyncTailWatermark (guildId, channelId) {
|
||||
const health = this._guildSyncHealthByGuild.get(guildId)
|
||||
if (health?.syncTailWatermark) return health.syncTailWatermark
|
||||
return this._getGuildSyncWatermark(guildId)
|
||||
},
|
||||
|
||||
async _refreshGuildSyncTailWatermark (guildId, channelId) {
|
||||
const gid = guildId || this.guild?.guild?.id
|
||||
if (!gid) return null
|
||||
const chId = channelId || this.activeChannelId
|
||||
let rows = []
|
||||
if (chId) {
|
||||
rows = await this.db.find(sharedScope.COLLECTIONS.MESSAGES, { channelId: chId })
|
||||
} else {
|
||||
const channels = (await this.guild?.listChannels?.()) || []
|
||||
for (const ch of channels) {
|
||||
if (!['text', 'announcement', 'forum', 'voice', 'stage'].includes(ch.type)) continue
|
||||
const part = await this.db.find(sharedScope.COLLECTIONS.MESSAGES, { channelId: ch.id })
|
||||
rows = rows.concat(part)
|
||||
}
|
||||
}
|
||||
if (!rows.length) return null
|
||||
const sorted = rows.sort((a, b) => (a.createdAt || 0) - (b.createdAt || 0))
|
||||
const tail = sorted[sorted.length - 1]
|
||||
const syncTailWatermark = {
|
||||
sinceTimestamp: Number(tail.createdAt) || 0,
|
||||
sinceMessageId: tail.id
|
||||
}
|
||||
this._patchGuildSyncHealth({ guildId: gid, syncTailWatermark })
|
||||
return syncTailWatermark
|
||||
},
|
||||
|
||||
_isGuildSyncPushHalted (guildId) {
|
||||
return this._guildSyncPushHalted.has(guildId)
|
||||
},
|
||||
|
||||
_computeGuildSyncProgressPct (guildId, healthRow) {
|
||||
const row = healthRow || this._guildSyncHealthByGuild.get(guildId)
|
||||
if (!row) return 0
|
||||
const wm = row.syncWatermark || {}
|
||||
const tail = row.syncTailWatermark || wm
|
||||
const tailTs = Number(tail.sinceTimestamp) || 0
|
||||
const wmTs = Number(wm.sinceTimestamp) || 0
|
||||
if (!tailTs) return row.pending ? 0 : 100
|
||||
if (wmTs >= tailTs) return 100
|
||||
return Math.min(99, Math.round((wmTs / tailTs) * 100))
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { platformGuildSyncWatermarkMixin }
|
||||
@@ -4,7 +4,7 @@
|
||||
* Ordered PearcordPlatform prototype mixin registration (Phase 730).
|
||||
* Preserves assign order from index.js / apply-platform-mixins (Phase 727).
|
||||
*/
|
||||
const PLATFORM_MIXIN_ASSIGNMENT_COUNT = 73
|
||||
const PLATFORM_MIXIN_ASSIGNMENT_COUNT = 76
|
||||
|
||||
const PLATFORM_MIXIN_ASSIGNMENTS = [
|
||||
{ module: './polls-scheduling', export: 'pollSchedulingMixin' },
|
||||
@@ -79,6 +79,9 @@ const PLATFORM_MIXIN_ASSIGNMENTS = [
|
||||
{ module: './platform-mesh-view-mixin', export: 'platformMeshViewMixin' },
|
||||
{ module: './platform-mesh-logging-mixin', export: 'platformMeshLoggingMixin' },
|
||||
{ module: './platform-member-page-burst-mixin', export: 'platformMemberPageBurstMixin' },
|
||||
{ module: './platform-guild-sync-watermark-mixin', export: 'platformGuildSyncWatermarkMixin' },
|
||||
{ module: './platform-guild-gossip-outbox-mixin', export: 'platformGuildGossipOutboxMixin' },
|
||||
{ module: './platform-guild-sync-bandwidth-mixin', export: 'platformGuildSyncBandwidthMixin' },
|
||||
{ module: './platform-diagnostics-mixin', export: 'platformDiagnosticsMixin' }
|
||||
]
|
||||
|
||||
|
||||
@@ -5282,205 +5282,29 @@ class PearcordPlatform extends EventEmitter {
|
||||
return hits
|
||||
}
|
||||
|
||||
_getGuildSyncLimits () {
|
||||
const perChannel = Number(process.env.PEARCORD_GUILD_SYNC_PER_CHANNEL)
|
||||
const total = Number(process.env.PEARCORD_GUILD_SYNC_MAX_MESSAGES)
|
||||
return {
|
||||
perChannel: Number.isFinite(perChannel) && perChannel > 0 ? perChannel : 50,
|
||||
total: Number.isFinite(total) && total > 0 ? total : 200
|
||||
}
|
||||
}
|
||||
|
||||
_getGuildSyncWatermark (guildId) {
|
||||
const id = guildId || this.guild?.guild?.id
|
||||
if (!id) return { sinceTimestamp: 0, sinceMessageId: null }
|
||||
const row = this._guildSyncHealthByGuild.get(id)
|
||||
return row?.syncWatermark || { sinceTimestamp: 0, sinceMessageId: null }
|
||||
}
|
||||
|
||||
_updateGuildSyncWatermark (guildId, messages = []) {
|
||||
if (!guildId || !messages.length) return null
|
||||
let sinceTimestamp = 0
|
||||
let sinceMessageId = null
|
||||
for (const m of messages) {
|
||||
const ts = Number(m.createdAt) || 0
|
||||
if (ts > sinceTimestamp) {
|
||||
sinceTimestamp = ts
|
||||
sinceMessageId = m.id
|
||||
} else if (ts === sinceTimestamp && sinceMessageId && String(m.id) > String(sinceMessageId)) {
|
||||
sinceMessageId = m.id
|
||||
} else if (ts === sinceTimestamp && !sinceMessageId) {
|
||||
sinceMessageId = m.id
|
||||
}
|
||||
}
|
||||
if (!sinceTimestamp && !sinceMessageId) return null
|
||||
return this._patchGuildSyncHealth({
|
||||
guildId,
|
||||
syncWatermark: { sinceTimestamp, sinceMessageId }
|
||||
})
|
||||
}
|
||||
|
||||
_filterMessagesForDeltaSync (messages, sinceTimestamp, sinceMessageId) {
|
||||
const sinceTs = Number(sinceTimestamp) || 0
|
||||
const sinceId = sinceMessageId || null
|
||||
if (!sinceTs && !sinceId) return messages
|
||||
return messages.filter((m) => {
|
||||
const ts = Number(m.createdAt) || 0
|
||||
if (sinceTs && ts > sinceTs) return true
|
||||
if (sinceTs && ts === sinceTs && sinceId && String(m.id) > String(sinceId)) return true
|
||||
if (!sinceTs && sinceId && String(m.id) > String(sinceId)) return true
|
||||
return false
|
||||
})
|
||||
}
|
||||
|
||||
_ackMeetsWatermark (ack, watermark) {
|
||||
if (!ack || !watermark) return false
|
||||
const ackTs = Number(ack.sinceTimestamp) || 0
|
||||
const wmTs = Number(watermark.sinceTimestamp) || 0
|
||||
if (ackTs > wmTs) return true
|
||||
if (ackTs === wmTs && ack.sinceMessageId && watermark.sinceMessageId) {
|
||||
return String(ack.sinceMessageId) >= String(watermark.sinceMessageId)
|
||||
}
|
||||
return ackTs === wmTs && !!ack.sinceMessageId
|
||||
}
|
||||
|
||||
_recordGuildSyncAck (guildId, payload) {
|
||||
if (!guildId || !payload?.userId) return null
|
||||
if (!this._guildSyncAcks.has(guildId)) this._guildSyncAcks.set(guildId, new Map())
|
||||
const row = {
|
||||
userId: payload.userId,
|
||||
sinceTimestamp: Number(payload.sinceTimestamp) || 0,
|
||||
sinceMessageId: payload.sinceMessageId || null,
|
||||
channelId: payload.channelId || null,
|
||||
at: Date.now()
|
||||
}
|
||||
this._guildSyncAcks.get(guildId).set(payload.userId, row)
|
||||
const cachedTail = this._guildSyncHealthByGuild.get(guildId)?.syncTailWatermark
|
||||
if (cachedTail && this._ackMeetsWatermark(row, cachedTail)) {
|
||||
this._guildSyncPushHalted.add(guildId)
|
||||
}
|
||||
void this._refreshGuildSyncTailWatermark(guildId, row.channelId)
|
||||
.then((tail) => {
|
||||
if (tail && this._ackMeetsWatermark(row, tail)) {
|
||||
this._guildSyncPushHalted.add(guildId)
|
||||
}
|
||||
})
|
||||
.catch(() => {})
|
||||
return row
|
||||
}
|
||||
|
||||
_getGuildSyncTailWatermark (guildId, channelId) {
|
||||
const health = this._guildSyncHealthByGuild.get(guildId)
|
||||
if (health?.syncTailWatermark) return health.syncTailWatermark
|
||||
return this._getGuildSyncWatermark(guildId)
|
||||
}
|
||||
|
||||
async _refreshGuildSyncTailWatermark (guildId, channelId) {
|
||||
const gid = guildId || this.guild?.guild?.id
|
||||
if (!gid) return null
|
||||
const chId = channelId || this.activeChannelId
|
||||
let rows = []
|
||||
if (chId) {
|
||||
rows = await this.db.find(sharedScope.COLLECTIONS.MESSAGES, { channelId: chId })
|
||||
} else {
|
||||
const channels = (await this.guild?.listChannels?.()) || []
|
||||
for (const ch of channels) {
|
||||
if (!['text', 'announcement', 'forum', 'voice', 'stage'].includes(ch.type)) continue
|
||||
const part = await this.db.find(sharedScope.COLLECTIONS.MESSAGES, { channelId: ch.id })
|
||||
rows = rows.concat(part)
|
||||
}
|
||||
}
|
||||
if (!rows.length) return null
|
||||
const sorted = rows.sort((a, b) => (a.createdAt || 0) - (b.createdAt || 0))
|
||||
const tail = sorted[sorted.length - 1]
|
||||
const syncTailWatermark = {
|
||||
sinceTimestamp: Number(tail.createdAt) || 0,
|
||||
sinceMessageId: tail.id
|
||||
}
|
||||
this._patchGuildSyncHealth({ guildId: gid, syncTailWatermark })
|
||||
return syncTailWatermark
|
||||
}
|
||||
|
||||
_isGuildSyncPushHalted (guildId) {
|
||||
return this._guildSyncPushHalted.has(guildId)
|
||||
}
|
||||
|
||||
_computeGuildSyncProgressPct (guildId, healthRow) {
|
||||
const row = healthRow || this._guildSyncHealthByGuild.get(guildId)
|
||||
if (!row) return 0
|
||||
const wm = row.syncWatermark || {}
|
||||
const tail = row.syncTailWatermark || wm
|
||||
const tailTs = Number(tail.sinceTimestamp) || 0
|
||||
const wmTs = Number(wm.sinceTimestamp) || 0
|
||||
if (!tailTs) return row.pending ? 0 : 100
|
||||
if (wmTs >= tailTs) return 100
|
||||
return Math.min(99, Math.round((wmTs / tailTs) * 100))
|
||||
}
|
||||
|
||||
async _guildGossipOrQueue (label, sendFn) {
|
||||
if (!this.guild || typeof sendFn !== 'function') return null
|
||||
const peers = this.guild.peers?.size ?? 0
|
||||
if (peers > 0) {
|
||||
await sendFn()
|
||||
return { sent: true }
|
||||
}
|
||||
this._guildGossipOutbox.enqueue({
|
||||
label,
|
||||
send: sendFn,
|
||||
priority: swarmScope.gossipLaneForLabel(label)
|
||||
})
|
||||
const gossipStats = this._guildGossipOutbox.stats?.()
|
||||
if (gossipStats?.lastDrop?.at && Date.now() - gossipStats.lastDrop.at < 50) {
|
||||
this.log.info('guild.sync.sparse-qos', {
|
||||
spanKind: 'guild.sync.sparse-qos',
|
||||
action: 'gossip-lane-drop',
|
||||
label: gossipStats.lastDrop.label,
|
||||
dropCount: gossipStats.dropCount
|
||||
})
|
||||
}
|
||||
this._patchGuildSyncHealth({
|
||||
guildId: this.guild.guild?.id,
|
||||
pendingGossipCount: this._guildGossipOutbox.size,
|
||||
gossipOutboxDropCount: gossipStats?.dropCount || 0
|
||||
})
|
||||
return { queued: true }
|
||||
}
|
||||
|
||||
async _flushGuildGossipOutbox (opts = {}) {
|
||||
const guildId = this.guild?.guild?.id
|
||||
const before = this._guildGossipOutbox.size
|
||||
const out = await this._guildGossipOutbox.flush()
|
||||
this._patchGuildSyncHealth({
|
||||
guildId,
|
||||
pendingGossipCount: this._guildGossipOutbox.size
|
||||
})
|
||||
const flushed = Number(out?.flushed) || 0
|
||||
if (guildId && flushed > 0) {
|
||||
void this._persistGossipOutboxReplayWatermark(guildId).catch(() => {})
|
||||
const laneCounts = this._guildGossipOutbox.laneCounts?.() || null
|
||||
this._logMeshResilience('gossip-outbox-flush', {
|
||||
guildId,
|
||||
flushed,
|
||||
remaining: Number(out?.remaining) || this._guildGossipOutbox.size,
|
||||
reason: opts.reason || null,
|
||||
laneCounts
|
||||
})
|
||||
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
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
async _canModeratorPushGuildSync () {
|
||||
const guildId = this.guild?.guild?.id
|
||||
@@ -5579,131 +5403,19 @@ class PearcordPlatform extends EventEmitter {
|
||||
}))
|
||||
}
|
||||
|
||||
_getGuildSyncBandwidthCap () {
|
||||
const cap = Number(process.env.PEARCORD_GUILD_SYNC_BYTES_PER_SEC)
|
||||
return Number.isFinite(cap) && cap > 0 ? Math.floor(cap) : 0
|
||||
}
|
||||
|
||||
_getGuildSyncBandwidthCapEffective () {
|
||||
const cap = this._getGuildSyncBandwidthCap()
|
||||
if (!cap) return 0
|
||||
let effective = cap
|
||||
if (process.env.PEARCORD_GUILD_SYNC_FAIR_SHARE !== '0') {
|
||||
const peers = Math.max(1, this.guild?.peers?.size ?? 1)
|
||||
effective = Math.max(4096, Math.floor(cap / peers))
|
||||
}
|
||||
if (process.env.PEARCORD_GUILD_SYNC_MULTI_GUILD_QOS !== '0') {
|
||||
const guildLoad = Math.max(1, (this.guilds || []).length)
|
||||
if (guildLoad > 2) {
|
||||
effective = Math.max(4096, Math.floor(effective / Math.min(guildLoad, 4)))
|
||||
}
|
||||
}
|
||||
if (process.env.PEARCORD_GUILD_SYNC_RTT_ADAPTIVE !== '0') {
|
||||
const rtt =
|
||||
Number(this._guildSyncBurstRttMs) ||
|
||||
Number(process.env.PEARCORD_GUILD_SYNC_RTT_MS) ||
|
||||
0
|
||||
if (rtt > 80) {
|
||||
const scale = rtt < 200 ? 0.85 : rtt < 400 ? 0.7 : 0.55
|
||||
effective = Math.max(4096, Math.floor(effective * scale))
|
||||
}
|
||||
}
|
||||
if (process.env.PEARCORD_GUILD_SYNC_ROSTER_BANDWIDTH_SCALE !== '0') {
|
||||
const health = this.getGuildSyncHealth()
|
||||
const rosterTotal = Math.max(
|
||||
0,
|
||||
Number(health?.memberTotal) || Number(health?.memberCount) || 0
|
||||
)
|
||||
if (rosterTotal > 100) {
|
||||
const rosterScale =
|
||||
rosterTotal < 200 ? 0.92 : rosterTotal < 500 ? 0.78 : 0.62
|
||||
effective = Math.max(4096, Math.floor(effective * rosterScale))
|
||||
}
|
||||
}
|
||||
return effective
|
||||
}
|
||||
|
||||
_reserveGuildSyncBandwidth (bytes) {
|
||||
const cap = this._getGuildSyncBandwidthCapEffective()
|
||||
if (!cap) return true
|
||||
const n = Math.max(0, Number(bytes) || 0)
|
||||
const now = Date.now()
|
||||
const win = this._guildSyncBandwidthWindow
|
||||
if (!win || now - win.at >= 1000) {
|
||||
this._guildSyncBandwidthWindow = { at: now, used: 0 }
|
||||
}
|
||||
const w = this._guildSyncBandwidthWindow
|
||||
if (w.used + n > cap) return false
|
||||
w.used += n
|
||||
return true
|
||||
}
|
||||
|
||||
_getGuildSyncMemberRosterCap () {
|
||||
return Math.max(8, Number(process.env.PEARCORD_GUILD_SYNC_MEMBER_CAP) || 100)
|
||||
}
|
||||
|
||||
_getGuildSyncMemberPageSize () {
|
||||
return Math.max(8, Number(process.env.PEARCORD_GUILD_SYNC_MEMBER_PAGE_SIZE) || 50)
|
||||
}
|
||||
|
||||
_sliceMembersForGuildSync (allMembers = [], opts = {}) {
|
||||
const cap = this._getGuildSyncMemberRosterCap()
|
||||
const pageSize = this._getGuildSyncMemberPageSize()
|
||||
const offset = Math.max(0, Number(opts.memberPageOffset) || 0)
|
||||
const sinceTs = Number(opts.sinceTimestamp) || 0
|
||||
if (opts.syncDelta && sinceTs > 0) {
|
||||
const churned = allMembers.filter((m) => {
|
||||
const j = Number(m.joinedAt) || 0
|
||||
const u = Number(m.updatedAt) || Number(m.createdAt) || 0
|
||||
return j > sinceTs || u > sinceTs
|
||||
})
|
||||
if (churned.length > 0 && churned.length < allMembers.length) {
|
||||
return {
|
||||
members: churned.map((m) => hyperdbScope.sanitizeMemberForHyperDb(m)),
|
||||
membersPartial: false,
|
||||
memberChurnOnly: true,
|
||||
memberTotal: allMembers.length,
|
||||
memberPageOffset: 0,
|
||||
memberPageSize: pageSize,
|
||||
memberPageNextOffset: null
|
||||
}
|
||||
}
|
||||
}
|
||||
const sorted = [...allMembers].sort(
|
||||
(a, b) => (Number(a.joinedAt) || 0) - (Number(b.joinedAt) || 0)
|
||||
)
|
||||
if (sorted.length <= cap && offset === 0) {
|
||||
return {
|
||||
members: sorted.map((m) => hyperdbScope.sanitizeMemberForHyperDb(m)),
|
||||
membersPartial: false,
|
||||
memberTotal: sorted.length,
|
||||
memberPageOffset: 0,
|
||||
memberPageSize: pageSize,
|
||||
memberPageNextOffset: null
|
||||
}
|
||||
}
|
||||
const page = sorted.slice(offset, offset + pageSize)
|
||||
const next = offset + page.length
|
||||
return {
|
||||
members: page.map((m) => hyperdbScope.sanitizeMemberForHyperDb(m)),
|
||||
membersPartial: true,
|
||||
memberTotal: sorted.length,
|
||||
memberPageOffset: offset,
|
||||
memberPageSize: pageSize,
|
||||
memberPageNextOffset: next < sorted.length ? next : null
|
||||
}
|
||||
}
|
||||
|
||||
_guildSyncBurstDelaysFromRtt () {
|
||||
const envRtt = Number(process.env.PEARCORD_GUILD_SYNC_RTT_MS)
|
||||
const rtt =
|
||||
Number(this._guildSyncBurstRttMs) > 0
|
||||
? Number(this._guildSyncBurstRttMs)
|
||||
: Number.isFinite(envRtt) && envRtt > 0
|
||||
? envRtt
|
||||
: 0
|
||||
return swarmScope.adaptiveBurstDelays(rtt)
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
async _exportForumIndexForSyncBundle (guildId, channels = [], sinceTs = 0) {
|
||||
if (!guildId) return []
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
'use strict'
|
||||
|
||||
/** Prototype mixins extracted from PearcordPlatform class body (Phases 738–743). */
|
||||
/** Prototype mixins extracted from PearcordPlatform class body (Phases 738–744). */
|
||||
const PLATFORM_RUNTIME_MIXIN_MODULES = [
|
||||
'./platform-hyperswarm-runtime-mixin',
|
||||
'./platform-session-startup-mixin',
|
||||
@@ -14,7 +14,10 @@ const PLATFORM_RUNTIME_MIXIN_MODULES = [
|
||||
'./platform-guild-mesh-peer-retry-mixin',
|
||||
'./platform-mesh-view-mixin',
|
||||
'./platform-mesh-logging-mixin',
|
||||
'./platform-member-page-burst-mixin'
|
||||
'./platform-member-page-burst-mixin',
|
||||
'./platform-guild-sync-watermark-mixin',
|
||||
'./platform-guild-gossip-outbox-mixin',
|
||||
'./platform-guild-sync-bandwidth-mixin'
|
||||
]
|
||||
|
||||
const PLATFORM_RUNTIME_MIXIN_COUNT = PLATFORM_RUNTIME_MIXIN_MODULES.length
|
||||
|
||||
Reference in New Issue
Block a user