nuthatch/src/Nuthatch/Protocol.hs

-- | Short Postgres transactions for mutation authority; store I/O stays outside them.
module Nuthatch.Protocol
  ( AdmitOutcome (..)
  , ExistingDisposition (..)
  , MutationRecord (..)
  , ObjectObservation (..)
  , admitMutation
  , markApplying
  , markExternalPending
  , markExternalSettled
  , markSettledFromIdle
  , finalizeSuccess
  , finalizeFailure
  , claimActiveBatch
  , lookupMutation
  , lookupObjectMeta
  , sameFingerprint
  , existingDisposition
  , growthBytes
  , quotaAllows
  , terminalExpiryEligible
  , terminalExpired
  , recoveryEligible
  , desiredMatches
  , priorMatches
  , finalizeFromObservation
  ) where

import           Data.Maybe (listToMaybe, mapMaybe)
import           Data.Text (Text)
import           Database.PostgreSQL.Simple (Connection, Only (..), execute, query)
import           Nuthatch.Types

data MutationRecord = MutationRecord
  { mrUserId        :: UserId
  , mrKind          :: MutationKind
  , mrOperationId   :: OperationId
  , mrPoolId        :: PoolId
  , mrObjectKey     :: ObjectKey
  , mrRelativePath  :: RelativePath
  , mrDeclaredSize  :: Integer
  , mrContentSha256 :: Maybe ContentSha256
  , mrPriorSize     :: Integer
  , mrPriorSha256   :: Maybe ContentSha256
  , mrGrowth        :: Integer
  , mrState         :: MutationState
  , mrExternal      :: ExternalState
  }
  deriving stock (Show)

-- | Unavailable must not drive terminal finalize.
data ObjectObservation
  = ObjectPresent Integer (Maybe ContentSha256)
  | ObjectAbsent
  | ObjectUnavailable Text
  deriving stock (Eq, Show)

data AdmitOutcome
  = AdmitPrepared MutationRecord
  | AdmitReplay MutationResult MutationRecord
  | AdmitRejected Text MutationRecord
  | AdmitConflict
  deriving stock (Show)

data ExistingDisposition
  = NewMutationEpoch
  | ExistingReplay MutationResult MutationRecord
  | ExistingConflict
  deriving stock (Show)

type ResultRow =
  ( Text, Text, Text, Text, Text, Text, Integer, Maybe Text
  , Integer, Maybe Text, Integer, Text, Text
  )

rowToRecord :: ResultRow -> Maybe MutationRecord
rowToRecord (user, kindT, op, pool, key, rel, size, csha, prior, psha, growth, stateT, extT) = do
  kind <- parseKind kindT
  state <- parseState stateT
  ext <- parseExternal extT
  pure MutationRecord
    { mrUserId = UserId user
    , mrKind = kind
    , mrOperationId = OperationId op
    , mrPoolId = PoolId pool
    , mrObjectKey = ObjectKey key
    , mrRelativePath = RelativePath rel
    , mrDeclaredSize = size
    , mrContentSha256 = ContentSha256 <$> csha
    , mrPriorSize = prior
    , mrPriorSha256 = ContentSha256 <$> psha
    , mrGrowth = growth
    , mrState = state
    , mrExternal = ext
    }

lookupMutation :: Connection -> UserId -> MutationKind -> OperationId -> IO (Maybe MutationRecord)
lookupMutation conn uid kind op = do
  rows <- query conn
    "SELECT user_id, kind, operation_id, pool_id, object_key, relative_path, \
    \declared_size, content_sha256, prior_size, prior_sha256, growth_bytes, \
    \state, external_state FROM mutations \
    \WHERE user_id = ? AND kind = ? AND operation_id = ?"
    (unUserId uid, kindLabel kind, unOperationId op)
  pure (listToMaybe rows >>= rowToRecord)

lookupObjectMeta :: Connection -> ObjectKey -> IO (Maybe (Integer, ContentSha256))
lookupObjectMeta conn (ObjectKey key) = do
  rows <- query conn
    "SELECT size_bytes, content_sha256 FROM objects_meta WHERE object_key = ?"
    (Only key)
  pure $ case rows of
    [(sz, sha)] -> Just (sz, ContentSha256 sha)
    _           -> Nothing

