Logical Decoding Output Plugin for pg_vault_tde — Technical Research Report

Date: 2026-02-28 Author: @Architect / @SecurityKMS research Status: Research complete, ready for implementation

1. Executive Summary

The WAL sender process reads raw WAL records (xl_heap_insert, xl_heap_update, xl_heap_delete) which contain the on-disk tuple representation — ciphertext for encrypted_heap tables. The Table Access Method callbacks are never invoked in the logical decoding path. A custom output plugin is required to intercept tuples before they are sent to subscribers and decrypt them in-place.

Key finding: The output plugin API is identical between PG17 and PG18 — no version guards needed for logical decoding code.

2. Output Plugin API — Exact Signatures

2.1 Entry Point

The plugin entry point is _PG_output_plugin_init, looked up via load_external_function():

/*
 * From src/include/replication/output_plugin.h:
 *
 * Type of the shared library symbol _PG_output_plugin_init that is
 * looked up when loading an output plugin shared library.
 */
typedef void (*LogicalOutputPluginInit)(struct OutputPluginCallbacks *cb);

extern PGDLLEXPORT void _PG_output_plugin_init(struct OutputPluginCallbacks *cb);

The loading mechanism (logical.c:LoadOutputPlugin): ```c static void LoadOutputPlugin(OutputPluginCallbacks callbacks, const char plugin) { LogicalOutputPluginInit plugin_init;

plugin_init = (LogicalOutputPluginInit)
    load_external_function(plugin, "_PG_output_plugin_init", false, NULL);

if (plugin_init == NULL)
    elog(ERROR, "output plugins have to declare the _PG_output_plugin_init symbol");

plugin_init(callbacks);

/* These 3 callbacks are MANDATORY: */
if (callbacks->begin_cb == NULL)
    elog(ERROR, "output plugins have to register a begin callback");
if (callbacks->change_cb == NULL)
    elog(ERROR, "output plugins have to register a change callback");
if (callbacks->commit_cb == NULL)
    elog(ERROR, "output plugins have to register a commit callback");

} ```

2.2 OutputPluginCallbacks Struct (PG17 & PG18 — identical)

typedef struct OutputPluginCallbacks
{
    /* MANDATORY — must be non-NULL */
    LogicalDecodeStartupCB       startup_cb;          /* optional but strongly recommended */
    LogicalDecodeBeginCB         begin_cb;             /* MANDATORY */
    LogicalDecodeChangeCB        change_cb;            /* MANDATORY */
    LogicalDecodeTruncateCB      truncate_cb;          /* optional */
    LogicalDecodeCommitCB        commit_cb;            /* MANDATORY */
    LogicalDecodeMessageCB       message_cb;           /* optional */
    LogicalDecodeFilterByOriginCB filter_by_origin_cb; /* optional */
    LogicalDecodeShutdownCB      shutdown_cb;          /* optional but recommended */

    /* Two-phase commit support (optional) */
    LogicalDecodeFilterPrepareCB     filter_prepare_cb;
    LogicalDecodeBeginPrepareCB      begin_prepare_cb;
    LogicalDecodePrepareCB           prepare_cb;
    LogicalDecodeCommitPreparedCB    commit_prepared_cb;
    LogicalDecodeRollbackPreparedCB  rollback_prepared_cb;

    /* Streaming of in-progress transactions (optional) */
    LogicalDecodeStreamStartCB       stream_start_cb;
    LogicalDecodeStreamStopCB        stream_stop_cb;
    LogicalDecodeStreamAbortCB       stream_abort_cb;
    LogicalDecodeStreamPrepareCB     stream_prepare_cb;
    LogicalDecodeStreamCommitCB      stream_commit_cb;
    LogicalDecodeStreamChangeCB      stream_change_cb;
    LogicalDecodeStreamMessageCB     stream_message_cb;
    LogicalDecodeStreamTruncateCB    stream_truncate_cb;
} OutputPluginCallbacks;

2.3 Required Callback Signatures

