Updates and Fixes

docs/README.md

@@ -60,6 +60,14 @@ Changes on the pgactive nodes are applied by a background worker. There are a nu
 
 All standard PostgreSQL authentication mechanisms can be utilized. This would include utilizing `pg_user_mapping` to map system users to PostgreSQL users and `.pgpass` to provide the password in a password file as opposed to the connection string.
 
+#### Apply-as-Table-Owner
+
+pgactive apply workers switch to the table owner before executing replicated DML
+(INSERT, UPDATE, DELETE) when `pgactive.apply_as_table_owner` is enabled (the
+default). This uses PostgreSQL's `SwitchToUntrustedUser`/`RestoreUserContext` API
+on PG 16+ and an equivalent shim on older versions, matching the upstream logical
+replication behavior introduced in PG 16 (commit `1e10d49b65d`).
+
 #### Connections
 
 Connections can use SSL to secure the data in flight.

meson.build

@@ -7,7 +7,7 @@ project(
     check: true,
   ).stdout().strip(),
   license: 'Apache-2.0',
-  meson_version: '>= 0.58.0',
+  meson_version: '>= 0.62.0',
   default_options: [
     'warning_level=2',
   ],
@@ -59,11 +59,13 @@ libpq = declare_dependency(
   ],
 )
 
-# pgfeutils (needed by pgactive_dump)
+# Static support libs from the PG install (needed by pgactive_dump and
+# pgactive_init_copy). Required: a missing one is a broken install, and failing
+# here gives a clear error instead of a confusing undefined-reference link error.
 cc = meson.get_compiler('c')
-libpgfeutils = cc.find_library('pgfeutils', dirs: [pg_libdir], required: false)
-libpgcommon  = cc.find_library('pgcommon',  dirs: [pg_libdir], required: false)
-libpgport    = cc.find_library('pgport',    dirs: [pg_libdir], required: false)
+libpgfeutils = cc.find_library('pgfeutils', dirs: [pg_libdir], required: true)
+libpgcommon  = cc.find_library('pgcommon',  dirs: [pg_libdir], required: true)
+libpgport    = cc.find_library('pgport',    dirs: [pg_libdir], required: true)
 
 # ---- Version header generation ----
 git = find_program('git', required: false)
@@ -78,7 +80,7 @@ endif
 
 version_date = run_command('date', '+%Y-%m-%d', check: true).stdout().strip()
 
