Make tablesync worker exit when apply dies while it was waiting for it

This avoids "orphaned" sync workers.

This was caused by a thinko in wait_for_sync_status_change.

Author: Petr Jelinek <petr.jelinek@2ndquadrant.com>
Reported-by: Masahiko Sawada <sawada.mshk@gmail.com>

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 85e480db4bdd6c95eceafac6f42a9ce0b8112d19..6e268f3521dba4fec25976c09030a183e04a9b37 100644 (file)
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -146,7 +146,12 @@ finish_sync_worker(void)
 /*
  * Wait until the table synchronization change.
  *
- * Returns false if the relation subscription state disappeared.
+ * If called from apply worker, it will wait for the synchronization worker to
+ * change table state in shmem.  If called from synchronization worker, it
+ * will wait for apply worker to change table state in shmem.
+ *
+ * Returns false if the opposite worker has disappeared or the table state has
+ * been reset.
  */
 static bool
 wait_for_sync_status_change(Oid relid, char origstate)
@@ -161,14 +166,27 @@ wait_for_sync_status_change(Oid relid, char origstate)
        CHECK_FOR_INTERRUPTS();
 
        LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+       /* Check if the opposite worker is still running and bail if not. */
        worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
-                                       relid, false);
+                                       am_tablesync_worker() ? InvalidOid : relid,
+                                       false);
        if (!worker)
        {
            LWLockRelease(LogicalRepWorkerLock);
            return false;
        }
+
+       /*
+        * If I'm the synchronization worker, look at my own state.  Otherwise
+        * look at the state of the synchronization worker we found above.
+        */
+       if (am_tablesync_worker())
+           worker = MyLogicalRepWorker;
+
+       Assert(worker->relid == relid);
        state = worker->relstate;
+
        LWLockRelease(LogicalRepWorkerLock);
 
        if (state == SUBREL_STATE_UNKNOWN)
@@ -179,7 +197,7 @@ wait_for_sync_status_change(Oid relid, char origstate)
 
        rc = WaitLatch(&MyProc->procLatch,
                       WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
-                      10000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
+                      1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
 
        /* emergency bailout if postmaster has died */
        if (rc & WL_POSTMASTER_DEATH)