redis-schema 0.1.0 → 0.2.0

raw patch · 6 files changed

+695/−113 lines, 6 filesdep +monadIOdep ~basedep ~hedisPVP ok

version bump matches the API change (PVP)

Dependencies added: monadIO

Dependency ranges changed: base, hedis

API changes (from Hackage documentation)

Files

• CHANGELOG.md +17/−0
• README.md +73/−2
• redis-schema.cabal +7/−5
• src/Database/Redis/Schema.hs +152/−84
• src/Database/Redis/Schema/Lock.hs +68/−22
• src/Database/Redis/Schema/RemoteJob.hs +378/−0

@@ -1,3 +1,20 @@+# v0.2++- Support GHC 9.8-9.12. Bump `hedis` to 0.16 (see [#11](https://github.com/chordify/redis-schema/pull/11)).+- Throw errors on silent discarding of errors from ByteString decoding (see [#10](https://github.com/chordify/redis-schema/pull/10)).++**Breaking changes** (since `hedis-0.16`).++1. Drop support of GHCs older than 9.6.+2. Non-empty lists used elsewhere in the code.++**Migration guide**++1. In case of compilation errors replace `[a]` with `NonEmpty a`, e.g. +  - `[v]` with `pure v` +  - or `(v :| [])` +  - or `NE.singleton v` (`import qualified Data.List.NonEmpty as NE`).+ # v0.1  First public version of `redis-schema`.

@@ -894,8 +894,79 @@  ### Remote jobs -Sadly, this library has not been published yet.-We'd like to, though.+In `Database.Redis.Schema.RemoteJob` a Redis-based worker queue is implemented, to run CPU+intensive jobs on remote machines. The queue is strongly typed, and can contain multiple+different jobs to be executed, with priorities, that workers can pick up.++As an example, we define a queue that can contain three types of jobs:+```haskell+newtype ComputeFactorial = CF Integer deriving ( Binary )+newtype ComputeSquare    = CS Integer deriving ( Binary )++data MyQueue+instance JobQueue MyQueue where+  type RPC MyQueue =+    '[ ComputeFactorial -> Integer+     , ComputeSquare -> Integer+     , String -> String+     ]+  keyPrefix = "myqueue"+```+Here the `MyQueue` type is used only during compile time to let the compiler find the right+instances. To distinguish between the two `Integer -> Integer` functions, we wrap them in+newtypes. A `Binary` instance must exist for all inputs and outputs, so that they can be put+into Redis.++Based on this queue, we can now define a worker that executes the jobs. This worker must+define a function for each the the types in `RPC`, and runs in a monadic context (which+we fixed to `IO` for the example).++```haskell+fac :: ComputeFactorial -> IO Integer+fac (CF n) = do+  putStrLn $ "Computing the factorial of " ++ show n+  pure $ product [1..n]++sm :: ComputeSquare -> IO Integer+sm (CS n) = pure $ n * n++runWorker :: IO ()+runWorker = do+  pool <- connect "redis:///" 10+  let myId = "localworker"+  let err e = error $ "Something went wrong: " ++ show e+  remoteJobWorker @MyQueue myId pool err fac sm (pure . reverse)+```+The arguments to `remoteJobWorker` are a unique identifier for this worker (for counting+the workers, executing jobs will work fine even with overlapping ids), a connection pool,+a logging function for exceptions, and then for each element in `RPC` the right function.++Now if we call `runWorker` it will block until work needs to be done, and it will never+return except when an async exception is thrown. In production cases it is adviced to use+`withRemoteJobWorker` instead, which forks off a worker thread and provides a `WorkerHandle`+to it's continuation, which can be passed to `gracefulShutdown` to handle the currently+running job and then gracefully return.++Now from another process or even other machine we can 'execute jobs', e.g. add them to the+queue and synchronously wait for their result. For example:+```haskell+runJobs :: IO ()+runJobs = do+  pool <- connect "redis:///" 10+  a <- runRemoteJob @MyQueue @String @String False pool 1 "test"+  print a+  b <- runRemoteJob @MyQueue @ComputeFactorial @Integer False pool 1 (CF 5)+  print b+```+This will print:+```ghci> runJobs+Right "tset"+Right 120+```+The underlying Redis implementation is based on blocking reads from sorted sets (`BZPOPMIN`),+which is concurrency safe and no polling is needed. An arbitrary amount of workers can be run+and jobs can be executed from arbitrary machines. Only the `countWorkers` implementation+is based on a keep-alive loop on the workers, to properly deal with TCP connection losses.  ## Future work 

@@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.34.7.+-- This file has been generated from package.yaml by hpack version 0.35.0. -- -- see: https://github.com/sol/hpack ----- hash: 8ad979f047b1d31267791ddddc75577141d0b1f972f265c586894ab5f99c498c+-- hash: 1876630159ac153904237e5ab7ec2b440a07213ac8ca4b6420a02fff46e1524d  name:           redis-schema-version:        0.1.0+version:        0.2.0 synopsis:       Typed, schema-based, composable Redis library description:    Typed, schema-based, composable Redis library category:       Database@@ -31,6 +31,7 @@   exposed-modules:       Database.Redis.Schema       Database.Redis.Schema.Lock+      Database.Redis.Schema.RemoteJob   other-modules:       Paths_redis_schema   hs-source-dirs:@@ -39,12 +40,13 @@       OverloadedStrings   ghc-options: -Wall   build-depends:-      base >=4.7 && <5+      base >=4.18 && <5     , binary     , bytestring     , containers     , exceptions-    , hedis+    , hedis >=0.16.1 && <0.17+    , monadIO     , mtl     , numeric-limits     , random