/* Startup: called when the replication slot is created or reused */
typedef void (*LogicalDecodeStartupCB)(
    struct LogicalDecodingContext *ctx,
    OutputPluginOptions *options,    /* set options->output_type here */
    bool is_init);                  /* true = slot just created */

/* BEGIN transaction */
typedef void (*LogicalDecodeBeginCB)(
    struct LogicalDecodingContext *ctx,
    ReorderBufferTXN *txn);

/* Per-row change — THIS IS WHERE DECRYPTION HAPPENS */
typedef void (*LogicalDecodeChangeCB)(
    struct LogicalDecodingContext *ctx,
    ReorderBufferTXN *txn,
    Relation relation,               /* fully opened Relation handle */
    ReorderBufferChange *change);

/* COMMIT transaction */
typedef void (*LogicalDecodeCommitCB)(
    struct LogicalDecodingContext *ctx,
    ReorderBufferTXN *txn,
    XLogRecPtr commit_lsn);

/* Shutdown: cleanup */
typedef void (*LogicalDecodeShutdownCB)(
    struct LogicalDecodingContext *ctx);

/* TRUNCATE (optional but recommended) */
typedef void (*LogicalDecodeTruncateCB)(
    struct LogicalDecodingContext *ctx,
    ReorderBufferTXN *txn,
    int nrelations,
    Relation relations[],
    ReorderBufferChange *change);

3. ReorderBufferChange — Tuple Access

3.1 Struct Layout (PG17 & PG18 — identical)

typedef struct ReorderBufferChange
{
    XLogRecPtr      lsn;
    ReorderBufferChangeType action;  /* INSERT, UPDATE, DELETE, TRUNCATE, ... */
    struct ReorderBufferTXN *txn;
    RepOriginId     origin_id;

    union
    {
        /* For INSERT, UPDATE, DELETE: */
        struct
        {
            RelFileLocator rlocator;       /* physical storage location */
            bool           clear_toast_afterwards;
            HeapTuple      oldtuple;       /* valid for DELETE, UPDATE (replica identity) */
            HeapTuple      newtuple;       /* valid for INSERT, UPDATE */
        } tp;

        /* For TRUNCATE: */
        struct { Size nrelids; bool cascade; bool restart_seqs; Oid *relids; } truncate;

        /* For MESSAGE: */
        struct { char *prefix; Size message_size; char *message; } msg;

        /* Internal types omitted */
    } data;

    dlist_node node;
} ReorderBufferChange;

3.2 Critical Detail: Tuples Are Direct HeapTuple

In both PG17 and PG18, change->data.tp.oldtuple and change->data.tp.newtuple are HeapTuple (i.e., HeapTupleData *), NOT the old ReorderBufferTupleBuf * from PG ≤ 16.

Verified from actual installed headers: - /usr/include/postgresql/17/server/replication/reorderbuffer.h:99: HeapTuple oldtuple; - /usr/include/postgresql/17/server/replication/reorderbuffer.h:101: HeapTuple newtuple; - /usr/include/postgresql/18/server/replication/reorderbuffer.h:104: HeapTuple oldtuple; - /usr/include/postgresql/18/server/replication/reorderbuffer.h:106: HeapTuple newtuple;

No casting needed — these are ready-to-use HeapTuple pointers.

The HeapTuple t_data pointer is valid and points to in-memory data (not a buffer page). The tuple layout is: [HeapTupleHeader (t_hoff bytes)] [user data].

For encrypted_heap tables, the user data portion [t_hoff..t_len) contains the wire format: [IV(12) | CIPHERTEXT(N) | GCM-TAG(16)].

3.3 How pgoutput Accesses Tuples

From pgoutput.c (the built-in output plugin): ```c / Store into a TupleTableSlot for column-level processing / if (change->data.tp.oldtuple) { old_slot = relentry->old_slot; ExecStoreHeapTuple(change->data.tp.oldtuple, old_slot, false); }

if (change->data.tp.newtuple) { new_slot = relentry->new_slot; ExecStoreHeapTuple(change->data.tp.newtuple, new_slot, false); } ```

