Fix fastjson2 concurrent ref deserialization

dubbo-serialization/dubbo-serialization-fastjson2/src/main/java/org/apache/dubbo/common/serialize/fastjson2/FastJson2ObjectInput.java

@@ -31,6 +31,16 @@
  */
 public class FastJson2ObjectInput implements ObjectInput {
 
+    // fastjson2 may corrupt JSONB $ref resolution when the same target class is parsed concurrently.
+    private static final Object NULL_PARSE_LOCK = new Object();
+
+    private static final ClassValue<Object> PARSE_LOCKS = new ClassValue<Object>() {
+        @Override
+        protected Object computeValue(Class<?> type) {
+            return new Object();
+        }
+    };
+
     private final Fastjson2CreatorManager fastjson2CreatorManager;
 
     private final Fastjson2SecurityManager fastjson2SecurityManager;
@@ -117,27 +127,7 @@ public <T> T readObject(Class<T> cls) throws IOException {
                     "deserialize failed. expected read length: " + length + " but actual read: " + read);
         }
         Fastjson2SecurityManager.Handler securityFilter = fastjson2SecurityManager.getSecurityFilter();
-        T result;
-        if (securityFilter.isCheckSerializable()) {
-            result = JSONB.parseObject(
-                    bytes,
-                    cls,
-                    securityFilter,
-                    JSONReader.Feature.UseDefaultConstructorAsPossible,
-                    JSONReader.Feature.ErrorOnNoneSerializable,
-                    JSONReader.Feature.IgnoreAutoTypeNotMatch,
-                    JSONReader.Feature.UseNativeObject,
-                    JSONReader.Feature.FieldBased);
-        } else {
-            result = JSONB.parseObject(
-                    bytes,
-                    cls,
-                    securityFilter,
-                    JSONReader.Feature.UseDefaultConstructorAsPossible,
-                    JSONReader.Feature.UseNativeObject,
-                    JSONReader.Feature.IgnoreAutoTypeNotMatch,
-                    JSONReader.Feature.FieldBased);
-        }
+        T result = parseObject(bytes, cls, securityFilter);
         if (result != null && cls != null && !ClassUtils.isMatch(result.getClass(), cls)) {
             throw new IllegalArgumentException(
                     "deserialize failed. expected class: " + cls + " but actual class: " + result.getClass());
@@ -156,27 +146,7 @@ public <T> T readObject(Class<T> cls, Type type) throws IOException, ClassNotFou
                     "deserialize failed. expected read length: " + length + " but actual read: " + read);
         }
         Fastjson2SecurityManager.Handler securityFilter = fastjson2SecurityManager.getSecurityFilter();
-        T result;
-        if (securityFilter.isCheckSerializable()) {
-            result = JSONB.parseObject(
-                    bytes,
-                    cls,
-                    securityFilter,
-                    JSONReader.Feature.UseDefaultConstructorAsPossible,
-                    JSONReader.Feature.ErrorOnNoneSerializable,
-                    JSONReader.Feature.IgnoreAutoTypeNotMatch,
-                    JSONReader.Feature.UseNativeObject,
-                    JSONReader.Feature.FieldBased);
-        } else {
-            result = JSONB.parseObject(
-                    bytes,
-                    cls,
-                    securityFilter,
-                    JSONReader.Feature.UseDefaultConstructorAsPossible,
-                    JSONReader.Feature.UseNativeObject,
-                    JSONReader.Feature.IgnoreAutoTypeNotMatch,
-                    JSONReader.Feature.FieldBased);
-        }
+        T result = parseObject(bytes, cls, securityFilter);
         if (result != null && cls != null && !ClassUtils.isMatch(result.getClass(), cls)) {
             throw new IllegalArgumentException(
                     "deserialize failed. expected class: " + cls + " but actual class: " + result.getClass());
@@ -205,4 +175,32 @@ private int readLength() throws IOException {
         }
         return value;
     }
+
+    private <T> T parseObject(byte[] bytes, Class<T> cls, Fastjson2SecurityManager.Handler securityFilter) {
+        synchronized (getParseLock(cls)) {
+            if (securityFilter.isCheckSerializable()) {
+                return JSONB.parseObject(
+                        bytes,
+                        cls,
+                        securityFilter,
+                        JSONReader.Feature.UseDefaultConstructorAsPossible,
+                        JSONReader.Feature.ErrorOnNoneSerializable,
+                        JSONReader.Feature.IgnoreAutoTypeNotMatch,
+                        JSONReader.Feature.UseNativeObject,
+                        JSONReader.Feature.FieldBased);
+            }
+            return JSONB.parseObject(
+                    bytes,
+                    cls,
+                    securityFilter,
+                    JSONReader.Feature.UseDefaultConstructorAsPossible,
+                    JSONReader.Feature.UseNativeObject,
+                    JSONReader.Feature.IgnoreAutoTypeNotMatch,
+                    JSONReader.Feature.FieldBased);
+        }
+    }
+
+    private Object getParseLock(Class<?> cls) {
+        return cls == null ? NULL_PARSE_LOCK : PARSE_LOCKS.get(cls);
+    }
 }

dubbo-serialization/dubbo-serialization-fastjson2/src/test/java/org/apache/dubbo/common/serialize/fastjson2/FastJson2SerializationTest.java

@@ -27,19 +27,61 @@
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.Serializable;
+import java.lang.reflect.Field;
 import java.time.LocalDate;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
