fix: no longer retries the transaction on 40001 errors

Author: laurenceisla

CHANGELOG.md

@@ -19,6 +19,7 @@ All notable changes to this project will be documented in this file. From versio
 
 - Shutdown should wait for in flight requests by @mkleczek in #4702
 - Fix login with uppercase and mixed case role names by @taimoorzaeem in #4678
+- Remove automatic transaction retries on `40001 (serialization_failure)` errors to prevent replication lag by @laurence in #3673
 
 ### Changed
 

nix/tools/withTools.nix

@@ -105,7 +105,8 @@ 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\" " \
+            # We set a low max_standby_streaming_delay to make the replication conflict fail faster in tests (otherwise it waits for the default 30s)
+            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

@@ -343,7 +343,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
           markSchemaCachePending appState

src/PostgREST/CLI.hs

@@ -62,7 +62,7 @@ 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)
+    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

@@ -94,7 +94,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 -> Session [(Text, Text)]
 queryDbSettings preConfFunc =
-  SQL.transaction SQL.ReadCommitted SQL.Read $ SQL.statement dbSettingsNames $ SQL.Statement sql (arrayParam HE.text) decodeSettings True
+  SQL.transactionNoRetry SQL.ReadCommitted SQL.Read $ SQL.statement dbSettingsNames $ SQL.Statement sql (arrayParam HE.text) decodeSettings True
   where
     sql = encodeUtf8 [trimming|
       WITH
@@ -134,7 +134,7 @@ queryDbSettings preConfFunc =
 
 queryRoleSettings :: PgVersion -> Session (RoleSettings, RoleIsolationLvl)
 queryRoleSettings pgVer =
-  SQL.transaction SQL.ReadCommitted SQL.Read $ SQL.statement mempty $ SQL.Statement sql HE.noParams (processRows <$> rows) True
+  SQL.transactionNoRetry SQL.ReadCommitted SQL.Read $ SQL.statement mempty $ SQL.Statement sql HE.noParams (processRows <$> rows) True
   where
     sql = encodeUtf8 [trimming|
       with

src/PostgREST/MainTx.hs

@@ -96,7 +96,7 @@ data ResultSet
 mainTx :: MainQuery -> AppConfig -> AuthResult -> ApiRequest -> ActionPlan -> SchemaCache -> MainTx
 mainTx _ _ _ _ (NoDb x) _ = NoDbTx $ NoDbResult x
 mainTx genQ@MainQuery{..} conf@AppConfig{..} AuthResult{..} apiReq (Db plan) sCache =
-  DbTx isoLvl txMode dbHandler SQL.transaction
+  DbTx isoLvl txMode dbHandler SQL.transactionNoRetry
   where
     isoLvl = planIsoLvl conf authRole plan
     txMode = planTxMode plan

test/io/fixtures/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,10 @@
 "IO tests for PostgREST started on replicas"
 
+import os
+import time
+
 from postgrest import run
+from util import Thread
 
 
 def test_sanity_replica(replicaenv):
@@ -22,3 +26,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()