For our plugin, we can work directly with the HeapTuple without going through slots, since we just need to decrypt the user data in-place.

4. Detecting encrypted_heap Tables

4.1 Using relation->rd_rel->relam

The change_cb callback receives a fully opened Relation handle. The AM is stored in relation->rd_rel->relam (type Oid).

From /usr/include/postgresql/18/server/catalog/pg_class.h:53: c Oid relam BKI_DEFAULT(heap) BKI_LOOKUP_OPT(pg_am);

4.2 Looking Up the encrypted_heap AM OID

#include "commands/defrem.h"   /* get_table_am_oid() */

/* At startup_cb time, resolve OID once: */
Oid encrypted_heap_am_oid = get_table_am_oid("encrypted_heap", true);
/* missing_ok=true: returns InvalidOid if extension not installed */

/* In change_cb, check each relation: */
static inline bool
relation_uses_encrypted_heap(Relation relation)
{
    return OidIsValid(encrypted_heap_am_oid) &&
           relation->rd_rel->relam == encrypted_heap_am_oid;
}

4.3 Alternative: Compare rd_tableam Pointer

We could compare relation->rd_tableam == pg_vault_tde_get_tableam_routine(), but this is fragile because the WAL sender’s relcache may not have loaded our AM handler yet. The OID comparison is more robust.

5. Execution Environment

5.1 WAL Sender Process

The output plugin runs in the WAL sender process, which is: - A regular PostgreSQL backend process forked from the postmaster - Has full access to shared memory (inherits shmem from postmaster) - Has a valid MemoryContext hierarchy - Can call all PostgreSQL backend functions (palloc, ereport, catalog lookups) - Has the relcache available (can do RelationIdGetRelation, etc.)

5.2 Shared Memory & DEK Access

Since pg_vault_tde is loaded via shared_preload_libraries: 1. The postmaster runs _PG_init() which chains shmem_request_hook and shmem_startup_hook 2. Shared memory (DEK cache, LWLock) is initialized before any backend forks 3. The WAL sender inherits the shared memory mappings 4. pg_vault_tde_kms_get_dek() works correctly in the WAL sender context 5. The per-backend EVP context pool and IV batch are initialized per-process 6. on_proc_exit(tde_backend_cleanup, ...) is also inherited and will fire on WAL sender exit

Conclusion: The DEK is accessible. The crypto functions work. No special initialization needed beyond what _PG_init already does.

6. PG17 vs PG18 API Differences

6.1 Logical Decoding: NO Differences

Verified by diffing the actual installed headers: - output_plugin.h: Identical (only copyright year changed) - logical.h: Identical (only copyright year changed) - reorderbuffer.h: Minor changes in RBTXN flags and prepare status macros; no changes to ReorderBufferChange struct layout or tuple types

6.2 ReorderBuffer API Function Names (Minor Difference)

API PG17 PG18
Allocate tuple buffer ReorderBufferGetTupleBuf() ReorderBufferAllocTupleBuf()
Free tuple buffer ReorderBufferReturnTupleBuf() ReorderBufferFreeTupleBuf()

These are not used by output plugins (only by the decoder internals), so they are irrelevant for our implementation.

Result: No #if PG_VERSION_NUM guards needed for the output plugin code.

7. Architecture Decision: Same .so vs Separate .so

7.1 Recommendation: Same .so (pg_vault_tde.so)

Both _PG_init and _PG_output_plugin_init can coexist in the same shared library. The test_decoding contrib module does exactly this:

/* test_decoding.c */
void _PG_init(void) { /* ... */ }

void _PG_output_plugin_init(OutputPluginCallbacks *cb)
{
    cb->startup_cb = pg_decode_startup;
    cb->change_cb = pg_decode_change;
    /* ... */
}

Advantages of same .so: 1. Direct access to tde_decrypt_heap_tuple() (static in tam.c — would need to be made non-static or a wrapper exposed) 2. Direct access to tde_gcm_decrypt() and pg_vault_tde_kms_get_dek() 3. No additional build artifacts or packaging complexity 4. The library is already loaded via shared_preload_libraries — dlopen() just increments the refcount and finds the symbol