+import com.alibaba.fastjson2.JSONFactory;
+import com.alibaba.fastjson2.reader.ObjectReaderProvider;
 import com.example.test.TestPojo;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class FastJson2SerializationTest {
 
+    public static class RefOuter implements Serializable {
+        private List<RefInner> items;
+
+        public List<RefInner> getItems() {
+            return items;
+        }
+
+        public void setItems(List<RefInner> items) {
+            this.items = items;
+        }
+    }
+
+    public static class RefInner implements Serializable {
+        private String name;
+        private List<Long> ids;
+
+        public String getName() {
+            return name;
+        }
+
+        public void setName(String name) {
+            this.name = name;
+        }
+
+        public List<Long> getIds() {
+            return ids;
+        }
+
+        public void setIds(List<Long> ids) {
+            this.ids = ids;
+        }
+    }
+
     @Test
     void testReadString() throws IOException {
         FrameworkModel frameworkModel = new FrameworkModel();
@@ -373,6 +415,22 @@ void testReadObject() throws IOException, ClassNotFoundException {
         frameworkModel.destroy();
     }
 
+    @Test
+    void testConcurrentReadObjectWithReferences() throws Exception {
+        FrameworkModel frameworkModel = new FrameworkModel();
+        try {
+            Serialization serialization =
+                    frameworkModel.getExtensionLoader(Serialization.class).getExtension("fastjson2");
+            URL url = URL.valueOf("").setScopeModel(frameworkModel);
+            byte[] bytes = serializeRefOuter(serialization, url);
+
+            Assertions.assertEquals(0, countNullIds(serialization, url, bytes));
+            Assertions.assertEquals(0, countConcurrentNullIds(serialization, url, bytes));
+        } finally {
+            frameworkModel.destroy();
+        }
+    }
+
     @Test
     void testReadObjectNotMatched() throws IOException, ClassNotFoundException {
         FrameworkModel frameworkModel = new FrameworkModel();
@@ -620,4 +678,95 @@ void testLimit5() throws IOException, ClassNotFoundException {
             frameworkModel.destroy();
         }
     }
+
+    private byte[] serializeRefOuter(Serialization serialization, URL url) throws IOException {
+        RefOuter outer = new RefOuter();
+        List<RefInner> items = new ArrayList<>();
+        List<Long> sharedIds = new ArrayList<>();
+        sharedIds.add(1L);
+        sharedIds.add(2L);
+        for (int i = 0; i < 20; i++) {
+            RefInner inner = new RefInner();
+            inner.setName("item-" + i);
+            inner.setIds(sharedIds);
+            items.add(inner);
+        }
+        outer.setItems(items);
+
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        ObjectOutput objectOutput = serialization.serialize(url, outputStream);
+        objectOutput.writeObject(outer);
+        objectOutput.flushBuffer();
+        return outputStream.toByteArray();
+    }
+
+    private int countConcurrentNullIds(Serialization serialization, URL url, byte[] bytes) throws Exception {
+        int rounds = 10;
+        int threadCount = 200;
+        AtomicInteger totalNullTasks = new AtomicInteger();
+        AtomicReference<Throwable> failure = new AtomicReference<>();
+
+        for (int round = 0; round < rounds; round++) {
+            clearObjectReaderCache();
+            CyclicBarrier barrier = new CyclicBarrier(threadCount);
+            CountDownLatch endLatch = new CountDownLatch(threadCount);
+            AtomicInteger roundNullTasks = new AtomicInteger();
+            for (int i = 0; i < threadCount; i++) {
+                Thread thread = new Thread(() -> {
+                    try {
+                        barrier.await();
+                        if (countNullIds(serialization, url, bytes) > 0) {
+                            roundNullTasks.incrementAndGet();
+                        }
+                    } catch (Throwable throwable) {
+                        failure.compareAndSet(null, throwable);
+                    } finally {
+                        endLatch.countDown();
+                    }
+                });
+                thread.start();
+            }
+            endLatch.await();
+            if (failure.get() != null) {
+                throw new AssertionError("Concurrent deserialization failed", failure.get());
+            }
+            totalNullTasks.addAndGet(roundNullTasks.get());
+        }
+
+        return totalNullTasks.get();
+    }
+
+    private int countNullIds(Serialization serialization, URL url, byte[] bytes) throws Exception {
+        ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
+        ObjectInput objectInput = serialization.deserialize(url, inputStream);
+        RefOuter outer = objectInput.readObject(RefOuter.class);
+        if (outer == null || outer.getItems() == null) {
+            return 1;
+        }
+
+        int nullCount = 0;
+        for (RefInner inner : outer.getItems()) {
+            if (inner == null || inner.getIds() == null) {
+                nullCount++;
+            }
+        }
+        return nullCount;
+    }
+
+    @SuppressWarnings("unchecked")
+    private void clearObjectReaderCache() throws Exception {
+        ObjectReaderProvider provider = JSONFactory.getDefaultObjectReaderProvider();
+        for (String fieldName : new String[] {"cache", "cacheFieldBased"}) {
+            Field field = ObjectReaderProvider.class.getDeclaredField(fieldName);
+            field.setAccessible(true);
+            Object cache = field.get(provider);
+            if (cache instanceof Map) {
+                ((Map<?, ?>) cache).clear();
+            }
+        }
+
+        Field readerCacheField = ObjectReaderProvider.class.getDeclaredField("readerCache");
+        readerCacheField.setAccessible(true);
+        readerCacheField.set(null, null);
+    }
 }