perf: eliminate per-message topic string alloc in comms receive path

Explorer/Assets/DCL/Infrastructure/SceneRuntime/Apis/Modules/CommsApi/CommsApiWrap.cs

@@ -8,6 +8,7 @@
 using SceneRunner.Scene.ExceptionsHandling;
 using System;
 using System.Buffers.Binary;
+using System.Collections.Generic;
 using System.IO;
 using System.Text;
 using System.Threading;
@@ -40,6 +41,15 @@ public sealed class CommsApiWrap : JsApiWrapper
         private readonly DCLConcurrentDictionary<string, DCLConcurrentQueue<BufferedDataMessage>> topicBuffers = new ();
         private readonly DCLConcurrentDictionary<string, (int count, int windowStartMs)> publishRateLimiters = new ();
 
+        private readonly object topicLookupLock = new ();
+
+        // Byte-keyed copy-on-write snapshot of topicBuffers (queues shared, not copied) so OnDataReceived
+        // can match the wire topic without materializing a string per message — Unity's BCL has no
+        // span-keyed dictionary lookup (GetAlternateLookup is .NET 9+). Writers rebuild under
+        // topicLookupLock and publish a new immutable array; the LiveKit-thread reader sees either
+        // the previous or the new snapshot, never a partial one.
+        private volatile TopicLookupEntry[] topicLookup = Array.Empty<TopicLookupEntry>();
+
         public CommsApiWrap(
             IRoomHub roomHub,
             ISceneCommunicationPipe sceneCommunicationPipe,
@@ -60,6 +70,7 @@ public override void Dispose()
         {
             sceneCommunicationPipe.RemoveSceneMessageHandler(sceneId, ISceneCommunicationPipe.MsgType.CommsData, onDataReceivedCached);
             topicBuffers.Clear();
+            topicLookup = Array.Empty<TopicLookupEntry>();
             publishRateLimiters.Clear();
             commsWriter.Dispose();
         }
@@ -128,7 +139,7 @@ public string GetActiveVideoStreams()
         /// Called from JS via ClearScript. Rate-limited to <see cref="MAX_MESSAGES_PER_SECOND"/> per topic.
         /// </summary>
         [UsedImplicitly]
-        public void PublishData(string topic, string data)
+        public void PublishData(string topic, string? data)
         {
             try
             {
@@ -191,7 +202,8 @@ public void PublishData(string topic, string data)
         public void SubscribeToTopic(string topic)
         {
             // method is called relatively rare, allocation new Queue is acceptable, pooling not required
-            topicBuffers.TryAdd(topic, new DCLConcurrentQueue<BufferedDataMessage>());
+            if (topicBuffers.TryAdd(topic, new DCLConcurrentQueue<BufferedDataMessage>()))
+                RebuildTopicLookup();
         }
 
         /// <summary>
@@ -201,8 +213,10 @@ public void SubscribeToTopic(string topic)
         [UsedImplicitly]
         public void UnsubscribeFromTopic(string topic)
         {
-            topicBuffers.TryRemove(topic, out DCLConcurrentQueue<BufferedDataMessage> _output);
-            // 'output' object is droped and will be collected by GC (it's assumed nothing else holds the reference)
+            if (topicBuffers.TryRemove(topic, out _))
+                RebuildTopicLookup();
+
+            // the removed queue is dropped and will be collected by GC (it's assumed nothing else holds the reference)
         }
 
         /// <summary>
@@ -248,13 +262,13 @@ public string ConsumeMessages(string topic)
 
         /// <summary>
         /// Runs on the LiveKit callback thread (ORIGIN_THREAD), not the main thread.
-        /// Only thread-safe types (DCLConcurrentQueue, Encoding) are used here.
+        /// Only thread-safe access is used here: a volatile read of the immutable topicLookup
+        /// snapshot, DCLConcurrentQueue and Encoding. Must not allocate for messages on
+        /// unsubscribed topics — all CommsData traffic for the scene reaches this handler.
         /// Decodes wire format: [topicLen 2 bytes LE][topic UTF-8][data UTF-8].
         /// </summary>
         private void OnDataReceived(ISceneCommunicationPipe.DecodedMessage message)
         {
-            // TODO: implement GetAlternateLookup on ReadOnlySpan<char/byte> to avoid allocation of temp string instances
-            // Reference: https://learn.microsoft.com/en-us/dotnet/api/system.collections.generic.dictionary-2.getalternatelookup
             ReadOnlySpan<byte> span = message.Data;
 
             if (span.Length < TOPIC_LENGTH_PREFIX_BYTES) return;
@@ -263,10 +277,19 @@ private void OnDataReceived(ISceneCommunicationPipe.DecodedMessage message)
 
             if (span.Length < TOPIC_LENGTH_PREFIX_BYTES + topicLength) return;
 
-            string topic = Encoding.UTF8.GetString(span.Slice(TOPIC_LENGTH_PREFIX_BYTES, topicLength));
+            ReadOnlySpan<byte> topicSpan = span.Slice(TOPIC_LENGTH_PREFIX_BYTES, topicLength);
 
-            if (topicBuffers.TryGetValue(topic, out DCLConcurrentQueue<BufferedDataMessage> queue))
+            // Scenes subscribe to a handful of topics, so a linear scan over the snapshot
+            // beats hashing (which would require materializing a string key).
+            TopicLookupEntry[] lookup = topicLookup;
+
+            for (var i = 0; i < lookup.Length; i++)
             {
+                if (!topicSpan.SequenceEqual(lookup[i].Utf8Topic))
+                    continue;
+
+                DCLConcurrentQueue<BufferedDataMessage> queue = lookup[i].Queue;
+
                 // DROP OLD POLICY. Dequeues oldest item to insert new one
                 if (queue.Count >= TOPIC_BUFFER_MAX_MESSAGE_COUNT)
                 {
@@ -275,6 +298,20 @@ private void OnDataReceived(ISceneCommunicationPipe.DecodedMessage message)
 
                 string data = Encoding.UTF8.GetString(span[(TOPIC_LENGTH_PREFIX_BYTES + topicLength)..]);
                 queue.Enqueue(new BufferedDataMessage(message.FromWalletId, data));
+                return;
+            }
+        }
+
+        private void RebuildTopicLookup()
+        {
+            lock (topicLookupLock)
+            {
+                var entries = new List<TopicLookupEntry>(topicBuffers.Count);
+
+                foreach (KeyValuePair<string, DCLConcurrentQueue<BufferedDataMessage>> pair in topicBuffers)
+                    entries.Add(new TopicLookupEntry(Encoding.UTF8.GetBytes(pair.Key), pair.Value));
+
+                topicLookup = entries.ToArray();
             }
         }
 
@@ -303,6 +340,18 @@ private bool TryConsumeRateLimit(string topic)
             return true;
         }
 
+        private readonly struct TopicLookupEntry
+        {
+            public readonly byte[] Utf8Topic;
+            public readonly DCLConcurrentQueue<BufferedDataMessage> Queue;
+
+            public TopicLookupEntry(byte[] utf8Topic, DCLConcurrentQueue<BufferedDataMessage> queue)
+            {
+                Utf8Topic = utf8Topic;
+                Queue = queue;
+            }
+        }
+
         private readonly struct BufferedDataMessage
         {
             public readonly string SenderIdentity;

Explorer/Assets/DCL/Infrastructure/SceneRuntime/Tests/CommsApiWrapShould.cs

@@ -5,24 +5,24 @@
 using NUnit.Framework;
 using SceneRunner.Scene;
 using SceneRunner.Scene.ExceptionsHandling;
-using SceneRuntime;
 using SceneRuntime.Apis.Modules.CommsApi;
 using System;
 using System.Collections.Generic;
 using System.Text;
 using System.Threading;
+using UnityEngine.Profiling;
 
 namespace SceneRuntime.Tests
 {
     public class CommsApiWrapShould
     {
         private const string TEST_SCENE_ID = "test-scene-123";
 
-        private CommsApiWrap commsApi;
-        private TestSceneCommunicationPipe pipe;
-        private IRoomHub roomHub;
-        private ISceneExceptionsHandler exceptionsHandler;
-        private CancellationTokenSource cts;
+        private CommsApiWrap commsApi = null!;
+        private TestSceneCommunicationPipe pipe = null!;
+        private IRoomHub roomHub = null!;
+        private ISceneExceptionsHandler exceptionsHandler = null!;
+        private CancellationTokenSource cts = null!;
 
         [SetUp]
         public void SetUp()
@@ -52,7 +52,7 @@ public void RegisterHandlerOnConstruction()
             //Assert
             Assert.AreEqual(TEST_SCENE_ID, pipe.registeredSceneId);
             Assert.AreEqual(ISceneCommunicationPipe.MsgType.CommsData, pipe.registeredMsgType);
-            Assert.IsNotNull(pipe.onSceneMessage);
+            Assert.IsNotNull(pipe.sceneMessageHandler);
         }
 
         [Test]
@@ -111,7 +111,7 @@ public void PublishAndReceiveRoundTrip()
 
             // Simulate receive: SceneCommunicationPipe.DecodeMessage strips byte[0] (MsgType).
             ReadOnlySpan<byte> afterMsgType = wireBytes.AsSpan(1);
-            pipe.onSceneMessage.Invoke(new ISceneCommunicationPipe.DecodedMessage(afterMsgType, senderIdentity, isTrustedSource: true));
+            pipe.sceneMessageHandler.Invoke(new ISceneCommunicationPipe.DecodedMessage(afterMsgType, senderIdentity, isTrustedSource: true));
 
             //Assert — ConsumeMessages returns JSON; inner data string is JSON-escaped by JsonTextWriter.
             string json = commsApi.ConsumeMessages(topic);
@@ -175,6 +175,78 @@ public void DropMessagesForUnsubscribedTopics()
             Assert.AreEqual("[]", result, "Messages before subscription should be dropped.");
         }
 
+        [Test]
+        public void ReceiveForUnsubscribedTopicDoesNotAllocate()
+        {
+            // GC.GetAllocatedBytesForCurrentThread is inert on the editor Mono runtime, and the
+            // strict AllocatingGCMemory constraint trips on runtime noise outside OnDataReceived —
+            // so this uses the budgeted GC.Alloc Recorder idiom with a liveness canary: the bug
+            // allocates at least one sample per message, the budget stays far under one per message.
+            const int MEASURED_INVOKES = 1000;
+            const int CANARY_ALLOCS = 16;
+            const int ALLOC_SAMPLE_BUDGET = 100;
+
+            //Arrange — capture real wire bytes via PublishData (round-trip pattern); topic is never subscribed.
+            commsApi.PublishData("never-subscribed-topic", "{\"type\":\"noise\"}");
+            Assert.AreEqual(1, pipe.sendMessageCalls.Count);
+            byte[] wireBytes = pipe.sendMessageCalls[0];
+
+            // SceneCommunicationPipe.DecodeMessage strips byte[0] (MsgType) before the handler sees it.
+            // DecodedMessage is a ref struct, so the span is rebuilt per call instead of captured.
+            // Warm-up: JIT the receive path outside the measured region.
+            for (var i = 0; i < 64; i++)
+                pipe.sceneMessageHandler.Invoke(new ISceneCommunicationPipe.DecodedMessage(wireBytes.AsSpan(1), "0xSENDER", isTrustedSource: true));
+
+            Recorder gcAllocRecorder = Recorder.Get("GC.Alloc");
+            gcAllocRecorder.FilterToCurrentThread();
+            gcAllocRecorder.enabled = false;
+            gcAllocRecorder.enabled = true;
+
+            //Act
+            for (var i = 0; i < MEASURED_INVOKES; i++)
+                pipe.sceneMessageHandler.Invoke(new ISceneCommunicationPipe.DecodedMessage(wireBytes.AsSpan(1), "0xSENDER", isTrustedSource: true));
+
+            byte[]? canary = null;
+
+            for (var i = 0; i < CANARY_ALLOCS; i++)
+                canary = new byte[16];
+
+            gcAllocRecorder.enabled = false;
+            int measured = gcAllocRecorder.sampleBlockCount;
+            GC.KeepAlive(canary);
+
+            Assert.GreaterOrEqual(measured, CANARY_ALLOCS,
+                "GC.Alloc recorder did not observe the deliberate canary allocations — the probe is inert on this runtime.");
+
+            //Assert — OnDataReceived runs on the LiveKit callback thread for ALL scene CommsData
+            // traffic; the unsubscribed-topic path must not allocate (e.g. a temp topic string).
+            Assert.Less(measured, ALLOC_SAMPLE_BUDGET,
+                $"OnDataReceived allocated GC memory for messages on a topic that was never subscribed ({measured} GC.Alloc samples over {MEASURED_INVOKES} messages).");
+        }
+
+        [Test]
+        public void ResubscribeAfterUnsubscribeReceivesAgain()
+        {
+            //Arrange
+            const string TOPIC = "flip-flop";
+
+            commsApi.SubscribeToTopic(TOPIC);
+            SimulateIncomingMessage(TOPIC, "{\"v\":1}", "sender1");
+
+            //Act — unsubscribe drops both the buffer and delivery; resubscribe restores delivery.
+            commsApi.UnsubscribeFromTopic(TOPIC);
+            SimulateIncomingMessage(TOPIC, "{\"v\":2}", "sender2");
+            commsApi.SubscribeToTopic(TOPIC);
+            SimulateIncomingMessage(TOPIC, "{\"v\":3}", "sender3");
+
+            string json = commsApi.ConsumeMessages(TOPIC);
+
+            //Assert — only the post-resubscribe message survives.
+            Assert.That(json, Does.Not.Contain("sender1"));
+            Assert.That(json, Does.Not.Contain("sender2"));
+            Assert.That(json, Does.Contain("sender3"));
+        }
+
         [Test]
         public void RejectNullData()
         {
@@ -248,7 +320,7 @@ private void SimulateIncomingMessage(string topic, string data, string senderIde
             System.Buffers.Binary.BinaryPrimitives.WriteUInt16LittleEndian(encoded, (ushort)topicBytes.Length);
             topicBytes.CopyTo(encoded, 2);
             dataBytes.CopyTo(encoded, 2 + topicBytes.Length);
-            pipe.onSceneMessage.Invoke(new ISceneCommunicationPipe.DecodedMessage(encoded, senderIdentity, isTrustedSource: true));
+            pipe.sceneMessageHandler.Invoke(new ISceneCommunicationPipe.DecodedMessage(encoded, senderIdentity, isTrustedSource: true));
         }
 
         /// <summary>
@@ -258,24 +330,24 @@ private void SimulateIncomingMessage(string topic, string data, string senderIde
         private class TestSceneCommunicationPipe : ISceneCommunicationPipe
         {
             internal readonly List<byte[]> sendMessageCalls = new ();
-            internal ISceneCommunicationPipe.SceneMessageHandler onSceneMessage;
-            internal string registeredSceneId;
+            internal ISceneCommunicationPipe.SceneMessageHandler sceneMessageHandler = null!;
+            internal string registeredSceneId = null!;
             internal ISceneCommunicationPipe.MsgType registeredMsgType;
             internal bool handlerRemoved;
 
             public void AddSceneMessageHandler(string sceneId, ISceneCommunicationPipe.MsgType msgType, ISceneCommunicationPipe.SceneMessageHandler onSceneMessage)
             {
                 registeredSceneId = sceneId;
                 registeredMsgType = msgType;
-                this.onSceneMessage = onSceneMessage;
+                sceneMessageHandler = onSceneMessage;
             }
 
             public void RemoveSceneMessageHandler(string sceneId, ISceneCommunicationPipe.MsgType msgType, ISceneCommunicationPipe.SceneMessageHandler onSceneMessage)
             {
                 handlerRemoved = true;
             }
 
-            public void SendMessage(ReadOnlySpan<byte> message, string sceneId, ISceneCommunicationPipe.ConnectivityAssertiveness assertiveness, CancellationToken ct, string specialRecipient = null)
+            public void SendMessage(ReadOnlySpan<byte> message, string sceneId, ISceneCommunicationPipe.ConnectivityAssertiveness assertiveness, CancellationToken ct, string? specialRecipient = null)
             {
                 sendMessageCalls.Add(message.ToArray());
             }