Add seq_sync_error_count to subscription statistics.

vigneshwaran-c · Commitfest Bot · commit cc3ac5aeec68 · 2025-10-30T17:05:32.000Z
This commit introduces a new column seq_sync_error_count to subscription
statistics. The new field tracks the number of errors encountered during
sequence synchronization for each subscription.
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
@@ -2193,6 +2193,15 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       </para></entry>
      </row>
 
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>sequence_sync_error_count</structfield> <type>bigint</type>
+      </para>
+      <para>
+       Number of times an error occurred during the sequence synchronization
+      </para></entry>
+     </row>
+
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>sync_error_count</structfield> <type>bigint</type>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
@@ -1415,6 +1415,7 @@ CREATE VIEW pg_stat_subscription_stats AS
         ss.subid,
         s.subname,
         ss.apply_error_count,
+        ss.seq_sync_error_count,
         ss.sync_error_count,
         ss.confl_insert_exists,
         ss.confl_update_origin_differs,
diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c
@@ -670,6 +670,9 @@ start_sequence_sync()
 			 * idle state.
 			 */
 			AbortOutOfAnyTransaction();
+			pgstat_report_subscription_error(MySubscription->oid,
+											 WORKERTYPE_SEQUENCESYNC);
+
 			PG_RE_THROW();
 		}
 	}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
@@ -1530,7 +1530,8 @@ start_table_sync(XLogRecPtr *origin_startpos, char **slotname)
 			 * idle state.
 			 */
 			AbortOutOfAnyTransaction();
-			pgstat_report_subscription_error(MySubscription->oid, false);
+			pgstat_report_subscription_error(MySubscription->oid,
+											 WORKERTYPE_TABLESYNC);
 
 			PG_RE_THROW();
 		}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
@@ -5607,7 +5607,7 @@ start_apply(XLogRecPtr origin_startpos)
 			 */
 			AbortOutOfAnyTransaction();
 			pgstat_report_subscription_error(MySubscription->oid,
-											 !am_tablesync_worker());
+											 MyLogicalRepWorker->type);
 
 			PG_RE_THROW();
 		}
@@ -5954,15 +5954,12 @@ DisableSubscriptionAndExit(void)
 
 	RESUME_INTERRUPTS();
 
-	if (am_leader_apply_worker() || am_tablesync_worker())
-	{
-		/*
-		 * Report the worker failed during either table synchronization or
-		 * apply.
-		 */
-		pgstat_report_subscription_error(MyLogicalRepWorker->subid,
-										 !am_tablesync_worker());
-	}
+	/*
+	 * Report the worker failed during either sequence synchronization or
+	 * table synchronization or apply.
+	 */
+	pgstat_report_subscription_error(MyLogicalRepWorker->subid,
+									 MyLogicalRepWorker->type);
 
 	/* Disable the subscription */
 	StartTransactionCommand();
diff --git a/src/backend/utils/activity/pgstat_subscription.c b/src/backend/utils/activity/pgstat_subscription.c
@@ -17,14 +17,15 @@
 
 #include "postgres.h"
 
+#include "replication/worker_internal.h"
 #include "utils/pgstat_internal.h"
 
 
 /*
  * Report a subscription error.
  */
 void
-pgstat_report_subscription_error(Oid subid, bool is_apply_error)
+pgstat_report_subscription_error(Oid subid, LogicalRepWorkerType wtype)
 {
 	PgStat_EntryRef *entry_ref;
 	PgStat_BackendSubEntry *pending;
@@ -33,10 +34,25 @@ pgstat_report_subscription_error(Oid subid, bool is_apply_error)
 										  InvalidOid, subid, NULL);
 	pending = entry_ref->pending;
 
