From 52583f2056dc42b55cec4e96b71d78d5ba7cdf42 Mon Sep 17 00:00:00 2001 From: Octoturge Date: Sun, 9 Aug 2026 19:30:32 +0200 Subject: [PATCH] Add periodic reconciliation sweep (Phase 7) Adds a rotating, non-overlapping-batch ReconciliationScheduler (common/, loader-agnostic) and wires it into the 1.12.2 leaf: a new tick timer (reconciliationIntervalTicks, mirroring the existing delta-flush timer) periodically re-reads and resends a bounded slice of currently-loaded chunks, catching mutations that never fire a block event (world-gen, other mods writing blocks directly, /fill, etc.). ChunkAdapter gains loadedChunkKeys(); Forge1122ChunkAdapter implements it via ChunkProviderServer.getLoadedChunks(). MapperConfig gains reconciliationChunksPerSweep to bound sweep cost. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_015tKdPZt78zbPUZMXWzKEKt --- .../mcmapper/common/ChunkAdapter.java | 8 ++ .../common/ReconciliationScheduler.java | 48 +++++++ .../mcmapper/common/config/MapperConfig.java | 1 + .../common/ReconciliationSchedulerTest.java | 123 ++++++++++++++++++ .../forge1122/Forge1122ChunkAdapter.java | 10 ++ .../mcmapper/forge1122/MCMapperMod.java | 48 ++++++- 6 files changed, 232 insertions(+), 6 deletions(-) create mode 100644 common/src/main/java/com/octoturge/mcmapper/common/ReconciliationScheduler.java create mode 100644 common/src/test/java/com/octoturge/mcmapper/common/ReconciliationSchedulerTest.java diff --git a/common/src/main/java/com/octoturge/mcmapper/common/ChunkAdapter.java b/common/src/main/java/com/octoturge/mcmapper/common/ChunkAdapter.java index 54bc462..9b4ffe4 100644 --- a/common/src/main/java/com/octoturge/mcmapper/common/ChunkAdapter.java +++ b/common/src/main/java/com/octoturge/mcmapper/common/ChunkAdapter.java @@ -21,6 +21,14 @@ public interface ChunkAdapter { /** Register the loader-specific hooks (block place/break, chunk load/unload) that feed the dirty buffer. */ void registerEventHooks(DeltaSink sink); + /** + * Currently-loaded chunk coordinates, packed as {@code (chunkX << 32) | (chunkZ & 0xFFFFFFFFL)} + * — the same packing the leaf modules' own dirty-chunk sets already use. Feeds + * {@link ReconciliationScheduler}, which picks a bounded rotating slice of this set to + * re-read and resend each periodic sweep (see the plan's "hybrid" change-detection decision). + */ + java.util.List loadedChunkKeys(); + interface DeltaSink { void onDelta(DeltaEvent event); diff --git a/common/src/main/java/com/octoturge/mcmapper/common/ReconciliationScheduler.java b/common/src/main/java/com/octoturge/mcmapper/common/ReconciliationScheduler.java new file mode 100644 index 0000000..4aea6c0 --- /dev/null +++ b/common/src/main/java/com/octoturge/mcmapper/common/ReconciliationScheduler.java @@ -0,0 +1,48 @@ +package com.octoturge.mcmapper.common; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +/** + * Picks a bounded-size, rotating slice of the currently-loaded chunk set to re-read and resend + * on each periodic reconciliation sweep (see the plan's "hybrid" change-detection decision) — + * catches mutations that never fire a {@code BlockEvent} (world-gen, other mods writing blocks + * directly, {@code /fill}, piston pushes into unloaded-at-the-time chunks, etc.) without paying + * the cost of re-reading every loaded chunk on every sweep. Pure/loader-agnostic on purpose — + * enumerating "currently loaded chunks" stays in each leaf's {@link ChunkAdapter}. + */ +public class ReconciliationScheduler { + private int cursor = 0; + + /** + * @param loadedChunkKeys all currently-loaded chunks, packed as {@code (chunkX << 32) | (chunkZ & 0xFFFFFFFFL)} + * (same packing MCMapperMod already uses for its own dirty-chunk sets) + * @param maxPerSweep upper bound on how many chunks to return this call + * @return up to {@code maxPerSweep} keys to reconcile this sweep; the internal cursor + * advances so the next call continues where this one left off, wrapping around once + * every key has been covered. Not a strict guarantee if the loaded set changes + * between calls (chunks unloading/loading) — best-effort coverage is the point, a + * missed chunk just gets picked up on a later sweep or the next event-driven change. + */ + public List next(Collection loadedChunkKeys, int maxPerSweep) { + if (loadedChunkKeys.isEmpty() || maxPerSweep <= 0) { + return Collections.emptyList(); + } + + List sorted = new ArrayList<>(loadedChunkKeys); + Collections.sort(sorted); + int size = sorted.size(); + + // A batch never wraps mid-call — it stops at the end of the current cycle instead of + // splicing the start back in, so no key is ever reconciled twice before every other key + // has had a turn. This means a batch can come back smaller than maxPerSweep right at a + // cycle boundary; the next call resumes at index 0 for a full-size batch again. + int start = cursor % size; + int count = Math.min(maxPerSweep, size - start); + List batch = new ArrayList<>(sorted.subList(start, start + count)); + cursor = (start + count) % size; + return batch; + } +} diff --git a/common/src/main/java/com/octoturge/mcmapper/common/config/MapperConfig.java b/common/src/main/java/com/octoturge/mcmapper/common/config/MapperConfig.java index 1ccdc24..5ca18ac 100644 --- a/common/src/main/java/com/octoturge/mcmapper/common/config/MapperConfig.java +++ b/common/src/main/java/com/octoturge/mcmapper/common/config/MapperConfig.java @@ -12,4 +12,5 @@ public class MapperConfig { public boolean playerTrackingEnabled = true; public int deltaFlushIntervalTicks = 20; public int reconciliationIntervalTicks = 20 * 60 * 5; + public int reconciliationChunksPerSweep = 50; } diff --git a/common/src/test/java/com/octoturge/mcmapper/common/ReconciliationSchedulerTest.java b/common/src/test/java/com/octoturge/mcmapper/common/ReconciliationSchedulerTest.java new file mode 100644 index 0000000..f74bac3 --- /dev/null +++ b/common/src/test/java/com/octoturge/mcmapper/common/ReconciliationSchedulerTest.java @@ -0,0 +1,123 @@ +package com.octoturge.mcmapper.common; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +public class ReconciliationSchedulerTest { + private static int passed = 0; + private static int failed = 0; + + public static void main(String[] args) { + test("empty loaded set returns an empty batch", ReconciliationSchedulerTest::emptySetReturnsEmpty); + test("maxPerSweep of zero (or negative) returns an empty batch", ReconciliationSchedulerTest::zeroMaxReturnsEmpty); + test("a batch is never larger than maxPerSweep", ReconciliationSchedulerTest::batchNeverExceedsMax); + test("maxPerSweep >= set size returns every key exactly once", ReconciliationSchedulerTest::coversWholeSetInOneCallWhenItFits); + test("repeated calls eventually cover every key exactly once per cycle, then wrap", ReconciliationSchedulerTest::rotatesThroughWholeSetWithoutRepeatsWithinACycle); + test("the cursor wraps around the end of the set back to the start", ReconciliationSchedulerTest::wrapsAroundTheEnd); + test("a shrinking loaded set doesn't throw or return stale keys", ReconciliationSchedulerTest::toleratesShrinkingSet); + + System.out.println(); + System.out.println(passed + " passed, " + failed + " failed"); + if (failed > 0) { + System.exit(1); + } + } + + private static void emptySetReturnsEmpty() { + ReconciliationScheduler scheduler = new ReconciliationScheduler(); + List batch = scheduler.next(new ArrayList<>(), 5); + assertEquals(0, batch.size()); + } + + private static void zeroMaxReturnsEmpty() { + ReconciliationScheduler scheduler = new ReconciliationScheduler(); + Set loaded = new LinkedHashSet<>(Arrays.asList(1L, 2L, 3L)); + assertEquals(0, scheduler.next(loaded, 0).size()); + assertEquals(0, scheduler.next(loaded, -1).size()); + } + + private static void batchNeverExceedsMax() { + ReconciliationScheduler scheduler = new ReconciliationScheduler(); + Set loaded = new LinkedHashSet<>(); + for (long i = 0; i < 50; i++) loaded.add(i); + for (int i = 0; i < 10; i++) { + List batch = scheduler.next(loaded, 7); + if (batch.size() > 7) { + throw new AssertionError("batch size " + batch.size() + " exceeds maxPerSweep 7"); + } + } + } + + private static void coversWholeSetInOneCallWhenItFits() { + ReconciliationScheduler scheduler = new ReconciliationScheduler(); + Set loaded = new LinkedHashSet<>(Arrays.asList(10L, 20L, 30L)); + List batch = scheduler.next(loaded, 100); + assertEquals(3, batch.size()); + assertEquals(new LinkedHashSet<>(Arrays.asList(10L, 20L, 30L)), new LinkedHashSet<>(batch)); + } + + private static void rotatesThroughWholeSetWithoutRepeatsWithinACycle() { + ReconciliationScheduler scheduler = new ReconciliationScheduler(); + Set loaded = new LinkedHashSet<>(); + for (long i = 0; i < 10; i++) loaded.add(i); + + Set seenThisCycle = new LinkedHashSet<>(); + // 10 keys, batches of 3 -> 4 calls (3+3+3+1) to complete exactly one cycle. + for (int call = 0; call < 4; call++) { + List batch = scheduler.next(loaded, 3); + for (Long key : batch) { + if (!seenThisCycle.add(key)) { + throw new AssertionError("key " + key + " reconciled twice within a single cycle"); + } + } + } + assertEquals(10, seenThisCycle.size()); + } + + private static void wrapsAroundTheEnd() { + ReconciliationScheduler scheduler = new ReconciliationScheduler(); + Set loaded = new LinkedHashSet<>(Arrays.asList(1L, 2L, 3L, 4L, 5L)); + scheduler.next(loaded, 5); // consumes the whole set in one call — cursor wraps to 0 + List batch = scheduler.next(loaded, 2); // should start a fresh cycle from the top + assertEquals(2, batch.size()); + if (!batch.contains(1L) || !batch.contains(2L)) { + throw new AssertionError("expected the cursor to wrap back to the start of the set, got " + batch); + } + } + + private static void toleratesShrinkingSet() { + ReconciliationScheduler scheduler = new ReconciliationScheduler(); + Set loaded = new LinkedHashSet<>(); + for (long i = 0; i < 20; i++) loaded.add(i); + scheduler.next(loaded, 15); + + Set shrunk = new LinkedHashSet<>(Arrays.asList(0L, 1L, 2L)); + List batch = scheduler.next(shrunk, 15); // must not throw despite cursor now out of range + assertEquals(3, batch.size()); + for (Long key : batch) { + if (!shrunk.contains(key)) { + throw new AssertionError("batch contained a key no longer in the loaded set: " + key); + } + } + } + + private static void test(String name, Runnable body) { + try { + body.run(); + passed++; + System.out.println("PASS " + name); + } catch (AssertionError e) { + failed++; + System.out.println("FAIL " + name + " — " + e.getMessage()); + } + } + + private static void assertEquals(Object expected, Object actual) { + if (expected == null ? actual != null : !expected.equals(actual)) { + throw new AssertionError("expected <" + expected + "> but got <" + actual + ">"); + } + } +} diff --git a/forge-1_12_2/src/main/java/com/octoturge/mcmapper/forge1122/Forge1122ChunkAdapter.java b/forge-1_12_2/src/main/java/com/octoturge/mcmapper/forge1122/Forge1122ChunkAdapter.java index 0a5c85d..14a9c77 100644 --- a/forge-1_12_2/src/main/java/com/octoturge/mcmapper/forge1122/Forge1122ChunkAdapter.java +++ b/forge-1_12_2/src/main/java/com/octoturge/mcmapper/forge1122/Forge1122ChunkAdapter.java @@ -10,6 +10,7 @@ import net.minecraft.world.World; import net.minecraft.world.WorldServer; import net.minecraft.world.chunk.Chunk; import net.minecraft.world.chunk.storage.ExtendedBlockStorage; +import net.minecraft.world.gen.ChunkProviderServer; import net.minecraftforge.common.MinecraftForge; import net.minecraftforge.common.util.BlockSnapshot; import net.minecraftforge.event.world.BlockEvent; @@ -101,6 +102,15 @@ public class Forge1122ChunkAdapter implements ChunkAdapter { MinecraftForge.EVENT_BUS.register(new EventHooks(sink)); } + @Override + public List loadedChunkKeys() { + List keys = new ArrayList<>(); + for (Chunk chunk : ((ChunkProviderServer) world.getChunkProvider()).getLoadedChunks()) { + keys.add((((long) chunk.x) << 32) | (chunk.z & 0xFFFFFFFFL)); + } + return keys; + } + /** * {@code BlockEvent.BreakEvent} fires *before* the block is actually removed (it's * cancellable), so reading world state synchronously inside that handler would see the diff --git a/forge-1_12_2/src/main/java/com/octoturge/mcmapper/forge1122/MCMapperMod.java b/forge-1_12_2/src/main/java/com/octoturge/mcmapper/forge1122/MCMapperMod.java index a90770f..691df9a 100644 --- a/forge-1_12_2/src/main/java/com/octoturge/mcmapper/forge1122/MCMapperMod.java +++ b/forge-1_12_2/src/main/java/com/octoturge/mcmapper/forge1122/MCMapperMod.java @@ -2,6 +2,7 @@ package com.octoturge.mcmapper.forge1122; import com.octoturge.mcmapper.common.ChunkAdapter; import com.octoturge.mcmapper.common.DefaultBackendConnection; +import com.octoturge.mcmapper.common.ReconciliationScheduler; import com.octoturge.mcmapper.common.config.MapperConfig; import com.octoturge.mcmapper.common.protocol.DeltaEvent; import com.octoturge.mcmapper.common.protocol.SectionData; @@ -28,9 +29,11 @@ import java.util.Set; /** * Entry point for the 1.12.2 leaf — the primary/first-implemented target (Enigmatica 2). * Phase 1: connects to the backend over WS, backfills the overworld's loaded chunks, and - * flushes event-driven column deltas on a timer (no periodic reconciliation sweep yet — see - * plan's Phase 7). Phase 2 adds full-section backfill/flush for 3D mesh rendering. Phase 3 adds - * `/mcmapper link` and the two-way chat bridge (see LinkCommand, Forge1122ChatBridge). + * flushes event-driven column deltas on a timer. Phase 2 adds full-section backfill/flush for + * 3D mesh rendering. Phase 3 adds `/mcmapper link` and the two-way chat bridge (see LinkCommand, + * Forge1122ChatBridge). Phase 7 adds a periodic reconciliation sweep (see {@link + * ReconciliationScheduler}) that catches non-event mutations (world-gen, other mods, `/fill`) + * the event-driven hooks in {@link Forge1122ChunkAdapter} never see. */ @Mod(modid = MCMapperMod.MOD_ID, name = "MCMapper", version = MCMapperMod.VERSION) public class MCMapperMod { @@ -45,7 +48,9 @@ public class MCMapperMod { private Forge1122ChunkAdapter adapter; private final List pendingDeltas = Collections.synchronizedList(new ArrayList<>()); private final Set pendingSectionChunks = Collections.synchronizedSet(new LinkedHashSet<>()); + private final ReconciliationScheduler reconciliationScheduler = new ReconciliationScheduler(); private int ticksSinceFlush = 0; + private int ticksSinceReconciliation = 0; @Mod.EventHandler public void preInit(FMLPreInitializationEvent event) { @@ -62,6 +67,14 @@ public class MCMapperMod { config.deltaFlushIntervalTicks = forgeConfig.getInt("deltaFlushIntervalTicks", "network", config.deltaFlushIntervalTicks, 1, 20 * 60, "How often (in ticks) to batch and flush block-change deltas to the backend"); + config.reconciliationIntervalTicks = forgeConfig.getInt("reconciliationIntervalTicks", "network", + config.reconciliationIntervalTicks, 20, 20 * 60 * 60, + "How often (in ticks) to run a reconciliation sweep, re-reading and resending a " + + "rotating slice of loaded chunks to catch changes event hooks miss " + + "(world-gen, other mods, /fill, etc.)"); + config.reconciliationChunksPerSweep = forgeConfig.getInt("reconciliationChunksPerSweep", "network", + config.reconciliationChunksPerSweep, 1, 5000, + "Max chunks to re-read and resend per reconciliation sweep"); if (forgeConfig.hasChanged()) forgeConfig.save(); } @@ -107,9 +120,32 @@ public class MCMapperMod { @SubscribeEvent public void onServerTick(TickEvent.ServerTickEvent event) { if (event.phase != TickEvent.Phase.END) return; - if (++ticksSinceFlush < config.deltaFlushIntervalTicks) return; - ticksSinceFlush = 0; - flush(); + + if (++ticksSinceFlush >= config.deltaFlushIntervalTicks) { + ticksSinceFlush = 0; + flush(); + } + + if (++ticksSinceReconciliation >= config.reconciliationIntervalTicks) { + ticksSinceReconciliation = 0; + reconcile(); + } + } + + private void reconcile() { + List chunkKeys = reconciliationScheduler.next(adapter.loadedChunkKeys(), + config.reconciliationChunksPerSweep); + if (chunkKeys.isEmpty()) return; + + String dimensionId = adapter.getDimensionId(); + for (long key : chunkKeys) { + int chunkX = (int) (key >> 32); + int chunkZ = (int) key; + List columns = adapter.readChunk(dimensionId, chunkX, chunkZ); + if (!columns.isEmpty()) connection.sendDeltas(columns); + List sections = adapter.readSections(chunkX, chunkZ); + if (!sections.isEmpty()) connection.sendSections(dimensionId, chunkX, chunkZ, sections); + } } private void flush() {