From a19909f1525d43a664a2bd1db526e46e5a3fc6ff Mon Sep 17 00:00:00 2001 From: Dylan Garvis Date: Thu, 8 Oct 2026 09:26:27 -0400 Subject: [PATCH] fix(events): recover partial appends without restarting --- .../fantasystick/events/LocalEventLog.java | 57 ++- .../LocalEventPartialWriteNativeTest.java | 359 ++++++++++++++++++ 2 files changed, 402 insertions(+), 14 deletions(-) create mode 100644 src/nativeTest/java/games/dmg/fantasystick/LocalEventPartialWriteNativeTest.java diff --git a/src/main/java/games/dmg/fantasystick/events/LocalEventLog.java b/src/main/java/games/dmg/fantasystick/events/LocalEventLog.java index ce734f6..ac93f4a 100644 --- a/src/main/java/games/dmg/fantasystick/events/LocalEventLog.java +++ b/src/main/java/games/dmg/fantasystick/events/LocalEventLog.java @@ -60,6 +60,7 @@ public final class LocalEventLog implements AutoCloseable { private final java.nio.channels.FileLock writerLock; private long sequence; private Path segment; + private Path damagedSegment; private LocalDate segmentDate; private long segmentBytes; private int segmentNumber; @@ -203,24 +204,40 @@ public final class LocalEventLog implements AutoCloseable { } boolean recovered = false; for (var path : files) { - if (Files.isSymbolicLink(path)) throw new IOException("Symbolic link in Fantasy Stick event segments"); - var bytes = Files.readAllBytes(path); - int prefix=0,cursor=0;while(cursor files; try (var stream = Files.list(pending)) { @@ -247,6 +264,10 @@ public final class LocalEventLog implements AutoCloseable { } private void append(byte[] line) throws IOException { + if (damagedSegment != null) { + recoverInterruptedTail(damagedSegment); + damagedSegment = null; + } var date = LocalDate.ofInstant(clock.instant(), ZoneOffset.UTC); if (segment == null || !date.equals(segmentDate) || segmentBytes > 0 && segmentBytes + line.length > maximumSegmentBytes) { @@ -256,6 +277,14 @@ public final class LocalEventLog implements AutoCloseable { var buffer = ByteBuffer.wrap(line); while (buffer.hasRemaining()) channel.write(buffer); channel.force(true); + } catch (IOException failure) { + // Even open/force/close failures can leave uncertain bytes. Never reuse this + // segment or acknowledge pending work until its tail has been repaired. + damagedSegment = segment; + segment = null; + segmentDate = null; + segmentBytes = 0; + throw failure; } segmentBytes += line.length; } diff --git a/src/nativeTest/java/games/dmg/fantasystick/LocalEventPartialWriteNativeTest.java b/src/nativeTest/java/games/dmg/fantasystick/LocalEventPartialWriteNativeTest.java new file mode 100644 index 0000000..599b75e --- /dev/null +++ b/src/nativeTest/java/games/dmg/fantasystick/LocalEventPartialWriteNativeTest.java @@ -0,0 +1,359 @@ +package games.dmg.fantasystick; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import games.dmg.fantasystick.events.EventInput; +import games.dmg.fantasystick.events.LocalEventLog; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.MockedStatic; + +/** Real file bytes and owned writer callbacks; only the append channel's failure is injected. */ +final class LocalEventPartialWriteNativeTest { + @TempDir Path data; + @BeforeAll static void bootstrap() throws Exception { NativeRuntime.bootstrap(); } + + @Test void sameLifetimeRetryPreservesPrefixAndExportsExactPendingRecordAfterPartialWrite() throws Exception { + var clock = Clock.fixed(Instant.parse("2026-09-26T12:00:00Z"), ZoneOffset.UTC); + var diagnostics = new ArrayList(); + try (var log = LocalEventLog.open(data, clock, + new LocalEventLog.RuntimeInfo("Season 4", "26.2", "test"), 8, 134_217_728L, diagnostics::add)) { + assertEquals(LocalEventLog.Delivery.EXPORTED, log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS)); + var original = segments().getFirst(); + var prefix = Files.readAllBytes(original); + var input = event(clock, "original\nUnicode ☃"); + var pending = data.resolve("events/pending/" + input.id() + ".pending"); + byte[] exact; + try (var fault = new AppendFault(log)) { + fault.failNext(1); + assertEquals(LocalEventLog.Delivery.DURABLE_PENDING, + log.emit(input).toCompletableFuture().get(5, TimeUnit.SECONDS)); + exact = Files.readAllBytes(pending); + assertEquals(1, fault.failures.get()); + assertArrayEquals(concat(prefix, Arrays.copyOf(exact, AppendFault.PARTIAL_BYTES)), + Files.readAllBytes(original), "the injected fault must really persist a partial record"); + assertEquals(LocalEventLog.Delivery.EXPORTED, + log.emit(event(clock, "next")).toCompletableFuture().get(5, TimeUnit.SECONDS)); + } + assertFalse(Files.exists(pending)); + var rows = exported(); // Strict JSON parser: concatenating the pending record onto its fragment is RED. + var recovered = rows.stream().filter(row -> input.id().toString().equals(row.get("id").getAsString())).toList(); + assertEquals(1, recovered.size()); + assertEquals(JsonParser.parseString(new String(exact, java.nio.charset.StandardCharsets.UTF_8)), recovered.getFirst()); + assertTrue(Files.readAllBytes(original).length >= prefix.length); + assertArrayEquals(prefix, Arrays.copyOf(Files.readAllBytes(original), prefix.length)); + assertTrue(corruptEvidence().stream().anyMatch(bytes -> + Arrays.equals(Arrays.copyOf(exact, AppendFault.PARTIAL_BYTES), bytes))); + assertTrue(diagnostics.stream().anyMatch(message -> message.contains("durable pending"))); + } + exported(); + } + + @Test void partialWriteDuringPendingReplayRetainsExactProofUntilASecondRecoverySucceeds() throws Exception { + var clock = Clock.fixed(Instant.parse("2026-09-26T12:00:00Z"), ZoneOffset.UTC); + try (var log = LocalEventLog.open(data, clock, + new LocalEventLog.RuntimeInfo("Season 4", "26.2", "test"), 8, 134_217_728L, ignored -> { })) { + log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS); + var input = event(clock, "pending replay"); + var pending = data.resolve("events/pending/" + input.id() + ".pending"); + try (var fault = new AppendFault(log)) { + fault.failNext(2); + assertEquals(LocalEventLog.Delivery.DURABLE_PENDING, + log.emit(input).toCompletableFuture().get(5, TimeUnit.SECONDS)); + var exact = Files.readAllBytes(pending); + var unaccepted = event(clock, "not yet accepted"); + assertEquals(LocalEventLog.Delivery.REJECTED, + log.emit(unaccepted).toCompletableFuture().get(5, TimeUnit.SECONDS)); + assertArrayEquals(exact, Files.readAllBytes(pending)); + assertEquals(2, fault.failures.get()); + assertEquals(LocalEventLog.Delivery.EXPORTED, + log.emit(event(clock, "retry")).toCompletableFuture().get(5, TimeUnit.SECONDS)); + assertFalse(Files.exists(pending)); + var rows = exported(); + assertEquals(List.of(JsonParser.parseString(new String(exact, java.nio.charset.StandardCharsets.UTF_8))), + rows.stream().filter(row -> input.id().toString().equals(row.get("id").getAsString())).toList()); + assertTrue(rows.stream().noneMatch(row -> unaccepted.id().toString().equals(row.get("id").getAsString()))); + assertEquals(2, corruptEvidence().stream().filter(bytes -> + Arrays.equals(Arrays.copyOf(exact, AppendFault.PARTIAL_BYTES), bytes)).count()); + } + } + } + + @Test void failedQuarantineForceCannotTruncateDamageDeletePendingOrAcknowledgeReplay() throws Exception { + var clock = Clock.fixed(Instant.parse("2026-09-26T12:00:00Z"), ZoneOffset.UTC); + try (var log = LocalEventLog.open(data, clock, + new LocalEventLog.RuntimeInfo("Season 4", "26.2", "test"), 8, 134_217_728L, ignored -> { })) { + log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS); + var original = segments().getFirst(); + var input = event(clock, "quarantine failure"); + var pending = data.resolve("events/pending/" + input.id() + ".pending"); + try (var fault = new AppendFault(log)) { + fault.failNext(1); + assertEquals(LocalEventLog.Delivery.DURABLE_PENDING, + log.emit(input).toCompletableFuture().get(5, TimeUnit.SECONDS)); + var exact = Files.readAllBytes(pending); + var damaged = Files.readAllBytes(original); + fault.quarantineForceFailures.set(1); + assertEquals(LocalEventLog.Delivery.REJECTED, + log.emit(event(clock, "repair unavailable")).toCompletableFuture().get(5, TimeUnit.SECONDS)); + assertEquals(1, fault.failedQuarantines.get()); + assertArrayEquals(damaged, Files.readAllBytes(original), "unforced quarantine must not authorize truncation"); + assertArrayEquals(exact, Files.readAllBytes(pending)); + assertEquals(LocalEventLog.Delivery.EXPORTED, + log.emit(event(clock, "repair available")).toCompletableFuture().get(5, TimeUnit.SECONDS)); + assertFalse(Files.exists(pending)); + assertEquals(1, exported().stream().filter(row -> input.id().toString().equals(row.get("id").getAsString())).count()); + assertTrue(corruptEvidence().stream().anyMatch(bytes -> + Arrays.equals(Arrays.copyOf(exact, AppendFault.PARTIAL_BYTES), bytes))); + } + } + } + + @Test void completeWriteWithFailedForceRetainsPendingAndPermitsOnlyIdenticalAtLeastOnceDuplicates() throws Exception { + var clock = Clock.fixed(Instant.parse("2026-09-26T12:00:00Z"), ZoneOffset.UTC); + try (var log = LocalEventLog.open(data, clock, + new LocalEventLog.RuntimeInfo("Season 4", "26.2", "test"), 8, 134_217_728L, ignored -> { })) { + log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS); + var input = event(clock, "uncertain force"); + var pending = data.resolve("events/pending/" + input.id() + ".pending"); + try (var fault = new AppendFault(log)) { + fault.appendForceFailures.set(1); + assertEquals(LocalEventLog.Delivery.DURABLE_PENDING, + log.emit(input).toCompletableFuture().get(5, TimeUnit.SECONDS)); + var exact = Files.readAllBytes(pending); + assertEquals(1, fault.failedAppendForces.get()); + assertEquals(LocalEventLog.Delivery.EXPORTED, + log.emit(event(clock, "retry")).toCompletableFuture().get(5, TimeUnit.SECONDS)); + var repeated = exported().stream().filter(row -> input.id().toString().equals(row.get("id").getAsString())).toList(); + assertEquals(2, repeated.size(), "an uncertain completed append may repeat, never promise exactly once"); + for (var row : repeated) assertEquals(JsonParser.parseString(new String(exact, java.nio.charset.StandardCharsets.UTF_8)), row); + assertFalse(Files.exists(pending)); + assertTrue(corruptEvidence().isEmpty(), "complete valid records must not be quarantined or rewritten"); + } + } + } + + @Test void registeredCancelledNativeActionRecoversItsOriginalContextWithoutRestartOrGameplayMutation() throws Exception { + try (var fixture = new FantasyStickLifecycleNativeTest.Fixture(data); + var nativeFailures = new NativeFailures(fixture)) { + fixture.plugin.setEnabled(true); + var actor = new FantasyStickLifecycleNativeTest.Actor(data); + fixture.players.put(actor.core.id, actor.player); + fixture.activate(actor); + doReturn(null).when(actor.player).getFishHook(); // No hook exists for this ordinary held-slot event. + var log = (LocalEventLog) FantasyStickLifecycleNativeTest.Fixture.field(fixture.plugin, "eventLog"); + log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS); + var before = actor.core.snapshot(); + var event = new org.bukkit.event.player.PlayerItemHeldEvent(actor.player, 0, 1); + event.setCancelled(true); + try (var fault = new AppendFault(log)) { + fault.onlyContaining = "\"entry_point\":\"PlayerItemHeldEvent\""; + fault.failNext(1); + fixture.dispatcher.accept(event); + fault.worker.submit(() -> { }).get(5, TimeUnit.SECONDS); // Both real registered capture stages settle. + assertEquals(1, fault.failures.get()); + assertTrue(event.isCancelled()); + assertArrayEquals(before, actor.core.snapshot()); + var exact = JsonParser.parseString(new String(fault.failedLine, java.nio.charset.StandardCharsets.UTF_8)).getAsJsonObject(); + assertTrue(exact.get("type").getAsString().endsWith("action.observed")); + assertEquals(actor.core.id.toString(), exact.get("playerid").getAsString()); + assertEquals(actor.core.record.id().toString(), exact.get("characterid").getAsString()); + var correlated = fault.worker.submit(this::exported).get(5, TimeUnit.SECONDS).stream() + .filter(row -> exact.getAsJsonObject("data").get("action_id") + .equals(row.getAsJsonObject("data").get("action_id"))).toList(); + assertEquals(2, correlated.size(), "one observation and one resolution, never a duplicated success"); + assertEquals(List.of(exact), correlated.stream().filter(row -> row.get("id").equals(exact.get("id"))).toList()); + var resolved = correlated.stream().filter(row -> row.get("type").getAsString().endsWith("action.resolved")) + .findFirst().orElseThrow().getAsJsonObject("data"); + assertTrue(resolved.get("cancelled").getAsBoolean()); + assertEquals("cancelled", resolved.get("outcome").getAsString()); + nativeFailures.assertClean(); + assertFalse(Files.exists(data.resolve("events/pending/" + exact.get("id").getAsString() + ".pending"))); + assertTrue(corruptEvidence().stream().anyMatch(bytes -> + Arrays.equals(Arrays.copyOf(fault.failedLine, AppendFault.PARTIAL_BYTES), bytes))); + } + } + assertEquals(1, exported().stream().filter(row -> row.get("type").getAsString().endsWith("game.stopped")).count()); + } + + private static EventInput event(Clock clock, String marker) { + return new EventInput(UUID.randomUUID(), UUID.randomUUID(), clock.instant(), "action.observed", null, + Map.of("action_kind", "partial_write_probe", "marker", marker, "outcome", "observed")); + } + + private List segments() throws IOException { + try (var files = Files.walk(data.resolve("events"))) { + return files.filter(path -> path.getFileName().toString().endsWith(".jsonl")).sorted().toList(); + } + } + + private List exported() throws IOException { + var rows = new ArrayList(); + for (var path : segments()) { + var bytes = Files.readAllBytes(path); + if (bytes.length == 0) continue; + assertEquals('\n', bytes[bytes.length - 1], "every exported record must be LF terminated"); + for (var line : Files.readAllLines(path)) rows.add(JsonParser.parseString(line).getAsJsonObject()); + } + return rows; + } + + private List corruptEvidence() throws IOException { + var evidence = new ArrayList(); + try (var files = Files.walk(data.resolve("events"))) { + for (var path : files.filter(path -> path.getFileName().toString().contains(".corrupt-")).toList()) { + assertFalse(path.getFileName().toString().endsWith(".jsonl")); + evidence.add(Files.readAllBytes(path)); + } + } + return evidence; + } + + private static byte[] concat(byte[] first, byte[] second) { + var bytes = Arrays.copyOf(first, first.length + second.length); + System.arraycopy(second, 0, bytes, first.length, second.length); + return bytes; + } + + /** Capture, never suppress, failures from this fixture's native callback boundaries. */ + private static final class NativeFailures implements AutoCloseable { + private final java.util.logging.Logger server; + private final java.util.logging.Logger plugin; + private final List failures = new java.util.concurrent.CopyOnWriteArrayList<>(); + private final java.util.logging.Handler handler = new java.util.logging.Handler() { + @Override public void publish(java.util.logging.LogRecord record) { + if (record.getLevel().intValue() >= java.util.logging.Level.SEVERE.intValue()) failures.add(record); + } + @Override public void flush() { } + @Override public void close() { } + }; + private NativeFailures(FantasyStickLifecycleNativeTest.Fixture fixture) { + server = fixture.server.getLogger(); + plugin = fixture.plugin.getLogger(); + server.addHandler(handler); + if (plugin != server) plugin.addHandler(handler); + } + private void assertClean() { + assertTrue(failures.isEmpty(), () -> failures.stream() + .map(record -> record.getMessage() + ": " + record.getThrown()).toList().toString()); + } + @Override public void close() { + server.removeHandler(handler); + if (plugin != server) plugin.removeHandler(handler); + assertClean(); + } + } + + /** Scoped to this log's serial worker, never installed globally or on the native/main thread. */ + private static final class AppendFault implements AutoCloseable { + private static final int PARTIAL_BYTES = 47; + private final ExecutorService worker; + private final MockedStatic channels; + private final AtomicInteger remaining = new AtomicInteger(); + private final AtomicInteger failures = new AtomicInteger(); + private final AtomicInteger appendForceFailures = new AtomicInteger(); + private final AtomicInteger failedAppendForces = new AtomicInteger(); + private final AtomicInteger quarantineForceFailures = new AtomicInteger(); + private final AtomicInteger failedQuarantines = new AtomicInteger(); + private volatile String onlyContaining; + private volatile byte[] failedLine; + + private AppendFault(LocalEventLog log) throws Exception { + worker = (ExecutorService) FantasyStickLifecycleNativeTest.Fixture.field(log, "worker"); + channels = worker.submit(() -> { + var scoped = mockStatic(FileChannel.class, CALLS_REAL_METHODS); + try { + scoped.when(() -> FileChannel.open(any(Path.class), eq(StandardOpenOption.WRITE), + eq(StandardOpenOption.APPEND))).thenAnswer(call -> { + if (remaining.get() == 0 && appendForceFailures.get() == 0) return call.callRealMethod(); + Path path = call.getArgument(0); + var real = path.getFileSystem().provider().newFileChannel(path, + Set.of(StandardOpenOption.WRITE, StandardOpenOption.APPEND)); + var failing = mock(FileChannel.class); + doAnswer(write -> { + ByteBuffer source = write.getArgument(0); + var complete = new byte[source.remaining()]; + source.duplicate().get(complete); + if ((onlyContaining == null || new String(complete, java.nio.charset.StandardCharsets.UTF_8) + .contains(onlyContaining)) && take(remaining)) { + failedLine = complete; + var partial = source.slice(); + partial.limit(Math.min(PARTIAL_BYTES, partial.remaining())); + while (partial.hasRemaining()) real.write(partial); + real.force(true); + failures.incrementAndGet(); + throw new IOException("controlled partial append"); + } + return real.write(source); + }).when(failing).write(any(ByteBuffer.class)); + doAnswer(force -> { + if (take(appendForceFailures)) { + failedAppendForces.incrementAndGet(); + throw new IOException("controlled append force failure"); + } + real.force(force.getArgument(0)); + return null; + }).when(failing).force(anyBoolean()); + doAnswer(close -> { real.close(); return null; }).when(failing).close(); + return failing; + }); + scoped.when(() -> FileChannel.open(any(Path.class), eq(StandardOpenOption.CREATE_NEW), + eq(StandardOpenOption.WRITE))).thenAnswer(call -> { + Path path = call.getArgument(0); + if (!path.getFileName().toString().contains(".corrupt-") || !take(quarantineForceFailures)) + return call.callRealMethod(); + var real = path.getFileSystem().provider().newFileChannel(path, + Set.of(StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)); + var failing = mock(FileChannel.class); + doAnswer(write -> real.write((ByteBuffer) write.getArgument(0))).when(failing).write(any(ByteBuffer.class)); + doAnswer(force -> { + failedQuarantines.incrementAndGet(); + throw new IOException("controlled quarantine force failure"); + }).when(failing).force(anyBoolean()); + doAnswer(close -> { real.close(); return null; }).when(failing).close(); + return failing; + }); + return scoped; + } catch (Throwable failure) { + scoped.close(); + throw failure; + } + }).get(5, TimeUnit.SECONDS); + } + + private static boolean take(AtomicInteger count) { + return count.getAndUpdate(value -> Math.max(0, value - 1)) > 0; + } + + private void failNext(int count) { remaining.set(count); } + + @Override public void close() { + try { worker.submit(channels::close).get(5, TimeUnit.SECONDS); } + catch (InterruptedException interrupted) { Thread.currentThread().interrupt(); throw new AssertionError(interrupted); } + catch (Exception failure) { throw new AssertionError(failure); } + } + } +}