wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'refs/remotes/origin/main' into dev/tclinkenbeard/private-header-encapsulation-20260715

This commit is contained in:
Trevor Clinkenbeard 2026-07-20 13:01:51 -07:00
parent 7753767a89 a04ee7f37f
commit 3d852e8ead
12 changed files with 218 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
fdbclient/BackupContainer.cpp
View File

@ -45,15 +45,12 @@ Future<Void> appendStringRefWithLen(Reference<IBackupFile> file, Standalone<Stri
co_await file->append(s.begin(), s.size());
}
// Writes data in chunks of at most BACKUP_MANIFEST_WRITE_CHUNK_SIZE bytes. This is necessary because
// IBackupFile::append() takes an int length, so passing a size_t larger than INT_MAX would silently
// truncate to a negative value and corrupt the write.
Future<Void> appendChunked(Reference<IBackupFile> file, const void* data, size_t len) {
Future<Void> append(Reference<IBackupFile> file, const void* data, size_t len) {
const char* ptr = static_cast<const char*>(data);
size_t chunkLimit = static_cast<size_t>(CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE);
for (size_t offset = 0; offset < len;) {
int chunkSize = static_cast<int>(
std::min(len - offset, static_cast<size_t>(CLIENT_KNOBS->BACKUP_MANIFEST_WRITE_CHUNK_SIZE)));
co_await file->append(ptr + offset, chunkSize);
size_t chunkSize = std::min(len - offset, chunkLimit);
co_await file->appendImpl(ptr + offset, chunkSize);
offset += chunkSize;
}
}
@ -65,7 +62,7 @@ Future<Void> IBackupFile::appendStringRefWithLen(Standalone<StringRef> s) {
}
Future<Void> IBackupFile::append(const void* data, size_t len) {
return IBackupFile_impl::appendChunked(Reference<IBackupFile>::addRef(this), data, len);
return IBackupFile_impl::append(Reference<IBackupFile>::addRef(this), data, len);
}
bool isBlobstoreUrl(const std::string& url) {

4
fdbclient/BackupContainerBlobStore.cpp
View File

@ -52,8 +52,8 @@ public:
BackupFile(std::string fileName, Reference<IAsyncFile> file)
: IBackupFile(fileName), m_file(file), m_offset(0) {}
Future<Void> append(const void* data, int len) override {
Future<Void> r = m_file->write(data, len, m_offset);
Future<Void> appendImpl(const void* data, size_t len) override {
Future<Void> r = m_file->write(data, static_cast<int>(len), m_offset);
m_offset += len;
return r;
}

99
fdbclient/BackupContainerFileSystem.cpp
View File

@ -38,6 +38,20 @@
class BackupContainerFileSystemImpl {
public:
// A snapshot manifest is normally a few hundred MB. Warn as it grows and error before it gets dangerously
// large, so we see the problem in the logs with time to act before a manifest actually becomes too large
// to handle.
static void traceManifestSize(const std::string& fileName, int64_t bytes) {
constexpr int64_t MB = 1048576; // 1024 * 1024
if (bytes >= 750 * MB) {
TraceEvent(SevError, "BackupSnapshotManifestTooLarge").detail("FileName", fileName).detail("Bytes", bytes);
} else if (bytes >= 500 * MB) {
TraceEvent(SevWarnAlways, "BackupSnapshotManifestLarge")
.detail("FileName", fileName)
.detail("Bytes", bytes);
}
}
// TODO: Do this more efficiently, as the range file list for a snapshot could potentially be hundreds of
// megabytes.
static Future<std::pair<std::vector<RangeFile>, std::map<std::string, KeyRange>>> readKeyspaceSnapshot(
@ -55,10 +69,22 @@ public:
// return them.
Reference<IAsyncFile> f = co_await bc->readFile(snapshot.fileName);
int64_t size = co_await f->size();
Standalone<StringRef> buf = makeString(size);
co_await f->read(mutateString(buf), buf.size(), 0);
traceManifestSize(snapshot.fileName, size);
// A manifest is normally a few hundred MB. Read it into a std::string in chunks; std::string and the
// chunked reads guard against an unexpectedly large manifest overflowing the int length that read() takes.
// TODO (optimization): the whole manifest is loaded into memory before parsing. Explore if a streaming JSON
// parser would avoid this.
std::string buf;
buf.resize(size);
for (int64_t offset = 0; offset < size;) {
int toRead = static_cast<int>(std::min<int64_t>(CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE, size - offset));
int r = co_await f->read((uint8_t*)buf.data() + offset, toRead, offset);
if (r != toRead)
throw restore_corrupted_data();
offset += r;
}
json_spirit::mValue json;
if (!json_spirit::read_string(buf.toString(), json)) {
if (!json_spirit::read_string(buf, json)) {
fprintf(stderr,
"ERROR: Failed to read data. Verify that backup and restore encryption keys match (if provided) or "
"the data is corrupted.\n");
@ -228,6 +254,8 @@ public:
}
co_await yield();
// TODO (optimization): the whole manifest is built and serialized in memory before writing. Explore if a
// streaming approach would avoid this.
std::string docString = json_spirit::write_string(json);
// Generate filename - add suffixes only when 'both' mode is active to prevent collision
@ -277,6 +305,7 @@ public:
Reference<IBackupFile> f = co_await bc->writeFile(fileName);
traceManifestSize(fileName, docString.size());
co_await f->append(docString.data(), docString.size());
co_await f->finish();
@ -2769,12 +2798,12 @@ TEST_CASE("/backup/containers/localdir/expireProgressVersions") {
}
// Verify that writeKeyspaceSnapshotFile correctly writes and reads back a snapshot manifest even when the
// JSON document is larger than BACKUP_MANIFEST_WRITE_CHUNK_SIZE, exercising the chunked-append path.
// JSON document is larger than BACKUP_MANIFEST_CHUNK_SIZE, exercising the chunked-append path.
TEST_CASE("/backup/containers/localdir/writeKeyspaceSnapshotFile/chunked") {
// Force a tiny chunk size so a normal-sized manifest triggers multiple append() calls.
int savedChunkSize = CLIENT_KNOBS->BACKUP_MANIFEST_WRITE_CHUNK_SIZE;
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_WRITE_CHUNK_SIZE = 64;
ASSERT_EQ(CLIENT_KNOBS->BACKUP_MANIFEST_WRITE_CHUNK_SIZE, 64);
int savedChunkSize = CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE;
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_CHUNK_SIZE = 64;
ASSERT_EQ(CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE, 64);
std::string url = format("file://%s/fdb_backups/%llx", params.getDataDir().c_str(), timer_int());
Reference<IBackupContainer> c = IBackupContainer::openContainer(url, {}, {}, 0);
@ -2804,7 +2833,61 @@ TEST_CASE("/backup/containers/localdir/writeKeyspaceSnapshotFile/chunked") {
ASSERT_EQ(listing.snapshots[0].beginVersion, 1000);
ASSERT_EQ(listing.snapshots[0].endVersion, 1004);
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_WRITE_CHUNK_SIZE = savedChunkSize;
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_CHUNK_SIZE = savedChunkSize;
co_await c->deleteContainer();
}
// Verify that readKeyspaceSnapshot correctly reassembles and parses a snapshot manifest when it is read
// back in many small pieces, exercising the chunked-read path. A tiny chunk size (that does not divide the
// manifest evenly) forces the read loop to run many iterations with a partial final chunk, which catches
// off-by-one / wrong-offset / short-read bugs in the loop. Note: a unit test cannot allocate a >2 GB
// manifest to reproduce the original int overflow, so this validates the chunking logic instead.
TEST_CASE("/backup/containers/localdir/readKeyspaceSnapshot/chunked") {
int savedChunkSize = CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE;
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_CHUNK_SIZE = 7;
std::string url = format("file://%s/fdb_backups/%llx", params.getDataDir().c_str(), timer_int());
Reference<IBackupContainer> c = IBackupContainer::openContainer(url, {}, {}, 0);
co_await c->create();
Version v = 1000;
int blockSize = 64;
// Write several range files with distinct, non-empty key ranges so the manifest also contains a
// populated keyRanges section (exercising that part of the read path too).
std::vector<std::string> rangeFileNames;
std::vector<std::pair<Key, Key>> beginEndKeys;
std::map<std::string, std::pair<std::string, std::string>> expected;
for (int i = 0; i < 5; ++i) {
Key begin = StringRef(format("begin-%d", i));
Key end = StringRef(format("end-%d", i));
Reference<IBackupFile> range = co_await c->writeRangeFile(v, 0, v, blockSize);
co_await testWriteSnapshotFile(range, begin, end, blockSize);
rangeFileNames.push_back(range->getFileName());
beginEndKeys.push_back({ begin, end });
expected[range->getFileName()] = { begin.toString(), end.toString() };
++v;
}
int64_t totalSize = 99999;
co_await c->writeKeyspaceSnapshotFile(rangeFileNames, beginEndKeys, totalSize, IncludeKeyRangeMap::True);
// Read the manifest back through the chunked-read path and verify every range file and key range.
Reference<BackupContainerFileSystem> bcfs = c.castTo<BackupContainerFileSystem>();
std::vector<KeyspaceSnapshotFile> snapshots = co_await bcfs->listKeyspaceSnapshots();
ASSERT_EQ(snapshots.size(), 1);
auto [files, keyRanges] = co_await bcfs->readKeyspaceSnapshot(snapshots[0]);
ASSERT_EQ(files.size(), rangeFileNames.size());
ASSERT_EQ(keyRanges.size(), expected.size());
for (const auto& [fileName, range] : expected) {
auto it = keyRanges.find(fileName);
ASSERT(it != keyRanges.end());
ASSERT(it->second.begin == StringRef(range.first));
ASSERT(it->second.end == StringRef(range.second));
}
const_cast<ClientKnobs*>(CLIENT_KNOBS)->BACKUP_MANIFEST_CHUNK_SIZE = savedChunkSize;
co_await c->deleteContainer();
}

4
fdbclient/BackupContainerLocalDirectory.cpp
View File

@ -40,8 +40,8 @@ public:
m_buffer.reserve(m_buffer.arena(), m_blockSize);
}
Future<Void> append(const void* data, int len) override {
m_buffer.append(m_buffer.arena(), (const uint8_t*)data, len);
Future<Void> appendImpl(const void* data, size_t len) override {
m_buffer.append(m_buffer.arena(), (const uint8_t*)data, static_cast<int>(len));
if (m_buffer.size() >= m_blockSize) {
return flush(m_blockSize);

2
fdbclient/ClientKnobs.cpp
View File

@ -258,7 +258,7 @@ void ClientKnobs::initialize(Randomize randomize, IsSimulated isSimulated) {
//Backup
init( BACKUP_LOCAL_FILE_WRITE_BLOCK, 024*1024 );
init( BACKUP_MANIFEST_WRITE_CHUNK_SIZE, std::numeric_limits<int>::max() ); if( randomize && buggify() ) BACKUP_MANIFEST_WRITE_CHUNK_SIZE = 64;
init( BACKUP_MANIFEST_CHUNK_SIZE, std::numeric_limits<int>::max() ); if( randomize && buggify() ) BACKUP_MANIFEST_CHUNK_SIZE = 64;
init( BACKUP_CONCURRENT_DELETES, 100 );
init( BACKUP_SIMULATED_LIMIT_BYTES, 1e6 ); if( randomize && buggify() ) BACKUP_SIMULATED_LIMIT_BYTES = 1000;
init( BACKUP_GET_RANGE_LIMIT_BYTES, 1e6 );

8
fdbclient/include/fdbclient/BackupContainer.h
View File

@ -48,9 +48,11 @@ public:
explicit IBackupFile(const std::string& fileName) : m_fileName(fileName) {}
virtual ~IBackupFile() = default;
// Backup files are append-only and cannot have more than 1 append outstanding at once.
virtual Future<Void> append(const void* data, int len) = 0;
// Non-virtual size_t overload: safely chunks large writes so len never overflows the int parameter
// of the virtual append(). Uses CLIENT_KNOBS->BACKUP_MANIFEST_WRITE_CHUNK_SIZE as the chunk size.
// Backend hook that writes a single chunk. len is bounded by the chunk size (see append()), so
// backends may safely narrow it to the int length taken by IAsyncFile::write().
virtual Future<Void> appendImpl(const void* data, size_t len) = 0;
// Writes len bytes, slicing them into chunks of at most CLIENT_KNOBS->BACKUP_MANIFEST_CHUNK_SIZE
// so appendImpl() never receives more than INT_MAX bytes.
Future<Void> append(const void* data, size_t len);
virtual Future<Void> finish() = 0;
inline std::string getFileName() const { return m_fileName; }

2
fdbclient/include/fdbclient/Knobs.h
View File

@ -160,7 +160,7 @@ public:
// Backup
int BACKUP_LOCAL_FILE_WRITE_BLOCK;
int BACKUP_MANIFEST_WRITE_CHUNK_SIZE;
int BACKUP_MANIFEST_CHUNK_SIZE;
int BACKUP_CONCURRENT_DELETES;
int BACKUP_SIMULATED_LIMIT_BYTES;
int BACKUP_GET_RANGE_LIMIT_BYTES;

39
fdbserver/kvstore/KeyValueStoreShardedRocksDB.cpp
View File

@ -1033,7 +1033,6 @@ struct PhysicalShard {
if (!s.ok()) {
logRocksDBError(s, "DestroyShard");
logShardEvent(id, ShardOp::DESTROY, SevError, s.ToString());
return;
}
}
auto s = db->DestroyColumnFamilyHandle(cf);
@ -1888,8 +1887,13 @@ public:
void closeAllShards() {
columnFamilyMap.clear();
physicalShards.clear();
if (db == nullptr) {
return;
}
// Close DB.
auto s = db->Close();
delete db;
db = nullptr;
if (!s.ok()) {
logRocksDBError(s, "Close");
return;
@ -1898,6 +1902,9 @@ public:
}
void destroyAllShards() {
if (db == nullptr) {
return;
}
auto metadataShard = getMetaDataShard();
KeyRange metadataRange = prefixRange(shardMappingPrefix);
rocksdb::WriteOptions options;
@ -1908,6 +1915,8 @@ public:
physicalShards.clear();
// Close DB.
auto s = db->Close();
delete db;
db = nullptr;
if (!s.ok()) {
logRocksDBError(s, "Close");
return;
@ -2327,7 +2336,7 @@ struct ShardedRocksDBKeyValueStore : IKeyValueStore {
struct CompactShardsAction : TypedAction<CompactionWorker, CompactShardsAction> {
std::vector<std::shared_ptr<PhysicalShard>> shards;
std::shared_ptr<PhysicalShard> metadataShard;
PhysicalShard* metadataShard;
ThreadReturnPromise<Void> done;
CompactShardsAction(std::vector<std::shared_ptr<PhysicalShard>> shards, PhysicalShard* metadataShard)
: shards(shards), metadataShard(metadataShard) {}
@ -3381,6 +3390,7 @@ struct ShardedRocksDBKeyValueStore : IKeyValueStore {
self->refreshHolder.cancel();
self->refreshRocksDBBackgroundWorkHolder.cancel();
self->cleanUpJob.cancel();
self->compactionJob.cancel();
self->counterLogger.cancel();
try {
@ -3388,6 +3398,12 @@ struct ShardedRocksDBKeyValueStore : IKeyValueStore {
} catch (Error& e) {
TraceEvent(SevError, "ShardedRocksCloseReadThreadError").errorUnsuppressed(e);
}
try {
co_await self->compactionThread->stop();
} catch (Error& e) {
TraceEvent(SevError, "ShardedRocksCloseCompactionThreadError").errorUnsuppressed(e);
}
self->compactionThread.clear();
TraceEvent("CloseKeyValueStore").detail("DeleteKVS", deleteOnClose);
self->iteratorPool->clear();
@ -3402,7 +3418,6 @@ struct ShardedRocksDBKeyValueStore : IKeyValueStore {
try {
co_await self->writeThread->stop();
co_await self->compactionThread->stop();
} catch (Error& e) {
TraceEvent(SevError, "ShardedRocksCloseWriteThreadError").errorUnsuppressed(e);
}
@ -3929,6 +3944,24 @@ TEST_CASE("noSim/ShardedRocksDB/Initialization") {
ASSERT(!directoryExists(rocksDBTestDir));
}
TEST_CASE("noSim/ShardedRocksDB/CloseWithoutInit") {
const std::string rocksDBTestDir = joinPath(params.getDataDir(), "sharded-rocksdb-close-without-init");
platform::eraseDirectoryRecursive(rocksDBTestDir);
for (bool dispose : { false, true }) {
IKeyValueStore* kvStore =
new ShardedRocksDBKeyValueStore(rocksDBTestDir, deterministicRandom()->randomUniqueID());
Future<Void> closed = kvStore->onClosed();
if (dispose) {
kvStore->dispose();
} else {
kvStore->close();
}
co_await closed;
ASSERT(!directoryExists(rocksDBTestDir));
}
}
TEST_CASE("noSim/ShardedRocksDB/SingleShardRead") {
const std::string rocksDBTestDir = "sharded-rocksdb-test-db";
platform::eraseDirectoryRecursive(rocksDBTestDir);

20
fdbserver/storageserver/storageserver.cpp
View File

@ -12583,7 +12583,15 @@ Future<Void> storageServer(IKeyValueStore* persistentData,
}
ssCore.cancel();
self.actors = ActorCollection(false);
co_await delay(0);
try {
co_await delay(0);
} catch (Error& cleanupError) {
// A rollback keeps the KVS open for its rebooter, which cannot reclaim it after cancellation.
if (cleanupError.code() == error_code_actor_cancelled && err.code() == error_code_please_reboot) {
persistentData->close();
}
throw;
}
throw err;
}
}
@ -12702,7 +12710,15 @@ Future<Void> storageServer(IKeyValueStore* persistentData,
}
ssCore.cancel();
self.actors = ActorCollection(false);
co_await delay(0);
try {
co_await delay(0);
} catch (Error& cleanupError) {
// A rollback keeps the KVS open for its rebooter, which cannot reclaim it after cancellation.
if (cleanupError.code() == error_code_actor_cancelled && err.code() == error_code_please_reboot) {
persistentData->close();
}
throw;
}
throw err;
}

6
fdbserver/tlog/TLogServer.cpp
View File

@ -3871,6 +3871,7 @@ Future<Void> restorePersistentState(TLogData* self,
Version ver = BinaryReader::fromStringRef<Version>(fVers.get()[idx].value, Unversioned());
logData->persistentDataVersion = ver;
logData->persistentDataDurableVersion = ver;
logData->queuePoppedVersion = ver;
logData->version.set(ver);
logData->recoveryCount =
BinaryReader::fromStringRef<DBRecoveryCount>(fRecoverCounts.get()[idx].value, Unversioned());
@ -3903,6 +3904,10 @@ Future<Void> restorePersistentState(TLogData* self,
logData->createTagData(
tag, popped, NothingPersistent::False, PoppedRecently::False, UnpoppedRecovered::False);
logData->getTagData(tag)->persistentPopped = popped;
// Reference-spilled data can still pin disk queue entries before the restored durable version.
if (logData->shouldSpillByReference(tag)) {
logData->queuePoppedVersion = std::min(logData->queuePoppedVersion, popped);
}
}
}
}
@ -4154,6 +4159,7 @@ Future<Void> tLogStart(TLogData* self, InitializeTLogRequest req, LocalityData l
logData->persistentDataVersion = logData->unrecoveredBefore - 1;
logData->persistentDataDurableVersion = logData->unrecoveredBefore - 1;
logData->queueCommittedVersion.set(logData->unrecoveredBefore - 1);
logData->queuePoppedVersion = logData->unrecoveredBefore - 1;
logData->version.set(logData->unrecoveredBefore - 1);
logData->unpoppedRecoveredTagCount = req.allTags.size();

72
fdbserver/workloads/GcGenerations.cpp
View File

@ -32,6 +32,7 @@
#include "fdbrpc/simulator.h"
#include "flow/CodeProbe.h"
#include "flow/NetworkAddress.h"
#include "flow/ScopeExit.h"
#include "flow/Error.h"
#include "flow/Trace.h"
#include "flow/flow.h"
@ -44,12 +45,16 @@ struct GcGenerationsWorkload : TestWorkload {
bool enabled;
double testDuration;
double startDelay;
bool completed = false;
bool forceCloggedDcMasterRetry;
std::vector<std::pair<IPAddress, IPAddress>> cloggedPairs;
Optional<Standalone<StringRef>> cloggedDcId;
explicit GcGenerationsWorkload(WorkloadContext const& wcx) : TestWorkload(wcx) {
enabled = !clientId; // only do this on the "first" client
testDuration = getOption(options, "testDuration"_sr, 1000.0);
startDelay = getOption(options, "startDelay"_sr, 30.0);
forceCloggedDcMasterRetry = getOption(options, "forceCloggedDcMasterRetry"_sr, false);
}
void disableFailureInjectionWorkloads(std::set<std::string>& out) const override {
@ -64,7 +69,7 @@ struct GcGenerationsWorkload : TestWorkload {
else
return Void();
}
Future<bool> check(Database const& cx) override { return true; }
Future<bool> check(Database const& cx) override { return !g_network->isSimulated() || !enabled || completed; }
void getMetrics(std::vector<PerfMetric>& m) override {}
// Ensure simulator state is cleaned up even if the workload is cancelled by timeout.
@ -85,6 +90,7 @@ struct GcGenerationsWorkload : TestWorkload {
g_simulator->unclogPair(pair.first, pair.second);
}
cloggedPairs.clear();
cloggedDcId.reset();
}
Future<Void> clogRemoteDc(GcGenerationsWorkload* self, Database cx) {
@ -104,12 +110,20 @@ struct GcGenerationsWorkload : TestWorkload {
return false;
};
auto& simPolicy = fdbSimulationPolicyState();
Optional<Standalone<StringRef>> inactiveDcId = simPolicy.remoteDcId;
// A region failover can make the configured remote DC the active primary. Always partition the inactive DC.
if (self->dbInfo->get().master.locality.dcId() == inactiveDcId) {
inactiveDcId = simPolicy.primaryDcId;
}
self->cloggedDcId = inactiveDcId;
std::vector<IPAddress> ips; // all non-remote process IPs
std::vector<IPAddress> remoteIps; // all remote process IPs
for (const auto& process : g_simulator->getAllProcesses()) {
const auto& ip = process->address.ip;
if (process->locality.dcId().present() &&
process->locality.dcId() == fdbSimulationPolicyState().remoteDcId && !isCoordinator(coordinators, ip)) {
if (process->locality.dcId().present() && process->locality.dcId() == inactiveDcId &&
!isCoordinator(coordinators, ip)) {
remoteIps.push_back(ip);
} else {
ips.push_back(ip);
@ -128,28 +142,28 @@ struct GcGenerationsWorkload : TestWorkload {
}
TraceEvent("PartitionRemoteDc")
.detail("RemoteDc", fdbSimulationPolicyState().remoteDcId)
.detail("RemoteDc", inactiveDcId)
.detail("CloggedRemoteProcess", describe(remoteIps));
}
bool isMasterInRemoteDc(GcGenerationsWorkload* self) {
bool isMasterInCloggedDc(GcGenerationsWorkload* self) {
auto masterAddr = self->dbInfo->get().master.address();
auto* masterProc = g_simulator->getProcessByAddress(masterAddr);
return !masterProc || !masterProc->locality.dcId().present() ||
masterProc->locality.dcId() == fdbSimulationPolicyState().remoteDcId;
masterProc->locality.dcId() == self->cloggedDcId;
}
// Wait for the DB to reach ACCEPTING_COMMITS. If rebootRemoteDcMaster is true and
// the master is in the remote DC, reboot it to force the CC to elect a primary DC
// master. This is required when the remote DC is clogged (otherwise recovery can
// never complete), but must be disabled once the remote DC is unclogged — otherwise
// every CC re-election that lands in the remote DC triggers another reboot, producing
// Wait for the DB to reach ACCEPTING_COMMITS. If rebootCloggedDcMaster is true and
// the master is in the clogged DC, reboot it to force the CC to elect an active DC
// master. This is required when the inactive DC is clogged (otherwise recovery can
// never complete), but must be disabled once that DC is unclogged — otherwise
// every CC re-election that lands there triggers another reboot, producing
// a tight loop that prevents recovery from ever reaching ACCEPTING_COMMITS.
Future<Void> dbAvailable(GcGenerationsWorkload* self, bool rebootRemoteDcMaster) {
Future<Void> dbAvailable(GcGenerationsWorkload* self, bool rebootCloggedDcMaster) {
while (self->dbInfo->get().recoveryState < RecoveryState::ACCEPTING_COMMITS) {
co_await self->dbInfo->onChange();
if (rebootRemoteDcMaster && self->dbInfo->get().recoveryState < RecoveryState::ACCEPTING_COMMITS &&
self->isMasterInRemoteDc(self)) {
if (rebootCloggedDcMaster && self->dbInfo->get().recoveryState < RecoveryState::ACCEPTING_COMMITS &&
self->isMasterInCloggedDc(self)) {
auto masterAddr = self->dbInfo->get().master.address();
auto* masterProc = g_simulator->getProcessByAddress(masterAddr);
TraceEvent("DbAvailableRebootRemoteMaster").detail("MasterAddr", masterAddr);
@ -174,15 +188,23 @@ struct GcGenerationsWorkload : TestWorkload {
TraceEvent("WaitingForDbAvailable")
.detail("Iteration", successfulReboots)
.detail("RecoveryState", self->dbInfo->get().recoveryState);
co_await self->dbAvailable(self, /*rebootRemoteDcMaster=*/true);
co_await self->dbAvailable(self, /*rebootCloggedDcMaster=*/true);
// Only reboot the master if it's in the primary DC. If it's in the clogged
// remote DC, recovery will stall because the master can't communicate with
// primary DC processes. Loop back and try again.
if (self->isMasterInRemoteDc(self)) {
// Only reboot the master if it's in the active DC. If it's in the clogged
// DC, recovery will stall because the master can't communicate with active
// DC processes. Force a new master election before retrying.
const bool forcedRetry = self->forceCloggedDcMasterRetry;
self->forceCloggedDcMasterRetry = false;
if (forcedRetry || self->isMasterInCloggedDc(self)) {
auto masterAddr = self->dbInfo->get().master.address();
auto* masterProc = g_simulator->getProcessByAddress(masterAddr);
TraceEvent("RetryingRemoteDcMaster")
.detail("Iteration", successfulReboots)
.detail("MasterAddr", self->dbInfo->get().master.address());
.detail("MasterAddr", masterAddr)
.detail("Forced", forcedRetry);
if (masterProc) {
g_simulator->rebootProcess(masterProc, ISimulator::KillType::Reboot);
}
continue;
}
@ -223,6 +245,11 @@ struct GcGenerationsWorkload : TestWorkload {
TraceEvent("GcGenerations").detail("StartTime", startTime).detail("EndTime", workloadEnd);
// Block TLog recovery while creating generations to test generation accumulation during recovery
ScopeExit cleanup([self]() {
self->unclogAll();
disableConnectionFailures("GcGenerations");
fdbSimulationPolicyState().disableTLogRecoveryFinish = false;
});
fdbSimulationPolicyState().disableTLogRecoveryFinish = true;
co_await self->generateMultipleTxnGenerations(self, cx);
@ -244,13 +271,13 @@ struct GcGenerationsWorkload : TestWorkload {
// Note: the remote DC is unclogged now, so any master (including remote DC)
// can coordinate recovery. No need for the primary-DC-only guard here.
while (self->dbInfo->get().logSystemConfig.oldTLogs.size() > 1) {
co_await self->dbAvailable(self, /*rebootRemoteDcMaster=*/false);
co_await self->dbAvailable(self, /*rebootCloggedDcMaster=*/false);
auto masterAddr = self->dbInfo->get().master.address();
TraceEvent("RebootMasterForGC").detail("Master", masterAddr);
g_simulator->rebootProcess(g_simulator->getProcessByAddress(masterAddr), ISimulator::KillType::Reboot);
// Give this recovery cycle time to GC before retrying.
co_await delay(60);
co_await self->dbAvailable(self, /*rebootRemoteDcMaster=*/false);
co_await self->dbAvailable(self, /*rebootCloggedDcMaster=*/false);
TraceEvent("GcGenerationsWaitingForReduction")
.detail("OldTLogs", self->dbInfo->get().logSystemConfig.oldTLogs.size())
.detail("RecoveryState", self->dbInfo->get().recoveryState);
@ -261,6 +288,7 @@ struct GcGenerationsWorkload : TestWorkload {
co_await self->dbInfo->onChange();
}
self->completed = true;
TraceEvent("GcGenerationsWorkloadFinish").log();
}
};

1
tests/slow/GcGenerations.toml
View File

@ -31,3 +31,4 @@ testTitle = 'GcGenerations'
[[test.workload]]
testName = 'GcGenerations'
testDuration = 1000.0
forceCloggedDcMasterRetry = true