Hi, On Thu, Sep 17, 2026 at 5:43 PM Masahiko Sawada <[email protected]> wrote: > > Thanks for reviewing the patch! > > While I also think the patch is in good shape, I'd like to raise a > security risk this feature might introduce, particularly around > pg_logical_emit_message():
Nice catch. Thanks for bringing it up. > EXECUTE on that function is granted to PUBLIC, so any role that can > connect to the publisher database can emit a logical decoding message. > With this patch the apply worker hands the message to a handler that > runs with the privileges of the subscription owner, and unlike > insert/update/delete there is no table owner to switch to. One use > case I have in mind is DDL replication. If an extension implements it > on top of this hook and the subscription is owned by a superuser, any > role on the publisher can choose what the handler is given and have it > executed with superuser privileges on the subscriber. So extensions > should carefully consider this case. I think the same applies to other > extensions that might use this feature. The real problem is that there > is no reliable way for the subscriber to tell whether a message came > from the source it expects. Right. This lets a less privileged user on the publisher get more privileged things done on the subscriber. > A practical solution is to revoke EXECUTE on pg_logical_emit_message() > from PUBLIC on the publisher and grant it to a role created for that > purpose. That seems like a good approach. Since the message emitted by pg_logical_emit_message() can carry anything the user wants and goes into the WAL stream, careful production systems may already restrict the function this way. But we can't assume that is always the case. > I think that covers most cases I did consider having the > server record the emitting role in the message so that the subscriber > could check it, but I'm not sure this feature alone justifies it. I lean towards this idea, with one change. Record what the emitter can do, not who it is. Role names and OIDs of ordinary roles on the publisher mean nothing on the subscriber. What the apply side wants to know is simple. What privileges did the emitter have, with role memberships expanded recursively, so they can be compared against what the role applying the message has. So record the emitter's role attributes and the predefined roles it had the privileges of, and hand that to the hook as-is. With this, a hook implementer can say "the emitter was less privileged than me, so I'll skip this logical message", or apply a stricter rule per message type. It also answers the real problem mentioned above, since the subscriber can now tell whether the logical message came from a source it trusts. I can also think of another place where this helps. The logical message today is replayed as-is during crash recovery, on read replicas, and in setups where replay is moved to storage away from compute. I quickly played with this idea using Claude Code to illustrate it better. I haven't reviewed it in depth. Attached for reference. Please have a look. That said, I don't want the hook addition to be blocked on this. It can be discussed separately since there can be other valid use cases. > So > my current thought is to document these risks and add nothing special > for these cases. Documenting the risks is good enough to proceed for now IMHO. But hook implementers can easily miss it. I'm also just thinking out loud here. How about a predefined role for pg_logical_emit_message()? The apply side could switch to that role before calling the hook, which gives logical messages the same kind of switch that apply gets with the table owner, and limits what any logical message can do to whatever that role is granted on the subscriber. Or the apply side could call the hook only when the subscription owner has been granted that role, so the hook stays off unless someone chose to give the subscription owner that role. > I've added the documentation changes and rebased the patch. Any ideas > and feedback is very welcome. I reviewed the v5 patch and the diff from the v4 patch and it looks good to me. -- Bharath Rupireddy Amazon Web Services: https://aws.amazon.com
From 3abb1cdb6c57643f0108bdc3d455094063f87b30 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Sat, 19 Sep 2026 23:19:38 +0000 Subject: [PATCH v1] WIP: Record emitter capabilities in logical decoding messages. pg_logical_emit_message() is executable by PUBLIC, so any role that can connect to a database can put a logical decoding message into WAL. A message has no object, no ACL and no owner, so a consumer has no way to tell whether one came from a privileged role or from any role that happened to be able to connect. Row changes do not have this problem: they are authorized by the ACL on the table they touch and are scoped by publications. Record the capabilities of the emitting role in the xl_logical_message record at emit time: its role attributes from pg_authid as flags, and the names of the predefined (pg_*) roles whose privileges it could exercise, with memberships expanded recursively through inheritable grants. Only capabilities are recorded, never identities. User-defined roles are left out because their names and OIDs mean nothing outside the emitting cluster, while the names of predefined roles are fixed and need no headers to interpret. A separate VALID flag keeps "nothing is known about the emitter" distinct from "the emitter holds no privileges". The capabilities reach the message and stream_message output plugin callbacks, so that a consumer can compare them against the privileges of the role applying the message. Core attaches no policy to them. pgoutput sends them under a new protocol version 5, which the built-in apply worker does not request, so nothing changes for existing subscribers. test_decoding grows an include-message-emitter option, off by default, used by the new tests. The role membership closure is exposed as get_role_privs_closure() in acl.c, a thin wrapper around roles_is_member_of(). Bump XLOG_PAGE_MAGIC for the WAL record format change. --- contrib/test_decoding/expected/messages.out | 66 ++++++++++ contrib/test_decoding/sql/messages.sql | 35 ++++++ contrib/test_decoding/test_decoding.c | 100 ++++++++++++--- src/backend/access/rmgrdesc/logicalmsgdesc.c | 25 +++- src/backend/replication/logical/decode.c | 12 +- src/backend/replication/logical/logical.c | 16 ++- src/backend/replication/logical/message.c | 118 ++++++++++++++++++ src/backend/replication/logical/proto.c | 25 +++- .../replication/logical/reorderbuffer.c | 50 +++++++- src/backend/replication/pgoutput/pgoutput.c | 9 +- src/backend/utils/adt/acl.c | 17 +++ src/include/access/xlog_internal.h | 2 +- src/include/replication/logicalproto.h | 10 +- src/include/replication/message.h | 45 ++++++- src/include/replication/output_plugin.h | 9 +- src/include/replication/reorderbuffer.h | 11 +- src/include/utils/acl.h | 1 + src/tools/pgindent/typedefs.list | 1 + 18 files changed, 505 insertions(+), 47 deletions(-) diff --git a/contrib/test_decoding/expected/messages.out b/contrib/test_decoding/expected/messages.out index 84baf8af3ee..5ed542a871d 100644 --- a/contrib/test_decoding/expected/messages.out +++ b/contrib/test_decoding/expected/messages.out @@ -98,6 +98,72 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'for ------ (0 rows) +-- Memberships are expanded recursively, so regress_msg_reader shows +-- pg_read_all_data, which it holds only through regress_msg_parent. The +-- user-defined roles themselves are not recorded. +CREATE ROLE regress_msg_lowpriv; +CREATE ROLE regress_msg_parent IN ROLE pg_read_all_data; +CREATE ROLE regress_msg_reader IN ROLE regress_msg_parent; +SET SESSION AUTHORIZATION regress_msg_lowpriv; +SELECT 'msg8' FROM pg_logical_emit_message(true, 'test', 'from lowpriv'); + ?column? +---------- + msg8 +(1 row) + +RESET SESSION AUTHORIZATION; +SET SESSION AUTHORIZATION regress_msg_reader; +SELECT 'msg9' FROM pg_logical_emit_message(true, 'test', 'from reader'); + ?column? +---------- + msg9 +(1 row) + +RESET SESSION AUTHORIZATION; +SELECT 'msg10' FROM pg_logical_emit_message(true, 'test', 'from superuser'); + ?column? +---------- + msg10 +(1 row) + +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0', 'include-message-emitter', '1'); + data +----------------------------------------------------------------------------------------------------------------------------------------------------------------------- + BEGIN + message: transactional: 1 prefix: test, sz: 12 emitter: inherit content:from lowpriv + COMMIT + BEGIN + message: transactional: 1 prefix: test, sz: 11 emitter: inherit,pg_read_all_data content:from reader + COMMIT + BEGIN + message: transactional: 1 prefix: test, sz: 14 emitter: superuser,inherit,createrole,createdb,canlogin,replication,bypassrls,pg_database_owner content:from superuser + COMMIT +(9 rows) + +-- Same, but in a transaction large enough to spill to disk, so that the +-- capabilities go through serialize and restore. +CREATE TABLE messages_spill(data text); +BEGIN; +SET SESSION AUTHORIZATION regress_msg_reader; +SELECT 'msg11' FROM pg_logical_emit_message(true, 'test', 'spilled'); + ?column? +---------- + msg11 +(1 row) + +RESET SESSION AUTHORIZATION; +INSERT INTO messages_spill SELECT g.i FROM generate_series(1, 5000) g(i); +COMMIT; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0', 'include-message-emitter', '1') WHERE data LIKE 'message:%'; + data +------------------------------------------------------------------------------------------------- + message: transactional: 1 prefix: test, sz: 7 emitter: inherit,pg_read_all_data content:spilled +(1 row) + +DROP TABLE messages_spill; +DROP ROLE regress_msg_reader; +DROP ROLE regress_msg_parent; +DROP ROLE regress_msg_lowpriv; SELECT 'cleanup' FROM pg_drop_replication_slot('regression_slot'); ?column? ---------- diff --git a/contrib/test_decoding/sql/messages.sql b/contrib/test_decoding/sql/messages.sql index 1f3dcb63ee7..0acf3228e07 100644 --- a/contrib/test_decoding/sql/messages.sql +++ b/contrib/test_decoding/sql/messages.sql @@ -32,4 +32,39 @@ SELECT 'otherdb2' FROM pg_logical_emit_message(true, 'test', 'otherdb2'); \c :prevdb SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1'); +-- Memberships are expanded recursively, so regress_msg_reader shows +-- pg_read_all_data, which it holds only through regress_msg_parent. The +-- user-defined roles themselves are not recorded. +CREATE ROLE regress_msg_lowpriv; +CREATE ROLE regress_msg_parent IN ROLE pg_read_all_data; +CREATE ROLE regress_msg_reader IN ROLE regress_msg_parent; + +SET SESSION AUTHORIZATION regress_msg_lowpriv; +SELECT 'msg8' FROM pg_logical_emit_message(true, 'test', 'from lowpriv'); +RESET SESSION AUTHORIZATION; + +SET SESSION AUTHORIZATION regress_msg_reader; +SELECT 'msg9' FROM pg_logical_emit_message(true, 'test', 'from reader'); +RESET SESSION AUTHORIZATION; + +SELECT 'msg10' FROM pg_logical_emit_message(true, 'test', 'from superuser'); + +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0', 'include-message-emitter', '1'); + +-- Same, but in a transaction large enough to spill to disk, so that the +-- capabilities go through serialize and restore. +CREATE TABLE messages_spill(data text); +BEGIN; +SET SESSION AUTHORIZATION regress_msg_reader; +SELECT 'msg11' FROM pg_logical_emit_message(true, 'test', 'spilled'); +RESET SESSION AUTHORIZATION; +INSERT INTO messages_spill SELECT g.i FROM generate_series(1, 5000) g(i); +COMMIT; +SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'force-binary', '0', 'skip-empty-xacts', '1', 'include-xids', '0', 'include-message-emitter', '1') WHERE data LIKE 'message:%'; +DROP TABLE messages_spill; + +DROP ROLE regress_msg_reader; +DROP ROLE regress_msg_parent; +DROP ROLE regress_msg_lowpriv; + SELECT 'cleanup' FROM pg_drop_replication_slot('regression_slot'); diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c index 78185837aef..339aafd910c 100644 --- a/contrib/test_decoding/test_decoding.c +++ b/contrib/test_decoding/test_decoding.c @@ -15,6 +15,7 @@ #include "catalog/pg_type.h" #include "replication/logical.h" +#include "replication/message.h" #include "replication/origin.h" #include "utils/builtins.h" @@ -34,6 +35,7 @@ typedef struct bool include_timestamp; bool skip_empty_xacts; bool only_local; + bool include_message_emitter; } TestDecodingData; /* @@ -74,7 +76,8 @@ static bool pg_decode_filter(LogicalDecodingContext *ctx, static void pg_decode_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional, const char *prefix, - Size sz, const char *message); + Size sz, const char *message, + const LogicalMessageEmitter *emitter); static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx, TransactionId xid, const char *gid); @@ -114,7 +117,8 @@ static void pg_decode_stream_change(LogicalDecodingContext *ctx, static void pg_decode_stream_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional, const char *prefix, - Size sz, const char *message); + Size sz, const char *message, + const LogicalMessageEmitter *emitter); static void pg_decode_stream_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, int nrelations, Relation relations[], @@ -171,6 +175,7 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, data->include_timestamp = false; data->skip_empty_xacts = false; data->only_local = false; + data->include_message_emitter = false; ctx->output_plugin_private = data; @@ -252,6 +257,16 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, errmsg("could not parse value \"%s\" for parameter \"%s\"", strVal(elem->arg), elem->defname))); } + else if (strcmp(elem->defname, "include-message-emitter") == 0) + { + if (elem->arg == NULL) + data->include_message_emitter = true; + else if (!parse_bool(strVal(elem->arg), &data->include_message_emitter)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("could not parse value \"%s\" for parameter \"%s\"", + strVal(elem->arg), elem->defname))); + } else if (strcmp(elem->defname, "stream-changes") == 0) { if (elem->arg == NULL) @@ -751,10 +766,62 @@ pg_decode_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, OutputPluginWrite(ctx, true); } +/* + * Print the emitter's role attributes, then its predefined roles. + */ +static void +print_emitter(StringInfo s, const LogicalMessageEmitter *emitter) +{ + static const struct + { + uint32 flag; + const char *name; + } attributes[] = { + {LOGICALMSG_EMITTER_SUPERUSER, "superuser"}, + {LOGICALMSG_EMITTER_INHERIT, "inherit"}, + {LOGICALMSG_EMITTER_CREATEROLE, "createrole"}, + {LOGICALMSG_EMITTER_CREATEDB, "createdb"}, + {LOGICALMSG_EMITTER_CANLOGIN, "canlogin"}, + {LOGICALMSG_EMITTER_REPLICATION, "replication"}, + {LOGICALMSG_EMITTER_BYPASSRLS, "bypassrls"}, + }; + const char *rolename = emitter->roles; + const char *sep = ""; + + appendStringInfoString(s, " emitter: "); + + if (!(emitter->flags & LOGICALMSG_EMITTER_VALID)) + { + appendStringInfoString(s, "unknown"); + return; + } + + for (int i = 0; i < lengthof(attributes); i++) + { + if (emitter->flags & attributes[i].flag) + { + appendStringInfo(s, "%s%s", sep, attributes[i].name); + sep = ","; + } + } + + for (uint32 i = 0; i < emitter->nroles; i++) + { + appendStringInfo(s, "%s%s", sep, rolename); + rolename += strlen(rolename) + 1; + sep = ","; + } + + /* no attributes and no predefined roles */ + if (sep[0] == '\0') + appendStringInfoString(s, "unprivileged"); +} + static void pg_decode_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional, - const char *prefix, Size sz, const char *message) + const char *prefix, Size sz, const char *message, + const LogicalMessageEmitter *emitter) { TestDecodingData *data = ctx->output_plugin_private; TestDecodingTxnData *txndata; @@ -769,8 +836,11 @@ pg_decode_message(LogicalDecodingContext *ctx, txndata->xact_wrote_changes = true; OutputPluginPrepareWrite(ctx, true); - appendStringInfo(ctx->out, "message: transactional: %d prefix: %s, sz: %zu content:", + appendStringInfo(ctx->out, "message: transactional: %d prefix: %s, sz: %zu", transactional, prefix, sz); + if (data->include_message_emitter) + print_emitter(ctx->out, emitter); + appendStringInfoString(ctx->out, " content:"); appendBinaryStringInfo(ctx->out, message, sz); OutputPluginWrite(ctx, true); } @@ -955,12 +1025,14 @@ pg_decode_stream_change(LogicalDecodingContext *ctx, static void pg_decode_stream_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional, - const char *prefix, Size sz, const char *message) + const char *prefix, Size sz, const char *message, + const LogicalMessageEmitter *emitter) { + TestDecodingData *data = ctx->output_plugin_private; + /* Output stream start if we haven't yet for transactional messages. */ if (transactional) { - TestDecodingData *data = ctx->output_plugin_private; TestDecodingTxnData *txndata = txn->output_plugin_private; if (data->skip_empty_xacts && !txndata->stream_wrote_changes) @@ -972,15 +1044,15 @@ pg_decode_stream_message(LogicalDecodingContext *ctx, OutputPluginPrepareWrite(ctx, true); - if (transactional) - { - appendStringInfo(ctx->out, "streaming message: transactional: %d prefix: %s, sz: %zu", - transactional, prefix, sz); - } - else + appendStringInfo(ctx->out, "streaming message: transactional: %d prefix: %s, sz: %zu", + transactional, prefix, sz); + + if (data->include_message_emitter) + print_emitter(ctx->out, emitter); + + if (!transactional) { - appendStringInfo(ctx->out, "streaming message: transactional: %d prefix: %s, sz: %zu content:", - transactional, prefix, sz); + appendStringInfoString(ctx->out, " content:"); appendBinaryStringInfo(ctx->out, message, sz); } diff --git a/src/backend/access/rmgrdesc/logicalmsgdesc.c b/src/backend/access/rmgrdesc/logicalmsgdesc.c index 3e8a4f33ee6..b6e7e2c2bf2 100644 --- a/src/backend/access/rmgrdesc/logicalmsgdesc.c +++ b/src/backend/access/rmgrdesc/logicalmsgdesc.c @@ -24,15 +24,32 @@ logicalmsg_desc(StringInfo buf, XLogReaderState *record) if (info == XLOG_LOGICAL_MESSAGE) { xl_logical_message *xlrec = (xl_logical_message *) rec; - char *prefix = xlrec->message; - char *message = xlrec->message + xlrec->prefix_size; + char *prefix = XLLogicalMessagePrefix(xlrec); + char *message = XLLogicalMessagePayload(xlrec); char *sep = ""; Assert(prefix[xlrec->prefix_size - 1] == '\0'); - appendStringInfo(buf, "%s, prefix \"%s\"; payload (%zu bytes): ", + appendStringInfo(buf, "%s, prefix \"%s\"; emitter flags %X", xlrec->transactional ? "transactional" : "non-transactional", - prefix, xlrec->message_size); + prefix, xlrec->emitter_flags); + + if (xlrec->emitter_nroles > 0) + { + char *rolename = XLLogicalMessageRoles(xlrec); + char *rolesep = ""; + + appendStringInfoString(buf, ", emitter roles "); + for (uint32 i = 0; i < xlrec->emitter_nroles; i++) + { + appendStringInfo(buf, "%s%s", rolesep, rolename); + rolename += strlen(rolename) + 1; + rolesep = ","; + } + } + + appendStringInfo(buf, "; payload (%zu bytes): ", + xlrec->message_size); /* Write message payload as a series of hex bytes */ for (Size cnt = 0; cnt < xlrec->message_size; cnt++) { diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 4a739230264..c22d8d5f54e 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -613,6 +613,7 @@ logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) ReplOriginId origin_id = XLogRecGetOrigin(r); Snapshot snapshot = NULL; xl_logical_message *message; + LogicalMessageEmitter emitter; if (info != XLOG_LOGICAL_MESSAGE) elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info); @@ -668,12 +669,17 @@ logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) if (!message->transactional) snapshot = SnapBuildGetOrBuildSnapshot(builder); + emitter.flags = message->emitter_flags; + emitter.nroles = message->emitter_nroles; + emitter.roles_size = message->emitter_roles_size; + emitter.roles = XLLogicalMessageRoles(message); + ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr, message->transactional, - message->message, /* first part of message is - * prefix */ + XLLogicalMessagePrefix(message), message->message_size, - message->message + message->prefix_size); + XLLogicalMessagePayload(message), + &emitter); } /* diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 98e5f1dd8f9..034cbf3a9d8 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -81,7 +81,8 @@ static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change); static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, - const char *prefix, Size message_size, const char *message); + const char *prefix, Size message_size, const char *message, + const LogicalMessageEmitter *emitter); /* streaming callbacks */ static void stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, @@ -98,7 +99,8 @@ static void stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn Relation relation, ReorderBufferChange *change); static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, - const char *prefix, Size message_size, const char *message); + const char *prefix, Size message_size, const char *message, + const LogicalMessageEmitter *emitter); static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, int nrelations, Relation relations[], ReorderBufferChange *change); @@ -1311,7 +1313,8 @@ filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, ReplOriginId origin_id) static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, - const char *prefix, Size message_size, const char *message) + const char *prefix, Size message_size, const char *message, + const LogicalMessageEmitter *emitter) { LogicalDecodingContext *ctx = cache->private_data; LogicalErrorCallbackState state; @@ -1339,7 +1342,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, /* do the actual work: call callback */ ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix, - message_size, message); + message_size, message, emitter); /* Pop the error context stack */ error_context_stack = errcallback.previous; @@ -1622,7 +1625,8 @@ stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, - const char *prefix, Size message_size, const char *message) + const char *prefix, Size message_size, const char *message, + const LogicalMessageEmitter *emitter) { LogicalDecodingContext *ctx = cache->private_data; LogicalErrorCallbackState state; @@ -1654,7 +1658,7 @@ stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, /* do the actual work: call callback */ ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix, - message_size, message); + message_size, message, emitter); /* Pop the error context stack */ error_context_stack = errcallback.previous; diff --git a/src/backend/replication/logical/message.c b/src/backend/replication/logical/message.c index 06825d66e7f..a1f2f3ce467 100644 --- a/src/backend/replication/logical/message.c +++ b/src/backend/replication/logical/message.c @@ -31,10 +31,117 @@ #include "postgres.h" +#include "access/htup_details.h" +#include "access/transam.h" #include "access/xact.h" #include "access/xloginsert.h" +#include "catalog/pg_authid.h" #include "miscadmin.h" #include "replication/message.h" +#include "utils/acl.h" +#include "utils/syscache.h" + +/* + * Collect the capabilities of the role emitting a message; see + * LogicalMessageEmitter. The role names are sorted, so that the same + * capabilities always produce the same bytes. + */ +static void +GetEmitterCapabilities(LogicalMessageEmitter *emitter) +{ + Oid userid = GetUserId(); + HeapTuple tuple; + List *memberships; + ListCell *lc; + char **names; + int nnames = 0; + StringInfoData buf; + + emitter->flags = 0; + emitter->nroles = 0; + emitter->roles_size = 0; + emitter->roles = NULL; + + /* + * Role lookups need catalog access, which we don't have when C code emits + * a non-transactional message outside a transaction. Record nothing + * rather than something wrong. + */ + if (!IsTransactionState()) + return; + + emitter->flags = LOGICALMSG_EMITTER_VALID; + + tuple = SearchSysCache1(AUTHOID, ObjectIdGetDatum(userid)); + if (HeapTupleIsValid(tuple)) + { + Form_pg_authid authform = (Form_pg_authid) GETSTRUCT(tuple); + + if (authform->rolsuper) + emitter->flags |= LOGICALMSG_EMITTER_SUPERUSER; + if (authform->rolinherit) + emitter->flags |= LOGICALMSG_EMITTER_INHERIT; + if (authform->rolcreaterole) + emitter->flags |= LOGICALMSG_EMITTER_CREATEROLE; + if (authform->rolcreatedb) + emitter->flags |= LOGICALMSG_EMITTER_CREATEDB; + if (authform->rolcanlogin) + emitter->flags |= LOGICALMSG_EMITTER_CANLOGIN; + if (authform->rolreplication) + emitter->flags |= LOGICALMSG_EMITTER_REPLICATION; + if (authform->rolbypassrls) + emitter->flags |= LOGICALMSG_EMITTER_BYPASSRLS; + + ReleaseSysCache(tuple); + } + + /* pick the predefined roles out of the emitter's memberships */ + memberships = get_role_privs_closure(userid); + names = palloc(list_length(memberships) * sizeof(char *)); + + foreach(lc, memberships) + { + Oid roleid = lfirst_oid(lc); + char *rolename; + + /* every predefined role has an OID assigned at bootstrap */ + if (roleid >= FirstNormalObjectId) + continue; + + rolename = GetUserNameFromId(roleid, true); + if (rolename == NULL) + continue; /* concurrently dropped */ + + /* the "pg_" prefix is reserved, so this is not a user-defined role */ + if (strncmp(rolename, "pg_", 3) == 0) + names[nnames++] = rolename; + else + pfree(rolename); + } + + list_free(memberships); + + if (nnames == 0) + { + pfree(names); + return; + } + + qsort(names, nnames, sizeof(char *), pg_qsort_strcmp); + + initStringInfo(&buf); + for (int i = 0; i < nnames; i++) + { + appendStringInfoString(&buf, names[i]); + appendStringInfoChar(&buf, '\0'); + pfree(names[i]); + } + pfree(names); + + emitter->nroles = nnames; + emitter->roles_size = buf.len; + emitter->roles = buf.data; +} /* * Write logical decoding message into XLog. @@ -45,6 +152,7 @@ LogLogicalMessage(const char *prefix, const char *message, size_t size, { xl_logical_message xlrec; XLogRecPtr lsn; + LogicalMessageEmitter emitter; /* * Force xid to be allocated if we're emitting a transactional message. @@ -55,7 +163,12 @@ LogLogicalMessage(const char *prefix, const char *message, size_t size, GetCurrentTransactionId(); } + GetEmitterCapabilities(&emitter); + xlrec.dbId = MyDatabaseId; + xlrec.emitter_flags = emitter.flags; + xlrec.emitter_nroles = emitter.nroles; + xlrec.emitter_roles_size = emitter.roles_size; xlrec.transactional = transactional; /* trailing zero is critical; see logicalmsg_desc */ xlrec.prefix_size = strlen(prefix) + 1; @@ -65,12 +178,17 @@ LogLogicalMessage(const char *prefix, const char *message, size_t size, XLogRegisterData(&xlrec, SizeOfLogicalMessage); XLogRegisterData(prefix, xlrec.prefix_size); XLogRegisterData(message, size); + if (emitter.roles_size > 0) + XLogRegisterData(emitter.roles, emitter.roles_size); /* allow origin filtering */ XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN); lsn = XLogInsert(RM_LOGICALMSG_ID, XLOG_LOGICAL_MESSAGE); + if (emitter.roles != NULL) + pfree(emitter.roles); + /* * Make sure that the message hits disk before leaving if emitting a * non-transactional message when flush is requested. diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 86ad97cd937..a88eb723ff0 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -26,6 +26,7 @@ #define LOGICALREP_IS_REPLICA_IDENTITY 1 #define MESSAGE_TRANSACTIONAL (1<<0) +#define MESSAGE_EMITTER_INFO (1<<1) #define TRUNCATE_CASCADE (1<<0) #define TRUNCATE_RESTART_SEQS (1<<1) @@ -635,11 +636,17 @@ logicalrep_read_truncate(StringInfo in, /* * Write MESSAGE to stream + * + * include_emitter says whether to send the emitter's capabilities, which only + * a subscriber on protocol version + * LOGICALREP_PROTO_MESSAGE_EMITTER_VERSION_NUM or above gets. The + * MESSAGE_EMITTER_INFO flag tells the reader whether they are there. */ void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, bool transactional, const char *prefix, Size sz, - const char *message) + const char *message, bool include_emitter, + const LogicalMessageEmitter *emitter) { uint8 flags = 0; @@ -648,12 +655,28 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, /* encode and send message flags */ if (transactional) flags |= MESSAGE_TRANSACTIONAL; + if (include_emitter) + flags |= MESSAGE_EMITTER_INFO; /* transaction ID (if not valid, we're not streaming) */ if (TransactionIdIsValid(xid)) pq_sendint32(out, xid); pq_sendint8(out, flags); + + if (include_emitter) + { + const char *rolename = emitter->roles; + + pq_sendint32(out, emitter->flags); + pq_sendint32(out, emitter->nroles); + for (uint32 i = 0; i < emitter->nroles; i++) + { + pq_sendstring(out, rolename); + rolename += strlen(rolename) + 1; + } + } + pq_sendint64(out, lsn); pq_sendstring(out, prefix); pq_sendint32(out, sz); diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index ede117d339e..34e8ba30f27 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -554,6 +554,9 @@ ReorderBufferFreeChange(ReorderBuffer *rb, ReorderBufferChange *change, if (change->data.msg.message != NULL) pfree(change->data.msg.message); change->data.msg.message = NULL; + if (change->data.msg.emitter.roles != NULL) + pfree(change->data.msg.emitter.roles); + change->data.msg.emitter.roles = NULL; break; case REORDER_BUFFER_CHANGE_INVALIDATION: if (change->data.inval.invalidations) @@ -874,7 +877,8 @@ void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snap, XLogRecPtr lsn, bool transactional, const char *prefix, - Size message_size, const char *message) + Size message_size, const char *message, + const LogicalMessageEmitter *emitter) { if (transactional) { @@ -898,6 +902,15 @@ ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, change->data.msg.message_size = message_size; change->data.msg.message = palloc(message_size); memcpy(change->data.msg.message, message, message_size); + change->data.msg.emitter = *emitter; + if (emitter->roles_size > 0) + { + change->data.msg.emitter.roles = palloc(emitter->roles_size); + memcpy(change->data.msg.emitter.roles, emitter->roles, + emitter->roles_size); + } + else + change->data.msg.emitter.roles = NULL; ReorderBufferQueueChange(rb, xid, lsn, change, false); @@ -918,7 +931,8 @@ ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, SetupHistoricSnapshot(snapshot_now, NULL); PG_TRY(); { - rb->message(rb, txn, lsn, false, prefix, message_size, message); + rb->message(rb, txn, lsn, false, prefix, message_size, message, + emitter); TeardownHistoricSnapshot(false); } @@ -2105,12 +2119,14 @@ ReorderBufferApplyMessage(ReorderBuffer *rb, ReorderBufferTXN *txn, rb->stream_message(rb, txn, change->lsn, true, change->data.msg.prefix, change->data.msg.message_size, - change->data.msg.message); + change->data.msg.message, + &change->data.msg.emitter); else rb->message(rb, txn, change->lsn, true, change->data.msg.prefix, change->data.msg.message_size, - change->data.msg.message); + change->data.msg.message, + &change->data.msg.emitter); } /* @@ -4193,9 +4209,10 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn, { char *data; Size prefix_size = strlen(change->data.msg.prefix) + 1; + Size roles_size = change->data.msg.emitter.roles_size; sz += prefix_size + change->data.msg.message_size + - sizeof(Size) + sizeof(Size); + sizeof(Size) + sizeof(Size) + roles_size; ReorderBufferSerializeReserve(rb, sz); data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange); @@ -4217,6 +4234,13 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn, change->data.msg.message_size); data += change->data.msg.message_size; + /* write the role names; their size is in the change itself */ + if (roles_size > 0) + { + memcpy(data, change->data.msg.emitter.roles, roles_size); + data += roles_size; + } + break; } case REORDER_BUFFER_CHANGE_INVALIDATION: @@ -4529,7 +4553,8 @@ ReorderBufferChangeSize(ReorderBufferChange *change) Size prefix_size = strlen(change->data.msg.prefix) + 1; sz += prefix_size + change->data.msg.message_size + - sizeof(Size) + sizeof(Size); + sizeof(Size) + sizeof(Size) + + change->data.msg.emitter.roles_size; break; } @@ -4808,6 +4833,19 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn, change->data.msg.message_size); data += change->data.msg.message_size; + /* read the role names, whose size we already have */ + if (change->data.msg.emitter.roles_size > 0) + { + change->data.msg.emitter.roles = + MemoryContextAlloc(rb->context, + change->data.msg.emitter.roles_size); + memcpy(change->data.msg.emitter.roles, data, + change->data.msg.emitter.roles_size); + data += change->data.msg.emitter.roles_size; + } + else + change->data.msg.emitter.roles = NULL; + break; } case REORDER_BUFFER_CHANGE_INVALIDATION: diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index b481d5bb388..144fc18d960 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -57,7 +57,8 @@ static void pgoutput_truncate(LogicalDecodingContext *ctx, static void pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, - Size sz, const char *message); + Size sz, const char *message, + const LogicalMessageEmitter *emitter); static bool pgoutput_origin_filter(LogicalDecodingContext *ctx, ReplOriginId origin_id); static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, @@ -1724,7 +1725,7 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, static void pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, - const char *message) + const char *message, const LogicalMessageEmitter *emitter) { PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; TransactionId xid = InvalidTransactionId; @@ -1758,7 +1759,9 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, transactional, prefix, sz, - message); + message, + data->protocol_version >= LOGICALREP_PROTO_MESSAGE_EMITTER_VERSION_NUM, + emitter); OutputPluginWrite(ctx, true); } diff --git a/src/backend/utils/adt/acl.c b/src/backend/utils/adt/acl.c index a01d8f4bc41..1fb91ce1a72 100644 --- a/src/backend/utils/adt/acl.c +++ b/src/backend/utils/adt/acl.c @@ -5298,6 +5298,23 @@ roles_is_member_of(Oid roleid, enum RoleRecurseType type, } +/* + * get_role_privs_closure + * Return the OIDs of all roles whose privileges roleid can exercise: + * roleid itself plus its memberships expanded recursively through + * inheritable grants. + * + * The result is a copy the caller owns, not the cached list that + * roles_is_member_of() returns. Note that only actual memberships are + * returned; a superuser has every role's privileges implicitly. + */ +List * +get_role_privs_closure(Oid roleid) +{ + return list_copy(roles_is_member_of(roleid, ROLERECURSE_PRIVS, + InvalidOid, NULL)); +} + /* * Does member have the privileges of role (directly or indirectly)? * diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h index bf609c3c703..ed8f488dda5 100644 --- a/src/include/access/xlog_internal.h +++ b/src/include/access/xlog_internal.h @@ -32,7 +32,7 @@ /* * Each page of XLOG file has a header like this: */ -#define XLOG_PAGE_MAGIC 0xD122 /* can be used as WAL version indicator */ +#define XLOG_PAGE_MAGIC 0xD123 /* can be used as WAL version indicator */ typedef struct XLogPageHeaderData { diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index 058a955e20c..fe9f7b0acc9 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -36,13 +36,17 @@ * LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM is the minimum protocol version * where we support applying large streaming transactions in parallel. * Introduced in PG16. + * + * LOGICALREP_PROTO_MESSAGE_EMITTER_VERSION_NUM is the minimum protocol + * version where a message carries the capabilities of its emitter. */ #define LOGICALREP_PROTO_MIN_VERSION_NUM 1 #define LOGICALREP_PROTO_VERSION_NUM 1 #define LOGICALREP_PROTO_STREAM_VERSION_NUM 2 #define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM 3 #define LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM 4 -#define LOGICALREP_PROTO_MAX_VERSION_NUM LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM +#define LOGICALREP_PROTO_MESSAGE_EMITTER_VERSION_NUM 5 +#define LOGICALREP_PROTO_MAX_VERSION_NUM LOGICALREP_PROTO_MESSAGE_EMITTER_VERSION_NUM /* * Logical message types @@ -247,7 +251,9 @@ extern void logicalrep_write_truncate(StringInfo out, TransactionId xid, extern List *logicalrep_read_truncate(StringInfo in, bool *cascade, bool *restart_seqs); extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, - bool transactional, const char *prefix, Size sz, const char *message); + bool transactional, const char *prefix, Size sz, + const char *message, bool include_emitter, + const LogicalMessageEmitter *emitter); extern void logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel, Bitmapset *columns, PublishGencolsType include_gencols_type); diff --git a/src/include/replication/message.h b/src/include/replication/message.h index d1c72755084..2243a4eb43b 100644 --- a/src/include/replication/message.h +++ b/src/include/replication/message.h @@ -14,21 +14,64 @@ #include "access/xlogdefs.h" #include "access/xlogreader.h" +/* + * Capabilities of the role that emitted a logical decoding message, as of the + * time it was emitted. + * + * flags carries the emitting role's own attributes, and roles the names of + * the predefined (pg_*) roles whose privileges it could exercise, that is, + * its memberships expanded recursively through inheritable grants. + * User-defined roles are left out because their names and OIDs mean nothing + * outside the emitting cluster. + * + * LOGICALMSG_EMITTER_VALID separates an emitter known to hold no privileges + * from one that nothing is known about. + */ +typedef struct LogicalMessageEmitter +{ + uint32 flags; /* LOGICALMSG_EMITTER_* bits */ + uint32 nroles; /* number of predefined role names */ + uint32 roles_size; /* total size of the packed names */ + char *roles; /* nroles null-terminated names, sorted */ +} LogicalMessageEmitter; + +/* the attribute bits follow pg_authid's columns, in column order */ +#define LOGICALMSG_EMITTER_VALID (1 << 0) +#define LOGICALMSG_EMITTER_SUPERUSER (1 << 1) /* rolsuper */ +#define LOGICALMSG_EMITTER_INHERIT (1 << 2) /* rolinherit */ +#define LOGICALMSG_EMITTER_CREATEROLE (1 << 3) /* rolcreaterole */ +#define LOGICALMSG_EMITTER_CREATEDB (1 << 4) /* rolcreatedb */ +#define LOGICALMSG_EMITTER_CANLOGIN (1 << 5) /* rolcanlogin */ +#define LOGICALMSG_EMITTER_REPLICATION (1 << 6) /* rolreplication */ +#define LOGICALMSG_EMITTER_BYPASSRLS (1 << 7) /* rolbypassrls */ + /* * Generic logical decoding message wal record. */ typedef struct xl_logical_message { Oid dbId; /* database Oid emitted from */ + uint32 emitter_flags; /* LOGICALMSG_EMITTER_* bits */ + uint32 emitter_nroles; /* number of predefined role names */ + uint32 emitter_roles_size; /* total size of the packed names */ bool transactional; /* is message transactional? */ Size prefix_size; /* length of prefix */ Size message_size; /* size of the message */ - /* payload, including null-terminated prefix of length prefix_size */ + + /* payload: prefix, then message, then emitter_nroles role names */ char message[FLEXIBLE_ARRAY_MEMBER]; } xl_logical_message; #define SizeOfLogicalMessage (offsetof(xl_logical_message, message)) +/* payload accessors */ +#define XLLogicalMessagePrefix(xlrec) \ + ((xlrec)->message) +#define XLLogicalMessagePayload(xlrec) \ + ((xlrec)->message + (xlrec)->prefix_size) +#define XLLogicalMessageRoles(xlrec) \ + (XLLogicalMessagePayload(xlrec) + (xlrec)->message_size) + extern XLogRecPtr LogLogicalMessage(const char *prefix, const char *message, size_t size, bool transactional, bool flush); diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h index 842fcde67f9..6a3fb100a52 100644 --- a/src/include/replication/output_plugin.h +++ b/src/include/replication/output_plugin.h @@ -80,7 +80,8 @@ typedef void (*LogicalDecodeCommitCB) (struct LogicalDecodingContext *ctx, XLogRecPtr commit_lsn); /* - * Called for the generic logical decoding messages. + * Called for the generic logical decoding messages. emitter describes what + * the emitting role was allowed to do; see LogicalMessageEmitter. */ typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, @@ -88,7 +89,8 @@ typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx, bool transactional, const char *prefix, Size message_size, - const char *message); + const char *message, + const LogicalMessageEmitter *emitter); /* * Filter changes by origin. @@ -199,7 +201,8 @@ typedef void (*LogicalDecodeStreamMessageCB) (struct LogicalDecodingContext *ctx bool transactional, const char *prefix, Size message_size, - const char *message); + const char *message, + const LogicalMessageEmitter *emitter); /* * Callback for streaming truncates from in-progress transactions. diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index ff825e4b7b2..0ed7f2f1cea 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -12,6 +12,7 @@ #include "access/htup_details.h" #include "lib/ilist.h" #include "lib/pairingheap.h" +#include "replication/message.h" #include "storage/sinval.h" #include "utils/hsearch.h" #include "utils/relcache.h" @@ -124,6 +125,7 @@ typedef struct ReorderBufferChange char *prefix; Size message_size; char *message; + LogicalMessageEmitter emitter; /* capabilities of the emitter */ } msg; /* New snapshot, set when action == *_INTERNAL_SNAPSHOT */ @@ -498,7 +500,8 @@ typedef void (*ReorderBufferMessageCB) (ReorderBuffer *rb, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, - const char *message); + const char *message, + const LogicalMessageEmitter *emitter); /* begin prepare callback signature */ typedef void (*ReorderBufferBeginPrepareCB) (ReorderBuffer *rb, @@ -557,7 +560,8 @@ typedef void (*ReorderBufferStreamMessageCB) (ReorderBuffer *rb, XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz, - const char *message); + const char *message, + const LogicalMessageEmitter *emitter); /* stream truncate callback signature */ typedef void (*ReorderBufferStreamTruncateCB) (ReorderBuffer *rb, @@ -721,7 +725,8 @@ extern void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, extern void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snap, XLogRecPtr lsn, bool transactional, const char *prefix, - Size message_size, const char *message); + Size message_size, const char *message, + const LogicalMessageEmitter *emitter); extern void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, ReplOriginId origin_id, XLogRecPtr origin_lsn); diff --git a/src/include/utils/acl.h b/src/include/utils/acl.h index 13fe6df0988..8f61151b145 100644 --- a/src/include/utils/acl.h +++ b/src/include/utils/acl.h @@ -209,6 +209,7 @@ extern AclMode aclmask(const Acl *acl, Oid roleid, Oid ownerId, AclMode mask, AclMaskHow how); extern int aclmembers(const Acl *acl, Oid **roleids); +extern List *get_role_privs_closure(Oid roleid); extern bool has_privs_of_role(Oid member, Oid role); extern bool member_can_set_role(Oid member, Oid role); extern void check_can_set_role(Oid member, Oid role); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 5d432074c2c..75848653349 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1676,6 +1676,7 @@ LogicalDecodeTruncateCB LogicalDecodingContext LogicalDecodingCtlData LogicalErrorCallbackState +LogicalMessageEmitter LogicalOutputPluginInit LogicalOutputPluginWriterPrepareWrite LogicalOutputPluginWriterUpdateProgress -- 2.47.3