Usage: ```sql – Create a logical replication slot using our plugin: SELECT pg_create_logical_replication_slot(‘tde_slot’, ‘pg_vault_tde’);

– Or with pg_recvlogical: pg_recvlogical –plugin=pg_vault_tde –slot=tde_slot –start ```

7.2 What Needs to Change

  1. tde_decrypt_heap_tuple() is currently static in pg_vault_tde_tam.c. Either:

    • (Preferred) Create a new wrapper in a separate file (src/logicaldec/) that calls tde_gcm_decrypt() directly (same logic, fewer dependencies)
    • Or expose tde_decrypt_heap_tuple() via pg_vault_tde_tam.h

  2. Add _PG_output_plugin_init to src/pg_vault_tde.c alongside the existing _PG_init

  3. No new OBJS needed if we add the output plugin code inline in pg_vault_tde.c, or add one new .o to the Makefile OBJS list

8. Concrete Implementation Skeleton

/*
 * pg_vault_tde_logicaldec.c — Logical decoding output plugin
 *
 * Decrypts tuples from encrypted_heap tables before sending them
 * to logical replication subscribers or pg_recvlogical consumers.
 *
 * The WAL sender reads raw WAL records (xl_heap_insert etc.) which
 * contain ciphertext — the TAM callbacks are never invoked.  This
 * output plugin intercepts the change_cb and transparently replaces
 * encrypted HeapTuples with their decrypted equivalents.
 *
 * Copyright (c) 2026 Miriade S.r.l.
 * Licensed under the PostgreSQL License (BSD).
 */
#include "postgres.h"

#include "access/heapam.h"
#include "access/htup_details.h"
#include "catalog/pg_class.h"
#include "commands/defrem.h"          /* get_table_am_oid */
#include "fmgr.h"
#include "replication/logical.h"
#include "replication/output_plugin.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"

#include "src/include/pg_vault_tde_crypto.h"
#include "src/include/pg_vault_tde_kms.h"

/*
 * Per-plugin private state, stored in ctx->output_plugin_private.
 */
typedef struct TdeOutputPluginState
{
    MemoryContext context;             /* private transient memory context */
    Oid           encrypted_heap_oid;  /* cached OID of encrypted_heap AM */
    bool          text_mode;           /* true = textual output, false = binary */
} TdeOutputPluginState;


/* ----------------------------------------------------------------
 * Helper: Decrypt a HeapTuple from an encrypted_heap table
 *
 * Takes a HeapTuple whose user-data region [t_hoff .. t_len) contains
 * [IV(12) | CIPHERTEXT | TAG(16)] and returns a palloc'd HeapTuple
 * with plaintext user data.
 *
 * This duplicates the logic of tde_decrypt_heap_tuple() from tam.c
 * to avoid making that static function externally visible.
 * ----------------------------------------------------------------
 */
static HeapTuple
tde_logicaldec_decrypt_tuple(HeapTuple enc)
{
    Size        hdr_len   = enc->t_data->t_hoff;
    char       *enc_data  = (char *) enc->t_data + hdr_len;
    Size        enc_len   = enc->t_len - hdr_len;
    Size        pt_len    = 0;
    char       *pt_buf;
    HeapTuple   plain;

    if (enc_len < (Size) TDE_GCM_OVERHEAD)
        ereport(ERROR,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("pg_vault_tde: logical decoding: encrypted tuple "
                        "too short (%zu bytes)", enc_len)));

    pt_buf = tde_gcm_decrypt(enc_data, enc_len, &pt_len);

    plain = (HeapTuple) palloc0(HEAPTUPLESIZE + hdr_len + pt_len);
    plain->t_len      = (uint32) (hdr_len + pt_len);
    plain->t_self     = enc->t_self;
    plain->t_tableOid = enc->t_tableOid;
    plain->t_data     = (HeapTupleHeader) ((char *) plain + HEAPTUPLESIZE);

    memcpy(plain->t_data, enc->t_data, hdr_len);                  /* header verbatim */
    memcpy((char *) plain->t_data + hdr_len, pt_buf, pt_len);     /* decrypted payload */

    OPENSSL_cleanse(pt_buf, pt_len);
    pfree(pt_buf);

    return plain;
}


