gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix partial sync failures

Browse Source 
Former-commit-id: 7e9f7c0c4f520392a930ab72951e287f52c711ab
This commit is contained in:
John Sully 2021-12-26 05:16:58 -05:00
parent cc0417d790
commit 30bba7f7de
2 changed files with 24 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
src/replication.cpp
View File

@ -2165,18 +2165,20 @@ void saveMasterStatusToStorage(bool fShutdown)
g_pserver->metadataDb->insert("repl-offset", 11, &g_pserver->master_repl_offset, sizeof(g_pserver->master_repl_offset), true); g_pserver->metadataDb->insert("repl-offset", 11, &g_pserver->master_repl_offset, sizeof(g_pserver->master_repl_offset), true);
else else
g_pserver->metadataDb->insert("repl-offset", 11, &tmp, sizeof(g_pserver->master_repl_offset), true); g_pserver->metadataDb->insert("repl-offset", 11, &tmp, sizeof(g_pserver->master_repl_offset), true);
if (g_pserver->fActiveReplica || (!listLength(g_pserver->masters) && g_pserver->repl_backlog)) { if (g_pserver->fActiveReplica || (!listLength(g_pserver->masters) && g_pserver->repl_backlog)) {
g_pserver->metadataDb->insert("repl-stream-db", 14, g_pserver->replicaseldb == -1 ? 0 : &g_pserver->replicaseldb, int zero = 0;
g_pserver->replicaseldb == -1 ? 0 : sizeof(g_pserver->replicaseldb), true); g_pserver->metadataDb->insert("repl-stream-db", 14, g_pserver->replicaseldb == -1 ? &zero : &g_pserver->replicaseldb,
} sizeof(g_pserver->replicaseldb), true);
} else {
struct redisMaster *miFirst = (redisMaster*)(listLength(g_pserver->masters) ? listNodeValue(listFirst(g_pserver->masters)) : NULL);
struct redisMaster *miFirst = (redisMaster*)(listLength(g_pserver->masters) ? listNodeValue(listFirst(g_pserver->masters)) : NULL); if (miFirst && miFirst->master) {
g_pserver->metadataDb->insert("repl-stream-db", 14, &miFirst->master->db->id, sizeof(miFirst->master->db->id), true);
if (miFirst && miFirst->master) { }
g_pserver->metadataDb->insert("repl-stream-db", 14, &miFirst->master->db->id, sizeof(miFirst->master->db->id), true); else if (miFirst && miFirst->cached_master) {
} g_pserver->metadataDb->insert("repl-stream-db", 14, &miFirst->cached_master->db->id, sizeof(miFirst->cached_master->db->id), true);
else if (miFirst && miFirst->cached_master) { }
g_pserver->metadataDb->insert("repl-stream-db", 14, &miFirst->cached_master->db->id, sizeof(miFirst->cached_master->db->id), true);
} }
if (listLength(g_pserver->masters) == 0) { if (listLength(g_pserver->masters) == 0) {

18
src/server.cpp
View File

@ -3397,7 +3397,6 @@ int restartServer(int flags, mstime_t delay) {
if (flags & RESTART_SERVER_GRACEFULLY) { if (flags & RESTART_SERVER_GRACEFULLY) {
if (g_pserver->m_pstorageFactory) { if (g_pserver->m_pstorageFactory) {
saveMasterStatusToStorage(true);
for (int idb = 0; idb < cserver.dbnum; ++idb) { for (int idb = 0; idb < cserver.dbnum; ++idb) {
g_pserver->db[idb]->storageProviderDelete(); g_pserver->db[idb]->storageProviderDelete();
} }
@ -4003,14 +4002,21 @@ void initServer(void) {
} }
}); });
g_pserver->metadataDb->retrieve("repl-offset", 11, [&](const char *, size_t, const void *data, size_t cb){ g_pserver->metadataDb->retrieve("repl-offset", 11, [&](const char *, size_t, const void *data, size_t cb){
if (cb == sizeof(g_pserver->replid)) { if (cb == sizeof(g_pserver->master_repl_offset)) {
g_pserver->master_repl_offset = *(long long*)data; g_pserver->master_repl_offset = *(long long*)data;
} }
}); });
int repl_stream_db = -1;
g_pserver->metadataDb->retrieve("repl-stream-db", 14, [&](const char *, size_t, const void *data, size_t){
repl_stream_db = *(int*)data;
});
/* !!! AFTER THIS POINT WE CAN NO LONGER READ FROM THE META DB AS IT WILL BE OVERWRITTEN !!! */
// replicationCacheMasterUsingMyself triggers the overwrite
listIter li; listIter li;
listNode *ln; listNode *ln;
listRewind(g_pserver->masters, &li); listRewind(g_pserver->masters, &li);
while ((ln = listNext(&li))) while ((ln = listNext(&li)))
{ {
@ -4019,9 +4025,7 @@ void initServer(void) {
* information, in order to allow partial resynchronizations * information, in order to allow partial resynchronizations
* with masters. */ * with masters. */
replicationCacheMasterUsingMyself(mi); replicationCacheMasterUsingMyself(mi);
g_pserver->metadataDb->retrieve("repl-stream-db", 14, [&](const char *, size_t, const void *data, size_t){ selectDb(mi->cached_master, repl_stream_db);
selectDb(mi->cached_master, *(int*)data);
});
} }
} }
} }
@ -4031,6 +4035,8 @@ void initServer(void) {
{ {
g_pserver->db[idb]->storageProviderInitialize(); g_pserver->db[idb]->storageProviderInitialize();
} }
saveMasterStatusToStorage(false); // eliminate the repl-offset field
/* Initialize ACL default password if it exists */ /* Initialize ACL default password if it exists */
ACLUpdateDefaultUserPassword(g_pserver->requirepass); ACLUpdateDefaultUserPassword(g_pserver->requirepass);