sameFingerprint :: MutationRecord -> MutationKind -> ObjectKey -> Integer -> Maybe ContentSha256 -> Bool
sameFingerprint rec kind key size msha =
  mrKind rec == kind
    && mrObjectKey rec == key
    && mrDeclaredSize rec == size
    && mrContentSha256 rec == msha

existingDisposition
  :: Maybe MutationRecord
  -> MutationKind
  -> ObjectKey
  -> Integer
  -> Maybe ContentSha256
  -> ExistingDisposition
existingDisposition existing kind key size msha =
  case existing of
    Nothing -> NewMutationEpoch
    Just rec
      | sameFingerprint rec kind key size msha ->
          ExistingReplay (stateToResult (mrState rec)) rec
      | otherwise -> ExistingConflict

growthBytes :: Integer -> Integer -> Integer
growthBytes priorSize desiredSize = max 0 (desiredSize - priorSize)

quotaAllows :: Integer -> Integer -> Integer -> Integer -> Bool
quotaAllows limitBytes usedBytes reservedBytes growth =
  usedBytes + reservedBytes + growth <= limitBytes

terminalExpiryEligible :: MutationState -> ExternalState -> Bool
terminalExpiryEligible state external =
  state `elem` [Succeeded, Failed, Rejected] && external == ExtIdle

terminalExpired :: MutationState -> ExternalState -> Integer -> Integer -> Bool
terminalExpired state external elapsedSeconds windowSeconds =
  terminalExpiryEligible state external && elapsedSeconds > windowSeconds

recoveryEligible :: MutationState -> Integer -> Integer -> Bool
recoveryEligible state elapsedSeconds graceSeconds =
  state `elem` [Prepared, Applying] && elapsedSeconds > graceSeconds

admitMutation
  :: Connection
  -> UserId
  -> MutationKind
  -> OperationId
  -> PoolId
  -> ObjectKey
  -> RelativePath
  -> Integer
  -> Maybe ContentSha256
  -> Integer
  -> Integer
  -> IO AdmitOutcome
admitMutation conn uid kind op pool key rel size msha maxObject 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 ()]
  _ <- execute conn
    "DELETE FROM mutations \
    \WHERE user_id = ? AND kind = ? AND operation_id = ? \
    \AND state IN ('succeeded', 'failed', 'rejected') \
    \AND external_state = 'idle' \
    \AND updated_at < now() - (? * interval '1 second')"
    (unUserId uid, kindLabel kind, unOperationId op, idempotencyWindow)
  existing <- lookupMutation conn uid kind op
  case existingDisposition existing kind key size msha of
    ExistingReplay result rec -> pure (AdmitReplay result rec)
    ExistingConflict -> pure AdmitConflict
    NewMutationEpoch -> do
      priorMeta <- lookupObjectMeta conn key
      let priorSize = maybe 0 fst priorMeta
          priorSha = snd <$> priorMeta
          desiredSize = case kind of
            Put    -> size
            Delete -> 0
          growth = growthBytes priorSize desiredSize
          baseRec state =
            MutationRecord
              { mrUserId = uid
              , mrKind = kind
              , mrOperationId = op
              , mrPoolId = pool
              , mrObjectKey = key
              , mrRelativePath = rel
              , mrDeclaredSize = size
              , mrContentSha256 = msha
              , mrPriorSize = priorSize
              , mrPriorSha256 = priorSha
              , mrGrowth = 0
              , mrState = state
              , mrExternal = ExtIdle
              }

      poolRows <- query conn
        "SELECT 1 FROM pools WHERE pool_id = ? FOR UPDATE"
        (Only (unPoolId pool)) :: IO [Only Int]

      if null poolRows
        then pure AdmitConflict
        else if kind == Put && (size <= 0 || size > maxObject)
        then do
          rec <- insertTerminalRejected conn uid kind op pool key rel size msha priorSize priorSha
          pure (AdmitRejected "invalid object size" rec)
        else do
          locked <- query conn
            "SELECT 1 FROM key_locks WHERE object_key = ? FOR UPDATE"
            (Only (unObjectKey key)) :: IO [Only Int]
          if not (null locked)
            then do
              rec <- insertTerminalRejected conn uid kind op pool key rel size msha priorSize priorSha
              pure (AdmitRejected "key locked by another mutation" rec)
            else do
              okQuota <- query conn
                "SELECT 1 FROM pools \
                \WHERE pool_id = ? AND used_bytes + reserved_bytes + ? <= limit_bytes"
                (unPoolId pool, growth) :: IO [Only Int]
              if null okQuota
                then do
                  rec <- insertTerminalRejected conn uid kind op pool key rel size msha priorSize priorSha
                  pure (AdmitRejected "quota exceeded" rec)
                else do
                  n <- execute conn
                    "INSERT INTO mutations \
                    \(user_id, kind, operation_id, pool_id, object_key, relative_path, \
                    \ declared_size, content_sha256, prior_size, prior_sha256, growth_bytes, \
                    \ state, external_state) \
                    \VALUES (?,?,?,?,?,?,?,?,?,?,?,'prepared','idle')"
                    ( unUserId uid
                    , kindLabel kind
                    , unOperationId op
                    , unPoolId pool
                    , unObjectKey key
                    , unRelativePath rel
                    , size
                    , unContentSha256 <$> msha
                    , priorSize
                    , unContentSha256 <$> priorSha
                    , growth
                    )
                  if n /= 1
                    then pure AdmitConflict
                    else do
                      lockN <- execute conn
                        "INSERT INTO key_locks (object_key, user_id, kind, operation_id) \
                        \VALUES (?,?,?,?) ON CONFLICT DO NOTHING"
                        (unObjectKey key, unUserId uid, kindLabel kind, unOperationId op)
                      if lockN /= 1
                        then do
                          _ <- execute conn
                            "UPDATE mutations SET state = 'rejected', growth_bytes = 0, \
                            \updated_at = now() \
                            \WHERE user_id = ? AND kind = ? AND operation_id = ?"
                            (unUserId uid, kindLabel kind, unOperationId op)
                          pure $ AdmitRejected "key locked by another mutation"
                            (baseRec Rejected)
                        else do
                          _ <- execute conn
                            "UPDATE pools SET reserved_bytes = reserved_bytes + ? WHERE pool_id = ?"
                            (growth, unPoolId pool)
                          pure $ AdmitPrepared MutationRecord
                            { mrUserId = uid
                            , mrKind = kind
                            , mrOperationId = op
                            , mrPoolId = pool
                            , mrObjectKey = key
                            , mrRelativePath = rel
                            , mrDeclaredSize = size
                            , mrContentSha256 = msha
                            , mrPriorSize = priorSize
                            , mrPriorSha256 = priorSha
                            , mrGrowth = growth
                            , mrState = Prepared
                            , mrExternal = ExtIdle
                            }