/* ----------------------------------------------------------------
 * Helper: decrypt a tuple in-place inside a ReorderBufferChange,
 * replacing the ciphertext HeapTuple with a plaintext HeapTuple.
 * ----------------------------------------------------------------
 */
static void
tde_logicaldec_decrypt_change(ReorderBufferChange *change, Relation relation,
                              Oid encrypted_heap_oid)
{
    HeapTuple   plain;

    /* Only process tables using encrypted_heap AM */
    if (!OidIsValid(encrypted_heap_oid) ||
        relation->rd_rel->relam != encrypted_heap_oid)
        return;

    /*
     * Decrypt newtuple (INSERT, UPDATE).
     *
     * The HeapTuple in the ReorderBuffer is palloc'd in the reorder
     * buffer's memory context.  We allocate a new plaintext tuple and
     * swap it in, freeing the encrypted original.
     */
    if (change->data.tp.newtuple != NULL)
    {
        plain = tde_logicaldec_decrypt_tuple(change->data.tp.newtuple);
        /* The original encrypted tuple is owned by the ReorderBuffer;
         * we can free it here because the RB won't access it again
         * after handing it to the change_cb.
         * Note: ReorderBufferFreeTupleBuf / ReorderBufferReturnTupleBuf
         * is the proper API but it's not accessible here.  Since these
         * tuples are palloc'd, pfree suffices.
         */
        pfree(change->data.tp.newtuple);
        change->data.tp.newtuple = plain;
    }

    /*
     * Decrypt oldtuple (UPDATE with replica identity, DELETE).
     *
     * The old tuple contains only replica-identity columns, which are
     * also encrypted in the WAL record.
     */
    if (change->data.tp.oldtuple != NULL)
    {
        plain = tde_logicaldec_decrypt_tuple(change->data.tp.oldtuple);
        pfree(change->data.tp.oldtuple);
        change->data.tp.oldtuple = plain;
    }
}


/* ================================================================
 * Output Plugin Callbacks
 * ================================================================ */

/*
 * Startup callback — called when the replication slot is created or reused.
 */
static void
tde_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
                   bool is_init)
{
    TdeOutputPluginState *state;

    state = palloc0(sizeof(TdeOutputPluginState));
    state->context = AllocSetContextCreate(ctx->context,
                                           "pg_vault_tde logical decoding",
                                           ALLOCSET_DEFAULT_SIZES);

    /*
     * Cache the encrypted_heap AM OID once at startup.
     * missing_ok=true: if the extension isn't installed in this DB
     * (shouldn't happen, since we're loaded in shared_preload_libraries),
     * we get InvalidOid and skip all decryption.
     */
    state->encrypted_heap_oid = get_table_am_oid("encrypted_heap", true);

    ctx->output_plugin_private = state;

    /* Use textual output by default (like test_decoding).
     * A production version may want to use binary for efficiency. */
    opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT;
    /* Don't need rewrites (DDL rewriting during initial snapshot) */
    opt->receive_rewrites = false;

    if (!OidIsValid(state->encrypted_heap_oid))
        ereport(LOG,
                (errmsg("pg_vault_tde: logical decoding plugin active, "
                        "but encrypted_heap AM not found in this database")));
    else
        ereport(LOG,
                (errmsg("pg_vault_tde: logical decoding plugin active, "
                        "encrypted_heap AM OID = %u",
                        state->encrypted_heap_oid)));
}


/*
 * BEGIN callback.
 */
static void
tde_decode_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
{
    OutputPluginPrepareWrite(ctx, true);
    appendStringInfo(ctx->out, "BEGIN %u", txn->xid);
    OutputPluginWrite(ctx, true);
}


/*
 * COMMIT callback.
 */
static void
tde_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                      XLogRecPtr commit_lsn)
{
    OutputPluginPrepareWrite(ctx, true);
    appendStringInfo(ctx->out, "COMMIT %u", txn->xid);
    OutputPluginWrite(ctx, true);
}


