fix(events): recover partial appends without restarting
CI / build (push) Successful in 43s
CI / build (pull_request) Successful in 1m1s

This commit is contained in:
dmg committed 2026-10-08 09:26:27 -04:00
1 parent 7b6ea939a5
commit a19909f152
2 files changed
+402 -14

No files matched your search

@@ -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<bytes.length){int end=cursor;while(end<bytes.length&&bytes[end]!='\n')end++;if(end==bytes.length)break;var line=java.util.Arrays.copyOfRange(bytes,cursor,end+1);if(!validJsonObjectLine(line))break;prefix=end+1;cursor=end+1;}if(prefix==bytes.length)continue;
recovered = true;
var quarantine = path.resolveSibling(path.getFileName() + ".corrupt-" + UUID.randomUUID());
try (var channel = FileChannel.open(quarantine, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
var tail = ByteBuffer.wrap(bytes, prefix, bytes.length - prefix);
while (tail.hasRemaining()) channel.write(tail);
channel.force(true);
}
try (var channel = FileChannel.open(path, StandardOpenOption.WRITE)) {
channel.truncate(prefix);
channel.force(true);
}
recovered |= recoverInterruptedTail(path);
}
return recovered;
}
/** Quarantine must be forced before truncation; failure leaves pending work and repair state intact. */
private boolean recoverInterruptedTail(Path path) throws IOException {
if (Files.isSymbolicLink(path)) throw new IOException("Symbolic link in Fantasy Stick event segments");
var bytes = Files.readAllBytes(path);
int prefix = 0;
int cursor = 0;
while (cursor < bytes.length) {
int end = cursor;
while (end < bytes.length && bytes[end] != '\n') end++;
if (end == bytes.length) break;
var line = java.util.Arrays.copyOfRange(bytes, cursor, end + 1);
if (!validJsonObjectLine(line)) break;
prefix = end + 1;
cursor = end + 1;
}
if (prefix == bytes.length) return false;
var quarantine = path.resolveSibling(path.getFileName() + ".corrupt-" + UUID.randomUUID());
try (var channel = FileChannel.open(quarantine, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
var tail = ByteBuffer.wrap(bytes, prefix, bytes.length - prefix);
while (tail.hasRemaining()) channel.write(tail);
channel.force(true);
}
try (var channel = FileChannel.open(path, StandardOpenOption.WRITE)) {
channel.truncate(prefix);
channel.force(true);
}
return true;
}
private int recoverPending() throws IOException {
List<Path> 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;
}
@@ -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<String>();
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<Path> segments() throws IOException {
try (var files = Files.walk(data.resolve("events"))) {
return files.filter(path -> path.getFileName().toString().endsWith(".jsonl")).sorted().toList();
}
}
private List<JsonObject> exported() throws IOException {
var rows = new ArrayList<JsonObject>();
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<byte[]> corruptEvidence() throws IOException {
var evidence = new ArrayList<byte[]>();
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<java.util.logging.LogRecord> 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<FileChannel> 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); }
}
}
}