TT#14008 restore foreign calls from the appropriate DB

closes #1257

Change-Id: I299475a01470cc9d4910d9d304bdda9bbbee6e83

daemon/main.c

@@ -1063,13 +1063,16 @@ no_kernel:
 			// active-active mode: the main DB has our own calls, while
 			// the "notify" DB has the "foreign" calls. "foreign" DB goes
 			// first as the "owned" DB can do a stray update back to Redis
-			if (redis_restore(rtpe_redis_notify, 1))
-				ilog(LOG_WARN, "Unable to restore calls from the active-active peer");
-			if (redis_restore(rtpe_redis_write, 0))
+			for (GList *l = rtpe_config.redis_subscribed_keyspaces.head; l; l = l->next) {
+				int db = GPOINTER_TO_INT(l->data);
+				if (redis_restore(rtpe_redis_notify, 1, db))
+					ilog(LOG_WARN, "Unable to restore calls from the active-active peer");
+			}
+			if (redis_restore(rtpe_redis_write, 0, -1))
 				die("Refusing to continue without working Redis database");
 		}
 		else {
-			if (redis_restore(rtpe_redis, 0))
+			if (redis_restore(rtpe_redis, 0, -1))
 				die("Refusing to continue without working Redis database");
 		}
 

daemon/redis.c

@@ -691,7 +691,7 @@ static int redis_notify(struct redis *r) {
 	// subscribe to the values in the configured keyspaces
 	rwlock_lock_r(&rtpe_config.config_lock);
 	for (l = rtpe_config.redis_subscribed_keyspaces.head; l; l = l->next) {
-		redis_notify_subscribe_action(r, SUBSCRIBE_KEYSPACE, GPOINTER_TO_UINT(l->data));
+		redis_notify_subscribe_action(r, SUBSCRIBE_KEYSPACE, GPOINTER_TO_INT(l->data));
 	}
 	rwlock_unlock_r(&rtpe_config.config_lock);
 
@@ -1926,7 +1926,7 @@ static void restore_thread(void *call_p, void *ctx_p) {
 	mutex_unlock(&ctx->r_m);
 }
 
-int redis_restore(struct redis *r, int foreign) {
+int redis_restore(struct redis *r, int foreign, int db) {
 	redisReply *calls = NULL, *call;
 	int i, ret = -1;
 	GThreadPool *gtp;
@@ -1947,10 +1947,18 @@ int redis_restore(struct redis *r, int foreign) {
 		ret = 0;
 		goto err;
 	}
-	mutex_unlock(&r->lock);
+	if (db != -1)
+		redis_select_db(r, db);
 
 	calls = redis_get(r, REDIS_REPLY_ARRAY, "KEYS *");
 
+	if (db != -1)
+		redis_select_db(r, r->db);
+	else
+		db = r->db;
+
+	mutex_unlock(&r->lock);
+
 	if (!calls) {
 		rlog(LOG_ERR, "Could not retrieve call list from Redis: %s",
 				r->ctx ? r->ctx->errstr : "No redis context");
@@ -1962,7 +1970,7 @@ int redis_restore(struct redis *r, int foreign) {
 	ctx.foreign = foreign;
 	for (i = 0; i < rtpe_config.redis_num_threads; i++)
 		g_queue_push_tail(&ctx.r_q,
-				redis_new(&r->endpoint, r->db, r->auth, r->role, r->no_redis_required));
+				redis_new(&r->endpoint, db, r->auth, r->role, r->no_redis_required));
 	gtp = g_thread_pool_new(restore_thread, &ctx, rtpe_config.redis_num_threads, TRUE, NULL);
 
 	for (i = 0; i < calls->elements; i++) {

include/redis.h

@@ -108,7 +108,7 @@ void redis_delete_async_loop(void *d);
 
 struct redis *redis_new(const endpoint_t *, int, const char *, enum redis_role, int);
 void redis_close(struct redis *r);
-int redis_restore(struct redis *, int foreign);
+int redis_restore(struct redis *, int foreign, int db);
 void redis_update(struct call *, struct redis *);
 void redis_update_onekey(struct call *c, struct redis *r);
 void redis_delete(struct call *, struct redis *);