Add bounded protocol and target conformance gate

src/Nuthatch/App.hs

@@ -6,6 +6,9 @@ module Nuthatch.App
   , authenticationHandler
   , loginResponse
   , dispatchAndFinalize
+  , PutFailure (..)
+  , runPut
+  , runDelete
   , runRecovery
   ) where
 
@@ -68,6 +71,7 @@ import           Nuthatch.S3
   , getObjectSource
   , headObject
   , listObjects
+  , objectStoreAvailable
   , observeObjectFully
   , putObjectStream
   , renderS3Error
@@ -550,32 +554,42 @@ runPut env user root rel op size sha body = do
       pool = case root of
         Personal -> personalPool uid
         Shared   -> sharedPool
-  outcome <- withTransaction (envDb env) $ \conn ->
-    admitMutation conn uid Put op pool key rel size (Just sha)
-      (cfgMaxObjectBytes (envConfig env))
-      (cfgIdempotencyWindowSec (envConfig env))
-  case outcome of
-    AdmitConflict -> pure (Right MutationConflict)
-    AdmitRejected reason _ -> pure (Left (PutRejected reason))
-    AdmitReplay MutationPending _ -> pure (Right MutationPending)
-    AdmitReplay result _ -> pure (Right result)
-    AdmitPrepared rec ->
-      dispatchAndFinalize
-        env
-        user
-        root
-        rec
-        PutSettlementFailed
-        (PutDispatchFailed (PutObjectStoreError S3TransportError))
-        (putObjectStream
-          (envS3 env)
-          key
-          size
-          (Just sha)
-          (transferPieceSize env)
-          (inspectBandwidth env user)
-          body
-          >>= pure . either (Left . PutDispatchFailed) Right)
+  admitted <- withTransaction (envDb env) $ \conn -> do
+    existing <- mutationEpochExists
+      conn uid Put op (cfgIdempotencyWindowSec (envConfig env))
+    available <-
+      if existing then pure True else objectStoreAvailable (envS3 env)
+    if available
+      then Just <$> admitMutation conn uid Put op pool key rel size (Just sha)
+        (cfgMaxObjectBytes (envConfig env))
+        (cfgIdempotencyWindowSec (envConfig env))
+      else pure Nothing
+  case admitted of
+    Nothing ->
+      pure (Left (PutDispatchFailed (PutObjectStoreError S3TransportError)))
+    Just outcome ->
+      case outcome of
+        AdmitConflict -> pure (Right MutationConflict)
+        AdmitRejected reason _ -> pure (Left (PutRejected reason))
+        AdmitReplay MutationPending _ -> pure (Right MutationPending)
+        AdmitReplay result _ -> pure (Right result)
+        AdmitPrepared rec ->
+          dispatchAndFinalize
+            env
+            user
+            root
+            rec
+            PutSettlementFailed
+            (PutDispatchFailed (PutObjectStoreError S3TransportError))
+            (putObjectStream
+              (envS3 env)
+              key
+              size
+              (Just sha)
+              (transferPieceSize env)
+              (inspectBandwidth env user)
+              body
+              >>= pure . either (Left . PutDispatchFailed) Right)
 
 data PutFailure
   = PutRejected Text
@@ -623,25 +637,34 @@ runDelete env user root rel op = do
       pool = case root of
         Personal -> personalPool uid
         Shared   -> sharedPool
