| 2038 | } |
| 2039 | |
| 2040 | ACTOR Future<Void> TagPartitionedLogSystem::epochEnd(Reference<AsyncVar<Reference<ILogSystem>>> outLogSystem, |
| 2041 | UID dbgid, |
| 2042 | DBCoreState prevState, |
| 2043 | FutureStream<TLogRejoinRequest> rejoinRequests, |
| 2044 | LocalityData locality, |
| 2045 | bool* forceRecovery) { |
| 2046 | // Stops a co-quorum of tlogs so that no further versions can be committed until the DBCoreState coordination |
| 2047 | // state is changed Creates a new logSystem representing the (now frozen) epoch No other important side effects. |
| 2048 | // The writeQuorum in the master info is from the previous configuration |
| 2049 | |
| 2050 | if (!prevState.tLogs.size()) { |
| 2051 | // This is a brand new database |
| 2052 | auto logSystem = makeReference<TagPartitionedLogSystem>(dbgid, locality, 0); |
| 2053 | logSystem->logSystemType = prevState.logSystemType; |
| 2054 | logSystem->recoverAt = 0; |
| 2055 | logSystem->knownCommittedVersion = 0; |
| 2056 | logSystem->stopped = true; |
| 2057 | outLogSystem->set(logSystem); |
| 2058 | wait(Future<Void>(Never())); |
| 2059 | throw internal_error(); |
| 2060 | } |
| 2061 | |
| 2062 | if (*forceRecovery) { |
| 2063 | DBCoreState modifiedState = prevState; |
| 2064 | |
| 2065 | int8_t primaryLocality = -1; |
| 2066 | for (auto& coreSet : modifiedState.tLogs) { |
| 2067 | if (coreSet.isLocal && coreSet.locality >= 0 && coreSet.tLogLocalities[0].dcId() != locality.dcId()) { |
| 2068 | primaryLocality = coreSet.locality; |
| 2069 | break; |
| 2070 | } |
| 2071 | } |
| 2072 | |
| 2073 | bool foundRemote = false; |
| 2074 | int8_t remoteLocality = -1; |
| 2075 | int modifiedLogSets = 0; |
| 2076 | int removedLogSets = 0; |
| 2077 | if (primaryLocality >= 0) { |
| 2078 | auto copiedLogs = modifiedState.tLogs; |
| 2079 | for (auto& coreSet : copiedLogs) { |
| 2080 | if (coreSet.locality != primaryLocality && coreSet.locality >= 0) { |
| 2081 | foundRemote = true; |
| 2082 | remoteLocality = coreSet.locality; |
| 2083 | modifiedState.tLogs.clear(); |
| 2084 | modifiedState.tLogs.push_back(coreSet); |
| 2085 | modifiedState.tLogs[0].isLocal = true; |
| 2086 | modifiedState.logRouterTags = 0; |
| 2087 | modifiedLogSets++; |
| 2088 | break; |
| 2089 | } |
| 2090 | } |
| 2091 | |
| 2092 | while (!foundRemote && modifiedState.oldTLogData.size()) { |
| 2093 | for (auto& coreSet : modifiedState.oldTLogData[0].tLogs) { |
| 2094 | if (coreSet.locality != primaryLocality && coreSet.locality >= tagLocalitySpecial) { |
| 2095 | foundRemote = true; |
| 2096 | remoteLocality = coreSet.locality; |
| 2097 | modifiedState.tLogs.clear(); |
nothing calls this directly
no test coverage detected