fix(recording): reschedule packet send recording onto server thread

CHANGELOG.md

@@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 - `activeSessions` in `RecorderManager` changed to `ConcurrentHashMap` to prevent `ConcurrentModificationException` (#33)
+- PacketEvents block-break recording is now rescheduled onto the server thread to avoid Netty-thread contention and unsafe shared-state mutation (#43)
 - Nested replay inventory loss when starting a replay during an active replay (#31)
 - Replay controls getting stuck after replay ends (#27)
 - Static `Replay.getInstance()` NPEs in test environments (#32)

src/main/java/me/justindevb/replay/RecordingSession.java

@@ -73,7 +73,11 @@ public RecordingSession(String name, File folder, Collection<Player> players, in
         this.tracker = new EntityTracker(players);
         this.builder = new TimelineBuilder(appendLogWriter, false);
         this.eventHandler = new RecordingEventHandler(tracker, builder, this::getTick);
-        this.packetHandler = new RecordingPacketHandler(tracker, builder, this::getTick);
+        this.packetHandler = new RecordingPacketHandler(
+            tracker,
+            builder,
+            this::getTick,
+            runnable -> replay.getFoliaLib().getScheduler().runNextTick(task -> runnable.run()));
     }
 
     public void start() {

src/main/java/me/justindevb/replay/recording/RecordingPacketHandler.java

@@ -5,67 +5,105 @@
 import com.github.retrooper.packetevents.protocol.packettype.PacketType;
 import com.github.retrooper.packetevents.wrapper.play.server.WrapperPlayServerBlockBreakAnimation;
 import io.github.retrooper.packetevents.util.SpigotConversionUtil;
+import org.bukkit.Bukkit;
 import org.bukkit.entity.Entity;
 import org.bukkit.entity.Player;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.UUID;
 
 /**
  * Handles PacketEvents packet interception during a recording session.
  * Captures block break animation stages for the timeline.
  */
 public class RecordingPacketHandler implements PacketListener {
 
+    @FunctionalInterface
+    public interface MainThreadScheduler {
+        void execute(Runnable task);
+    }
+
+    record BlockBreakAnimation(UUID viewerUuid, int entityId, int x, int y, int z, int stage) {
+    }
+
     private final EntityTracker tracker;
     private final TimelineBuilder builder;
     private final RecordingEventHandler.TickProvider tickProvider;
+    private final MainThreadScheduler mainThreadScheduler;
     private final Map<String, Integer> breakStageDedup = new HashMap<>();
 
     public RecordingPacketHandler(EntityTracker tracker, TimelineBuilder builder, RecordingEventHandler.TickProvider tickProvider) {
+        this(tracker, builder, tickProvider, Runnable::run);
+    }
+
+    public RecordingPacketHandler(EntityTracker tracker,
+                                  TimelineBuilder builder,
+                                  RecordingEventHandler.TickProvider tickProvider,
+                                  MainThreadScheduler mainThreadScheduler) {
         this.tracker = tracker;
         this.builder = builder;
         this.tickProvider = tickProvider;
+        this.mainThreadScheduler = mainThreadScheduler;
     }
 
     @Override
     public void onPacketSend(PacketSendEvent e) {
-        if (e.getPacketType() == PacketType.Play.Server.BLOCK_BREAK_ANIMATION) {
-            WrapperPlayServerBlockBreakAnimation packet = new WrapperPlayServerBlockBreakAnimation(e);
-            Player p = e.getPlayer();
-
-            String world = p.getWorld().getName();
-
-            int stage = packet.getDestroyStage();
-            int x = packet.getBlockPosition().getX();
-            int y = packet.getBlockPosition().getY();
-            int z = packet.getBlockPosition().getZ();
-
-            int tick = tickProvider.getTick();
-
-            String dedupKey = world + ":" + x + ":" + y + ":" + z + ":" + stage;
-            Integer lastTick = breakStageDedup.get(dedupKey);
-            if (lastTick != null && lastTick == tick) {
-                return;
-            }
-            breakStageDedup.put(dedupKey, tick);
-            if (breakStageDedup.size() > 4000) {
-                breakStageDedup.entrySet().removeIf(entry -> entry.getValue() < tick - 40);
-            }
-
-            String breakerUuid = null;
-            Entity entity = SpigotConversionUtil.getEntityById(p.getWorld(), packet.getEntityId());
-            if (entity instanceof Player breaker && tracker.isTrackedPlayer(breaker.getUniqueId())) {
-                breakerUuid = breaker.getUniqueId().toString();
-            }
-
-            if (breakerUuid == null && !tracker.isTrackedPlayer(p.getUniqueId())) {
-                return;
-            }
-
-            builder.addEvent(new TimelineEvent.BlockBreakStage(
-                    tick, breakerUuid, world, x, y, z, stage
-            ));
+        if (e.getPacketType() != PacketType.Play.Server.BLOCK_BREAK_ANIMATION) {
+            return;
         }
+
+        Player viewer = e.getPlayer();
+        if (viewer == null) {
+            return;
+        }
+
+        WrapperPlayServerBlockBreakAnimation packet = new WrapperPlayServerBlockBreakAnimation(e);
+        scheduleBlockBreakAnimation(new BlockBreakAnimation(
+                viewer.getUniqueId(),
+                packet.getEntityId(),
+                packet.getBlockPosition().getX(),
+                packet.getBlockPosition().getY(),
+                packet.getBlockPosition().getZ(),
+                packet.getDestroyStage()
+        ));
+    }
+
+    void scheduleBlockBreakAnimation(BlockBreakAnimation animation) {
+        mainThreadScheduler.execute(() -> recordBlockBreakAnimation(animation));
+    }
+
+    void recordBlockBreakAnimation(BlockBreakAnimation animation) {
+        Player viewer = Bukkit.getPlayer(animation.viewerUuid());
+        if (viewer == null || !viewer.isOnline()) {
+            return;
+        }
+
+        String world = viewer.getWorld().getName();
+        int tick = tickProvider.getTick();
+
+        String dedupKey = world + ":" + animation.x() + ":" + animation.y() + ":" + animation.z() + ":" + animation.stage();
+        Integer lastTick = breakStageDedup.get(dedupKey);
+        if (lastTick != null && lastTick == tick) {
+            return;
+        }
+        breakStageDedup.put(dedupKey, tick);
+        if (breakStageDedup.size() > 4000) {
+            breakStageDedup.entrySet().removeIf(entry -> entry.getValue() < tick - 40);
+        }
+
+        String breakerUuid = null;
+        Entity entity = SpigotConversionUtil.getEntityById(viewer.getWorld(), animation.entityId());
+        if (entity instanceof Player breaker && tracker.isTrackedPlayer(breaker.getUniqueId())) {
+            breakerUuid = breaker.getUniqueId().toString();
+        }
+
+        if (breakerUuid == null && !tracker.isTrackedPlayer(animation.viewerUuid())) {
+            return;
+        }
+
+        builder.addEvent(new TimelineEvent.BlockBreakStage(
+                tick, breakerUuid, world, animation.x(), animation.y(), animation.z(), animation.stage()
+        ));
     }
 }

src/test/java/me/justindevb/replay/recording/RecordingPacketHandlerTest.java

@@ -2,19 +2,30 @@
 
 import com.github.retrooper.packetevents.event.PacketSendEvent;
 import com.github.retrooper.packetevents.protocol.packettype.PacketType;
+import io.github.retrooper.packetevents.util.SpigotConversionUtil;
+import org.bukkit.Bukkit;
+import org.bukkit.World;
+import org.bukkit.entity.Player;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
+import org.mockito.MockedStatic;
 import org.mockito.junit.jupiter.MockitoExtension;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
 @ExtendWith(MockitoExtension.class)
 class RecordingPacketHandlerTest {
 
     @Mock private EntityTracker tracker;
+    @Mock private Player player;
+    @Mock private World world;
     private TimelineBuilder builder;
     private int tick = 5;
     private RecordingPacketHandler handler;
@@ -43,4 +54,59 @@ void nonBlockBreakAnimationPacket_ignored() {
     void handler_constructsWithoutError() {
         assertNotNull(handler);
     }
+
+    @Test
+    void scheduleBlockBreakAnimation_defersTimelineMutationUntilMainThread() {
+        List<Runnable> scheduledTasks = new ArrayList<>();
+        handler = new RecordingPacketHandler(tracker, builder, () -> tick, scheduledTasks::add);
+
+        handler.scheduleBlockBreakAnimation(new RecordingPacketHandler.BlockBreakAnimation(
+                UUID.randomUUID(),
+                17,
+                10,
+                64,
+                20,
+                4));
+
+        assertEquals(1, scheduledTasks.size());
+        assertTrue(builder.getTimeline().isEmpty());
+    }
+
+    @Test
+    void recordBlockBreakAnimation_addsEventForTrackedViewer() {
+        UUID viewerUuid = UUID.randomUUID();
+
+        try (MockedStatic<Bukkit> bukkit = mockStatic(Bukkit.class);
+             MockedStatic<SpigotConversionUtil> conversion = mockStatic(SpigotConversionUtil.class)) {
+
+            bukkit.when(() -> Bukkit.getPlayer(viewerUuid)).thenReturn(player);
+            conversion.when(() -> SpigotConversionUtil.getEntityById(world, 99)).thenReturn(null);
+
+            when(player.isOnline()).thenReturn(true);
+            when(player.getWorld()).thenReturn(world);
+            when(world.getName()).thenReturn("world");
+            when(tracker.isTrackedPlayer(viewerUuid)).thenReturn(true);
+
+            handler.recordBlockBreakAnimation(new RecordingPacketHandler.BlockBreakAnimation(
+                    viewerUuid,
+                    99,
+                    1,
+                    65,
+                    2,
+                    7));
+
+            List<TimelineEvent> timeline = builder.getTimeline();
+            assertEquals(1, timeline.size());
+            assertInstanceOf(TimelineEvent.BlockBreakStage.class, timeline.getFirst());
+
+            TimelineEvent.BlockBreakStage event = (TimelineEvent.BlockBreakStage) timeline.getFirst();
+            assertEquals(tick, event.tick());
+            assertNull(event.uuid());
+            assertEquals("world", event.world());
+            assertEquals(1, event.x());
+            assertEquals(65, event.y());
+            assertEquals(2, event.z());
+            assertEquals(7, event.stage());
+        }
+    }
 }