public inbox for [email protected]
help / color / mirror / Atom feedFrom: Álvaro Herrera <[email protected]>
Subject: [PATCH v2] Disallow direct use of the pgrepack plugin
Date: Mon, 8 Jun 2026 20:57:29 +0200
Nothing can be gained from using pgrepack outside of REPACK
(CONCURRENTLY), and dragons may be there. Reject it.
Reported-by: Nikita Kalinin
Discussion: https://postgr.es/m/[email protected]
---
contrib/test_decoding/expected/repack.out | 11 +++
contrib/test_decoding/sql/repack.sql | 7 ++
src/backend/commands/repack_worker.c | 100 ++++++++------------
src/backend/replication/pgrepack/pgrepack.c | 20 +++-
src/include/commands/repack_internal.h | 2 -
5 files changed, 79 insertions(+), 61 deletions(-)
diff --git a/contrib/test_decoding/expected/repack.out b/contrib/test_decoding/expected/repack.out
index 6204e620b43..c4ff41be690 100644
--- a/contrib/test_decoding/expected/repack.out
+++ b/contrib/test_decoding/expected/repack.out
@@ -101,3 +101,14 @@ DETAIL: REPACK (CONCURRENTLY) does not support deferrable primary keys.
HINT: Use ALTER TABLE ... REPLICA IDENTITY USING INDEX to designate another index as replica identity.
-- clean up
DROP TABLE repack_conc_replident, clstrpart;
+-- verify that the pgrepack plugin cannot be called directly
+CREATE TABLE repack_plugin (a int);
+SELECT * FROM pg_create_logical_replication_slot('s_repack', 'pgrepack');
+ERROR: unsupported use of logical decoding plugin "pgrepack"
+DETAIL: This plugin can only be used by REPACK (CONCURRENTLY).
+CONTEXT: slot "s_repack", output plugin "pgrepack", in the startup callback
+INSERT INTO repack_plugin VALUES (1);
+SELECT * FROM pg_logical_slot_get_binary_changes('s_repack', NULL, NULL);
+ERROR: replication slot "s_repack" does not exist
+SELECT pg_drop_replication_slot('s_repack');
+ERROR: replication slot "s_repack" does not exist
diff --git a/contrib/test_decoding/sql/repack.sql b/contrib/test_decoding/sql/repack.sql
index cea3bd33689..f461f5479f4 100644
--- a/contrib/test_decoding/sql/repack.sql
+++ b/contrib/test_decoding/sql/repack.sql
@@ -75,3 +75,10 @@ REPACK (CONCURRENTLY) repack_conc_replident;
-- clean up
DROP TABLE repack_conc_replident, clstrpart;
+
+-- verify that the pgrepack plugin cannot be called directly
+CREATE TABLE repack_plugin (a int);
+SELECT * FROM pg_create_logical_replication_slot('s_repack', 'pgrepack');
+INSERT INTO repack_plugin VALUES (1);
+SELECT * FROM pg_logical_slot_get_binary_changes('s_repack', NULL, NULL);
+SELECT pg_drop_replication_slot('s_repack');
diff --git a/src/backend/commands/repack_worker.c b/src/backend/commands/repack_worker.c
index b6b7b604b4f..3a8ddff9124 100644
--- a/src/backend/commands/repack_worker.c
+++ b/src/backend/commands/repack_worker.c
@@ -197,7 +197,7 @@ repack_setup_logical_decoding(Oid relid)
Relation rel;
Oid toastrelid;
LogicalDecodingContext *ctx;
- NameData slotname;
+ char slotname[NAMEDATALEN];
RepackDecodingState *dstate;
MemoryContext oldcxt;
@@ -207,43 +207,26 @@ repack_setup_logical_decoding(Oid relid)
*/
Assert(!TransactionIdIsValid(GetTopTransactionIdIfAny()));
- /*
- * Make sure we can use logical decoding.
- */
+ /* Make sure we can use logical decoding */
CheckLogicalDecodingRequirements(true);
/*
- * A single backend should not execute multiple REPACK commands at a time,
- * so use PID to make the slot unique.
+ * Create the replication slot we'll use, and enable logical decoding in
+ * case it isn't already on.
*
- * RS_TEMPORARY so that the slot gets cleaned up on ERROR.
+ * Make it RS_TEMPORARY so that it's removed on ERROR. A single backend
+ * should not execute multiple REPACK commands at a time, so use the PID
+ * to make the slot name unique.
*/
- snprintf(NameStr(slotname), NAMEDATALEN, "repack_%d", MyProcPid);
- ReplicationSlotCreate(NameStr(slotname), true, RS_TEMPORARY, false, true,
+ snprintf(slotname, NAMEDATALEN, "repack_%d", MyProcPid);
+ ReplicationSlotCreate(slotname, true, RS_TEMPORARY, false, true,
false, false);
-
EnsureLogicalDecodingEnabled();
/*
- * Neither prepare_write nor do_write callback nor update_progress is
- * useful for us.
+ * Setup repacked_rel_locator and repacked_rel_toast_locator, which we use
+ * to skip decoding of unrelated relations.
*/
- ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME,
- NIL,
- true,
- true,
- InvalidXLogRecPtr,
- XL_ROUTINE(.page_read = read_local_xlog_page,
- .segment_open = wal_segment_open,
- .segment_close = wal_segment_close),
- NULL, NULL, NULL);
-
- /*
- * We don't have control on setting fast_forward, so at least check it.
- */
- Assert(!ctx->fast_forward);
-
- /* Avoid logical decoding of other relations. */
rel = table_open(relid, AccessShareLock);
repacked_rel_locator = rel->rd_locator;
toastrelid = rel->rd_rel->reltoastrelid;
@@ -258,16 +241,38 @@ repack_setup_logical_decoding(Oid relid)
}
table_close(rel, AccessShareLock);
- DecodingContextFindStartpoint(ctx);
-
/*
- * decode_concurrent_changes() needs non-blocking callback.
+ * Set up our logical decoding context. We initially use the blocking
+ * read_local_xlog_page until we have found the start point, and switch to
+ * the non-blocking one after.
*/
- ctx->reader->routine.page_read = read_local_xlog_page_no_wait;
+ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME,
+ NIL,
+ true,
+ true,
+ InvalidXLogRecPtr,
+ XL_ROUTINE(.page_read = read_local_xlog_page,
+ .segment_open = wal_segment_open,
+ .segment_close = wal_segment_close),
+ NULL, NULL, NULL);
+
+ /* Complete setup of output_writer_private */
+ dstate = (RepackDecodingState *) ctx->output_writer_private;
+ dstate->relid = relid;
+ dstate->worker_cxt = CurrentMemoryContext;
+ dstate->worker_resowner = CurrentResourceOwner;
+
+ /* We don't have control on fast_forward, but verify it's sane */
+ Assert(!ctx->fast_forward);
+
+ DecodingContextFindStartpoint(ctx);
/* Some WAL records should have been read. */
Assert(XLogRecPtrIsValid(ctx->reader->EndRecPtr));
+ /* Switch to using non-blocking WAL reads now. */
+ ctx->reader->routine.page_read = read_local_xlog_page_no_wait;
+
/*
* Initialize repack_current_segment so that we can notice WAL segment
* boundaries.
@@ -275,36 +280,15 @@ repack_setup_logical_decoding(Oid relid)
XLByteToSeg(ctx->reader->EndRecPtr, repack_current_segment,
wal_segment_size);
- /* Our private state belongs to the decoding context. */
+ /*
+ * Set up our reader private state to let the page-read callback notify
+ * when end-of-WAL has been reached. This lives in the same context as
+ * the logical decoding itself.
+ */
oldcxt = MemoryContextSwitchTo(ctx->context);
-
- /*
- * read_local_xlog_page_no_wait() needs to be able to indicate the end of
- * WAL.
- */
ctx->reader->private_data = palloc0_object(ReadLocalXLogPageNoWaitPrivate);
- dstate = palloc0_object(RepackDecodingState);
MemoryContextSwitchTo(oldcxt);
-#ifdef USE_ASSERT_CHECKING
- dstate->relid = relid;
-#endif
-
- dstate->change_cxt = AllocSetContextCreate(ctx->context,
- "REPACK - change",
- ALLOCSET_DEFAULT_SIZES);
-
- /* The file will be set as soon as we have it opened. */
- dstate->file = NULL;
-
- /*
- * Memory context and resource owner for long-lived resources.
- */
- dstate->worker_cxt = CurrentMemoryContext;
- dstate->worker_resowner = CurrentResourceOwner;
-
- ctx->output_writer_private = dstate;
-
return ctx;
}
diff --git a/src/backend/replication/pgrepack/pgrepack.c b/src/backend/replication/pgrepack/pgrepack.c
index a2615ce54c1..5d0f28a65a9 100644
--- a/src/backend/replication/pgrepack/pgrepack.c
+++ b/src/backend/replication/pgrepack/pgrepack.c
@@ -13,6 +13,7 @@
#include "postgres.h"
#include "access/detoast.h"
+#include "commands/repack.h"
#include "commands/repack_internal.h"
#include "replication/snapbuild.h"
#include "utils/memutils.h"
@@ -47,7 +48,24 @@ static void
repack_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
bool is_init)
{
- ctx->output_plugin_private = NULL;
+ RepackDecodingState *dstate;
+
+ if (!AmRepackWorker())
+ ereport(ERROR,
+ errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("unsupported use of logical decoding plugin \"%s\"",
+ "pgrepack"),
+ errdetail("This plugin can only be used by %s.",
+ "REPACK (CONCURRENTLY)"));
+
+ /* Set up our private state. */
+ Assert(CurrentMemoryContext == ctx->context);
+ dstate = palloc0_object(RepackDecodingState);
+ dstate->change_cxt = AllocSetContextCreate(ctx->context,
+ "REPACK - change",
+ ALLOCSET_DEFAULT_SIZES);
+ /* repack_setup_logical_decoding fills in the rest */
+ ctx->output_writer_private = dstate;
/* Probably unnecessary, as we don't use the SQL interface ... */
opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
diff --git a/src/include/commands/repack_internal.h b/src/include/commands/repack_internal.h
index 6a85cee8910..42111aa4ae3 100644
--- a/src/include/commands/repack_internal.h
+++ b/src/include/commands/repack_internal.h
@@ -39,10 +39,8 @@ typedef char ConcurrentChangeKind;
*/
typedef struct RepackDecodingState
{
-#ifdef USE_ASSERT_CHECKING
/* The relation whose changes we're decoding. */
Oid relid;
-#endif
/* Per-change memory context. */
MemoryContext change_cxt;
--
2.47.3
--qnq35le53r53khhg--
view thread (736+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH v2] Disallow direct use of the pgrepack plugin
In-Reply-To: <no-message-id-20509@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox