fix(mesh): coalesce guild mesh joins across open/retry paths

Route guild.open, mesh-bounded bootstrap, create, and peer-retry joins through _requestGuildMeshJoin so startup/switch races skip stale loadGen attempts and share one in-flight join per guild instance. Co-authored-by: Cursor <[email protected]>
2026-06-02 11:19:28 -04:00
parent 95ba2a2272
commit eb9122b7ad
2 changed files with 151 additions and 57 deletions
@@ -12,6 +12,7 @@ Application facade: one `PearcordPlatform` class that wires identity, database,
 **Phase 410 hardening (v0.8.643):** `view()` now supports explicit detail levels (`minimal`/`light`/`full`/`diagnostic`) through `opts.detail`, keeps legacy `opts.light` compatibility, and surfaces the resolved mode as `viewDetailLevel` in payload diagnostics.
 **Phase 410 hardening (v0.8.643):** `guildSyncHealthRail` now carries bounded-retry metadata (`retryAttempt`, `retryDelayMs`, `retryMaxAttempts`, `nextRetryAt`, `retryCountdownMs`) so UI can show live “mesh bounded mode” retry countdown/tooltip state during background join retries.
 **Phase 410 hardening (v0.8.643):** Guild mesh peer retry scheduling now uses `_computeGuildMeshPeerRetryDelayMs` (last-peer-contact staleness + RTT bounds) instead of fixed attempt-index spacing; health rows track `lastPeerSeenAt` on peer join.
+**Phase 410 hardening (v0.8.643):** `_requestGuildMeshJoin` coalesces duplicate guild mesh joins during `guild.open`, peer retry, and startup/switch races; stale `loadGen` joins are skipped; `pearcord-guild#joinMesh` deduplicates in-flight `swarm.join`. Smoke: `npm run test:platform-guild-mesh-join-coalesce`.

 **Phase 659 (v0.8.638):** DM scheduled messages slice — `sendMessage` parses `/schedule` and `/sendlater` in DM mode; queue APIs `createDmScheduledMessage`, `listDmScheduledMessages`, `cancelDmScheduledMessage`, `sendNowDmScheduledMessage`, plus due runner `runDueDmScheduledMessages()` and local timer flush. See [DM_SCHEDULED_MESSAGES.md](../../docs/DM_SCHEDULED_MESSAGES.md).

