Expire terminal mutation identities by configured epoch

test/Integration.hs

@@ -246,6 +246,52 @@ verifyAdmission database config = do
     [("applying", "settled")]
     durableState
 
+  verifyIdempotencyExpiry database
+
+verifyIdempotencyExpiry :: Db -> IO ()
+verifyIdempotencyExpiry database = do
+  resetAuthority database 100
+  first <- admitWithWindow database "expiring-operation" "expiring-key" 10 1
+  let rec = preparedRecord [first]
+  finalized <- withTransaction database $ \connection -> do
+    applying <- markApplying connection rec
+    pending <- markExternalPending connection rec
+    settled <- markExternalSettled connection rec
+    succeeded <- finalizeSuccess connection rec 10 (Just fixedIdentity)
+    pure (applying && pending && settled && succeeded)
+  assert "terminal expiry fixture finalized" finalized
+  _ <- Db.execute database
+    "UPDATE mutations SET updated_at = now() - interval '5 seconds' \
+    \WHERE user_id = ? AND kind = ? AND operation_id = ?"
+    ( "integration-user" :: Text
+    , "put" :: Text
+    , "expiring-operation" :: Text
+    )
+  outcomes <- mapConcurrently
+    (\size ->
+      admitWithWindow database "expiring-operation" "expiring-key" size 1)
+    [11, 12]
+  assertEqual "one first request after terminal expiry" 1 (countPrepared outcomes)
+  assertEqual "atomic conflicting reuse after terminal expiry" 1
+    (countConflict outcomes)
+  assertAuthority database 1 (mrGrowth (preparedRecord outcomes))
+
+  resetAuthority database 100
+  active <- admitWithWindow database "active-operation" "active-key" 10 1
+  _ <- Db.execute database
+    "UPDATE mutations SET updated_at = now() - interval '5 seconds' \
+    \WHERE user_id = ? AND kind = ? AND operation_id = ?"
+    ( "integration-user" :: Text
+    , "put" :: Text
+    , "active-operation" :: Text
+    )
+  replay <- admitWithWindow database "active-operation" "active-key" 10 1
+  blocked <- admitWithWindow database "other-operation" "active-key" 10 1
+  assertEqual "aged active identity remains replayable" 1
+    (countReplay [active, replay])
+  assertEqual "aged active key remains locked" 1 (countRejected [blocked])
+  assertAuthority database 1 10
+
 verifyStorage :: Config -> S3 -> IO ()
 verifyStorage config store = do
   let fallbackKey = ObjectKey "shared/fallback.bin"
@@ -258,8 +304,8 @@ verifyStorage config store = do
       (toInteger (BL.length fallbackBody))
       Nothing
       fallbackBody
-  fallbackHead <- expectJust "observe object without identity metadata"
-    =<< headObject store fallbackKey
+  fallbackHead <- expectRight "observe object without identity metadata"
+    =<< observeObjectFully store fallbackKey
   assertEqual "streaming digest fallback size"
     (toInteger (BL.length fallbackBody))
     (oiSize fallbackHead)
@@ -329,6 +375,10 @@ collectPages store prefix marker accumulated = do
 
 admit :: Db -> Text -> Text -> Integer -> IO AdmitOutcome
 admit database operation key size =
+  admitWithWindow database operation key size 3600
+
+admitWithWindow :: Db -> Text -> Text -> Integer -> Integer -> IO AdmitOutcome
+admitWithWindow database operation key size window =
   withTransaction database $ \connection ->
     admitMutation
       connection
@@ -341,6 +391,7 @@ admit database operation key size =
       size
       (Just fixedIdentity)
       1000
+      window
 
 resetAuthority :: Db -> Integer -> IO ()
 resetAuthority database limitBytes = do