Commit 95ba01b5 authored by Oran Agra's avatar Oran Agra
Browse files

RESTORE ABSTTL won't store expired keys into the db (#7472)

Similarly to EXPIREAT with TTL in the past, which implicitly deletes the
key and return success, RESTORE should not store key that are already
expired into the db.
When used together with REPLACE it should emit a DEL to keyspace
notification and replication stream.

(cherry picked from commit 5977a948)
parent 7b21b8c3
...@@ -4988,7 +4988,8 @@ void restoreCommand(client *c) { ...@@ -4988,7 +4988,8 @@ void restoreCommand(client *c) {
} }
/* Make sure this key does not already exist here... */ /* Make sure this key does not already exist here... */
if (!replace && lookupKeyWrite(c->db,c->argv[1]) != NULL) { robj *key = c->argv[1];
if (!replace && lookupKeyWrite(c->db,key) != NULL) {
addReply(c,shared.busykeyerr); addReply(c,shared.busykeyerr);
return; return;
} }
...@@ -5010,24 +5011,37 @@ void restoreCommand(client *c) { ...@@ -5010,24 +5011,37 @@ void restoreCommand(client *c) {
rioInitWithBuffer(&payload,c->argv[3]->ptr); rioInitWithBuffer(&payload,c->argv[3]->ptr);
if (((type = rdbLoadObjectType(&payload)) == -1) || if (((type = rdbLoadObjectType(&payload)) == -1) ||
((obj = rdbLoadObject(type,&payload,c->argv[1]->ptr)) == NULL)) ((obj = rdbLoadObject(type,&payload,key->ptr)) == NULL))
{ {
addReplyError(c,"Bad data format"); addReplyError(c,"Bad data format");
return; return;
} }
/* Remove the old key if needed. */ /* Remove the old key if needed. */
if (replace) dbDelete(c->db,c->argv[1]); int deleted = 0;
if (replace)
deleted = dbDelete(c->db,key);
if (ttl && !absttl) ttl+=mstime();
if (ttl && checkAlreadyExpired(ttl)) {
if (deleted) {
rewriteClientCommandVector(c,2,shared.del,key);
signalModifiedKey(c,c->db,key);
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
server.dirty++;
}
addReply(c, shared.ok);
return;
}
/* Create the key and set the TTL if any */ /* Create the key and set the TTL if any */
dbAdd(c->db,c->argv[1],obj); dbAdd(c->db,key,obj);
if (ttl) { if (ttl) {
if (!absttl) ttl+=mstime(); setExpire(c,c->db,key,ttl);
setExpire(c,c->db,c->argv[1],ttl);
} }
objectSetLRUOrLFU(obj,lfu_freq,lru_idle,lru_clock,1000); objectSetLRUOrLFU(obj,lfu_freq,lru_idle,lru_clock,1000);
signalModifiedKey(c,c->db,c->argv[1]); signalModifiedKey(c,c->db,key);
notifyKeyspaceEvent(NOTIFY_GENERIC,"restore",c->argv[1],c->db->id); notifyKeyspaceEvent(NOTIFY_GENERIC,"restore",key,c->db->id);
addReply(c,shared.ok); addReply(c,shared.ok);
server.dirty++; server.dirty++;
} }
......
...@@ -475,6 +475,16 @@ void flushSlaveKeysWithExpireList(void) { ...@@ -475,6 +475,16 @@ void flushSlaveKeysWithExpireList(void) {
} }
} }
int checkAlreadyExpired(long long when) {
/* EXPIRE with negative TTL, or EXPIREAT with a timestamp into the past
* should never be executed as a DEL when load the AOF or in the context
* of a slave instance.
*
* Instead we add the already expired key to the database with expire time
* (possibly in the past) and wait for an explicit DEL from the master. */
return (when <= mstime() && !server.loading && !server.masterhost);
}
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
* Expires Commands * Expires Commands
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
...@@ -502,13 +512,7 @@ void expireGenericCommand(client *c, long long basetime, int unit) { ...@@ -502,13 +512,7 @@ void expireGenericCommand(client *c, long long basetime, int unit) {
return; return;
} }
/* EXPIRE with negative TTL, or EXPIREAT with a timestamp into the past if (checkAlreadyExpired(when)) {
* should never be executed as a DEL when load the AOF or in the context
* of a slave instance.
*
* Instead we take the other branch of the IF statement setting an expire
* (possibly in the past) and wait for an explicit DEL from the master. */
if (when <= mstime() && !server.loading && !server.masterhost) {
robj *aux; robj *aux;
int deleted = server.lazyfree_lazy_expire ? dbAsyncDelete(c->db,key) : int deleted = server.lazyfree_lazy_expire ? dbAsyncDelete(c->db,key) :
......
...@@ -2070,6 +2070,7 @@ void propagateExpire(redisDb *db, robj *key, int lazy); ...@@ -2070,6 +2070,7 @@ void propagateExpire(redisDb *db, robj *key, int lazy);
int expireIfNeeded(redisDb *db, robj *key); int expireIfNeeded(redisDb *db, robj *key);
long long getExpire(redisDb *db, robj *key); long long getExpire(redisDb *db, robj *key);
void setExpire(client *c, redisDb *db, robj *key, long long when); void setExpire(client *c, redisDb *db, robj *key, long long when);
int checkAlreadyExpired(long long when);
robj *lookupKey(redisDb *db, robj *key, int flags); robj *lookupKey(redisDb *db, robj *key, int flags);
robj *lookupKeyRead(redisDb *db, robj *key); robj *lookupKeyRead(redisDb *db, robj *key);
robj *lookupKeyWrite(redisDb *db, robj *key); robj *lookupKeyWrite(redisDb *db, robj *key);
......
...@@ -36,7 +36,18 @@ start_server {tags {"dump"}} { ...@@ -36,7 +36,18 @@ start_server {tags {"dump"}} {
assert {$ttl >= 2900 && $ttl <= 3100} assert {$ttl >= 2900 && $ttl <= 3100}
r get foo r get foo
} {bar} } {bar}
test {RESTORE with ABSTTL in the past} {
r set foo bar
set encoded [r dump foo]
set now [clock milliseconds]
r debug set-active-expire 0
r restore foo [expr $now-3000] $encoded absttl REPLACE
catch {r debug object foo} e
r debug set-active-expire 1
set e
} {ERR no such key}
test {RESTORE can set LRU} { test {RESTORE can set LRU} {
r set foo bar r set foo bar
set encoded [r dump foo] set encoded [r dump foo]
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment