From bef6fbe5c2a59e5e014c2f6f0f5f651900d1457e Mon Sep 17 00:00:00 2001 From: Nikhil Chawla Date: Mon, 23 Mar 2026 17:15:00 +0530 Subject: [PATCH] Add prepared_orphaned_transaction_timeout GUC This adds a new GUC parameter that automatically rolls back prepared transactions that have remained unresolved beyond the configured timeout. The cleanup is performed by the checkpointer process. This is useful for preventing orphaned two-phase transactions from holding locks and preventing VACUUM from reclaiming dead tuples indefinitely. The timeout is disabled by default (0) and can be changed with a configuration reload (SIGHUP). --- doc/src/sgml/config.sgml | 35 +++ src/backend/access/transam/twophase.c | 252 ++++++++++++++++++ src/backend/postmaster/checkpointer.c | 7 + src/backend/utils/misc/guc_parameters.dat | 10 + src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/access/twophase.h | 5 +- src/test/modules/test_misc/meson.build | 1 + .../t/011_prepared_orphaned_timeout.pl | 203 ++++++++++++++ 8 files changed, 513 insertions(+), 1 deletion(-) create mode 100644 src/test/modules/test_misc/t/011_prepared_orphaned_timeout.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 8cdd826fbd3..2938e91ce1c 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -10467,6 +10467,41 @@ COPY postgres_log FROM '/full/path/to/logfile.csv' WITH csv; + + prepared_orphaned_transaction_timeout (integer) + + prepared_orphaned_transaction_timeout configuration parameter + + + + + Automatically roll back any prepared transaction that has remained + in the prepared state for longer than the specified amount of time. + If this value is specified without units, it is taken as milliseconds. + A value of zero (the default) disables the timeout. + + + + Prepared transactions (from PREPARE TRANSACTION) + persist even after the originating session disconnects, and hold + locks and prevent vacuum cleanup of recently-dead tuples. If the + application that issued the PREPARE TRANSACTION + fails to follow up with COMMIT PREPARED or + ROLLBACK PREPARED, these orphaned prepared + transactions can accumulate and impact database performance. + This setting provides an automatic safety net to clean up such + orphaned transactions. + + + + The cleanup is performed by the checkpointer process, which + periodically scans for prepared transactions exceeding the timeout + and rolls them back. A log message is emitted for each transaction + that is automatically rolled back. + + + + bytea_output (enum) diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index d468c9774b3..9113dbddf06 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -116,6 +116,9 @@ /* GUC variable, can't be changed after startup */ int max_prepared_xacts = 0; +/* GUC variable for orphaned prepared transaction timeout (in ms, 0 = disabled) */ +int prepared_orphaned_transaction_timeout = 0; + /* * This struct describes one global transaction that is in prepared state * or attempting to become prepared. @@ -2873,3 +2876,252 @@ TwoPhaseGetOldestXidInCommit(void) return oldestRunningXid; } + +/* + * LockGXactForCleanup + * Locate the prepared transaction by GID and mark it busy for cleanup. + * + * This is similar to LockGXact but does not check database identity or + * ownership, since it is used by the background cleanup of orphaned prepared + * transactions which operates as a superuser-like facility. + * + * Returns the GlobalTransaction on success, or NULL if the GID is not found + * or is already locked by another backend. + */ +static GlobalTransaction +LockGXactForCleanup(const char *gid) +{ + int i; + + /* on first call, register the exit hook */ + if (!twophaseExitRegistered) + { + before_shmem_exit(AtProcExit_Twophase, 0); + twophaseExitRegistered = true; + } + + LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); + + for (i = 0; i < TwoPhaseState->numPrepXacts; i++) + { + GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; + + /* Ignore not-yet-valid GIDs */ + if (!gxact->valid) + continue; + if (strcmp(gxact->gid, gid) != 0) + continue; + + /* Already locked by another backend? Skip it. */ + if (gxact->locking_backend != INVALID_PROC_NUMBER) + { + LWLockRelease(TwoPhaseStateLock); + return NULL; + } + + /* OK for us to lock it */ + gxact->locking_backend = MyProcNumber; + MyLockedGxact = gxact; + + LWLockRelease(TwoPhaseStateLock); + + return gxact; + } + + LWLockRelease(TwoPhaseStateLock); + + return NULL; +} + +/* + * CleanupOrphanedPreparedTransactions + * Roll back prepared transactions that have exceeded the + * prepared_orphaned_transaction_timeout. + * + * This is called from the checkpointer's main loop to periodically scan + * for prepared transactions that have been in the prepared state for + * longer than the configured timeout. Such transactions are considered + * orphaned and are automatically rolled back. + * + * The function first collects GIDs of candidate transactions while holding + * TwoPhaseStateLock in shared mode, then processes each one individually. + */ +void +CleanupOrphanedPreparedTransactions(void) +{ + int timeout_ms = prepared_orphaned_transaction_timeout; + TimestampTz now; + int num_orphaned = 0; + char **orphaned_gids = NULL; + int i; + + /* Quick exit if the feature is disabled or no prepared xacts configured */ + if (timeout_ms <= 0 || max_prepared_xacts == 0) + return; + + now = GetCurrentTimestamp(); + + /* + * Phase 1: Collect GIDs of prepared transactions that have exceeded the + * timeout. We only hold the lock in shared mode for the scan. + */ + LWLockAcquire(TwoPhaseStateLock, LW_SHARED); + + if (TwoPhaseState->numPrepXacts > 0) + { + orphaned_gids = (char **) palloc(TwoPhaseState->numPrepXacts * sizeof(char *)); + + for (i = 0; i < TwoPhaseState->numPrepXacts; i++) + { + GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; + + if (!gxact->valid) + continue; + + /* Skip transactions currently being worked on */ + if (gxact->locking_backend != INVALID_PROC_NUMBER) + continue; + + /* Check if this transaction has exceeded the timeout */ + if (TimestampDifferenceExceeds(gxact->prepared_at, now, timeout_ms)) + { + orphaned_gids[num_orphaned] = pstrdup(gxact->gid); + num_orphaned++; + } + } + } + + LWLockRelease(TwoPhaseStateLock); + + /* + * Phase 2: Roll back each orphaned prepared transaction. We do this + * outside the lock to avoid holding TwoPhaseStateLock for too long. + * + * We use LockGXactForCleanup which does not enforce database or owner + * checks, since the checkpointer process is not connected to any + * particular database. + */ + for (i = 0; i < num_orphaned; i++) + { + GlobalTransaction gxact; + PGPROC *proc; + FullTransactionId fxid; + TransactionId xid; + bool ondisk; + char *buf; + char *bufptr; + TwoPhaseFileHeader *hdr; + TransactionId latestXid; + TransactionId *children; + RelFileLocator *abortrels; + int ndelrels; + xl_xact_stats_item *abortstats; + + ereport(LOG, + (errmsg("rolling back orphaned prepared transaction \"%s\"", + orphaned_gids[i]), + errdetail("This prepared transaction has exceeded the prepared_orphaned_transaction_timeout of %d ms.", + timeout_ms))); + + /* Try to lock the gxact; skip if someone else got it first */ + gxact = LockGXactForCleanup(orphaned_gids[i]); + if (gxact == NULL) + { + pfree(orphaned_gids[i]); + continue; + } + + proc = GetPGProcByNumber(gxact->pgprocno); + fxid = gxact->fxid; + xid = XidFromFullTransactionId(fxid); + + /* + * Read and validate 2PC state data. + */ + if (gxact->ondisk) + buf = ReadTwoPhaseFile(fxid, false); + else + XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL); + + /* + * Disassemble the header area + */ + hdr = (TwoPhaseFileHeader *) buf; + Assert(TransactionIdEquals(hdr->xid, xid)); + bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader)); + bufptr += MAXALIGN(hdr->gidlen); + children = (TransactionId *) bufptr; + bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId)); + /* skip commitrels */ + bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileLocator)); + abortrels = (RelFileLocator *) bufptr; + bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileLocator)); + /* skip commitstats */ + bufptr += MAXALIGN(hdr->ncommitstats * sizeof(xl_xact_stats_item)); + abortstats = (xl_xact_stats_item *) bufptr; + bufptr += MAXALIGN(hdr->nabortstats * sizeof(xl_xact_stats_item)); + /* skip invalmsgs */ + bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage)); + + /* compute latestXid among all children */ + latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children); + + /* Prevent cancel/die interrupt while cleaning up */ + HOLD_INTERRUPTS(); + + /* Record the abort in WAL and mark transaction as aborted in pg_xact */ + RecordTransactionAbortPrepared(xid, + hdr->nsubxacts, children, + hdr->nabortrels, abortrels, + hdr->nabortstats, + abortstats, + orphaned_gids[i]); + + ProcArrayRemove(proc, latestXid); + + /* + * Mark the gxact invalid so no one else will try to commit/rollback. + */ + gxact->valid = false; + + /* Drop files that were supposed to be dropped on abort */ + ndelrels = hdr->nabortrels; + DropRelationFiles(abortrels, ndelrels, false); + + pgstat_execute_transactional_drops(hdr->nabortstats, abortstats, false); + + /* + * Acquire the two-phase lock for callbacks and cleanup. + */ + LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); + + /* Run post-abort callbacks */ + ProcessRecords(bufptr, fxid, twophase_postabort_callbacks); + + PredicateLockTwoPhaseFinish(fxid, false); + + ondisk = gxact->ondisk; + + /* Clear shared memory state */ + RemoveGXact(gxact); + + LWLockRelease(TwoPhaseStateLock); + + /* Count the prepared xact as aborted */ + AtEOXact_PgStat(false, false); + + /* Remove on-disk state file if any */ + if (ondisk) + RemoveTwoPhaseFile(fxid, true); + + MyLockedGxact = NULL; + + RESUME_INTERRUPTS(); + + pfree(buf); + pfree(orphaned_gids[i]); + } + + if (orphaned_gids) + pfree(orphaned_gids); +} diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c index 3c982c6ffac..9139d64b3c5 100644 --- a/src/backend/postmaster/checkpointer.c +++ b/src/backend/postmaster/checkpointer.c @@ -39,6 +39,7 @@ #include #include +#include "access/twophase.h" #include "access/xlog.h" #include "access/xlog_internal.h" #include "access/xlogrecovery.h" @@ -568,6 +569,12 @@ CheckpointerMain(const void *startup_data, size_t startup_data_len) /* Check for archive_timeout and switch xlog files if necessary. */ CheckArchiveTimeout(); + /* + * Clean up orphaned prepared transactions that have exceeded the + * prepared_orphaned_transaction_timeout. + */ + CleanupOrphanedPreparedTransactions(); + /* Report pending statistics to the cumulative stats system */ pgstat_report_checkpointer(); pgstat_report_wal(true); diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index 0c9854ad8fc..225130b2595 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -2306,6 +2306,16 @@ max => '60', }, +{ name => 'prepared_orphaned_transaction_timeout', type => 'int', context => 'PGC_SIGHUP', group => 'CLIENT_CONN_STATEMENT', + short_desc => 'Sets the maximum time that a prepared transaction can remain unresolved before it is automatically rolled back.', + long_desc => 'A value of 0 (the default) disables the timeout.', + flags => 'GUC_UNIT_MS', + variable => 'prepared_orphaned_transaction_timeout', + boot_val => '0', + min => '0', + max => 'INT_MAX', +}, + { name => 'primary_conninfo', type => 'string', context => 'PGC_SIGHUP', group => 'REPLICATION_STANDBY', short_desc => 'Sets the connection string to be used to connect to the sending server.', flags => 'GUC_SUPERUSER_ONLY', diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index e4abe6c0077..25469078ee0 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -144,6 +144,7 @@ # (change requires restart) # Caution: it is not advisable to set max_prepared_transactions nonzero unless # you actively intend to use prepared transactions. +#prepared_orphaned_transaction_timeout = 0 # in milliseconds, 0 is disabled #work_mem = 4MB # min 64kB #hash_mem_multiplier = 2.0 # 1-1000.0 multiplier on hash table work_mem #maintenance_work_mem = 64MB # min 64kB diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h index e312514ba87..b660aedcaff 100644 --- a/src/include/access/twophase.h +++ b/src/include/access/twophase.h @@ -25,8 +25,9 @@ */ typedef struct GlobalTransactionData *GlobalTransaction; -/* GUC variable */ +/* GUC variables */ extern PGDLLIMPORT int max_prepared_xacts; +extern PGDLLIMPORT int prepared_orphaned_transaction_timeout; extern Size TwoPhaseShmemSize(void); extern void TwoPhaseShmemInit(void); @@ -70,4 +71,6 @@ extern bool LookupGXactBySubid(Oid subid); extern TransactionId TwoPhaseGetOldestXidInCommit(void); +extern void CleanupOrphanedPreparedTransactions(void); + #endif /* TWOPHASE_H */ diff --git a/src/test/modules/test_misc/meson.build b/src/test/modules/test_misc/meson.build index 6e8db1621a7..8f71efcfe69 100644 --- a/src/test/modules/test_misc/meson.build +++ b/src/test/modules/test_misc/meson.build @@ -19,6 +19,7 @@ tests += { 't/008_replslot_single_user.pl', 't/009_log_temp_files.pl', 't/010_index_concurrently_upsert.pl', + 't/011_prepared_orphaned_timeout.pl', ], # The injection points are cluster-wide, so disable installcheck 'runningcheck': false, diff --git a/src/test/modules/test_misc/t/011_prepared_orphaned_timeout.pl b/src/test/modules/test_misc/t/011_prepared_orphaned_timeout.pl new file mode 100644 index 00000000000..a47c0595f7e --- /dev/null +++ b/src/test/modules/test_misc/t/011_prepared_orphaned_timeout.pl @@ -0,0 +1,203 @@ + +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Test for prepared_orphaned_transaction_timeout GUC. +# Verifies that orphaned prepared transactions are automatically +# rolled back by the checkpointer when they exceed the configured timeout. + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# +# Test 1: Basic orphaned prepared transaction cleanup +# +# Set up a node with prepared transactions enabled and a short timeout. +my $node = PostgreSQL::Test::Cluster->new('primary'); +$node->init; +$node->append_conf( + 'postgresql.conf', qq( + max_prepared_transactions = 5 + prepared_orphaned_transaction_timeout = '2s' + log_min_messages = log +)); +$node->start; + +# Create a test table and prepare a transaction. +$node->safe_psql('postgres', 'CREATE TABLE t_orphan_test (id int, msg text)'); +$node->safe_psql( + 'postgres', " + BEGIN; + INSERT INTO t_orphan_test VALUES (1, 'orphaned'); + PREPARE TRANSACTION 'orphan_xact_1'; +"); + +# Verify the prepared transaction exists. +my $result = $node->safe_psql('postgres', + "SELECT count(*) FROM pg_prepared_xacts WHERE gid = 'orphan_xact_1'"); +is($result, '1', 'prepared transaction orphan_xact_1 exists'); + +# Record log position before the timeout fires. +my $log_offset = -s $node->logfile; + +# Wait for the timeout to elapse, then trigger the checkpointer loop +# by issuing a CHECKPOINT. The checkpointer will scan for orphaned +# prepared transactions as part of its main loop iteration. +sleep(3); +$node->safe_psql('postgres', 'CHECKPOINT'); + +# Give the checkpointer a moment to process the cleanup. +sleep(1); + +# Verify the prepared transaction was rolled back. +ok( $node->poll_query_until( + 'postgres', + "SELECT count(*) FROM pg_prepared_xacts WHERE gid = 'orphan_xact_1'", + '0'), + 'orphaned prepared transaction was rolled back'); + +# Verify that the inserted data was NOT committed (rolled back). +$result = $node->safe_psql('postgres', + "SELECT count(*) FROM t_orphan_test WHERE id = 1"); +is($result, '0', 'orphaned transaction data was not committed'); + +# Verify the rollback was logged. +$node->wait_for_log( + qr/rolling back orphaned prepared transaction "orphan_xact_1"/, + $log_offset); +ok(1, 'orphaned transaction rollback was logged'); + + +# +# Test 2: Prepared transaction committed before timeout is not affected +# +$node->safe_psql( + 'postgres', " + BEGIN; + INSERT INTO t_orphan_test VALUES (2, 'committed_in_time'); + PREPARE TRANSACTION 'timely_xact'; +"); + +# Commit the prepared transaction immediately (before timeout). +$node->safe_psql('postgres', "COMMIT PREPARED 'timely_xact'"); + +# Verify the data was committed. +$result = $node->safe_psql('postgres', + "SELECT count(*) FROM t_orphan_test WHERE id = 2"); +is($result, '1', + 'prepared transaction committed before timeout preserved data'); + + +# +# Test 3: Timeout of 0 disables the feature +# +$node->safe_psql('postgres', + "ALTER SYSTEM SET prepared_orphaned_transaction_timeout = '0'"); +$node->safe_psql('postgres', "SELECT pg_reload_conf()"); + +$node->safe_psql( + 'postgres', " + BEGIN; + INSERT INTO t_orphan_test VALUES (3, 'should_persist'); + PREPARE TRANSACTION 'persist_xact'; +"); + +# Wait and trigger checkpointer - the transaction should NOT be rolled back. +sleep(3); +$node->safe_psql('postgres', 'CHECKPOINT'); +sleep(1); + +$result = $node->safe_psql('postgres', + "SELECT count(*) FROM pg_prepared_xacts WHERE gid = 'persist_xact'"); +is($result, '1', + 'prepared transaction not rolled back when timeout is disabled'); + +# Clean up. +$node->safe_psql('postgres', "ROLLBACK PREPARED 'persist_xact'"); + + +# +# Test 4: Multiple orphaned transactions are all cleaned up +# +$node->safe_psql('postgres', + "ALTER SYSTEM SET prepared_orphaned_transaction_timeout = '2s'"); +$node->safe_psql('postgres', "SELECT pg_reload_conf()"); + +$node->safe_psql( + 'postgres', " + BEGIN; + INSERT INTO t_orphan_test VALUES (10, 'multi_1'); + PREPARE TRANSACTION 'multi_orphan_1'; +"); + +$node->safe_psql( + 'postgres', " + BEGIN; + INSERT INTO t_orphan_test VALUES (11, 'multi_2'); + PREPARE TRANSACTION 'multi_orphan_2'; +"); + +$node->safe_psql( + 'postgres', " + BEGIN; + INSERT INTO t_orphan_test VALUES (12, 'multi_3'); + PREPARE TRANSACTION 'multi_orphan_3'; +"); + +$result = $node->safe_psql('postgres', + "SELECT count(*) FROM pg_prepared_xacts"); +is($result, '3', 'three prepared transactions exist'); + +# Wait for timeout, trigger checkpointer. +sleep(3); +$node->safe_psql('postgres', 'CHECKPOINT'); + +# All three should be cleaned up. +ok( $node->poll_query_until( + 'postgres', "SELECT count(*) FROM pg_prepared_xacts", '0'), + 'all orphaned prepared transactions were rolled back'); + + +# +# Test 5: Timeout change via reload takes effect +# +# Set a very long timeout so nothing gets cleaned. +$node->safe_psql('postgres', + "ALTER SYSTEM SET prepared_orphaned_transaction_timeout = '1h'"); +$node->safe_psql('postgres', "SELECT pg_reload_conf()"); + +$node->safe_psql( + 'postgres', " + BEGIN; + INSERT INTO t_orphan_test VALUES (20, 'reload_test'); + PREPARE TRANSACTION 'reload_xact'; +"); + +sleep(3); +$node->safe_psql('postgres', 'CHECKPOINT'); +sleep(1); + +$result = $node->safe_psql('postgres', + "SELECT count(*) FROM pg_prepared_xacts WHERE gid = 'reload_xact'"); +is($result, '1', + 'prepared transaction persists with long timeout'); + +# Now lower the timeout and reload. +$node->safe_psql('postgres', + "ALTER SYSTEM SET prepared_orphaned_transaction_timeout = '1s'"); +$node->safe_psql('postgres', "SELECT pg_reload_conf()"); + +sleep(2); +$node->safe_psql('postgres', 'CHECKPOINT'); + +ok( $node->poll_query_until( + 'postgres', + "SELECT count(*) FROM pg_prepared_xacts WHERE gid = 'reload_xact'", + '0'), + 'prepared transaction cleaned up after lowering timeout via reload'); + +$node->stop; +done_testing(); -- 2.53.0.windows.2