/*
 * CHANGE callback — the core of the output plugin.
 *
 * For encrypted_heap tables, we decrypt the tuple data before
 * outputting it.  For standard heap tables, we pass through unchanged.
 */
static void
tde_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                  Relation relation, ReorderBufferChange *change)
{
    TdeOutputPluginState *state = ctx->output_plugin_private;
    Form_pg_class class_form = RelationGetForm(relation);
    TupleDesc tupdesc = RelationGetDescr(relation);
    MemoryContext old;

    old = MemoryContextSwitchTo(state->context);

    /*
     * Decrypt encrypted_heap tuples before any output processing.
     *
     * After this call, change->data.tp.newtuple and .oldtuple
     * contain plaintext HeapTuples (or are unchanged for non-encrypted
     * tables).
     */
    tde_logicaldec_decrypt_change(change, relation, state->encrypted_heap_oid);

    OutputPluginPrepareWrite(ctx, true);

    appendStringInfoString(ctx->out, "table ");
    appendStringInfoString(ctx->out,
        quote_qualified_identifier(
            get_namespace_name(RelationGetNamespace(relation)),
            NameStr(class_form->relname)));
    appendStringInfoChar(ctx->out, ':');

    switch (change->action)
    {
        case REORDER_BUFFER_CHANGE_INSERT:
            appendStringInfoString(ctx->out, " INSERT:");
            if (change->data.tp.newtuple != NULL)
            {
                /* TODO: format the decrypted tuple for output */
                appendStringInfoString(ctx->out, " (decrypted tuple data)");
            }
            break;

        case REORDER_BUFFER_CHANGE_UPDATE:
            appendStringInfoString(ctx->out, " UPDATE:");
            if (change->data.tp.oldtuple != NULL)
                appendStringInfoString(ctx->out, " old-key: (decrypted)");
            if (change->data.tp.newtuple != NULL)
                appendStringInfoString(ctx->out, " new-tuple: (decrypted)");
            break;

        case REORDER_BUFFER_CHANGE_DELETE:
            appendStringInfoString(ctx->out, " DELETE:");
            if (change->data.tp.oldtuple != NULL)
                appendStringInfoString(ctx->out, " (decrypted old-key)");
            else
                appendStringInfoString(ctx->out, " (no tuple data)");
            break;

        default:
            Assert(false);
    }

    OutputPluginWrite(ctx, true);

    MemoryContextSwitchTo(old);
    MemoryContextReset(state->context);
}


/*
 * TRUNCATE callback — pass through, no decryption needed.
 */
static void
tde_decode_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
                    int nrelations, Relation relations[],
                    ReorderBufferChange *change)
{
    OutputPluginPrepareWrite(ctx, true);
    appendStringInfoString(ctx->out, "TRUNCATE (pass-through)");
    OutputPluginWrite(ctx, true);
}


/*
 * Shutdown callback — cleanup.
 */
static void
tde_decode_shutdown(LogicalDecodingContext *ctx)
{
    TdeOutputPluginState *state = ctx->output_plugin_private;

    if (state != NULL)
        MemoryContextDelete(state->context);
}


/* ================================================================
 * Plugin Registration
 *
 * This function is looked up by name when the replication slot is
 * created with:  SELECT pg_create_logical_replication_slot('s', 'pg_vault_tde');
 *
 * It coexists with _PG_init in the same .so — both are valid
 * simultaneously.  PostgreSQL uses _PG_init for shared_preload_libraries
 * and _PG_output_plugin_init for logical decoding plugin loading.
 * ================================================================
 */
void
_PG_output_plugin_init(OutputPluginCallbacks *cb)
{
    cb->startup_cb  = tde_decode_startup;
    cb->begin_cb    = tde_decode_begin_txn;
    cb->change_cb   = tde_decode_change;
    cb->truncate_cb = tde_decode_truncate;
    cb->commit_cb   = tde_decode_commit_txn;
    cb->shutdown_cb = tde_decode_shutdown;
}

