Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : * logicalctl.c
3 : : * Functionality to control logical decoding status online.
4 : : *
5 : : * This module enables dynamic control of logical decoding availability.
6 : : * Logical decoding becomes active under two conditions: when the wal_level
7 : : * parameter is set to 'logical', or when at least one valid logical replication
8 : : * slot exists with wal_level set to 'replica'. The system disables logical
9 : : * decoding when neither condition is met. Therefore, the dynamic control
10 : : * of logical decoding availability is required only when wal_level is set
11 : : * to 'replica'. Logical decoding is always enabled when wal_level='logical'
12 : : * and always disabled when wal_level='minimal'.
13 : : *
14 : : * The core concept of dynamically enabling and disabling logical decoding
15 : : * is to separately control two aspects: writing information required for
16 : : * logical decoding to WAL records, and using logical decoding itself. During
17 : : * activation, we first enable logical WAL writing while keeping logical
18 : : * decoding disabled. This change is reflected in the read-only
19 : : * effective_wal_level GUC parameter. Once we ensure that all processes have
20 : : * updated to the latest effective_wal_level value, we then enable logical
21 : : * decoding. Deactivation follows a similar careful, multi-step process
22 : : * in reverse order.
23 : : *
24 : : * While activation occurs synchronously right after creating the first
25 : : * logical slot, deactivation happens asynchronously through the checkpointer
26 : : * process. This design avoids a race condition at the end of recovery; see
27 : : * the comments in UpdateLogicalDecodingStatusEndOfRecovery() for details.
28 : : * Asynchronous deactivation also avoids excessive toggling of the logical
29 : : * decoding status in workloads that repeatedly create and drop a single
30 : : * logical slot. On the other hand, this lazy approach can delay changes
31 : : * to effective_wal_level and the disabling of logical decoding, especially
32 : : * when the checkpointer is busy with other tasks. We chose this lazy approach
33 : : * in all deactivation paths to keep the implementation simple, even though
34 : : * laziness is strictly required only for end-of-recovery cases. Future work
35 : : * might address this limitation either by using a dedicated worker instead
36 : : * of the checkpointer, or by implementing synchronous waiting during slot
37 : : * drops if workloads are significantly affected by the lazy deactivation
38 : : * of logical decoding.
39 : : *
40 : : * Standby servers use the primary server's effective_wal_level and logical
41 : : * decoding status. Unlike normal activation and deactivation, these
42 : : * are updated simultaneously without status change coordination, solely by
43 : : * replaying XLOG_LOGICAL_DECODING_STATUS_CHANGE records. The local wal_level
44 : : * setting has no effect during this time. Upon promotion, we update the
45 : : * logical decoding status based on local conditions: the wal_level value and
46 : : * the presence of logical slots.
47 : : *
48 : : * In the future, we could extend support to include automatic transitions
49 : : * of effective_wal_level between 'minimal' and 'logical' WAL levels. However,
50 : : * this enhancement would require additional coordination mechanisms and
51 : : * careful implementation of operations such as terminating walsenders and
52 : : * archiver processes while carefully considering the sequence of operations
53 : : * to ensure system stability during these transitions.
54 : : *
55 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
56 : : * Portions Copyright (c) 1994, Regents of the University of California
57 : : *
58 : : * IDENTIFICATION
59 : : * src/backend/replication/logical/logicalctl.c
60 : : *
61 : : *-------------------------------------------------------------------------
62 : : */
63 : :
64 : : #include "postgres.h"
65 : :
66 : : #include "access/xloginsert.h"
67 : : #include "access/xlogrecovery.h"
68 : : #include "catalog/pg_control.h"
69 : : #include "miscadmin.h"
70 : : #include "replication/slot.h"
71 : : #include "storage/ipc.h"
72 : : #include "storage/lmgr.h"
73 : : #include "storage/proc.h"
74 : : #include "storage/procarray.h"
75 : : #include "storage/procsignal.h"
76 : : #include "storage/subsystems.h"
77 : : #include "utils/injection_point.h"
78 : :
79 : : /*
80 : : * Struct for controlling the logical decoding status.
81 : : *
82 : : * This struct is protected by LogicalDecodingControlLock.
83 : : */
84 : : typedef struct LogicalDecodingCtlData
85 : : {
86 : : /*
87 : : * This is the authoritative value used by all processes to determine
88 : : * whether to write additional information required by logical decoding to
89 : : * WAL. Since this information could be checked frequently, each process
90 : : * caches this value in XLogLogicalInfo for better performance.
91 : : */
92 : : bool xlog_logical_info;
93 : :
94 : : /* True if logical decoding is available in the system */
95 : : bool logical_decoding_enabled;
96 : :
97 : : /* True if logical decoding might need to be disabled */
98 : : bool pending_disable;
99 : :
100 : : /*
101 : : * End LSN of the last XLOG_LOGICAL_DECODING_STATUS_CHANGE record that
102 : : * enabled logical decoding, or InvalidXLogRecPtr if none has been
103 : : * replayed since the server started. Checking it with
104 : : * logical_decoding_enabled tells from which point on WAL was written with
105 : : * logical decoding enabled; see StandbyLogicalDecodingEnabledSince().
106 : : *
107 : : * WAL records that disable logical decoding are deliberately not tracked
108 : : * here. logical_decoding_enabled is false while decoding is off, which is
109 : : * all the check needs.
110 : : *
111 : : * This is maintained only during recovery and is not persisted, so it
112 : : * says nothing about a status change replayed in an earlier server run.
113 : : */
114 : : XLogRecPtr last_replayed_enable_lsn;
115 : : } LogicalDecodingCtlData;
116 : :
117 : : static LogicalDecodingCtlData *LogicalDecodingCtl = NULL;
118 : :
119 : : static void LogicalDecodingCtlShmemRequest(void *arg);
120 : : static void LogicalDecodingCtlShmemInit(void *arg);
121 : :
122 : : const ShmemCallbacks LogicalDecodingCtlShmemCallbacks = {
123 : : .request_fn = LogicalDecodingCtlShmemRequest,
124 : : .init_fn = LogicalDecodingCtlShmemInit,
125 : : };
126 : :
127 : : /*
128 : : * A process-local cache of LogicalDecodingCtl->xlog_logical_info. This is
129 : : * initialized at process startup, and updated when processing the process
130 : : * barrier signal in ProcessBarrierUpdateXLogLogicalInfo(). If the process
131 : : * is in an XID-assigned transaction, the cache update is delayed until the
132 : : * transaction ends. See the comments for XLogLogicalInfoUpdatePending for details.
133 : : */
134 : : bool XLogLogicalInfo = false;
135 : :
136 : : /*
137 : : * When receiving the PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO signal, if
138 : : * an XID is assigned to the current transaction, the process sets this flag and
139 : : * delays the XLogLogicalInfo update until the transaction ends. This ensures
140 : : * that the XLogLogicalInfo value (typically accessed via XLogLogicalInfoActive)
141 : : * remains consistent throughout the transaction.
142 : : */
143 : : static bool XLogLogicalInfoUpdatePending = false;
144 : :
145 : : static void update_xlog_logical_info(void);
146 : : static void abort_logical_decoding_activation(int code, Datum arg);
147 : : static void write_logical_decoding_status_update_record(bool status);
148 : :
149 : : static void
189 heikki.linnakangas@i 150 :CBC 1287 : LogicalDecodingCtlShmemRequest(void *arg)
151 : : {
152 : 1287 : ShmemRequestStruct(.name = "Logical decoding control",
153 : : .size = sizeof(LogicalDecodingCtlData),
154 : : .ptr = (void **) &LogicalDecodingCtl,
155 : : );
293 msawada@postgresql.o 156 : 1287 : }
157 : :
158 : : static void
5 159 : 1283 : LogicalDecodingCtlShmemInit(void *arg)
160 : : {
161 : 1283 : LogicalDecodingCtl->xlog_logical_info = false;
162 : 1283 : LogicalDecodingCtl->logical_decoding_enabled = false;
163 : 1283 : LogicalDecodingCtl->pending_disable = false;
164 : 1283 : LogicalDecodingCtl->last_replayed_enable_lsn = InvalidXLogRecPtr;
165 : 1283 : }
166 : :
167 : : /*
168 : : * Initialize the logical decoding status in shmem at server startup. This
169 : : * must be called ONCE during postmaster or standalone-backend startup.
170 : : */
171 : : void
293 172 : 1107 : StartupLogicalDecodingStatus(bool last_status)
173 : : {
174 : : /* Logical decoding is always disabled when 'minimal' WAL level */
175 [ + + ]: 1107 : if (wal_level == WAL_LEVEL_MINIMAL)
176 : 381 : return;
177 : :
178 : : /*
179 : : * Set the initial logical decoding status based on the last status. If
180 : : * logical decoding was enabled before the last shutdown, it remains
181 : : * enabled as we might have set wal_level='logical' or have at least one
182 : : * logical slot.
183 : : */
184 : 726 : LogicalDecodingCtl->xlog_logical_info = last_status;
185 : 726 : LogicalDecodingCtl->logical_decoding_enabled = last_status;
186 : : }
187 : :
188 : : /*
189 : : * Update the XLogLogicalInfo cache.
190 : : */
191 : : static inline void
192 : 27217 : update_xlog_logical_info(void)
193 : : {
194 : 27217 : XLogLogicalInfo = IsXLogLogicalInfoEnabled();
195 : 27217 : }
196 : :
197 : : /*
198 : : * Initialize XLogLogicalInfo backend-private cache. This routine is called
199 : : * during process initialization.
200 : : */
201 : : void
202 : 23913 : InitializeProcessXLogLogicalInfo(void)
203 : : {
204 : 23913 : update_xlog_logical_info();
205 : 23913 : }
206 : :
207 : : /*
208 : : * This routine is called when we are told to update XLogLogicalInfo
209 : : * by a ProcSignalBarrier.
210 : : */
211 : : bool
212 : 3304 : ProcessBarrierUpdateXLogLogicalInfo(void)
213 : : {
214 [ + + ]: 3304 : if (GetTopTransactionIdIfAny() != InvalidTransactionId)
215 : : {
216 : : /* Delay updating XLogLogicalInfo until the transaction end */
217 : 13 : XLogLogicalInfoUpdatePending = true;
218 : : }
219 : : else
220 : 3291 : update_xlog_logical_info();
221 : :
222 : 3304 : return true;
223 : : }
224 : :
225 : : /*
226 : : * Check the shared memory state and return true if logical decoding is
227 : : * enabled on the system.
228 : : */
229 : : bool
230 : 20841 : IsLogicalDecodingEnabled(void)
231 : : {
232 : : bool enabled;
233 : :
234 : 20841 : LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
235 : 20841 : enabled = LogicalDecodingCtl->logical_decoding_enabled;
236 : 20841 : LWLockRelease(LogicalDecodingControlLock);
237 : :
238 : 20841 : return enabled;
239 : : }
240 : :
241 : : /*
242 : : * Return true if logical decoding has been enabled continuously from the given
243 : : * LSN up to the current replay position, that is, if the WAL in that range
244 : : * was written with logical decoding enabled.
245 : : *
246 : : * The given LSN must have been replayed already; nothing can be said about
247 : : * WAL this server has not replayed yet. The caller is responsible for checking
248 : : * that.
249 : : */
250 : : bool
5 251 : 12 : StandbyLogicalDecodingEnabledSince(XLogRecPtr lsn)
252 : : {
253 : : bool result;
254 : :
255 [ - + ]: 12 : Assert(RecoveryInProgress());
256 [ - + ]: 12 : Assert(lsn <= GetXLogReplayRecPtr(NULL));
257 : :
258 : 12 : LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
259 [ + + ]: 23 : result = LogicalDecodingCtl->logical_decoding_enabled &&
260 [ + + ]: 11 : lsn >= LogicalDecodingCtl->last_replayed_enable_lsn;
261 : 12 : LWLockRelease(LogicalDecodingControlLock);
262 : :
263 : 12 : return result;
264 : : }
265 : :
266 : : /*
267 : : * Returns true if logical WAL logging is enabled based on the shared memory
268 : : * status.
269 : : */
270 : : bool
293 271 : 27260 : IsXLogLogicalInfoEnabled(void)
272 : : {
273 : : bool xlog_logical_info;
274 : :
275 : 27260 : LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
276 : 27260 : xlog_logical_info = LogicalDecodingCtl->xlog_logical_info;
277 : 27260 : LWLockRelease(LogicalDecodingControlLock);
278 : :
279 : 27260 : return xlog_logical_info;
280 : : }
281 : :
282 : : /*
283 : : * Apply a pending XLogLogicalInfo update at end of the top-level transaction.
284 : : *
285 : : * This is called from CommitTransaction(), PrepareTransaction(), and
286 : : * CleanupTransaction(), which are the only places where the top-level XID is
287 : : * reset, so the next transaction always starts with the latest value.
288 : : */
289 : : void
290 : 427040 : AtEOXact_LogicalCtl(void)
291 : : {
292 : : /* Update the local cache if there is a pending update */
293 [ + + ]: 427040 : if (XLogLogicalInfoUpdatePending)
294 : : {
295 : 13 : update_xlog_logical_info();
296 : 13 : XLogLogicalInfoUpdatePending = false;
297 : : }
298 : 427040 : }
299 : :
300 : : /*
301 : : * Writes an XLOG_LOGICAL_DECODING_STATUS_CHANGE WAL record with the given
302 : : * status.
303 : : */
304 : : static void
305 : 122 : write_logical_decoding_status_update_record(bool status)
306 : : {
307 : : XLogRecPtr recptr;
308 : :
309 : 122 : XLogBeginInsert();
310 : 122 : XLogRegisterData(&status, sizeof(bool));
311 : 122 : recptr = XLogInsert(RM_XLOG_ID, XLOG_LOGICAL_DECODING_STATUS_CHANGE);
312 : 122 : XLogFlush(recptr);
313 : 122 : }
314 : :
315 : : /*
316 : : * A PG_ENSURE_ERROR_CLEANUP callback for activating logical decoding.
317 : : *
318 : : * Rather than directly reverting xlog_logical_info here, we request
319 : : * that the checkpointer handle it via the normal disable path. This
320 : : * avoids race conditions when multiple backends attempt concurrent
321 : : * activation: the checkpointer will reset xlog_logical_info when
322 : : * no valid logical slots exist.
323 : : */
324 : : static void
325 : 2 : abort_logical_decoding_activation(int code, Datum arg)
326 : : {
327 [ + - ]: 2 : elog(DEBUG1, "aborting logical decoding activation process");
125 328 : 2 : RequestDisableLogicalDecoding();
293 329 : 2 : }
330 : :
331 : : /*
332 : : * Enable logical decoding if disabled.
333 : : *
334 : : * If this function is called during recovery, it just checks that logical
335 : : * decoding is still enabled, since the logical decoding status cannot be
336 : : * changed during this time. The logical decoding status depends on the
337 : : * status on the primary.
338 : : *
339 : : * Note that there is no interlock between logical decoding activation
340 : : * and slot creation. To ensure enabling logical decoding, the caller
341 : : * needs to call this function after creating a logical slot before
342 : : * initializing the logical decoding context.
343 : : */
344 : : void
345 : 545 : EnsureLogicalDecodingEnabled(void)
346 : : {
347 [ - + ]: 545 : Assert(MyReplicationSlot);
348 [ - + ]: 545 : Assert(wal_level >= WAL_LEVEL_REPLICA);
349 : :
350 : : /* Logical decoding is always enabled */
351 [ + + ]: 545 : if (wal_level >= WAL_LEVEL_LOGICAL)
352 : 510 : return;
353 : :
354 [ + + ]: 35 : if (RecoveryInProgress())
355 : : {
356 : : /*
357 : : * The caller has already checked that logical decoding is enabled via
358 : : * CheckLogicalDecodingRequirements(), but the status could have been
359 : : * disabled concurrently before we created our slot: either by
360 : : * replaying an XLOG_LOGICAL_DECODING_STATUS_CHANGE record, or by
361 : : * UpdateLogicalDecodingStatusEndOfRecovery() upon promotion. We
362 : : * cannot enable logical decoding during recovery, so raise an error.
363 : : *
364 : : * Our slot has already been created, so its in_use flag is set and
365 : : * the slot scans performed by a deactivation can see it. It
366 : : * guarantees that this check doesn't miss a concurrent deactivation:
367 : : * UpdateLogicalDecodingStatusEndOfRecovery() won't disable logical
368 : : * decoding since CheckLogicalSlotExists() finds our valid slot, and
369 : : * replaying a status change record after this check invalidates our
370 : : * slot, so this slot creation fails afterwards anyway (by a recovery
371 : : * conflict or the requirement re-check in
372 : : * CreateInitDecodingContext()). Hence, this check only needs to catch
373 : : * deactivations that completed before our slot's in_use flag was set.
374 : : */
74 375 [ + + ]: 4 : if (!IsLogicalDecodingEnabled())
376 [ + - ]: 1 : ereport(ERROR,
377 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
378 : : errmsg("logical decoding on standby requires \"effective_wal_level\" >= \"logical\" on the primary"),
379 : : errdetail("Logical decoding was concurrently disabled during the logical replication slot creation.")));
380 : :
293 381 : 3 : return;
382 : : }
383 : :
384 : : /*
385 : : * Ensure to abort the activation process in cases where there is an
386 : : * interruption during the wait.
387 : : */
388 [ + + ]: 31 : PG_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0);
389 : : {
5 390 : 31 : EnableLogicalDecoding(InvalidXLogRecPtr);
391 : : }
293 392 [ - + ]: 31 : PG_END_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0);
393 : : }
394 : :
395 : : /*
396 : : * A workhorse function to enable logical decoding.
397 : : *
398 : : * lsn is the end LSN of the XLOG_LOGICAL_DECODING_STATUS_CHANGE record
399 : : * being replayed, and is InvalidXLogRecPtr when not called from redo.
400 : : */
401 : : void
5 402 : 48 : EnableLogicalDecoding(XLogRecPtr lsn)
403 : : {
404 : : bool in_recovery;
405 : :
293 406 : 48 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
407 : :
408 : : /*
409 : : * Remember where logical decoding was enabled. This has to happen before
410 : : * the early return below, because replay can reach here with the status
411 : : * already on. For example, CreateCheckPoint() fixes the redo point before
412 : : * it records logicalDecodingEnabled, so a checkpoint can claim logical
413 : : * decoding is enabled while the record that enabled it still follows the
414 : : * redo point.
415 : : */
5 416 [ - + ]: 48 : Assert(RecoveryInProgress() == XLogRecPtrIsValid(lsn));
417 [ + + ]: 48 : if (XLogRecPtrIsValid(lsn))
418 : 17 : LogicalDecodingCtl->last_replayed_enable_lsn = lsn;
419 : :
420 : : /* Return if it is already enabled */
293 421 [ + + ]: 48 : if (LogicalDecodingCtl->logical_decoding_enabled)
422 : : {
423 : 5 : LogicalDecodingCtl->pending_disable = false;
424 : 5 : LWLockRelease(LogicalDecodingControlLock);
425 : 5 : return;
426 : : }
427 : :
428 : : /*
429 : : * Set logical info WAL logging in shmem. All processes starting after
430 : : * this point will include the information required by logical decoding in
431 : : * WAL records.
432 : : */
433 : 43 : LogicalDecodingCtl->xlog_logical_info = true;
434 : :
435 : 43 : LWLockRelease(LogicalDecodingControlLock);
436 : :
437 : : /*
438 : : * Tell all running processes to reflect the xlog_logical_info update, and
439 : : * wait. This ensures that all running processes have enabled logical
440 : : * information WAL logging.
441 : : */
442 : 43 : WaitForProcSignalBarrier(
443 : : EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO));
444 : :
445 : 43 : INJECTION_POINT("logical-decoding-activation", NULL);
446 : :
447 : 41 : in_recovery = RecoveryInProgress();
448 : :
449 : : /*
450 : : * There could be some transactions that might have started with the old
451 : : * status, but we don't need to wait for these transactions to complete as
452 : : * long as they have valid XIDs. These transactions will appear in the
453 : : * xl_running_xacts record and therefore the snapshot builder will not try
454 : : * to decode the transaction during the logical decoding initialization.
455 : : *
456 : : * There is a theoretical case where a transaction decides whether to
457 : : * include logical-info in WAL records before getting an XID. In this
458 : : * case, the transaction won't appear in xl_running_xacts.
459 : : *
460 : : * For operations that do not require an XID assignment, the process
461 : : * starts including logical-info immediately upon receiving the signal
462 : : * (barrier). If such an operation checks the effective_wal_level multiple
463 : : * times within a single execution, the resulting WAL records might be
464 : : * inconsistent (i.e., logical-info is included in some records but not in
465 : : * others). However, this is harmless because logical decoding generally
466 : : * ignores WAL records that are not associated with an assigned XID.
467 : : *
468 : : * One might think we need to wait for all running transactions, including
469 : : * those without XIDs and read-only transactions, to finish before
470 : : * enabling logical decoding. However, such a requirement would force the
471 : : * slot creation to wait for a potentially very long time due to
472 : : * long-running read queries, which is practically unacceptable.
473 : : */
474 : :
125 475 : 41 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
476 : :
477 : : /*
478 : : * Re-check whether logical decoding got enabled while we waited for the
479 : : * barrier above.
480 : : */
77 481 [ + + ]: 41 : if (LogicalDecodingCtl->logical_decoding_enabled)
482 : : {
483 : 1 : LogicalDecodingCtl->pending_disable = false;
484 : 1 : LWLockRelease(LogicalDecodingControlLock);
485 : 1 : return;
486 : : }
487 : :
293 488 : 40 : START_CRIT_SECTION();
489 : :
490 : : /*
491 : : * We enable logical decoding first, followed by writing the WAL record.
492 : : * This sequence ensures logical decoding becomes available on the primary
493 : : * first.
494 : : */
495 : 40 : LogicalDecodingCtl->logical_decoding_enabled = true;
496 : :
497 [ + + ]: 40 : if (!in_recovery)
498 : 23 : write_logical_decoding_status_update_record(true);
499 : :
500 : 40 : LogicalDecodingCtl->pending_disable = false;
501 : :
502 [ - + ]: 40 : END_CRIT_SECTION();
503 : :
125 504 : 40 : LWLockRelease(LogicalDecodingControlLock);
505 : :
506 : : /*
507 : : * We log the activation message after releasing the slot lock. This is
508 : : * safe because the activation is performed while holding a logical slot,
509 : : * meaning, a concurrent deactivation cannot interleave its log message
510 : : * ahead of ours.
511 : : */
293 512 [ + + ]: 40 : if (!in_recovery)
513 [ + - ]: 23 : ereport(LOG,
514 : : errmsg("logical decoding is enabled upon creating a new logical replication slot"));
515 : : }
516 : :
517 : : /*
518 : : * Initiate a request for disabling logical decoding.
519 : : *
520 : : * Note that this function does not verify whether logical slots exist. The
521 : : * checkpointer will verify if logical decoding should actually be disabled.
522 : : *
523 : : * This may be called during recovery, for example when a standby invalidates
524 : : * its last valid logical slot. That is safe because the queued request is only
525 : : * acted upon outside recovery. See the RecoveryInProgress() check in
526 : : * DisableLogicalDecodingIfNecessary().
527 : : */
528 : : void
529 : 472 : RequestDisableLogicalDecoding(void)
530 : : {
531 [ + + ]: 472 : if (wal_level != WAL_LEVEL_REPLICA)
532 : 436 : return;
533 : :
534 : : /*
535 : : * It's possible that we might not actually need to disable logical
536 : : * decoding if someone creates a new logical slot concurrently. We set the
537 : : * flag anyway and the checkpointer will check it and disable logical
538 : : * decoding if necessary.
539 : : */
540 : 36 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
541 : 36 : LogicalDecodingCtl->pending_disable = true;
542 : 36 : LWLockRelease(LogicalDecodingControlLock);
543 : :
544 : 36 : WakeupCheckpointer();
545 : :
546 [ + + ]: 36 : elog(DEBUG1, "requested disabling logical decoding");
547 : : }
548 : :
549 : : /*
550 : : * Disable logical decoding if necessary.
551 : : *
552 : : * This function disables logical decoding upon a request initiated by
553 : : * RequestDisableLogicalDecoding(). Otherwise, it performs no action.
554 : : */
555 : : void
556 : 6267 : DisableLogicalDecodingIfNecessary(void)
557 : : {
558 : : bool pending_disable;
559 : :
560 [ + + ]: 6267 : if (wal_level != WAL_LEVEL_REPLICA)
561 : 1484 : return;
562 : :
563 : : /*
564 : : * Sanity check as we cannot disable logical decoding while holding a
565 : : * logical slot.
566 : : */
567 [ - + ]: 4783 : Assert(!MyReplicationSlot);
568 : :
569 : : /*
570 : : * During recovery the logical decoding status follows the primary via WAL
571 : : * replay, so we must not disable it here. A pending_disable request
572 : : * queued during recovery, for example by a local slot invalidation, is
573 : : * intentionally left for the end-of-recovery transition or the
574 : : * post-promotion checkpointer to act on. See
575 : : * UpdateLogicalDecodingStatusEndOfRecovery().
576 : : */
577 [ + + ]: 4783 : if (RecoveryInProgress())
578 : 2059 : return;
579 : :
580 : 2724 : LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
581 : 2724 : pending_disable = LogicalDecodingCtl->pending_disable;
582 : 2724 : LWLockRelease(LogicalDecodingControlLock);
583 : :
584 : : /* Quick return if no pending disable request */
585 [ + + ]: 2724 : if (!pending_disable)
586 : 2691 : return;
587 : :
588 : 33 : DisableLogicalDecoding();
589 : : }
590 : :
591 : : /*
592 : : * A workhorse function to disable logical decoding.
593 : : */
594 : : void
595 : 50 : DisableLogicalDecoding(void)
596 : : {
597 : 50 : bool in_recovery = RecoveryInProgress();
598 : : bool was_enabled;
599 : :
600 : 50 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
601 : :
602 : : /*
603 : : * Check if we can disable logical decoding.
604 : : *
605 : : * Nothing to do if both flags are already off, or if valid slots exist
606 : : * (skip the slot check during recovery because the existing slots will be
607 : : * invalidated after disabling logical decoding.)
608 : : */
125 609 [ + + ]: 50 : if ((!LogicalDecodingCtl->logical_decoding_enabled &&
610 [ + + ]: 3 : !LogicalDecodingCtl->xlog_logical_info) ||
293 611 [ + + + + ]: 48 : (!in_recovery && CheckLogicalSlotExists()))
612 : : {
613 : 9 : LogicalDecodingCtl->pending_disable = false;
614 : 9 : LWLockRelease(LogicalDecodingControlLock);
615 : 9 : return;
616 : : }
617 : :
618 : : /*
619 : : * Remember if logical decoding was enabled. An interrupted activation can
620 : : * leave xlog_logical_info=true while logical_decoding_enabled remains
621 : : * false.
622 : : */
125 623 : 41 : was_enabled = LogicalDecodingCtl->logical_decoding_enabled;
624 : :
293 625 : 41 : START_CRIT_SECTION();
626 : :
627 : : /*
628 : : * We need to disable logical decoding first and then disable logical
629 : : * information WAL logging in order to ensure that no logical decoding
630 : : * processes WAL records with insufficient information.
631 : : */
632 : 41 : LogicalDecodingCtl->logical_decoding_enabled = false;
633 : :
634 : : /* Write the WAL to disable logical decoding on standbys too */
125 635 [ + + + + ]: 41 : if (!in_recovery && was_enabled)
293 636 : 23 : write_logical_decoding_status_update_record(false);
637 : :
638 : : /* Now disable logical information WAL logging */
639 : 41 : LogicalDecodingCtl->xlog_logical_info = false;
640 : 41 : LogicalDecodingCtl->pending_disable = false;
641 : :
642 [ - + ]: 41 : END_CRIT_SECTION();
643 : :
644 : : /*
645 : : * Logging under the lock guarantees our "is disabled" message appears in
646 : : * the server log before its eventual "is enabled", making server log
647 : : * diagnostics easy.
648 : : */
125 649 [ + + + + ]: 41 : if (!in_recovery && was_enabled)
293 650 [ + - ]: 23 : ereport(LOG,
651 : : errmsg("logical decoding is disabled because there are no valid logical replication slots"));
652 : :
653 : 41 : LWLockRelease(LogicalDecodingControlLock);
654 : :
655 : : /*
656 : : * Tell all running processes to reflect the xlog_logical_info update.
657 : : * Unlike when enabling logical decoding, we don't need to wait for all
658 : : * processes to complete it in this case. We already disabled logical
659 : : * decoding and it's always safe to write logical information to WAL
660 : : * records, even when not strictly required. Therefore, we don't need to
661 : : * wait for all running transactions to finish either.
662 : : */
663 : 41 : EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO);
664 : : }
665 : :
666 : : /*
667 : : * Updates the logical decoding status at end of recovery, and ensures that
668 : : * all running processes have the updated XLogLogicalInfo status. This
669 : : * function must be called before accepting writes.
670 : : */
671 : : void
672 : 1034 : UpdateLogicalDecodingStatusEndOfRecovery(void)
673 : : {
674 : 1034 : bool new_status = false;
675 : :
676 [ - + ]: 1034 : Assert(RecoveryInProgress());
677 : :
678 : 1034 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
679 : :
680 : : /*
681 : : * With 'minimal' WAL level, no logical replication slot can exist (see
682 : : * RestoreSlotFromDisk()), so the new status is always false. However,
683 : : * logical decoding could have been enabled during recovery by replaying
684 : : * an XLOG_LOGICAL_DECODING_STATUS_CHANGE record from WAL generated with a
685 : : * higher wal_level, e.g. if the server crashed right after the last
686 : : * logical slot was dropped and then restarted with wal_level='minimal'.
687 : : * The code below disables logical decoding in that case.
688 : : */
689 [ + + + + ]: 1034 : if (wal_level == WAL_LEVEL_LOGICAL || CheckLogicalSlotExists())
690 : 141 : new_status = true;
691 : :
692 : : /*
693 : : * When recovery ends, we need to either enable or disable logical
694 : : * decoding based on the wal_level setting and the presence of logical
695 : : * slots. We need to note that concurrent slot creation and deletion could
696 : : * happen but WAL writes are still not permitted until recovery fully
697 : : * completes. Here's how we handle concurrent toggling of logical
698 : : * decoding:
699 : : *
700 : : * For 'enable' case, if there's a concurrent disable request before
701 : : * recovery fully completes, the checkpointer will handle it after
702 : : * recovery is done. This means there might be a brief period after
703 : : * recovery where logical decoding remains enabled even with no logical
704 : : * replication slots present. This temporary state is not new - it can
705 : : * already occur due to the checkpointer's asynchronous deactivation
706 : : * process.
707 : : *
708 : : * For 'disable' case, a backend concurrently creating a logical slot on a
709 : : * standby could have passed its CheckLogicalDecodingRequirements() check
710 : : * when creating its slot only after our slot check above. Such a backend
711 : : * rechecks the status after creating the slot in
712 : : * EnsureLogicalDecodingEnabled() and raises an error if logical decoding
713 : : * has been disabled meanwhile, so it cannot end up with a logical slot
714 : : * while logical decoding remains disabled. (If recovery has already ended
715 : : * by the time of the recheck, the backend instead enables logical
716 : : * decoding by itself, which is fine after promotion.)
717 : : */
718 [ + + ]: 1034 : if (new_status != LogicalDecodingCtl->logical_decoding_enabled)
719 : : {
720 : : /*
721 : : * Update both the logical decoding status and logical WAL logging
722 : : * status. Unlike toggling these status during non-recovery, we don't
723 : : * need to worry about the operation order as WAL writes are still not
724 : : * permitted.
725 : : */
726 : 76 : LogicalDecodingCtl->xlog_logical_info = new_status;
727 : 76 : LogicalDecodingCtl->logical_decoding_enabled = new_status;
728 : :
729 [ + + ]: 76 : elog(DEBUG1,
730 : : "update logical decoding status to %d at the end of recovery",
731 : : new_status);
732 : :
733 : : /*
734 : : * Now that we updated the logical decoding status, clear the pending
735 : : * disable flag. It's possible that a concurrent process drops the
736 : : * last logical slot and initiates the pending disable again. The
737 : : * checkpointer process will check it.
738 : : */
739 : 76 : LogicalDecodingCtl->pending_disable = false;
740 : :
741 : 76 : LWLockRelease(LogicalDecodingControlLock);
742 : :
743 : 76 : write_logical_decoding_status_update_record(new_status);
744 : : }
745 : : else
746 : 958 : LWLockRelease(LogicalDecodingControlLock);
747 : :
748 : : /*
749 : : * Ensure all running processes have the updated status. We don't need to
750 : : * wait for running transactions to finish as we don't accept any writes
751 : : * yet. On the other hand, we need to wait for synchronizing
752 : : * XLogLogicalInfo even if we've not updated the status above as the
753 : : * status have been turned on and off during recovery, having running
754 : : * processes have different status on their local caches.
755 : : */
756 [ + + ]: 1034 : if (IsUnderPostmaster)
757 : 897 : WaitForProcSignalBarrier(
758 : : EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO));
759 : :
760 : 1034 : INJECTION_POINT("startup-logical-decoding-status-change-end-of-recovery", NULL);
761 : 1034 : }
|