@@ -41,7 +41,7 @@   , run   , connect   , incrementBy, incrementByFloat-  , txIncrementBy+  , txIncrementBy, txIncrementByFloat   , get, set, getSet   , txGet, txSet, txExpect   , setWithTTL, setIfNotExists, setIfNotExists_@@ -56,7 +56,7 @@   , day, hour, minute, second   , throw, throwMsg   , sInsert, sDelete, sContains, sSize-  , Priority(..), zInsert, zSize, zCount, zDelete, zPopMin, bzPopMin, zRangeByScoreLimit+  , Priority(..), zInsert, zSize, zCount, zDelete, zIncrBy, zPopMax, zPopMin, bzPopMin, zRangeByScoreLimit, zRange, zRevRange, zScanOpts, zUnionStoreWeights   , txSInsert, txSDelete, txSContains, txSSize   , MapItem(..)   , RecordField(..), RecordItem(..), Record@@ -70,6 +70,7 @@ import GHC.Word         ( Word32  ) import Data.Functor     ( void, (<&>) ) import Data.Function    ( (&) )+import Data.List.NonEmpty ( NonEmpty (..) ) import Data.Time        ( UTCTime, LocalTime, Day ) import Text.Read        ( readMaybe ) import Data.ByteString  ( ByteString )@@ -94,6 +95,7 @@ import qualified Database.Redis as Hedis import qualified Data.ByteString.Char8 as BS import qualified Data.ByteString.Lazy as BSL+import qualified Data.List.NonEmpty as NE import qualified Data.Map as Map import qualified Data.Set as Set import qualified System.IO.Error as IOE@@ -133,13 +135,13 @@   | TransactionAborted   | TransactionError String   | CouldNotDecodeValue (Maybe ByteString)-  | LockAcquireTimeout   | UnexpectedStatus String Hedis.Status   | EmptyAlternative  -- for 'instance Alternative Tx'   deriving (Show, Exception)  -- | Time-To-Live for Redis values. The Num instance works in (integral) seconds. newtype TTL = TTLSec { ttlToSeconds :: Integer }+  deriving stock (Show)   deriving newtype (Eq, Ord, Num)  run :: MonadIO m => Pool inst -> RedisM inst a -> m a@@ -332,8 +334,8 @@   txValGet :: Identifier val -> Tx inst (Maybe val)    default txValGet :: SimpleValue inst val => Identifier val -> Tx inst (Maybe val)-  txValGet (SviTopLevel keyBS) = fmap (fromBS =<<) . txWrap $ Hedis.get keyBS-  txValGet (SviHash keyBS hkeyBS) = fmap (fromBS =<<) . txWrap $ Hedis.hget keyBS hkeyBS+  txValGet (SviTopLevel keyBS) = txFromBSorExcept $ txWrap $ Hedis.get keyBS+  txValGet (SviHash keyBS hkeyBS) = txFromBSorExcept $ txWrap $ Hedis.hget keyBS hkeyBS    -- | Write a value to Redis in a transaction.   txValSet :: Identifier val -> val -> Tx inst ()@@ -344,14 +346,14 @@       $ txWrap (Hedis.set keyBS $ toBS val)   txValSet (SviHash keyBS hkeyBS) val =     void-      $ txWrap (Hedis.hset keyBS hkeyBS $ toBS val)+      $ txWrap (Hedis.hset keyBS $ (hkeyBS, toBS val) :| [])    -- | Delete a value from Redis in a transaction.   txValDelete :: Identifier val -> Tx inst ()    default txValDelete :: SimpleValue inst val => Identifier val -> Tx inst ()-  txValDelete (SviTopLevel keyBS) = void . txWrap $ Hedis.del [keyBS]-  txValDelete (SviHash keyBS hkeyBS) = void . txWrap $ Hedis.hdel keyBS [hkeyBS]+  txValDelete (SviTopLevel keyBS) = void . txWrap $ Hedis.del (keyBS :| [])+  txValDelete (SviHash keyBS hkeyBS) = void . txWrap $ Hedis.hdel keyBS (hkeyBS :| [])    -- | Set time-to-live for a value in a transaction. Return 'True' if the value exists.   txValSetTTLIfExists :: Identifier val -> TTL -> Tx inst Bool@@ -368,9 +370,9 @@    default valGet :: SimpleValue inst val => Identifier val -> RedisM inst (Maybe val)   valGet (SviTopLevel keyBS) =-    fmap (fromBS =<<) . expectRight "valGet/plain" =<< Hedis.get keyBS+    traverse rFromBSorThrow =<< expectRight "valGet/plain" =<< Hedis.get keyBS   valGet (SviHash keyBS hkeyBS) =-    fmap (fromBS =<<) . expectRight "valGet/hash" =<< Hedis.hget keyBS hkeyBS+    traverse rFromBSorThrow =<< expectRight "valGet/hash" =<< Hedis.hget keyBS hkeyBS    -- | Write a value.   valSet :: Identifier val -> val -> RedisM inst ()@@ -379,7 +381,7 @@   valSet (SviTopLevel keyBS) val =     expect "valSet/plain" (Right Hedis.Ok) =<< Hedis.set keyBS (toBS val)   valSet (SviHash keyBS hkeyBS) val =-    ignore {- @Integer -} =<< expectRight "valSet/hash" =<< Hedis.hset keyBS hkeyBS (toBS val)+    ignore {- @Integer -} =<< expectRight "valSet/hash" =<< Hedis.hset keyBS ((hkeyBS, toBS val) :| [])       --   ^- this is Bool in some versions of Hedis and Integer in others    -- | Delete a value.@@ -387,9 +389,9 @@    default valDelete :: SimpleValue inst val => Identifier val -> RedisM inst ()   valDelete (SviTopLevel keyBS) =-    ignore @Integer =<< expectRight "valDelete/plain" =<< Hedis.del [keyBS]+    ignore @Integer =<< expectRight "valDelete/plain" =<< Hedis.del (keyBS :| [])   valDelete (SviHash keyBS hkeyBS) =-    ignore @Integer =<< expectRight "valDelete/hash" =<< Hedis.hdel keyBS [hkeyBS]+    ignore @Integer =<< expectRight "valDelete/hash" =<< Hedis.hdel keyBS (hkeyBS :| [])    -- | Set time-to-live for a value. Return 'True' if the value exists.   valSetTTLIfExists :: Identifier val -> TTL -> RedisM inst Bool@@ -400,9 +402,17 @@   valSetTTLIfExists (SviHash keyBS _hkeyBS) (TTLSec ttlSec) =     expectRight "valSetTTLIfExists/hash" =<< Hedis.expire keyBS ttlSec +-- | Ensure the fromBS conversion does not fail silently due to a change of the internal Haskell ValueType of a Ref+rFromBSorThrow :: Serializable a => ByteString -> RedisM inst a+rFromBSorThrow bs = maybe (throw $ CouldNotDecodeValue $ Just bs) pure $ fromBS bs++txFromBSorExcept :: Serializable a => Tx inst (Maybe ByteString) -> Tx inst (Maybe a)+txFromBSorExcept = txCheckMap $ traverse $ \bs -> maybe (Left $ CouldNotDecodeValue $ Just bs) Right $ fromBS bs+ data SimpleValueIdentifier   = SviTopLevel ByteString         -- ^ Stored in a top-level key.   | SviHash ByteString ByteString  -- ^ Stored in a hash field.+  deriving stock (Show)  -- | Simple values, like strings, integers or enums, -- that be represented as a single bytestring.@@ -450,8 +460,9 @@ getSet :: forall ref. SimpleRef ref => ref -> ValueType ref -> RedisM (RefInstance ref) (Maybe (ValueType ref)) getSet ref val = case toIdentifier ref of   SviTopLevel keyBS ->-    fmap (fromBS =<<) . expectRight "getSet/plain"-      =<< Hedis.getset keyBS (toBS val)+    traverse rFromBSorThrow+    =<< expectRight "getSet/plain"+    =<< Hedis.getset keyBS (toBS val)    -- no native Redis call for this   SviHash _ _ -> atomically (txGet ref <* txSet ref val)@@ -492,12 +503,14 @@ incrementBy :: (SimpleRef ref, Num (ValueType ref)) => ref -> Integer -> RedisM (RefInstance ref) (ValueType ref) incrementBy ref val = fmap fromInteger . expectRight "incrementBy" =<< case toIdentifier ref of   SviTopLevel keyBS -> Hedis.incrby keyBS val-  SviHash keyBS hkeyBS -> Hedis.hincrby keyBS hkeyBS val+  SviHash keyBS hkeyBS -> fmap (fromIntegral @Int64 @Integer)+    <$> Hedis.hincrby keyBS hkeyBS (fromIntegral @Integer @Int64 val)  txIncrementBy :: (SimpleRef ref, Num (ValueType ref)) => ref -> Integer -> Tx (RefInstance ref) (ValueType ref) txIncrementBy ref val = fmap fromInteger . txWrap $ case toIdentifier ref of   SviTopLevel keyBS -> Hedis.incrby keyBS val-  SviHash keyBS hkeyBS -> Hedis.hincrby keyBS hkeyBS val+  SviHash keyBS hkeyBS -> fmap (fromIntegral @Int64 @Integer)+    <$> Hedis.hincrby keyBS hkeyBS (fromIntegral @Integer @Int64 val)  -- | Increment the value under the given ref. incrementByFloat :: (SimpleRef ref, Floating (ValueType ref)) => ref -> Double -> RedisM (RefInstance ref) (ValueType ref)@@ -505,6 +518,11 @@   SviTopLevel keyBS -> Hedis.incrbyfloat keyBS val   SviHash keyBS hkeyBS -> Hedis.hincrbyfloat keyBS hkeyBS val +txIncrementByFloat :: (SimpleRef ref, Floating (ValueType ref)) => ref -> Double -> Tx (RefInstance ref) (ValueType ref)+txIncrementByFloat ref val = fmap realToFrac . txWrap $ case toIdentifier ref of+  SviTopLevel keyBS -> Hedis.incrbyfloat keyBS val+  SviHash keyBS hkeyBS -> Hedis.hincrbyfloat keyBS hkeyBS val+ setIfNotExists :: forall ref. SimpleRef ref => ref -> ValueType ref -> RedisM (RefInstance ref) Bool setIfNotExists ref val = expectRight "setIfNotExists" =<< case toIdentifier ref of   SviTopLevel keyBS -> Hedis.setnx keyBS (toBS val)@@ -529,6 +547,9 @@       { Hedis.setSeconds      = Just ttlSec       , Hedis.setMilliseconds = Nothing       , Hedis.setCondition    = Just Hedis.Nx+      , Hedis.setUnixMilliseconds = Nothing+      , Hedis.setUnixSeconds  = Nothing+      , Hedis.setKeepTTL      = False       }  deleteIfEqual :: forall ref. SimpleRef ref => ref -> ValueType ref -> RedisM (RefInstance ref) Bool@@ -810,8 +831,10 @@     txWrap (Hedis.lrange keyBS 0 (-1))     & txFromBSMany     & fmap Just-  txValSet keyBS vs = void $ txWrap (Hedis.del [keyBS] *> Hedis.rpush keyBS (map toBS vs))-  txValDelete keyBS = void $ txWrap (Hedis.del [keyBS])+  txValSet _ [] = txThrow $ UserException "txValSet: empty list"+  txValSet keyBS (v : vs) = void $ txWrap+    (Hedis.del (keyBS :| []) *> Hedis.rpush keyBS (NE.map toBS (v :| vs)))+  txValDelete keyBS = void $ txWrap (Hedis.del $ keyBS :| [])   txValSetTTLIfExists keyBS (TTLSec ttlSec) = txWrap (Hedis.expire keyBS ttlSec)    valGet keyBS =@@ -821,12 +844,14 @@         Left badBS -> throw $ CouldNotDecodeValue (Just badBS)         Right vs -> pure (Just vs)) -  valSet keyBS vs =-    Redis (Hedis.multiExec (Hedis.del [keyBS] *> Hedis.rpush keyBS (map toBS vs)))+  valSet _ [] = throwMsg "valSet: empty list"+  valSet keyBS (v : vs) =+    Redis (Hedis.multiExec+      (Hedis.del (keyBS :| []) *> Hedis.rpush keyBS (NE.map toBS (v :| vs))))       >>= expectTxSuccess       >>= ignore @Integer   valDelete keyBS =-    Redis (Hedis.del [keyBS])+    Redis (Hedis.del (keyBS :| []))       >>= expectRight "valDelete/[a]"       >>= ignore @Integer   valSetTTLIfExists keyBS (TTLSec ttlSec) =@@ -834,16 +859,15 @@       >>= expectRight "valSetTTLIfExists/[a]"  -- | Append to a Redis list.-lAppend :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => ref -> [a] -> RedisM (RefInstance ref) ()+lAppend :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => ref -> NonEmpty a -> RedisM (RefInstance ref) () lAppend (toIdentifier -> keyBS) vals =-  Redis (Hedis.rpush keyBS (map toBS vals))+  Redis (Hedis.rpush keyBS (NE.map toBS vals))     >>= expectRight "rpush"     >>= ignore @Integer  -- | Append to a Redis list in a transaction.-txLAppend :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => ref -> [a] -> Tx (RefInstance ref) ()-txLAppend (toIdentifier -> keyBS) vals =-  void . txWrap $ Hedis.rpush keyBS (map toBS vals)+txLAppend :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => ref -> NonEmpty a -> Tx (RefInstance ref) ()+txLAppend (toIdentifier -> keyBS) vals = void . txWrap $ Hedis.rpush keyBS $ NE.map toBS vals  -- | Length of a Redis list lLength :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => ref -> RedisM (RefInstance ref) Integer@@ -852,9 +876,9 @@     >>= expectRight "llen"  -- | Prepend to a Redis list.-lPushLeft :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => ref -> [a] -> RedisM (RefInstance ref) ()+lPushLeft :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => ref -> NonEmpty a -> RedisM (RefInstance ref) () lPushLeft (toIdentifier -> keyBS) vals =-  Redis (Hedis.lpush keyBS (map toBS vals))+  Redis (Hedis.lpush keyBS (NE.map toBS vals))     >>= expectRight "lpush"     >>= ignore @Integer @@ -862,19 +886,15 @@ lPopRight :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => ref -> RedisM (RefInstance ref) (Maybe a) lPopRight (toIdentifier -> keyBS) =   Redis (Hedis.rpop keyBS)-  >>= fmap (fromBS =<<) . expectRight "rpop"+  >>= expectRight "rpop"+  >>= traverse rFromBSorThrow  -- | Pop from the right, blocking. lPopRightBlocking :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => TTL -> ref -> RedisM (RefInstance ref) (Maybe a) lPopRightBlocking (TTLSec timeoutSec) (toIdentifier -> keyBS) =-  Redis (Hedis.brpop [keyBS] timeoutSec)+  Redis (Hedis.brpop (keyBS :| []) timeoutSec)     >>= expectRight "brpop"-    >>= \case-      Nothing -> pure Nothing -- timeout-      Just (_listName, valBS) ->-        case fromBS valBS of-          Just val -> pure $ Just val-          Nothing -> throw $ CouldNotDecodeValue (Just valBS)+    >>= traverse (rFromBSorThrow . snd)  -- | Delete from a Redis list lRem :: forall ref a. (Ref ref, ValueType ref ~ [a], Serializable a) => ref -> Integer -> a -> RedisM (RefInstance ref) ()@@ -893,13 +913,11 @@     & txFromBSMany     & fmap (Just . Set.fromList) -  txValSet keyBS vs =-    void $ txWrap (-      Hedis.del [keyBS]-      *> Hedis.sadd keyBS (map toBS $ Set.toList vs)-    )+  txValSet keyBS vs = case map toBS $ Set.toList vs of+    [] -> txThrow $ UserException "txValSet: empty set"+    (x : xs) -> void $ txWrap (Hedis.del (keyBS :| []) *> Hedis.sadd keyBS (x :| xs)) -  txValDelete keyBS = void $ txWrap (Hedis.del [keyBS])+  txValDelete keyBS = void $ txWrap (Hedis.del (keyBS :| []))   txValSetTTLIfExists keyBS (TTLSec ttlSec) = txWrap (Hedis.expire keyBS ttlSec)    valGet keyBS =@@ -909,15 +927,14 @@         Left badBS -> throw $ CouldNotDecodeValue (Just badBS)         Right vs -> pure (Just $ Set.fromList vs)) -  valSet keyBS vs =-    Redis (Hedis.multiExec (-      Hedis.del [keyBS]-      *> Hedis.sadd keyBS (map toBS $ Set.toList vs)-    ))-      >>= expectTxSuccess-      >>= ignore @Integer+  valSet keyBS vs = case (map toBS $ Set.toList vs) of+    [] -> throwMsg "valSet: empty set"+    (x : xs) -> +      Redis (Hedis.multiExec (Hedis.del (keyBS :| []) *> Hedis.sadd keyBS (x :| xs)))+        >>= expectTxSuccess+        >>= ignore @Integer -  valDelete keyBS = Redis (Hedis.del [keyBS])+  valDelete keyBS = Redis (Hedis.del (keyBS :| []))     >>= expectRight "valDelete/Set a"     >>= ignore @Integer @@ -926,30 +943,26 @@       >>= expectRight "valSetTTLIfExists/Set a"  -- | Insert into a Redis set.-sInsert :: forall ref a. (Ref ref, ValueType ref ~ Set a, Serializable a) => ref -> [a] -> RedisM (RefInstance ref) ()+sInsert :: forall ref a. (Ref ref, ValueType ref ~ Set a, Serializable a) => ref -> NonEmpty a -> RedisM (RefInstance ref) () sInsert ref vals =-  Redis (Hedis.sadd (toIdentifier ref) (map toBS vals))+  Redis (Hedis.sadd (toIdentifier ref) (NE.map toBS vals))     >>= expectRight "setInsert"     >>= ignore @Integer  -- | Insert into a Redis set in a transaction.-txSInsert :: forall ref a. (Ref ref, ValueType ref ~ Set a, Serializable a) => ref -> [a] -> Tx (RefInstance ref) ()-txSInsert ref vals =-  void . txWrap-    $ Hedis.sadd (toIdentifier ref) (map toBS vals)+txSInsert :: forall ref a. (Ref ref, ValueType ref ~ Set a, Serializable a) => ref -> NonEmpty a -> Tx (RefInstance ref) ()+txSInsert ref vals = void . txWrap $ Hedis.sadd (toIdentifier ref) (NE.map toBS vals)  -- | Delete from a Redis set.-sDelete :: forall ref a. (Ref ref, ValueType ref ~ Set a, Serializable a) => ref -> [a] -> RedisM (RefInstance ref) ()+sDelete :: forall ref a. (Ref ref, ValueType ref ~ Set a, Serializable a) => ref -> NonEmpty a -> RedisM (RefInstance ref) () sDelete ref vals =-  Redis (Hedis.srem (toIdentifier ref) (map toBS vals))+  Redis (Hedis.srem (toIdentifier ref) (NE.map toBS vals))     >>= expectRight "hashSetDelete"     >>= ignore @Integer  -- | Delete from a Redis set in a transaction.-txSDelete :: forall ref a. (Ref ref, ValueType ref ~ Set a, Serializable a) => ref -> [a] -> Tx (RefInstance ref) ()-txSDelete ref vals =-  void . txWrap-    $ Hedis.srem (toIdentifier ref) (map toBS vals)+txSDelete :: forall ref a. (Ref ref, ValueType ref ~ Set a, Serializable a) => ref -> NonEmpty a -> Tx (RefInstance ref) ()+txSDelete ref vals = void . txWrap $ Hedis.srem (toIdentifier ref) (NE.map toBS vals)  -- | Check membership in a Redis set. sContains :: forall ref a. (Ref ref, ValueType ref ~ Set a, Serializable a) => ref -> a -> RedisM (RefInstance ref) Bool@@ -972,6 +985,7 @@  -- | Priority for a sorted set newtype Priority = Priority { unPriority :: Double }+  deriving newtype (Eq, Ord, Num, Real, Fractional, RealFrac)  instance Serializable Priority where   fromBS = fmap Priority . fromBS@@ -991,7 +1005,7 @@ -- | Delete from a Redis sorted set zDelete :: forall ref a. (Ref ref, ValueType ref ~ [(Priority, a)], Serializable a) => ref -> a -> RedisM (RefInstance ref) () zDelete (toIdentifier -> keyBS) val =-  Redis (Hedis.zrem keyBS [toBS val])+  Redis (Hedis.zrem keyBS (toBS val :| []))     >>= expectRight "zrem"     >>= ignore @Integer @@ -1008,6 +1022,28 @@   Redis (Hedis.zcount keyBS minScore maxScore)     >>= expectRight "zcount" +-- | Increment the value in the sorted set by the given amount. Note that if the value is not present this will add the+-- the value to the sorted list with the given amount as its priority.+zIncrBy :: forall ref a. (Ref ref, ValueType ref ~ [(Priority, a)], Serializable a) => ref -> Integer -> a -> RedisM (RefInstance ref) Priority+zIncrBy (toIdentifier -> keyBS) incr (toBS -> val)=+  Hedis.zincrby keyBS incr val+    >>= expectRight "zincrby"+    <&> Priority++-- | Remove given number of largest elements from a sorted set.+--   Available since Redis 5.0.0+zPopMax :: forall ref a. (Ref ref, ValueType ref ~ [(Priority, a)], Serializable a) => ref -> Integer -> RedisM (RefInstance ref) [(Priority, a)]+zPopMax (toIdentifier -> keyBS) cnt =+  Redis (zpopmax keyBS cnt)+  >>= expectRight "zpopmax call"+  >>= expectRight "zpopmax decode" . fromBSMany'+  where fromBSMany' = traverse $ \(valBS,sc) -> maybe (Left valBS) (Right . (Priority sc,)) $ fromBS valBS++-- | ZPOPMAX as it should be in the Hedis library (but it isn't yet)+--   Available since Redis 5.0.0+zpopmax :: Hedis.RedisCtx m f => ByteString -> Integer -> m (f [(ByteString, Double)])+zpopmax k c = Hedis.sendRequest ["ZPOPMAX", k, toBS c]+ -- | Remove given number of smallest elements from a sorted set. --   Available since Redis 5.0.0 zPopMin :: forall ref a. (Ref ref, ValueType ref ~ [(Priority, a)], Serializable a) => ref -> Integer -> RedisM (RefInstance ref) [(Priority, a)]@@ -1051,6 +1087,34 @@   >>= expectRight "zrangebyscoreLimit call"   >>= expectRight "zrangebyscoreLimit decode" . fromBSMany +zRange :: forall ref a. (Ref ref, ValueType ref ~ [(Priority, a)], Serializable a)+          => ref -> Integer -> Integer -> RedisM (RefInstance ref) [a]+zRange (toIdentifier -> keyBS) start end =+  Hedis.zrange keyBS start end+  >>= expectRight "zrange call"+  >>= expectRight "zrange decode" . fromBSMany++zRevRange :: forall ref a. (Ref ref, ValueType ref ~ [(Priority, a)], Serializable a)+          => ref -> Integer -> Integer -> RedisM (RefInstance ref) [a]+zRevRange (toIdentifier -> keyBS) start end =+  Hedis.zrevrange keyBS start end+  >>= expectRight "zrevrange call"+  >>= expectRight "zrevrange decode" . fromBSMany++-- | Scan the sorted set by reference using an optional match and count.+zScanOpts :: forall ref a. (Ref ref, ValueType ref ~ [(Priority, a)], Serializable a)+          => ref -> Maybe Text -> Maybe Integer -> RedisM (RefInstance ref) [a]+zScanOpts (toIdentifier -> keyBS) mMatch mCount =+  Hedis.zscanOpts keyBS Hedis.cursor0 Hedis.ScanOpts { Hedis.scanMatch = toBS <$> mMatch, Hedis.scanCount = mCount}+  >>= expectRight "zscanOpts call"+  >>= expectRight "zscanOpts decode" . fromBSMany . map fst . snd++zUnionStoreWeights :: forall ref a. (Ref ref, ValueType ref ~ [(Priority, a)], Serializable a) => ref -> NonEmpty (ref, Double) -> RedisM (RefInstance ref) ()+zUnionStoreWeights (toIdentifier -> keyBS) refWithWeights =+  Hedis.zunionstoreWeights keyBS [(toIdentifier k, d) | (k, d) <- NE.toList refWithWeights] Hedis.Sum+  >>= expectRight "zUnionStoreWeights call"+  >>= ignore @Integer+ parseMap :: (Ord k, Serializable k, Serializable v)   => [(ByteString, ByteString)] -> Maybe (Map k v) parseMap kvsBS = Map.fromList <$> sequence@@ -1071,14 +1135,15 @@           . parseMap         ) -  txValSet keyBS m =-    void $ txWrap (-      Hedis.del [keyBS]-      *> Hedis.hmset keyBS-        [(toBS ref, toBS val) | (ref, val) <- Map.toList m]-    )+  txValSet keyBS m = case [(toBS ref, toBS val) | (ref, val) <- Map.toList m] of+    [] -> txThrow $ UserException "txValSet: empty map"+    (v : vs) -> +      void $ txWrap (+        Hedis.del (keyBS :| [])+        *> Hedis.hmset keyBS (v :| vs)+      ) -  txValDelete keyBS = void . txWrap $ Hedis.del [keyBS]+  txValDelete keyBS = void . txWrap $ Hedis.del (keyBS :| [])   txValSetTTLIfExists keyBS (TTLSec ttlSec) =     txWrap $ Hedis.expire keyBS ttlSec @@ -1089,17 +1154,18 @@         Just m -> pure (Just m)         Nothing -> throw $ CouldNotDecodeValue Nothing -  valSet keyBS m =-    Redis (Hedis.multiExec (-      Hedis.del [keyBS]-      *> Hedis.hmset keyBS-        [(toBS ref, toBS val) | (ref, val) <- Map.toList m]-    ))-      >>= expectTxSuccess-      >>= expect "valSet/Map k v" Hedis.Ok+  valSet keyBS m = case [(toBS ref, toBS val) | (ref, val) <- Map.toList m] of+    [] -> throwMsg "Redis.valSet: map must not be empty"+    (x : xs) ->+      Redis (Hedis.multiExec (+        Hedis.del (keyBS :| [])+        *> Hedis.hmset keyBS (x :| xs)+      ))+        >>= expectTxSuccess+        >>= expect "valSet/Map k v" Hedis.Ok    valDelete keyBS =-    Redis (Hedis.del [keyBS])+    Redis (Hedis.del (keyBS :| []))       >>= expectRight "valDelete/Map k v"       >>= ignore @Integer @@ -1132,6 +1198,7 @@   , ValueType ref ~ Map k v   , Serializable k   , SimpleValue (RefInstance ref) v+  , Value (RefInstance ref) v   ) => Ref (MapItem ref k v) where    type ValueType (MapItem ref k v) = v@@ -1155,6 +1222,7 @@   , ValueType ref ~ Record fieldF   , SimpleValue (RefInstance ref) val   , RecordField fieldF+  , Value (RefInstance ref) val   ) => Ref (RecordItem ref fieldF val) where    type ValueType (RecordItem ref fieldF val) = val@@ -1192,11 +1260,11 @@   type Identifier (Record fieldF) = ByteString   txValGet _ = error "Record is not meant to be read"   txValSet _ _ = error "Record is not meant to be written"-  txValDelete keyBS = void . txWrap $ Hedis.del [keyBS]+  txValDelete keyBS = void . txWrap $ Hedis.del (keyBS :| [])   txValSetTTLIfExists keyBS (TTLSec ttlSec) = txWrap $ Hedis.expire keyBS ttlSec   valGet _ = error "Record is not meant to be read"   valSet _ _ = error "Record is not meant to be written"-  valDelete keyBS = Hedis.del [keyBS]+  valDelete keyBS = Hedis.del (keyBS :| [])     >>= expectRight "valDelete/Record" >>= ignore @Integer   valSetTTLIfExists keyBS (TTLSec ttlSec) =     Hedis.expire keyBS ttlSec >>= expectRight "setTTLIfExists/Record"@@ -1214,11 +1282,11 @@   type Identifier (PubSub msg) = ByteString   txValGet _ = error "PubSub is not meant to be read"   txValSet _ _ = error "PubSub is not meant to be written"-  txValDelete keyBS = void . txWrap $ Hedis.del [keyBS]+  txValDelete keyBS = void . txWrap $ Hedis.del (keyBS :| [])   txValSetTTLIfExists keyBS (TTLSec ttlSec) = txWrap $ Hedis.expire keyBS ttlSec   valGet _ = error "PubSub is not meant to be read"   valSet _ _ = error "PubSub is not meant to be written"-  valDelete keyBS = Hedis.del [keyBS]+  valDelete keyBS = Hedis.del (keyBS :| [])     >>= expectRight "valDelete/PubSub" >>= ignore @Integer   valSetTTLIfExists keyBS (TTLSec ttlSec) =     Hedis.expire keyBS ttlSec >>= expectRight "setTTLIfExists/PubSub"

@@ -23,40 +23,68 @@   , defaultMetaParams   , ExclusiveLock, withExclusiveLock   , ShareableLock, withShareableLock, LockSharing(..)+  , ExclusiveLockAcquireTimeout(..)+  , ShareableLockAcquireTimeout(..)   )   where  import GHC.Generics-import Data.Functor     ( void )-import Data.Kind        ( Type )-import Data.Maybe       ( fromMaybe )-import Data.Time        ( NominalDiffTime, addUTCTime, getCurrentTime )-import Data.Set         ( Set )-import Data.ByteString  ( ByteString )+import Data.Functor       ( void )+import Data.List.NonEmpty ( NonEmpty (..) )+import Data.Kind          ( Type )+import Data.Maybe         ( fromMaybe )+import Data.Time          ( NominalDiffTime, addUTCTime, getCurrentTime )+import Data.Set           ( Set )+import Data.ByteString    ( ByteString ) import qualified Data.Set as Set import qualified Data.ByteString.Char8 as BS  import System.Random    ( randomIO )  import Control.Concurrent  ( threadDelay, myThreadId )+import Control.Exception   ( Exception ) import Control.Monad.Fix   ( fix ) import Control.Monad.Catch ( MonadThrow(..), MonadCatch(..), MonadMask(..), throwM, finally ) import Control.Monad.IO.Class ( liftIO, MonadIO )  import qualified Database.Redis.Schema as Redis +data ExclusiveLockAcquireTimeout = ExclusiveLockAcquireTimeout+  { elatRefIdentifier :: Redis.SimpleValueIdentifier+  , elatLockParams :: LockParams+  , elatOurId :: LockOwnerId+  }+  deriving stock (Show)++instance Exception ExclusiveLockAcquireTimeout++data ShareableLockAcquireTimeout = ShareableLockAcquireTimeout+  { slatRefIdentifier :: ByteString+  , slatLockParams :: ShareableLockParams+  , slatSharing :: LockSharing+  , slatOurId :: LockOwnerId+  }+  deriving stock (Show)++instance Exception ShareableLockAcquireTimeout+ data LockParams = LockParams   { lpMeanRetryInterval :: NominalDiffTime   , lpAcquireTimeout    :: NominalDiffTime   , lpLockTTL           :: Redis.TTL   }+  deriving stock (Show)  -- | ID of the process that owns the Redis lock. newtype LockOwnerId = LockOwnerId { _unLockOwnerId :: ByteString }+  deriving stock (Show)   deriving newtype (Eq, Ord, Redis.Serializable) instance Redis.Value inst LockOwnerId instance Redis.SimpleValue inst LockOwnerId +data AcquireResult = Acquired | AcquireTimeout+  deriving stock (Show)+ -------------------- -- Exclusive lock -- --------------------@@ -82,6 +110,9 @@ -- -- * For exclusive locks, 'withExclusiveLock' is more efficient. --+-- This throws 'ExclusiveLockAcquireTimeout' if we fail to acquire the lock before the+-- timeout specified in the 'LockParams'.+-- withExclusiveLock ::   ( MonadCatch m, MonadThrow m, MonadMask m, MonadIO m   , Redis.Ref ref, Redis.ValueType ref ~ ExclusiveLock@@ -92,9 +123,14 @@   -> m a         -- ^ The action to perform under lock   -> m a withExclusiveLock redis lp ref action = do-  exclusiveLockAcquire redis lp ref >>= \case-    Nothing -> throwM Redis.LockAcquireTimeout-    Just ourId -> action `finally` exclusiveLockRelease redis ref ourId+  (result, ourId) <- exclusiveLockAcquire redis lp ref+  case result of+    AcquireTimeout -> throwM ExclusiveLockAcquireTimeout+      { elatRefIdentifier = Redis.toIdentifier ref+      , elatLockParams = lp+      , elatOurId = ourId+      }+    Acquired -> action `finally` exclusiveLockRelease redis ref ourId  -- | Acquire a distributed exclusive lock. -- Returns Nothing on timeout. Otherwise it returns the unique client ID used for the lock.@@ -102,7 +138,7 @@   ( MonadCatch m, MonadThrow m, MonadMask m, MonadIO m   , Redis.Ref ref, Redis.ValueType ref ~ ExclusiveLock   )-  => Redis.Pool (Redis.RefInstance ref) -> LockParams -> ref -> m (Maybe LockOwnerId)+  => Redis.Pool (Redis.RefInstance ref) -> LockParams -> ref -> m (AcquireResult, LockOwnerId) exclusiveLockAcquire redis lp ref = do   -- this is unique only if we have only one instance of HConductor running   ourId <- LockOwnerId . BS.pack . show <$> liftIO myThreadId  -- unique client id@@ -110,13 +146,13 @@   fix $ \ ~retry -> do  -- ~ makes the lambda lazy     tsNow <- liftIO getCurrentTime     if tsNow >= tsDeadline-      then return Nothing  -- didn't manage to acquire the lock before timeout+      then return (AcquireTimeout, ourId)  -- didn't manage to acquire the lock before timeout       else do         -- set the lock if it does not exist         didNotExist <- Redis.run redis $           Redis.setIfNotExistsTTL ref (ExclusiveLock ourId) (lpLockTTL lp)         if didNotExist-          then return (Just ourId)  -- everything went well+          then return (Acquired, ourId)  -- everything went well           else do             -- someone got there first; wait a bit and try again             fuzzySleep (lpMeanRetryInterval lp)@@ -225,6 +261,7 @@   { slpParams :: LockParams   , slpMetaParams :: LockParams   }+  deriving (Show)  defaultMetaParams :: LockParams defaultMetaParams = LockParams@@ -244,6 +281,9 @@ -- -- * For exclusive locks, withExclusiveLock is more efficient. --+-- This throws 'ShareableLockAcquireTimeout' if we fail to acquire the lock before the+-- timeout specified in the 'ShareableLockParams'.+-- -- NOTE: the shareable lock seems to have quite a lot of performance overhead. -- Always benchmark first whether the exclusive lock would perform better in your scenario, -- even when a shareable lock would be sufficient in theory.@@ -258,18 +298,24 @@   -> ref         -- ^ Lock ref   -> m a         -- ^ The action to perform under lock   -> m a-withShareableLock redis slp lockSharing ref action =-  shareableLockAcquire redis slp lockSharing ref >>= \case-    Nothing -> throwM Redis.LockAcquireTimeout-    Just ourId -> action-      `finally` shareableLockRelease redis slp ref lockSharing ourId+withShareableLock redis slp lockSharing ref action = do+  (result, ourId) <- shareableLockAcquire redis slp lockSharing ref+  case result of+    AcquireTimeout -> throwM ShareableLockAcquireTimeout+      { slatRefIdentifier = Redis.toIdentifier ref+      , slatLockParams = slp+      , slatSharing = lockSharing+      , slatOurId = ourId+      }+    Acquired ->+      action `finally` shareableLockRelease redis slp ref lockSharing ourId  shareableLockAcquire ::   forall m ref.   ( MonadCatch m, MonadThrow m, MonadMask m, MonadIO m   , Redis.Ref ref, Redis.ValueType ref ~ ShareableLock   , Redis.SimpleValue (Redis.RefInstance ref) (MetaLock ref)-  ) => Redis.Pool (Redis.RefInstance ref) -> ShareableLockParams -> LockSharing -> ref -> m (Maybe LockOwnerId)+  ) => Redis.Pool (Redis.RefInstance ref) -> ShareableLockParams -> LockSharing -> ref -> m (AcquireResult, LockOwnerId) shareableLockAcquire redis slp lockSharing ref = do   -- this is unique only if we have only one instance of HConductor running   ourId <- LockOwnerId . BS.pack . show <$> liftIO myThreadId  -- unique client id@@ -277,7 +323,7 @@   fix $ \ ~retry -> do  -- ~ makes the lambda lazy     tsNow <- liftIO getCurrentTime     if tsNow >= tsDeadline-      then return Nothing  -- didn't manage to acquire the lock before timeout+      then return (AcquireTimeout, ourId)  -- didn't manage to acquire the lock before timeout       else do         -- acquire the lock if possible, using the meta lock to synchronise access         success <- withExclusiveLock redis (slpMetaParams slp) (MetaLock ref) $@@ -294,7 +340,7 @@               -- we want to share               -- so we can acquire               Just Shared | lockSharing == Shared -> do-                Redis.sInsert (lockField LockFieldOwners) [ourId]+                Redis.sInsert (lockField LockFieldOwners) (ourId :| [])                 return True                -- can't acquire lock otherwise@@ -304,7 +350,7 @@           then do             -- everything went well, set ttl and return             Redis.run redis $ Redis.setTTL ref (lpLockTTL $ slpParams slp)-            return (Just ourId)+            return (Acquired, ourId)           else do             -- someone got there first; wait a bit and try again             fuzzySleep $ lpMeanRetryInterval (slpParams slp)@@ -339,7 +385,7 @@             -- delete the whole lock             then Redis.delete_ ref             -- just remove ourselves from the list of owners-            else Redis.sDelete (lockField LockFieldOwners) [ourId]+            else Redis.sDelete (lockField LockFieldOwners) (ourId :| [])   where     lockField :: LockFieldName ty -> LockField (Redis.RefInstance ref) ty     lockField = LockField (Redis.toIdentifier ref)

