feat: isolation level for roles/functions

CHANGELOG.md

@@ -37,6 +37,9 @@ This project adheres to [Semantic Versioning](http://semver.org/).
      `ALTER ROLE anon SET statement_timeout TO '5s'` will result in that `statement_timeout` getting applied for that role.
    - Works when switching roles when a JWT is sent
    - Settings can be reloaded with `NOTIFY pgrst, 'reload config'`.
+ - #2468, Configurable transaction isolation level with `default_transaction_isolation` - @steve-chavez
+   - Can be set per function `create function .. set default_transaction_isolation = 'repeatable read'`
+   - Or per role `alter role .. set default_transaction_isolation = 'serializable'`
 
 ### Fixed
 

src/PostgREST/App.hs

@@ -27,6 +27,7 @@ import Network.Wai.Handler.Warp (defaultSettings, setHost, setPort,
                                  setServerName)
 import System.Posix.Types       (FileMode)
 
+import qualified Data.HashMap.Strict        as HM
 import qualified Hasql.Pool                 as SQL
 import qualified Hasql.Transaction.Sessions as SQL
 import qualified Network.Wai                as Wai
@@ -44,16 +45,17 @@ import qualified PostgREST.Query            as Query
 import qualified PostgREST.Response         as Response
 import qualified PostgREST.Workers          as Workers
 
-import PostgREST.ApiRequest       (Action (..), ApiRequest (..),
-                                   Mutation (..), Target (..))
-import PostgREST.AppState         (AppState)
-import PostgREST.Auth             (AuthResult (..))
-import PostgREST.Config           (AppConfig (..))
-import PostgREST.Config.PgVersion (PgVersion (..))
-import PostgREST.Error            (Error)
-import PostgREST.Query            (DbHandler)
-import PostgREST.SchemaCache      (SchemaCache (..))
-import PostgREST.Version          (prettyVersion)
+import PostgREST.ApiRequest          (Action (..), ApiRequest (..),
+                                      Mutation (..), Target (..))
+import PostgREST.AppState            (AppState)
+import PostgREST.Auth                (AuthResult (..))
+import PostgREST.Config              (AppConfig (..))
+import PostgREST.Config.PgVersion    (PgVersion (..))
+import PostgREST.Error               (Error)
+import PostgREST.Query               (DbHandler)
+import PostgREST.SchemaCache         (SchemaCache (..))
+import PostgREST.SchemaCache.Routine (Routine (..))
+import PostgREST.Version             (prettyVersion)
 
 import Protolude hiding (Handler)
 
@@ -152,11 +154,11 @@ postgrestResponse appState conf@AppConfig{..} maybeSchemaCache pgVer authResult@
   Response.optionalRollback conf apiRequest $
     handleRequest authResult conf appState (Just authRole /= configDbAnonRole) configDbPreparedStatements pgVer apiRequest sCache
 
-runDbHandler :: AppState.AppState -> SQL.Mode -> Bool -> Bool -> DbHandler b -> Handler IO b
-runDbHandler appState mode authenticated prepared handler = do
+runDbHandler :: AppState.AppState -> Maybe Text -> SQL.Mode -> Bool -> Bool -> DbHandler b -> Handler IO b
+runDbHandler appState isoLvl mode authenticated prepared handler = do
   dbResp <- lift $ do
     let transaction = if prepared then SQL.transaction else SQL.unpreparedTransaction
-    res <- AppState.usePool appState . transaction SQL.ReadCommitted mode $ runExceptT handler
+    res <- AppState.usePool appState . transaction (toIsolationLevel isoLvl) mode $ runExceptT handler
     whenLeft res (\case
       SQL.AcquisitionTimeoutUsageError -> AppState.debounceLogAcquisitionTimeout appState -- this can happen rapidly for many requests, so we debounce
       _ -> pure ())
@@ -167,42 +169,48 @@ runDbHandler appState mode authenticated prepared handler = do
       mapLeft (Error.PgError authenticated) dbResp
 
   liftEither resp
+  where
+    toIsolationLevel = \case
+      Nothing                -> SQL.ReadCommitted
+      Just "repeatable read" -> SQL.RepeatableRead
+      Just "serializable"    -> SQL.Serializable
+      _                      -> SQL.ReadCommitted
 
 handleRequest :: AuthResult -> AppConfig -> AppState.AppState -> Bool -> Bool -> PgVersion -> ApiRequest -> SchemaCache -> Handler IO Wai.Response
 handleRequest AuthResult{..} conf appState authenticated prepared pgVer apiReq@ApiRequest{..} sCache =
   case (iAction, iTarget) of
     (ActionRead headersOnly, TargetIdent identifier) -> do
       wrPlan <- liftEither $ Plan.wrappedReadPlan identifier conf sCache apiReq
-      resultSet <- runQuery (Plan.wrTxMode wrPlan) $ Query.readQuery wrPlan conf apiReq
+      resultSet <- runQuery roleIsoLvl (Plan.wrTxMode wrPlan) $ Query.readQuery wrPlan conf apiReq
       return $ Response.readResponse headersOnly identifier apiReq resultSet
 
     (ActionMutate MutationCreate, TargetIdent identifier) -> do
       mrPlan <- liftEither $ Plan.mutateReadPlan MutationCreate apiReq identifier conf sCache
-      resultSet <- runQuery (Plan.mrTxMode mrPlan) $ Query.createQuery mrPlan apiReq conf
+      resultSet <- runQuery roleIsoLvl (Plan.mrTxMode mrPlan) $ Query.createQuery mrPlan apiReq conf
       return $ Response.createResponse identifier mrPlan apiReq resultSet
 
     (ActionMutate MutationUpdate, TargetIdent identifier) -> do
       mrPlan <- liftEither $ Plan.mutateReadPlan MutationUpdate apiReq identifier conf sCache
-      resultSet <- runQuery (Plan.mrTxMode mrPlan) $ Query.updateQuery mrPlan apiReq conf
+      resultSet <- runQuery roleIsoLvl (Plan.mrTxMode mrPlan) $ Query.updateQuery mrPlan apiReq conf
       return $ Response.updateResponse apiReq resultSet
 
     (ActionMutate MutationSingleUpsert, TargetIdent identifier) -> do
       mrPlan <- liftEither $ Plan.mutateReadPlan MutationSingleUpsert apiReq identifier conf sCache
-      resultSet <- runQuery (Plan.mrTxMode mrPlan) $ Query.singleUpsertQuery mrPlan apiReq conf
+      resultSet <- runQuery roleIsoLvl (Plan.mrTxMode mrPlan) $ Query.singleUpsertQuery mrPlan apiReq conf
       return $ Response.singleUpsertResponse apiReq resultSet
 
     (ActionMutate MutationDelete, TargetIdent identifier) -> do
       mrPlan <- liftEither $ Plan.mutateReadPlan MutationDelete apiReq identifier conf sCache
-      resultSet <- runQuery (Plan.mrTxMode mrPlan) $ Query.deleteQuery mrPlan apiReq conf
+      resultSet <- runQuery roleIsoLvl (Plan.mrTxMode mrPlan) $ Query.deleteQuery mrPlan apiReq conf
       return $ Response.deleteResponse apiReq resultSet
 
     (ActionInvoke invMethod, TargetProc identifier _) -> do
       cPlan <- liftEither $ Plan.callReadPlan identifier conf sCache apiReq invMethod
-      resultSet <- runQuery (Plan.crTxMode cPlan) $ Query.invokeQuery (Plan.crProc cPlan) cPlan apiReq conf pgVer
+      resultSet <- runQuery (roleIsoLvl <|> pdIsoLvl (Plan.crProc cPlan))(Plan.crTxMode cPlan) $ Query.invokeQuery (Plan.crProc cPlan) cPlan apiReq conf pgVer
       return $ Response.invokeResponse invMethod (Plan.crProc cPlan) apiReq resultSet
 
     (ActionInspect headersOnly, TargetDefaultSpec tSchema) -> do
-      oaiResult <- runQuery Plan.inspectPlanTxMode $ Query.openApiQuery sCache pgVer conf tSchema
+      oaiResult <- runQuery roleIsoLvl Plan.inspectPlanTxMode $ Query.openApiQuery sCache pgVer conf tSchema
       return $ Response.openApiResponse headersOnly oaiResult conf sCache iSchema iNegotiatedByProfile
 
     (ActionInfo, TargetIdent identifier) ->
@@ -220,8 +228,10 @@ handleRequest AuthResult{..} conf appState authenticated prepared pgVer apiReq@A
       -- TODO Refactor the Action/Target types to remove this line
       throwError $ Error.ApiRequestError ApiRequestTypes.NotFound
   where
-    runQuery mode query =
-      runDbHandler appState mode authenticated prepared $ do
-        Query.setPgLocals conf authClaims authRole apiReq pgVer
+    roleSettings = fromMaybe mempty (HM.lookup authRole $ configRoleSettings conf)
+    roleIsoLvl = decodeUtf8 <$> HM.lookup "default_transaction_isolation" roleSettings
+    runQuery isoLvl mode query =
+      runDbHandler appState isoLvl mode authenticated prepared $ do
+        Query.setPgLocals conf authClaims authRole (HM.toList roleSettings) apiReq pgVer
         Query.runPreReq conf
         query

src/PostgREST/Config/Database.hs

@@ -25,7 +25,7 @@ import Text.InterpolatedString.Perl6 (q)
 
 import Protolude
 
-type RoleSettings = (HM.HashMap ByteString [(ByteString, ByteString)])
+type RoleSettings = (HM.HashMap ByteString (HM.HashMap ByteString ByteString))
 
 queryPgVersion :: Bool -> Session PgVersion
 queryPgVersion prepared = statement mempty $ pgVersionStatement prepared
@@ -75,7 +75,7 @@ queryRoleSettings prepared =
   transaction SQL.ReadCommitted SQL.Read $ SQL.statement mempty $ roleSettingsStatement prepared
 
 roleSettingsStatement :: Bool -> SQL.Statement () RoleSettings
-roleSettingsStatement = SQL.Statement sql HE.noParams decodeSettings
+roleSettingsStatement = SQL.Statement sql HE.noParams decodeRoleSettings
   where
     sql = [q|
       with
@@ -89,14 +89,14 @@ roleSettingsStatement = SQL.Statement sql HE.noParams decodeSettings
         SELECT
           rolname,
           substr(setting, 1, strpos(setting, '=') - 1) as key,
-          substr(setting, strpos(setting, '=') + 1) as value
+          lower(substr(setting, strpos(setting, '=') + 1)) as value
         FROM role_setting
       )
       select rolname, array_agg(row(key, value))
       from kv_settings
       group by rolname;
     |]
