decentraland · pravusjif · Jun 12, 2026
@@ -39,6 +39,14 @@ public int CompareTo(CRDTEntity other) =>
         public bool Equals(CRDTEntity other) =>
             Id.Equals(other.Id);
 
+        public override bool Equals(object obj) =>
+            obj is CRDTEntity other && Equals(other);
+
+        // Without the override the default ValueType.GetHashCode kicks in (boxing/reflection-based on Mono)
+        // when no explicit comparer is provided to a dictionary
+        public override int GetHashCode() =>
+            Id;
+
         public override string ToString() =>
             $"E: number {EntityNumber} version {EntityVersion}";
 

@@ -1,6 +1,5 @@
 using CRDT.Memory;
 using CRDT.Protocol;
-using DCL.Diagnostics;
 using System;
 using System.Buffers;
 using System.Collections.Generic;
@@ -149,10 +148,6 @@ public bool TryDeserializePutComponent(ref ReadOnlyMemory<byte> memory, CRDTMess
             memory = memory.Slice(shift + dataLength);
 
             crdtMessage = new CRDTMessage(messageType, entityId, componentId, timestamp, memoryOwner);
-
-            if (messageType == CRDTMessageType.AUTHORITATIVE_PUT_COMPONENT)
-                ReportHub.Log(ReportCategory.CRDT, $"Deserialized {nameof(CRDTMessageType.AUTHORITATIVE_PUT_COMPONENT)} - Entity: {crdtMessage.EntityId}, Component: {crdtMessage.ComponentId}, Timestamp: {crdtMessage.Timestamp}");
-
             return true;
         }
 

@@ -1,5 +1,7 @@
 using Collections.Pooled;
+using CRDT.Memory;
 using CRDT.Protocol.Factory;
+using CRDT.Serializer;
 using System;
 using System.Buffers;
 using System.Collections.Generic;
@@ -11,16 +13,31 @@ public class CRDTProtocol : ICRDTProtocol, IComparer<CRDTProtocol.EntityComponen
     {
         private const int MAX_APPEND_COMPONENTS_COUNT = 100;
 
+        /// <summary>
+        ///     Roughly the number of distinct SDK component types a scene touches: presizing avoids rehash cascades
+        ///     during scene-load PUT storms (every rehash re-rents and copies the pooled backing arrays)
+        /// </summary>
+        private const int LWW_COMPONENTS_CAPACITY = 64;
+
+        /// <summary>
+        ///     Entities per component bucket; hot buckets (e.g. Transform) will still grow beyond it
+        /// </summary>
+        private const int LWW_ENTITIES_CAPACITY = 128;
+
+        private const int APPEND_COMPONENTS_CAPACITY = 8;
+        private const int APPEND_ENTITIES_CAPACITY = 32;
+        private const int DELETED_ENTITIES_CAPACITY = 256;
+
         private State crdtState;
 
         internal ref State CRDTState => ref crdtState;
 
         public CRDTProtocol()
         {
             crdtState = new State(
-                new PooledDictionary<int, int>(),
-                new PooledDictionary<int, PooledDictionary<CRDTEntity, EntityComponentData>>(),
-                new PooledDictionary<int, PooledDictionary<CRDTEntity, PooledList<EntityComponentData>>>());
+                new PooledDictionary<int, int>(DELETED_ENTITIES_CAPACITY),
+                new PooledDictionary<int, PooledDictionary<CRDTEntity, EntityComponentData>>(LWW_COMPONENTS_CAPACITY),
+                new PooledDictionary<int, PooledDictionary<CRDTEntity, PooledList<EntityComponentData>>>(APPEND_COMPONENTS_CAPACITY));
         }
 
         public void Dispose()