insertTerminalRejected
  :: Connection
  -> UserId
  -> MutationKind
  -> OperationId
  -> PoolId
  -> ObjectKey
  -> RelativePath
  -> Integer
  -> Maybe ContentSha256
  -> Integer
  -> Maybe ContentSha256
  -> IO MutationRecord
insertTerminalRejected conn uid kind op pool key rel size msha priorSize priorSha = do
  _ <- execute conn
    "INSERT INTO mutations \
    \(user_id, kind, operation_id, pool_id, object_key, relative_path, \
    \ declared_size, content_sha256, prior_size, prior_sha256, growth_bytes, \
    \ state, external_state) \
    \VALUES (?,?,?,?,?,?,?,?,?,?,0,'rejected','idle') \
    \ON CONFLICT DO NOTHING"
    ( unUserId uid
    , kindLabel kind
    , unOperationId op
    , unPoolId pool
    , unObjectKey key
    , unRelativePath rel
    , size
    , unContentSha256 <$> msha
    , priorSize
    , unContentSha256 <$> priorSha
    )
  pure MutationRecord
    { mrUserId = uid
    , mrKind = kind
    , mrOperationId = op
    , mrPoolId = pool
    , mrObjectKey = key
    , mrRelativePath = rel
    , mrDeclaredSize = size
    , mrContentSha256 = msha
    , mrPriorSize = priorSize
    , mrPriorSha256 = priorSha
    , mrGrowth = 0
    , mrState = Rejected
    , mrExternal = ExtIdle
    }

stateToResult :: MutationState -> MutationResult
stateToResult Prepared  = MutationPending
stateToResult Applying  = MutationPending
stateToResult Succeeded = MutationSucceeded
stateToResult Failed    = MutationFailed
stateToResult Rejected  = MutationRejected

markApplying :: Connection -> MutationRecord -> IO Bool
markApplying conn rec = do
  n <- execute conn
    "UPDATE mutations SET state = 'applying', updated_at = now() \
    \WHERE user_id = ? AND kind = ? AND operation_id = ? AND state = 'prepared'"
    (unUserId (mrUserId rec), kindLabel (mrKind rec), unOperationId (mrOperationId rec))
  pure (n == 1)

