Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : * syncutils.c
3 : : * PostgreSQL logical replication: common synchronization code
4 : : *
5 : : * Copyright (c) 2025-2026, PostgreSQL Global Development Group
6 : : *
7 : : * IDENTIFICATION
8 : : * src/backend/replication/logical/syncutils.c
9 : : *
10 : : * NOTES
11 : : * This file contains code common for synchronization workers.
12 : : *-------------------------------------------------------------------------
13 : : */
14 : :
15 : : #include "postgres.h"
16 : :
17 : : #include "catalog/pg_subscription_rel.h"
18 : : #include "pgstat.h"
19 : : #include "replication/logicallauncher.h"
20 : : #include "replication/worker_internal.h"
21 : : #include "storage/ipc.h"
22 : : #include "utils/lsyscache.h"
23 : : #include "utils/memutils.h"
24 : :
25 : : /*
26 : : * Enum for phases of the subscription relations state.
27 : : *
28 : : * SYNC_RELATIONS_STATE_NEEDS_REBUILD indicates that the subscription relations
29 : : * state is no longer valid, and the subscription relations should be rebuilt.
30 : : *
31 : : * SYNC_RELATIONS_STATE_REBUILD_STARTED indicates that the subscription
32 : : * relations state is being rebuilt.
33 : : *
34 : : * SYNC_RELATIONS_STATE_VALID indicates that the subscription relation state is
35 : : * up-to-date and valid.
36 : : */
37 : : typedef enum
38 : : {
39 : : SYNC_RELATIONS_STATE_NEEDS_REBUILD,
40 : : SYNC_RELATIONS_STATE_REBUILD_STARTED,
41 : : SYNC_RELATIONS_STATE_VALID,
42 : : } SyncingRelationsState;
43 : :
44 : : static SyncingRelationsState relation_states_validity = SYNC_RELATIONS_STATE_NEEDS_REBUILD;
45 : :
46 : : /*
47 : : * Exit routine for synchronization worker.
48 : : */
49 : : pg_noreturn void
50 : 207 : FinishSyncWorker(void)
51 : : {
52 : : Assert(am_sequencesync_worker() || am_tablesync_worker());
53 : :
54 : : /*
55 : : * Commit any outstanding transaction. This is the usual case, unless
56 : : * there was nothing to do for the table.
57 : : */
58 [ + + ]: 207 : if (IsTransactionState())
59 : : {
60 : 200 : CommitTransactionCommand();
61 : 200 : pgstat_report_stat(true);
62 : : }
63 : :
64 : : /* And flush all writes. */
65 : 207 : XLogFlush(GetXLogWriteRecPtr());
66 : :
67 [ + + ]: 207 : if (am_sequencesync_worker())
68 : : {
69 [ + - ]: 7 : ereport(LOG,
70 : : errmsg("logical replication sequence synchronization worker for subscription \"%s\" has finished",
71 : : MySubscription->name));
72 : :
73 : : /*
74 : : * Reset last_seqsync_start_time, so that next time a sequencesync
75 : : * worker is needed it can be started promptly.
76 : : */
77 : 7 : logicalrep_reset_seqsync_start_time();
78 : : }
79 : : else
80 : : {
81 : 200 : StartTransactionCommand();
82 [ + - ]: 200 : ereport(LOG,
83 : : errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
84 : : MySubscription->name,
85 : : get_rel_name(MyLogicalRepWorker->relid)));
86 : 200 : CommitTransactionCommand();
87 : :
88 : : /* Find the leader apply worker and signal it. */
89 : 200 : logicalrep_worker_wakeup(WORKERTYPE_APPLY, MyLogicalRepWorker->subid,
90 : : InvalidOid);
91 : : }
92 : :
93 : : /* Stop gracefully */
94 : 207 : proc_exit(0);
95 : : }
96 : :
97 : : /*
98 : : * Callback from syscache invalidation.
99 : : */
100 : : void
101 : 2127 : InvalidateSyncingRelStates(Datum arg, SysCacheIdentifier cacheid,
102 : : uint32 hashvalue)
103 : : {
104 : 2127 : relation_states_validity = SYNC_RELATIONS_STATE_NEEDS_REBUILD;
105 : 2127 : }
106 : :
107 : : /*
108 : : * Attempt to launch a sync worker for one or more sequences or a table, if
109 : : * a worker slot is available and the retry interval has elapsed.
110 : : *
111 : : * wtype: sync worker type.
112 : : * nsyncworkers: Number of currently running sync workers for the subscription.
113 : : * relid: InvalidOid for sequencesync worker, actual relid for tablesync
114 : : * worker.
115 : : * last_start_time: Pointer to the last start time of the worker.
116 : : */
117 : : void
118 : 922 : launch_sync_worker(LogicalRepWorkerType wtype, int nsyncworkers, Oid relid,
119 : : TimestampTz *last_start_time)
120 : : {
121 : : TimestampTz now;
122 : :
123 : : Assert((wtype == WORKERTYPE_TABLESYNC && OidIsValid(relid)) ||
124 : : (wtype == WORKERTYPE_SEQUENCESYNC && !OidIsValid(relid)));
125 : :
126 : : /* If there is a free sync worker slot, start a new sync worker */
127 [ + + ]: 922 : if (nsyncworkers >= max_sync_workers_per_subscription)
128 : 610 : return;
129 : :
130 : 312 : now = GetCurrentTimestamp();
131 : :
132 [ + + + + ]: 411 : if (!(*last_start_time) ||
133 : 99 : TimestampDifferenceExceeds(*last_start_time, now,
134 : : wal_retrieve_retry_interval))
135 : : {
136 : : /*
137 : : * Set the last_start_time even if we fail to start the worker, so
138 : : * that we won't retry until wal_retrieve_retry_interval has elapsed.
139 : : */
140 : 234 : *last_start_time = now;
141 : 234 : (void) logicalrep_worker_launch(wtype,
142 : 234 : MyLogicalRepWorker->dbid,
143 : 234 : MySubscription->oid,
144 : 234 : MySubscription->name,
145 : 234 : MyLogicalRepWorker->userid,
146 : : relid, DSM_HANDLE_INVALID, false);
147 : : }
148 : : }
149 : :
150 : : /*
151 : : * Process possible state change(s) of relations that are being synchronized
152 : : * and start new tablesync workers for the newly added tables. Also, start a
153 : : * new sequencesync worker for the newly added sequences.
154 : : */
155 : : void
156 : 10127 : ProcessSyncingRelations(XLogRecPtr current_lsn)
157 : : {
158 [ + + + - : 10127 : switch (MyLogicalRepWorker->type)
- - ]
159 : : {
160 : 23 : case WORKERTYPE_PARALLEL_APPLY:
161 : :
162 : : /*
163 : : * Skip for parallel apply workers because they only operate on
164 : : * tables that are in a READY state. See pa_can_start() and
165 : : * should_apply_changes_for_rel().
166 : : */
167 : 23 : break;
168 : :
169 : 221 : case WORKERTYPE_TABLESYNC:
170 : 221 : ProcessSyncingTablesForSync(current_lsn);
171 : 21 : break;
172 : :
173 : 9883 : case WORKERTYPE_APPLY:
174 : 9883 : ProcessSyncingTablesForApply(current_lsn);
175 : 9877 : ProcessSequencesForSync();
176 : 9877 : break;
177 : :
178 : 0 : case WORKERTYPE_SEQUENCESYNC:
179 : : /* Should never happen. */
180 [ # # ]: 0 : elog(ERROR, "sequence synchronization worker is not expected to process relations");
181 : : break;
182 : :
183 : 0 : case WORKERTYPE_UNKNOWN:
184 : : /* Should never happen. */
185 [ # # ]: 0 : elog(ERROR, "Unknown worker type");
186 : : }
187 : 9921 : }
188 : :
189 : : /*
190 : : * Common code to fetch the up-to-date sync state info for tables and sequences.
191 : : *
192 : : * The pg_subscription_rel catalog is shared by tables and sequences. Changes
193 : : * to either sequences or tables can affect the validity of relation states, so
194 : : * we identify non-READY tables and non-READY sequences together to ensure
195 : : * consistency.
196 : : *
197 : : * has_pending_subtables: true if the subscription has one or more tables that
198 : : * are not in READY state, otherwise false.
199 : : * has_pending_subsequences: true if the subscription has one or more sequences
200 : : * that are not in READY state, otherwise false.
201 : : */
202 : : void
203 : 20065 : FetchRelationStates(bool *has_pending_subtables,
204 : : bool *has_pending_subsequences,
205 : : bool *started_tx)
206 : : {
207 : : /*
208 : : * has_subtables and has_subsequences_non_ready are declared as static,
209 : : * since the same value can be used until the system table is invalidated.
210 : : */
211 : : static bool has_subtables = false;
212 : : static bool has_subsequences_non_ready = false;
213 : :
214 : 20065 : *started_tx = false;
215 : :
216 [ + + ]: 20065 : if (relation_states_validity != SYNC_RELATIONS_STATE_VALID)
217 : : {
218 : : MemoryContext oldctx;
219 : : List *rstates;
220 : : SubscriptionRelState *rstate;
221 : :
222 : 1053 : relation_states_validity = SYNC_RELATIONS_STATE_REBUILD_STARTED;
223 : 1053 : has_subsequences_non_ready = false;
224 : :
225 : : /* Clean the old lists. */
226 : 1053 : list_free_deep(table_states_not_ready);
227 : 1053 : table_states_not_ready = NIL;
228 : :
229 [ + + ]: 1053 : if (!IsTransactionState())
230 : : {
231 : 1036 : StartTransactionCommand();
232 : 1036 : *started_tx = true;
233 : : }
234 : :
235 : : /* Fetch tables and sequences that are in non-READY state. */
236 : 1053 : rstates = GetSubscriptionRelations(MySubscription->oid, true, true,
237 : : true);
238 : :
239 : : /* Allocate the tracking info in a permanent memory context. */
240 : 1053 : oldctx = MemoryContextSwitchTo(CacheMemoryContext);
241 [ + + + + : 3906 : foreach_ptr(SubscriptionRelState, subrel, rstates)
+ + ]
242 : : {
243 [ + + ]: 1800 : if (get_rel_relkind(subrel->relid) == RELKIND_SEQUENCE)
244 : 33 : has_subsequences_non_ready = true;
245 : : else
246 : : {
247 : 1767 : rstate = palloc_object(SubscriptionRelState);
248 : 1767 : memcpy(rstate, subrel, sizeof(SubscriptionRelState));
249 : 1767 : table_states_not_ready = lappend(table_states_not_ready,
250 : : rstate);
251 : : }
252 : : }
253 : 1053 : MemoryContextSwitchTo(oldctx);
254 : :
255 : : /*
256 : : * Does the subscription have tables?
257 : : *
258 : : * If there were not-READY tables found then we know it does. But if
259 : : * table_states_not_ready was empty we still need to check again to
260 : : * see if there are 0 tables.
261 : : */
262 [ + + + + ]: 1369 : has_subtables = (table_states_not_ready != NIL) ||
263 : 316 : HasSubscriptionTables(MySubscription->oid);
264 : :
265 : : /*
266 : : * If the subscription relation cache has been invalidated since we
267 : : * entered this routine, we still use and return the relations we just
268 : : * finished constructing, to avoid infinite loops, but we leave the
269 : : * table states marked as stale so that we'll rebuild it again on next
270 : : * access. Otherwise, we mark the table states as valid.
271 : : */
272 [ + + ]: 1053 : if (relation_states_validity == SYNC_RELATIONS_STATE_REBUILD_STARTED)
273 : 1050 : relation_states_validity = SYNC_RELATIONS_STATE_VALID;
274 : : }
275 : :
276 [ + + ]: 20065 : if (has_pending_subtables)
277 : 305 : *has_pending_subtables = has_subtables;
278 : :
279 [ + + ]: 20065 : if (has_pending_subsequences)
280 : 9877 : *has_pending_subsequences = has_subsequences_non_ready;
281 : 20065 : }
|
