Re: Logical Replication of sequences

On Mon, 27 Oct 2025 at 14:42, Chao Li <li(dot)evan(dot)chao(at)gmail(dot)com> wrote:
>
>
> > On Oct 24, 2025, at 23:22, vignesh C <vignesh21(at)gmail(dot)com> wrote:
> >
> > Regards,
> > Vignesh
> > <v20251024-0001-Rename-sync_error_count-to-tbl_sync_error_.patch><v20251024-0002-Add-worker-type-argument-to-logicalrep_wor.patch><v20251024-0003-New-worker-for-sequence-synchronization-du.patch><v20251024-0004-Documentation-for-sequence-synchronization.patch>
>
>
> The changes in 0001 are straightforward, looks good. I haven’t reviewed 0004 yet. Got a few comments for 0002 and 0003.

> 5 - 0003
> ```
> +/*
> + * Reset the last_seqsync_start_time of the sequencesync worker in the
> + * subscription's apply worker.
> + */
> +void
> +logicalrep_reset_seqsync_start_time(void)
> +{
> + LogicalRepWorker *worker;
> +
> + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
> +
> + /*
> + * Set the last_seqsync_start_time for the sequence worker in the apply
> + * worker instead of the sequence sync worker, as the sequence sync worker
> + * has finished and is about to exit.
> + */
> + worker = logicalrep_worker_find(MyLogicalRepWorker->subid, InvalidOid,
> + WORKERTYPE_APPLY, true);
> + if (worker)
> + worker->last_seqsync_start_time = 0;
> +
> + LWLockRelease(LogicalRepWorkerLock);
> +}
> ```
>
> Two comments for this new function:
>
> * The function comment and in-code comment are redundant. Suggesting move the in-code comment to function comment.
> * Why LW_SHARED is used? We are writing worker->last_seqsync_start_time, shouldn’t LW_EXCLUSIVE be used?

There will be only one sequence sync worker and only this process is
going to update this, so LW_SHARED is enough to find the apply worker.

> 6 - 0003
> ```
> + /*
> + * Count running sync workers for this subscription, while we have the
> + * lock.
> + */
> + nsyncworkers = logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
> + LWLockRelease(LogicalRepWorkerLock);
> +
> + launch_sync_worker(nsyncworkers, InvalidOid,
> + &MyLogicalRepWorker->last_seqsync_start_time);
> ```
>
> I think here could be a race condition. Because the lock is acquired in LW_SHARED, meaning multiple caller may get the same nsyncworkers. Then it launches sync worker based on nsyncworkers, which would use inaccurate nsyncworkers, because between LWLockRelease() and launch_sync_worker(), another worker might be started.
>
> But if that is not the case, only one caller should call ProcessSyncingSequencesForApply(), then why the lock is needed?

Sequence sync worker will be started only by the apply worker, another
worker cannot be started for this subscription between LWLockRelease()
and launch_sync_worker() as this apply worker is responsible for it
and the apply worker is active with current work. Same logic is used
for table sync workers too.

> 7 - 0003
> ```
> + if (insuffperm_seqs->len)
> + {
> + appendStringInfo(combined_error_detail, "Insufficient permission for sequence(s): (%s)",
> + insuffperm_seqs->data);
> + appendStringInfoString(combined_error_hint, "Grant permissions for the sequence(s).");
> + }
> ```
>
> “Grant permissions” is unclear. Should it be “Grant UPDATE privilege”?

Modified

> 8 - 0003
> ```
> + appendStringInfoString(combined_error_hint, " For mismatched sequences, alter or re-create local sequences to have matching parameters as publishers.");
> ```
>
> “To have matching parameters as publishers” grammatically sound not good. Maybe revision to “to match the publisher’s parameters”.

Modified

> 9 - 0003
> ```
> + /*
> + * current_indexes is not incremented sequentially because some
> + * sequences may be missing, and the number of fetched rows may not
> + * match the batch size. The `hash_search` with HASH_REMOVE takes care
> + * of the count.
> + */
> ```
>
> Typo: current_indexes => current_index

Modified

> 10 - 0003
> ```
> - /* Find the leader apply worker and signal it. */
> - logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
> + /*
> + * This is a clean exit of the sequencesync worker; reset the
> + * last_seqsync_start_time.
> + */
> + if (wtype == WORKERTYPE_SEQUENCESYNC)
> + logicalrep_reset_seqsync_start_time();
> + else
> + /* Find the leader apply worker and signal it. */
> + logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
> ```
>
> The comment “this is a clean exist of sequencsync worker” is specific to “if”, so suggesting moving into “if”. And “this is a clean exis of the sequencesyc worker” is not needed, keep consistent with the comment in “else”.

Modified

> 11 - 0003
> ```
> +void
> +launch_sync_worker(int nsyncworkers, Oid relid, TimestampTz *last_start_time)
> +{
> + /* If there is a free sync worker slot, start a new sync worker */
> + if (nsyncworkers < max_sync_workers_per_subscription)
> + {
> ```
>
> The entire function is under an “if”, so we can do “if (!…) return”, so saves a level of indent.

Modified

Peter's comments from [1] have also been addressed. The attached
v20251029 version patch has the changes for the same.

[1] - https://www.postgresql.org/message-id/CAHut%2BPtMc1fr6cQvUAnxRE%2Bbuim5m-d9M2dM0YAeEHNkS9KzBw%40mail.gmail.com

Regards,
Vignesh