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 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 "catalog/pg_control.h"
68 : : #include "miscadmin.h"
69 : : #include "replication/slot.h"
70 : : #include "storage/ipc.h"
71 : : #include "storage/lmgr.h"
72 : : #include "storage/proc.h"
73 : : #include "storage/procarray.h"
74 : : #include "storage/procsignal.h"
75 : : #include "storage/subsystems.h"
76 : : #include "utils/injection_point.h"
77 : :
78 : : /*
79 : : * Struct for controlling the logical decoding status.
80 : : *
81 : : * This struct is protected by LogicalDecodingControlLock.
82 : : */
83 : : typedef struct LogicalDecodingCtlData
84 : : {
85 : : /*
86 : : * This is the authoritative value used by all processes to determine
87 : : * whether to write additional information required by logical decoding to
88 : : * WAL. Since this information could be checked frequently, each process
89 : : * caches this value in XLogLogicalInfo for better performance.
90 : : */
91 : : bool xlog_logical_info;
92 : :
93 : : /* True if logical decoding is available in the system */
94 : : bool logical_decoding_enabled;
95 : :
96 : : /* True if logical decoding might need to be disabled */
97 : : bool pending_disable;
98 : : } LogicalDecodingCtlData;
99 : :
100 : : static LogicalDecodingCtlData *LogicalDecodingCtl = NULL;
101 : :
102 : : static void LogicalDecodingCtlShmemRequest(void *arg);
103 : :
104 : : const ShmemCallbacks LogicalDecodingCtlShmemCallbacks = {
105 : : .request_fn = LogicalDecodingCtlShmemRequest,
106 : : };
107 : :
108 : : /*
109 : : * A process-local cache of LogicalDecodingCtl->xlog_logical_info. This is
110 : : * initialized at process startup, and updated when processing the process
111 : : * barrier signal in ProcessBarrierUpdateXLogLogicalInfo(). If the process
112 : : * is in an XID-assigned transaction, the cache update is delayed until the
113 : : * transaction ends. See the comments for XLogLogicalInfoUpdatePending for details.
114 : : */
115 : : bool XLogLogicalInfo = false;
116 : :
117 : : /*
118 : : * When receiving the PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO signal, if
119 : : * an XID is assigned to the current transaction, the process sets this flag and
120 : : * delays the XLogLogicalInfo update until the transaction ends. This ensures
121 : : * that the XLogLogicalInfo value (typically accessed via XLogLogicalInfoActive)
122 : : * remains consistent throughout the transaction.
123 : : */
124 : : static bool XLogLogicalInfoUpdatePending = false;
125 : :
126 : : static void update_xlog_logical_info(void);
127 : : static void abort_logical_decoding_activation(int code, Datum arg);
128 : : static void write_logical_decoding_status_update_record(bool status);
129 : :
130 : : static void
143 heikki.linnakangas@i 131 :CBC 1239 : LogicalDecodingCtlShmemRequest(void *arg)
132 : : {
133 : 1239 : ShmemRequestStruct(.name = "Logical decoding control",
134 : : .size = sizeof(LogicalDecodingCtlData),
135 : : .ptr = (void **) &LogicalDecodingCtl,
136 : : );
247 msawada@postgresql.o 137 : 1239 : }
138 : :
139 : : /*
140 : : * Initialize the logical decoding status in shmem at server startup. This
141 : : * must be called ONCE during postmaster or standalone-backend startup.
142 : : */
143 : : void
144 : 1069 : StartupLogicalDecodingStatus(bool last_status)
145 : : {
146 : : /* Logical decoding is always disabled when 'minimal' WAL level */
147 [ + + ]: 1069 : if (wal_level == WAL_LEVEL_MINIMAL)
148 : 370 : return;
149 : :
150 : : /*
151 : : * Set the initial logical decoding status based on the last status. If
152 : : * logical decoding was enabled before the last shutdown, it remains
153 : : * enabled as we might have set wal_level='logical' or have at least one
154 : : * logical slot.
155 : : */
156 : 699 : LogicalDecodingCtl->xlog_logical_info = last_status;
157 : 699 : LogicalDecodingCtl->logical_decoding_enabled = last_status;
158 : : }
159 : :
160 : : /*
161 : : * Update the XLogLogicalInfo cache.
162 : : */
163 : : static inline void
164 : 26376 : update_xlog_logical_info(void)
165 : : {
166 : 26376 : XLogLogicalInfo = IsXLogLogicalInfoEnabled();
167 : 26376 : }
168 : :
169 : : /*
170 : : * Initialize XLogLogicalInfo backend-private cache. This routine is called
171 : : * during process initialization.
172 : : */
173 : : void
174 : 23298 : InitializeProcessXLogLogicalInfo(void)
175 : : {
176 : 23298 : update_xlog_logical_info();
177 : 23298 : }
178 : :
179 : : /*
180 : : * This routine is called when we are told to update XLogLogicalInfo
181 : : * by a ProcSignalBarrier.
182 : : */
183 : : bool
184 : 3078 : ProcessBarrierUpdateXLogLogicalInfo(void)
185 : : {
186 [ + + ]: 3078 : if (GetTopTransactionIdIfAny() != InvalidTransactionId)
187 : : {
188 : : /* Delay updating XLogLogicalInfo until the transaction end */
189 : 5 : XLogLogicalInfoUpdatePending = true;
190 : : }
191 : : else
192 : 3073 : update_xlog_logical_info();
193 : :
194 : 3078 : return true;
195 : : }
196 : :
197 : : /*
198 : : * Check the shared memory state and return true if logical decoding is
199 : : * enabled on the system.
200 : : */
201 : : bool
202 : 20425 : IsLogicalDecodingEnabled(void)
203 : : {
204 : : bool enabled;
205 : :
206 : 20425 : LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
207 : 20425 : enabled = LogicalDecodingCtl->logical_decoding_enabled;
208 : 20425 : LWLockRelease(LogicalDecodingControlLock);
209 : :
210 : 20425 : return enabled;
211 : : }
212 : :
213 : : /*
214 : : * Returns true if logical WAL logging is enabled based on the shared memory
215 : : * status.
216 : : */
217 : : bool
218 : 26411 : IsXLogLogicalInfoEnabled(void)
219 : : {
220 : : bool xlog_logical_info;
221 : :
222 : 26411 : LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
223 : 26411 : xlog_logical_info = LogicalDecodingCtl->xlog_logical_info;
224 : 26411 : LWLockRelease(LogicalDecodingControlLock);
225 : :
226 : 26411 : return xlog_logical_info;
227 : : }
228 : :
229 : : /*
230 : : * Reset the local cache at end of the transaction.
231 : : */
232 : : void
233 : 431037 : AtEOXact_LogicalCtl(void)
234 : : {
235 : : /* Update the local cache if there is a pending update */
236 [ + + ]: 431037 : if (XLogLogicalInfoUpdatePending)
237 : : {
238 : 5 : update_xlog_logical_info();
239 : 5 : XLogLogicalInfoUpdatePending = false;
240 : : }
241 : 431037 : }
242 : :
243 : : /*
244 : : * Writes an XLOG_LOGICAL_DECODING_STATUS_CHANGE WAL record with the given
245 : : * status.
246 : : */
247 : : static void
248 : 103 : write_logical_decoding_status_update_record(bool status)
249 : : {
250 : : XLogRecPtr recptr;
251 : :
252 : 103 : XLogBeginInsert();
253 : 103 : XLogRegisterData(&status, sizeof(bool));
254 : 103 : recptr = XLogInsert(RM_XLOG_ID, XLOG_LOGICAL_DECODING_STATUS_CHANGE);
255 : 103 : XLogFlush(recptr);
256 : 103 : }
257 : :
258 : : /*
259 : : * A PG_ENSURE_ERROR_CLEANUP callback for activating logical decoding.
260 : : *
261 : : * Rather than directly reverting xlog_logical_info here, we request
262 : : * that the checkpointer handle it via the normal disable path. This
263 : : * avoids race conditions when multiple backends attempt concurrent
264 : : * activation: the checkpointer will reset xlog_logical_info when
265 : : * no valid logical slots exist.
266 : : */
267 : : static void
268 : 2 : abort_logical_decoding_activation(int code, Datum arg)
269 : : {
270 [ + - ]: 2 : elog(DEBUG1, "aborting logical decoding activation process");
79 271 : 2 : RequestDisableLogicalDecoding();
247 272 : 2 : }
273 : :
274 : : /*
275 : : * Enable logical decoding if disabled.
276 : : *
277 : : * If this function is called during recovery, it just checks that logical
278 : : * decoding is still enabled, since the logical decoding status cannot be
279 : : * changed during this time. The logical decoding status depends on the
280 : : * status on the primary.
281 : : *
282 : : * Note that there is no interlock between logical decoding activation
283 : : * and slot creation. To ensure enabling logical decoding, the caller
284 : : * needs to call this function after creating a logical slot before
285 : : * initializing the logical decoding context.
286 : : */
287 : : void
288 : 523 : EnsureLogicalDecodingEnabled(void)
289 : : {
290 [ - + ]: 523 : Assert(MyReplicationSlot);
291 [ - + ]: 523 : Assert(wal_level >= WAL_LEVEL_REPLICA);
292 : :
293 : : /* Logical decoding is always enabled */
294 [ + + ]: 523 : if (wal_level >= WAL_LEVEL_LOGICAL)
295 : 500 : return;
296 : :
297 [ + + ]: 23 : if (RecoveryInProgress())
298 : : {
299 : : /*
300 : : * The caller has already checked that logical decoding is enabled via
301 : : * CheckLogicalDecodingRequirements(), but the status could have been
302 : : * disabled concurrently before we created our slot: either by
303 : : * replaying an XLOG_LOGICAL_DECODING_STATUS_CHANGE record, or by
304 : : * UpdateLogicalDecodingStatusEndOfRecovery() upon promotion. We
305 : : * cannot enable logical decoding during recovery, so raise an error.
306 : : *
307 : : * Our slot has already been created, so its in_use flag is set and
308 : : * the slot scans performed by a deactivation can see it. It
309 : : * guarantees that this check doesn't miss a concurrent deactivation:
310 : : * UpdateLogicalDecodingStatusEndOfRecovery() won't disable logical
311 : : * decoding since CheckLogicalSlotExists() finds our valid slot, and
312 : : * replaying a status change record after this check invalidates our
313 : : * slot, so this slot creation fails afterwards anyway (by a recovery
314 : : * conflict or the requirement re-check in
315 : : * CreateInitDecodingContext()). Hence, this check only needs to catch
316 : : * deactivations that completed before our slot's in_use flag was set.
317 : : */
28 318 [ + + ]: 4 : if (!IsLogicalDecodingEnabled())
319 [ + - ]: 1 : ereport(ERROR,
320 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
321 : : errmsg("logical decoding on standby requires \"effective_wal_level\" >= \"logical\" on the primary"),
322 : : errdetail("Logical decoding was concurrently disabled during the logical replication slot creation.")));
323 : :
247 324 : 3 : return;
325 : : }
326 : :
327 : : /*
328 : : * Ensure to abort the activation process in cases where there in an
329 : : * interruption during the wait.
330 : : */
331 [ + + ]: 19 : PG_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0);
332 : : {
333 : 19 : EnableLogicalDecoding();
334 : : }
335 [ - + ]: 19 : PG_END_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0);
336 : : }
337 : :
338 : : /*
339 : : * A workhorse function to enable logical decoding.
340 : : */
341 : : void
342 : 34 : EnableLogicalDecoding(void)
343 : : {
344 : : bool in_recovery;
345 : :
346 : 34 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
347 : :
348 : : /* Return if it is already enabled */
349 [ + + ]: 34 : if (LogicalDecodingCtl->logical_decoding_enabled)
350 : : {
351 : 2 : LogicalDecodingCtl->pending_disable = false;
352 : 2 : LWLockRelease(LogicalDecodingControlLock);
353 : 2 : return;
354 : : }
355 : :
356 : : /*
357 : : * Set logical info WAL logging in shmem. All process starts after this
358 : : * point will include the information required by logical decoding to WAL
359 : : * records.
360 : : */
361 : 32 : LogicalDecodingCtl->xlog_logical_info = true;
362 : :
363 : 32 : LWLockRelease(LogicalDecodingControlLock);
364 : :
365 : : /*
366 : : * Tell all running processes to reflect the xlog_logical_info update, and
367 : : * wait. This ensures that all running processes have enabled logical
368 : : * information WAL logging.
369 : : */
370 : 32 : WaitForProcSignalBarrier(
371 : : EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO));
372 : :
373 : 32 : INJECTION_POINT("logical-decoding-activation", NULL);
374 : :
375 : 30 : in_recovery = RecoveryInProgress();
376 : :
377 : : /*
378 : : * There could be some transactions that might have started with the old
379 : : * status, but we don't need to wait for these transactions to complete as
380 : : * long as they have valid XIDs. These transactions will appear in the
381 : : * xl_running_xacts record and therefore the snapshot builder will not try
382 : : * to decode the transaction during the logical decoding initialization.
383 : : *
384 : : * There is a theoretical case where a transaction decides whether to
385 : : * include logical-info to WAL records before getting an XID. In this
386 : : * case, the transaction won't appear in xl_running_xacts.
387 : : *
388 : : * For operations that do not require an XID assignment, the process
389 : : * starts including logical-info immediately upon receiving the signal
390 : : * (barrier). If such an operation checks the effective_wal_level multiple
391 : : * times within a single execution, the resulting WAL records might be
392 : : * inconsistent (i.e., logical-info is included in some records but not in
393 : : * others). However, this is harmless because logical decoding generally
394 : : * ignores WAL records that are not associated with an assigned XID.
395 : : *
396 : : * One might think we need to wait for all running transactions, including
397 : : * those without XIDs and read-only transactions, to finish before
398 : : * enabling logical decoding. However, such a requirement would force the
399 : : * slot creation to wait for a potentially very long time due to
400 : : * long-running read queries, which is practically unacceptable.
401 : : */
402 : :
79 403 : 30 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
404 : :
405 : : /*
406 : : * Re-check whether logical decoding got enabled while we waited for the
407 : : * barrier above.
408 : : */
31 409 [ + + ]: 30 : if (LogicalDecodingCtl->logical_decoding_enabled)
410 : : {
411 : 1 : LogicalDecodingCtl->pending_disable = false;
412 : 1 : LWLockRelease(LogicalDecodingControlLock);
413 : 1 : return;
414 : : }
415 : :
247 416 : 29 : START_CRIT_SECTION();
417 : :
418 : : /*
419 : : * We enable logical decoding first, followed by writing the WAL record.
420 : : * This sequence ensures logical decoding becomes available on the primary
421 : : * first.
422 : : */
423 : 29 : LogicalDecodingCtl->logical_decoding_enabled = true;
424 : :
425 [ + + ]: 29 : if (!in_recovery)
426 : 14 : write_logical_decoding_status_update_record(true);
427 : :
428 : 29 : LogicalDecodingCtl->pending_disable = false;
429 : :
430 [ - + ]: 29 : END_CRIT_SECTION();
431 : :
79 432 : 29 : LWLockRelease(LogicalDecodingControlLock);
433 : :
434 : : /*
435 : : * We log the activation message after releasing the slot lock. This is
436 : : * safe because the activation is performed while holding a logical slot,
437 : : * meaning, a concurrent deactivation cannot interleave its log message
438 : : * ahead of ours.
439 : : */
247 440 [ + + ]: 29 : if (!in_recovery)
441 [ + - ]: 14 : ereport(LOG,
442 : : errmsg("logical decoding is enabled upon creating a new logical replication slot"));
443 : : }
444 : :
445 : : /*
446 : : * Initiate a request for disabling logical decoding.
447 : : *
448 : : * Note that this function does not verify whether logical slots exist. The
449 : : * checkpointer will verify if logical decoding should actually be disabled.
450 : : *
451 : : * This may be called during recovery, for example when a standby invalidates
452 : : * its last valid logical slot. That is safe because the queued request is only
453 : : * acted upon outside recovery. See the RecoveryInProgress() check in
454 : : * DisableLogicalDecodingIfNecessary().
455 : : */
456 : : void
457 : 453 : RequestDisableLogicalDecoding(void)
458 : : {
459 [ + + ]: 453 : if (wal_level != WAL_LEVEL_REPLICA)
460 : 429 : return;
461 : :
462 : : /*
463 : : * It's possible that we might not actually need to disable logical
464 : : * decoding if someone creates a new logical slot concurrently. We set the
465 : : * flag anyway and the checkpointer will check it and disable logical
466 : : * decoding if necessary.
467 : : */
468 : 24 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
469 : 24 : LogicalDecodingCtl->pending_disable = true;
470 : 24 : LWLockRelease(LogicalDecodingControlLock);
471 : :
472 : 24 : WakeupCheckpointer();
473 : :
474 [ + + ]: 24 : elog(DEBUG1, "requested disabling logical decoding");
475 : : }
476 : :
477 : : /*
478 : : * Disable logical decoding if necessary.
479 : : *
480 : : * This function disables logical decoding upon a request initiated by
481 : : * RequestDisableLogicalDecoding(). Otherwise, it performs no action.
482 : : */
483 : : void
484 : 6179 : DisableLogicalDecodingIfNecessary(void)
485 : : {
486 : : bool pending_disable;
487 : :
488 [ + + ]: 6179 : if (wal_level != WAL_LEVEL_REPLICA)
489 : 1476 : return;
490 : :
491 : : /*
492 : : * Sanity check as we cannot disable logical decoding while holding a
493 : : * logical slot.
494 : : */
495 [ - + ]: 4703 : Assert(!MyReplicationSlot);
496 : :
497 : : /*
498 : : * During recovery the logical decoding status follows the primary via WAL
499 : : * replay, so we must not disable it here. A pending_disable request
500 : : * queued during recovery, for example by a local slot invalidation, is
501 : : * intentionally left for the end-of-recovery transition or the
502 : : * post-promotion checkpointer to act on. See
503 : : * UpdateLogicalDecodingStatusEndOfRecovery().
504 : : */
505 [ + + ]: 4703 : if (RecoveryInProgress())
506 : 2002 : return;
507 : :
508 : 2701 : LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
509 : 2701 : pending_disable = LogicalDecodingCtl->pending_disable;
510 : 2701 : LWLockRelease(LogicalDecodingControlLock);
511 : :
512 : : /* Quick return if no pending disable request */
513 [ + + ]: 2701 : if (!pending_disable)
514 : 2680 : return;
515 : :
516 : 21 : DisableLogicalDecoding();
517 : : }
518 : :
519 : : /*
520 : : * A workhorse function to disable logical decoding.
521 : : */
522 : : void
523 : 36 : DisableLogicalDecoding(void)
524 : : {
525 : 36 : bool in_recovery = RecoveryInProgress();
526 : : bool was_enabled;
527 : :
528 : 36 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
529 : :
530 : : /*
531 : : * Check if we can disable logical decoding.
532 : : *
533 : : * Nothing to do if both flags are already off, or if valid slots exist
534 : : * (skip the slot check during recovery because the existing slots will be
535 : : * invalidated after disabling logical decoding.)
536 : : */
79 537 [ + + ]: 36 : if ((!LogicalDecodingCtl->logical_decoding_enabled &&
538 [ + + ]: 3 : !LogicalDecodingCtl->xlog_logical_info) ||
247 539 [ + + + + ]: 34 : (!in_recovery && CheckLogicalSlotExists()))
540 : : {
541 : 6 : LogicalDecodingCtl->pending_disable = false;
542 : 6 : LWLockRelease(LogicalDecodingControlLock);
543 : 6 : return;
544 : : }
545 : :
546 : : /*
547 : : * Remember if logical decoding was enabled. An interrupted activation can
548 : : * leave xlog_logical_info=true while logical_decoding_enabled remains
549 : : * false.
550 : : */
79 551 : 30 : was_enabled = LogicalDecodingCtl->logical_decoding_enabled;
552 : :
247 553 : 30 : START_CRIT_SECTION();
554 : :
555 : : /*
556 : : * We need to disable logical decoding first and then disable logical
557 : : * information WAL logging in order to ensure that no logical decoding
558 : : * processes WAL records with insufficient information.
559 : : */
560 : 30 : LogicalDecodingCtl->logical_decoding_enabled = false;
561 : :
562 : : /* Write the WAL to disable logical decoding on standbys too */
79 563 [ + + + + ]: 30 : if (!in_recovery && was_enabled)
247 564 : 14 : write_logical_decoding_status_update_record(false);
565 : :
566 : : /* Now disable logical information WAL logging */
567 : 30 : LogicalDecodingCtl->xlog_logical_info = false;
568 : 30 : LogicalDecodingCtl->pending_disable = false;
569 : :
570 [ - + ]: 30 : END_CRIT_SECTION();
571 : :
572 : : /*
573 : : * Logging under the lock guarantees our "is disabled" message appears in
574 : : * the server log before its eventual "is enabled", making server log
575 : : * diagnostics easy.
576 : : */
79 577 [ + + + + ]: 30 : if (!in_recovery && was_enabled)
247 578 [ + - ]: 14 : ereport(LOG,
579 : : errmsg("logical decoding is disabled because there are no valid logical replication slots"));
580 : :
581 : 30 : LWLockRelease(LogicalDecodingControlLock);
582 : :
583 : : /*
584 : : * Tell all running processes to reflect the xlog_logical_info update.
585 : : * Unlike when enabling logical decoding, we don't need to wait for all
586 : : * processes to complete it in this case. We already disabled logical
587 : : * decoding and it's always safe to write logical information to WAL
588 : : * records, even when not strictly required. Therefore, we don't need to
589 : : * wait for all running transactions to finish either.
590 : : */
591 : 30 : EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO);
592 : : }
593 : :
594 : : /*
595 : : * Updates the logical decoding status at end of recovery, and ensures that
596 : : * all running processes have the updated XLogLogicalInfo status. This
597 : : * function must be called before accepting writes.
598 : : */
599 : : void
600 : 1001 : UpdateLogicalDecodingStatusEndOfRecovery(void)
601 : : {
602 : 1001 : bool new_status = false;
603 : :
604 [ - + ]: 1001 : Assert(RecoveryInProgress());
605 : :
606 : 1001 : LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
607 : :
608 : : /*
609 : : * With 'minimal' WAL level, no logical replication slot can exist (see
610 : : * RestoreSlotFromDisk()), so the new status is always false. However,
611 : : * logical decoding could have been enabled during recovery by replaying
612 : : * an XLOG_LOGICAL_DECODING_STATUS_CHANGE record from WAL generated with a
613 : : * higher wal_level, e.g. if the server crashed right after the last
614 : : * logical slot was dropped and then restarted with wal_level='minimal'.
615 : : * The code below disables logical decoding in that case.
616 : : */
617 [ + + + + ]: 1001 : if (wal_level == WAL_LEVEL_LOGICAL || CheckLogicalSlotExists())
618 : 140 : new_status = true;
619 : :
620 : : /*
621 : : * When recovery ends, we need to either enable or disable logical
622 : : * decoding based on the wal_level setting and the presence of logical
623 : : * slots. We need to note that concurrent slot creation and deletion could
624 : : * happen but WAL writes are still not permitted until recovery fully
625 : : * completes. Here's how we handle concurrent toggling of logical
626 : : * decoding:
627 : : *
628 : : * For 'enable' case, if there's a concurrent disable request before
629 : : * recovery fully completes, the checkpointer will handle it after
630 : : * recovery is done. This means there might be a brief period after
631 : : * recovery where logical decoding remains enabled even with no logical
632 : : * replication slots present. This temporary state is not new - it can
633 : : * already occur due to the checkpointer's asynchronous deactivation
634 : : * process.
635 : : *
636 : : * For 'disable' case, a backend concurrently creating a logical slot on a
637 : : * standby could have passed its CheckLogicalDecodingRequirements() check
638 : : * when creating its slot only after our slot check above. Such a backend
639 : : * rechecks the status after creating the slot in
640 : : * EnsureLogicalDecodingEnabled() and raises an error if logical decoding
641 : : * has been disabled meanwhile, so it cannot end up with a logical slot
642 : : * while logical decoding remains disabled. (If recovery has already ended
643 : : * by the time of the recheck, the backend instead enables logical
644 : : * decoding by itself, which is fine after promotion.)
645 : : */
646 [ + + ]: 1001 : if (new_status != LogicalDecodingCtl->logical_decoding_enabled)
647 : : {
648 : : /*
649 : : * Update both the logical decoding status and logical WAL logging
650 : : * status. Unlike toggling these status during non-recovery, we don't
651 : : * need to worry about the operation order as WAL writes are still not
652 : : * permitted.
653 : : */
654 : 75 : LogicalDecodingCtl->xlog_logical_info = new_status;
655 : 75 : LogicalDecodingCtl->logical_decoding_enabled = new_status;
656 : :
657 [ + + ]: 75 : elog(DEBUG1,
658 : : "update logical decoding status to %d at the end of recovery",
659 : : new_status);
660 : :
661 : : /*
662 : : * Now that we updated the logical decoding status, clear the pending
663 : : * disable flag. It's possible that a concurrent process drops the
664 : : * last logical slot and initiates the pending disable again. The
665 : : * checkpointer process will check it.
666 : : */
667 : 75 : LogicalDecodingCtl->pending_disable = false;
668 : :
669 : 75 : LWLockRelease(LogicalDecodingControlLock);
670 : :
671 : 75 : write_logical_decoding_status_update_record(new_status);
672 : : }
673 : : else
674 : 926 : LWLockRelease(LogicalDecodingControlLock);
675 : :
676 : : /*
677 : : * Ensure all running processes have the updated status. We don't need to
678 : : * wait for running transactions to finish as we don't accept any writes
679 : : * yet. On the other hand, we need to wait for synchronizing
680 : : * XLogLogicalInfo even if we've not updated the status above as the
681 : : * status have been turned on and off during recovery, having running
682 : : * processes have different status on their local caches.
683 : : */
684 [ + + ]: 1001 : if (IsUnderPostmaster)
685 : 871 : WaitForProcSignalBarrier(
686 : : EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO));
687 : :
688 : 1001 : INJECTION_POINT("startup-logical-decoding-status-change-end-of-recovery", NULL);
689 : 1001 : }
|