-    decodeSettings = HM.fromList . map (bimap encodeUtf8 ((encodeUtf8 *** encodeUtf8) <$>)) <$> HD.rowList aRow
+    decodeRoleSettings = HM.fromList . map (bimap encodeUtf8 (HM.fromList . ((encodeUtf8 *** encodeUtf8) <$>))) <$> HD.rowList aRow
     aRow :: HD.Row (Text, [(Text, Text)])
     aRow = (,) <$> column HD.text <*> compositeArrayColumn ((,) <$> compositeField HD.text <*> compositeField HD.text)
 

src/PostgREST/Query.hs

@@ -235,9 +235,9 @@ optionalRollback AppConfig{..} ApiRequest{iPreferences=Preferences{..}} = do
       configDbTxAllowOverride && preferTransaction == Just Rollback
 
 -- | Runs local (transaction scoped) GUCs for every request.
-setPgLocals :: AppConfig  -> KM.KeyMap JSON.Value -> BS.ByteString ->
+setPgLocals :: AppConfig  -> KM.KeyMap JSON.Value -> BS.ByteString -> [(ByteString, ByteString)] ->
                ApiRequest -> PgVersion -> DbHandler ()
-setPgLocals AppConfig{..} claims role req actualPgVersion = lift $
+setPgLocals AppConfig{..} claims role roleSettings req actualPgVersion = lift $
   SQL.statement mempty $ SQL.dynamicallyParameterized
     ("select " <> intercalateSnippet ", " (searchPathSql : roleSql ++ roleSettingsSql ++ claimsSql ++ [methodSql, pathSql] ++ headersSql ++ cookiesSql ++ appSettingsSql))
     HD.noResult configDbPreparedStatements
