Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : * sequencesync.c
3 : : * PostgreSQL logical replication: sequence synchronization
4 : : *
5 : : * Copyright (c) 2025-2026, PostgreSQL Global Development Group
6 : : *
7 : : * IDENTIFICATION
8 : : * src/backend/replication/logical/sequencesync.c
9 : : *
10 : : * NOTES
11 : : * This file contains code for sequence synchronization for
12 : : * logical replication.
13 : : *
14 : : * Sequences requiring synchronization are tracked in the pg_subscription_rel
15 : : * catalog.
16 : : *
17 : : * Sequences to be synchronized will be added with state INIT when either of
18 : : * the following commands is executed:
19 : : * CREATE SUBSCRIPTION
20 : : * ALTER SUBSCRIPTION ... REFRESH PUBLICATION
21 : : *
22 : : * Executing the following command resets all sequences in the subscription to
23 : : * state INIT, triggering re-synchronization:
24 : : * ALTER SUBSCRIPTION ... REFRESH SEQUENCES
25 : : *
26 : : * The apply worker periodically scans pg_subscription_rel for sequences in
27 : : * INIT state. When such sequences are found, it spawns a sequencesync worker
28 : : * to handle synchronization.
29 : : *
30 : : * A single sequencesync worker is responsible for synchronizing all sequences.
31 : : * It begins by retrieving the list of sequences that are flagged for
32 : : * synchronization, i.e., those in the INIT state. These sequences are then
33 : : * processed in batches, allowing multiple entries to be synchronized within a
34 : : * single transaction. The worker fetches the current sequence values and page
35 : : * LSNs from the remote publisher, updates the corresponding sequences on the
36 : : * local subscriber, and finally marks each sequence as READY upon successful
37 : : * synchronization.
38 : : *
39 : : * Sequence state transitions follow this pattern:
40 : : * INIT -> READY
41 : : *
42 : : * To avoid creating too many transactions, up to MAX_SEQUENCES_SYNC_PER_BATCH
43 : : * sequences are synchronized per transaction. The locks on the sequence
44 : : * relation will be periodically released at each transaction commit.
45 : : *
46 : : * XXX: We didn't choose launcher process to maintain the launch of sequencesync
47 : : * worker as it didn't have database connection to access the sequences from the
48 : : * pg_subscription_rel system catalog that need to be synchronized.
49 : : *-------------------------------------------------------------------------
50 : : */
51 : :
52 : : #include "postgres.h"
53 : :
54 : : #include "access/genam.h"
55 : : #include "access/table.h"
56 : : #include "catalog/pg_sequence.h"
57 : : #include "catalog/pg_subscription_rel.h"
58 : : #include "commands/sequence.h"
59 : : #include "pgstat.h"
60 : : #include "postmaster/interrupt.h"
61 : : #include "replication/logicalworker.h"
62 : : #include "replication/worker_internal.h"
63 : : #include "storage/lmgr.h"
64 : : #include "storage/lwlock.h"
65 : : #include "utils/acl.h"
66 : : #include "utils/builtins.h"
67 : : #include "utils/fmgroids.h"
68 : : #include "utils/guc.h"
69 : : #include "utils/inval.h"
70 : : #include "utils/lsyscache.h"
71 : : #include "utils/memutils.h"
72 : : #include "utils/pg_lsn.h"
73 : : #include "utils/syscache.h"
74 : : #include "utils/usercontext.h"
75 : :
76 : : #define REMOTE_SEQ_COL_COUNT 11
77 : :
78 : : typedef enum CopySeqResult
79 : : {
80 : : COPYSEQ_SUCCESS,
81 : : COPYSEQ_MISMATCH,
82 : : COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM,
83 : : COPYSEQ_PUBLISHER_INSUFFICIENT_PERM,
84 : : COPYSEQ_SKIPPED,
85 : : COPYSEQ_NOT_SUBSCRIBED
86 : : } CopySeqResult;
87 : :
88 : : static List *seqinfos = NIL;
89 : :
90 : : /*
91 : : * Apply worker determines if sequence synchronization is needed.
92 : : *
93 : : * Start a sequencesync worker if one is not already running. The active
94 : : * sequencesync worker will handle all pending sequence synchronization. If any
95 : : * sequences remain unsynchronized after it exits, a new worker can be started
96 : : * in the next iteration.
97 : : */
98 : : void
99 : 7243 : ProcessSequencesForSync(void)
100 : : {
101 : : LogicalRepWorker *sequencesync_worker;
102 : : int nsyncworkers;
103 : : bool has_pending_sequences;
104 : : bool started_tx;
105 : :
106 : 7243 : FetchRelationStates(NULL, &has_pending_sequences, &started_tx);
107 : :
108 [ + + ]: 7243 : if (started_tx)
109 : : {
110 : 199 : CommitTransactionCommand();
111 : 199 : pgstat_report_stat(true);
112 : : }
113 : :
114 [ + + ]: 7243 : if (!has_pending_sequences)
115 : 7210 : return;
116 : :
117 : 58 : LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
118 : :
119 : : /* Check if there is a sequencesync worker already running? */
120 : 58 : sequencesync_worker = logicalrep_worker_find(WORKERTYPE_SEQUENCESYNC,
121 : 58 : MyLogicalRepWorker->subid,
122 : : InvalidOid, true);
123 [ + + ]: 58 : if (sequencesync_worker)
124 : : {
125 : 25 : LWLockRelease(LogicalRepWorkerLock);
126 : 25 : return;
127 : : }
128 : :
129 : : /*
130 : : * Count running sync workers for this subscription, while we have the
131 : : * lock.
132 : : */
133 : 33 : nsyncworkers = logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
134 : 33 : LWLockRelease(LogicalRepWorkerLock);
135 : :
136 : : /*
137 : : * It is okay to read/update last_seqsync_start_time here in apply worker
138 : : * as we have already ensured that sync worker doesn't exist.
139 : : */
140 : 33 : launch_sync_worker(WORKERTYPE_SEQUENCESYNC, nsyncworkers, InvalidOid,
141 : 33 : &MyLogicalRepWorker->last_seqsync_start_time);
142 : : }
143 : :
144 : : /*
145 : : * get_sequences_string
146 : : *
147 : : * Build a comma-separated string of schema-qualified sequence names
148 : : * for the given list of sequence indexes.
149 : : */
150 : : static void
151 : 8 : get_sequences_string(List *seqindexes, StringInfo buf)
152 : : {
153 : 8 : resetStringInfo(buf);
154 [ + - + + : 24 : foreach_int(seqidx, seqindexes)
+ + ]
155 : : {
156 : : LogicalRepSequenceInfo *seqinfo =
157 : 8 : (LogicalRepSequenceInfo *) list_nth(seqinfos, seqidx);
158 : :
159 [ - + ]: 8 : if (buf->len > 0)
160 : 0 : appendStringInfoString(buf, ", ");
161 : :
162 : 8 : appendStringInfo(buf, "\"%s.%s\"", seqinfo->nspname, seqinfo->seqname);
163 : : }
164 : 8 : }
165 : :
166 : : /*
167 : : * Error callback to give more context info about the sequence being
168 : : * synchronized.
169 : : */
170 : : static void
171 : 1 : sequence_sync_error_callback(void *arg)
172 : : {
173 : 1 : LogicalRepSequenceInfo *seqinfo = *(LogicalRepSequenceInfo **) arg;
174 : :
175 [ - + ]: 1 : if (seqinfo == NULL)
176 : 0 : return;
177 : :
178 : 1 : errcontext("synchronizing sequence \"%s.%s\" for subscription \"%s\"",
179 : 1 : seqinfo->nspname, seqinfo->seqname, MySubscription->name);
180 : : }
181 : :
182 : : /*
183 : : * report_sequence_errors
184 : : *
185 : : * Report discrepancies found during sequence synchronization between
186 : : * the publisher and subscriber. Emits warnings for:
187 : : * a) mismatched definitions or concurrent rename
188 : : * b) insufficient privileges on the subscriber
189 : : * c) insufficient privileges on the publisher
190 : : * d) missing sequences on the publisher
191 : : * Then raises an ERROR to indicate synchronization failure.
192 : : */
193 : : static void
194 : 16 : report_sequence_errors(List *mismatched_seqs_idx,
195 : : List *sub_insuffperm_seqs_idx,
196 : : List *pub_insuffperm_seqs_idx,
197 : : List *missing_seqs_idx)
198 : : {
199 : : StringInfoData seqstr;
200 : :
201 : : /* Quick exit if there are no errors to report */
202 [ + + + - : 16 : if (!mismatched_seqs_idx && !sub_insuffperm_seqs_idx &&
+ + ]
203 [ + + ]: 11 : !pub_insuffperm_seqs_idx && !missing_seqs_idx)
204 : 8 : return;
205 : :
206 : 8 : initStringInfo(&seqstr);
207 : :
208 [ + + ]: 8 : if (mismatched_seqs_idx)
209 : : {
210 : 4 : get_sequences_string(mismatched_seqs_idx, &seqstr);
211 [ + - ]: 4 : ereport(WARNING,
212 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
213 : : errmsg_plural("mismatched or renamed sequence on subscriber (%s)",
214 : : "mismatched or renamed sequences on subscriber (%s)",
215 : : list_length(mismatched_seqs_idx),
216 : : seqstr.data));
217 : : }
218 : :
219 [ - + ]: 8 : if (sub_insuffperm_seqs_idx)
220 : : {
221 : 0 : get_sequences_string(sub_insuffperm_seqs_idx, &seqstr);
222 : :
223 : : /*
224 : : * With run_as_owner enabled, sequence synchronization runs as the
225 : : * subscription owner, so a missing UPDATE privilege should be granted
226 : : * to that role. Otherwise, the worker switches to the sequence owner
227 : : * before checking privileges, so no useful GRANT hint can be
228 : : * provided.
229 : : */
230 [ # # # # ]: 0 : ereport(WARNING,
231 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
232 : : errmsg_plural("insufficient privileges on subscriber sequence (%s)",
233 : : "insufficient privileges on subscriber sequences (%s)",
234 : : list_length(sub_insuffperm_seqs_idx),
235 : : seqstr.data),
236 : : MySubscription->runasowner ?
237 : : errhint_plural("Grant UPDATE on the sequence to the subscription "
238 : : "owner on the subscriber.",
239 : : "Grant UPDATE on the sequences to the subscription "
240 : : "owner on the subscriber.",
241 : : list_length(sub_insuffperm_seqs_idx)) : 0);
242 : : }
243 : :
244 [ + + ]: 8 : if (pub_insuffperm_seqs_idx)
245 : : {
246 : 1 : get_sequences_string(pub_insuffperm_seqs_idx, &seqstr);
247 [ + - ]: 1 : ereport(WARNING,
248 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
249 : : errmsg_plural("insufficient privileges on publisher sequence (%s)",
250 : : "insufficient privileges on publisher sequences (%s)",
251 : : list_length(pub_insuffperm_seqs_idx),
252 : : seqstr.data),
253 : : errhint_plural("Grant SELECT on the sequence to the role used for "
254 : : "the replication connection on the publisher.",
255 : : "Grant SELECT on the sequences to the role used for "
256 : : "the replication connection on the publisher.",
257 : : list_length(pub_insuffperm_seqs_idx)));
258 : : }
259 : :
260 [ + + ]: 8 : if (missing_seqs_idx)
261 : : {
262 : 3 : get_sequences_string(missing_seqs_idx, &seqstr);
263 [ + - ]: 3 : ereport(WARNING,
264 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
265 : : errmsg_plural("missing sequence on publisher (%s)",
266 : : "missing sequences on publisher (%s)",
267 : : list_length(missing_seqs_idx),
268 : : seqstr.data));
269 : : }
270 : :
271 [ + - ]: 8 : ereport(ERROR,
272 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
273 : : errmsg("logical replication sequence synchronization failed for subscription \"%s\"",
274 : : MySubscription->name));
275 : : }
276 : :
277 : : /*
278 : : * get_and_validate_seq_info
279 : : *
280 : : * Extracts remote sequence information from the tuple slot received from the
281 : : * publisher, and validates it against the corresponding local sequence
282 : : * definition.
283 : : */
284 : : static CopySeqResult
285 : 36 : get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel,
286 : : LogicalRepSequenceInfo **seqinfo, int *seqidx)
287 : : {
288 : : bool isnull;
289 : 36 : int col = 0;
290 : : Datum datum;
291 : : bool remote_has_select_priv;
292 : : Oid remote_typid;
293 : : int64 remote_start;
294 : : int64 remote_increment;
295 : : int64 remote_min;
296 : : int64 remote_max;
297 : : bool remote_cycle;
298 : 36 : CopySeqResult result = COPYSEQ_SUCCESS;
299 : : HeapTuple tup;
300 : : Form_pg_sequence local_seq;
301 : : LogicalRepSequenceInfo *seqinfo_local;
302 : :
303 : 36 : *seqidx = DatumGetInt32(slot_getattr(slot, ++col, &isnull));
304 : : Assert(!isnull);
305 : :
306 : : /*
307 : : * The publisher only echoes back an index that we put in the VALUES list,
308 : : * so this should always identify an entry of seqinfos. Check it anyway
309 : : * before using it as a list subscript, since list_nth() does not
310 : : * bounds-check on non-assert builds and we would then write the remote
311 : : * sequence state through a pointer fetched from beyond the list.
312 : : *
313 : : * This only keeps the subscript inside the list. An index that is wrong
314 : : * but still in range is not detected, and cannot be; the sequence it
315 : : * points at then receives another sequence's data. That is the same kind
316 : : * of damage as the publisher reporting a wrong value in any other column,
317 : : * and is likewise beyond what we can check.
318 : : */
319 [ + - - + ]: 36 : if (*seqidx < 0 || *seqidx >= list_length(seqinfos))
320 [ # # ]: 0 : elog(ERROR, "invalid sequence index %d received from the publisher",
321 : : *seqidx);
322 : :
323 : : /* Identify the corresponding local sequence for the given index. */
324 : 36 : *seqinfo = seqinfo_local =
325 : 36 : (LogicalRepSequenceInfo *) list_nth(seqinfos, *seqidx);
326 : :
327 : : /*
328 : : * has_sequence_privilege() itself returns NULL, rather than false, when
329 : : * the sequence has been dropped concurrently after it was identified in
330 : : * the catalog snapshot (see has_sequence_privilege_id()). Treat that as a
331 : : * missing sequence on the publisher.
332 : : */
333 : 36 : datum = slot_getattr(slot, ++col, &isnull);
334 [ + + ]: 36 : if (isnull)
335 : 1 : return COPYSEQ_SKIPPED;
336 : :
337 : 35 : remote_has_select_priv = DatumGetBool(datum);
338 : :
339 : : /*
340 : : * The remote sequence state can be NULL if the publisher lacks the
341 : : * required privileges or if the sequence was dropped concurrently after
342 : : * it was identified in the catalog snapshot (see pg_get_sequence_data()).
343 : : */
344 : 35 : datum = slot_getattr(slot, ++col, &isnull);
345 [ + + ]: 35 : if (isnull)
346 : : {
347 : : /*
348 : : * The sequence was dropped concurrently after it was identified in
349 : : * the catalog snapshot. Treat it as skipped (and, since it no longer
350 : : * exists on the publisher, ultimately missing).
351 : : */
352 [ - + ]: 1 : if (remote_has_select_priv)
353 : 0 : return COPYSEQ_SKIPPED;
354 : :
355 : : /*
356 : : * The publisher lacks the SELECT privilege required by
357 : : * pg_get_sequence_data(). Since has_sequence_privilege() returned
358 : : * false, not NULL, do not classify this sequence as missing on the
359 : : * publisher.
360 : : */
361 : 1 : seqinfo_local->found_on_pub = true;
362 : 1 : return COPYSEQ_PUBLISHER_INSUFFICIENT_PERM;
363 : : }
364 : :
365 : 34 : seqinfo_local->last_value = DatumGetInt64(datum);
366 : :
367 : 34 : seqinfo_local->is_called = DatumGetBool(slot_getattr(slot, ++col, &isnull));
368 : : Assert(!isnull);
369 : :
370 : 34 : seqinfo_local->page_lsn = DatumGetLSN(slot_getattr(slot, ++col, &isnull));
371 : : Assert(!isnull);
372 : :
373 : 34 : remote_typid = DatumGetObjectId(slot_getattr(slot, ++col, &isnull));
374 : : Assert(!isnull);
375 : :
376 : 34 : remote_start = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
377 : : Assert(!isnull);
378 : :
379 : 34 : remote_increment = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
380 : : Assert(!isnull);
381 : :
382 : 34 : remote_min = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
383 : : Assert(!isnull);
384 : :
385 : 34 : remote_max = DatumGetInt64(slot_getattr(slot, ++col, &isnull));
386 : : Assert(!isnull);
387 : :
388 : 34 : remote_cycle = DatumGetBool(slot_getattr(slot, ++col, &isnull));
389 : : Assert(!isnull);
390 : :
391 : : /* Sanity check */
392 : : Assert(col == REMOTE_SEQ_COL_COUNT);
393 : :
394 : 34 : seqinfo_local->found_on_pub = true;
395 : :
396 : 34 : *sequence_rel = try_table_open(seqinfo_local->localrelid, RowExclusiveLock);
397 : :
398 : : /* Sequence was concurrently dropped? */
399 [ - + ]: 34 : if (!*sequence_rel)
400 : 0 : return COPYSEQ_SKIPPED;
401 : :
402 : 34 : tup = SearchSysCache1(SEQRELID, ObjectIdGetDatum(seqinfo_local->localrelid));
403 : :
404 : : /* Sequence was concurrently dropped? */
405 [ - + ]: 34 : if (!HeapTupleIsValid(tup))
406 [ # # ]: 0 : elog(ERROR, "cache lookup failed for sequence %u",
407 : : seqinfo_local->localrelid);
408 : :
409 : 34 : local_seq = (Form_pg_sequence) GETSTRUCT(tup);
410 : :
411 : : /* Sequence parameters for remote/local are the same? */
412 [ + - ]: 34 : if (local_seq->seqtypid != remote_typid ||
413 [ + + ]: 34 : local_seq->seqstart != remote_start ||
414 [ + + ]: 33 : local_seq->seqincrement != remote_increment ||
415 [ + - ]: 30 : local_seq->seqmin != remote_min ||
416 [ + - ]: 30 : local_seq->seqmax != remote_max ||
417 [ - + ]: 30 : local_seq->seqcycle != remote_cycle)
418 : 4 : result = COPYSEQ_MISMATCH;
419 : :
420 : : /* Sequence was concurrently renamed? */
421 [ + - ]: 34 : if (strcmp(seqinfo_local->nspname,
422 : 34 : get_namespace_name(RelationGetNamespace(*sequence_rel))) ||
423 [ - + ]: 34 : strcmp(seqinfo_local->seqname, RelationGetRelationName(*sequence_rel)))
424 : 0 : result = COPYSEQ_MISMATCH;
425 : :
426 : 34 : ReleaseSysCache(tup);
427 : 34 : return result;
428 : : }
429 : :
430 : : /*
431 : : * Apply remote sequence state to local sequence and mark it as
432 : : * synchronized (READY).
433 : : */
434 : : static CopySeqResult
435 : 30 : copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner)
436 : : {
437 : : UserContext ucxt;
438 : : AclResult aclresult;
439 : 30 : bool run_as_owner = MySubscription->runasowner;
440 : 30 : Oid seqoid = seqinfo->localrelid;
441 : : Relation rel;
442 : :
443 : : /*
444 : : * Take the subscription object lock before checking whether this sequence
445 : : * is still part of the subscription. The lock is held until the end of
446 : : * the transaction, so the check and the state update below are protected
447 : : * from a concurrent ALTER SUBSCRIPTION ... REFRESH PUBLICATION.
448 : : *
449 : : * AlterSubscription() takes this lock in AccessExclusiveLock mode while
450 : : * removing pg_subscription_rel rows, so the row cannot be removed between
451 : : * the check and the state update.
452 : : */
453 : 30 : LockSharedObject(SubscriptionRelationId, MySubscription->oid, 0,
454 : : AccessShareLock);
455 : :
456 : : /*
457 : : * The sequence may no longer be part of the subscription, in which case
458 : : * there is nothing to synchronize and the caller just skips it.
459 : : */
460 [ - + ]: 30 : if (!SearchSysCacheExists2(SUBSCRIPTIONRELMAP,
461 : : ObjectIdGetDatum(seqoid),
462 : : ObjectIdGetDatum(MySubscription->oid)))
463 : 0 : return COPYSEQ_NOT_SUBSCRIBED;
464 : :
465 : : /*
466 : : * If the user did not opt to run as the owner of the subscription
467 : : * ('run_as_owner'), then copy the sequence as the owner of the sequence.
468 : : */
469 [ + - ]: 30 : if (!run_as_owner)
470 : 30 : SwitchToUntrustedUser(seqowner, &ucxt);
471 : :
472 : 29 : aclresult = pg_class_aclcheck(seqoid, GetUserId(), ACL_UPDATE);
473 : :
474 [ - + ]: 29 : if (aclresult != ACLCHECK_OK)
475 : : {
476 [ # # ]: 0 : if (!run_as_owner)
477 : 0 : RestoreUserContext(&ucxt);
478 : :
479 : 0 : return COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM;
480 : : }
481 : :
482 : : /*
483 : : * The log counter (log_cnt) tracks how many sequence values are still
484 : : * unused locally. It is only relevant to the local node and managed
485 : : * internally by nextval() when allocating new ranges. Since log_cnt does
486 : : * not affect the visible sequence state (like last_value or is_called)
487 : : * and is only used for local caching, it need not be copied to the
488 : : * subscriber during synchronization.
489 : : */
490 : 29 : SetSequence(seqoid, seqinfo->last_value, seqinfo->is_called);
491 : :
492 [ + - ]: 29 : if (!run_as_owner)
493 : 29 : RestoreUserContext(&ucxt);
494 : :
495 : 29 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
496 : :
497 : : /*
498 : : * Record the remote sequence's LSN in pg_subscription_rel and mark the
499 : : * sequence as READY. Both locks it needs are held already, the object
500 : : * lock from further up and the relation lock just taken, so say so rather
501 : : * than have it take and release them again.
502 : : */
503 : 29 : UpdateSubscriptionRelState(MySubscription->oid, seqoid, SUBREL_STATE_READY,
504 : : seqinfo->page_lsn, true);
505 : :
506 : 29 : table_close(rel, NoLock);
507 : :
508 : 29 : return COPYSEQ_SUCCESS;
509 : : }
510 : :
511 : : /*
512 : : * Copy existing data of sequences from the publisher.
513 : : */
514 : : static void
515 : 17 : copy_sequences(WalReceiverConn *conn)
516 : : {
517 : 17 : int cur_batch_base_index = 0;
518 : 17 : int n_seqinfos = list_length(seqinfos);
519 : 17 : List *mismatched_seqs_idx = NIL;
520 : 17 : List *missing_seqs_idx = NIL;
521 : 17 : List *sub_insuffperm_seqs_idx = NIL;
522 : 17 : List *pub_insuffperm_seqs_idx = NIL;
523 : : StringInfoData seqstr;
524 : : StringInfoData cmd;
525 : : MemoryContext oldctx;
526 : :
527 : : /*
528 : : * Sequence synchronization depends on publisher-side functionality
529 : : * introduced in PostgreSQL 19, so it cannot work against an older
530 : : * publisher.
531 : : */
532 [ - + ]: 17 : if (walrcv_server_version(conn) < 190000)
533 [ # # ]: 0 : ereport(ERROR,
534 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
535 : : errmsg("cannot synchronize sequences if the publisher is running a version earlier than PostgreSQL 19"));
536 : :
537 : 17 : initStringInfo(&seqstr);
538 : 17 : initStringInfo(&cmd);
539 : :
540 : : #define MAX_SEQUENCES_SYNC_PER_BATCH 100
541 : :
542 [ - + ]: 17 : elog(DEBUG1,
543 : : "logical replication sequence synchronization for subscription \"%s\" - total unsynchronized: %d",
544 : : MySubscription->name, n_seqinfos);
545 : :
546 [ + + ]: 33 : while (cur_batch_base_index < n_seqinfos)
547 : : {
548 : 17 : Oid seqRow[REMOTE_SEQ_COL_COUNT] = {INT4OID, BOOLOID, INT8OID,
549 : : BOOLOID, LSNOID, OIDOID, INT8OID, INT8OID, INT8OID, INT8OID, BOOLOID};
550 : 17 : int batch_size = 0;
551 : 17 : int batch_succeeded_count = 0;
552 : 17 : int batch_mismatched_count = 0;
553 : 17 : int batch_skipped_count = 0;
554 : 17 : int batch_sub_insuffperm_count = 0;
555 : 17 : int batch_pub_insuffperm_count = 0;
556 : : int batch_missing_count;
557 : :
558 : : WalRcvExecResult *res;
559 : : TupleTableSlot *slot;
560 : :
561 : 17 : StartTransactionCommand();
562 : 17 : maybe_reread_subscription();
563 : :
564 [ + + ]: 55 : for (int idx = cur_batch_base_index; idx < n_seqinfos; idx++)
565 : : {
566 : : char *nspname_literal;
567 : : char *seqname_literal;
568 : :
569 : : LogicalRepSequenceInfo *seqinfo =
570 : 38 : (LogicalRepSequenceInfo *) list_nth(seqinfos, idx);
571 : :
572 [ + + ]: 38 : if (seqstr.len > 0)
573 : 21 : appendStringInfoString(&seqstr, ", ");
574 : :
575 : 38 : nspname_literal = quote_literal_cstr(seqinfo->nspname);
576 : 38 : seqname_literal = quote_literal_cstr(seqinfo->seqname);
577 : :
578 : 38 : appendStringInfo(&seqstr, "(%s, %s, %d)",
579 : : nspname_literal, seqname_literal, idx);
580 : :
581 [ - + ]: 38 : if (++batch_size == MAX_SEQUENCES_SYNC_PER_BATCH)
582 : 0 : break;
583 : : }
584 : :
585 : : /*
586 : : * We deliberately avoid acquiring a local lock on the sequence before
587 : : * querying the publisher to prevent potential distributed deadlocks
588 : : * in bi-directional replication setups.
589 : : *
590 : : * Example scenario:
591 : : *
592 : : * - On each node, a background worker acquires a lock on a sequence
593 : : * as part of a sync operation.
594 : : *
595 : : * - Concurrently, a user transaction attempts to alter the same
596 : : * sequence, waiting on the background worker's lock.
597 : : *
598 : : * - Meanwhile, a query from the other node tries to access metadata
599 : : * that depends on the completion of the alter operation.
600 : : *
601 : : * - This creates a circular wait across nodes:
602 : : *
603 : : * Node-1: Query -> waits on Alter -> waits on Sync Worker
604 : : *
605 : : * Node-2: Query -> waits on Alter -> waits on Sync Worker
606 : : *
607 : : * Since each node only sees part of the wait graph, the deadlock may
608 : : * go undetected, leading to indefinite blocking.
609 : : *
610 : : * Note: Each entry in VALUES includes an index 'seqidx' that
611 : : * represents the sequence's position in the local 'seqinfos' list.
612 : : * This index is propagated to the query results and later used to
613 : : * directly map the fetched publisher sequence rows back to their
614 : : * corresponding local entries without relying on result order or name
615 : : * matching.
616 : : */
617 : 17 : appendStringInfo(&cmd,
618 : : "SELECT s.seqidx, has_sequence_privilege(c.oid, 'SELECT'),\n"
619 : : " ps.*, seq.seqtypid,\n"
620 : : " seq.seqstart, seq.seqincrement, seq.seqmin,\n"
621 : : " seq.seqmax, seq.seqcycle\n"
622 : : "FROM ( VALUES %s ) AS s (schname, seqname, seqidx)\n"
623 : : "JOIN pg_namespace n ON n.nspname = s.schname\n"
624 : : "JOIN pg_class c ON c.relnamespace = n.oid AND c.relname = s.seqname\n"
625 : : "JOIN pg_sequence seq ON seq.seqrelid = c.oid\n"
626 : : "JOIN LATERAL pg_get_sequence_data(seq.seqrelid) AS ps ON true\n",
627 : : seqstr.data);
628 : :
629 : 17 : res = walrcv_exec(conn, cmd.data, lengthof(seqRow), seqRow);
630 [ - + ]: 17 : if (res->status != WALRCV_OK_TUPLES)
631 [ # # ]: 0 : ereport(ERROR,
632 : : errcode(ERRCODE_CONNECTION_FAILURE),
633 : : errmsg("could not fetch sequence information from the publisher: %s",
634 : : res->err));
635 : :
636 : 17 : slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
637 [ + + ]: 52 : while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
638 : : {
639 : : CopySeqResult sync_status;
640 : 36 : LogicalRepSequenceInfo *seqinfo = NULL;
641 : 36 : Relation sequence_rel = NULL;
642 : : int seqidx;
643 : : ErrorContextCallback errcallback;
644 : :
645 [ - + ]: 36 : CHECK_FOR_INTERRUPTS();
646 : :
647 [ - + ]: 36 : if (ConfigReloadPending)
648 : : {
649 : 0 : ConfigReloadPending = false;
650 : 0 : ProcessConfigFile(PGC_SIGHUP);
651 : : }
652 : :
653 : : /*
654 : : * Set up callback to identify the sequence in error messages.
655 : : * seqinfo will be filled once get_and_validate_seq_info()
656 : : * resolves the sequence index.
657 : : */
658 : 36 : errcallback.callback = sequence_sync_error_callback;
659 : 36 : errcallback.arg = &seqinfo;
660 : 36 : errcallback.previous = error_context_stack;
661 : 36 : error_context_stack = &errcallback;
662 : :
663 : 36 : sync_status = get_and_validate_seq_info(slot, &sequence_rel,
664 : : &seqinfo, &seqidx);
665 [ + + ]: 36 : if (sync_status == COPYSEQ_SUCCESS)
666 : 30 : sync_status = copy_sequence(seqinfo,
667 : 30 : sequence_rel->rd_rel->relowner);
668 : :
669 : : /* Pop the error context stack */
670 : 35 : error_context_stack = errcallback.previous;
671 : :
672 [ + + - + : 35 : switch (sync_status)
+ - - ]
673 : : {
674 : 29 : case COPYSEQ_SUCCESS:
675 [ - + ]: 29 : elog(DEBUG1,
676 : : "logical replication synchronization for subscription \"%s\", sequence \"%s.%s\" has finished",
677 : : MySubscription->name, seqinfo->nspname,
678 : : seqinfo->seqname);
679 : 29 : batch_succeeded_count++;
680 : 29 : break;
681 : 4 : case COPYSEQ_MISMATCH:
682 : :
683 : : /*
684 : : * Remember mismatched sequences in a long-lived memory
685 : : * context since these will be used after the transaction
686 : : * is committed.
687 : : */
688 : 4 : oldctx = MemoryContextSwitchTo(ApplyContext);
689 : 4 : mismatched_seqs_idx = lappend_int(mismatched_seqs_idx,
690 : : seqidx);
691 : 4 : MemoryContextSwitchTo(oldctx);
692 : 4 : batch_mismatched_count++;
693 : 4 : break;
694 : 0 : case COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM:
695 : :
696 : : /*
697 : : * Remember sequences with insufficient privileges in a
698 : : * long-lived memory context since these will be used
699 : : * after the transaction is committed.
700 : : */
701 : 0 : oldctx = MemoryContextSwitchTo(ApplyContext);
702 : 0 : sub_insuffperm_seqs_idx = lappend_int(sub_insuffperm_seqs_idx,
703 : : seqidx);
704 : 0 : MemoryContextSwitchTo(oldctx);
705 : 0 : batch_sub_insuffperm_count++;
706 : 0 : break;
707 : 1 : case COPYSEQ_PUBLISHER_INSUFFICIENT_PERM:
708 : :
709 : : /*
710 : : * Remember sequences for which the publisher lacks the
711 : : * privileges required by pg_get_sequence_data().
712 : : */
713 : 1 : oldctx = MemoryContextSwitchTo(ApplyContext);
714 : 1 : pub_insuffperm_seqs_idx = lappend_int(pub_insuffperm_seqs_idx,
715 : : seqidx);
716 : 1 : MemoryContextSwitchTo(oldctx);
717 : 1 : batch_pub_insuffperm_count++;
718 : 1 : break;
719 : 1 : case COPYSEQ_SKIPPED:
720 : :
721 : : /*
722 : : * Concurrent removal of a sequence on the subscriber is
723 : : * treated as success, since the only viable action is to
724 : : * skip the corresponding sequence data. Missing sequences
725 : : * on the publisher are treated as ERROR.
726 : : */
727 [ - + ]: 1 : if (seqinfo->found_on_pub)
728 : : {
729 [ # # ]: 0 : ereport(LOG,
730 : : errmsg("skip synchronization of sequence \"%s.%s\" because it has been dropped concurrently",
731 : : seqinfo->nspname,
732 : : seqinfo->seqname));
733 : 0 : batch_skipped_count++;
734 : : }
735 : 1 : break;
736 : 0 : case COPYSEQ_NOT_SUBSCRIBED:
737 : :
738 : : /*
739 : : * A concurrent refresh removed this sequence from the
740 : : * subscription. Skipping it is the only sensible action,
741 : : * and it must not be treated as an error.
742 : : */
743 [ # # ]: 0 : ereport(LOG,
744 : : errmsg("skip synchronization of sequence \"%s.%s\" because it is no longer part of subscription \"%s\"",
745 : : seqinfo->nspname, seqinfo->seqname,
746 : : MySubscription->name));
747 : 0 : batch_skipped_count++;
748 : 0 : break;
749 : : }
750 : :
751 [ + + ]: 35 : if (sequence_rel)
752 : 33 : table_close(sequence_rel, NoLock);
753 : : }
754 : :
755 : 16 : ExecDropSingleTupleTableSlot(slot);
756 : 16 : walrcv_clear_result(res);
757 : 16 : resetStringInfo(&seqstr);
758 : 16 : resetStringInfo(&cmd);
759 : :
760 : 16 : batch_missing_count = batch_size - (batch_succeeded_count +
761 : 16 : batch_mismatched_count +
762 : 16 : batch_sub_insuffperm_count +
763 : 16 : batch_pub_insuffperm_count +
764 : : batch_skipped_count);
765 : :
766 [ - + ]: 16 : elog(DEBUG1,
767 : : "logical replication sequence synchronization for subscription \"%s\" - batch #%d = %d attempted, %d succeeded, %d mismatched, %d subscriber insufficient permission, %d publisher insufficient permission, %d missing from publisher, %d skipped",
768 : : MySubscription->name,
769 : : (cur_batch_base_index / MAX_SEQUENCES_SYNC_PER_BATCH) + 1,
770 : : batch_size, batch_succeeded_count, batch_mismatched_count,
771 : : batch_sub_insuffperm_count, batch_pub_insuffperm_count, batch_missing_count, batch_skipped_count);
772 : :
773 : : /* Commit this batch, and prepare for next batch */
774 : 16 : CommitTransactionCommand();
775 : :
776 [ + + ]: 16 : if (batch_missing_count)
777 : : {
778 [ + + ]: 13 : for (int idx = cur_batch_base_index; idx < cur_batch_base_index + batch_size; idx++)
779 : : {
780 : : LogicalRepSequenceInfo *seqinfo =
781 : 10 : (LogicalRepSequenceInfo *) list_nth(seqinfos, idx);
782 : :
783 : : /* If the sequence was not found on publisher, record it */
784 [ + + ]: 10 : if (!seqinfo->found_on_pub)
785 : 3 : missing_seqs_idx = lappend_int(missing_seqs_idx, idx);
786 : : }
787 : : }
788 : :
789 : : /*
790 : : * cur_batch_base_index is not incremented sequentially because some
791 : : * sequences may be missing, and the number of fetched rows may not
792 : : * match the batch size.
793 : : */
794 : 16 : cur_batch_base_index += batch_size;
795 : : }
796 : :
797 : : /* Report mismatches, permission issues, or missing sequences */
798 : 16 : report_sequence_errors(mismatched_seqs_idx, sub_insuffperm_seqs_idx,
799 : : pub_insuffperm_seqs_idx, missing_seqs_idx);
800 : 8 : }
801 : :
802 : : /*
803 : : * Identifies sequences that require synchronization and initiates the
804 : : * synchronization process.
805 : : */
806 : : static void
807 : 17 : LogicalRepSyncSequences(void)
808 : : {
809 : : char *err;
810 : : bool must_use_password;
811 : : Relation rel;
812 : : HeapTuple tup;
813 : : ScanKeyData skey[2];
814 : : SysScanDesc scan;
815 : 17 : Oid subid = MyLogicalRepWorker->subid;
816 : : StringInfoData app_name;
817 : :
818 : 17 : StartTransactionCommand();
819 : 17 : maybe_reread_subscription();
820 : :
821 : 17 : rel = table_open(SubscriptionRelRelationId, AccessShareLock);
822 : :
823 : 17 : ScanKeyInit(&skey[0],
824 : : Anum_pg_subscription_rel_srsubid,
825 : : BTEqualStrategyNumber, F_OIDEQ,
826 : : ObjectIdGetDatum(subid));
827 : :
828 : 17 : ScanKeyInit(&skey[1],
829 : : Anum_pg_subscription_rel_srsubstate,
830 : : BTEqualStrategyNumber, F_CHAREQ,
831 : : CharGetDatum(SUBREL_STATE_INIT));
832 : :
833 : 17 : scan = systable_beginscan(rel, InvalidOid, false,
834 : : NULL, 2, skey);
835 [ + + ]: 57 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
836 : : {
837 : : Form_pg_subscription_rel subrel;
838 : : LogicalRepSequenceInfo *seq;
839 : : Relation sequence_rel;
840 : : MemoryContext oldctx;
841 : :
842 [ - + ]: 40 : CHECK_FOR_INTERRUPTS();
843 : :
844 : 40 : subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
845 : :
846 : : /*
847 : : * Lock the sequence so its identity (namespace and name) cannot
848 : : * change under us via a concurrent DROP, RENAME or SET SCHEMA. The
849 : : * lock is released immediately rather than at the transaction end.
850 : : * The later synchronization does not depend on this captured identity
851 : : * remaining valid, as it re-opens the sequence and tolerates
852 : : * concurrent changes. Releasing early also avoids holding one lock
853 : : * per sequence, which could exhaust the lock table.
854 : : */
855 : 40 : sequence_rel = try_table_open(subrel->srrelid, AccessShareLock);
856 : :
857 : : /* Skip if sequence was dropped concurrently */
858 [ - + ]: 40 : if (!sequence_rel)
859 : 0 : continue;
860 : :
861 : : /* Skip if the relation is not a sequence */
862 [ + + ]: 40 : if (sequence_rel->rd_rel->relkind != RELKIND_SEQUENCE)
863 : : {
864 : 2 : table_close(sequence_rel, AccessShareLock);
865 : 2 : continue;
866 : : }
867 : :
868 : : /*
869 : : * Worker needs to process sequences across transaction boundary, so
870 : : * allocate them under long-lived context.
871 : : */
872 : 38 : oldctx = MemoryContextSwitchTo(ApplyContext);
873 : :
874 : 38 : seq = palloc0_object(LogicalRepSequenceInfo);
875 : 38 : seq->localrelid = subrel->srrelid;
876 : 38 : seq->nspname = get_namespace_name(RelationGetNamespace(sequence_rel));
877 : 38 : seq->seqname = pstrdup(RelationGetRelationName(sequence_rel));
878 : 38 : seqinfos = lappend(seqinfos, seq);
879 : :
880 : 38 : MemoryContextSwitchTo(oldctx);
881 : :
882 : 38 : table_close(sequence_rel, AccessShareLock);
883 : : }
884 : :
885 : : /* Cleanup */
886 : 17 : systable_endscan(scan);
887 : 17 : table_close(rel, AccessShareLock);
888 : :
889 : 17 : CommitTransactionCommand();
890 : :
891 : : /*
892 : : * Exit early if no catalog entries found, likely due to concurrent drops.
893 : : */
894 [ - + ]: 17 : if (!seqinfos)
895 : 0 : return;
896 : :
897 : : /* Is the use of a password mandatory? */
898 [ + + ]: 33 : must_use_password = MySubscription->passwordrequired &&
899 [ - + ]: 16 : !MySubscription->ownersuperuser;
900 : :
901 : 17 : initStringInfo(&app_name);
902 : 17 : appendStringInfo(&app_name, "pg_%u_sequence_sync_" UINT64_FORMAT,
903 : 17 : MySubscription->oid, GetSystemIdentifier());
904 : :
905 : : /*
906 : : * Establish the connection to the publisher for sequence synchronization.
907 : : */
908 : 17 : LogRepWorkerWalRcvConn =
909 : 17 : walrcv_connect(MySubscriptionConninfo, true, true,
910 : : must_use_password,
911 : : app_name.data, &err);
912 [ - + ]: 17 : if (LogRepWorkerWalRcvConn == NULL)
913 [ # # ]: 0 : ereport(ERROR,
914 : : errcode(ERRCODE_CONNECTION_FAILURE),
915 : : errmsg("sequencesync worker for subscription \"%s\" could not connect to the publisher: %s",
916 : : MySubscription->name, err));
917 : :
918 : 17 : pfree(app_name.data);
919 : :
920 : 17 : copy_sequences(LogRepWorkerWalRcvConn);
921 : : }
922 : :
923 : : /*
924 : : * Execute the initial sync with error handling. Disable the subscription,
925 : : * if required.
926 : : *
927 : : * Note that we don't handle FATAL errors which are probably because of system
928 : : * resource error and are not repeatable.
929 : : */
930 : : static void
931 : 17 : start_sequence_sync(void)
932 : : {
933 : : Assert(am_sequencesync_worker());
934 : :
935 [ + + ]: 17 : PG_TRY();
936 : : {
937 : : /* Call initial sync. */
938 : 17 : LogicalRepSyncSequences();
939 : : }
940 : 9 : PG_CATCH();
941 : : {
942 [ - + ]: 9 : if (MySubscription->disableonerr)
943 : 0 : DisableSubscriptionAndExit();
944 : : else
945 : : {
946 : : /*
947 : : * Report the worker failed during sequence synchronization. Abort
948 : : * the current transaction so that the stats message is sent in an
949 : : * idle state.
950 : : */
951 : 9 : AbortOutOfAnyTransaction();
952 : 9 : pgstat_report_subscription_error(MySubscription->oid);
953 : :
954 : 9 : PG_RE_THROW();
955 : : }
956 : : }
957 [ - + ]: 8 : PG_END_TRY();
958 : 8 : }
959 : :
960 : : /* Logical Replication sequencesync worker entry point */
961 : : void
962 : 17 : SequenceSyncWorkerMain(Datum main_arg)
963 : : {
964 : 17 : int worker_slot = DatumGetInt32(main_arg);
965 : :
966 : 17 : SetupApplyOrSyncWorker(worker_slot);
967 : :
968 : 17 : start_sequence_sync();
969 : :
970 : 8 : FinishSyncWorker();
971 : : }
|
