fix: calm mesh join thrash and gossip outbox delivery

Serialize/retry outbox flushes, longer wire probes for messages, quiet view-path multi-guild diagnostics, light-poll skip of heavy export blocks, and soft peer-replace handling without full join storms.
2026-07-13 03:01:56 -04:00
parent f9cbc5bf5f
commit de62098365
6 changed files with 173 additions and 73 deletions
@@ -1,6 +1,7 @@
 'use strict'

-const MESH_PEER_DISPLAY_ZERO_DROP_HOLD_MS = 2500
+/** Hold brief 0-peer drops so header mesh status does not flicker during dual-connection swaps. */
+const MESH_PEER_DISPLAY_ZERO_DROP_HOLD_MS = 4500

 function createMeshPeerDisplayStabilizer (initialPeers = 0) {
  const peers = Math.max(0, Number(initialPeers) || 0)
@@ -6360,7 +6360,7 @@ async _onDiscoveryListingExportJsonGossip (payload) {
 async _federationSyncJsonRegistryEntries () {
  const gid = this.guild?.guild?.id
  if (!gid) return []
-  const diag = await this.exportFederationSyncDiagnostics().catch(() => null)
+  const diag = await this.exportFederationSyncDiagnostics({ quiet: true, fromView: true }).catch(() => null)
  const agg = diag?.federationAggregate || this._federationSyncAggregateForView()
  const entries = [{
    federationId: gid,
@@ -6627,7 +6627,7 @@ async _guildWidgetJsonRegistryEntries () {
 },

 async _multiGuildSyncJsonRegistryEntries () {
-  const multi = this.exportMultiGuildSyncDiagnostics()
+  const multi = this.exportMultiGuildSyncDiagnostics({ quiet: true, fromView: true })
  const entries = (multi.guildIds || []).slice(0, 32).map((gid) => {
    const g = multi.guilds?.[gid] || {}
    return {
@@ -6662,7 +6662,7 @@ async _partitionHealJsonRegistryEntries () {
    meshBounded: !!active.meshBounded,
    exportedAt: active.exportedAt || Date.now()
  })
-  const multi = this.exportMultiGuildSyncDiagnostics()
+  const multi = this.exportMultiGuildSyncDiagnostics({ quiet: true, fromView: true })
  for (const id of (multi.guildIds || []).slice(0, 16)) {
    if (id === gid) continue
    const ph = multi.guilds?.[id]?.partitionHeal
@@ -6,9 +6,11 @@ const { wireReady } = require('pearcord-drive/mux-wire')
 const swarmScope = require('../../../platform-swarm-manager-imports')
 const sharedScope = require('pearcord-shared')

-const GOSSIP_SEND_WIRE_PROBE_MS = 250
+const GOSSIP_SEND_WIRE_PROBE_MS = 800
 /** Longer probe after peer join so first handshake is not false-queued. */
 const GOSSIP_JOIN_WIRE_PROBE_MS = 2500
+/** Message/presence labels get a longer wire probe so live peers are not missed. */
+const GOSSIP_IMPORTANT_WIRE_PROBE_MS = 2000

 const platformGuildGossipOutboxMixin = {
  async _ensureGuildGossipWireReady (timeoutMs) {
@@ -55,6 +57,22 @@ const platformGuildGossipOutboxMixin = {
    return { queued: true }
  },

+  _wireProbeMsForLabel (label, opts = {}) {
+    if (Number(opts.wireTimeoutMs) > 0) return Number(opts.wireTimeoutMs)
+    const l = String(label || '')
+    if (
+      l === 'message' ||
+      l === 'bot-message' ||
+      l === 'member-join' ||
+      l === 'presence' ||
+      l === 'reaction' ||
+      l === 'pin'
+    ) {
+      return GOSSIP_IMPORTANT_WIRE_PROBE_MS
+    }
+    return GOSSIP_SEND_WIRE_PROBE_MS
+  },
+
  /**
   * @param {string} label
   * @param {() => any} sendFn
@@ -64,17 +82,14 @@ const platformGuildGossipOutboxMixin = {
    if (!this.guild || typeof sendFn !== 'function') return null
    const peers = this.guild.peers?.size ?? 0
    if (peers > 0) {
-      const probeMs =
-        Number(opts.wireTimeoutMs) > 0
-          ? Number(opts.wireTimeoutMs)
-          : GOSSIP_SEND_WIRE_PROBE_MS
+      const probeMs = this._wireProbeMsForLabel(label, opts)
      const wireOk = await this._ensureGuildGossipWireReady(probeMs).catch(() => false)
      if (wireOk) {
        try {
          await sendFn()
          return { sent: true }
        } catch {
-          // peer map stale or wire dropped mid-send — queue for peer-join flush
+          // peer map stale or no delivery — queue for peer-join flush
        }
      }
    }
@@ -82,13 +97,29 @@ const platformGuildGossipOutboxMixin = {
  },

  async _flushGuildGossipOutbox (opts = {}) {
+    // Serialize flushes so concurrent peer-join handshakes do not race the queue.
+    if (this._guildGossipFlushInFlight) {
+      this._guildGossipFlushQueued = {
+        ...(this._guildGossipFlushQueued || {}),
+        ...opts,
+        reason: opts.reason || this._guildGossipFlushQueued?.reason || null
+      }
+      return this._guildGossipFlushInFlight
+    }
+    const run = async () => {
      const guildId = this.guild?.guild?.id
      const before = this._guildGossipOutbox.size
      // Join/recover: wait for wire so flush does not no-op while sessions open.
-    if (opts.reason === 'peer-join' || opts.reason === 'mesh-handshake') {
+      if (
+        opts.reason === 'peer-join' ||
+        opts.reason === 'mesh-handshake' ||
+        opts.reason === 'peer-join-retry'
+      ) {
        await this._ensureGuildGossipWireReady(
          Number(opts.wireTimeoutMs) || GOSSIP_JOIN_WIRE_PROBE_MS
        ).catch(() => false)
+      } else if ((this.guild?.peers?.size ?? 0) > 0 && before > 0) {
+        await this._ensureGuildGossipWireReady(GOSSIP_SEND_WIRE_PROBE_MS).catch(() => false)
      }
      const out = await this._guildGossipOutbox.flush()
      this._patchGuildSyncHealth({
@@ -119,8 +150,32 @@ const platformGuildGossipOutboxMixin = {
          remaining: before,
          reason: 'peer-join-empty'
        }))
+        // Retry once after wire settles — common on first dual-connection swap.
+        if ((this.guild?.peers?.size ?? 0) > 0 && !opts._retried) {
+          setTimeout(() => {
+            if (this.guild?.guild?.id !== guildId) return
+            if ((this._guildGossipOutbox?.size ?? 0) === 0) return
+            void this._flushGuildGossipOutbox({
+              reason: 'peer-join-retry',
+              wireTimeoutMs: GOSSIP_JOIN_WIRE_PROBE_MS,
+              _retried: true
+            }).catch(() => {})
+          }, 700)
+        }
      }
      return out
+    }
+    this._guildGossipFlushInFlight = run()
+      .catch((err) => ({ flushed: 0, error: err?.message || String(err) }))
+      .finally(() => {
+        this._guildGossipFlushInFlight = null
+        const queued = this._guildGossipFlushQueued
+        this._guildGossipFlushQueued = null
+        if (queued && (this._guildGossipOutbox?.size ?? 0) > 0) {
+          void this._flushGuildGossipOutbox(queued).catch(() => {})
+        }
+      })
+    return this._guildGossipFlushInFlight
  },

  async _gossipMemberJoinForUser (guildId, userId, opts = {}) {
@@ -474,6 +474,25 @@ _wireGuild (guildInstance) {
      }
      return
    }
+    // Connection swap for an already-known peer — rewire without full join storms.
+    if (type === 'replace') {
+      const gid = this.guild?.guild?.id
+      if (gid && peerId) {
+        this._touchGuildMeshPeerSeen(gid, peerId, 'peer-replace')
+        this._patchGuildSyncHealth({
+          guildId: gid,
+          peers: this.guild?.peers?.size ?? 0,
+          lastPeerSeenAt: Date.now(),
+          lastSuccessfulPeerContactAt: Date.now()
+        })
+      }
+      if (typeof this.guild?.reconcileMeshConnectionsFromSwarm === 'function') {
+        this.guild.reconcileMeshConnectionsFromSwarm()
+      }
+      void this._ensureGuildGossipWireReady?.(1500).catch(() => false)
+      void this._flushGuildGossipOutbox?.({ reason: 'peer-replace' }).catch(() => {})
+      return
+    }
    if (type === 'join') {
      const gid = this.guild?.guild?.id
      const hadZeroPeer = !!(gid && this._meshZeroPeerSinceByGuild.has(gid))
@@ -1309,7 +1309,7 @@ exportPartitionHealDiagnostics (guildId) {
  }
 },

-exportMultiGuildSyncDiagnostics () {
+exportMultiGuildSyncDiagnostics (opts = {}) {
  const guildIds = [
    ...new Set([
      ...(this.guilds || []).map((g) => g.id),
@@ -1317,12 +1317,21 @@ exportMultiGuildSyncDiagnostics () {
    ])
  ]
  const activeGuildId = this.guild?.guild?.id || null
-  const span = this.log.time('guild.mesh.multi-scale', {
+  const quiet = opts.quiet === true || opts.fromView === true
+  // Rate-limit INFO spam: view builders used to call this every state push.
+  const now = Date.now()
+  const minLogMs = Number(process.env.PEARCORD_MULTI_GUILD_DIAG_LOG_MS) || 15000
+  const shouldLog =
+    !quiet &&
+    (!this._lastMultiGuildDiagLogAt || now - this._lastMultiGuildDiagLogAt >= minLogMs)
+  const span = shouldLog
+    ? this.log.time('guild.mesh.multi-scale', {
      spanKind: 'guild.mesh.multi-scale',
      action: 'export-diagnostics',
      guildCount: guildIds.length,
      activeGuildId
    })
+    : null
  const guilds = {}
  for (const id of guildIds) {
    guilds[id] = buildMultiGuildPerGuildDiagnosticsEntry({
@@ -1339,22 +1348,33 @@ exportMultiGuildSyncDiagnostics () {
    guilds,
    activeGuildMeshBounded: activeGuildId ? this._isGuildMeshBounded(activeGuildId) : false
  })
+  if (span) {
    span.end({ guildCount: guildIds.length, activeGuildId })
+    this._lastMultiGuildDiagLogAt = now
    this._logMultiGuildMesh('export-diagnostics', {
      guildCount: guildIds.length,
      activeGuildId
    })
+  }
  return out
 },

 /** Re-announce public listing TTL for one guild (guild focus — does not global-stall mesh). */

-async exportFederationSyncDiagnostics () {
-  const span = this.log.time('discovery.mesh.federation', {
+async exportFederationSyncDiagnostics (opts = {}) {
+  const quiet = opts.quiet === true || opts.fromView === true
+  const now = Date.now()
+  const minLogMs = Number(process.env.PEARCORD_FEDERATION_DIAG_LOG_MS) || 15000
+  const shouldLog =
+    !quiet &&
+    (!this._lastFederationDiagLogAt || now - this._lastFederationDiagLogAt >= minLogMs)
+  const span = shouldLog
+    ? this.log.time('discovery.mesh.federation', {
      spanKind: 'discovery.mesh.federation',
      action: 'export-diagnostics'
    })
-  const multi = this.exportMultiGuildSyncDiagnostics()
+    : null
+  const multi = this.exportMultiGuildSyncDiagnostics({ quiet: true, fromView: quiet })
  const federationAggregate = this._federationSyncAggregateForView()
  const listingRefreshByGuild = Object.fromEntries(
    this._discoveryListingRefreshByGuild.entries()
@@ -1391,6 +1411,8 @@ async exportFederationSyncDiagnostics () {
    announcementHub,
    discoveryListingExpiry: this._discoveryListingExpiryForView()
  }
+  if (span) {
+    this._lastFederationDiagLogAt = now
    span.end({
      guildCount: federationAggregate.guildCount,
      discoveryPeers: federationAggregate.discoveryPeers,
@@ -1400,6 +1422,7 @@ async exportFederationSyncDiagnostics () {
      guildCount: federationAggregate.guildCount,
      syncingGuilds: federationAggregate.syncingGuilds
    })
+  }
  return out
 },

@@ -1600,7 +1600,9 @@ async view (opts = {}) {
  let auditArchiveFetchPeers = null
  let workerReleaseHint = null
  let workerInstallPlan = null
-  if (guild) {
+  // Heavy JSON-export / multi-guild diagnostics — full views only.
+  // Light polls used to re-export every 1–2s, flooding logs and thrashing the UI.
+  if (guild && !light) {
    hookFailureDigest = await this.getHookFailureDigest()
    auditExportSchedule = await this.getAuditExportSchedule()
    digestExportSchedule = await this.getDigestExportSchedule()