fix: heal missed Pulse peer discovery without widening LiveKit fan-out

Remote avatars intermittently disappear after teleporting to a World
and back to Genesis Plaza (and, per the 2026-08-05 recurrence, even
without teleporting, under archipelago fragmentation). The remote-avatar
pipeline is announcement-driven with no reconciliation loop: since the
Pulse integration (PR #7291), a peer is announced to an observer exactly
once per interest-set entry, and while Pulse is active LiveKit's
AnnounceProfileVersion broadcast is targeted only at wallets that
already announced to us. Any dropped, negated, or realm-raced Pulse
join therefore produces a permanently invisible peer -- there is no
client- or server-side membership reconciliation. (The filed v0.154/
v0.155 100%-repro mechanism, PR #9041's ForceRelease of Pulse-sourced
avatars on teleport, is already reverted at the pin; this addresses the
residual one-shot-discovery fragility class that still recurs.)

- LiveKitMessagesBroadcaster: while Pulse is active, fall back to an
  untargeted AnnounceProfileVersion broadcast on the island+scene pipes
  at a slow cadence (~10s) instead of only the targeted send, so any
  peer whose Pulse join we missed self-heals over a LiveKit room we
  demonstrably share. To avoid duplicating the Pulse movement/emote
  stream room-wide, Send's recipient lists now skip any wallet with a
  live PeerIdCache session -- recruitment into announcedWallets stays
  unconditional, but effective fan-out composition for Pulse-live peers
  is unchanged from HEAD.
- PulseMultiplayerBus.PlayerState: HandlePlayerLeft now ignores a leave
  whose SubjectId no longer matches the wallet's current peer id (a
  late/re-ordered leave for a superseded session), closing the window
  where a stale leave deletes a freshly re-joined avatar.
- PeerIdCache: Remove/RemoveWhereNotInRealm only clear the wallet's
  reverse mapping when it still points at the peer id being removed, so
  a dangling forward entry from a re-join can't destroy the live
  session's reverse lookup out from under the new guard.
- Thread PeerIdCache through PulseContainer -> MultiplayerContainer ->
  LiveKitMultiplayerContainer so the broadcaster can filter by live
  session.

Adversarial review (NEEDS-FIX) found the first pass's announce fallback
recruited every Pulse-active peer into announcedWallets within one
interval, turning targeted LiveKit movement/emotes into full-room
fan-out that duplicated the Pulse stream. This revision keeps the
fallback announce-only and filters Send by live PeerIdCache membership,
so only genuinely missing peers get materialized and fan-out
composition matches HEAD.

Add EditMode tests for the join-epoch guard (join/re-join/stale-leave
sequencing) and for the broadcaster's fallback-materializes /
send-path-excludes-live-session behavior.

Fixes #9337.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: eordano

Explorer/Assets/DCL/Multiplayer/Connections/Pulse/PeerIdCache.cs

@@ -31,11 +31,18 @@ public void Remove(uint peerId)
         {
             lock (sync)
             {
-                if (peersByWallet.Remove(peerId, out (Web3Address wallet, string realm) entry))
-                    walletsByPeerId.Remove(entry.wallet);
+                RemoveEntry(peerId);
             }
         }
 
+        private void RemoveEntry(uint peerId)
+        {
+            if (peersByWallet.Remove(peerId, out (Web3Address wallet, string realm) entry)
+                && walletsByPeerId.TryGetValue(entry.wallet, out uint currentPeerId)
+                && currentPeerId == peerId)
+                walletsByPeerId.Remove(entry.wallet);
+        }
+
         /// <summary>
         ///     Atomically iterates all wallets, invokes the callback for each, then clears both caches.
         /// </summary>
@@ -121,8 +128,7 @@ public void RemoveWhereNotInRealm(string realm, Action<uint> onPeerRemoved)
 
                 foreach (uint peerId in removalBuffer)
                 {
-                    if (peersByWallet.Remove(peerId, out (Web3Address wallet, string realm) entry))
-                        walletsByPeerId.Remove(entry.wallet);
+                    RemoveEntry(peerId);
 
                     onPeerRemoved(peerId);
                 }

