Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1lrPld-00072R-7e for pgsql-hackers@arkaria.postgresql.org; Thu, 10 Jun 2021 18:51:01 +0000 Received: from localhost ([127.0.0.1] helo=malur.postgresql.org) by malur.postgresql.org with esmtp (Exim 4.92) (envelope-from ) id 1lrPlc-0003Os-1Y for pgsql-hackers@arkaria.postgresql.org; Thu, 10 Jun 2021 18:51:00 +0000 Received: from magus.postgresql.org ([2a02:c0:301:0:ffff::29]) by malur.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1lrPlb-0003Ok-QG for pgsql-hackers@lists.postgresql.org; Thu, 10 Jun 2021 18:50:59 +0000 Received: from sss.pgh.pa.us ([66.207.139.130]) by magus.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1lrPlX-0005Yp-BQ for pgsql-hackers@lists.postgresql.org; Thu, 10 Jun 2021 18:50:59 +0000 Received: from sss1.sss.pgh.pa.us (localhost [127.0.0.1]) by sss.pgh.pa.us (8.15.2/8.15.2) with ESMTP id 15AIoqHk1719084; Thu, 10 Jun 2021 14:50:52 -0400 From: Tom Lane To: Amit Kapila cc: Mark Dilger , PostgreSQL Hackers , Peter Eisentraut , Simon Riggs Subject: Re: logical replication of truncate command with trigger causes Assert In-reply-to: References: <1250962.1623192946@sss.pgh.pa.us> <1254157.1623196736@sss.pgh.pa.us> <1356183.1623250358@sss.pgh.pa.us> Comments: In-reply-to Amit Kapila message dated "Thu, 10 Jun 2021 09:40:04 +0530" MIME-Version: 1.0 Content-Type: multipart/mixed; boundary="----- =_aaaaaaaaaa0" Content-ID: <1719052.1623351016.0@sss.pgh.pa.us> Date: Thu, 10 Jun 2021 14:50:52 -0400 Message-ID: <1719083.1623351052@sss.pgh.pa.us> List-Id: List-Help: List-Subscribe: List-Post: List-Owner: List-Archive: Archived-At: Precedence: bulk ------- =_aaaaaaaaaa0 Content-Type: text/plain; charset="us-ascii" Content-ID: <1719052.1623351016.1@sss.pgh.pa.us> Content-Transfer-Encoding: quoted-printable Amit Kapila writes: > On Wed, Jun 9, 2021 at 8:44 PM Mark Dilger wrote: >> On Jun 9, 2021, at 7:52 AM, Tom Lane wrote: >>> Somewhat unrelated, but ... am I reading the code correctly that >>> apply_handle_stream_start and related routines are using Asserts >>> to check that the remote sent stream-control messages in the correct >>> order? > This also needs to be changed to test-and-elog. Here's a proposed patch for this. It looks like pretty much all of the bogosity is new with the streaming business. You might quibble with which things I thought deserved elog versus ereport. Another thing I'm wondering is how many of these messages really need to be translated. We could use errmsg_internal and avoid burdening the translators, perhaps. regards, tom lane ------- =_aaaaaaaaaa0 Content-Type: text/x-diff; name="clean-up-dubious-assert-usage.patch"; charset="us-ascii" Content-ID: <1719052.1623351016.2@sss.pgh.pa.us> Content-Description: clean-up-dubious-assert-usage.patch Content-Transfer-Encoding: quoted-printable diff --git a/src/backend/replication/logical/worker.c b/src/backend/replic= ation/logical/worker.c index 98c26002e8..26738d3589 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -177,7 +177,7 @@ bool in_remote_transaction =3D false; static XLogRecPtr remote_final_lsn =3D InvalidXLogRecPtr; = /* fields valid only when processing streamed transaction */ -bool in_streamed_transaction =3D false; +static bool in_streamed_transaction =3D false; = static TransactionId stream_xid =3D InvalidTransactionId; = @@ -345,7 +345,8 @@ handle_streamed_transaction(LogicalRepMsgType action, = StringInfo s) */ xid =3D pq_getmsgint(s, 4); = - Assert(TransactionIdIsValid(xid)); + if (!TransactionIdIsValid(xid)) + elog(ERROR, "invalid transaction ID in streamed replication transaction= "); = /* Add the new subxact to the array (unless already there). */ subxact_info_add(xid); @@ -785,7 +786,10 @@ apply_handle_commit(StringInfo s) = logicalrep_read_commit(s, &commit_data); = - Assert(commit_data.commit_lsn =3D=3D remote_final_lsn); + if (commit_data.commit_lsn !=3D remote_final_lsn) + elog(ERROR, "incorrect commit LSN %X/%X in commit message (expected %X/= %X)", + LSN_FORMAT_ARGS(commit_data.commit_lsn), + LSN_FORMAT_ARGS(remote_final_lsn)); = apply_handle_commit_internal(s, &commit_data); = @@ -824,7 +828,10 @@ apply_handle_stream_start(StringInfo s) bool first_segment; HASHCTL hash_ctl; = - Assert(!in_streamed_transaction); + if (in_streamed_transaction) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("duplicate STREAM START message"))); = /* * Start a transaction on stream start, this transaction will be committ= ed @@ -873,7 +880,10 @@ apply_handle_stream_start(StringInfo s) static void apply_handle_stream_stop(StringInfo s) { - Assert(in_streamed_transaction); + if (!in_streamed_transaction) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("STREAM STOP message without STREAM START"))); = /* * Close the file with serialized changes, and serialize information abo= ut @@ -905,7 +915,10 @@ apply_handle_stream_abort(StringInfo s) TransactionId xid; TransactionId subxid; = - Assert(!in_streamed_transaction); + if (in_streamed_transaction) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("STREAM ABORT message without STREAM STOP"))); = logicalrep_read_stream_abort(s, &xid, &subxid); = @@ -967,13 +980,12 @@ apply_handle_stream_abort(StringInfo s) return; } = - Assert((subidx >=3D 0) && (subidx < subxact_data.nsubxacts)); - ent =3D (StreamXidHash *) hash_search(xidhash, (void *) &xid, HASH_FIND, &found); - Assert(found); + if (!found) + elog(ERROR, "transaction %u not found in hash table", xid); = /* open the changes file */ changes_filename(path, MyLogicalRepWorker->subid, xid); @@ -1012,7 +1024,10 @@ apply_handle_stream_commit(StringInfo s) MemoryContext oldcxt; BufFile *fd; = - Assert(!in_streamed_transaction); + if (in_streamed_transaction) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("STREAM COMMIT message without STREAM STOP"))); = xid =3D logicalrep_read_stream_commit(s, &commit_data); = @@ -1031,11 +1046,14 @@ apply_handle_stream_commit(StringInfo s) /* open the spool file for the committed transaction */ changes_filename(path, MyLogicalRepWorker->subid, xid); elog(DEBUG1, "replaying changes from file \"%s\"", path); + ent =3D (StreamXidHash *) hash_search(xidhash, (void *) &xid, HASH_FIND, &found); - Assert(found); + if (!found) + elog(ERROR, "transaction %u not found in hash table", xid); + fd =3D BufFileOpenShared(ent->stream_fileset, path, O_RDONLY); = buffer =3D palloc(BLCKSZ); @@ -1080,7 +1098,9 @@ apply_handle_stream_commit(StringInfo s) errmsg("could not read from streaming transaction's changes file \"= %s\": %m", path))); = - Assert(len > 0); + if (len <=3D 0) + elog(ERROR, "incorrect length %d in streaming transaction's changes fi= le \"%s\"", + len, path); = /* make sure we have sufficiently large buffer */ buffer =3D repalloc(buffer, len); @@ -1108,7 +1128,7 @@ apply_handle_stream_commit(StringInfo s) nchanges++; = if (nchanges % 1000 =3D=3D 0) - elog(DEBUG1, "replayed %d changes from file '%s'", + elog(DEBUG1, "replayed %d changes from file \"%s\"", nchanges, path); } = ------- =_aaaaaaaaaa0--