Subject: Re: Time delayed LR (WAS Re: logical replication restrictions)

From: Masahiko Sawada <[email-masked]>
Date: Mon, 27 Feb 2023 14:40:36 +0900

On Thu, Feb 23, 2023 at 9:10 PM Hayato Kuroda (Fujitsu)
<[email-masked]> wrote:
>
> Dear Shi,
>
> Thank you for reviewing! PSA new version.
>
> > +             elog(DEBUG2, "time-delayed replication for txid %u, delay_time
> > = %d ms, remaining wait time: %ld ms",
> > +                      ctx->write_xid, (int) ctx->min_send_delay,
> > +                      remaining_wait_time_ms);
> >
> > I tried and saw that the xid here looks wrong, what it got is the xid of the
> > previous transaction. It seems `ctx->write_xid` has not been updated and we
> > can't use it.
> >
>
> Good catch. There are several approaches to fix that, I choose the simplest way.
> TransactionId was added as an argument of functions.
>

Thank you for updating the patch. Here are some comments on v7 patch:

+ *
+ * LOGICALREP_PROTO_MIN_SEND_DELAY_VERSION_NUM is the minimum protocol version
+ * with support for delaying to send transactions. Introduced in PG16.
  */
 #define LOGICALREP_PROTO_MIN_VERSION_NUM 1
 #define LOGICALREP_PROTO_VERSION_NUM 1
 #define LOGICALREP_PROTO_STREAM_VERSION_NUM 2
 #define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM 3
 #define LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM 4
-#define LOGICALREP_PROTO_MAX_VERSION_NUM
LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM
+#define LOGICALREP_PROTO_MIN_SEND_DELAY_VERSION_NUM 4
+#define LOGICALREP_PROTO_MAX_VERSION_NUM
LOGICALREP_PROTO_MIN_SEND_DELAY_VERSION_NUM

What is the usecase of the old macro,
LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM, after adding
LOGICALREP_PROTO_MIN_SEND_DELAY_VERSION_NUM ? I think if we go this
way, we will end up adding macros every time when adding a new option,
which seems not a good idea. I'm really not sure we need to change the
protocol version or the macro. Commit
366283961ac0ed6d89014444c6090f3fd02fce0a introduced the 'origin'
subscription parameter that is also sent to the publisher, but we
didn't touch the protocol version at all.

---
Why do we not to delay sending COMMIT PREPARED messages?

---
+                /*
+                 * If we've requested to shut down, exit the process.
+                 *
+                 * Note that WalSndDone() cannot be used here because
the delaying
+                 * changes will be sent in the function.
+                 */
+                if (got_STOPPING)
+                        WalSndShutdown();

Since the walsender exits without sending the done message at a server
shutdown, we get the following log message on the subscriber:

ERROR:  could not receive data from WAL stream: server closed the
connection unexpectedly

I think that since the walsender is just waiting for sending data, it
can send the done message if the socket is writable.

---
+                delayUntil = TimestampTzPlusMilliseconds(delay_start,
ctx->min_send_delay);
+                remaining_wait_time_ms =
TimestampDifferenceMilliseconds(GetCurrentTimestamp(), delayUntil);
+
(snip)
+
+                /* Sleep until appropriate time. */
+                timeout_sleeptime_ms =
WalSndComputeSleeptime(GetCurrentTimestamp());

I think it's better to call GetCurrentTimestamp() only once.

---
+# This test is successful only if at least the configured delay has elapsed.
+ok( time() - $publisher_insert_time >= $delay,
+        "subscriber applies WAL only after replication delay for
non-streaming transaction"
+);

The subscriber doesn't actually apply WAL records, but logically
replicated changes. How about "subscriber applies changes only
after..."?

Regards,

--
Masahiko Sawada
Amazon Web Services: https://aws.amazon.com

---

From: Amit Kapila <[email-masked]>
Date: Mon, 27 Feb 2023 12:03:48 +0530

> didn't touch the protocol version at all.

Right, I also don't see a reason to do anything for this. We have
previously bumped the protocol version when we send extra/additional
information from walsender but here that is not the requirement, so
this change doesn't seem to be required.

> Why do we not to delay sending COMMIT PREPARED messages?