-	if (is_apply_error)
-		pending->apply_error_count++;
-	else
-		pending->sync_error_count++;
+	switch (wtype)
+	{
+		case WORKERTYPE_APPLY:
+			pending->apply_error_count++;
+			break;
+
+		case WORKERTYPE_SEQUENCESYNC:
+			pending->seq_sync_error_count++;
+			break;
+
+		case WORKERTYPE_TABLESYNC:
+			pending->sync_error_count++;
+			break;
+
+		default:
+			/* Should never happen. */
+			Assert(0);
+			break;
+	}
 }
 
 /*
@@ -115,6 +131,7 @@ pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait)
 
 #define SUB_ACC(fld) shsubent->stats.fld += localent->fld
 	SUB_ACC(apply_error_count);
+	SUB_ACC(seq_sync_error_count);
 	SUB_ACC(sync_error_count);
 	for (int i = 0; i < CONFLICT_NUM_TYPES; i++)
 		SUB_ACC(conflict_count[i]);
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
@@ -2203,7 +2203,7 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS	12
+#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS	13
 	Oid			subid = PG_GETARG_OID(0);
 	TupleDesc	tupdesc;
 	Datum		values[PG_STAT_GET_SUBSCRIPTION_STATS_COLS] = {0};
@@ -2221,25 +2221,27 @@ pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
 					   OIDOID, -1, 0);
 	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "apply_error_count",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "sync_error_count",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "seq_sync_error_count",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "confl_insert_exists",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "sync_error_count",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "confl_update_origin_differs",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 5, "confl_insert_exists",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "confl_update_exists",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 6, "confl_update_origin_differs",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "confl_update_deleted",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 7, "confl_update_exists",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "confl_update_missing",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 8, "confl_update_deleted",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "confl_delete_origin_differs",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 9, "confl_update_missing",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "confl_delete_missing",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 10, "confl_delete_origin_differs",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 11, "confl_multiple_unique_conflicts",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 11, "confl_delete_missing",
 					   INT8OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 12, "stats_reset",
+	TupleDescInitEntry(tupdesc, (AttrNumber) 12, "confl_multiple_unique_conflicts",
+					   INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 13, "stats_reset",
 					   TIMESTAMPTZOID, -1, 0);
 	BlessTupleDesc(tupdesc);
 
@@ -2256,6 +2258,9 @@ pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
 	/* apply_error_count */
 	values[i++] = Int64GetDatum(subentry->apply_error_count);
 