Explorer/Assets/DCL/Multiplayer/Movement/Systems/LiveKitMultiplayerContainer.cs

@@ -28,9 +28,10 @@ internal LiveKitMultiplayerContainer(
             ISelfProfile selfProfile,
             IUserBlockingCache userBlockingCache,
             MultiplayerDebugSettings multiplayerDebugSettings,
-            PulseActivation pulseActivation)
+            PulseActivation pulseActivation,
+            PeerIdCache peerIdCache)
         {
-            var broadcaster = new LiveKitMessagesBroadcaster(roomHub.SceneRoom(), messagePipesHub, pulseActivation);
+            var broadcaster = new LiveKitMessagesBroadcaster(roomHub.SceneRoom(), messagePipesHub, pulseActivation, peerIdCache);
 
             RemoteAnnouncements = new LiveKitRemoteAnnouncements(messagePipesHub, broadcaster);
             ProfileBroadcast = new DebounceLiveKitProfileBroadcast(new LiveKitProfileBroadcast(selfProfile, broadcaster));

Explorer/Assets/DCL/Multiplayer/Movement/Systems/MultiplayerContainer.cs

@@ -234,7 +234,7 @@ public static async UniTask<MultiplayerContainer> CreateAsync(
             var pulseActivation = new PulseActivation(FeaturesRegistry.Instance.IsEnabled(FeatureId.Pulse));
 
             PulseContainer pulseContainer = await PulseContainer.CreateAsync(pluginSettingsContainer, identityCache, movementInbox, landscapeData, urlsSource, selfProfile, realmData, pulseActivation, ct);
-            var liveKitContainer = new LiveKitMultiplayerContainer(roomHub, messagePipesHub, movementInbox, selfProfile, userBlockingCache, multiplayerDebugSettings, pulseActivation);
+            var liveKitContainer = new LiveKitMultiplayerContainer(roomHub, messagePipesHub, movementInbox, selfProfile, userBlockingCache, multiplayerDebugSettings, pulseActivation, pulseContainer.peerIdCache);
 
             return new MultiplayerContainer(pulseContainer, liveKitContainer, selfProfile, pulseActivation);
         }

Explorer/Assets/DCL/Multiplayer/Movement/Systems/PulseContainer.cs

@@ -20,7 +20,7 @@ internal class PulseContainer : DCLWorldContainer<PulseContainer.Settings>
     {
         private readonly IWeb3IdentityCache identityCache;
         private readonly MovementInbox movementInbox;
-        private readonly PeerIdCache peerIdCache = new ();
+        internal readonly PeerIdCache peerIdCache = new ();
         private readonly MessagePipe messagePipe = new ();
 
         internal ENetTransport? transport;

Explorer/Assets/DCL/Multiplayer/Movement/Systems/PulseMultiplayerBus.PlayerState.cs

@@ -83,7 +83,12 @@ private void HandlePlayerLeft(IncomingMessage message)
 
             // Realm-agnostic on purpose: removals must process even for peers whose stored realm is stale
             if (peerIdCache.TryGetWallet(playerLeft.SubjectId, out Web3Address wallet))
-                removeIntentions.Enqueue(wallet);
+            {
+                if (peerIdCache.TryGetPeerId(wallet, out uint currentPeerId) && currentPeerId != playerLeft.SubjectId)
+                    ReportHub.Log(ReportCategory.MULTIPLAYER, $"Ignoring stale PlayerLeft for {playerLeft.SubjectId}: {wallet} re-joined as {currentPeerId}");
+                else
+                    removeIntentions.Enqueue(wallet);
+            }
 
             peerIdCache.Remove(playerLeft.SubjectId);
             PurgeQueues(playerLeft.SubjectId);

Explorer/Assets/DCL/Multiplayer/Movement/Tests/PulseMultiplayerBusRealmFilteringShould.cs

@@ -292,6 +292,63 @@ public void ProcessPlayerLeftForStaleRealmPeer()
             Assert.IsFalse(peerIdCache.TryGetWallet(7, out _));
         }
 
+        // Regression coverage for unity-explorer#9337 (join-epoch guard, potential-fix.patch site 3+4):
+        // a PlayerLeft for a superseded session (a wallet that already re-joined under a new subject id)
+        // must not delete the freshly re-joined avatar. At pin, HandlePlayerLeft enqueues the remove
+        // unconditionally from the dangling forward entry, and PeerIdCache.Remove(7) then deletes the
+        // *live* wallet->peerId reverse mapping too (peersByWallet[7] still resolves to the wallet even
+        // though walletsByPeerId[wallet] already points at 9) - exactly the "re-join cancels the stale
+        // pending leave" gap the report's [INVISIBLE_AVATAR] diagnosis (03b82789c) named.
+        [Test]
+        public void IgnoreStalePlayerLeftForSupersededSession()
+        {
+            Handle(PlayerJoinedMessage(7, WALLET_1, REALM_A));
+            DrainAnnouncements();
+
+            // Wallet re-joins under a new subject id (e.g. a reconnect burst) before the old session's
+            // leave is processed; the routing thread always serializes these messages in arrival order.
+            Handle(PlayerJoinedMessage(9, WALLET_1, REALM_A));
+            DrainAnnouncements();
+
+            // Late/re-ordered leave for the superseded session (subject id 7).
+            Handle(new ServerMessage { PlayerLeft = new PlayerLeft { SubjectId = 7 } });
+
+            using (OwnedBunch<RemoveIntention> bunch = removeIntentions.Bunch())
+                Assert.IsFalse(bunch.Available(),
+                    "A PlayerLeft for a superseded session must not delete the peer's live re-joined avatar.");
+
+            Assert.IsTrue(peerIdCache.TryGetWallet(9, out Web3Address wallet));
+            Assert.IsTrue(wallet.Equals(WALLET_1));
+
+            Assert.IsTrue(peerIdCache.TryGetPeerId(new Web3Address(WALLET_1), out uint currentPeerId),
+                "The re-join's reverse mapping (wallet -> current peer id) must survive the stale leave; " +
+                "at pin, PeerIdCache.Remove(7) deletes the *live* session's wallet->peerId entry too.");
+            Assert.AreEqual(9u, currentPeerId);
+        }
+
+        // Companion to IgnoreStalePlayerLeftForSupersededSession: the guard must only reject leaves for
+        // superseded sessions, not swallow every future leave for a wallet that has ever re-joined.
+        [Test]
+        public void ProcessPlayerLeftForCurrentSessionAfterRejoin()
+        {
+            Handle(PlayerJoinedMessage(7, WALLET_1, REALM_A));
+            DrainAnnouncements();
+
+            Handle(PlayerJoinedMessage(9, WALLET_1, REALM_A));
+            DrainAnnouncements();
+
+            // Stale leave for the superseded session - ignored (see IgnoreStalePlayerLeftForSupersededSession).
+            Handle(new ServerMessage { PlayerLeft = new PlayerLeft { SubjectId = 7 } });
+
+            // The eventual leave for the CURRENT session (9) must still be processed normally.
+            Handle(new ServerMessage { PlayerLeft = new PlayerLeft { SubjectId = 9 } });
+
+            using (OwnedBunch<RemoveIntention> bunch = removeIntentions.Bunch())
+                CollectionAssert.AreEquivalent(new[] { new RemoveIntention(WALLET_1, RoomSource.Pulse) }, bunch.Collection());
+
+            Assert.IsFalse(peerIdCache.TryGetWallet(9, out _));
+        }
+
         private void Handle(ServerMessage serverMessage)
         {
             byte[] bytes = serverMessage.ToByteArray();

Explorer/Assets/DCL/Multiplayer/Profiles/BroadcastProfiles/LiveKitMessagesBroadcaster.cs

@@ -5,6 +5,7 @@
 using DCL.Multiplayer.Connections.Messaging.Pipe;
 using DCL.Multiplayer.Connections.Pulse;
 using DCL.Multiplayer.Connections.Rooms;
+using DCL.Web3;
 using Google.Protobuf;
 using LiveKit.Rooms;
 using System;
@@ -35,13 +36,36 @@ public class LiveKitMessagesBroadcaster
         /// </summary>
         private readonly PulseActivation pulseActivation;
 
-        private readonly Dictionary<string, RoomSource> announcedWallets = new ();
+        private readonly PeerIdCache peerIdCache;
 
-        public LiveKitMessagesBroadcaster(IGateKeeperSceneRoom sceneRoom, IMessagePipesHub messagePipesHub, PulseActivation pulseActivation)
+        private readonly Dictionary<string, (RoomSource rooms, Web3Address wallet)> announcedWallets = new ();
+
+        private readonly TimeSpan untargetedAnnounceInterval;
+        private DateTime previousUntargetedAnnounce;
+
+        public LiveKitMessagesBroadcaster(IGateKeeperSceneRoom sceneRoom, IMessagePipesHub messagePipesHub, PulseActivation pulseActivation, PeerIdCache peerIdCache)
+            : this(sceneRoom, messagePipesHub, pulseActivation, peerIdCache, TimeSpan.FromSeconds(10)) { }
+
+        public LiveKitMessagesBroadcaster(IGateKeeperSceneRoom sceneRoom, IMessagePipesHub messagePipesHub, PulseActivation pulseActivation, PeerIdCache peerIdCache, TimeSpan untargetedAnnounceInterval)
         {
             this.sceneRoom = sceneRoom;
             this.messagePipesHub = messagePipesHub;
             this.pulseActivation = pulseActivation;
+            this.peerIdCache = peerIdCache;
+            this.untargetedAnnounceInterval = untargetedAnnounceInterval;
+        }
+
+        public void SendProfileAnnouncement<TInput, TMessage>(Action<TInput, TMessage> buildMessage, TInput args,
+            LKDataPacketKind packetKind, CancellationToken ct) where TMessage: class, IMessage, new()
+        {
+            if (pulseActivation.IsActive && DateTime.UtcNow - previousUntargetedAnnounce < untargetedAnnounceInterval)
+            {
+                Send(buildMessage, args, packetKind, ct);
+                return;
+            }
+
+            previousUntargetedAnnounce = DateTime.UtcNow;
+            SendUntargeted(buildMessage, args, packetKind, ct);
         }
 
         public void Send<TInput, TMessage>(Action<TInput, TMessage> buildMessage, TInput args,
@@ -54,8 +78,11 @@ public void Send<TInput, TMessage>(Action<TInput, TMessage> buildMessage, TInput
                 using PooledObject<List<string>> _ = ListPool<string>.Get(out List<string>? islandList);
                 using PooledObject<List<string>> __ = ListPool<string>.Get(out List<string>? sceneList);
 
-                foreach ((string walletId, RoomSource rooms) in announcedWallets)
+                foreach ((string walletId, (RoomSource rooms, Web3Address wallet)) in announcedWallets)
                 {
+                    if (peerIdCache.TryGetPeerId(wallet, out uint _))
+                        continue;
+
                     if (EnumUtils.HasFlag(rooms, RoomSource.Island))
                         islandList.Add(walletId);
 
@@ -67,49 +94,57 @@ public void Send<TInput, TMessage>(Action<TInput, TMessage> buildMessage, TInput
                     sceneList.Add(AUTH_SERVER_IDENTITY);
 
                 if (islandList.Count > 0)
-                    BuildMessageAndSend(messagePipesHub.IslandPipe(), islandList);
+                    BuildMessageAndSend(messagePipesHub.IslandPipe(), islandList, buildMessage, args, packetKind, ct);
 
                 if (sceneList.Count > 0)
-                    BuildMessageAndSend(messagePipesHub.ScenePipe(), sceneList);
+                    BuildMessageAndSend(messagePipesHub.ScenePipe(), sceneList, buildMessage, args, packetKind, ct);
             }
             else
             {
                 // Broadcast as before
-                BuildMessageAndSend(messagePipesHub.IslandPipe(), null);
-                BuildMessageAndSend(messagePipesHub.ScenePipe(), null);
+                SendUntargeted(buildMessage, args, packetKind, ct);
             }
+        }
 
-            void BuildMessageAndSend(IMessagePipe messagePipe, IReadOnlyList<string>? recipients)
-            {
-                MessageWrap<TMessage> message = messagePipe.NewMessage<TMessage>();
-                buildMessage(args, message.Payload);
+        private void SendUntargeted<TInput, TMessage>(Action<TInput, TMessage> buildMessage, TInput args,
+            LKDataPacketKind packetKind, CancellationToken ct) where TMessage: class, IMessage, new()
+        {
+            BuildMessageAndSend(messagePipesHub.IslandPipe(), null, buildMessage, args, packetKind, ct);
+            BuildMessageAndSend(messagePipesHub.ScenePipe(), null, buildMessage, args, packetKind, ct);
+        }
 
-                if (recipients != null)
-                    foreach (string recipient in recipients)
-                        message.AddSpecialRecipient(recipient);
+        private void BuildMessageAndSend<TInput, TMessage>(IMessagePipe messagePipe, IReadOnlyList<string>? recipients,
+            Action<TInput, TMessage> buildMessage, TInput args, LKDataPacketKind packetKind, CancellationToken ct) where TMessage: class, IMessage, new()
+        {
+            MessageWrap<TMessage> message = messagePipe.NewMessage<TMessage>();
+            buildMessage(args, message.Payload);
 
-                message.SendAndDisposeAsync(ct, packetKind).Forget();
-            }
+            if (recipients != null)
+                foreach (string recipient in recipients)
+                    message.AddSpecialRecipient(recipient);
+
+            message.SendAndDisposeAsync(ct, packetKind).Forget();
         }
 
         public void Add(string walletId, RoomSource from)
         {
-            if (announcedWallets.TryGetValue(walletId, out RoomSource source))
-                from |= source;
-
-            announcedWallets[walletId] = from;
+            if (announcedWallets.TryGetValue(walletId, out (RoomSource rooms, Web3Address wallet) entry))
+                announcedWallets[walletId] = (entry.rooms | from, entry.wallet);
+            else
+                announcedWallets[walletId] = (from, new Web3Address(walletId));
         }
 
         public void Remove(string walletId, RoomSource roomSource)
         {
-            if (announcedWallets.TryGetValue(walletId, out RoomSource currentSource))
+            if (announcedWallets.TryGetValue(walletId, out (RoomSource rooms, Web3Address wallet) entry))
             {
+                RoomSource currentSource = entry.rooms;
                 currentSource.RemoveFlag(roomSource);
 
                 if (currentSource == RoomSource.None)
                     announcedWallets.Remove(walletId);
                 else
-                    announcedWallets[walletId] = currentSource;
+                    announcedWallets[walletId] = (currentSource, entry.wallet);
             }
         }
     }

Explorer/Assets/DCL/Multiplayer/Profiles/BroadcastProfiles/LiveKitProfileBroadcast.cs

@@ -33,7 +33,7 @@ async UniTaskVoid GetProfileVersionThenSendAsync(CancellationToken ct)
             {
                 Profile? profile = await selfProfile.ProfileAsync(ct);
 
-                broadcaster.Send<Profile?, AnnounceProfileVersion>(static (p, version) => BuildMessage(p, version), profile, LKDataPacketKind.KindReliable, ct);
+                broadcaster.SendProfileAnnouncement<Profile?, AnnounceProfileVersion>(static (p, version) => BuildMessage(p, version), profile, LKDataPacketKind.KindReliable, ct);
             }
 
             GetProfileVersionThenSendAsync(cancellationTokenSource.Token).Forget();

Explorer/Assets/DCL/Multiplayer/Profiles/BroadcastProfiles/Tests.meta

@@ -0,0 +1,3 @@
+{
+    "reference": "GUID:da80994a355e49d5b84f91c0a84a721f"
+}