I think we need to either add delay for prepare or commit prepared as
otherwise, it will lead to delaying the xact more than required. The
patch seems to add a delay before sending a PREPARE as that is the
time when the subscriber will apply the changes.
-- 
With Regards,
Amit Kapila.

---

From: Masahiko Sawada <[email-masked]>
Date: Mon, 27 Feb 2023 17:20:21 +0900

On Mon, Feb 27, 2023 at 3:34 PM Amit Kapila <[email-masked]> wrote:
> otherwise, it will lead to delaying the xact more than required.

Agreed.

> The

Considering the purpose of this feature mentioned in the commit
message "particularly to fix errors that might cause data loss",
delaying sending PREPARE would really help that situation? For
example, even after (mistakenly) executing PREPARE for a transaction
executing DELETE without WHERE clause on the publisher the user still
can rollback the transaction. They don't lose data on both nodes yet.
After executing (and replicating) COMMIT PREPARED for that
transaction, they lose the data on both nodes. IIUC the time-delayed
logical replication should help this situation by delaying sending
COMMIT PREPARED so that, for example, the user can stop logical
replication before COMMIT PREPARED message arrives to the subscriber.
So I think we should delay sending COMMIT PREPARED (and COMMIT)
instead of PREPARE. This would help users to correct data loss errors,
and would be more consistent with what recovery_min_apply_delay does.

Regards,

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 27 Feb 2023 08:51:29 +0000

Dear Sawada-san, Amit,

Thank you for reviewing!

> didn't touch the protocol version at all.

I removed the protocol number.

I checked the previous discussion[1]. According to it, the protocol version must
be modified when new message is added or exiting messages are changed.
This patch intentionally make walsenders delay sending data, and at that time no
extra information is added. Therefore I think it is not needed.

> Why do we not to delay sending COMMIT PREPARED messages?

This is motivated by the comment[2] but I preferred your opinion[3].
Now COMMIT PREPARED is delayed instead of PREPARE message.

> can send the done message if the socket is writable.

You are right. I was confused with the previous implementation that workers cannot
accept any messages. I make walsenders send the end-command message directly.
Is it what you expeced?

> I think it's better to call GetCurrentTimestamp() only once.

Right, fixed.

> replicated changes. How about "subscriber applies changes only

I grepped other tests, and I could not find the same usage of the word "WAL".
So fixed as you said.

In next version I will use grammar checker like Chat-GPT to modify commit messages...

[1]: https://www.postgresql.org/message-id/CAA4eK1LjOm6-OHggYVH35dQ_v40jOXrJW0GFy3kuwTd2J48%3DUg%40mail.gmail.com
[2]: https://www.postgresql.org/message-id/CAA4eK1K4uPbudrNdH%2B%3D_vN-Hpe9wYh%3D3vBS5Ww9dHn-LOWMV0g%40mail.gmail.com
[3]: https://www.postgresql.org/message-id/CAD21AoA0mPq_m6USfAC8DAkvFfwjqGvGq++Uv=[email-masked]

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Amit Kapila <[email-masked]>
Date: Mon, 27 Feb 2023 14:56:19 +0530

> and would be more consistent with what recovery_min_apply_delay does.

The one difference w.r.t recovery_min_apply_delay is that here we will
hold locks for the duration of the delay which didn't seem to be a
good idea. This will also probably lead to more bloat as we will keep
transactions open for a long time. Doing it before DecodePrepare won't
have such problems. This is the reason that we decide to perform a
delay at the start of the transaction instead at commit/prepare in the
subscriber-side approach.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Tue, 28 Feb 2023 14:05:52 +0530

On Mon, Feb 27, 2023 at 2:21 PM Hayato Kuroda (Fujitsu)
<[email-masked]> wrote:

Few comments:
1.
+ /*
+ * If we've requested to shut down, exit the process.
+ *
+ * Note that WalSndDone() cannot be used here because the delaying
+ * changes will be sent in the function.
+ */
+ if (got_STOPPING)
+ {
+ QueryCompletion qc;
+
+ /* Inform the standby that XLOG streaming is done */
+ SetQueryCompletion(&qc, CMDTAG_COPY, 0);
+ EndCommand(&qc, DestRemote, false);
+ pq_flush();

Do we really need to do anything except for breaking the loop and let
the exit handling happen in the main loop when 'got_STOPPING' is set?
AFAICS, this is what we are doing in some other palces (See
WalSndWaitForWal). Won't that work? It seems that will help us sending
all the pending WAL.

2.
+ /* Try to flush pending output to the client */
+ if (pq_flush_if_writable() != 0)
+ WalSndShutdown();

Is there a reason to try flushing here?

Apart from the above, I have made a few changes in the comments in the
attached diff patch. If you agree with those then please include them
in the next version.

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Tue, 28 Feb 2023 15:51:32 +0000

Dear Amit,

Thank you for reviewing! PSA new version.
Note that the starting point of delay for 2PC was not changed,
I think it has been under discussion.

> WalSndWaitForWal). Won't that work? It seems that will help us sending

If we exit the loop after got_STOPPING is set, as you said, the walsender will
send delaying changes and then exit. The behavior is same as the case that WalSndDone()
is called. But I think it is not suitable for the motivation of the feature.
If users notice the miss operation like TRUNCATE, they must shut down the publisher
once and then recovery from back up or old subscriber. If the walsender sends all
pending changes, miss operations will be also propagated to subscriber and data
cannot be protected. So currently I want to keep the style.
FYI - In case of physical replication, received WALs are not applied when the
secondary is shutted down.

> Is there a reason to try flushing here?

IIUC if pq_flush_if_writable() returns non-zero (EOF), it means that there is a
trouble and walsender fails to send messages to subscriber.

In Linux, the stuck trace from pq_flush_if_writable() will finally reach the send() system call.
And according to man page[1], it will be triggered by some unexpected state or the connection is closed.

Based on above, I think the returned value should be confirmed.

> attached diff patch. If you agree with those then please include them

Thanks! I checked and I thought all of them should be included.

Moreover, I used grammar checker and slightly reworded the commit message.

[1]: https://man7.org/linux/man-pages/man3/send.3p.html

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Peter Smith <[email-masked]>
Date: Wed, 1 Mar 2023 12:53:49 +1100

Here are some review comments for v9-0001, but these are only very trivial.

======
Commit Message

1.
Nitpick. The new text is jagged-looking. It should wrap at ~80 chars.

~~~

2.
2. Another reason is for that parallel streaming, the transaction will be opened
immediately by the parallel apply worker. Therefore, if the walsender
is delayed in sending the final record of the transaction, the
parallel apply worker must wait to receive it with an open
transaction. This would result in the locks acquired during the
transaction not being released until the min_send_delay has elapsed.

~

The text already said there are "two reasons", and already this is
numbered as reason 2. So it doesn't need to keep saying "Another
reason" here.

"Another reason is for that parallel streaming" --> "For parallel streaming..."

======
src/backend/replication/walsender.c

3. WalSndDelay

+ /* die if timeout was reached */
+ WalSndCheckTimeOut();

Other nearby comments start uppercase, so this should too.

======
src/include/replication/walreceiver.h

4. WalRcvStreamOptions

@@ -187,6 +187,7 @@ typedef struct
  * prepare time */
  char    *origin; /* Only publish data originating from the
  * specified origin */
+ int32 min_send_delay; /* The minimum send delay */
  } logical;
  } proto;
 } WalRcvStreamOptions;

~

Should that comment mention the units are "(ms)"

------
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: Masahiko Sawada <[email-masked]>
Date: Wed, 1 Mar 2023 11:48:20 +0900

> Moreover, I used grammar checker and slightly reworded the commit message.

Thinking of side effects of this feature (no matter where we delay
applying the changes), on the publisher, vacuum cannot collect garbage
and WAL cannot be recycled. Is that okay in the first place? The point
is that the subscription setting affects the publisher. That is,
min_send_delay is specified on the subscriber but the symptoms that
could ultimately lead to a server crash appear on the publisher, which
sounds dangerous to me.

Imagine a service or system like where there is a publication server
and it's somewhat exposed so that a user (or a subsystem) arbitrarily
can create a subscriber to replicate a subset of the data. A malicious
user can have the publisher crash by creating a subscription with,
say, min_send_delay = 20d. max_slot_wal_keep_size helps this situation
but it's -1 by default.