+	/* seq_sync_error_count */
+	values[i++] = Int64GetDatum(subentry->seq_sync_error_count);
+
 	/* sync_error_count */
 	values[i++] = Int64GetDatum(subentry->sync_error_count);
 
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
@@ -5704,9 +5704,9 @@
 { oid => '6231', descr => 'statistics: information about subscription stats',
   proname => 'pg_stat_get_subscription_stats', provolatile => 's',
   proparallel => 'r', prorettype => 'record', proargtypes => 'oid',
-  proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}',
-  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{subid,subid,apply_error_count,sync_error_count,confl_insert_exists,confl_update_origin_differs,confl_update_exists,confl_update_deleted,confl_update_missing,confl_delete_origin_differs,confl_delete_missing,confl_multiple_unique_conflicts,stats_reset}',
+  proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{subid,subid,apply_error_count,seq_sync_error_count,sync_error_count,confl_insert_exists,confl_update_origin_differs,confl_update_exists,confl_update_deleted,confl_update_missing,confl_delete_origin_differs,confl_delete_missing,confl_multiple_unique_conflicts,stats_reset}',
   prosrc => 'pg_stat_get_subscription_stats' },
 { oid => '6118', descr => 'statistics: information about subscription',
   proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
@@ -16,6 +16,7 @@
 #include "portability/instr_time.h"
 #include "postmaster/pgarch.h"	/* for MAX_XFN_CHARS */
 #include "replication/conflict.h"
+#include "replication/worker_internal.h"
 #include "utils/backend_progress.h" /* for backward compatibility */	/* IWYU pragma: export */
 #include "utils/backend_status.h"	/* for backward compatibility */	/* IWYU pragma: export */
 #include "utils/pgstat_kind.h"
@@ -108,6 +109,7 @@ typedef struct PgStat_FunctionCallUsage
 typedef struct PgStat_BackendSubEntry
 {
 	PgStat_Counter apply_error_count;
+	PgStat_Counter seq_sync_error_count;
 	PgStat_Counter sync_error_count;
 	PgStat_Counter conflict_count[CONFLICT_NUM_TYPES];
 } PgStat_BackendSubEntry;
@@ -416,6 +418,7 @@ typedef struct PgStat_SLRUStats
 typedef struct PgStat_StatSubEntry
 {
 	PgStat_Counter apply_error_count;
+	PgStat_Counter seq_sync_error_count;
 	PgStat_Counter sync_error_count;
 	PgStat_Counter conflict_count[CONFLICT_NUM_TYPES];
 	TimestampTz stat_reset_timestamp;
@@ -769,7 +772,8 @@ extern PgStat_SLRUStats *pgstat_fetch_slru(void);
  * Functions in pgstat_subscription.c
  */
 
-extern void pgstat_report_subscription_error(Oid subid, bool is_apply_error);
+extern void pgstat_report_subscription_error(Oid subid,
+											 LogicalRepWorkerType wtype);
 extern void pgstat_report_subscription_conflict(Oid subid, ConflictType type);
 extern void pgstat_create_subscription(Oid subid);
 extern void pgstat_drop_subscription(Oid subid);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
@@ -2191,6 +2191,7 @@ pg_stat_subscription| SELECT su.oid AS subid,
 pg_stat_subscription_stats| SELECT ss.subid,
     s.subname,
     ss.apply_error_count,
+    ss.seq_sync_error_count,
     ss.sync_error_count,
     ss.confl_insert_exists,
     ss.confl_update_origin_differs,
@@ -2202,7 +2203,7 @@ pg_stat_subscription_stats| SELECT ss.subid,
     ss.confl_multiple_unique_conflicts,
     ss.stats_reset
    FROM pg_subscription s,
-    LATERAL pg_stat_get_subscription_stats(s.oid) ss(subid, apply_error_count, sync_error_count, confl_insert_exists, confl_update_origin_differs, confl_update_exists, confl_update_deleted, confl_update_missing, confl_delete_origin_differs, confl_delete_missing, confl_multiple_unique_conflicts, stats_reset);
+    LATERAL pg_stat_get_subscription_stats(s.oid) ss(subid, apply_error_count, seq_sync_error_count, sync_error_count, confl_insert_exists, confl_update_origin_differs, confl_update_exists, confl_update_deleted, confl_update_missing, confl_delete_origin_differs, confl_delete_missing, confl_multiple_unique_conflicts, stats_reset);
 pg_stat_sys_indexes| SELECT relid,
     indexrelid,
     schemaname,
diff --git a/src/test/subscription/t/026_stats.pl b/src/test/subscription/t/026_stats.pl

Original file line number	Diff line number	Diff line change
`@@ -670,6 +670,9 @@ start_sequence_sync()`
`670`	`670`	`* idle state.`
`671`	`671`	`*/`
`672`	`672`	`AbortOutOfAnyTransaction();`
	`673`	`+ pgstat_report_subscription_error(MySubscription->oid,`
	`674`	`+ WORKERTYPE_SEQUENCESYNC);`
	`675`	`+`
`673`	`676`	`PG_RE_THROW();`
`674`	`677`	`}`
`675`	`678`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1530,7 +1530,8 @@ start_table_sync(XLogRecPtr origin_startpos, char *slotname)`
`1530`	`1530`	`* idle state.`
`1531`	`1531`	`*/`
`1532`	`1532`	`AbortOutOfAnyTransaction();`
`1533`		`- pgstat_report_subscription_error(MySubscription->oid, false);`
	`1533`	`+ pgstat_report_subscription_error(MySubscription->oid,`
	`1534`	`+ WORKERTYPE_TABLESYNC);`
`1534`	`1535`
`1535`	`1536`	`PG_RE_THROW();`
`1536`	`1537`	`}`