Reset the logical worker type while cleaning up other worker info.

Amit Kapila · Amit Kapila · commit 9c13b6814ac7 · 2023-08-25T08:57:55.000+05:30
Commit 2a8b40e introduces the worker type field for logical replication workers, but forgot to reset the type when the worker exits. This can lead to recognizing a stopped worker as a valid logical replication worker. Fix it by resetting the worker type and additionally adding the safeguard to not use LogicalRepWorker until ->in_use is verified. Reported-by: Thomas Munro based on cfbot reports. Author: Hou Zhijie, Alvaro Herrera Reviewed-by: Amit Kapila Discussion: http://postgr.es/m/CA+hUKGK2RQh4LifVgBmkHsCYChP-65UwGXOmnCzYVa5aAt4GWg@mail.gmail.com
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
@@ -793,6 +793,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)
 {
 	Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_EXCLUSIVE));
 
+	worker->type = WORKERTYPE_UNKNOWN;
 	worker->in_use = false;
 	worker->proc = NULL;
 	worker->dbid = InvalidOid;
@@ -862,7 +863,7 @@ logicalrep_sync_worker_count(Oid subid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (w->subid == subid && isTablesyncWorker(w))
+		if (isTablesyncWorker(w) && w->subid == subid)
 			res++;
 	}
 
@@ -889,7 +890,7 @@ logicalrep_pa_worker_count(Oid subid)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
-		if (w->subid == subid && isParallelApplyWorker(w))
+		if (isParallelApplyWorker(w) && w->subid == subid)
 			res++;
 	}
 
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
@@ -327,8 +327,10 @@ extern void pa_decr_and_wait_stream_block(void);
 extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
 						   XLogRecPtr remote_lsn);
 
-#define isParallelApplyWorker(worker) ((worker)->type == WORKERTYPE_PARALLEL_APPLY)
-#define isTablesyncWorker(worker) ((worker)->type == WORKERTYPE_TABLESYNC)
+#define isParallelApplyWorker(worker) ((worker)->in_use && \
+									   (worker)->type == WORKERTYPE_PARALLEL_APPLY)
+#define isTablesyncWorker(worker) ((worker)->in_use && \
+								   (worker)->type == WORKERTYPE_TABLESYNC)
 
 static inline bool
 am_tablesync_worker(void)
@@ -339,12 +341,14 @@ am_tablesync_worker(void)
 static inline bool
 am_leader_apply_worker(void)
 {
+	Assert(MyLogicalRepWorker->in_use);
 	return (MyLogicalRepWorker->type == WORKERTYPE_APPLY);
 }
 
 static inline bool
 am_parallel_apply_worker(void)
 {
+	Assert(MyLogicalRepWorker->in_use);
 	return isParallelApplyWorker(MyLogicalRepWorker);
 }
 

Original file line number	Diff line number	Diff line change
`@@ -793,6 +793,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)`
`793`	`793`	`{`
`794`	`794`	`Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_EXCLUSIVE));`
`795`	`795`
	`796`	`+ worker->type = WORKERTYPE_UNKNOWN;`
`796`	`797`	`worker->in_use = false;`
`797`	`798`	`worker->proc = NULL;`
`798`	`799`	`worker->dbid = InvalidOid;`
`@@ -862,7 +863,7 @@ logicalrep_sync_worker_count(Oid subid)`
`862`	`863`	`{`
`863`	`864`	`LogicalRepWorker *w = &LogicalRepCtx->workers[i];`
`864`	`865`
`865`		`- if (w->subid == subid && isTablesyncWorker(w))`
	`866`	`+ if (isTablesyncWorker(w) && w->subid == subid)`
`866`	`867`	`res++;`
`867`	`868`	`}`
`868`	`869`
`@@ -889,7 +890,7 @@ logicalrep_pa_worker_count(Oid subid)`
`889`	`890`	`{`
`890`	`891`	`LogicalRepWorker *w = &LogicalRepCtx->workers[i];`
`891`	`892`
`892`		`- if (w->subid == subid && isParallelApplyWorker(w))`
	`893`	`+ if (isParallelApplyWorker(w) && w->subid == subid)`
`893`	`894`	`res++;`
`894`	`895`	`}`
`895`	`896`
Original file line number	Diff line number	Diff line change
`@@ -327,8 +327,10 @@ extern void pa_decr_and_wait_stream_block(void);`
`327`	`327`	`extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,`
`328`	`328`	`XLogRecPtr remote_lsn);`
`329`	`329`
`330`		`-#define isParallelApplyWorker(worker) ((worker)->type == WORKERTYPE_PARALLEL_APPLY)`
`331`		`-#define isTablesyncWorker(worker) ((worker)->type == WORKERTYPE_TABLESYNC)`
	`330`	`+#define isParallelApplyWorker(worker) ((worker)->in_use && \`
	`331`	`+ (worker)->type == WORKERTYPE_PARALLEL_APPLY)`
	`332`	`+#define isTablesyncWorker(worker) ((worker)->in_use && \`
	`333`	`+ (worker)->type == WORKERTYPE_TABLESYNC)`
`332`	`334`
`333`	`335`	`static inline bool`
`334`	`336`	`am_tablesync_worker(void)`
`@@ -339,12 +341,14 @@ am_tablesync_worker(void)`
`339`	`341`	`static inline bool`
`340`	`342`	`am_leader_apply_worker(void)`
`341`	`343`	`{`
	`344`	`+ Assert(MyLogicalRepWorker->in_use);`
`342`	`345`	`return (MyLogicalRepWorker->type == WORKERTYPE_APPLY);`
`343`	`346`	`}`
`344`	`347`
`345`	`348`	`static inline bool`
`346`	`349`	`am_parallel_apply_worker(void)`
`347`	`350`	`{`
	`351`	`+ Assert(MyLogicalRepWorker->in_use);`
`348`	`352`	`return isParallelApplyWorker(MyLogicalRepWorker);`
`349`	`353`	`}`
`350`	`354`