Regards,

---

From: Amit Kapila <[email-masked]>
Date: Wed, 1 Mar 2023 10:24:50 +0530

> say, min_send_delay = 20d. max_slot_wal_keep_size helps this situation

By publisher crash, do you mean due to the disk full situation, it can
lead the publisher to stop/panic? Won't a malicious user can block the
replication in other ways as well and let the publisher stall (or
crash the publisher) even without setting min_send_delay? Basically,
one needs to either disable the subscription or create a
constraint-violating row in the table to make that happen. If the
system is exposed for arbitrarily allowing the creation of a
subscription then a malicious user can create a subscription similar
to one existing subscription and block the replication due to
constraint violations. I don't think it would be so easy to bypass the
current system that a malicious user will be allowed to create/alter
subscriptions arbitrarily. Similarly, if there is a network issue
(unreachable or slow), one will see similar symptoms. I think
retention of data and WAL on publisher do rely on acknowledgment from
subscribers and delay in that due to any reason can lead to the
symptoms you describe above. We have documented at least one such case
already where during Drop Subscription, if the network is not
reachable then also, a similar problem can happen and users need to be
careful about it [1].

[1] - https://www.postgresql.org/docs/devel/logical-replication-subscription.html

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Wed, 1 Mar 2023 10:44:49 +0530

> secondary is shutted down.

Fair point but I think the current comment should explain why we are
doing something different here. How about extending the existing
comments to something like: "If we've requested to shut down, exit the
process. This is unlike handling at other places where we allow
complete WAL to be sent before shutdown because we don't want the
delayed transactions to be applied downstream. This will allow one to
use the data from downstream in case of some unwanted operations on
the current node."

-- 
With Regards,
Amit Kapila.

---

From: vignesh C <[email-masked]>
Date: Wed, 1 Mar 2023 21:59:13 +0530

> Thank you for reviewing! PSA new version.

Thanks for the updated patch, few comments:
1) Currently we have added the delay during the decode of commit,
while decoding the commit walsender process will stop decoding any
further transaction until delay is completed. There might be a
possibility that a lot of transactions will happen in parallel and
there will be a lot of transactions to be decoded after the delay is
completed.
Will it be possible to decode the WAL if any WAL is generated instead
of staying idle in the meantime, I'm not sure if this is feasible just
throwing my thought to see if it might be possible.
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -676,6 +676,15 @@ DecodeCommit(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf,

buf->origptr, buf->endptr);
        }

+       /*
+        * Delay sending the changes if required. For streaming transactions,
+        * this means a delay in sending the last stream but that is OK
+        * because on the downstream the changes will be applied only after
+        * receiving the last stream.
+        */
+       if (ctx->min_send_delay > 0 && ctx->delay_send)
+               ctx->delay_send(ctx, xid, commit_time);
+

2) Generally single line comments are not terminated by ".", The
comment "/* Sleep until appropriate time. */" can be changed
appropriately:
+
+               /* Sleep until appropriate time. */
+               timeout_sleeptime_ms = WalSndComputeSleeptime(now);
+
+               elog(DEBUG2, "time-delayed replication for txid %u,
delay_time = %d ms, remaining wait time: %ld ms",
+                        xid, (int) ctx->min_send_delay,
remaining_wait_time_ms);
+
+               /* Sleep until we get reply from worker or we time out */
+               WalSndWait(WL_SOCKET_READABLE,

3) In some places we mention as min_send_delay and in some places we
mention it as time-delayed replication, we can keep the comment
consistent by using the similar wordings.
+-- fail - specifying streaming = parallel with time-delayed replication is not
+-- supported
+CREATE SUBSCRIPTION regress_testsub CONNECTION
'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect =
false, streaming = parallel, min_send_delay = 123);

+-- fail - alter subscription with streaming = parallel should fail when
+-- time-delayed replication is set
+ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);

+-- fail - alter subscription with min_send_delay should fail when
+-- streaming = parallel is set

4) Since the value is stored in ms, we need not add ms again as the
default value is in ms:
@@ -4686,6 +4694,9 @@ dumpSubscription(Archive *fout, const
SubscriptionInfo *subinfo)
        if (strcmp(subinfo->subsynccommit, "off") != 0)
                appendPQExpBuffer(query, ", synchronous_commit = %s",
fmtId(subinfo->subsynccommit));