9. Build Integration

9.1 Same .so Approach (Recommended)

Add the new source file to the existing OBJS in the Makefile:

OBJS = \
    src/pg_vault_tde.o \
    src/kms/pg_vault_tde_kms.o \
    src/crypto/pg_vault_tde_crypto.o \
    src/crypto/pg_vault_tde_hw_accel.o \
    src/tam/pg_vault_tde_tam.o \
    src/tam/pg_vault_tde_toast.o \
    src/iam/pg_vault_tde_iam.o \
    src/backup/pg_vault_tde_backup.o \
    src/logicaldec/pg_vault_tde_logicaldec.o    # NEW

The _PG_output_plugin_init symbol is exported via PGDLLEXPORT (the function is declared extern in output_plugin.h).

9.2 No Separate _PG_init Needed

The output plugin shares the same .so as the extension. When PostgreSQL loads it via load_external_function("pg_vault_tde", "_PG_output_plugin_init", ...):

  1. Since pg_vault_tde is in shared_preload_libraries, the .so is already loaded in every backend (including the WAL sender).
  2. dlopen() returns the already-loaded handle; dlsym() finds the symbol.
  3. _PG_init has already run — shmem hooks are chained, DEK cache is live, EVP contexts are available.
  4. No second _PG_init call happens (the dynamic linker’s dlopen with RTLD_NOW on an already-opened library just bumps the refcount).

9.3 Directory Structure

src/
  logicaldec/
    logicaldec.instructions.md    # module-specific instructions
    pg_vault_tde_logicaldec.c     # output plugin implementation
src/include/
    pg_vault_tde_logicaldec.h     # (optional) header for any shared declarations

10. Key Implementation Considerations

10.1 Memory Ownership of Decrypted Tuples

The change->data.tp.newtuple and .oldtuple are allocated by the ReorderBuffer in its own memory context. When we replace them with decrypted copies, we must: 1. pfree() the original encrypted HeapTuple (it’s palloc’d, not buffer-backed) 2. Allocate the decrypted tuple in the same memory context 3. The ReorderBuffer will free our replacement tuple when it cleans up the change

10.2 TOAST Handling

The ReorderBuffer reassembles TOAST values before calling change_cb. This means: - External TOAST chunks are already detoasted and assembled inline - For encrypted_heap tables with the TOAST AM override (HEAP_TABLE_AM_OID), TOAST data is stored unencrypted (documented v1 limitation) - The inline (compressed/plain) portion of the tuple IS encrypted - After TOAST reassembly, the tuple’s user data portion is a mix of: - Encrypted inline columns - Unencrypted TOAST-fetched columns (already reassembled)

Issue: If the TOAST reassembly inserts plain varlena data into an otherwise-encrypted tuple body, the wire format [IV|CT|TAG] is disrupted.

Resolution: Since TOAST tables use HEAP_TABLE_AM_OID (not encrypted_heap), the TOAST chunks themselves are stored as plaintext. The main tuple’s inline data is encrypted. The ReorderBuffer’s ReorderBufferToastReplace() function deforms the tuple using heap_deform_tuple() — this will fail on an encrypted tuple because the datum boundaries are encrypted.

This is a critical design issue. Two possible solutions: 1. Decrypt before TOAST reassembly: Hook into the ReorderBuffer’s toast processing (not possible via the output plugin API alone) 2. Ensure TOAST columns are always external: For encrypted_heap tables, the encrypted user data portion is opaque; column-level TOAST compression cannot work. This means tables with large columns must rely on the TOAST AM override to store values externally (unencrypted).

Practical impact: For rows where ALL column data fits inline (< ~2 kB after header), no TOAST is involved and decryption-in-output-plugin works perfectly. For rows with large columns, the v1 TOAST limitation (unencrypted external storage) means TOAST reassembly receives the detoasted value from the unencrypted TOAST table, and the inline data is the encrypted portion.

Further investigation needed: We must verify how ReorderBufferToastReplace() interacts with encrypted inline data. If it calls heap_deform_tuple() on the encrypted tuple, it will read garbage for attribute boundaries. This may require the decryption to happen at the ReorderBuffer level, not the output plugin level.

