…lines 1–190 unchanged (190 lines)
191/*
192 * If necessary, update the local synced slot's metadata based on the data
193 * from the remote slot.
194 *
195 * If no update was needed (the data of the remote slot is the same as the
196 * local slot) return false, otherwise true.
197 *
198 * *found_consistent_snapshot will be true iff the remote slot's LSN or xmin is
199 * modified, and decoding from the corresponding LSN's can reach a
200 * consistent snapshot.
201 *
202 * *remote_slot_precedes will be true if the remote slot's LSN or xmin
203 * precedes locally reserved position.
204 */
205static bool
206update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid,
207 bool *found_consistent_snapshot,
208 bool *remote_slot_precedes)
209{
210 ReplicationSlot *slot = MyReplicationSlot;
211 bool updated_xmin_or_lsn = false;
212 bool updated_config = false;
213 SlotSyncSkipReason skip_reason = SS_SKIP_NONE;
214
215 Assert(slot->data.invalidated == RS_INVAL_NONE);
216
217 if (found_consistent_snapshot)
218 *found_consistent_snapshot = false;
219
220 if (remote_slot_precedes)
221 *remote_slot_precedes = false;
222
223 /*
224 * Don't overwrite if we already have a newer catalog_xmin and
225 * restart_lsn.
226 */
227 if (remote_slot->restart_lsn < slot->data.restart_lsn ||
…lines 228–258 unchanged (31 lines)
259 errdetail("Synchronization could lead to data loss, because the remote slot needs WAL at LSN %X/%08X and catalog xmin %u, but the standby has LSN %X/%08X and catalog xmin %u.",
260 LSN_FORMAT_ARGS(remote_slot->restart_lsn),
261 remote_slot->catalog_xmin,
262 LSN_FORMAT_ARGS(slot->data.restart_lsn),
263 slot->data.catalog_xmin));
264
265 if (remote_slot_precedes)
266 *remote_slot_precedes = true;
267
268 /*
269 * Skip updating the configuration. This is required to avoid syncing
270 * two_phase_at without syncing confirmed_lsn. Otherwise, the prepared
271 * transaction between old confirmed_lsn and two_phase_at will
272 * unexpectedly get decoded and sent to the downstream after
…lines 273–301 unchanged (29 lines)
302 SpinLockAcquire(&slot->mutex);
303 slot->data.restart_lsn = remote_slot->restart_lsn;
304 slot->data.confirmed_flush = remote_slot->confirmed_lsn;
305 slot->data.catalog_xmin = remote_slot->catalog_xmin;
306 SpinLockRelease(&slot->mutex);
307
308 if (found_consistent_snapshot)
309 *found_consistent_snapshot = true;
310 }
311 else
312 {
313 LogicalSlotAdvanceAndCheckSnapState(remote_slot->confirmed_lsn,
314 found_consistent_snapshot);
315
316 /* Sanity check */
317 if (slot->data.confirmed_flush != remote_slot->confirmed_lsn)
318 ereport(ERROR,
319 errmsg_internal("synchronized confirmed_flush for slot \"%s\" differs from remote slot",
320 remote_slot->name),
321 errdetail_internal("Remote slot has LSN %X/%08X but local slot has LSN %X/%08X.",
322 LSN_FORMAT_ARGS(remote_slot->confirmed_lsn),
323 LSN_FORMAT_ARGS(slot->data.confirmed_flush)));
324
325 /*
326 * If we can't reach a consistent snapshot, the slot won't be
327 * persisted. See update_and_persist_local_synced_slot().
328 */
329 if (found_consistent_snapshot && !(*found_consistent_snapshot))
330 skip_reason = SS_SKIP_NO_CONSISTENT_SNAPSHOT;
331 }
332
333 updated_xmin_or_lsn = true;
334 }
335
336 /* Update slot sync skip stats */
…lines 337–615 unchanged (279 lines)
616 */
617static bool
618update_and_persist_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid,
619 bool *slot_persistence_pending)
620{
621 ReplicationSlot *slot = MyReplicationSlot;
622 bool found_consistent_snapshot = false;
623 bool remote_slot_precedes = false;
624
625 /* Slotsync skip stats are handled in function update_local_synced_slot() */
626 (void) update_local_synced_slot(remote_slot, remote_dbid,
627 &found_consistent_snapshot,
628 &remote_slot_precedes);
629
630 /*
631 * Check if the primary server has caught up. Refer to the comment atop
632 * the file for details on this check.
633 */
634 if (remote_slot_precedes)
635 {
636 /*
637 * The remote slot didn't catch up to locally reserved position.
638 *
639 * We do not drop the slot because the restart_lsn can be ahead of the
640 * current location when recreating the slot in the next cycle. It may
641 * take more time to create such a slot. Therefore, we keep this slot
642 * and attempt the synchronization in the next cycle.
643 *
644 * We also update the slot_persistence_pending parameter, so the SQL
645 * function can retry.
646 */
647 if (slot_persistence_pending)
648 *slot_persistence_pending = true;
649
650 return false;
651 }
652
653 /*
654 * Don't persist the slot if it cannot reach the consistent point from the
655 * restart_lsn. See comments atop this file.
656 */
657 if (!found_consistent_snapshot)
658 {
659 ereport(LOG,
660 errmsg("could not synchronize replication slot \"%s\"", remote_slot->name),
661 errdetail("Synchronization could lead to data loss, because the standby could not build a consistent snapshot to decode WALs at LSN %X/%08X.",
662 LSN_FORMAT_ARGS(slot->data.restart_lsn)));
663
664 /* Set this, so that SQL function can retry */
665 if (slot_persistence_pending)
666 *slot_persistence_pending = true;
667
668 return false;
669 }
670
671 ReplicationSlotPersist();
672
673 ereport(LOG,
674 errmsg("newly created replication slot \"%s\" is sync-ready now",
675 remote_slot->name));
…lines 676–694 unchanged (19 lines)
695 */
696static bool
697synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid,
698 bool *slot_persistence_pending)
699{
700 ReplicationSlot *slot;
701 XLogRecPtr latestFlushPtr = GetStandbyFlushRecPtr(NULL);
702 bool slot_updated = false;
703
704 /* Search for the named slot */
705 if ((slot = SearchNamedReplicationSlot(remote_slot->name, true)))
706 {
707 bool synced;
…lines 708–815 unchanged (108 lines)
816 errmsg_internal("cannot synchronize local slot \"%s\"",
817 remote_slot->name),
818 errdetail_internal("Local slot's start streaming location LSN(%X/%08X) is ahead of remote slot's LSN(%X/%08X).",
819 LSN_FORMAT_ARGS(slot->data.confirmed_flush),
820 LSN_FORMAT_ARGS(remote_slot->confirmed_lsn)));
821
822 slot_updated = update_local_synced_slot(remote_slot, remote_dbid,
823 NULL, NULL);
824 }
825 }
826 /* Otherwise create the slot first. */
827 else
828 {
829 NameData plugin_name;
…lines 830–2052 unchanged (1223 lines)