MT#55283 redis: never delete a foreign call's key after a failed restore

Ref #2211

When json_restore_call() fails, it sends "DEL <call-id>" twice: on the
write connection, in its own DB, and on the notification connection, in
whatever DB the last keyspace notification selected. For a foreign call
(a keyspace notification, or the restore of a subscribed keyspace at
startup) that key belongs to the instance that owns the call, which
keeps writing it: a standby that fails to restore one record deletes the
owner's copy, and the call cannot be taken over any more. For an own
call, the DEL on the notification connection can hit a peer's DB.

Only delete an unrestorable record when the call is our own, and only on
the write connection, in our own DB. A foreign record that can never be
restored now stays until its owner rewrites or deletes it, or until it
expires (--redis-expires). Seen in a lab with Redis Sentinel, where a
subscriber's DEL was refused only because its node had just become a
replica.

Tests, in auto-daemon-tests-redis-subscribe.pl: a keyspace notification
for a foreign call whose record cannot be parsed is followed by no
command on either connection, and an own record that cannot be parsed at
startup is deleted once, on the write connection only. Without this
change both get a DEL on both connections.

Change-Id: I5d8a53f3515596cf3fc2c58ea97847476385b876
(cherry picked from commit 6b3bf8299b)
(cherry picked from commit fef185612e)

daemon/redis.c

@@ -2280,21 +2280,19 @@ err1:
 		if (c)
 			call_destroy(c);
 		release_closed_sockets();
-		if (must_release_pop) // avoid deadlock with redis_notify->lock below
+		if (must_release_pop) // avoid deadlock with rtpe_redis_write->lock below
 			redis_ports_release_pop(false);
 		must_release_pop = false;
 
-		mutex_lock(&rtpe_redis_write->lock);
+		// a foreign call's key is its owner's; ours is deleted in our own
+		// DB (the notification connection has the last notification's DB)
+		if (!foreign) {
+			mutex_lock(&rtpe_redis_write->lock);
 
-		redis_select_db(rtpe_redis_write, rtpe_redis_write->db);
+			redis_select_db(rtpe_redis_write, rtpe_redis_write->db);
 
-		redisCommandNR(rtpe_redis_write->ctx, "DEL " PB, PBSTR(callid));
-		mutex_unlock(&rtpe_redis_write->lock);
-
-		if (rtpe_redis_notify) {
-			mutex_lock(&rtpe_redis_notify->lock);
-			redisCommandNR(rtpe_redis_notify->ctx, "DEL " PB, PBSTR(callid));
-			mutex_unlock(&rtpe_redis_notify->lock);
+			redisCommandNR(rtpe_redis_write->ctx, "DEL " PB, PBSTR(callid));
+			mutex_unlock(&rtpe_redis_write->lock);
 		}
 	}
 	if (c)

t/auto-daemon-tests-redis-subscribe.pl

@@ -141,8 +141,46 @@ $NGCP::Rtpengine::AutoTest::launch_cb = sub {
 		"+PONG\r\n",
 		"PING 5");
 	redis_io($redis_fd, "*2\r\n\$4\r\nKEYS\r\n\$1\r\n*\r\n",
-		"*0\r\n",
+		"*1\r\n\$6\r\nbadown\r\n",
 		"KEYS 5");
+
+	# one own record, which cannot be restored: the restore thread's own
+	# connection reads it ...
+
+	my $restore_fd;
+	accept($restore_fd, $redis_listener) or die;
+	redis_io($restore_fd, "*1\r\n\$4\r\nPING\r\n",
+		"+PONG\r\n",
+		"PING 6");
+	redis_io($restore_fd, "*2\r\n\$6\r\nSELECT\r\n\$1\r\n2\r\n",
+		"+OK\r\n",
+		"SELECT 6");
+	redis_io($restore_fd, "*1\r\n\$4\r\nINFO\r\n",
+		"\$13\r\nrole:master\r\n\r\n",
+		"INFO 6");
+	redis_io($restore_fd, "*2\r\n\$4\r\nTYPE\r\n\$5\r\ncalls\r\n",
+		"+none\r\n",
+		"TYPE 6");
+	redis_io($restore_fd, "*2\r\n\$3\r\nGET\r\n\$6\r\nbadown\r\n",
+		"\$9\r\nnot-a-rec\r\n",
+		"GET 6");
+
+	# ... and the record is deleted from our own DB on the write connection,
+	# and on no other connection
+
+	redis_io($redis_fd, "*2\r\n\$3\r\nDEL\r\n\$6\r\nbadown\r\n",
+		":1\r\n",
+		"DEL of an unrestorable own record");
+	my $rin = '';
+	vec($rin, fileno($redis_notify), 1) = 1;
+	my $buf = '';
+	if (select(my $rout = $rin, undef, undef, 0.5) > 0) {
+		alarm(1);
+		recv($redis_notify, $buf, 1000, 0);
+		alarm(0);
+		send($redis_notify, ":1\r\n", 0);
+	}
+	is($buf, '', 'no DEL of an own record on the notification connection');
 };
 
 
@@ -150,6 +188,7 @@ autotest_start(qw(--config-file=none -t -1 -i 203.0.113.1
 			-n 2223 -f -L 7 -E
 			--redis=203.0.113.42:6379/2
 			--subscribe-keyspace=3
+			--redis-num-threads=1
 		))
 		or die;
 
@@ -742,4 +781,38 @@ $resp = rtpe_req('list', 'list', {});
 is($#{$resp->{calls}}, 0, 'calls len');
 is($resp->{calls}[0], cid(), 'call ID');
 
+
+
+# foreign call whose record cannot be restored: the key belongs to its owner,
+# so a failed restore must not delete it, on either connection
+
+send($redis_subscribe, "*4\r\n\$8\r\npmessage\r\n\$16\r\n__keyspace\@3__:*\r\n\$22\r\n__keyspace\@3__:badcall\r\n\$3\r\nset\r\n", 0);
+
+redis_io($redis_notify, "*1\r\n\$4\r\nPING\r\n",
+	"+PONG\r\n",
+	"PING");
+redis_io($redis_notify, "*2\r\n\$3\r\nGET\r\n\$7\r\nbadcall\r\n",
+	"\$9\r\nnot-a-rec\r\n",
+	"GET unrestorable");
+
+# nothing may follow on the write or the notification connection
+for my $fd ([$redis_fd, 'write connection'], [$redis_notify, 'notification connection']) {
+	my $rin = '';
+	vec($rin, fileno($fd->[0]), 1) = 1;
+	my $n = select(my $rout = $rin, undef, undef, 0.5);
+	my $buf = '';
+	if ($n > 0) {
+		alarm(1);
+		recv($fd->[0], $buf, 1000, 0);
+		alarm(0);
+		# unblock the daemon, which waits for a reply
+		send($fd->[0], ":1\r\n", 0);
+	}
+	is($buf, '', "no DEL of a foreign key on the $fd->[1]");
+}
+
+$resp = rtpe_req('list', 'list', {});
+is($#{$resp->{calls}}, 0, 'calls len');
+is($resp->{calls}[0], cid(), 'call ID');
+
 done_testing();