Reserve SQLite writer before reading subscription event receipts

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);
 

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<String> acknowledged = transactions.execute(status -> {
+                events.reserveWriter();
                 List<String> 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");
         });
     }

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<java.util.concurrent.Future<?>> 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'");