diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -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`.
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -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
 
diff --git a/redis-schema.cabal b/redis-schema.cabal
--- a/redis-schema.cabal
+++ b/redis-schema.cabal
@@ -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
diff --git a/src/Database/Redis/Schema.hs b/src/Database/Redis/Schema.hs
--- a/src/Database/Redis/Schema.hs
+++ b/src/Database/Redis/Schema.hs
@@ -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"
diff --git a/src/Database/Redis/Schema/Lock.hs b/src/Database/Redis/Schema/Lock.hs
--- a/src/Database/Redis/Schema/Lock.hs
+++ b/src/Database/Redis/Schema/Lock.hs
@@ -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)
diff --git a/src/Database/Redis/Schema/RemoteJob.hs b/src/Database/Redis/Schema/RemoteJob.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Redis/Schema/RemoteJob.hs
@@ -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)