+       if (subinfo->subminsenddelay > 0)
+               appendPQExpBuffer(query, ", min_send_delay = '%d ms'",
subinfo->subminsenddelay);
+

5) we can use the new error reporting style:
5.a) brackets around errcode can be removed
+               ereport(ERROR,
+                               (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                                errmsg("invalid value for parameter
\"%s\": \"%s\"",
+                                               "min_send_delay", input_string),
+                                hintmsg ? errhint("%s", _(hintmsg)) : 0));

5.b) Similarly here too;
+       if (result < 0 || result > PG_INT32_MAX)
+               ereport(ERROR,
+                               (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                                errmsg("%d ms is outside the valid
range for parameter \"%s\" (%d .. %d)",
+                                               result,
+                                               "min_send_delay",
+                                               0, PG_INT32_MAX)));

5.c) Similarly here too;
+                       delay_val = strtoul(strVal(defel->arg), &endptr, 10);
+                       if (errno != 0 || *endptr != '\0')
+                               ereport(ERROR,
+
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                                                errmsg("invalid
min_send_delay")));

5.d) Similarly here too;
+                       if (delay_val > PG_INT32_MAX)
+                               ereport(ERROR,
+
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+
errmsg("min_send_delay \"%s\" out of range",
+
strVal(defel->arg))));

6) This can be changed to a single line comment:
+       /*
+        * Parse given string as parameter which has millisecond unit
+        */
+       if (!parse_int(input_string, &result, GUC_UNIT_MS, &hintmsg))
+               ereport(ERROR,

7) In the expect we have specifically mention "for non-streaming
transaction", is the behavior different for streaming transaction, if
not we can change the message accordingly
+# The publisher waits for the replication to complete
+$node_publisher->wait_for_catchup('tap_sub_renamed');
+
+# This test is successful only if at least the configured delay has elapsed.
+ok( time() - $publisher_insert_time >= $delay,
+       "subscriber applies changes only after replication delay for
non-streaming transaction"
+);

Regards,
Vignesh

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Thu, 2 Mar 2023 13:25:17 +0000

Dear Peter,

> Nitpick. The new text is jagged-looking. It should wrap at ~80 chars.

Addressed.

> "Another reason is for that parallel streaming" --> "For parallel streaming..."

Changed.

> Other nearby comments start uppercase, so this should too.

I just picked from other part and they have lowercase, but fixed.

> Should that comment mention the units are "(ms)"

Added.

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Thu, 2 Mar 2023 13:27:16 +0000

Dear Vignesh,

Thank you for reviewing! New version can be available at [1].

> +               ctx->delay_send(ctx, xid, commit_time);

I see your point, but I think that extension can be done in future version if needed.
This is because we must change some parts and introduce some complexities.

If we have decoded but have not wanted to send changes yet, we must store them in
the memory one and skip sending. In order to do that we must add new data structure,
and we must add another path in DecodeCommit, DecodePrepare not to send changes
and in WalSndLoop() and other functions to send pending changes. These may not be sufficient. 

I'm now thinking aboves are not needed, we can modify later if the overhead of
decoding is quite large and we must do them very efficiently.

> +               WalSndWait(WL_SOCKET_READABLE,

Right, removed.

> +-- streaming = parallel is set

"time-delayed replication" was removed.

Right, fixed.

> input_string),

All of them are fixed.

> +       if (!parse_int(input_string, &result, GUC_UNIT_MS, &hintmsg))

Changed. I grepped ereport() in the patch and I thought there were no similar one.

> non-streaming transaction"

There is no difference, both of normal and streamed transaction could be delayed to apply.
So removed.

[1]: https://www.postgresql.org/message-id/flat/TYAPR01MB586606CF3B585B6F8BE13A9CF5B29%40TYAPR01MB5866.jpnprd01.prod.outlook.com

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Thu, 2 Mar 2023 13:27:49 +0000

Dear Amit,

> use the data from downstream in case of some unwanted operations on

Thank you for suggestion. I think it is better, so changed.
Please see new patch at [1]

Best Regards,
Hayato Kuroda
FUJITSU LIMITED