@@ -0,0 +1,378 @@+{-# LANGUAGE AllowAmbiguousTypes #-}  -- for remoteJobWorker+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE UndecidableInstances #-}+module Database.Redis.Schema.RemoteJob (+  -- * Types+  WorkerId (..),+  WorkerHandle,+  RemoteJobError (..),+  JobQueue (..),++  -- * Main functionality+  runRemoteJob,+  runRemoteJobAsync,+  remoteJobWorker,+  withRemoteJobWorker,+  gracefulShutdown,++  -- * Inspection+  countWorkers,+  queueLength,+  countRunningJobs,+) where++import Data.Binary ( decode, encode, Binary(..) )+import Data.Bifunctor as BF ( second )+import Data.List.NonEmpty ( NonEmpty (..) )+import Data.Kind ( Type )+import Data.Maybe ( isJust )+import Data.Proxy ( Proxy(..) )+import Data.String ( IsString )+import Data.Text ( Text )+import Data.Time ( UTCTime, getCurrentTime, addUTCTime )+import Data.Time.Clock.POSIX ( utcTimeToPOSIXSeconds )+import Data.UUID ( UUID )+import Data.UUID.V4 ( nextRandom )+import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy as BSL+import qualified Data.Set as Set++import Database.Redis ( ConnectionLostException )+import Database.Redis.Schema as Redis++import Control.Concurrent.MonadIO+import Control.Monad ( when, forever )+import Control.Monad.Catch+import Control.Exception ( SomeAsyncException )++import GHC.Generics ( Generic )+++-- | Errors that can occur in the remote job running.+data RemoteJobError+  = RemoteJobException String+  | NoActiveWorkers+  | Timeout+  deriving ( Show, Generic )+instance Binary RemoteJobError++instance Serializable RemoteJobError where+  fromBS = readBinary+  toBS = showBinary++-- | Identifier for the worker process, is only used for inspecting the queue+newtype WorkerId = WorkerId { unWorkerId :: Text }+  deriving newtype ( Show, Eq, Ord, IsString, Serializable, Binary )++-- | Handle of the worker process, which can be used in the graceful shutdown procedure+data WorkerHandle = WorkerHandle (MVar ()) ThreadId++class JobQueue jq where+  -- | The remote job protocol, a list of 'i -> o' entries indicating+  --   this queue can contains jobs taking type 'i' as input and+  --   returning type 'o'. Both 'i' and 'o' must have a binary instance.+  type RPC jq :: [Type]++  -- | Prefix for the Redis keys+  keyPrefix :: BS.ByteString++  -- | Which Redis instance the queue lives in.+  type RedisInstance jq :: Instance+  type RedisInstance jq = DefaultInstance+++-- | Type for representing a job request in Redis+data Job = Job+  { jobId         :: UUID+  , jobHandlerIdx :: Int+  , jobInput      :: BSL.ByteString+  } deriving ( Eq, Ord )++instance Serializable Job where+  toBS j = toBS (jobId j, jobHandlerIdx j, jobInput j)+  fromBS = fmap (\(jid,jidx,jinp) -> Job jid jidx jinp) . fromBS++-- | This queue contains many requests.+-- There is only one request queue and it's read by all workers.+data RequestQueue jq = RequestQueue+instance JobQueue jq => Ref (RequestQueue jq) where+  type RefInstance (RequestQueue jq) = RedisInstance jq+  type ValueType (RequestQueue jq) = [(Priority, Job)]+  toIdentifier RequestQueue =+    colonSep [keyPrefix @jq, "requests"]++-- | This set contains the requests that are currently being processed.+data RunningJobs jq = RunningJobs+instance JobQueue jq => Ref (RunningJobs jq) where+  type RefInstance (RunningJobs jq) = RedisInstance jq+  type ValueType (RunningJobs jq) = Set.Set (WorkerId, Job)+  toIdentifier RunningJobs =+    colonSep [keyPrefix @jq, "running"]++-- | A box that contains only one response.+-- For every response, a unique box is created (tagged with job ID).+newtype ResultBox jq = ResultBox UUID+instance JobQueue jq => Ref (ResultBox jq) where+  type RefInstance (ResultBox jq) = RedisInstance jq+  type ValueType (ResultBox jq) = [Either RemoteJobError BSL.ByteString]+  toIdentifier (ResultBox uuid) =+    colonSep [keyPrefix @jq, "result", toBS uuid]++-- | A registry of all active workers+data Workers jq = Workers+instance JobQueue jq => Ref (Workers jq) where+  type RefInstance (Workers jq) = RedisInstance jq+  type ValueType (Workers jq) = [(Priority, WorkerId)]+  toIdentifier Workers =+    Redis.colonSep [keyPrefix @jq, "workers"]++-- | Type class to check where in the 'RPC' list a i->o job occurs, which+--   is then used together with 'CanHandle' to use the right handler.+class HasHandler (i :: Type) (o :: Type) (xs :: [Type]) where+  handlerIdx :: Proxy (i -> o) -> Proxy xs -> Int++instance (HasHandler' i o xs (IsHead (i -> o) xs)) => HasHandler i o xs where+  handlerIdx = handlerIdx' (Proxy @(IsHead (i -> o) xs))++class HasHandler' (i :: Type) (o :: Type) (xs :: [Type]) (isHead :: Bool) where+  handlerIdx' :: Proxy isHead -> Proxy (i -> o) -> Proxy xs -> Int++instance HasHandler' i o ((i -> o) ': xs) 'True where+  handlerIdx' _ _ _ = 0++instance HasHandler i o xs => HasHandler' i o (x ': xs) 'False where+  handlerIdx' _ _ _ = 1 + handlerIdx (Proxy @(i -> o)) (Proxy @xs)++type family IsHead (x :: Type) (xs :: [Type]) :: Bool where+  IsHead x (x ': _) = 'True+  IsHead x _        = 'False+++-- | An instance 'CanHandle m xs' means that the list xs of i->o jobs+--   can be handled in monad m, e.g. there exists Binary instances for all+--   i and o, and the instances take care of encoding and decoding as the+--   right type.+class CanHandle (m :: Type -> Type) (xs :: [Type]) where+  type HandleTy m xs r :: Type+  doHandle :: Proxy m -> Proxy xs -> ((Int -> BSL.ByteString -> m BSL.ByteString) -> m r) -> HandleTy m xs r++instance CanHandle m '[] where+  type HandleTy m '[] r = m r+  doHandle Proxy Proxy cont = cont $ \_ _ -> error "remoteJobWorker: protocol broken"++instance (Monad m, Binary i, Binary o, CanHandle m xs) => CanHandle m ((i -> o) ': xs) where+  type HandleTy m ((i -> o) ': xs) r = (i -> m o) -> HandleTy m xs r+  doHandle Proxy Proxy cont f = doHandle (Proxy @m) (Proxy @xs) (cont . g) where+    g handler i bsi+      | i == 0    = encode <$> f (decode bsi)+      | otherwise = handler (i - 1) bsi++-- | Run a job on a remote worker. This will block until a 'remoteJobWorker' process picks up the+--   task. The 'Double' argument is the priority, jobs with a lower priority are picked up earlier.+runRemoteJob ::+  forall q i o m.+  (MonadCatch m, MonadIO m, JobQueue q, HasHandler i o (RPC q), Binary i, Binary o) =>+  Bool -> Pool (RedisInstance q) -> Priority -> i -> m (Either RemoteJobError o)+runRemoteJob waitForWorkers pool prio a = do+  -- Check that there are active workers+  abort <-+    if waitForWorkers+    then return False+    else (==0) <$> run pool (countWorkers @q)++  if abort then return $ Left NoActiveWorkers+  else do+    -- Add the job+    jid <- liftIO nextRandom+    let job = Job+          { jobId = jid+          , jobHandlerIdx = handlerIdx (Proxy @(i -> o)) (Proxy @(RPC q))+          , jobInput = encode a+          }++    -- Add to the queue and wait for the result. If any exception occurs at this point+    -- (which is then likely an async exception), we remove the element from the queue,+    -- because we will not listen to the result anymore anyway.+    popResult <- run pool+      ( do zInsert (RequestQueue @q) [(prio,job)]+           lPopRightBlocking 0 (ResultBox @q jid)+      ) `onException`+      run pool (zDelete (RequestQueue @q) job)++    -- Now look at the result and decode it.+    return $ case popResult of+      Just r  -> BF.second decode r+      Nothing -> Left Timeout++-- | Run a job on a remote worker but do not wait for any results. This assumes the remote+--   job has some side-effect, which is executed by a 'remoteJobWorker' process that picks+--   up this task.+runRemoteJobAsync ::+  forall q i m.+  (MonadCatch m, MonadIO m, JobQueue q, HasHandler i () (RPC q), Binary i) =>+  Pool (RedisInstance q) -> Priority -> i -> m ()+runRemoteJobAsync pool prio a = do+  -- Add to the queue and forget about it.+  jid <- liftIO nextRandom+  let job = Job+        { jobId = jid+        , jobHandlerIdx = handlerIdx (Proxy @(i -> ())) (Proxy @(RPC q))+        , jobInput = encode a+        }+  run pool $ zInsert (RequestQueue @q) [(prio,job)]++-- | The actual worker loop, this generalizes over 'remoteJobWorker' and 'forkRemoteJobWorker'+remoteJobWorker' :: forall q m r. (MonadIO m, MonadCatch m, MonadMask m, JobQueue q, CanHandle m (RPC q)) =>+  (MVar () -> m () -> m r) -> WorkerId -> Pool (RedisInstance q) -> (SomeException -> m ()) -> HandleTy m (RPC q) r+remoteJobWorker' cont wid pool logger = doHandle (Proxy @m) (Proxy @(RPC q)) $ \handler -> do+  -- MVar that is used for the graceful shutdown procedure. When it is full, the worker+  -- thread is not doing anything and can be killed. As soon as the worker starts working+  -- it takes the value and puts it back when the work is done.+  workerFree <- liftIO $ newMVar ()+  let+    -- Main loop, pop elements from the queue and handle them+    loop :: m ()+    loop = run pool (bzPopMin (RequestQueue @q) 0) >>= \case+      Just (_, job) -> do+        -- Update the RunningJobs queue at the start and end of this block,+        -- and keep the workerFree var up to date+        bracket_+          (run pool (sInsert (RunningJobs @q) ((wid,job) :| [])) >> liftIO (takeMVar workerFree))+          (run pool (sDelete (RunningJobs @q) ((wid,job) :| [])) >> liftIO (putMVar workerFree ())) $ do+            -- Call the actual handler+            resp <- fmap Right (handler (jobHandlerIdx job) (jobInput job))+                    `catchAll`+                    (return . Left)++            -- Send back the result+            let bso = case resp of+                  Left e  -> Left $ RemoteJobException $ show e+                  Right b -> Right b+            run pool $ do+              let box = ResultBox @q (jobId job)+              lPushLeft box (bso :| [])+              -- set ttl to ensure the data is not left behind in case of crashes,+              -- the caller should be awaiting this already, so it's either read+              -- directly or it is never read.+              setTTLIfExists_ box (5 * Redis.second)++            -- Check for exceptions+            case resp of+              Right _ -> return ()+              Left e -> do+                -- Call the parent logger+                logger e++                -- And in case of an async exception, rethrow+                let mbAsync :: Maybe SomeAsyncException+                    mbAsync = fromException e+                when (isJust mbAsync) $ throwM e++        -- Sleep for a tiny bit, to allow the graceful shutdown procedure to interrupt when needed+        liftIO $ threadDelay 1000 -- 1ms+        loop++      -- With BRPOP and no timeout there should always be a result+      _ ->  error "remoteJobWorker: Got no result with BRPOP and timeout 0"++    -- Fork a keep-alive loop that updates our latest-seen time every+    -- 5 seconds. We use the current UTC timestamp as priority, so+    -- that we can efficiently count the servers that checked in recently.+    signup = liftIO $ forkIO $ forever $ do+      t <- liftIO getCurrentTime+      run pool $ zInsert (Workers @q) [(utcTimeToPriority t, wid)]+      threadDelay 5_000_000 -- 5s++    -- Kill the keep-alive loop and remove ourselves from the list.+    signout tid = do+      liftIO $ killThread tid+      run pool $ zDelete (Workers @q) wid++  -- Signup and signout in an exception-safe way.+  -- When the connection is lost (which will also throw exceptions+  -- in signup/signout), we sleep for a while and try again+  let outerLoop = bracket signup signout (const loop)+        `catch`+        \(e :: ConnectionLostException) -> do+          -- Make sure to log the exception+          logger (toException e)++          -- Sleep for 10s+          liftIO $ threadDelay 10_000_000++          -- And run the loop again+          outerLoop++  -- Pass the continuation the function to start up the outer loop+  cont workerFree outerLoop+++-- | Worker for handling jobs from the queue. The first function that matches the given input/output types+--   will be executed. Multiple workers can be run in parallel.+--   When exceptions appear in the handling function, this exception will be sent to the caller+--   as a String and this exception is thrown from the worker process.+remoteJobWorker :: forall q m. (MonadIO m, MonadCatch m, MonadMask m, JobQueue q, CanHandle m (RPC q)) =>+  WorkerId -> Pool (RedisInstance q) -> (SomeException -> m ()) -> HandleTy m (RPC q) ()+remoteJobWorker = remoteJobWorker' @q cont where+  cont :: MVar () -> m () -> m ()+  cont _ runLoop = runLoop++-- | Forking version of 'remoteJobWorker', which forks a thread to do the work and returns a 'WorkerHandle'+--   that can be passed to 'gracefulShutdown' to end the worker thread in a graceful way.+--   This function ensures that on exceptions, the worker is cleaned up properly.+withRemoteJobWorker :: forall q m a. (HasFork m, MonadIO m, MonadCatch m, MonadMask m, JobQueue q, CanHandle m (RPC q)) =>+  WorkerId -> Pool (RedisInstance q) -> (SomeException -> m ()) -> (WorkerHandle -> m a) -> HandleTy m (RPC q) a+withRemoteJobWorker wid pool logger outerCont = remoteJobWorker' @q cont wid pool logger where+  cont :: MVar () -> m () -> m a+  cont workerFree runLoop = do+    tid <- fork runLoop+    let hd = WorkerHandle workerFree tid+    outerCont hd+      `finally` -- on exceptions or when the outer thread finishes, make sure to clean up+      hardShutdown hd++-- | Gracefully shut down the worker, which means waiting for the current job to complete+--   There is a tiny race condition, so there is a small chance the worker just took+--   a new job when it is killed.+gracefulShutdown :: MonadIO m => WorkerHandle -> m ()+gracefulShutdown (WorkerHandle workerFree tid) = liftIO $ do+  takeMVar workerFree+  killThread tid++-- | Send an async exception to the worker thread to kill it and clean up. The remote job+--   caller will receive a 'RemoteJobException' if a job is running.+hardShutdown :: MonadIO m => WorkerHandle -> m ()+hardShutdown (WorkerHandle _ tid) = liftIO $ killThread tid++-- | Returns the number of workers that are currently connected to the job queue.+countWorkers :: forall jq. JobQueue jq => RedisM (RedisInstance jq) Integer+countWorkers = do+  -- Count workers that checked in at most 10s ago. Workers are supposed to+  -- do this every 5 seconds, so we allow missing one beat.+  t <- liftIO getCurrentTime+  zCount (Workers @jq) (utcTimeToPriority $ addUTCTime (-10) t) maxBound++-- | Helper for worker list, which converts the current timestamp to a priority+utcTimeToPriority :: UTCTime -> Priority+utcTimeToPriority = Priority . realToFrac . utcTimeToPOSIXSeconds++-- | Returns the number of jobs that are currently queued+queueLength :: forall jq. JobQueue jq => RedisM (RedisInstance jq) Integer+queueLength = zSize (RequestQueue @jq)++-- | Returns the number of jobs that are currently being processed+countRunningJobs :: forall jq. JobQueue jq => RedisM (RedisInstance jq) Integer+countRunningJobs = sSize (RunningJobs @jq)