@@ -254,9 +254,7 @@ setPgLocals AppConfig{..} claims role req actualPgVersion = lift $
                   then setConfigLocal "request.jwt.claim." <$> [(toUtf8 $ K.toText c, toUtf8 $ unquoted v) | (c,v) <- KM.toList claims]
                   else [setConfigLocal mempty ("request.jwt.claims", LBS.toStrict $ JSON.encode claims)]
     roleSql = [setConfigLocal mempty ("role", role)]
-    roleSettingsSql = if null configRoleSettings
-      then mempty
-      else setConfigLocal mempty <$> fromMaybe mempty (HM.lookup role configRoleSettings)
+    roleSettingsSql = setConfigLocal mempty <$> roleSettings
     appSettingsSql = setConfigLocal mempty <$> (join bimap toUtf8 <$> configAppSettings)
     searchPathSql =
       let schemas = pgFmtIdentList (iSchema req : configDbExtraSearchPath) in

src/PostgREST/SchemaCache.hs

@@ -250,6 +250,7 @@ decodeFuncs =
                   <*> column HD.bool)
               <*> (parseVolatility <$> column HD.char)
               <*> column HD.bool
+              <*> nullableColumn HD.text
 
     addKey :: Routine -> (QualifiedIdentifier, Routine)
     addKey pd = (QualifiedIdentifier (pdSchema pd) (pdName pd), pd)
@@ -339,7 +340,8 @@ funcsSqlQuery pgVer = [q|
     ) AS rettype_is_composite,
     bt.oid <> bt.base as rettype_is_composite_alias,
     p.provolatile,
-    p.provariadic > 0 as hasvariadic
+    p.provariadic > 0 as hasvariadic,
+    lower((regexp_split_to_array((regexp_split_to_array(config, '='))[2], ','))[1]) AS transaction_isolation_level
   FROM pg_proc p
   LEFT JOIN arguments a ON a.oid = p.oid
   JOIN pg_namespace pn ON pn.oid = p.pronamespace
@@ -348,6 +350,7 @@ funcsSqlQuery pgVer = [q|
   JOIN pg_namespace tn ON tn.oid = t.typnamespace
   LEFT JOIN pg_class comp ON comp.oid = t.typrelid
   LEFT JOIN pg_description as d ON d.objoid = p.oid
+  LEFT JOIN LATERAL unnest(proconfig) config ON config like 'default_transaction_isolation%'
   WHERE t.oid <> 'trigger'::regtype AND COALESCE(a.callable, true)
 |] <> (if pgVer >= pgVersion110 then "AND prokind = 'f'" else "AND NOT (proisagg OR proiswindow)")
 

src/PostgREST/SchemaCache/Routine.hs

@@ -48,6 +48,7 @@ data Routine = Function
   , pdReturnType  :: RetType
   , pdVolatility  :: FuncVolatility
   , pdHasVariadic :: Bool
+  , pdIsoLvl      :: Maybe Text
   }
   deriving (Eq, Generic, JSON.ToJSON)
 
@@ -61,10 +62,10 @@ data RoutineParam = RoutineParam
 
 -- Order by least number of params in the case of overloaded functions
 instance Ord Routine where
-  Function schema1 name1 des1 prms1 rt1 vol1 hasVar1 `compare` Function schema2 name2 des2 prms2 rt2 vol2 hasVar2
+  Function schema1 name1 des1 prms1 rt1 vol1 hasVar1 iso1 `compare` Function schema2 name2 des2 prms2 rt2 vol2 hasVar2 iso2
     | schema1 == schema2 && name1 == name2 && length prms1 < length prms2  = LT
     | schema2 == schema2 && name1 == name2 && length prms1 > length prms2  = GT
-    | otherwise = (schema1, name1, des1, prms1, rt1, vol1, hasVar1) `compare` (schema2, name2, des2, prms2, rt2, vol2, hasVar2)
+    | otherwise = (schema1, name1, des1, prms1, rt1, vol1, hasVar1, iso1) `compare` (schema2, name2, des2, prms2, rt2, vol2, hasVar2, iso2)
 
 -- | A map of all procs, all of which can be overloaded(one entry will have more than one Routine).
 -- | It uses a HashMap for a faster lookup.

test/io/fixtures.sql

@@ -8,7 +8,13 @@ ALTER ROLE :USER SET pgrst.db_anon_role = 'postgrest_test_anonymous';
 
 CREATE ROLE postgrest_test_author;
 
-GRANT postgrest_test_anonymous, postgrest_test_author TO :USER;
+CREATE ROLE postgrest_test_serializable;
+alter role postgrest_test_serializable set default_transaction_isolation = 'serializable';
+
+CREATE ROLE postgrest_test_repeatable_read;
+alter role postgrest_test_repeatable_read set default_transaction_isolation = 'REPEATABLE READ';
+
+GRANT postgrest_test_anonymous, postgrest_test_author, postgrest_test_serializable, postgrest_test_repeatable_read TO :USER;
 
 CREATE SCHEMA v1;
 GRANT USAGE ON SCHEMA v1 TO postgrest_test_anonymous;
