Always replicate time-to-live(TTL) as absolute timestamps in milliseconds (#8474)

Till now, on replica full-sync we used to transfer absolute time for TTL,
however when a command arrived (EXPIRE or EXPIREAT),
we used to propagate it as is to replicas (possibly with relative time),
but always translate it to EXPIREAT (absolute time) to AOF.

This commit changes that and will always use absolute time for propagation.
see discussion in #8433

Furthermore, we Introduce new commands: `EXPIRETIME/PEXPIRETIME`
that allow extracting the absolute TTL time from a key.

10 files changed, 390 insertions, 214 deletions

diff --git a/src/aof.c b/src/aof.c
index 35856f49b..f4209c6c5 100644
--- a/src/aof.c
+++ b/src/aof.c
@@ -575,43 +575,7 @@ sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) {
     return dst;
 }
 
-/* Create the sds representation of a PEXPIREAT command, using
- * 'seconds' as time to live and 'cmd' to understand what command
- * we are translating into a PEXPIREAT.
- *
- * This command is used in order to translate EXPIRE and PEXPIRE commands
- * into PEXPIREAT command so that we retain precision in the append only
- * file, and the time is always absolute and not relative. */
-sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) {
-    long long when;
-    robj *argv[3];
-
-    /* Make sure we can use strtoll */
-    seconds = getDecodedObject(seconds);
-    when = strtoll(seconds->ptr,NULL,10);
-    /* Convert argument into milliseconds for EXPIRE, SETEX, EXPIREAT */
-    if (cmd->proc == expireCommand || cmd->proc == setexCommand ||
-        cmd->proc == expireatCommand)
-    {
-        when *= 1000;
-    }
-    /* Convert into absolute time for EXPIRE, PEXPIRE, SETEX, PSETEX */
-    if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
-        cmd->proc == setexCommand || cmd->proc == psetexCommand)
-    {
-        when += mstime();
-    }
-    decrRefCount(seconds);
-
-    argv[0] = shared.pexpireat;
-    argv[1] = key;
-    argv[2] = createStringObjectFromLongLong(when);
-    buf = catAppendOnlyGenericCommand(buf, 3, argv);
-    decrRefCount(argv[2]);
-    return buf;
-}
-
-void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
+void feedAppendOnlyFile(int dictid, robj **argv, int argc) {
     sds buf = sdsempty();
     /* The DB this command was targeting is not the same as the last command
      * we appended. To issue a SELECT command is needed. */
@@ -624,44 +588,9 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
         server.aof_selected_db = dictid;
     }
 
-    if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
-        cmd->proc == expireatCommand) {
-        /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */
-        buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
-    } else if (cmd->proc == setCommand && argc > 3) {
-        robj *pxarg = NULL;
-        /* When SET is used with EX/PX argument setGenericCommand propagates them with PX millisecond argument.
-         * So since the command arguments are re-written there, we can rely here on the index of PX being 3. */
-        if (!strcasecmp(argv[3]->ptr, "px")) {
-            pxarg = argv[4];
-        }
-        /* For AOF we convert SET key value relative time in milliseconds to SET key value absolute time in
-         * millisecond. Whenever the condition is true it implies that original SET has been transformed
-         * to SET PX with millisecond time argument so we do not need to worry about unit here.*/
-        if (pxarg) {
-            robj *millisecond = getDecodedObject(pxarg);
-            long long when = strtoll(millisecond->ptr,NULL,10);
-            when += mstime();
-
-            decrRefCount(millisecond);
-
-            robj *newargs[5];
-            newargs[0] = argv[0];
-            newargs[1] = argv[1];
-            newargs[2] = argv[2];
-            newargs[3] = shared.pxat;
-            newargs[4] = createStringObjectFromLongLong(when);
-            buf = catAppendOnlyGenericCommand(buf,5,newargs);
-            decrRefCount(newargs[4]);
-        } else {
-            buf = catAppendOnlyGenericCommand(buf,argc,argv);
-        }
-    } else {
-        /* All the other commands don't need translation or need the
-         * same translation already operated in the command vector
-         * for the replication itself. */
-        buf = catAppendOnlyGenericCommand(buf,argc,argv);
-    }
+    /* All commands should be propagated the same way in AOF as in replication.
+     * No need for AOF-specific translation. */
+    buf = catAppendOnlyGenericCommand(buf,argc,argv);
 
     /* Append to the AOF buffer. This will be flushed on disk just before
      * of re-entering the event loop, so before the client will get a
diff --git a/src/cluster.c b/src/cluster.c
index b29f3cdac..f34c33162 100644
--- a/src/cluster.c
+++ b/src/cluster.c
@@ -5206,6 +5206,13 @@ void restoreCommand(client *c) {
     dbAdd(c->db,key,obj);
     if (ttl) {
         setExpire(c,c->db,key,ttl);
+        if (!absttl) {
+            /* Propagate TTL as absolute timestamp */
+            robj *ttl_obj = createStringObjectFromLongLong(ttl);
+            rewriteClientCommandArgument(c,2,ttl_obj);
+            decrRefCount(ttl_obj);
+            rewriteClientCommandArgument(c,c->argc,shared.absttl);
+        }
     }
     objectSetLRUOrLFU(obj,lfu_freq,lru_idle,lru_clock,1000);
     signalModifiedKey(c,c->db,key);
