From cadd1ac728da22e444b8f83139b64577e1417684 Mon Sep 17 00:00:00 2001 From: stack Date: Wed, 23 Sep 2026 08:36:54 -0700 Subject: [PATCH] Re-recruit an old epoch's backup worker when it dies ae0706e016 (#13939) stopped an old-epoch backup worker failure from forcing a transaction-system recovery. Recovery was also the only caller of recruitBackupWorkers(), the sole consumer of BackupProgress::getUnfinishedBackup(), so nothing re-recruited the unfinished work and the dead worker's slot pinned oldestBackupEpoch indefinitely. That defers TLog pops and freezes latestBackupWorkerSavedVersion, which is the restorable version for PARTITIONED_LOG, so BackupLogsDispatchTask never satisfies stopWhenDone && restorableVersion.present() and such a backup never completes. Seen on tests/slow/ParallelRestoreNewBackupCorrectnessAtomicOp.toml seed 1003233285: epoch-8 tag -2:0 worker ac570a326c42885d died at t=287.4 before saveProgress, no further recovery followed, and RestorableVersion stayed -1 for 12,300 simsec until TracedTooManyLines aborted the run. The seed now passes in 593 simsec with oldestBackupEpoch advancing 8 -> 10 -> 12. replaceBackupWorker swaps the interface in place so the epoch keeps its hold on oldestBackupEpoch while its work is outstanding. Knob CC_RERECRUIT_BACKUP_WORKER_ENABLED (default true) disables the monitor. This does not address a second, independent cause of the same pin: a live old-epoch worker that never completes its range, reproducible on tests/slow/BackupNewAndOldRestore.toml seed 2702875306. --- fdbclient/ServerKnobs.cpp | 1 + fdbclient/include/fdbclient/ServerKnobs.h | 1 + fdbserver/ClusterRecovery.actor.cpp | 94 +++++++++++++++++++ fdbserver/TagPartitionedLogSystem.actor.cpp | 69 ++++++++++++-- fdbserver/include/fdbserver/LogSystem.h | 10 ++ .../fdbserver/TagPartitionedLogSystem.actor.h | 7 ++ 6 files changed, 175 insertions(+), 7 deletions(-) diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index ee1a4d4a7ea..bf680792935 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -794,6 +794,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( BACKUP_FILE_BLOCK_BYTES, 1024 * 1024 ); init( BACKUP_WORKER_LOCK_BYTES, 3e9 ); if(randomize && BUGGIFY) BACKUP_WORKER_LOCK_BYTES = deterministicRandom()->randomInt(2048, 4096) * 4096; init( BACKUP_UPLOAD_DELAY, 10.0 ); if(randomize && BUGGIFY) BACKUP_UPLOAD_DELAY = deterministicRandom()->random01() * 60; + init( CC_RERECRUIT_BACKUP_WORKER_ENABLED, true ); //Cluster Controller init( CLUSTER_CONTROLLER_LOGGING_DELAY, 5.0 ); diff --git a/fdbclient/include/fdbclient/ServerKnobs.h b/fdbclient/include/fdbclient/ServerKnobs.h index f53addcb426..b606b591c51 100644 --- a/fdbclient/include/fdbclient/ServerKnobs.h +++ b/fdbclient/include/fdbclient/ServerKnobs.h @@ -757,6 +757,7 @@ class SWIFT_CXX_IMMORTAL_SINGLETON_TYPE ServerKnobs : public KnobsImpl> getMinBackupVersion(Reference monitorOldEpochBackupWorker(Reference self, + Database cx, + InitializeBackupRequest req, + BackupInterface interf) { + state int nextWorker = 0; + loop { + wait(waitFailureClient(interf.waitFailure, + SERVER_KNOBS->BACKUP_TIMEOUT, + -SERVER_KNOBS->BACKUP_TIMEOUT / SERVER_KNOBS->SECONDS_BEFORE_NO_FAILURE_DELAY, + /*trace=*/true)); + + // A worker that completed its range also stops answering, so only durable progress can say + // whether work remains. + state Reference progress( + new BackupProgress(self->dbgid, self->logSystem->getOldEpochTagsVersionsInfo())); + wait(getBackupProgress(cx, self->dbgid, progress, /*logging=*/false)); + + std::map status = progress->getEpochStatus(req.backupEpoch); + auto it = status.find(req.routerTag); + Version savedVersion = it == status.end() ? invalidVersion : it->second; + + if (savedVersion >= req.endVersion.get()) { + // Durable progress covers the whole range, so those mutations reached the container and + // the slot can be released even though the worker died before reporting done. + CODE_PROBE(true, "Released an old epoch backup worker slot from durable progress"); + TraceEvent("BackupWorkerReplacementDone", self->dbgid) + .detail("BackupEpoch", req.backupEpoch) + .detail("Tag", req.routerTag.toString()) + .detail("WorkerID", interf.id()) + .detail("SavedVersion", savedVersion) + .detail("EndVersion", req.endVersion.get()); + self->logSystem->releaseBackupWorker(interf.id(), req.backupEpoch); + return Void(); + } + + InitializeBackupRequest replacement(deterministicRandom()->randomUniqueID()); + replacement.recruitedEpoch = req.recruitedEpoch; + replacement.backupEpoch = req.backupEpoch; + replacement.routerTag = req.routerTag; + replacement.totalTags = req.totalTags; + replacement.startVersion = std::max(req.startVersion, savedVersion + 1); + replacement.endVersion = req.endVersion; + + WorkerInterface worker = self->backupWorkers[nextWorker++ % self->backupWorkers.size()]; + CODE_PROBE(true, "Re-recruiting an old epoch backup worker that died before finishing"); + TraceEvent("BackupWorkerReplacement", self->dbgid) + .detail("RequestID", replacement.reqId) + .detail("Tag", replacement.routerTag.toString()) + .detail("Epoch", replacement.recruitedEpoch) + .detail("BackupEpoch", replacement.backupEpoch) + .detail("StartVersion", replacement.startVersion) + .detail("EndVersion", replacement.endVersion.get()) + .detail("DeadWorkerID", interf.id()); + + try { + InitializeBackupReply reply = wait(throwErrorOr( + worker.backup.getReplyUnlessFailedFor(replacement, + SERVER_KNOBS->BACKUP_TIMEOUT, + SERVER_KNOBS->MASTER_FAILURE_SLOPE_DURING_RECOVERY))); + if (!self->logSystem->replaceBackupWorker(interf.id(), reply)) { + return Void(); + } + interf = reply.interf; + } catch (Error& e) { + if (e.code() == error_code_actor_cancelled) { + throw; + } + CODE_PROBE(true, "Old epoch backup worker re-recruitment failed", probe::decoration::rare); + TraceEvent(SevWarn, "BackupWorkerReplacementFailed", self->dbgid) + .error(e) + .detail("BackupEpoch", req.backupEpoch) + .detail("Tag", req.routerTag.toString()) + .detail("DeadWorkerID", interf.id()); + wait(delay(SERVER_KNOBS->BACKUP_TIMEOUT)); + } + } +} + ACTOR static Future recruitBackupWorkers(Reference self, Database cx) { ASSERT(self->backupWorkers.size() > 0); @@ -676,6 +760,7 @@ ACTOR static Future recruitBackupWorkers(Reference se new BackupProgress(self->dbgid, self->logSystem->getOldEpochTagsVersionsInfo())); state Future gotProgress = getBackupProgress(cx, self->dbgid, backupProgress, /*logging=*/true); state std::vector> initializationReplies; + state std::vector oldEpochRequests; // requests for previous epochs' unfinished work state std::vector> idsTags; // worker IDs and tags for current epoch state int logRouterTags = self->logSystem->getLogRouterTags(); @@ -744,6 +829,7 @@ ACTOR static Future recruitBackupWorkers(Reference se throwErrorOr(worker.backup.getReplyUnlessFailedFor( req, SERVER_KNOBS->BACKUP_TIMEOUT, SERVER_KNOBS->MASTER_FAILURE_SLOPE_DURING_RECOVERY)), backup_worker_failed())); + oldEpochRequests.push_back(req); } } @@ -751,6 +837,14 @@ ACTOR static Future recruitBackupWorkers(Reference se self->logSystem->setBackupWorkers(newRecruits); TraceEvent("BackupRecruitmentDone", self->dbgid).log(); self->registrationTrigger.trigger(); + + ASSERT(newRecruits.size() == logRouterTags + oldEpochRequests.size()); + if (SERVER_KNOBS->CC_RERECRUIT_BACKUP_WORKER_ENABLED) { + for (int j = 0; j < oldEpochRequests.size(); j++) { + self->addActor.send(monitorOldEpochBackupWorker( + self, cx, oldEpochRequests[j], newRecruits[logRouterTags + j].interf)); + } + } return Void(); } diff --git a/fdbserver/TagPartitionedLogSystem.actor.cpp b/fdbserver/TagPartitionedLogSystem.actor.cpp index e52ad1aac08..f098e63febd 100644 --- a/fdbserver/TagPartitionedLogSystem.actor.cpp +++ b/fdbserver/TagPartitionedLogSystem.actor.cpp @@ -1939,13 +1939,7 @@ bool TagPartitionedLogSystem::removeBackupWorker(const BackupWorkerDoneRequest& } if (removed) { - oldestBackupEpoch = epoch; - for (const auto& old : oldLogData) { - if (old.epoch < oldestBackupEpoch && old.tLogs[0]->backupWorkers.size() > 0) { - oldestBackupEpoch = old.epoch; - } - } - backupWorkerChanged.trigger(); + recomputeOldestBackupEpoch(); } else { removedBackupWorkers.insert(req.workerUID); } @@ -1958,6 +1952,67 @@ bool TagPartitionedLogSystem::removeBackupWorker(const BackupWorkerDoneRequest& return removed; } +bool TagPartitionedLogSystem::replaceBackupWorker(UID deadWorker, const InitializeBackupReply& reply) { + Reference logset = getEpochLogSet(reply.backupEpoch); + if (!logset.isValid()) { + return false; + } + + for (auto& worker : logset->backupWorkers) { + if (worker->get().interf().id() != deadWorker) { + continue; + } + // Keeping the entry in place leaves the epoch's count unchanged, so it retains its hold on + // oldestBackupEpoch and no TLog data becomes collectable while its work is outstanding. + worker->setUnconditional(OptionalInterface(reply.interf)); + TraceEvent("ReplaceBackupWorker", dbgid) + .detail("BackupEpoch", reply.backupEpoch) + .detail("DeadWorkerID", deadWorker) + .detail("WorkerID", reply.interf.id()); + + // A replacement that finished before this call found no entry to erase and only recorded its + // UID. Honour that now, or the epoch keeps a slot for a worker that will never report again. + if (removedBackupWorkers.contains(reply.interf.id())) { + removedBackupWorkers.erase(reply.interf.id()); + releaseBackupWorker(reply.interf.id(), reply.backupEpoch); + return false; + } + + backupWorkerChanged.trigger(); + return true; + } + return false; +} + +void TagPartitionedLogSystem::releaseBackupWorker(UID worker, LogEpoch backupEpoch) { + Reference logset = getEpochLogSet(backupEpoch); + if (!logset.isValid()) { + return; + } + + for (auto it = logset->backupWorkers.begin(); it != logset->backupWorkers.end(); it++) { + if (it->getPtr()->get().interf().id() == worker) { + logset->backupWorkers.erase(it); + recomputeOldestBackupEpoch(); + TraceEvent("ReleaseBackupWorker", dbgid) + .detail("BackupEpoch", backupEpoch) + .detail("WorkerID", worker) + .detail("OldestBackupEpoch", oldestBackupEpoch); + return; + } + } +} + +void TagPartitionedLogSystem::recomputeOldestBackupEpoch() { + oldestBackupEpoch = epoch; + for (const auto& old : oldLogData) { + if (old.epoch < oldestBackupEpoch && old.tLogs[0]->backupWorkers.size() > 0) { + oldestBackupEpoch = old.epoch; + } + } + backupWorkerChanged.trigger(); +} + LogEpoch TagPartitionedLogSystem::getOldestBackupEpoch() const { return oldestBackupEpoch; } diff --git a/fdbserver/include/fdbserver/LogSystem.h b/fdbserver/include/fdbserver/LogSystem.h index ec65e2b41e1..eef177df50a 100644 --- a/fdbserver/include/fdbserver/LogSystem.h +++ b/fdbserver/include/fdbserver/LogSystem.h @@ -718,6 +718,16 @@ struct ILogSystem { // if the worker is not found. virtual bool removeBackupWorker(const BackupWorkerDoneRequest& req) = 0; + // Points an old epoch's backup worker slot at a re-recruited replacement, keeping the entry in + // place so the epoch retains its hold on oldestBackupEpoch. Returns false if the dead worker is + // no longer tracked, meaning the replacement must not be installed. + virtual bool replaceBackupWorker(UID deadWorker, const InitializeBackupReply& reply) = 0; + + // Frees a backup worker slot whose work durable progress already shows as complete, for a worker + // that died without reporting done. Unlike removeBackupWorker this records nothing for a later + // setBackupWorkers, since there is no done request in flight to reconcile with. + virtual void releaseBackupWorker(UID worker, LogEpoch backupEpoch) = 0; + virtual LogEpoch getOldestBackupEpoch() const = 0; virtual void setOldestBackupEpoch(LogEpoch epoch) = 0; }; diff --git a/fdbserver/include/fdbserver/TagPartitionedLogSystem.actor.h b/fdbserver/include/fdbserver/TagPartitionedLogSystem.actor.h index 55c59fa7d97..e3f1e8fe1ae 100644 --- a/fdbserver/include/fdbserver/TagPartitionedLogSystem.actor.h +++ b/fdbserver/include/fdbserver/TagPartitionedLogSystem.actor.h @@ -344,6 +344,13 @@ struct TagPartitionedLogSystem final : ILogSystem, ReferenceCounted