Fix race condition when enabling logical decoding concurrently.

With wal_level = 'replica', logical decoding is enabled on demand
when the first logical replication slot is created:

When enabling logical decoding, EnableLogicalDecoding() flips the
shared logical_decoding_enabled flag and writes an
XLOG_LOGICAL_DECODING_STATUS_CHANGE record so that standbys follow the
status change. The initial "already enabled?" check and the WAL record
write happen under two separate acquisitions of
LogicalDecodingControlLock, since the lock must be released while
waiting for the ProcSignalBarrier: processes absorbing the barrier
acquire the same lock in shared mode.

Consequently, if two backends concurrently created the first logical
slots, both could pass the initial check and both write a
status-change record. The redundant record lands after the decoding
start point already reserved by the other backend's slot, so decoding
that slot processes the record and fails with "unexpected logical
decoding status change", as xlog_decode() assumes that no such record
can appear within the WAL range any slot decodes.

Fix by re-checking the status after re-acquiring the lock, so that
only the backend that actually performs the disabled->enabled
transition writes the WAL record.

Reported-by: Srinath Reddy Sadipiralla <srinath2133@gmail.com>
Author: Srinath Reddy Sadipiralla <srinath2133@gmail.com>
Reviewed-by: Masahiko Sawada <sawada.mshk@gmail.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>
Discussion: https://postgr.es/m/CAFC+b6oYzmAgp7F0ivrhfZT46-CjvCTrU9pWuMNcem-52YjOTw@mail.gmail.com
Backpatch-through: 19

diff --git a/src/backend/replication/logical/logicalctl.c b/src/backend/replication/logical/logicalctl.c
index 624965ef95dce0af037dd2021e23847a29098ea0..4a690a631dae3c3ba580dc414c59ed2c9839967e 100644 (file)
--- a/src/backend/replication/logical/logicalctl.c
+++ b/src/backend/replication/logical/logicalctl.c
@@ -384,6 +384,17 @@ EnableLogicalDecoding(void)
 
        LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
 
+       /*
+        * Re-check whether logical decoding got enabled while we waited for the
+        * barrier above.
+        */
+       if (LogicalDecodingCtl->logical_decoding_enabled)
+       {
+               LogicalDecodingCtl->pending_disable = false;
+               LWLockRelease(LogicalDecodingControlLock);
+               return;
+       }
+
        START_CRIT_SECTION();
 
        /*

diff --git a/src/test/recovery/t/051_effective_wal_level.pl b/src/test/recovery/t/051_effective_wal_level.pl
index d4bc7f0aa407c139773d48026ea69b008cd5c3b3..2cf2ea6546ddcc2593b8c1740211da798279ec2e 100644 (file)
--- a/src/test/recovery/t/051_effective_wal_level.pl
+++ b/src/test/recovery/t/051_effective_wal_level.pl
@@ -326,11 +326,12 @@ $standby3->safe_psql('postgres',
 
 $standby3->stop;
 
-# Test the race condition at end of the recovery between the startup and logical
-# decoding status change. This test requires injection points enabled.
 if (   $ENV{enable_injection_points} eq 'yes'
        && $primary->check_extension('injection_points'))
 {
+       # Test the race condition at end of the recovery between the startup and logical
+       # decoding status change. This test requires injection points enabled.
+
        # Initialize standby4 and start it.
        my $standby4 = PostgreSQL::Test::Cluster->new('standby4');
        $standby4->init_from_backup($primary, 'my_backup', has_streaming => 1);
@@ -381,9 +382,63 @@ if (   $ENV{enable_injection_points} eq 'yes'
        test_wal_level($primary, "replica|replica",
                "effective_wal_level got decreased to 'replica' on primary");
 
+       # Test that concurrent activations don't write redundant status-change records.
+
+       # Start a psql session and stop it in the middle of the activation process.
+       my $psql_create_slot = $primary->background_psql('postgres');
+       $psql_create_slot->query_until(
+               qr/create_slot_1/,
+               q(\echo create_slot_1
+select injection_points_set_local();
+select injection_points_attach('logical-decoding-activation', 'wait');
+select pg_create_logical_replication_slot('slot_1', 'test_decoding');
+));
+       $primary->wait_for_event('client backend', 'logical-decoding-activation');
+       note("injection_point 'logical-decoding-activation' is reached");
+
+       # A second backend concurrently enables logical decoding and finishes creating
+       # its slot, writing the status-change record. The slot reserves its decoding
+       # start point after its own status-change record.
+       $primary->safe_psql('postgres',
+               qq[select pg_create_logical_replication_slot('slot_2', 'test_decoding')]
+       );
+       test_wal_level($primary, "replica|logical",
+               "logical decoding enabled by the first of two concurrent activations"
+       );
+
+       # Resume the first backend to complete the slot creation. It must not write
+       # a second redundant status-change record as logical decoding is already
+       # enabled.
+       $primary->safe_psql('postgres',
+               qq[select injection_points_wakeup('logical-decoding-activation')]);
+
+       # Let the released backend finish creating its slot.
+       $psql_create_slot->quit;
+
+       # Decode from slot_2, whose start point precedes where a redundant
+       # status-change record would have been written; this fails in xlog_decode()
+       # if one exists.
+       is( $primary->safe_psql(
+                       'postgres',
+                       qq[SELECT count(*) FROM pg_logical_slot_get_changes('slot_2', NULL, NULL, 'skip-empty-xacts', '1')]
+               ),
+               0,
+               'decoding a concurrently-created slot succeeds');
+
+       # Restore the disabled state for the tests that follow.
+       $primary->safe_psql(
+               'postgres',
+               qq[
+select pg_drop_replication_slot('slot_1');
+select pg_drop_replication_slot('slot_2');
+]);
+       wait_for_logical_decoding_disabled($primary);
+
+       # Test a race when logical decoding activation is concurrently interrupted.
+
        # Start a psql session to test the case where the activation process is
        # interrupted.
-       my $psql_create_slot = $primary->background_psql('postgres');
+       $psql_create_slot = $primary->background_psql('postgres');
 
        # Start the logical decoding activation process upon creating the logical
        # slot, but it will wait due to the injection point.