fix: retrying transaction on 40001 errors

Author: laurenceisla

nix/overlays/haskell-packages.nix

@@ -62,12 +62,13 @@ let
           }) "--subpath=."
           { }
       );
-      hasql-transaction = lib.dontCheck (prev.callHackageDirect {
-        pkg = "hasql-transaction";
-        ver = "1.2.1";
-        sha256 = "sha256-7Q7gt5ts4OoGU58dp6PJFZmVjfwjozANHNg2u1PJf6Q=";
-      }
-      { });
+      hasql-transaction = lib.dontCheck (prev.callHackageDirect
+        {
+          pkg = "hasql-transaction";
+          ver = "1.2.1";
+          sha256 = "sha256-7Q7gt5ts4OoGU58dp6PJFZmVjfwjozANHNg2u1PJf6Q=";
+        }
+        { });
 
       # Needed for hasql 1.9
       text-builder = lib.dontCheck prev.text-builder_1_0_0_3;

nix/tools/withTools.nix

@@ -85,7 +85,7 @@ let
           >&2 echo "${commandName}: You can tail the logs with: tail -f $tmpdir/db.log"
 
           if test "$_arg_replica" = "on"; then
-            replica_slot="replica_$RANDOM"
+            replica_slot="rr_$RANDOM"
             replica_dir="$tmpdir/$replica_slot"
             replica_host="$tmpdir/socket_$replica_slot"
 
@@ -100,7 +100,7 @@ let
 
             log "Starting replica on $replica_host"
 
-            pg_ctl -D "$replica_dir" -l "$replica_dblog" -w start -o "-F -c listen_addresses=\"\" -c hba_file=$HBA_FILE -k $replica_host -c log_statement=\"all\" " \
+            pg_ctl -D "$replica_dir" -l "$replica_dblog" -w start -o "-F -c listen_addresses=\"\" -c hba_file=$HBA_FILE -k $replica_host -c log_statement=\"all\" -c max_standby_streaming_delay=\"3s\"" \
               >> "$setuplog"
 
             >&2 echo "${commandName}: Replica enabled. You can connect to it with: psql 'postgres:///$PGDATABASE?host=$replica_host' -U postgres"

src/PostgREST/AppState.hs

@@ -408,7 +408,7 @@ retryingSchemaCacheLoad appState@AppState{stateObserver=observer, stateMainThrea
     qSchemaCache = do
       conf@AppConfig{..} <- getConfig appState
       (resultTime, result) <-
-        timeItT $ usePool appState (SQL.transaction SQL.ReadCommitted SQL.Read $ querySchemaCache conf)
+        timeItT $ usePool appState (SQL.transactionNoRetry SQL.ReadCommitted SQL.Read $ querySchemaCache conf)
       case result of
         Left e -> do
           putSCacheStatus appState SCPending

src/PostgREST/CLI.hs

@@ -53,7 +53,7 @@ main CLI{cliCommand, cliPath} = do
 dumpSchema :: AppState -> IO LBS.ByteString
 dumpSchema appState = do
   conf@AppConfig{..} <- AppState.getConfig appState
-  result <- AppState.usePool appState (SQL.transaction SQL.ReadCommitted SQL.Read $ querySchemaCache conf)
+  result <- AppState.usePool appState (SQL.transactionNoRetry SQL.ReadCommitted SQL.Read $ querySchemaCache conf)
   case result of
     Left e -> do
       let observer = AppState.getObserver appState

src/PostgREST/Config/Database.hs

@@ -92,7 +92,7 @@ pgVersionStatement = SQL.Statement sql HE.noParams versionRow
 -- A setting on the database only will have no effect: ALTER DATABASE postgres SET <prefix>jwt_aud = 'xx'
 queryDbSettings :: Maybe Text -> Bool -> Session [(Text, Text)]
 queryDbSettings preConfFunc prepared =
-  SQL.transaction SQL.ReadCommitted SQL.Read $ SQL.statement dbSettingsNames $ SQL.Statement sql (arrayParam HE.text) decodeSettings prepared
+  SQL.transactionNoRetry SQL.ReadCommitted SQL.Read $ SQL.statement dbSettingsNames $ SQL.Statement sql (arrayParam HE.text) decodeSettings prepared
   where
     sql = encodeUtf8 [trimming|
       WITH
@@ -132,7 +132,7 @@ queryDbSettings preConfFunc prepared =
 
 queryRoleSettings :: PgVersion -> Bool -> Session (RoleSettings, RoleIsolationLvl)
 queryRoleSettings pgVer prepared =
-  SQL.transaction SQL.ReadCommitted SQL.Read $ SQL.statement mempty $ SQL.Statement sql HE.noParams (processRows <$> rows) prepared
+  SQL.transactionNoRetry SQL.ReadCommitted SQL.Read $ SQL.statement mempty $ SQL.Statement sql HE.noParams (processRows <$> rows) prepared
   where
     sql = encodeUtf8 [trimming|
       with

src/PostgREST/Query.hs

@@ -76,7 +76,7 @@ data QueryResult
 query :: AppConfig -> AuthResult -> ApiRequest -> ActionPlan -> SchemaCache -> Query
 query _ _ _ (NoDb x) _ = NoDbQuery $ NoDbResult x
 query config AuthResult{..} apiReq (Db plan) sCache =
-  DbQuery isoLvl txMode dbHandler SQL.transaction mainSQLQuery
+  DbQuery isoLvl txMode dbHandler SQL.transactionNoRetry mainSQLQuery
   where
     isoLvl = planIsoLvl config authRole plan
     txMode = planTxMode plan

test/io/replica.sql

@@ -10,6 +10,10 @@ $$ language sql;
 
 create table replica.items as select x as id from generate_series(1, 10) x;
 
+create table replica.conflict as select x as id from generate_series(1, 1000000) x;
+
+create view replica.conflict_view as select * from replica.conflict where (pg_sleep(0.01) is not null);
+
 DROP ROLE IF EXISTS postgrest_test_anonymous;
 CREATE ROLE postgrest_test_anonymous;
 

test/io/test_replica.py

@@ -1,6 +1,8 @@
 "IO tests for PostgREST started on replicas"
 
+import os
 import pytest
+import time
 
 from config import *
 from util import *
@@ -26,3 +28,33 @@ def test_sanity_replica(replicaenv):
 
         response = postgrest.session.get("/items?select=count")
         assert response.text == '[{"count":10}]'
+
+
+def test_conflict_replica(replicaenv):
+    "Test that PostgREST does not retry the transaction on conflict with recovery (PG error code 40001)"
+
+    with run(env=replicaenv["replica"]) as postgrest:
+
+        def conflict():
+            response = postgrest.session.get("/conflict_view")
+            # Checks that the transaction stops and returns the 40001 error instead of retrying
+            assert response.json()["code"] == "40001"
+            assert response.status_code == 500
+
+        t = Thread(target=conflict)
+        t.start()
+
+        # make sure the request has started
+        time.sleep(0.1)
+
+        prienv = replicaenv["primary"]
+        connopts = f'-d {prienv["PGDATABASE"]} -U postgres -h {prienv["PGHOST"]}'
+
+        # Delete the table data while the request with the lock is running to trigger the recovery conflict
+        os.system(
+            f'psql {connopts} --set ON_ERROR_STOP=1 -a -c "DELETE FROM replica.conflict;"'
+        )
+        # Vacuum the table to accelerate the process
+        os.system(f"vacuumdb {connopts} -t replica.conflict")
+
+        t.join()