Tighten the concurrent abort check during decoding.

During decoding of an in-progress or prepared transaction, we detect
concurrent abort with an error code ERRCODE_TRANSACTION_ROLLBACK. That is
not sufficient because a callback can decide to throw that error code
at other times as well.

Reported-by: Tom Lane
Author: Amit Kapila
Reviewed-by: Dilip Kumar
Discussion: https://postgr.es/m/CAA4eK1KCjPRS4aZHB48QMM4J8XOC1+TD8jo-4Yu84E+MjwqVhA@mail.gmail.com

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index ee680e5e1b4a3cbf41d4108a62ac12522999393c..c79425fbb73887606e0e7afa4b8d46f582873f2e 100644 (file)
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -2492,17 +2492,18 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                 * abort of the (sub)transaction we are streaming or preparing. We
                 * need to do the cleanup and return gracefully on this error, see
                 * SetupCheckXidLive.
+                *
+                * This error code can be thrown by one of the callbacks we call during
+                * decoding so we need to ensure that we return gracefully only when we are
+                * sending the data in streaming mode and the streaming is not finished yet
+                * or when we are sending the data out on a PREPARE during a two-phase
+                * commit.
                 */
-               if (errdata->sqlerrcode == ERRCODE_TRANSACTION_ROLLBACK)
+               if (errdata->sqlerrcode == ERRCODE_TRANSACTION_ROLLBACK &&
+                       (stream_started || rbtxn_prepared(txn)))
                {
-                       /*
-                        * This error can occur either when we are sending the data in
-                        * streaming mode and the streaming is not finished yet or when we
-                        * are sending the data out on a PREPARE during a two-phase
-                        * commit.
-                        */
-                       Assert(streaming || rbtxn_prepared(txn));
-                       Assert(stream_started || rbtxn_prepared(txn));
+                       /* curtxn must be set for streaming or prepared transactions */
+                       Assert(curtxn);
 
                        /* Cleanup the temporary error state. */
                        FlushErrorState();