diff --git a/src/main/java/com/lion/lionwebsite/Dao/normal/SubscriptionEventMapper.java b/src/main/java/com/lion/lionwebsite/Dao/normal/SubscriptionEventMapper.java index 2420691..54a1772 100644 --- a/src/main/java/com/lion/lionwebsite/Dao/normal/SubscriptionEventMapper.java +++ b/src/main/java/com/lion/lionwebsite/Dao/normal/SubscriptionEventMapper.java @@ -7,6 +7,10 @@ import org.apache.ibatis.annotations.*; @Mapper public interface SubscriptionEventMapper { + /** Reserve SQLite's writer before the receipt read; a deferred read-to-write upgrade can fail immediately. */ + @Update("update subscription_event_receipt set ingested_at=ingested_at where 0") + void reserveWriter(); + @Select("select fingerprint from subscription_event_receipt where event_id=#{eventId}") String fingerprint(String eventId); diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionEventService.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionEventService.java index 7e8f320..44fda3c 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubscriptionEventService.java +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionEventService.java @@ -123,6 +123,7 @@ public class SubscriptionEventService { try { // Return acknowledgements only after execute has returned and the entire transaction committed. List acknowledged = transactions.execute(status -> { + events.reserveWriter(); List ids = new ArrayList<>(); for (SubscriptionEvent event : batch) if (persist(event, "standby", null)) ids.add(event.getEventId()); return ids; @@ -133,6 +134,7 @@ public class SubscriptionEventService { public void recordPrimary(SubscriptionEvent event, SubBind binding) { transactions.executeWithoutResult(status -> { + events.reserveWriter(); if (!persist(event, "primary", binding)) throw new IllegalStateException("subscription user unavailable"); }); } diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionEventIntegrationTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionEventIntegrationTest.java index c6e8ba8..810729b 100644 --- a/src/test/java/com/lion/lionwebsite/Service/SubscriptionEventIntegrationTest.java +++ b/src/test/java/com/lion/lionwebsite/Service/SubscriptionEventIntegrationTest.java @@ -94,6 +94,29 @@ class SubscriptionEventIntegrationTest { assertEquals(5, ((Number) row.get("user_id")).intValue()); } + @Test void concurrentPrimaryAndReplayTransactionsReserveTheWriterBeforeReading() throws Exception { + var start = new java.util.concurrent.CountDownLatch(1); + var executor = java.util.concurrent.Executors.newFixedThreadPool(8); + var binding = subscriptions.selectSubBind(KEY); + try { + List> futures = new ArrayList<>(); + for (int i = 0; i < 32; i++) { + SubscriptionEvent event = event(NOW - i); + boolean primary = i % 2 == 0; + futures.add(executor.submit(() -> { + try { start.await(); } + catch (InterruptedException interrupted) { Thread.currentThread().interrupt(); throw new RuntimeException(interrupted); } + if (primary) service.recordPrimary(event, binding); + else assertEquals(List.of(event.getEventId()), service.ingest(List.of(event)).get("acknowledged")); + })); + } + start.countDown(); + for (var future : futures) future.get(30, java.util.concurrent.TimeUnit.SECONDS); + assertEquals(32, count("subscription_event_receipt")); + assertEquals(10, count("sub_update_record")); + } finally { executor.shutdownNow(); } + } + @Test void keyRotationAndRenameRetainStableIdentity() { service.rememberBindings(subscriptions.selectAllSubBind()); db.update("update sub_bind set key='rotated-key'");