diff options
| -rw-r--r-- | src/backend/replication/logical/logical.c | 16 | ||||
| -rw-r--r-- | src/backend/replication/logical/logicalfuncs.c | 13 | ||||
| -rw-r--r-- | src/backend/replication/walsender.c | 7 |
3 files changed, 16 insertions, 20 deletions
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index c3ec97a0a62..6082d222d5d 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -518,6 +518,22 @@ CreateDecodingContext(XLogRecPtr start_lsn, errmsg("replication slot \"%s\" was not created in this database", NameStr(slot->data.name)))); + /* + * Check if slot has been invalidated due to max_slot_wal_keep_size. Avoid + * "cannot get changes" wording in this errmsg because that'd be + * confusingly ambiguous about no changes being available when called from + * pg_logical_slot_get_changes_guts(). + */ + if (MyReplicationSlot->data.invalidated == RS_INVAL_WAL_REMOVED) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("can no longer get changes from replication slot \"%s\"", + NameStr(MyReplicationSlot->data.name)), + errdetail("This slot has been invalidated because it exceeded the maximum reserved size."))); + + Assert(MyReplicationSlot->data.invalidated == RS_INVAL_NONE); + Assert(MyReplicationSlot->data.restart_lsn != InvalidXLogRecPtr); + if (start_lsn == InvalidXLogRecPtr) { /* continue from last position */ diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c index fa1b641a2b0..55a24c02c94 100644 --- a/src/backend/replication/logical/logicalfuncs.c +++ b/src/backend/replication/logical/logicalfuncs.c @@ -214,19 +214,6 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin LogicalOutputPrepareWrite, LogicalOutputWrite, NULL); - /* - * After the sanity checks in CreateDecodingContext, make sure the - * restart_lsn is valid. Avoid "cannot get changes" wording in this - * errmsg because that'd be confusingly ambiguous about no changes - * being available. - */ - if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn)) - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("can no longer get changes from replication slot \"%s\"", - NameStr(*name)), - errdetail("This slot has never previously reserved WAL, or it has been invalidated."))); - MemoryContextSwitchTo(oldcontext); /* diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 75e8363e248..e40a9b1ba7b 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1253,13 +1253,6 @@ StartLogicalReplication(StartReplicationCmd *cmd) ReplicationSlotAcquire(cmd->slotname, true); - if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn)) - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("cannot read from logical replication slot \"%s\"", - cmd->slotname), - errdetail("This slot has been invalidated because it exceeded the maximum reserved size."))); - /* * Force a disconnect, so that the decoding code doesn't need to care * about an eventual switch from running in recovery, to running in a |