Re: replication slot restart_lsn initialization

On 2015-07-07 09:42:54 -0700, Gurjeet Singh wrote:
> --- a/src/backend/replication/slot.c
> +++ b/src/backend/replication/slot.c
> @@ -40,10 +40,10 @@
> #include <sys/stat.h>
>
> #include "access/transam.h"
> +#include "access/xlog_internal.h"
> #include "common/string.h"
> #include "miscadmin.h"
> #include "replication/slot.h"
> -#include "storage/fd.h"
> #include "storage/proc.h"
> #include "storage/procarray.h"

Why did you remove fd.h? The file definitely uses infrastructure from
there. We're not terribly consistent about that but I'd rather not rely
on it being included via xlog_internal.h -> xlog.h.

> /*
> + * Grab and save an LSN value to prevent WAL recycling past that point.
> + */
> +void
> +ReplicationSlotRegisterRestartLSN()
> +{

Didn't like that description and function name very much. What does
'grabbing' mean here? Should probably mention that it works on the
currently active slot and modifies it.

It's now ReplicationSlotReserveWal()

> + ReplicationSlot *slot = MyReplicationSlot;
> +
> + Assert(slot != NULL);
> + Assert(slot->data.restart_lsn == InvalidXLogRecPtr);
> +
> + /*
> + * The replication slot mechanism is used to prevent removal of required
> + * WAL. As there is no interlock between this and checkpoints required, WAL
> + * segment could be removed before ReplicationSlotsComputeRequiredLSN() has
> + * been called to prevent that. In the very unlikely case that this happens
> + * we'll just retry.
> + */

You removed some punctuation in that sentence converting a sentence in
bad english into one without the original meaning ;). See the attached
for a new version.

> + while (true)
> + {
> + XLogSegNo segno;
> +
> + /*
> + * Let's start with enough information if we can, so log a standby
> + * snapshot and start logical decoding at exactly that position.
> + */
> + if (!RecoveryInProgress())
> + {
> + XLogRecPtr flushptr;
> +
> + /* start at current insert position */
> + slot->data.restart_lsn = GetXLogInsertRecPtr();
> +
> + /*
> + * Log an xid snapshot for logical replication. This snapshot is not
> + * needed for physical replication, as it relies on the snapshot
> + * created by checkpoint when the base backup starts.
> + */
> + if (slot->data.database != InvalidOid)
> + {
> + /* make sure we have enough information to start */
> + flushptr = LogStandbySnapshot();
> +
> + /* and make sure it's fsynced to disk */
> + XLogFlush(flushptr);
> + }
> + }
> + else
> + slot->data.restart_lsn = GetRedoRecPtr();
> +
> + /* prevent WAL removal as fast as possible */
> + ReplicationSlotsComputeRequiredLSN();
> +
> + /*
> + * If all required WAL is still there, great, otherwise retry. The
> + * slot should prevent further removal of WAL, unless there's a
> + * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
> + * the new restart_lsn above, so normally we should never need to loop
> + * more than twice.
> + */
> + XLByteToSeg(slot->data.restart_lsn, segno);
> + if (XLogGetLastRemovedSegno() < segno)
> + break;
> + }
> +}

The way you added the check for logical vs. physical slots in there
looks wrong to me. For a physical slot created !InRecovy we'll possibly
return a xlog position from the future (it's the insert position *and*
not flushed to disk), which then cannot be received.

> +/*
> * Flush all replication slots to disk.
> *
> * This needn't actually be part of a checkpoint, but it's a convenient
> @@ -876,7 +942,7 @@ StartupReplicationSlots(void)
> }
>
> /* ----
> - * Manipulation of ondisk state of replication slots
> + * Manipulation of on-disk state of replication slots
> *
> * NB: none of the routines below should take any notice whether a slot is the
> * current one or not, that's all handled a layer above.
> diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
> index 9a2793f..01b376a 100644
> --- a/src/backend/replication/slotfuncs.c
> +++ b/src/backend/replication/slotfuncs.c
> @@ -40,6 +40,7 @@ Datum
> pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
> {
> Name name = PG_GETARG_NAME(0);
> + bool immediately_reserve = PG_GETARG_BOOL(1);
> Datum values[2];
> bool nulls[2];
> TupleDesc tupdesc;
> @@ -58,10 +59,28 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
> /* acquire replication slot, this will check for conflicting names */
> ReplicationSlotCreate(NameStr(*name), false, RS_PERSISTENT);
>
> - values[0] = NameGetDatum(&MyReplicationSlot->data.name);
> + if (immediately_reserve)
> + {
> + /* Allocate restart-LSN, if the user asked for it */
> + ReplicationSlotRegisterRestartLSN();
> +
> + /* Write this slot to disk */
> + ReplicationSlotMarkDirty();
> + ReplicationSlotSave();
>
> - nulls[0] = false;
> - nulls[1] = true;
> + values[0] = NameGetDatum(&MyReplicationSlot->data.name);
> + values[1] = LSNGetDatum(MyReplicationSlot->data.restart_lsn);
> +
> + nulls[0] = false;
> + nulls[1] = false;
> + }
> + else
> + {
> + values[0] = NameGetDatum(&MyReplicationSlot->data.name);
> +
> + nulls[0] = false;
> + nulls[1] = true;
> + }

I moved
values[0] = NameGetDatum(&MyReplicationSlot->data.name);
nulls[0] = false;
to outside the conditional block, there seems no reason to have it in
there?

I also removed a bunch of unrelated minor cleanups that I plan to commit
& backpatch separately.

What do you think?

Andres