markExternalPending
  :: Connection
  -> MutationRecord
  -> Integer
  -> Integer
  -> IO Bool
markExternalPending conn rec settlementBoundSec workerGraceSec = do
  n <- execute conn
    "UPDATE mutations SET external_state = 'pending', \
    \dispatch_started_at = now(), \
    \request_deadline_at = now() + (? * interval '1 second'), \
    \worker_lease_until = now() + (? * interval '1 second'), \
    \updated_at = now() \
    \WHERE user_id = ? AND kind = ? AND operation_id = ? \
    \AND state = 'applying' AND external_state = 'idle'"
    ( settlementBoundSec
    , workerGraceSec
    , unUserId (mrUserId rec)
    , kindLabel (mrKind rec)
    , unOperationId (mrOperationId rec)
    )
  pure (n == 1)

markExternalSettled :: Connection -> MutationRecord -> IO Bool
markExternalSettled conn rec = do
  n <- execute conn
    "UPDATE mutations SET external_state = 'settled', settled_at = now(), \
    \updated_at = now() \
    \WHERE user_id = ? AND kind = ? AND operation_id = ? \
    \AND state = 'applying' AND external_state = 'pending'"
    (unUserId (mrUserId rec), kindLabel (mrKind rec), unOperationId (mrOperationId rec))
  pure (n == 1)

markSettledFromIdle :: Connection -> MutationRecord -> IO Bool
markSettledFromIdle conn rec = do
  n <- execute conn
    "UPDATE mutations SET external_state = 'settled', settled_at = now(), \
    \updated_at = now() \
    \WHERE user_id = ? AND kind = ? AND operation_id = ? \
    \AND state = 'applying' AND external_state = 'idle'"
    (unUserId (mrUserId rec), kindLabel (mrKind rec), unOperationId (mrOperationId rec))
  pure (n == 1)

finalizeSuccess :: Connection -> MutationRecord -> Integer -> Maybe ContentSha256 -> IO Bool
finalizeSuccess conn rec finalSize finalSha = do
  rows <- query conn
    "SELECT state, external_state, growth_bytes, prior_size FROM mutations \
    \WHERE user_id = ? AND kind = ? AND operation_id = ? FOR UPDATE"
    (unUserId (mrUserId rec), kindLabel (mrKind rec), unOperationId (mrOperationId rec))
      :: IO [(Text, Text, Integer, Integer)]
  case rows of
    [(st, ext, growth, prior)]
      | st == "applying" && ext == "settled" -> do
          let pool = mrPoolId rec
          _ <- execute conn
            "UPDATE pools SET used_bytes = used_bytes - ? + ?, \
            \reserved_bytes = reserved_bytes - ? WHERE pool_id = ?"
            (prior, finalSize, growth, unPoolId pool)
          case mrKind rec of
            Put -> do
              let sha = maybe "" unContentSha256 finalSha
              _ <- execute conn
                "INSERT INTO objects_meta (object_key, pool_id, size_bytes, content_sha256, updated_at) \
                \VALUES (?,?,?,?, now()) \
                \ON CONFLICT (object_key) DO UPDATE SET \
                \  size_bytes = EXCLUDED.size_bytes, \
                \  content_sha256 = EXCLUDED.content_sha256, \
                \  updated_at = now()"
                (unObjectKey (mrObjectKey rec), unPoolId pool, finalSize, sha)
              pure ()
            Delete -> do
              _ <- execute conn
                "DELETE FROM objects_meta WHERE object_key = ?"
                (Only (unObjectKey (mrObjectKey rec)))
              pure ()
          _ <- execute conn
            "UPDATE mutations SET state = 'succeeded', external_state = 'idle', updated_at = now() \
            \WHERE user_id = ? AND kind = ? AND operation_id = ?"
            (unUserId (mrUserId rec), kindLabel (mrKind rec), unOperationId (mrOperationId rec))
          _ <- execute conn
            "DELETE FROM key_locks WHERE object_key = ? AND user_id = ? AND kind = ? AND operation_id = ?"
            ( unObjectKey (mrObjectKey rec)
            , unUserId (mrUserId rec)
            , kindLabel (mrKind rec)
            , unOperationId (mrOperationId rec)
            )
          pure True
    _ -> pure False