-version_h = configure_file(
+configure_file(
   input: 'include/pgactive_version.h.in',
   output: 'pgactive_version.h',
   configuration: {

pgactive--2.1.8--2.1.9.sql

@@ -7,6 +7,21 @@ SET pgactive.skip_ddl_replication = true;
 SET LOCAL search_path = pgactive;
 -- Start Upgrade SQLs/Functions/Procedures
 
+REVOKE ALL ON FUNCTION pgactive_version() FROM public;
+REVOKE ALL ON FUNCTION pgactive_is_active_in_db() FROM public;
+REVOKE ALL ON FUNCTION pgactive_is_apply_paused() FROM public;
+REVOKE ALL ON FUNCTION pgactive_get_local_node_name() FROM public;
+REVOKE ALL ON FUNCTION pgactive_get_global_locks_info() FROM public;
+REVOKE ALL ON FUNCTION pgactive_get_last_applied_xact_info(text, oid, oid) FROM public;
+REVOKE ALL ON FUNCTION pgactive_get_replication_lag_info() FROM public;
+REVOKE ALL ON FUNCTION pgactive_get_table_replication_sets(regclass) FROM public;
+REVOKE ALL ON FUNCTION pgactive_set_table_replication_sets(regclass, boolean) FROM public;
+REVOKE ALL ON FUNCTION pgactive_include_table_replication_set(regclass) FROM public;
+REVOKE ALL ON FUNCTION pgactive_exclude_table_replication_set(regclass) FROM public;
+REVOKE ALL ON FUNCTION pgactive_create_conflict_handler(regclass, name, regprocedure, pgactive.pgactive_conflict_type, interval) FROM public;
+REVOKE ALL ON FUNCTION pgactive_drop_conflict_handler(regclass, name) FROM public;
+REVOKE ALL ON FUNCTION pgactive_replicate_ddl_command(text) FROM public;
+
 -- Finish Upgrade SQLs/Functions/Procedures
 RESET pgactive.skip_ddl_replication;
 RESET search_path;

pgactive--2.1.9.sql

@@ -22,6 +22,7 @@ CREATE FUNCTION pgactive_version()
 RETURNS TEXT
 AS 'MODULE_PATHNAME'
 LANGUAGE C;
+REVOKE ALL ON FUNCTION pgactive_version() FROM public;
 
 CREATE FUNCTION pgactive_variant()
 RETURNS TEXT
@@ -88,11 +89,13 @@ CREATE FUNCTION pgactive_create_conflict_handler (
 RETURNS VOID
 AS 'MODULE_PATHNAME'
 LANGUAGE C;
+REVOKE ALL ON FUNCTION pgactive_create_conflict_handler(regclass, name, regprocedure, pgactive.pgactive_conflict_type, interval) FROM public;
 
 CREATE FUNCTION pgactive_drop_conflict_handler(ch_rel REGCLASS, ch_name NAME)
 RETURNS VOID
 AS 'MODULE_PATHNAME'
 LANGUAGE C STRICT;
+REVOKE ALL ON FUNCTION pgactive_drop_conflict_handler(regclass, name) FROM public;
 
 CREATE TYPE pgactive_conflict_resolution AS ENUM (
     'conflict_trigger_skip_change',
@@ -265,6 +268,7 @@ CREATE FUNCTION pgactive_replicate_ddl_command(cmd TEXT)
 RETURNS VOID
 AS 'MODULE_PATHNAME'
 LANGUAGE C;
+REVOKE ALL ON FUNCTION pgactive_replicate_ddl_command(text) FROM public;
 
 CREATE FUNCTION pgactive_truncate_trigger_add()
 RETURNS event_trigger
@@ -324,6 +328,7 @@ CREATE FUNCTION pgactive_get_table_replication_sets(relation regclass, OUT sets
                 ), '["default"]'))
         )|| '{all}';
   $$;
+REVOKE ALL ON FUNCTION pgactive_get_table_replication_sets(regclass) FROM public;
 
 CREATE TABLE pgactive_replication_set_config (
     set_name name PRIMARY KEY,
@@ -1358,6 +1363,7 @@ WHERE n.node_sysid = i.sysid
   AND n.node_timeline = i.timeline
   AND n.node_dboid = i.dboid;
 $$;
+REVOKE ALL ON FUNCTION pgactive_get_local_node_name() FROM public;
 
 COMMENT ON FUNCTION pgactive_get_local_node_name() IS
 'Return the name from pgactive.pgactive_nodes for the local node, or null if no entry exists';
@@ -1366,6 +1372,7 @@ CREATE FUNCTION pgactive_is_apply_paused()
 RETURNS boolean
 AS 'MODULE_PATHNAME'
 LANGUAGE C;
+REVOKE ALL ON FUNCTION pgactive_is_apply_paused() FROM public;
 
 CREATE FUNCTION pgactive_set_node_read_only (node_name text, read_only boolean)
 RETURNS void
@@ -1688,6 +1695,7 @@ CREATE FUNCTION pgactive_is_active_in_db()
 RETURNS boolean
 AS 'MODULE_PATHNAME','pgactive_is_active_in_db'
 LANGUAGE C;
+REVOKE ALL ON FUNCTION pgactive_is_active_in_db() FROM public;
 
 CREATE EVENT TRIGGER pgactive_truncate_trigger_add
 ON ddl_command_end
@@ -1837,6 +1845,7 @@ CREATE FUNCTION pgactive_get_global_locks_info (
 RETURNS record
 AS 'MODULE_PATHNAME', 'pgactive_get_global_locks_info'
 LANGUAGE C VOLATILE;
+REVOKE ALL ON FUNCTION pgactive_get_global_locks_info() FROM public;
 
 COMMENT ON FUNCTION pgactive_get_global_locks_info() IS
 'Backing function for pgactive_global_locks_info view';
@@ -2129,6 +2138,7 @@ CREATE FUNCTION pgactive_get_last_applied_xact_info(
 RETURNS record
 AS 'MODULE_PATHNAME'
 LANGUAGE C STRICT;
+REVOKE ALL ON FUNCTION pgactive_get_last_applied_xact_info(text, oid, oid) FROM public;
 
 COMMENT ON FUNCTION pgactive_get_last_applied_xact_info(text, oid, oid) IS
 'Gets last applied transaction info of apply worker for a given node.';
@@ -2149,6 +2159,7 @@ CREATE FUNCTION pgactive_get_replication_lag_info(
 RETURNS SETOF record
 AS 'MODULE_PATHNAME'
 LANGUAGE C VOLATILE STRICT;
+REVOKE ALL ON FUNCTION pgactive_get_replication_lag_info() FROM public;
 
 COMMENT ON FUNCTION pgactive_get_replication_lag_info() IS
 'Gets replication lag info.';
@@ -2777,6 +2788,7 @@ BEGIN
 	END IF;
 END;
 $$;
+REVOKE ALL ON FUNCTION pgactive_set_table_replication_sets(regclass, boolean) FROM public;
 
 
 CREATE OR REPLACE FUNCTION pgactive_exclude_table_replication_set(p_relation regclass)
@@ -2791,6 +2803,7 @@ BEGIN
 	PERFORM pgactive.pgactive_set_table_replication_sets(p_relation, true);
 END;
 $$;
+REVOKE ALL ON FUNCTION pgactive_exclude_table_replication_set(regclass) FROM public;
 
 
 CREATE OR REPLACE FUNCTION pgactive_include_table_replication_set(p_relation regclass)
@@ -2805,6 +2818,7 @@ BEGIN
 	PERFORM pgactive.pgactive_set_table_replication_sets(p_relation, false);
 END;
 $$;
+REVOKE ALL ON FUNCTION pgactive_include_table_replication_set(regclass) FROM public;
 
 DROP FUNCTION pgactive_get_connection_replication_sets(
     text[],