Phase 654 batch 2: extend partition heal with sparse ACK resume, hyperbee rebuild, and sync fanout.

Adds heal hooks for member roles, forum/search, settings device sync, and extended voice/emoji/automod slices; wires retryWire on host mesh join and adaptive guild.create mesh caps.

Co-authored-by: Cursor <[email protected]>

index.js

@@ -567,6 +567,7 @@ class PearcordPlatform extends EventEmitter {
     this._meshAutoHealCheckTimer = null
     this._gossipOutboxReplayByGuild = new Map()
     this._lastGuildSyncChannelCoreKeys = []
+    this._forumIndexByGuild = new Map()
     this._guildOpenDurationSamples = []
     this._guildOpenFallbackThrottledUntil = 0
   }
@@ -1344,6 +1345,198 @@ class PearcordPlatform extends EventEmitter {
     }
   }
 
+  async _resumeSparseAckCursorsAfterHeal (guildId) {
+    const gid = guildId || this.guild?.guild?.id
+    if (!gid) return { resumed: 0, sparsePulled: 0 }
+    const cursors = { ...(this._sparseAckCursorByGuild.get(gid) || {}) }
+    const channels = (await this.guild?.listChannels?.().catch(() => [])) || []
+    let resumed = 0
+    for (const ch of channels) {
+      if (!['text', 'announcement', 'thread', 'forum'].includes(ch.type)) continue
+      const rows = await this.db.find(COLLECTIONS.MESSAGES, { channelId: ch.id }).catch(() => [])
+      const localTail = rows.length > 0 ? rows.length - 1 : 0
+      const prev = Number(cursors[ch.id]) || 0
+      const next = Math.max(prev, localTail)
+      if (next !== prev) resumed++
+      cursors[ch.id] = next
+    }
+    this._sparseAckCursorByGuild.set(gid, cursors)
+    let sparsePulled = 0
+    const rep = await this._getGuildReplicator(gid).catch(() => null)
+    if (rep) {
+      const keys = rep.exportChannelCoreKeys(Object.keys(cursors))
+      if (keys.length) {
+        const opened = await this._eagerOpenAnnouncedCores(gid, keys).catch(() => 0)
+        sparsePulled = Number(opened) || 0
+      }
+    }
+    const channelsList = channels
+    const threadSlice = this._exportThreadArchiveSlice(channelsList, 0)
+    const threadApplied = await this._ingestThreadArchiveSlice(gid, threadSlice).catch(() => 0)
+    const forumRows = await this._exportForumIndexForSyncBundle(gid, channelsList, 0)
+    const forumApplied = await this._ingestForumIndexBundle(gid, forumRows).catch(() => 0)
+    this._patchGuildSyncHealth({
+      guildId: gid,
+      sparseAckCursorCount: Object.keys(cursors).length
+    })
+    return { resumed, sparsePulled, threadApplied, forumApplied }
+  }
+
+  async _rebuildHyperbeeIndexesOnHeal (guildId) {
+    const gid = guildId || this.guild?.guild?.id
+    if (!gid || !GuildReplicator.isAvailable()) {
+      return { channelCount: 0, migrated: 0 }
+    }
+    const userId = this.identity?.user?.id
+    const isOwner = userId && this.guild?.guild?.ownerId === userId
+    if (isOwner) return { channelCount: 0, migrated: 0, skipped: 'owner' }
+    const cursors = this._sparseAckCursorByGuild.get(gid) || {}
+    const channelIds = Object.keys(cursors).slice(0, 24)
+    if (!channelIds.length) return { channelCount: 0, migrated: 0 }
+    const rep = await this._getGuildReplicator(gid).catch(() => null)
+    if (!rep?.rebuildIndexesForChannels) return { channelCount: 0, migrated: 0 }
+    const batch = await rep
+      .rebuildIndexesForChannels(channelIds, { dryRun: false })
+      .catch(() => ({ results: [] }))
+    const migrated = (batch.results || []).filter((r) => (Number(r.migrated) || 0) > 0).length
+    return { channelCount: channelIds.length, migrated, batch }
+  }
+
+  async _healMemberRolesLww (guildId) {
+    const gid = guildId || this.guild?.guild?.id
+    if (!gid || !this.guildRoles) return { links: 0 }
+    const roles = await this.guildRoles.listRoles(gid).catch(() => [])
+    const members = await this.guild?.listMembers?.().catch(() => [])
+    let links = 0
+    for (const m of members || []) {
+      const memberId = m.userId || m.id
+      if (!memberId) continue
+      const row = await this.db.get(COLLECTIONS.MEMBERS, { guildId: gid, userId: memberId })
+      if (!row) continue
+      const at = Number(row.updatedAt) || Number(row.joinedAt) || 0
+      const custom = Array.isArray(row.customRoleIds) ? row.customRoleIds : []
+      for (const roleId of custom) {
+        if (!roles.some((r) => r.id === roleId)) continue
+        links++
+      }
+      if (at > 0) {
+        await this.db
+          .insert(COLLECTIONS.MEMBERS, { ...row, updatedAt: at })
+          .catch(() => null)
+      }
+    }
+    if (this._memberPageBurstRemaining() > 0) {
+      this._recordMemberPageBurst()
+    }
+    return { links, roleCount: roles.length }
+  }
+
+  async _healForumSearchIndexOnPartition (guildId) {
+    const gid = guildId || this.guild?.guild?.id
+    if (!gid) return { forumRows: 0, searchUpdated: false }
+    const channels = (await this.guild?.listChannels?.().catch(() => [])) || []
+    const forumRows = await this._exportForumIndexForSyncBundle(gid, channels, 0)
+    const forumApplied = await this._ingestForumIndexBundle(gid, forumRows)
+    let searchUpdated = false
+    if (typeof this.rebuildGuildSearchIndex === 'function') {
+      await this.rebuildGuildSearchIndex(gid).catch(() => null)
+      searchUpdated = true
+    }
+    return { forumRows: forumRows.length, forumApplied, searchUpdated }
+  }
+
+  async _healSettingsMeshDeviceSync (guildId) {
+    const gid = guildId || this.guild?.guild?.id
+    if (!gid) return null
+    await this.syncSettingsMesh({ guildId: gid }).catch(() => null)
+    const prefs = await this.getPrefs().catch(() => null)
+    const lastChannelId =
+      prefs?.lastChannelId || prefs?.lastGuildChannelId || this.activeChannelId || null
+    const lastGuildId = prefs?.lastGuildId || gid
+    if (lastChannelId && this.notifications) {
+      const { RPC } = require('pearcord-shared')
+      const { broadcastSettingsGossip } = require('pearcord-notifications/mesh')
+      await broadcastSettingsGossip(this.notifications, RPC.SETTINGS_UPDATE, {
+        userId: this.identity?.user?.id,
+        deviceId: this.identity?.deviceId,
+        updatedAt: Date.now(),
+        prefs: {
+          lastGuildId,
+          lastChannelId,
+          partitionHealAt: Date.now()
+        }
+      }).catch(() => null)
+    }
+    return { lastChannelId, lastGuildId, settingsMesh: true }
+  }
+
+  async _healPartitionExtendedSlices (guildId) {
+    const gid = guildId || this.guild?.guild?.id
+    if (!gid) return {}
+    const channels = (await this.guild?.listChannels?.().catch(() => [])) || []
+    const voiceRows = await this._exportVoiceOccupancyForSync(gid)
+    const voiceApplied = await this._ingestVoiceOccupancyBundle(gid, voiceRows).catch(() => 0)
+    let emojiSlots = 0
+    if (this.emojiRegistry && this.guild?.gossipEmojiUpsert) {
+      const emojis = (await this.emojiRegistry.list().catch(() => [])).filter(
+        (r) => !r.guildId || r.guildId === gid
+      )
+      for (const row of emojis.slice(0, 64)) {
+        this.guild.gossipEmojiUpsert(row)
+        emojiSlots++
+      }
+    }
+    let automodReconciled = false
+    if (this.automod) {
+      await this.automod.getConfig().catch(() => null)
+      automodReconciled = true
+    }
+    const store = await this._getGuildSidecar(gid).catch(() => null)
+    if (store?.patchAuditRecent) {
+      const recent = await this._exportAuditSyncSlice(gid, 16, 0)
+      if (recent.length) await store.patchAuditRecent(recent).catch(() => null)
+    }
+    return { voiceApplied, emojiSlots, automodReconciled }
+  }
+
+  _fanoutGuildSyncRequestWithAckWatermarks (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(() => {})
+    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 }))
+        : []
+    }
+  }
+
+  requestGuildSyncFanout () {
+    const out = this._fanoutGuildSyncRequestWithAckWatermarks(this.activeChannelId)
+    this._scheduleGuildSyncRequestBurst()
+    return out
+  }
+
   _meshPeerQualityForView (guildId) {
     const gid = guildId || this.guild?.guild?.id
     if (!gid) return []
@@ -2618,6 +2811,7 @@ class PearcordPlatform extends EventEmitter {
     const force = opts.force === true
     const loadGen = opts.loadGen ?? null
     const bounded = opts.bounded !== false
+    const joinMeshCapMs = opts.joinMeshCapMs ?? null
     const fallbackThrottled = opts.fallbackThrottled === true
     const g = this.guild
     if (!g?.guild) {
@@ -2660,7 +2854,8 @@ class PearcordPlatform extends EventEmitter {
       loadGen,
       bounded,
       guildId,
-      fallbackThrottled
+      fallbackThrottled,
+      joinMeshCapMs
     })
     this._guildMeshJoinCoalesce = { guildId, topic, guild: g, loadGen, promise }
     return promise
@@ -2672,14 +2867,18 @@ class PearcordPlatform extends EventEmitter {
     loadGen,
     bounded,
     guildId,
-    fallbackThrottled = false
+    fallbackThrottled = false,
+    joinMeshCapMs: joinMeshCapMsOpt = null
   }) {
     const meshFlushMs = Number(
       process.env.PEARCORD_GUILD_SWARM_FLUSH_MS ??
         process.env.PEARCORD_SWARM_FLUSH_MS ??
         2500
     )
-    const joinMeshCapMs = fallbackThrottled
+    const joinMeshCapMs =
+      Number(joinMeshCapMsOpt) > 0
+        ? Number(joinMeshCapMsOpt)
+        : fallbackThrottled
           ? Math.max(
               500,
               Number(process.env.PEARCORD_GUILD_OPEN_THROTTLE_MESH_CAP_MS) || 1500
@@ -5385,7 +5584,18 @@ class PearcordPlatform extends EventEmitter {
     const flush = await this._flushGuildGossipOutbox().catch(() => ({
       flushed: 0
     }))
-    this._requestGuildSyncFromHost(this.activeChannelId)
+    const userId = this.identity?.user?.id
+    const isOwner = userId && this.guild.guild.ownerId === userId
+    if (!isOwner) {
+      const pk = await this._lookupGuildHostPublicKey(this.guild.guild).catch(() => null)
+      if (pk) {
+        await this._wireGuildMeshToHost(
+          { creatorPublicKey: pk, hostPublicKey: pk },
+          { hostPublicKey: pk }
+        ).catch(() => null)
+      }
+    }
+    this._fanoutGuildSyncRequestWithAckWatermarks(this.activeChannelId)
     this._scheduleGuildSyncRequestBurst([800, 2200, 5000])
     span.end({
       flushed: flush?.flushed || 0,
@@ -6524,7 +6734,23 @@ class PearcordPlatform extends EventEmitter {
 
   async _ingestForumIndexBundle (guildId, rows = []) {
     if (!guildId || !rows?.length) return 0
-    return rows.length
+    let map = this._forumIndexByGuild.get(guildId)
+    if (!map) {
+      map = new Map()
+      this._forumIndexByGuild.set(guildId, map)
+    }
+    let applied = 0
+    for (const row of rows) {
+      if (!row?.channelId) continue
+      const key = `${row.parentId || ''}:${row.channelId}`
+      const at = Number(row.createdAt) || 0
+      const prev = map.get(key)
+      if (!prev || at >= (Number(prev.createdAt) || 0)) {
+        map.set(key, { ...row, guildId })
+        applied++
+      }
+    }
+    return applied
   }
 
   async _ingestVoiceOccupancyBundle (guildId, rows = []) {
@@ -8340,33 +8566,22 @@ class PearcordPlatform extends EventEmitter {
     const userId = this.identity?.user?.id
     if (!guildId || !userId || !this.guild?.gossipGuildSyncRequest) return
     if (this.guild.guild.ownerId === userId) return
-    void this._flushGuildGossipOutbox().catch(() => {})
     const debKey = `${guildId}:${userId}`
     if (this._guildSyncRequestDebounce.has(debKey)) return
     this._guildSyncRequestDebounce.set(debKey, Date.now())
     setTimeout(() => this._guildSyncRequestDebounce.delete(debKey), 4000)
-    const ch = channelId || this.activeChannelId || null
+    this._fanoutGuildSyncRequestWithAckWatermarks(channelId)
-    const wm = this._getGuildSyncWatermark(guildId)
-    this.guild
-      .gossipGuildSyncRequest({
-        guildId,
-        userId,
-        channelId: ch,
-        sinceTimestamp: wm.sinceTimestamp || 0,
-        sinceMessageId: wm.sinceMessageId || null
-      })
-      .catch(() => {})
   }
 
   requestGuildSyncFromHost () {
     const guildId = this.guild?.guild?.id
     if (!guildId) throw new Error('no guild')
-    this._requestGuildSyncFromHost(this.activeChannelId)
+    const fanout = this.requestGuildSyncFanout()
-    this._scheduleGuildSyncRequestBurst()
     return {
       guildId,
       channelId: this.activeChannelId || null,
-      requested: true
+      requested: !!fanout?.requested,
+      ...fanout
     }
   }
 
@@ -8476,11 +8691,12 @@ class PearcordPlatform extends EventEmitter {
       }
     }
     this._sparseAckCursorByGuild.set(guildId, cursors)
+    const resume = await this._resumeSparseAckCursorsAfterHeal(guildId)
     this._patchGuildSyncHealth({
       guildId,
-      sparseAckCursorCount: Object.keys(cursors).length
+      sparseAckCursorCount: Object.keys(this._sparseAckCursorByGuild.get(guildId) || {}).length
     })
-    return { pruned, remaining: Object.keys(cursors).length }
+    return { pruned, remaining: Object.keys(cursors).length, ...resume }
   }
 
   async _healSidecarExportCursor (guildId) {
@@ -8545,7 +8761,10 @@ class PearcordPlatform extends EventEmitter {
       const presenceMerged = await this._healPresenceVectorPartition(guildId)
       const readReceiptsHealed = await this._healReadReceiptVectors(guildId)
       const sparseAckReconciled = await this._reconcileSparseAckCursorsAfterHeal(guildId)
+      const hyperbeeRebuilt = await this._rebuildHyperbeeIndexesOnHeal(guildId)
       const channelMetadataHealed = await this._healPartitionChannelMetadata(guildId)
+      const forumSearchHealed = await this._healForumSearchIndexOnPartition(guildId)
+      const memberRolesHealed = await this._healMemberRolesLww(guildId)
       const sidecarHealed = await this._healSidecarExportCursor(guildId)
       await this._persistGossipOutboxReplayWatermark(guildId).catch(() => null)
       const memberPageResumed = this._resumeMemberPageChainAfterHeal(guildId)
@@ -8553,6 +8772,8 @@ class PearcordPlatform extends EventEmitter {
       const contactsReconciled = await this._reconcileContactsPresenceWithGuild(guildId).catch(
         () => 0
       )
+      const extendedHeal = await this._healPartitionExtendedSlices(guildId)
+      const settingsDeviceSync = await this._healSettingsMeshDeviceSync(guildId).catch(() => null)
       const req = this.requestGuildSyncFromHost()
       const burst = this.requestGuildSyncBurst({ clearPushHalt: false, forcePush: true })
       const discoveryRefresh = await this._refreshDiscoveryAfterGuildMeshAction(
@@ -8566,12 +8787,18 @@ class PearcordPlatform extends EventEmitter {
         presenceMerged,
         readReceiptsHealed,
         sparseAckReconciled,
+        hyperbeeRebuilt,
         channelMetadataHealed,
+        forumSearchHealed,
+        memberRolesHealed,
         sidecarHealed,
         memberPageResumed,
         watermark,
         contactsReconciled,
+        extendedHeal,
+        settingsDeviceSync,
         syncRequested: !!req?.requested,
+        syncFanoutAckCount: req?.ackCount ?? 0,
         burst,
         settingsMesh: !!settingsMesh,
         discoveryScoped: !!discoveryRefresh?.scoped
@@ -14800,7 +15027,16 @@ class PearcordPlatform extends EventEmitter {
       }
     }
     this._wireGuild(this.guild)
-    await this._requestGuildMeshJoin({ source: 'guild-create', bounded: true })
+    const createMeshRtt =
+      Number(this._guildSyncBurstRttMs) ||
+      Number(process.env.PEARCORD_GUILD_MESH_RTT_MS) ||
+      120
+    const createCaps = adaptiveBurstDelays(createMeshRtt, [8500, 12000, 16000])
+    await this._requestGuildMeshJoin({
+      source: 'guild-create',
+      bounded: true,
+      joinMeshCapMs: createCaps[0]
+    })
     await this._initGuildVoice(result.guild.id)
     this.messages = new PearcordMessage({ authorId: user.id, db: this.db })
     await this.messages.ready()