Re: Logical replication timeout problem

On Fri, Mar 18, 2022 at 10:43 AM wangw(dot)fnst(at)fujitsu(dot)com
<wangw(dot)fnst(at)fujitsu(dot)com> wrote:
>
> On Thu, Mar 17, 2022 at 7:52 PM Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
> >
>
> Attach the new patch.
>

*
case REORDER_BUFFER_CHANGE_INVALIDATION:
- /* Execute the invalidation messages locally */
- ReorderBufferExecuteInvalidations(
- change->data.inval.ninvalidations,
- change->data.inval.invalidations);
- break;
+ {
+ LogicalDecodingContext *ctx = rb->private_data;
+
+ /* Try to send a keepalive message. */
+ UpdateProgress(ctx, true);

Calling UpdateProgress() here appears adhoc to me especially because
it calls OutputPluginUpdateProgress which appears to be called only
from plugin API. Am, I missing something? Also why the same handling
is missed in other similar messages like
REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID where we don't call any
plug-in API?

I am not sure what is a good way to achieve this but one idea that
occurred to me was shall we invent a new callback
ReorderBufferSkipChangeCB similar to ReorderBufferApplyChangeCB and
then pgoutput can register its API where we can have the logic similar
to what you have in UpdateProgress()? If we do so, then all the
cuurent callers of UpdateProgress in pgoutput can also call that API.
What do you think?

* Why don't you have a quick exit like below code in WalSndWriteData?
/* Try taking fast path unless we get too close to walsender timeout. */
if (now < TimestampTzPlusMilliseconds(last_reply_timestamp,
wal_sender_timeout / 2) &&
!pq_is_send_pending())
{
return;
}

* Can we rename variable 'is_send' to 'change_sent'?

--
With Regards,
Amit Kapila.