diff --git a/src/expire.c b/src/expire.c
index 982301542..8996ae57e 100644
--- a/src/expire.c
+++ b/src/expire.c
@@ -539,6 +539,10 @@ void expireGenericCommand(client *c, long long basetime, int unit) {
     } else {
         setExpire(c,c->db,key,when);
         addReply(c,shared.cone);
+        /* Propagate as PEXPIREAT millisecond-timestamp */
+        robj *when_obj = createStringObjectFromLongLong(when);
+        rewriteClientCommandVector(c, 3, shared.pexpireat, key, when_obj);
+        decrRefCount(when_obj);
         signalModifiedKey(c,c->db,key);
         notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);
         server.dirty++;
@@ -566,8 +570,8 @@ void pexpireatCommand(client *c) {
     expireGenericCommand(c,0,UNIT_MILLISECONDS);
 }
 
-/* Implements TTL and PTTL */
-void ttlGenericCommand(client *c, int output_ms) {
+/* Implements TTL, PTTL and EXPIRETIME */
+void ttlGenericCommand(client *c, int output_ms, int output_abs) {
     long long expire, ttl = -1;
 
     /* If the key does not exist at all, return -2 */
@@ -575,11 +579,12 @@ void ttlGenericCommand(client *c, int output_ms) {
         addReplyLongLong(c,-2);
         return;
     }
+
     /* The key exists. Return -1 if it has no expire, or the actual
      * TTL value otherwise. */
     expire = getExpire(c->db,c->argv[1]);
     if (expire != -1) {
-        ttl = expire-mstime();
+        ttl = output_abs ? expire : expire-mstime();
         if (ttl < 0) ttl = 0;
     }
     if (ttl == -1) {
@@ -591,12 +596,22 @@ void ttlGenericCommand(client *c, int output_ms) {
 
 /* TTL key */
 void ttlCommand(client *c) {
-    ttlGenericCommand(c, 0);
+    ttlGenericCommand(c, 0, 0);
 }
 
 /* PTTL key */
 void pttlCommand(client *c) {
-    ttlGenericCommand(c, 1);
+    ttlGenericCommand(c, 1, 0);
+}
+
+/* EXPIRETIME key */
+void expiretimeCommand(client *c) {
+    ttlGenericCommand(c, 0, 1);
+}
+
+/* PEXPIRETIME key */
+void pexpiretimeCommand(client *c) {
+    ttlGenericCommand(c, 1, 1);
 }
 
 /* PERSIST key */
diff --git a/src/server.c b/src/server.c
index 835872aeb..428e4aef1 100644
--- a/src/server.c
+++ b/src/server.c
@@ -800,6 +800,14 @@ struct redisCommand redisCommandTable[] = {
      "read-only fast random @keyspace",
      0,NULL,1,1,1,0,0,0},
 
+    {"expiretime",expiretimeCommand,2,
+     "read-only fast random @keyspace",
+     0,NULL,1,1,1,0,0,0},
+
+    {"pexpiretime",pexpiretimeCommand,2,
+     "read-only fast random @keyspace",
+     0,NULL,1,1,1,0,0,0},
+
     {"persist",persistCommand,2,
      "write fast @keyspace",
      0,NULL,1,1,1,0,0,0},
@@ -2601,7 +2609,6 @@ void createSharedObjects(void) {
     shared.left = createStringObject("left",4);
     shared.right = createStringObject("right",5);
     shared.pxat = createStringObject("PXAT", 4);
-    shared.px = createStringObject("PX",2);
     shared.time = createStringObject("TIME",4);
     shared.retrycount = createStringObject("RETRYCOUNT",10);
     shared.force = createStringObject("FORCE",5);
@@ -2611,6 +2618,7 @@ void createSharedObjects(void) {
     shared.ping = createStringObject("ping",4);
     shared.setid = createStringObject("SETID",5);
     shared.keepttl = createStringObject("KEEPTTL",7);
+    shared.absttl = createStringObject("ABSTTL",6);
     shared.load = createStringObject("LOAD",4);
     shared.createconsumer = createStringObject("CREATECONSUMER",14);
     shared.getack = createStringObject("GETACK",6);
@@ -3574,6 +3582,8 @@ struct redisCommand *lookupCommandOrOriginal(sds name) {
 void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
                int flags)
 {
+    UNUSED(cmd);
+
     if (!server.replication_allowed)
         return;
 
@@ -3590,7 +3600,7 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
     serverAssert(!(areClientsPaused() && !server.client_pause_in_transaction));
 
     if (server.aof_state != AOF_OFF && flags & PROPAGATE_AOF)
-        feedAppendOnlyFile(cmd,dbid,argv,argc);
+        feedAppendOnlyFile(dbid,argv,argc);
     if (flags & PROPAGATE_REPL)
         replicationFeedSlaves(server.slaves,dbid,argv,argc);
 }
diff --git a/src/server.h b/src/server.h
index aa0f580c6..52778d1ed 100644
--- a/src/server.h
+++ b/src/server.h
@@ -983,7 +983,7 @@ struct sharedObjectsStruct {
     *rpop, *lpop, *lpush, *rpoplpush, *lmove, *blmove, *zpopmin, *zpopmax,
     *emptyscan, *multi, *exec, *left, *right, *hset, *srem, *xgroup, *xclaim,  
     *script, *replconf, *eval, *persist, *set, *pexpireat, *pexpire, 
-    *time, *pxat, *px, *retrycount, *force, *justid, 
+    *time, *pxat, *absttl, *retrycount, *force, *justid, 
     *lastid, *ping, *setid, *keepttl, *load, *createconsumer,
     *getack, *special_asterick, *special_equals, *default_username, *redacted,
     *select[PROTO_SHARED_SELECT_CMDS],
@@ -2068,7 +2068,7 @@ int bg_unlink(const char *filename);
 
 /* AOF persistence */
 void flushAppendOnlyFile(int force);
-void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc);
+void feedAppendOnlyFile(int dictid, robj **argv, int argc);
 void aofRemoveTempFile(pid_t childpid);
 int rewriteAppendOnlyFileBackground(void);
 int loadAppendOnlyFile(char *filename);
@@ -2561,6 +2561,8 @@ void getsetCommand(client *c);
 void ttlCommand(client *c);
 void touchCommand(client *c);
 void pttlCommand(client *c);
+void expiretimeCommand(client *c);
+void pexpiretimeCommand(client *c);
 void persistCommand(client *c);
 void replicaofCommand(client *c);
 void roleCommand(client *c);
diff --git a/src/t_string.c b/src/t_string.c
index db6f7042e..99843c863 100644
--- a/src/t_string.c
+++ b/src/t_string.c
@@ -72,26 +72,13 @@ static int checkStringLength(client *c, long long size) {
 #define OBJ_PXAT (1<<7)            /* Set if timestamp in ms is given */
 #define OBJ_PERSIST (1<<8)         /* Set if we need to remove the ttl */
 
-void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
-    long long milliseconds = 0, when = 0; /* initialized to avoid any harmness warning */
+/* Forward declaration */
+static int getExpireMillisecondsOrReply(client *c, robj *expire, int flags, int unit, long long *milliseconds);
 
-    if (expire) {
-        if (getLongLongFromObjectOrReply(c, expire, &milliseconds, NULL) != C_OK)
-            return;
-        if (milliseconds <= 0 || (unit == UNIT_SECONDS && milliseconds > LLONG_MAX / 1000)) {
-            /* Negative value provided or multiplication is gonna overflow. */
-            addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
-            return;
-        }
-        if (unit == UNIT_SECONDS) milliseconds *= 1000;
-        when = milliseconds;
-        if ((flags & OBJ_PX) || (flags & OBJ_EX))
-            when += mstime();
-        if (when <= 0) {
-            /* Overflow detected. */
-            addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
-            return;
-        }
+void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
+    long long milliseconds = 0; /* initialized to avoid any harmness warning */
+    if (expire && getExpireMillisecondsOrReply(c, expire, flags, unit, &milliseconds) != C_OK) {
+        return;
     }
 
     if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,key) != NULL) ||
@@ -108,24 +95,17 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
     genericSetKey(c,c->db,key, val,flags & OBJ_KEEPTTL,1);
     server.dirty++;
     notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
+
     if (expire) {
-        setExpire(c,c->db,key,when);
+        setExpire(c,c->db,key,milliseconds);
+        /* Propagate as SET Key Value PXAT millisecond-timestamp if there is
+         * EX/PX/EXAT/PXAT flag. */
+        robj *milliseconds_obj = createStringObjectFromLongLong(milliseconds);
+        rewriteClientCommandVector(c, 5, shared.set, key, val, shared.pxat, milliseconds_obj);
+        decrRefCount(milliseconds_obj);
         notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);
-
-        /* Propagate as SET Key Value PXAT millisecond-timestamp if there is EXAT/PXAT or
-         * propagate as SET Key Value PX millisecond if there is EX/PX flag.
-         *
-         * Additionally when we propagate the SET with PX (relative millisecond) we translate
-         * it again to SET with PXAT for the AOF.
-         *
-         * Additional care is required while modifying the argument order. AOF relies on the
-         * exp argument being at index 3. (see feedAppendOnlyFile)
-         * */
-        robj *exp = (flags & OBJ_PXAT) || (flags & OBJ_EXAT) ? shared.pxat : shared.px;
-        robj *millisecondObj = createStringObjectFromLongLong(milliseconds);
-        rewriteClientCommandVector(c,5,shared.set,key,val,exp,millisecondObj);
-        decrRefCount(millisecondObj);
     }
+
     if (!(flags & OBJ_SET_GET)) {
         addReply(c, ok_reply ? ok_reply : shared.ok);
     }
@@ -150,6 +130,45 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
     }
 }
 
+/*
+ * Extract the `expire` argument of a given GET/SET command as an absolute timestamp in milliseconds.
+ *
+ * "client" is the client that sent the `expire` argument.
+ * "expire" is the `expire` argument to be extracted.
+ * "flags" represents the behavior of the command (e.g. PX or EX).
+ * "unit" is the original unit of the given `expire` argument (e.g. UNIT_SECONDS).
+ * "milliseconds" is output argument.
+ *
+ * If return C_OK, "milliseconds" output argument will be set to the resulting absolute timestamp.
+ * If return C_ERR, an error reply has been added to the given client.
+ */
+static int getExpireMillisecondsOrReply(client *c, robj *expire, int flags, int unit, long long *milliseconds) {
+    int ret = getLongLongFromObjectOrReply(c, expire, milliseconds, NULL);
+    if (ret != C_OK) {
+        return ret;
+    }
+
+    if (*milliseconds <= 0 || (unit == UNIT_SECONDS && *milliseconds > LLONG_MAX / 1000)) {
+        /* Negative value provided or multiplication is gonna overflow. */
+        addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
+        return C_ERR;
+    }
+
+    if (unit == UNIT_SECONDS) *milliseconds *= 1000;
+
+    if ((flags & OBJ_PX) || (flags & OBJ_EX)) {
+        *milliseconds += mstime();
+    }
+
+    if (*milliseconds <= 0) {
+        /* Overflow detected. */
+        addReplyErrorFormat(c,"invalid expire time in %s",c->cmd->name);
+        return C_ERR;
+    }
+
+    return C_OK;
+}
+
 #define COMMAND_GET 0
 #define COMMAND_SET 1
 /*
@@ -338,26 +357,10 @@ void getexCommand(client *c) {
         return;
     }
 
-    long long milliseconds = 0, when = 0;
-
     /* Validate the expiration time value first */
-    if (expire) {
-        if (getLongLongFromObjectOrReply(c, expire, &milliseconds, NULL) != C_OK)
-            return;
-        if (milliseconds <= 0 || (unit == UNIT_SECONDS && milliseconds > LLONG_MAX / 1000)) {
-            /* Negative value provided or multiplication is gonna overflow. */
-            addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
-            return;
-        }
-        if (unit == UNIT_SECONDS) milliseconds *= 1000;
-        when = milliseconds;
-        if ((flags & OBJ_PX) || (flags & OBJ_EX))
-            when += mstime();
-        if (when <= 0) {
-            /* Overflow detected. */
-            addReplyErrorFormat(c, "invalid expire time in %s", c->cmd->name);
-            return;
-        }
+    long long milliseconds = 0;
+    if (expire && getExpireMillisecondsOrReply(c, expire, flags, unit, &milliseconds) != C_OK) {
+        return;
     }
 
     /* We need to do this before we expire the key or delete it */
@@ -377,12 +380,12 @@ void getexCommand(client *c) {
         notifyKeyspaceEvent(NOTIFY_GENERIC, "del", c->argv[1], c->db->id);
         server.dirty++;
     } else if (expire) {
-        setExpire(c,c->db,c->argv[1],when);
-        /* Propagate */
-        robj *exp = (flags & OBJ_PXAT) || (flags & OBJ_EXAT) ? shared.pexpireat : shared.pexpire;
-        robj* millisecondObj = createStringObjectFromLongLong(milliseconds);
-        rewriteClientCommandVector(c,3,exp,c->argv[1],millisecondObj);
-        decrRefCount(millisecondObj);
+        setExpire(c,c->db,c->argv[1],milliseconds);
+        /* Propagate as PXEXPIREAT millisecond-timestamp if there is
+         * EX/PX/EXAT/PXAT flag and the key has not expired. */
+        robj *milliseconds_obj = createStringObjectFromLongLong(milliseconds);
+        rewriteClientCommandVector(c,3,shared.pexpireat,c->argv[1],milliseconds_obj);
+        decrRefCount(milliseconds_obj);
         signalModifiedKey(c, c->db, c->argv[1]);
         notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",c->argv[1],c->db->id);
         server.dirty++;
diff --git a/tests/cluster/tests/14-consistency-check.tcl b/tests/cluster/tests/14-consistency-check.tcl
index ddc0570e6..d182693d8 100644
--- a/tests/cluster/tests/14-consistency-check.tcl
+++ b/tests/cluster/tests/14-consistency-check.tcl
@@ -53,7 +53,7 @@ proc test_slave_load_expired_keys {aof} {
 
         # config the replica persistency and rewrite the config file to survive restart
         # note that this needs to be done before populating the volatile keys since
-        # that triggers and AOFRW, and we rather the AOF file to have SETEX commands
+        # that triggers and AOFRW, and we rather the AOF file to have 'SET PXAT' commands
         # rather than an RDB with volatile keys
         R $replica_id config set appendonly $aof
         R $replica_id config rewrite
diff --git a/tests/integration/aof.tcl b/tests/integration/aof.tcl
index abe2dc10c..4d427bc97 100644
--- a/tests/integration/aof.tcl
+++ b/tests/integration/aof.tcl
@@ -228,10 +228,10 @@ tags {"aof"} {
         }
     }
 
-    ## Test that EXPIREAT is loaded correctly
+    ## Test that PEXPIREAT is loaded correctly
     create_aof {
         append_to_aof [formatCommand rpush list foo]
-        append_to_aof [formatCommand expireat list 1000]
+        append_to_aof [formatCommand pexpireat list 1000]
         append_to_aof [formatCommand rpush list bar]
     }
 
diff --git a/tests/support/util.tcl b/tests/support/util.tcl
index 0f185d57a..8afb30cc4 100644
--- a/tests/support/util.tcl
+++ b/tests/support/util.tcl
@@ -776,3 +776,29 @@ proc punsubscribe {client {channels {}}} {
     $client punsubscribe {*}$channels
     consume_subscribe_messages $client punsubscribe $channels
 }
+
+proc read_from_aof {fp} {
+    # Input fp is a blocking binary file descriptor of an opened AOF file.
+    if {[gets $fp count] == -1} return ""
+    set count [string range $count 1 end]
+
+    # Return a list of arguments for the command.
+    set res {}
+    for {set j 0} {$j < $count} {incr j} {
+        read $fp 1
+        set arg [::redis::redis_bulk_read $fp]
+        if {$j == 0} {set arg [string tolower $arg]}
+        lappend res $arg
+    }
+    return $res
+}
+
+proc assert_aof_content {aof_path patterns} {
+    set fp [open $aof_path r]
+    fconfigure $fp -translation binary
+    fconfigure $fp -blocking 1
+
+    for {set j 0} {$j < [llength $patterns]} {incr j} {
+        assert_match [lindex $patterns $j] [read_from_aof $fp]
+    }
+}
diff --git a/tests/unit/expire.tcl b/tests/unit/expire.tcl
index 459503adc..a52299cde 100644
--- a/tests/unit/expire.tcl
+++ b/tests/unit/expire.tcl
@@ -142,16 +142,30 @@ start_server {tags {"expire"}} {
         assert {$ttl > 900 && $ttl <= 1000}
     }
 
-    test {TTL / PTTL return -1 if key has no expire} {
+    test {TTL / PTTL / EXPIRETIME / PEXPIRETIME return -1 if key has no expire} {
         r del x
         r set x hello
-        list [r ttl x] [r pttl x]
-    } {-1 -1}
+        list [r ttl x] [r pttl x] [r expiretime x] [r pexpiretime x]
+    } {-1 -1 -1 -1}
 
-    test {TTL / PTTL return -2 if key does not exit} {
+    test {TTL / PTTL / EXPIRETIME / PEXPIRETIME return -2 if key does not exit} {
         r del x
-        list [r ttl x] [r pttl x]
-    } {-2 -2}
+        list [r ttl x] [r pttl x] [r expiretime x] [r pexpiretime x]
+    } {-2 -2 -2 -2}
+
+    test {EXPIRETIME returns absolute expiration time in seconds} {
+        r del x
+        set abs_expire [expr [clock seconds] + 100]
+        r set x somevalue exat $abs_expire
+        assert_equal [r expiretime x] $abs_expire
+    }
+
+    test {PEXPIRETIME returns absolute expiration time in milliseconds} {
+        r del x
+        set abs_expire [expr [clock milliseconds] + 100000]
+        r set x somevalue pxat $abs_expire
+        assert_equal [r pexpiretime x] $abs_expire
+    }
 
     test {Redis should actively expire keys incrementally} {
         r flushdb
@@ -264,62 +278,159 @@ start_server {tags {"expire"}} {
         r ttl foo
     } {-2}
 
-    test {EXPIRE and SET/GETEX EX/PX/EXAT/PXAT option, TTL should not be reset after loadaof} {
-        # This test makes sure that expire times are propagated as absolute
-        # times to the AOF file and not as relative time, so that when the AOF
-        # is reloaded the TTLs are not being shifted forward to the future.
-        # We want the time to logically pass when the server is restarted!
-
-        r config set appendonly yes
-        r set foo1 bar EX 100
-        r set foo2 bar PX 100000
-        r set foo3 bar
-        r set foo4 bar
-        r expire foo3 100
-        r pexpire foo4 100000
-        r setex foo5 100 bar
-        r psetex foo6 100000 bar
-        r set foo7 bar EXAT [expr [clock seconds] + 100]
-        r set foo8 bar PXAT [expr [clock milliseconds] + 100000]
-        r set foo9 bar
-        r getex foo9 EX 100
-        r set foo10 bar
-        r getex foo10 PX 100000
-        r set foo11 bar
-        r getex foo11 EXAT [expr [clock seconds] + 100]
-        r set foo12 bar
-        r getex foo12 PXAT [expr [clock milliseconds] + 100000]
-
-        after 2000
-        r debug loadaof
-        assert_range [r ttl foo1] 90 98
-        assert_range [r ttl foo2] 90 98
-        assert_range [r ttl foo3] 90 98
-        assert_range [r ttl foo4] 90 98
-        assert_range [r ttl foo5] 90 98
-        assert_range [r ttl foo6] 90 98
-        assert_range [r ttl foo7] 90 98
-        assert_range [r ttl foo8] 90 98
-        assert_range [r ttl foo9] 90 98
-        assert_range [r ttl foo10] 90 98
-        assert_range [r ttl foo11] 90 98
-        assert_range [r ttl foo12] 90 98
+    # Start a new server with empty data and AOF file.
+    start_server {overrides {appendonly {yes} appendfilename {appendonly.aof} appendfsync always}} {
+        test {All time-to-live(TTL) in commands are propagated as absolute timestamp in milliseconds in AOF} {
+            # This test makes sure that expire times are propagated as absolute
+            # times to the AOF file and not as relative time, so that when the AOF
+            # is reloaded the TTLs are not being shifted forward to the future.
+            # We want the time to logically pass when the server is restarted!
+
+            set aof [file join [lindex [r config get dir] 1] [lindex [r config get appendfilename] 1]]
+
+            # Apply each TTL-related command to a unique key
+            # SET commands
+            r set foo1 bar ex 100
+            r set foo2 bar px 100000
+            r set foo3 bar exat [expr [clock seconds]+100]
+            r set foo4 bar pxat [expr [clock milliseconds]+100000]
+            r setex foo5 100 bar
+            r psetex foo6 100000 bar
+            # EXPIRE-family commands
+            r set foo7 bar
+            r expire foo7 100
+            r set foo8 bar
+            r pexpire foo8 100000
+            r set foo9 bar
+            r expireat foo9 [expr [clock seconds]+100]
+            r set foo10 bar
+            r pexpireat foo10 [expr [clock seconds]*1000+100000]
+            r set foo11 bar
+            r expireat foo11 [expr [clock seconds]-100]
+            # GETEX commands
+            r set foo12 bar
+            r getex foo12 ex 100
+            r set foo13 bar
+            r getex foo13 px 100000
+            r set foo14 bar
+            r getex foo14 exat [expr [clock seconds]+100]
+            r set foo15 bar
+            r getex foo15 pxat [expr [clock milliseconds]+100000]
+            # RESTORE commands
+            r set foo16 bar
+            set encoded [r dump foo16]
+            r restore foo17 100000 $encoded
+            r restore foo18 [expr [clock milliseconds]+100000] $encoded absttl
+
+            # Assert that each TTL-relatd command are persisted with absolute timestamps in AOF
+            assert_aof_content $aof {
+                {select *}
+                {set foo1 bar PXAT *}
+                {set foo2 bar PXAT *}
+                {set foo3 bar PXAT *}
+                {set foo4 bar PXAT *}
+                {set foo5 bar PXAT *}
+                {set foo6 bar PXAT *}
+                {set foo7 bar}
+                {pexpireat foo7 *}
+                {set foo8 bar}
+                {pexpireat foo8 *}
+                {set foo9 bar}
+                {pexpireat foo9 *}
+                {set foo10 bar}
+                {pexpireat foo10 *}
+                {set foo11 bar}
+                {del foo11}
+                {set foo12 bar}
+                {pexpireat foo12 *}
+                {set foo13 bar}
+                {pexpireat foo13 *}
+                {set foo14 bar}
+                {pexpireat foo14 *}
+                {set foo15 bar}
+                {pexpireat foo15 *}
+                {set foo16 bar}
+                {restore foo17 * {*} ABSTTL}
+                {restore foo18 * {*} absttl}
+            }
+
+            # Remember the absolute TTLs of all the keys
+            set ttl1 [r pexpiretime foo1]
+            set ttl2 [r pexpiretime foo2]
+            set ttl3 [r pexpiretime foo3]
+            set ttl4 [r pexpiretime foo4]
+            set ttl5 [r pexpiretime foo5]
+            set ttl6 [r pexpiretime foo6]
+            set ttl7 [r pexpiretime foo7]
+            set ttl8 [r pexpiretime foo8]
+            set ttl9 [r pexpiretime foo9]
+            set ttl10 [r pexpiretime foo10]
+            assert_equal "-2" [r pexpiretime foo11] ; # foo11 is gone
+            set ttl12 [r pexpiretime foo12]
+            set ttl13 [r pexpiretime foo13]
+            set ttl14 [r pexpiretime foo14]
+            set ttl15 [r pexpiretime foo15]
+            assert_equal "-1" [r pexpiretime foo16] ; # foo16 has no TTL
+            set ttl17 [r pexpiretime foo17]
+            set ttl18 [r pexpiretime foo18]
+
+            # Let some time pass and reload data from AOF
+            after 2000
+            r debug loadaof
+
+            # Assert that relative TTLs are roughly the same
+            assert_range [r ttl foo1] 90 98
+            assert_range [r ttl foo2] 90 98
+            assert_range [r ttl foo3] 90 98
+            assert_range [r ttl foo4] 90 98
+            assert_range [r ttl foo5] 90 98
+            assert_range [r ttl foo6] 90 98
+            assert_range [r ttl foo7] 90 98
+            assert_range [r ttl foo8] 90 98
+            assert_range [r ttl foo9] 90 98
+            assert_range [r ttl foo10] 90 98
+            assert_equal [r ttl foo11] "-2" ; # foo11 is gone
+            assert_range [r ttl foo12] 90 98
+            assert_range [r ttl foo13] 90 98
+            assert_range [r ttl foo14] 90 98
+            assert_range [r ttl foo15] 90 98
+            assert_equal [r ttl foo16] "-1" ; # foo16 has no TTL
+            assert_range [r ttl foo17] 90 98
+            assert_range [r ttl foo18] 90 98
+
+            # Assert that all keys have restored the same absolute TTLs from AOF
+            assert_equal [r pexpiretime foo1] $ttl1
+            assert_equal [r pexpiretime foo2] $ttl2
+            assert_equal [r pexpiretime foo3] $ttl3
+            assert_equal [r pexpiretime foo4] $ttl4
+            assert_equal [r pexpiretime foo5] $ttl5
+            assert_equal [r pexpiretime foo6] $ttl6
+            assert_equal [r pexpiretime foo7] $ttl7
+            assert_equal [r pexpiretime foo8] $ttl8
+            assert_equal [r pexpiretime foo9] $ttl9
+            assert_equal [r pexpiretime foo10] $ttl10
+            assert_equal [r pexpiretime foo11] "-2" ; # foo11 is gone
+            assert_equal [r pexpiretime foo12] $ttl12
+            assert_equal [r pexpiretime foo13] $ttl13
+            assert_equal [r pexpiretime foo14] $ttl14
+            assert_equal [r pexpiretime foo15] $ttl15
+            assert_equal [r pexpiretime foo16] "-1" ; # foo16 has no TTL
+            assert_equal [r pexpiretime foo17] $ttl17
+            assert_equal [r pexpiretime foo18] $ttl18
+        }
     }
 
-    test {EXPIRE relative and absolute propagation to replicas} {
-        # Make sure that relative and absolute expire commands are propagated
-        # "as is" to replicas.
-        # We want replicas to honor the same high level contract of expires that
-        # the master has, that is, we want the time to be counted logically
-        # starting from the moment the write was received. This usually provides
-        # the most coherent behavior from the point of view of the external
-        # users, with TTLs that are similar from the POV of the external observer.
-        #
-        # This test is here to stop some innocent / eager optimization or cleanup
-        # from doing the wrong thing without proper discussion, see:
-        # https://github.com/redis/redis/pull/5171#issuecomment-409553266
+    test {All TTL in commands are propagated as absolute timestamp in replication stream} {
+        # Make sure that both relative and absolute expire commands are propagated
+        # as absolute to replicas for two reasons:
+        # 1) We want to avoid replicas retaining data much longer than primary due
+        #    to replication lag.
+        # 2) We want to unify the way TTLs are replicated in both RDB and replication
+        #    stream, which is as absolute timestamps.
+        # See: https://github.com/redis/redis/issues/8433
 
         set repl [attach_to_replication_stream]
+        # SET commands
         r set foo1 bar ex 200
         r set foo1 bar px 100000
         r set foo1 bar exat [expr [clock seconds]+100]
@@ -327,37 +438,110 @@ start_server {tags {"expire"}} {
         r setex foo1 100 bar
         r psetex foo1 100000 bar
         r set foo2 bar
+        # EXPIRE-family commands
         r expire foo2 100
         r pexpire foo2 100000
         r set foo3 bar
         r expireat foo3 [expr [clock seconds]+100]
         r pexpireat foo3 [expr [clock seconds]*1000+100000]
         r expireat foo3 [expr [clock seconds]-100]
+        # GETEX-family commands
         r set foo4 bar
         r getex foo4 ex 200
         r getex foo4 px 200000
         r getex foo4 exat [expr [clock seconds]+100]
         r getex foo4 pxat [expr [clock milliseconds]+10000]
+        # RESTORE commands
+        r set foo5 bar
+        set encoded [r dump foo5]
+        r restore foo6 100000 $encoded
+        r restore foo7 [expr [clock milliseconds]+100000] $encoded absttl
+
         assert_replication_stream $repl {
             {select *}
-            {set foo1 bar PX 200000}
-            {set foo1 bar PX 100000}
             {set foo1 bar PXAT *}
             {set foo1 bar PXAT *}
-            {set foo1 bar PX 100000}
-            {set foo1 bar PX 100000}
+            {set foo1 bar PXAT *}
+            {set foo1 bar PXAT *}
+            {set foo1 bar PXAT *}
+            {set foo1 bar PXAT *}
             {set foo2 bar}
-            {expire foo2 100}
-            {pexpire foo2 100000}
+            {pexpireat foo2 *}
+            {pexpireat foo2 *}
             {set foo3 bar}
-            {expireat foo3 *}
+            {pexpireat foo3 *}
             {pexpireat foo3 *}
             {del foo3}
             {set foo4 bar}
-            {pexpire foo4 200000}
-            {pexpire foo4 200000}
             {pexpireat foo4 *}
             {pexpireat foo4 *}
+            {pexpireat foo4 *}
+            {pexpireat foo4 *}
+            {set foo5 bar}
+            {restore foo6 * {*} ABSTTL}
+            {restore foo7 * {*} absttl}
+        }
+    }
+
+    # Start another server to test replication of TTLs
+    start_server {} {
+        # Set the outer layer server as primary
+        set primary [srv -1 client]
+        set primary_host [srv -1 host]
+        set primary_port [srv -1 port]
+        # Set this inner layer server as replica
+        set replica [srv 0 client]
+
+        test {First server should have role slave after REPLICAOF} {
+            $replica replicaof $primary_host $primary_port
+            wait_for_condition 50 100 {
+                [s 0 role] eq {slave}
+            } else {
+                fail "Replication not started."
+            }
+        }
+
+        test {For all replicated TTL-related commands, absolute expire times are identical on primary and replica} {
+            # Apply each TTL-related command to a unique key on primary
+            # SET commands
+            $primary set foo1 bar ex 100
+            $primary set foo2 bar px 100000
+            $primary set foo3 bar exat [expr [clock seconds]+100]
+            $primary set foo4 bar pxat [expr [clock milliseconds]+100000]
+            $primary setex foo5 100 bar
+            $primary psetex foo6 100000 bar
+            # EXPIRE-family commands
+            $primary set foo7 bar
+            $primary expire foo7 100
+            $primary set foo8 bar
+            $primary pexpire foo8 100000
+            $primary set foo9 bar
+            $primary expireat foo9 [expr [clock seconds]+100]
+            $primary set foo10 bar
+            $primary pexpireat foo10 [expr [clock milliseconds]+100000]
+            # GETEX commands
+            $primary set foo11 bar
+            $primary getex foo11 ex 100
+            $primary set foo12 bar
+            $primary getex foo12 px 100000
+            $primary set foo13 bar
+            $primary getex foo13 exat [expr [clock seconds]+100]
+            $primary set foo14 bar
+            $primary getex foo14 pxat [expr [clock milliseconds]+100000]
+            # RESTORE commands
+            $primary set foo15 bar
+            set encoded [$primary dump foo15]
+            $primary restore foo16 100000 $encoded
+            $primary restore foo17 [expr [clock milliseconds]+100000] $encoded absttl
+
+            # Wait for replica to get the keys and TTLs
+            assert {[$primary wait 1 0] == 1}
+
+            # Verify absolute TTLs are identical on primary and replica for all keys
+            # This is because TTLs are always replicated as absolute values
+            foreach key [$primary keys *] {
+                assert_equal [$primary pexpiretime $key] [$replica pexpiretime $key]
+            }
         }
     }
 
@@ -406,7 +590,7 @@ start_server {tags {"expire"}} {
        r getex foo exat [expr [clock seconds]-100]
        assert_replication_stream $repl {
            {select *}
-           {set foo bar PX 100000}
+           {set foo bar PXAT *}
            {persist foo}
            {del foo}
         }