@@ -121,16 +138,38 @@ public int CreateMessagesFromTheCurrentState(ProcessedCRDTMessage[] preallocated
         public ProcessedCRDTMessage CreateAppendMessage(CRDTEntity entity, int componentId, int timestamp, in IMemoryOwner<byte> data) =>
             CRDTMessagesFactory.CreateAppendMessage(entity, componentId, timestamp, data);
 
-        public ProcessedCRDTMessage CreatePutMessage(CRDTEntity entity, int componentId, in IMemoryOwner<byte> data) =>
-            crdtState.CreatePutMessage(entity, componentId, data);
+        public ProcessedCRDTMessage CreateAndCommitPutMessage(CRDTEntity entity, int componentId, in IMemoryOwner<byte> data) =>
+            CreateAndCommitLwwMessage(CRDTMessageType.PUT_COMPONENT, entity, componentId, data);
 
-        public ProcessedCRDTMessage CreateDeleteMessage(CRDTEntity entity, int componentId) =>
-            crdtState.CreateDeleteMessage(entity, componentId);
+        public ProcessedCRDTMessage CreateAndCommitDeleteMessage(CRDTEntity entity, int componentId) =>
+            CreateAndCommitLwwMessage(CRDTMessageType.DELETE_COMPONENT, entity, componentId, EmptyMemoryOwner<byte>.EMPTY);
 
-        public void EnforceLWWState(in CRDTMessage message)
+        /// <summary>
+        ///     Creates the LWW message and writes it into the local CRDT state in a single probe of the state dictionaries.
+        ///     The local message is by construction the newest state (its timestamp is the stored one + 1) so the full
+        ///     reconciliation performed by <see cref="ProcessMessage" /> would be redundant.
+        ///     The CRDT state takes the ownership of <paramref name="data" />.
+        /// </summary>
+        private ProcessedCRDTMessage CreateAndCommitLwwMessage(CRDTMessageType messageType, CRDTEntity entity, int componentId, in IMemoryOwner<byte> data)
         {
-            GetReconciliationResultFromLWWMessage(in message, out CRDTReconciliationEffect overrideEffect, out CRDTReconciliationEffect newComponentEffect);
-            UpdateLWWState(in message, overrideEffect, newComponentEffect);
+            if (!crdtState.lwwComponents.TryGetValue(componentId, out PooledDictionary<CRDTEntity, EntityComponentData> inner))
+                crdtState.lwwComponents[componentId] = inner = new PooledDictionary<CRDTEntity, EntityComponentData>(LWW_ENTITIES_CAPACITY, CRDTEntityComparer.INSTANCE);
+
+            var timestamp = 0;
+
+            if (inner.TryGetValue(entity, out EntityComponentData storedData))
+            {
+                timestamp = storedData.Timestamp + 1;
+                storedData.Data.Dispose();
+            }
+            else
+                crdtState.messagesCount++;
+
+            inner[entity] = new EntityComponentData(timestamp, data, messageType);
+
+            return new ProcessedCRDTMessage(
+                new CRDTMessage(messageType, entity, componentId, timestamp, data),
+                CRDTMessageSerializationUtils.GetMessageDataLength(messageType, in data));
         }
 
         private static void GetReconciliationResultFromLWWMessage(in CRDTMessage message, out CRDTReconciliationEffect overrideEffect, out CRDTReconciliationEffect newComponentEffect)
@@ -154,7 +193,7 @@ private static void GetReconciliationResultFromLWWMessage(in CRDTMessage message
 
         private CRDTReconciliationResult UpdateLWWState(in CRDTMessage message, CRDTReconciliationEffect overrideEffect, CRDTReconciliationEffect newComponentEffect, bool ignoreTimestamp = false)
         {
-            bool innerSetExists = crdtState.TryGetLWWComponentState(message, out PooledDictionary<CRDTEntity, EntityComponentData> inner,
+            bool innerSetExists = crdtState.TryGetLWWComponentState(in message, out PooledDictionary<CRDTEntity, EntityComponentData> inner,
                 out bool componentExists, out EntityComponentData storedData);
 
             bool componentWasDeleted = componentExists && storedData.isDeleted;
@@ -201,7 +240,7 @@ private void UpdateLWWState(bool innerSetExists, bool componentExists, PooledDic
             in CRDTMessage crdtMessage, ref EntityComponentData componentData)
         {
             if (!innerSetExists)
-                crdtState.lwwComponents[crdtMessage.ComponentId] = inner = new PooledDictionary<CRDTEntity, EntityComponentData>(CRDTEntityComparer.INSTANCE);
+                crdtState.lwwComponents[crdtMessage.ComponentId] = inner = new PooledDictionary<CRDTEntity, EntityComponentData>(LWW_ENTITIES_CAPACITY, CRDTEntityComparer.INSTANCE);
 
             if (!componentExists)
                 crdtState.messagesCount++;
@@ -231,25 +270,24 @@ private void DeleteEntity(CRDTEntity entityId, int entityNumber, int entityVersi
 
             foreach (PooledDictionary<CRDTEntity, EntityComponentData> componentsStorage in crdtState.lwwComponents.Values)
             {
-                if (componentsStorage.TryGetValue(entityId, out EntityComponentData componentData))
+                // Remove with the out value probes the bucket once, unlike the TryGetValue + Remove pair
+                if (componentsStorage.Remove(entityId, out EntityComponentData componentData))
                 {
                     componentData.Data.Dispose();
-                    componentsStorage.Remove(entityId);
                     crdtState.messagesCount--;
                 }
             }
 
             foreach (KeyValuePair<int, PooledDictionary<CRDTEntity, PooledList<EntityComponentData>>> componentsSet in crdtState.appendComponents)
             {
-                if (componentsSet.Value.TryGetValue(entityId, out PooledList<EntityComponentData> list))
+                if (componentsSet.Value.Remove(entityId, out PooledList<EntityComponentData> list))
                 {
                     crdtState.messagesCount -= list.Count;
 
                     foreach (EntityComponentData entityComponentData in list)
                         entityComponentData.Data.Dispose();
 
                     list.Dispose();
-                    componentsSet.Value.Remove(entityId);
                 }
             }
         }
@@ -300,7 +338,7 @@ private bool TryAppendComponent(in CRDTMessage message)
 
                 // Pooled collection will take care of pooling an internal state, there will be a small garbage related to the class itself
                 // We can tolerate it
-                crdtState.appendComponents[message.ComponentId] = outer = new PooledDictionary<CRDTEntity, PooledList<EntityComponentData>>(CRDTEntityComparer.INSTANCE);
+                crdtState.appendComponents[message.ComponentId] = outer = new PooledDictionary<CRDTEntity, PooledList<EntityComponentData>>(APPEND_ENTITIES_CAPACITY, CRDTEntityComparer.INSTANCE);
 
             outer[message.EntityId] = existingSet;
             return true;
@@ -374,7 +412,7 @@ public State(PooledDictionary<int, int> deletedEntities, PooledDictionary<int, P
                 messagesCount = 0;
             }
 
-            internal readonly bool TryGetLWWComponentState(CRDTMessage message, out PooledDictionary<CRDTEntity, EntityComponentData> inner, out bool componentExists,
+            internal readonly bool TryGetLWWComponentState(in CRDTMessage message, out PooledDictionary<CRDTEntity, EntityComponentData> inner, out bool componentExists,
                 out EntityComponentData storedData) =>
                 TryGetLWWComponentState(message.EntityId, message.ComponentId, out inner, out componentExists, out storedData);
 

@@ -2,7 +2,6 @@
 using CRDT.Memory;
 using CRDT.Serializer;
 using System.Buffers;
-using System.Runtime.CompilerServices;
 
 namespace CRDT.Protocol.Factory
 {
@@ -15,12 +14,6 @@ internal static class CRDTMessagesFactory
         public static ProcessedCRDTMessage CreateAppendMessage(CRDTEntity entity, int componentId, int timestamp, in IMemoryOwner<byte> data) =>
             new (new CRDTMessage(CRDTMessageType.APPEND_COMPONENT, entity, componentId, timestamp, data), CRDTMessageSerializationUtils.GetMessageDataLength(CRDTMessageType.APPEND_COMPONENT, in data));
 
-        public static ProcessedCRDTMessage CreatePutMessage(this in CRDTProtocol.State state, CRDTEntity entity, int componentId, in IMemoryOwner<byte> data) =>
-            CreateLwwMessage(in state, CRDTMessageType.PUT_COMPONENT, entity, componentId, data);
-
-        public static ProcessedCRDTMessage CreateDeleteMessage(this in CRDTProtocol.State state, CRDTEntity entity, int componentId) =>
-            CreateLwwMessage(in state, CRDTMessageType.DELETE_COMPONENT, entity, componentId, EmptyMemoryOwner<byte>.EMPTY);
-
         /// <summary>
         ///     Fills the array with messages corresponding to the current CRDT state
         ///     Creates PUT, APPEND, DELETE COMPONENT and DELETE ENTITY messages
@@ -85,14 +78,5 @@ private static int AddCRDTMessage(ProcessedCRDTMessage[] preallocatedArray, CRDT
             index++;
             return numberOfBytes;
         }
-
-        [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        private static ProcessedCRDTMessage CreateLwwMessage(in CRDTProtocol.State state, CRDTMessageType messageType, CRDTEntity entity, int componentId, in IMemoryOwner<byte> data)
-        {
-            var timestamp = 0;
-
-            if (state.TryGetLWWComponentState(entity, componentId, out _, out _, out CRDTProtocol.EntityComponentData storedData)) timestamp = storedData.Timestamp + 1;
-            return new ProcessedCRDTMessage(new CRDTMessage(messageType, entity, componentId, timestamp, data), CRDTMessageSerializationUtils.GetMessageDataLength(messageType, in data));
-        }
     }
 }
@@ -17,11 +17,6 @@ public interface ICRDTProtocol : IDisposable
 
         CRDTReconciliationResult ProcessMessage(in CRDTMessage message);
 
-        /// <summary>
-        ///     Enforce LWW state update, it must be guaranteed that the message is PUT_COMPONENT or DELETE_COMPONENT
-        /// </summary>
-        void EnforceLWWState(in CRDTMessage message);
-
         /// <summary>
         ///     <inheritdoc cref="CRDTMessagesFactory.CreateMessagesFromTheCurrentState" />
         /// </summary>
@@ -36,13 +31,14 @@ public interface ICRDTProtocol : IDisposable
         ProcessedCRDTMessage CreateAppendMessage(CRDTEntity entity, int componentId, int timestamp, in IMemoryOwner<byte> data);
 
         /// <summary>
-        ///     Creates an LWW PUT Message but does not process it
+        ///     Creates an LWW PUT Message and commits it to the local CRDT state in a single pass.
+        ///     The state takes the ownership of <paramref name="data" />
         /// </summary>
-        ProcessedCRDTMessage CreatePutMessage(CRDTEntity entity, int componentId, in IMemoryOwner<byte> data);
+        ProcessedCRDTMessage CreateAndCommitPutMessage(CRDTEntity entity, int componentId, in IMemoryOwner<byte> data);
 
         /// <summary>
-        ///     Creates an LWW DELETE Message but does not process it
+        ///     Creates an LWW DELETE Message and commits it to the local CRDT state in a single pass
         /// </summary>
-        ProcessedCRDTMessage CreateDeleteMessage(CRDTEntity entity, int componentId);
+        ProcessedCRDTMessage CreateAndCommitDeleteMessage(CRDTEntity entity, int componentId);
     }
 }
@@ -44,7 +44,11 @@ public class EngineAPIImplementation : IEngineApi
         private readonly CustomSampler worldSyncBufferSampler;
         private readonly SceneRuntimeMetrics metrics;
 
-        private readonly Action<OutgoingCRDTMessagesProvider.PendingMessage> processPendingMessage;
+        /// <summary>
+        ///     Invoked for every pending outgoing message before serialization.
+        ///     Null by default so the serialization loop skips the delegate invocation entirely
+        /// </summary>
+        protected virtual Action<OutgoingCRDTMessagesProvider.PendingMessage>? PendingMessageProcessor => null;
 
         public EngineAPIImplementation(
             ISharedPoolsProvider poolsProvider,
@@ -78,8 +82,6 @@ public EngineAPIImplementation(
             outgoingMessagesSampler = CustomSampler.Create("OutgoingMessages");
             crdtProcessMessagesSampler = CustomSampler.Create("CRDTProcessMessage");
             applyBufferSampler = CustomSampler.Create(nameof(ApplySyncCommandBuffer));
-
-            processPendingMessage = ProcessPendingMessage;
         }
 
         public virtual void Dispose() { }
@@ -154,9 +156,11 @@ public PoolableByteArray CrdtGetState()
 
             try
             {
-                // Apply outgoing messages straight-away so they are reflected in the current CRDT state
+                // Drain the outgoing messages so they are reflected in the current CRDT state:
+                // LWW messages are committed to the state by the provider on creation,
+                // the rest (APPEND) is not kept in the state so it must be released straight-away
                 using (OutgoingCRDTMessagesSyncBlock outgoingMessagesSyncBlock = GetSerializationSyncBlock())
-                    SyncOutgoingCRDTMessages(outgoingMessagesSyncBlock.Messages);
+                    DisposeMessagesNotOwnedByState(outgoingMessagesSyncBlock.Messages);
 
                 // Create CRDT Messages from the current state
                 // we know exactly how big the array should be
@@ -190,9 +194,7 @@ public PoolableByteArray CrdtGetState()
         }
 
         private OutgoingCRDTMessagesSyncBlock GetSerializationSyncBlock() =>
-            outgoingCrtdMessagesProvider.GetSerializationSyncBlock(processPendingMessage);
-
-        protected virtual void ProcessPendingMessage(OutgoingCRDTMessagesProvider.PendingMessage pendingMessage) { }
+            outgoingCrtdMessagesProvider.GetSerializationSyncBlock(PendingMessageProcessor);
 
         private PoolableByteArray SerializeOutgoingCRDTMessages()
         {
@@ -209,7 +211,7 @@ private PoolableByteArray SerializeOutgoingCRDTMessages()
 
                     SerializeOutgoingCRDTMessages(outgoingMessagesSyncBlock.Messages, serializationBufferPoolable.Span);
 
-                    SyncOutgoingCRDTMessages(outgoingMessagesSyncBlock.Messages);
+                    DisposeMessagesNotOwnedByState(outgoingMessagesSyncBlock.Messages);
 
                     metrics.MessagesToScene.Add(outgoingMessagesSyncBlock.Messages.Count);
                 }
@@ -225,29 +227,18 @@ private PoolableByteArray SerializeOutgoingCRDTMessages()
         }
 
         /// <summary>
-        ///     If local messages are not written in the local CRDT, the final state
-        ///     including timestamp is incorrect. Subsequent creation of propagated messages will result in a wrong timestamp
+        ///     LWW messages (PUT/DELETE_COMPONENT) are committed to the local CRDT state when they are created
+        ///     (see <see cref="ICRDTProtocol.CreateAndCommitPutMessage" />) so their data is owned by the state.
+        ///     The data of the remaining messages (APPEND) is not kept in <see cref="ICRDTProtocol" /> so it must be released immediately
         /// </summary>
-        private void SyncOutgoingCRDTMessages(IReadOnlyList<ProcessedCRDTMessage> outgoingMessages)
+        private static void DisposeMessagesNotOwnedByState(IReadOnlyList<ProcessedCRDTMessage> outgoingMessages)
         {
-            for (var i = 0; i < outgoingMessages.Count; i++) { SyncCRDTMessage(outgoingMessages[i]); }
-        }
-
-        private void SyncCRDTMessage(ProcessedCRDTMessage message)
-        {
-            // We are interested in LWW messages only, in AUTHORITATIVE_PUT_COMPONENT we are not interested as it can't be originated from the client
-            switch (message.message.Type)
+            for (var i = 0; i < outgoingMessages.Count; i++)
             {
-                case CRDTMessageType.DELETE_COMPONENT:
-                case CRDTMessageType.PUT_COMPONENT:
-                    // instead of processing via CRDTProtocol.ProcessMessage
-                    // we can skip part of the logic as we guarantee that the local message is the final valid state (see OutgoingCRDTMessagesProvider.AddLwwMessage)
-                    crdtProtocol.EnforceLWWState(message.message);
-                    break;
-                default:
-                    // as this data is not kept in CRDTProtocol it must be released immediately
-                    message.message.Data.Dispose();
-                    break;
+                CRDTMessage message = outgoingMessages[i].message;
+
+                if (message.Type is not CRDTMessageType.PUT_COMPONENT and not CRDTMessageType.DELETE_COMPONENT)
+                    message.Data.Dispose();
             }
         }