@@ -108,3 +114,31 @@ begin
     alter role current_user set statement_timeout = %L;
   $$, timeout);
 end $_$ volatile language plpgsql ;
+
+create table items as select x as id from generate_series(1,5) x;
+
+create view items_w_isolation_level as
+select
+  id,
+  current_setting('transaction_isolation', true) as isolation_level
+from items;
+
+grant all on items_w_isolation_level to postgrest_test_anonymous, postgrest_test_repeatable_read, postgrest_test_serializable;
+
+create function default_isolation_level()
+returns text as $$
+  select current_setting('transaction_isolation', true);
+$$
+language sql;
+
+create function serializable_isolation_level()
+returns text as $$
+  select current_setting('transaction_isolation', true);
+$$
+language sql set default_transaction_isolation = 'serializable';
+
+create function repeatable_read_isolation_level()
+returns text as $$
+  select current_setting('transaction_isolation', true);
+$$
+language sql set default_transaction_isolation = 'REPEATABLE READ';

test/io/test_io.py

@@ -873,6 +873,79 @@ def test_role_settings(defaultenv):
         assert response.text == '"10s"'
 
 
+def test_isolation_level(defaultenv):
+    "isolation_level should be set per role and per function"
+
+    env = {
+        **defaultenv,
+        "PGRST_JWT_SECRET": SECRET,
+    }
+
+    with run(env=env) as postgrest:
+        # default isolation level for postgrest_test_anonymous
+        response = postgrest.session.get(
+            "/items_w_isolation_level?select=isolation_level&limit=1"
+        )
+        assert response.text == '[{"isolation_level":"read committed"}]'
+
+        # isolation level for postgrest_test_repeatable_read on GET
+        headers = jwtauthheader({"role": "postgrest_test_repeatable_read"}, SECRET)
+        response = postgrest.session.get(
+            "/items_w_isolation_level?select=isolation_level&limit=1", headers=headers
+        )
+        assert response.text == '[{"isolation_level":"repeatable read"}]'
+
+        # isolation level for postgrest_test_serializable on POST
+        headers = jwtauthheader({"role": "postgrest_test_serializable"}, SECRET)
+        headers["Prefer"] = "return=representation"
+        response = postgrest.session.post(
+            "/items_w_isolation_level?select=isolation_level",
+            json={"id": "666"},
+            headers=headers,
+        )
+        assert response.text == '[{"isolation_level":"serializable"}]'
+
+        # isolation level for postgrest_test_serializable on PATCH
+        headers = jwtauthheader({"role": "postgrest_test_serializable"}, SECRET)
+        headers["Prefer"] = "return=representation"
+        response = postgrest.session.patch(
+            "/items_w_isolation_level?select=isolation_level&id=eq.666",
+            json={"id": "666"},
+            headers=headers,
+        )
+        assert response.text == '[{"isolation_level":"serializable"}]'
+
+        # isolation level for postgrest_test_serializable on DELETE
+        headers = jwtauthheader({"role": "postgrest_test_serializable"}, SECRET)
+        headers["Prefer"] = "return=representation"
+        response = postgrest.session.delete(
+            "/items_w_isolation_level?select=isolation_level&id=eq.666", headers=headers
+        )
+        assert response.text == '[{"isolation_level":"serializable"}]'
+
+        # default isolation level for function
+        response = postgrest.session.get("/rpc/default_isolation_level")
+        assert response.text == '"read committed"'
+
+        # changes with role isolation level
+        headers = jwtauthheader({"role": "postgrest_test_repeatable_read"}, SECRET)
+        response = postgrest.session.get(
+            "/rpc/default_isolation_level", headers=headers
+        )
+        assert response.text == '"repeatable read"'
+
+        # isolation level can be set per function
+        response = postgrest.session.get("/rpc/serializable_isolation_level")
+        assert response.text == '"serializable"'
+        response = postgrest.session.get("/rpc/repeatable_read_isolation_level")
+        assert response.text == '"repeatable read"'
+
+        # isolation level for a function overrides the role isolation level
+        headers = jwtauthheader({"role": "postgrest_test_repeatable_read"}, SECRET)
+        response = postgrest.session.get("/rpc/serializable_isolation_level")
+        assert response.text == '"serializable"'
+
+
 # TODO: This test fails now because of https://github.com/PostgREST/postgrest/pull/2122
 # The stack size of 1K(-with-rtsopts=-K1K) is not enough and this fails with "stack overflow"
 # A stack size of 200K seems to be enough for succeess