|
|
|
@@ -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); }
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
dmg