finalizeFailure :: Connection -> MutationRecord -> IO Bool
finalizeFailure conn rec = do
  rows <- query conn
    "SELECT state, external_state, growth_bytes FROM mutations \
    \WHERE user_id = ? AND kind = ? AND operation_id = ? FOR UPDATE"
    (unUserId (mrUserId rec), kindLabel (mrKind rec), unOperationId (mrOperationId rec))
      :: IO [(Text, Text, Integer)]
  case rows of
    [(st, ext, growth)]
      | st == "applying" && ext == "settled" -> do
          _ <- execute conn
            "UPDATE pools SET reserved_bytes = reserved_bytes - ? WHERE pool_id = ?"
            (growth, unPoolId (mrPoolId rec))
          _ <- execute conn
            "UPDATE mutations SET state = 'failed', external_state = 'idle', updated_at = now() \
            \WHERE user_id = ? AND kind = ? AND operation_id = ?"
            (unUserId (mrUserId rec), kindLabel (mrKind rec), unOperationId (mrOperationId rec))
          _ <- execute conn
            "DELETE FROM key_locks WHERE object_key = ? AND user_id = ? AND kind = ? AND operation_id = ?"
            ( unObjectKey (mrObjectKey rec)
            , unUserId (mrUserId rec)
            , kindLabel (mrKind rec)
            , unOperationId (mrOperationId rec)
            )
          pure True
    _ -> pure False

-- | updated_at advances on state transitions only, not during a long put body.
claimActiveBatch :: Connection -> Int -> Integer -> IO [MutationRecord]
claimActiveBatch conn limit graceSec = do
  rows <- query conn
    "SELECT user_id, kind, operation_id, pool_id, object_key, relative_path, \
    \declared_size, content_sha256, prior_size, prior_sha256, growth_bytes, \
    \state, external_state FROM mutations \
    \WHERE state IN ('prepared', 'applying') \
    \AND ( \
    \  (external_state = 'idle' \
    \    AND updated_at < now() - (? * interval '1 second')) \
    \  OR (external_state = 'settled' \
    \    AND updated_at < now() - (? * interval '1 second')) \
    \  OR (external_state = 'pending' \
    \    AND worker_lease_until < now() \
    \    AND request_deadline_at < now()) \
    \) \
    \ORDER BY updated_at ASC \
    \FOR UPDATE SKIP LOCKED \
    \LIMIT ?"
    (graceSec, graceSec, limit)
  pure (mapMaybe rowToRecord rows)

desiredMatches :: MutationRecord -> ObjectObservation -> Bool
desiredMatches rec = \case
  ObjectUnavailable _ -> False
  ObjectAbsent ->
    case mrKind rec of
      Delete -> True
      Put    -> False
  ObjectPresent sz msha ->
    case mrKind rec of
      Delete -> False
      Put ->
        sz == mrDeclaredSize rec
          && case (mrContentSha256 rec, msha) of
               (Just want, Just got) -> want == got
               (Just _, Nothing)     -> False
               (Nothing, _)          -> True

priorMatches :: MutationRecord -> ObjectObservation -> Bool
priorMatches rec obs =
  let priorEmpty = mrPriorSize rec == 0 && mrPriorSha256 rec == Nothing
  in case obs of
       ObjectUnavailable _ -> False
       ObjectAbsent -> priorEmpty
       ObjectPresent sz msha ->
         sz == mrPriorSize rec
           && ( case (mrPriorSha256 rec, msha) of
                  (Just want, Just got) -> want == got
                  (Nothing, Nothing)    -> True
                  (Just _, Nothing)     -> False
                  (Nothing, Just _)     -> mrPriorSize rec == sz
              )

finalizeFromObservation
  :: Connection
  -> MutationRecord
  -> ObjectObservation
  -> IO (Maybe MutationResult)
finalizeFromObservation conn rec observed =
  case observed of
    ObjectUnavailable _ -> pure Nothing
    _ | desiredMatches rec observed -> do
      let (finalSize, finalSha) =
            case mrKind rec of
              Put ->
                ( mrDeclaredSize rec
                , mrContentSha256 rec
                )
              Delete -> (0, Nothing)
      ok <- finalizeSuccess conn rec finalSize finalSha
      pure $ if ok then Just MutationSucceeded else Nothing
    _ | priorMatches rec observed -> do
      ok <- finalizeFailure conn rec
      pure $ if ok then Just MutationFailed else Nothing
    _ -> pure Nothing