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