[SPR-842]: Fix deadlock between log reader and copy thread

include/pg_log_mgr/pg_log_mgr.hh

@@ -236,7 +236,16 @@ namespace springtail::pg_log_mgr {
         void _copy_thread();
 
         /** Process copy table results; insert into redis */
-        void _process_copy_results(const std::vector<PgCopyResultPtr> &res);
+        bool _process_copy_results(const std::vector<PgCopyResultPtr> &res);
+
+        /**
+         * @brief Pick copy table request from the redis queue
+         *
+         * @param timeout timeout in seconds (0 = use try_pop)
+         *
+         * @return pair<table_id, optional<XidLsn of the copy table>>
+         */
+        std::pair<uint32_t, std::optional<XidLsn>> _get_copy_table_ids(uint32_t timeout=0);
 
         /** Redis cache callback for watching database state change */
         RedisCache::RedisChangeWatcherPtr _cache_watcher_db_states;

include/pg_log_mgr/sync_tracker.hh

@@ -142,6 +142,15 @@ namespace springtail::pg_log_mgr {
          */
         SkipDetails should_skip(uint64_t db_id, uint64_t table_id, uint32_t pg_xid) const;
 
+        /**
+         * @brief Record PG_XID at which table is picked for resync
+         *
+         * @param db_id     Database ID
+         * @param table_id  Table ID
+         * @param xid       PG XID/LSN which will be marked
+         */
+        void pick_table_for_sync(uint64_t db_id, uint64_t table_id, const XidLsn &xid);
+
     protected:
         SyncTracker() : Singleton<SyncTracker>(ServiceId::SyncTrackerId) {}
         virtual ~SyncTracker() override = default;
@@ -269,6 +278,17 @@ namespace springtail::pg_log_mgr {
         /** Used to track all of the table syncs operating at a given snapshot XID. */
         DbMap<PgXidMap<std::shared_ptr<XidRecord>>> _sync_map;
 
+        /** db-> table indicating that a resync was issued but it hasn't been picked up by the copy
+          thread yet. */
+        using TableSyncXidsMap = std::map<uint64_t, std::set<XidLsn>>;
+        using DbTableSyncXidsMap = std::map<uint64_t, TableSyncXidsMap>;
+        DbTableSyncXidsMap _resync_map;
+
+        /** db -> table -> xid indicating the table @ XID is selected for resync, yet to in-flight*/
+        using TableSyncPickedXidMap = std::map<uint64_t, XidLsn>;
+        using DbTableSyncPickedXidMap = std::map<uint64_t, TableSyncPickedXidMap>;
+        DbTableSyncPickedXidMap _resync_picked_map;
+
         /** Entry is added here when a copy for the table is in-flight but hasn't completed. */
         DbMap<TableMap<std::shared_ptr<Inflight>>> _inflight_map;
 

include/pg_repl/pg_copy_table.hh

@@ -177,6 +177,16 @@ namespace springtail
          */
         void _reset_schema();
 
+        /**
+         * @brief Check if table is dropped in the primary
+         *
+         * @param schema_oid  Schema OID in which table is present
+         * @param table_oid   Table OID
+         *
+         * @return bool indicating if table is dropped or not
+         */
+        bool _is_table_dropped(uint64_t schema_oid, uint64_t table_oid);
+
         /**
          * @brief Extract schema from table and store in internal _schema object
          * @details Uses atttypid from pg_attribute table for identifier of the type.

include/redis/redis_containers.hh

@@ -89,6 +89,17 @@ namespace springtail {
             return nullptr;
         }
 
+        /**
+         * @brief Peek an item from the queue, doesnt move/remove it
+         */
+        std::shared_ptr<T> peek() const {
+            auto res = _redis->lindex(_key, -1);
+            if (res) {
+                return std::make_shared<T>(std::move(*res));
+            }
+            return nullptr;
+        }
+
         /**
          * @brief Try to pop an item from queue (list).
          *

src/pg_log_mgr/pg_log_mgr.cc

@@ -304,20 +304,20 @@ namespace springtail::pg_log_mgr {
         while (!_shutdown) {
             std::set<uint32_t> table_ids;
 
-            // block on redis table sync queue w/timeout for shutdown
-            auto request = _redis_sync_queue.pop(REDIS_WORKER_ID, constant::COORDINATOR_KEEP_ALIVE_TIMEOUT);
-            if (request == nullptr) {
-                continue; // timeout, check for shutdown
+            // _get_copy_table_ids block on redis table sync queue w/timeout for shutdown
+            auto [next_table_id, next_xid_lsn] = _get_copy_table_ids(constant::COORDINATOR_KEEP_ALIVE_TIMEOUT);
+            if (next_table_id == -1) {
+                continue; // check for shutdown
             }
-
             do {
                 // populate the tables to copy; check for more work
-                LOG_DEBUG(LOG_PG_LOG_MGR, "Table sync queue: {}@{}:{}", request->table_id(),
-                                    request->xid().xid, request->xid().lsn);
-                table_ids.insert(request->table_id());
+                table_ids.insert(next_table_id);
+
+                // Mark table's XID to be processed for resync
+                SyncTracker::get_instance()->pick_table_for_sync(_db_id, next_table_id, next_xid_lsn.value());
 
-                request = _redis_sync_queue.try_pop(REDIS_WORKER_ID);
-            } while (request != nullptr);
+                std::tie(next_table_id, next_xid_lsn) = _get_copy_table_ids();
+            } while (next_table_id != -1);
 
             CHECK(!table_ids.empty());
 
@@ -337,6 +337,24 @@ namespace springtail::pg_log_mgr {
         }
     }
 
+    std::pair<uint32_t, std::optional<XidLsn>>
+    PgLogMgr::_get_copy_table_ids(uint32_t timeout) {
+        uint32_t table_id = -1;
+        std::shared_ptr<TableSyncRequest> request;
+        if (timeout > 0) {
+            request = _redis_sync_queue.pop(REDIS_WORKER_ID, timeout);
+        } else {
+            request = _redis_sync_queue.try_pop(REDIS_WORKER_ID);
+        }
+        if (request != nullptr) {
+            // populate the tables to copy; check for more work
+            LOG_DEBUG(LOG_PG_LOG_MGR, "Table sync queue: {}@{}:{}", request->table_id(),
+                    request->xid().xid, request->xid().lsn);
+            return std::make_pair(request->table_id(), request->xid());
+        }
+        return std::make_pair(table_id, std::nullopt);
+    }
+
     void
     PgLogMgr::_do_table_copies(std::optional<std::set<uint32_t>> table_ids)
     {
@@ -348,6 +366,9 @@ namespace springtail::pg_log_mgr {
         // notify xact handler to rollover log
         _notify_xact_start_sync();
 
+        // ensure the pipeline was stalled before we complete
+        _internal_state.wait_for_state(STATE_SYNCING);
+
         // copy tables
         LOG_DEBUG(LOG_PG_LOG_MGR, "Copying tables for db {}; state=synchronizing", _db_id);
         std::vector<PgCopyResultPtr> res;
@@ -361,19 +382,37 @@ namespace springtail::pg_log_mgr {
             res = PgCopyTable::copy_db(_db_id, xid);
         }
 
-        // ensure the pipeline was stalled before we complete
-        _internal_state.wait_for_state(STATE_SYNCING);
-
         LOG_DEBUG(LOG_PG_LOG_MGR, "Table copy done; res size={}", res.size());
-        if (res.size() > 0) {
-            // process copy results
-            _process_copy_results(res);
-        } else {
-            // no tables copied
-            LOG_DEBUG(LOG_PG_LOG_MGR, "No tables copied; setting state=running");
-            // set to running this unblocks the xact handler
-            _internal_state.set(STATE_RUNNING);
-            Properties::set_db_state(_db_id, redis::db_state_change::REDIS_STATE_RUNNING);
+
+        bool copy_task_pending = true;
+        while(copy_task_pending) {
+            if (res.size() > 0) {
+                // process copy results
+                copy_task_pending = _process_copy_results(res);
+                if (copy_task_pending) {
+                    auto [next_table_id, next_xid_lsn] = _get_copy_table_ids();
+                    if (next_table_id != -1) {
+                        xid = _pg_log_reader->get_next_xid();
+                        SyncTracker::get_instance()->pick_table_for_sync(_db_id, next_table_id, next_xid_lsn.value());
+                        LOG_DEBUG(LOG_PG_LOG_MGR, "Copying more tables; target xid={}", xid);
+                        res = PgCopyTable::copy_tables(_db_id, xid, std::set<uint32_t>{next_table_id});
+                    } else {
+                        // Not able to fetch next table, so exit copy
+                        LOG_DEBUG(LOG_PG_LOG_MGR, "Couldn't fetch more tables; setting state=running");
+                        // set to running this unblocks the xact handler
+                        _internal_state.set(STATE_RUNNING);
+                        Properties::set_db_state(_db_id, redis::db_state_change::REDIS_STATE_RUNNING);
+                        copy_task_pending = false;
+                    }
+                }
+            } else {
+                // no tables copied
+                LOG_DEBUG(LOG_PG_LOG_MGR, "No more tables copied; setting state=running");
+                // set to running this unblocks the xact handler
+                _internal_state.set(STATE_RUNNING);
+                Properties::set_db_state(_db_id, redis::db_state_change::REDIS_STATE_RUNNING);
+                copy_task_pending = false;
+            }
         }
     }
 
@@ -387,7 +426,7 @@ namespace springtail::pg_log_mgr {
     }
 
 
-    void
+    bool
     PgLogMgr::_process_copy_results(const std::vector<PgCopyResultPtr> &res)
     {
         assert(_internal_state.is(STATE_SYNCING));
@@ -407,11 +446,16 @@ namespace springtail::pg_log_mgr {
             SyncTracker::get_instance()->add_sync(std::get<pg_log_mgr::PgXactMsg::TableSyncMsg>(redis_xact.msg));
         }
 
-        // process stalled messages; set state to replaying
-        _internal_state.set(STATE_REPLAYING);
-        _internal_state.wait_for_state(STATE_RUNNING);
-
-        LOG_DEBUG(LOG_PG_LOG_MGR, "Table copy done; state=replaying");
+        if (_redis_sync_queue.peek() == nullptr) {
+            // process stalled messages; set state to replaying
+            _internal_state.set(STATE_REPLAYING);
+            _internal_state.wait_for_state(STATE_RUNNING);
+            LOG_DEBUG(LOG_PG_LOG_MGR, "Table copy done; state=replaying");
+            return false;
+        } else {
+            LOG_DEBUG(LOG_PG_LOG_MGR, "Table copy done; More to process, dont wake up logreader");
+            return true;
+        }
     }
 
     bool

src/pg_log_mgr/pg_log_reader.cc

@@ -1189,7 +1189,6 @@ namespace springtail::pg_log_mgr {
             for (auto &entry : swap->table_info()) {
                 // update the existence cache for the referenced tables
                 _exists_cache->insert(db_id, entry->table_id, true);
-
                 auto copy_info = entry->info;
                 if (copy_info == nullptr) {
                     // During resync if the table is found to be invalid as part of the copy flow, the table
@@ -1198,36 +1197,51 @@ namespace springtail::pg_log_mgr {
                     LOG_DEBUG(LOG_PG_LOG_MGR, "Copy info not present for table {}", entry->table_id);
                     continue;
                 }
-                LOG_DEBUG(LOG_PG_LOG_MGR, "table_id {}", entry->table_id);
-
-                // perform the table swap
-                auto *namespace_req = copy_info->mutable_namespace_req();
-                namespace_req->set_xid(xid);
-                namespace_req->set_lsn(constant::RESYNC_NAMESPACE_LSN);
-
-                auto *create = copy_info->mutable_table_req();
-                create->set_xid(xid);
-                create->set_lsn(constant::RESYNC_CREATE_LSN);
-
-                auto *indexes = copy_info->mutable_index_reqs();
-                std::vector<proto::IndexRequest> indexes_vec;
-                for (auto &index : *indexes) {
-                    index.set_xid(xid);
-                    index.set_lsn(constant::RESYNC_CREATE_LSN);
-                    indexes_vec.push_back(index);
-                }
 
-                auto *roots = copy_info->mutable_roots_req();
-                roots->set_xid(xid);
+                if (copy_info->is_table_dropped()) {
+                    // Table is dropped when the sync was in queue/in-progress
+
+                    auto table_req = copy_info->table_req();
+                    auto table_info = table_req.table();
+                    PgMsgDropTable drop_msg;
+                    drop_msg.oid = table_info.id();
+                    drop_msg.namespace_name = table_info.namespace_name();
+                    drop_msg.table = table_info.name();
+                    std::string &&ddl_stmt = client->drop_table(_db_id, XidLsn{xid}, std::move(drop_msg));
+                    ddls.emplace_back(ddl_stmt);
+                } else {
+
+                    LOG_DEBUG(LOG_PG_LOG_MGR, "table_id {}", entry->table_id);
+
+                    // perform the table swap
+                    auto *namespace_req = copy_info->mutable_namespace_req();
+                    namespace_req->set_xid(xid);
+                    namespace_req->set_lsn(constant::RESYNC_NAMESPACE_LSN);
 
-                // note: this will also invalidate the table's client cache entry
-                auto ddl_str = client->swap_sync_table(*namespace_req, *create, indexes_vec, *roots);
+                    auto *create = copy_info->mutable_table_req();
+                    create->set_xid(xid);
+                    create->set_lsn(constant::RESYNC_CREATE_LSN);
 
-                // store the ddl mutations for the FDWs
-                auto ddl = nlohmann::json::parse(ddl_str);
-                assert(ddl.is_array());
-                ddls.insert(ddls.end(), ddl.begin(), ddl.end());
-                table_ids.emplace_back(static_cast<uint64_t>(entry->table_id));
+                    auto *indexes = copy_info->mutable_index_reqs();
+                    std::vector<proto::IndexRequest> indexes_vec;
+                    for (auto &index : *indexes) {
+                        index.set_xid(xid);
+                        index.set_lsn(constant::RESYNC_CREATE_LSN);
+                        indexes_vec.push_back(index);
+                    }
+
+                    auto *roots = copy_info->mutable_roots_req();
+                    roots->set_xid(xid);
+
+                    // note: this will also invalidate the table's client cache entry
+                    auto ddl_str = client->swap_sync_table(*namespace_req, *create, indexes_vec, *roots);
+
+                    // store the ddl mutations for the FDWs
+                    auto ddl = nlohmann::json::parse(ddl_str);
+                    assert(ddl.is_array());
+                    ddls.insert(ddls.end(), ddl.begin(), ddl.end());
+                    table_ids.emplace_back(static_cast<uint64_t>(entry->table_id));
+                }
             }
             LOG_INFO("Swapped synced tables: {}@{}", db_id, xid);