-  outcome <- withTransaction (envDb env) $ \conn ->
-    admitMutation conn uid Delete op pool key rel 0 Nothing
-      (cfgMaxObjectBytes (envConfig env))
-      (cfgIdempotencyWindowSec (envConfig env))
-  case outcome of
-    AdmitConflict -> pure (Right MutationConflict)
-    AdmitRejected reason _ -> pure (Left reason)
-    AdmitReplay MutationPending _ -> pure (Right MutationPending)
-    AdmitReplay result _ -> pure (Right result)
-    AdmitPrepared rec ->
-      dispatchAndFinalize
-        env
-        user
-        root
-        rec
-        "mutation failed after store settlement"
-        "object request timed out"
-        (deleteObject (envS3 env) key
-          >>= pure . either (Left . renderS3Error) Right)
+  admitted <- withTransaction (envDb env) $ \conn -> do
+    existing <- mutationEpochExists
+      conn uid Delete op (cfgIdempotencyWindowSec (envConfig env))
+    available <-
+      if existing then pure True else objectStoreAvailable (envS3 env)
+    if available
+      then Just <$> admitMutation conn uid Delete op pool key rel 0 Nothing
+        (cfgMaxObjectBytes (envConfig env))
+        (cfgIdempotencyWindowSec (envConfig env))
+      else pure Nothing
+  case admitted of
+    Nothing -> pure (Left "object store unavailable")
+    Just outcome ->
+      case outcome of
+        AdmitConflict -> pure (Right MutationConflict)
+        AdmitRejected reason _ -> pure (Left reason)
+        AdmitReplay MutationPending _ -> pure (Right MutationPending)
+        AdmitReplay result _ -> pure (Right result)
+        AdmitPrepared rec ->
+          dispatchAndFinalize
+            env
+            user
+            root
+            rec
+            "mutation failed after store settlement"
+            "object request timed out"
+            (deleteObject (envS3 env) key
+              >>= pure . either (Left . renderS3Error) Right)
 
 deletePrefix :: Env -> SessionUser -> Root -> Text -> IO (Either Text Int)
 deletePrefix env user root prefix = do

src/Nuthatch/Protocol.hs

@@ -13,6 +13,7 @@ module Nuthatch.Protocol
   , finalizeFailure
   , claimActiveBatch
   , lookupMutation
+  , mutationEpochExists
   , lookupObjectMeta
   , sameFingerprint
   , existingDisposition
@@ -104,6 +105,29 @@ lookupMutation conn uid kind op = do
     (unUserId uid, kindLabel kind, unOperationId op)
   pure (listToMaybe rows >>= rowToRecord)
 
+mutationEpochExists
+  :: Connection
+  -> UserId
+  -> MutationKind
+  -> OperationId
+  -> Integer
+  -> IO Bool
+mutationEpochExists conn uid kind op idempotencyWindow = do
+  let identity =
+        unUserId uid <> "\NUL" <> kindLabel kind <> "\NUL" <> unOperationId op
+  _ <- query conn
+    "SELECT pg_advisory_xact_lock(hashtextextended(?, 0))"
+    (Only identity) :: IO [Only ()]
+  rows <- query conn
+    "SELECT 1 FROM mutations \
+    \WHERE user_id = ? AND kind = ? AND operation_id = ? \
+    \AND (state IN ('prepared', 'applying') \
+    \  OR external_state <> 'idle' \
+    \  OR updated_at >= now() - (? * interval '1 second'))"
+    (unUserId uid, kindLabel kind, unOperationId op, idempotencyWindow)
+      :: IO [Only Int]
+  pure (not (null rows))
+
 lookupObjectMeta :: Connection -> ObjectKey -> IO (Maybe (Integer, ContentSha256))
 lookupObjectMeta conn (ObjectKey key) = do
   rows <- query conn

src/Nuthatch/S3.hs

@@ -13,6 +13,7 @@ module Nuthatch.S3
   , observeObjectFully
   , deleteObject
   , listObjects
+  , objectStoreAvailable
   , renderS3Error
   ) where
 
@@ -443,6 +444,11 @@ listObjects store prefix delimiter continuation maxKeys = do
         then S3.objectKey <$> lastMaybe (S3.gbrContents response)
         else Nothing
 
+objectStoreAvailable :: S3 -> IO Bool
+objectStoreAvailable store =
+  either (const False) (const True) <$>
+    listObjects store "" Nothing Nothing 1
+
 renderS3Error :: S3Error -> Text
 renderS3Error S3NotFound = "object not found"
 renderS3Error (S3ServiceError code) =