fix(events): coalesce in-flight feature intent replay
Release / release (push) Failing after 19m41s

This commit is contained in:
dmg committed 2026-09-27 10:53:50 -04:00
1 parent 9bb56f3a1c
commit e04a3152c1
4 files changed
+122 -9

No files matched your search

@@ -33,6 +33,7 @@ final class FeatureEventOutbox implements AutoCloseable {
private final ExecutorService worker = Executors.newSingleThreadExecutor(task ->
new Thread(task, "fantasy-feature-event-intents"));
private final CompletableFuture<Void> drained = new CompletableFuture<>();
private final Set<UUID> inFlight = java.util.concurrent.ConcurrentHashMap.newKeySet();
FeatureEventOutbox(Path data, String folder, Set<String> types,
Function<EventInput, CompletionStage<LocalEventLog.Delivery>> sink,
@@ -239,12 +240,18 @@ final class FeatureEventOutbox implements AutoCloseable {
}
private void publish(EventInput event) {
if (!inFlight.add(event.id())) return;
try { sink.apply(event).whenComplete((delivery, failure) -> {
if (failure != null) { report(failure); return; }
if (delivery == LocalEventLog.Delivery.REJECTED) return;
try { if (Files.deleteIfExists(path(event))) forceDirectory(); }
catch (IOException error) { report(error); }
}); } catch (RuntimeException failure) { report(failure); }
try {
if (failure != null) { report(failure); return; }
if (delivery == LocalEventLog.Delivery.REJECTED) return;
try { if (Files.deleteIfExists(path(event))) forceDirectory(); }
catch (IOException error) { report(error); }
} finally { inFlight.remove(event.id()); }
}); } catch (RuntimeException failure) {
inFlight.remove(event.id());
report(failure);
}
}
private void forceDirectory() throws IOException {
@@ -398,7 +398,9 @@ class AdventureQuestNativeTest {
}
var earned=exportedCampEvents(data).stream().filter(event->event.get("type").getAsString()
.endsWith("adventure.trophy_earned")).toList();
assertEquals(1,earned.size());
assertEquals(1,earned.size(),()->"Exported camp earned count="+earned.size()+" records="
+earned.stream().map(event->"id="+event.get("id").getAsString()+",receipt="
+event.getAsJsonObject("data").get("receipt_id").getAsString()).toList());
var details=earned.getFirst().getAsJsonObject("data");
assertEquals(receiptId.toString(),details.get("receipt_id").getAsString());
assertEquals(finalEnemy.toString(),details.get("final_enemy_id").getAsString());
@@ -782,13 +784,21 @@ class AdventureQuestNativeTest {
private static void awaitCampExport(FantasyStickLifecycleNativeTest.Fixture fixture,Path root,
UUID receipt)throws Exception{
var eventId=UUID.nameUUIDFromBytes(("fantasystick:fishing:"+receipt
+":adventure.trophy_earned").getBytes(java.nio.charset.StandardCharsets.UTF_8));
var ownerIntent=root.resolve("fishing-event-intents").resolve(eventId+".pending");
var logIntent=root.resolve("events/pending").resolve(eventId+".pending");
for(int attempt=0;attempt<200;attempt++){
if(hasExportedCampReceipt(root,receipt))return;
if(hasExportedCampReceipt(root,receipt)&&java.nio.file.Files.notExists(ownerIntent)
&&java.nio.file.Files.notExists(logIntent))return;
var task=fixture.tasks.poll(25,java.util.concurrent.TimeUnit.MILLISECONDS);
if(task!=null)task.run();
}
assertTrue(hasExportedCampReceipt(root,receipt),
"Committed camp receipt "+receipt+" did not reach exported JSONL before shutdown");
assertTrue(hasExportedCampReceipt(root,receipt)
&&java.nio.file.Files.notExists(ownerIntent)&&java.nio.file.Files.notExists(logIntent),
"Committed camp receipt "+receipt+" did not complete exported JSONL/intent handoff before shutdown"
+"; owner_pending="+java.nio.file.Files.exists(ownerIntent)
+"; log_pending="+java.nio.file.Files.exists(logIntent));
}
private static boolean hasExportedCampReceipt(Path root,UUID receipt){
try{return exportedCampEvents(root).stream().anyMatch(event->event.get("type").getAsString()
@@ -23,6 +23,52 @@ class FishingEventBridgePartsNativeTest {
@org.junit.jupiter.api.BeforeAll static void bootstrap() throws Exception { NativeRuntime.bootstrap(); }
@TempDir Path data;
@Test void campStartupReplayDoesNotRepublishAnAlreadyExportedUnacknowledgedIntent() throws Exception {
var character=UUID.randomUUID();var receiptId=UUID.randomUUID();
var receipt=new games.dmg.fantasystick.adventure.FishingState.Trophy(receiptId,
games.dmg.fantasystick.adventure.FishingState.TrophyType.CAMP,character,
UUID.randomUUID(),UUID.randomUUID(),UUID.randomUUID(),UUID.randomUUID(),
"Spoils of Restart Camp","Restart Camp",Map.of(character,
new games.dmg.fantasystick.adventure.FishingState.Reward(
java.math.BigInteger.ONE,java.math.BigInteger.ZERO)),
games.dmg.fantasystick.adventure.FishingState.TrophyPhase.EARNED,
Map.of("final_enemy_id",UUID.randomUUID().toString(),"unit_kind","solo",
"unit_id",character.toString(),"accepted_roster",character.toString(),
"arena_shape","circle","arena_x_blocks","10","arena_z_blocks","10"));
var log=LocalEventLog.open(data,Clock.systemUTC(),
new LocalEventLog.RuntimeInfo("test","test","test"),128,1_000_000,ignored->{});
log.ready().toCompletableFuture().get(5,TimeUnit.SECONDS);
var allowFirstAcknowledgement=new CompletableFuture<Void>();
var firstExported=new CompletableFuture<Void>();
var deliveries=new java.util.concurrent.atomic.AtomicInteger();
var bridge=new FishingEventBridge(event->{
var result=log.emit(event);
if(deliveries.incrementAndGet()==1)return result.thenCompose(status->{
firstExported.complete(null);
return allowFirstAcknowledgement.thenApply(ignored->status);
});
return result;
},Clock.systemUTC(),data,mock(EquipmentSessions.class),ignored->{});
bridge.trophyEarned(receipt);
firstExported.get(5,TimeUnit.SECONDS);
bridge.replay();
bridge.close();bridge.drained().toCompletableFuture().get(5,TimeUnit.SECONDS);
allowFirstAcknowledgement.complete(null);
log.close();
var earned=new ArrayList<com.google.gson.JsonObject>();
try(var files=java.nio.file.Files.walk(data.resolve("events"))){
for(var path:files.filter(value->value.toString().endsWith(".jsonl")).toList())
for(var line:java.nio.file.Files.readAllLines(path)){
var event=com.google.gson.JsonParser.parseString(line).getAsJsonObject();
if(event.get("type").getAsString().endsWith("adventure.trophy_earned"))earned.add(event);
}
}
assertEquals(1,earned.size(),()->"camp receipt="+receiptId+" exported IDs="
+earned.stream().map(event->event.get("id").getAsString()).toList());
assertEquals(receiptId.toString(),earned.getFirst().getAsJsonObject("data")
.get("receipt_id").getAsString());
}
@Test void rejectedChampionIntentReplaysFromFrozenReceiptIntoExportedJsonl() throws Exception {
var carrier=UUID.randomUUID();var zero=UUID.randomUUID();var receiptId=UUID.randomUUID();
var receipt=new games.dmg.fantasystick.adventure.FishingState.Trophy(receiptId,
@@ -17,6 +17,56 @@ import org.junit.jupiter.api.io.TempDir;
class FeatureEventOutboxCollectionsTest {
@TempDir Path data;
@Test void rejectedInFlightIntentCanReplayWithItsOriginalIdAndDifferentIdsStillDeliver() throws Exception {
var first=new EventInput(UUID.randomUUID(),UUID.randomUUID(),Instant.parse("2026-09-26T00:00:00Z"),
"adventure.trophy_earned",null,Map.of("receipt_id",UUID.randomUUID().toString()));
var other=new EventInput(UUID.randomUUID(),UUID.randomUUID(),Instant.parse("2026-09-26T00:00:01Z"),
"adventure.trophy_earned",null,Map.of("receipt_id",UUID.randomUUID().toString()));
var refused=new CompletableFuture<LocalEventLog.Delivery>();
var initial=new java.util.concurrent.CountDownLatch(2);
var deliveries=new java.util.concurrent.CopyOnWriteArrayList<EventInput>();
var failures=new ArrayList<Throwable>();
var writer=new FeatureEventOutbox(data,"fishing-event-intents",
Set.of("adventure.trophy_earned"),event->{
deliveries.add(event);
if(deliveries.size()<=2)initial.countDown();
return event.id().equals(first.id())&&deliveries.stream()
.filter(value->value.id().equals(first.id())).count()==1
?refused:CompletableFuture.completedFuture(LocalEventLog.Delivery.EXPORTED);
},failures::add);
writer.accepted(first);
writer.accepted(other);
assertTrue(initial.await(5,TimeUnit.SECONDS));
assertEquals(List.of(first,other),deliveries);
refused.complete(LocalEventLog.Delivery.REJECTED);
assertTrue(java.nio.file.Files.exists(data.resolve("fishing-event-intents")
.resolve(first.id()+".pending")));
writer.replay();writer.close();writer.drained().toCompletableFuture().get(5,TimeUnit.SECONDS);
assertEquals(List.of(first,other,first),deliveries);
assertTrue(failures.isEmpty(),failures.toString());
}
@Test void failedInFlightIntentRemainsRetryableWithItsOriginalId() throws Exception {
var original=new EventInput(UUID.randomUUID(),UUID.randomUUID(),
Instant.parse("2026-09-26T00:00:00Z"),"adventure.trophy_earned",null,
Map.of("receipt_id",UUID.randomUUID().toString()));
var deliveries=new java.util.concurrent.CopyOnWriteArrayList<EventInput>();
var failures=new java.util.concurrent.CopyOnWriteArrayList<Throwable>();
var writer=new FeatureEventOutbox(data,"fishing-event-intents",
Set.of("adventure.trophy_earned"),event->{
deliveries.add(event);
return deliveries.size()==1
?CompletableFuture.failedFuture(new java.io.IOException("sink unavailable"))
:CompletableFuture.completedFuture(LocalEventLog.Delivery.EXPORTED);
},failures::add);
writer.accepted(original);
writer.replay();writer.close();writer.drained().toCompletableFuture().get(5,TimeUnit.SECONDS);
assertEquals(List.of(original,original),deliveries);
assertEquals(1,failures.size());
assertFalse(java.nio.file.Files.exists(data.resolve("fishing-event-intents")
.resolve(original.id()+".pending")));
}
@Test void durableFishingRosterAndAllocationsReplayAsStructuredJsonValues() throws Exception {
var original = new EventInput(UUID.randomUUID(), UUID.randomUUID(), Instant.parse("2026-09-26T00:00:00Z"),
"adventure.items_submitted", null, Map.of("roster", List.of("first", "second"),