Merge pull request 'fix(events): recover partial appends without restarting' (#11) from us-092-write-local-character-and-system-events into main
Release / release (push) Successful in 1m30s
Release / release (push) Successful in 1m30s
This commit was merged in pull request #11.
This commit is contained in:
commit
7353436a77
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); }
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user