@@ -461,6 +461,7 @@ class PearcordPlatform extends EventEmitter {
    this._guildLoading = false
    this._guildLoadingGuildId = null
    this._guildLoadGeneration = 0
+    this._guildMeshJoinCoalesce = null
    this._guildDeleteInProgress = null
    this._guildOpenNoViewableChannels = false
    this._emojiAttachmentMissUntil = new Map()
@@ -1145,6 +1146,7 @@ class PearcordPlatform extends EventEmitter {
        ])
        if (guildLoadTimedOut && this._guildLoading) {
          this._guildLoadGeneration++
+          this._abortGuildMeshJoinCoalesce('session-startup-guild-timeout')
          this._guildLoading = false
          this._guildLoadingGuildId = null
          this.emit('guild-loading')
@@ -1615,7 +1617,136 @@ class PearcordPlatform extends EventEmitter {
    this._invalidateDiscoveryListingsCache()
  }

+  _abortGuildMeshJoinCoalesce (reason = '') {
+    const c = this._guildMeshJoinCoalesce
+    if (!c) return
+    this._guildMeshJoinCoalesce = null
+    this.log.debug('guild mesh join coalesce cleared', {
+      reason: reason || 'unknown',
+      guildId: c.guildId || null
+    })
+  }
+
+  async _requestGuildMeshJoin (opts = {}) {
+    const source = opts.source || 'unknown'
+    const force = opts.force === true
+    const loadGen = opts.loadGen ?? null
+    const bounded = opts.bounded !== false
+    const g = this.guild
+    if (!g?.guild) {
+      return { peers: 0, bounded: false, skipped: true, reason: 'no-guild' }
+    }
+    if (loadGen != null && !this._isGuildLoadCurrent(loadGen)) {
+      this.log.debug('guild mesh join skipped (stale load)', {
+        source,
+        loadGen,
+        guildId: g.guild.id
+      })
+      return {
+        peers: g.peers?.size ?? 0,
+        bounded: this._isGuildMeshBounded(g.guild.id),
+        skipped: true,
+        reason: 'stale-load'
+      }
+    }
+    const guildId = g.guild.id
+    const topic = g.guild.topic
+    const coalesce = this._guildMeshJoinCoalesce
+    if (
+      !force &&
+      coalesce &&
+      coalesce.guildId === guildId &&
+      coalesce.topic === topic &&
+      coalesce.guild === g &&
+      coalesce.promise
+    ) {
+      this.log.debug('guild mesh join coalesced', {
+        source,
+        guildId,
+        generation: loadGen
+      })
+      return coalesce.promise
+    }
+    const promise = this._executeGuildMeshJoin({
+      source,
+      force,
+      loadGen,
+      bounded,
+      guildId
+    })
+    this._guildMeshJoinCoalesce = { guildId, topic, guild: g, loadGen, promise }
+    return promise
+  }
+
+  async _executeGuildMeshJoin ({ source, force, loadGen, bounded, guildId }) {
+    const meshFlushMs = Number(
+      process.env.PEARCORD_GUILD_SWARM_FLUSH_MS ??
+        process.env.PEARCORD_SWARM_FLUSH_MS ??
+        2500
+    )
+    const joinMeshCapMs = meshFlushMs + 5000
+    try {
+      const g = this.guild
+      if (!g?.guild || g.guild.id !== guildId) {
+        return { peers: 0, bounded: false, skipped: true, reason: 'guild-gone' }
+      }
+      this._setGuildMeshBounded(guildId, false)
+      this.log.debug('guild mesh join start', { source, guildId, generation: loadGen })
+      const joinPromise = g.joinMesh(force ? { force: true } : {})
+      if (bounded) {
+        await Promise.race([
+          joinPromise,
+          new Promise((resolve) => {
+            setTimeout(() => {
+              this._setGuildMeshBounded(guildId, true)
+              this.log.info('guild joinMesh bounded; continuing in background', {
+                flushMs: meshFlushMs,
+                capMs: joinMeshCapMs,
+                guildId,
+                source
+              })
+              resolve()
+            }, joinMeshCapMs)
+          })
+        ])
+        await boundedFlush(() => g.swarm?.flush?.(), meshFlushMs).catch(() => {})
+      } else {
+        await joinPromise
+      }
+      if (loadGen != null && !this._isGuildLoadCurrent(loadGen)) {
+        this.log.debug('guild mesh join aborted after join (stale load)', {
+          source,
+          guildId,
+          generation: loadGen
+        })
+        return {
+          peers: g.peers?.size ?? 0,
+          bounded: this._isGuildMeshBounded(guildId),
+          skipped: true,
+          reason: 'stale-load-after'
+        }
+      }
+      if (this.guild !== g || this.guild?.guild?.id !== guildId) {
+        return { peers: 0, bounded: false, skipped: true, reason: 'guild-replaced' }
+      }
+      const peers = g.peers?.size ?? 0
+      return { peers, bounded: this._isGuildMeshBounded(guildId) }
+    } catch (err) {
+      this.log.error('guild.mesh error', {
+        guildId,
+        source,
+        err: err?.message || String(err)
+      })
+      throw err
+    } finally {
+      if (this._guildMeshJoinCoalesce?.guildId === guildId) {
+        this._guildMeshJoinCoalesce = null
+      }
+    }
+  }
+
  async _leaveMeshes () {
+    this._abortGuildMeshJoinCoalesce('leave-meshes')
    this._stopAuditExportScheduleTimer()
    this._stopDmScheduledMessageTimer()
    await this._leaveVoice({ gossip: true })
@@ -7313,40 +7444,14 @@ class PearcordPlatform extends EventEmitter {

  async _joinGuildMeshBounded () {
    if (!this.guild?.guild) return { peers: 0, bounded: false }
-    const meshFlushMs = Number(
-      process.env.PEARCORD_GUILD_SWARM_FLUSH_MS ??
-        process.env.PEARCORD_SWARM_FLUSH_MS ??
-        2500
-    )
-    const joinMeshCapMs = meshFlushMs + 5000
    const gid = this.guild.guild.id
-    this._setGuildMeshBounded(gid, false)
-    try {
-      await Promise.race([
-        this.guild.joinMesh(),
-        new Promise((resolve) => {
-          setTimeout(() => {
-            this._setGuildMeshBounded(gid, true)
-            this.log.info('guild joinMesh bounded; continuing in background', {
-              flushMs: meshFlushMs,
-              capMs: joinMeshCapMs,
-              guildId: gid
-            })
-            resolve()
-          }, joinMeshCapMs)
-        })
-      ])
-      await boundedFlush(() => this.guild.swarm?.flush?.(), meshFlushMs).catch(() => {})
-    } catch (err) {
-      this.log.error('guild.mesh error', {
-        guildId: gid,
-        err: err?.message || String(err)
-      })
-      throw err
-    }
-    const peers = this.guild.peers?.size ?? 0
-    if (peers === 0) this._scheduleGuildMeshPeerRetry(gid)
-    return { peers, bounded: this._isGuildMeshBounded(gid) }
+    const mesh = await this._requestGuildMeshJoin({
+      source: 'mesh-bounded',
+      bounded: true
+    })
+    const peers = mesh.peers ?? this.guild.peers?.size ?? 0
+    if (!mesh.skipped && peers === 0) this._scheduleGuildMeshPeerRetry(gid)
+    return { peers, bounded: mesh.bounded ?? this._isGuildMeshBounded(gid) }
  }

  async _waitForGuildSyncAfterJoin (opts = {}) {
@@ -12759,27 +12864,12 @@ class PearcordPlatform extends EventEmitter {
    this._setGuildMeshBounded(guildRecord.id, false)
    this._logMultiGuildMesh('guild-open', { guildId: guildRecord.id })
    const meshSpan = this.log.time('guild.open.joinMesh')
-    const meshFlushMs = Number(
-      process.env.PEARCORD_GUILD_SWARM_FLUSH_MS ??
-        process.env.PEARCORD_SWARM_FLUSH_MS ??
-        2500
-    )
-    const joinMeshCapMs = meshFlushMs + 5000
    try {
-      await Promise.race([
-        this.guild.joinMesh(),
-        new Promise((resolve) => {
-          setTimeout(() => {
-            this._setGuildMeshBounded(guildRecord.id, true)
-            this.log.info('guild joinMesh bounded; continuing in background', {
-              flushMs: meshFlushMs,
-              capMs: joinMeshCapMs,
-              guildId: guildRecord.id
-            })
-            resolve()
-          }, joinMeshCapMs)
-        })
-      ])
+      await this._requestGuildMeshJoin({
+        source: 'guild-open',
+        loadGen,
+        bounded: true
+      })
    } catch (err) {
      this.log.error('guild.mesh error', {
        guildId: guildRecord.id,
@@ -12903,7 +12993,7 @@ class PearcordPlatform extends EventEmitter {
    await this.guild.ready()
    const result = await this.guild.create({ name })
    this._wireGuild(this.guild)
-    await this.guild.joinMesh()
+    await this._requestGuildMeshJoin({ source: 'guild-create', bounded: true })
    await this._initGuildVoice(result.guild.id)
    this.messages = new PearcordMessage({ authorId: user.id, db: this.db })
    await this.messages.ready()
@@ -13056,9 +13146,11 @@ class PearcordPlatform extends EventEmitter {
        sinceLastPeerContactMs: lastContact ? Date.now() - lastContact : null,
        spacingSource: 'last-peer-contact'
      })
-      void this.guild.joinMesh().catch((err) => {
-        this.log.debug('guild mesh peer retry failed', { guildId, err })
-      })
+      void this._requestGuildMeshJoin({ source: 'peer-retry', bounded: false }).catch(
+        (err) => {
+          this.log.debug('guild mesh peer retry failed', { guildId, err })
+        }
+      )
      // Targeted resilience improvement (Phase 654 prep): non-owners trigger discovery mesh refresh on retry to aid peer discovery for replication/sync
      const userId2 = this.identity?.user?.id
      const isOwner2 = userId2 && this.guild.guild.ownerId === userId2
@@ -13149,6 +13241,7 @@ class PearcordPlatform extends EventEmitter {
      throw new Error('guild not found')
    }
    const gen = ++this._guildLoadGeneration
+    this._abortGuildMeshJoinCoalesce('guild-switch')
    this._guildLoading = true
    this._guildLoadingGuildId = id
    this._logMultiGuildMesh('guild-switch', { guildId: id, generation: gen })