10.3 DEK Security in WAL Sender

  • The WAL sender has shared memory access → can read the DEK
  • OPENSSL_cleanse is called on all DEK copies (existing pattern)
  • The on_proc_exit(tde_backend_cleanup, ...) handler fires on WAL sender exit
  • EVP contexts are per-process → the WAL sender gets its own pool

10.4 Key Rotation During Logical Decoding

If a key rotation happens while the WAL sender is streaming: - Old tuples in the WAL still use the old DEK - tde_gcm_decrypt() already handles prev_dek fallback (see crypto.c) - The decryption function tries the current DEK first, then falls back to prev_dek if GCM authentication fails - This is safe for logical decoding

10.5 Replica Identity Interaction

For UPDATE and DELETE, PostgreSQL logs the old tuple (or replica identity columns) in the WAL. For encrypted_heap tables: - The old tuple’s user data is also encrypted (it’s part of the heap page) - Both oldtuple and newtuple need decryption - The replica identity is a subset of columns, but the WAL record contains the full encrypted user data portion from t_hoff onwards - After decryption, column extraction works normally

11. PG Version Compatibility Summary

Component PG17 PG18 Guard Needed?
OutputPluginCallbacks struct Identical Identical No
_PG_output_plugin_init entry point Same Same No
ReorderBufferChange.data.tp.{old,new}tuple HeapTuple HeapTuple No
LogicalDecodingContext struct Identical Identical No
OutputPluginPrepareWrite / OutputPluginWrite Same Same No
get_table_am_oid() Available Available No
relation->rd_rel->relam Available Available No
ReorderBufferAllocTupleBuf name ReorderBufferGetTupleBuf ReorderBufferAllocTupleBuf No (not used by plugins)

Conclusion: Zero #if PG_VERSION_NUM guards required for logical decoding code.

12. Open Questions / Further Research

  1. TOAST interaction: Must verify whether ReorderBufferToastReplace() calls heap_deform_tuple() on the still-encrypted tuple. If so, the output plugin approach is insufficient — decryption must happen earlier in the pipeline (possibly by hooking into the ReorderBuffer layer, which would require a different architectural approach).

  2. Binary protocol: For production logical replication (not just test output), the plugin should support OUTPUT_PLUGIN_BINARY_OUTPUT and use logicalrep_write_* functions from replication/logicalproto.h.

  3. Initial table sync: The subscriber’s initial COPY uses table_scan_getnextslot which DOES invoke the TAM (correctly decrypts). No output plugin involvement needed for initial sync.

  4. Streaming transactions: For large transactions that stream changes before COMMIT, the stream_change_cb also needs decryption. This is the same decrypt logic but registered on a different callback.

  5. ALTER TABLE … SET ACCESS METHOD: If a table is changed from encrypted_heap to heap (or vice versa), the cached AM OID check remains correct because we check relation->rd_rel->relam per-change, which reflects the relation’s state at the time the WAL record was written.

13. References

File Location Purpose
output_plugin.h /usr/include/postgresql/{17,18}/server/replication/output_plugin.h Callback typedefs and OutputPluginCallbacks struct
logical.h /usr/include/postgresql/{17,18}/server/replication/logical.h LogicalDecodingContext struct
reorderbuffer.h /usr/include/postgresql/{17,18}/server/replication/reorderbuffer.h ReorderBufferChange, HeapTuple in data.tp
pgoutput.c src/backend/replication/pgoutput/pgoutput.c (PG source) Reference implementation of a production output plugin
test_decoding.c contrib/test_decoding/test_decoding.c (PG source) Simple reference for output plugin + _PG_init coexistence
decode.c src/backend/replication/logical/decode.c (PG source) WAL record → ReorderBufferChange conversion
defrem.h /usr/include/postgresql/18/server/commands/defrem.h get_table_am_oid() declaration
pg_class.h /usr/include/postgresql/18/server/catalog/pg_class.h relam field definition