PSYNC2: Save Lua scripts state into RDB file.

This is currently needed in order to fix #4483, but this can be
useful in other contexts, so maybe later we may want to remove the
conditionals and always save/load scripts.

Note that we are using the "lua" AUX field here, in order to guarantee
backward compatibility of the RDB file. The unknown AUX fields must be
discarded by past versions of Redis.

2 files changed, 48 insertions, 0 deletions

diff --git a/src/rdb.c b/src/rdb.c
index 00106cac4..d1495e79a 100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@ -943,6 +943,27 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
     }
     di = NULL; /* So that we don't release it again on error. */
 
+    /* If we are storing the replication information on disk, persist
+     * the script cache as well: on successful PSYNC after a restart, we need
+     * to be able to process any EVALSHA inside the replication backlog the
+     * master will send us. */
+    if (rsi && dictSize(server.lua_scripts)) {
+        di = dictGetIterator(server.lua_scripts);
+        while((de = dictNext(di)) != NULL) {
+            sds sha = dictGetKey(de);
+            robj *body = dictGetVal(de);
+            /* Concatenate the SHA1 and the Lua script together. Because the
+             * SHA1 is fixed length, we will always be able to load it back
+             * telling apart the name from the body. */
+            sds combo = sdsdup(sha);
+            combo = sdscatlen(combo,body->ptr,sdslen(body->ptr));
+            if (rdbSaveAuxField(rdb,"lua",3,combo,sdslen(combo)) == -1)
+                goto werr;
+            sdsfree(combo);
+        }
+        dictReleaseIterator(di);
+    }
+
     /* EOF opcode */
     if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr;
 
@@ -1589,6 +1610,32 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi) {
                 }
             } else if (!strcasecmp(auxkey->ptr,"repl-offset")) {
                 if (rsi) rsi->repl_offset = strtoll(auxval->ptr,NULL,10);
+            } else if (!strcasecmp(auxkey->ptr,"lua")) {
+                /* Load the string combining the function name and body
+                 * back in memory. The format is basically:
+                 * <sha><lua-script-bodybody>. To load it back we need
+                 * to create the function name as "f_<sha>" and load the
+                 * body as a Redis string object. */
+                sds combo = auxval->ptr;
+                if (sdslen(combo) < 40) {
+                    rdbExitReportCorruptRDB(
+                        "Lua script stored into the RDB file has invalid "
+                        "length < 40 bytes: '%s'", combo);
+                }
+                char funcname[42];
+                funcname[0] = 'f';
+                funcname[1] = '_';
+                memcpy(funcname+2,combo,40);
+                robj *body = createRawStringObject(combo+40,sdslen(combo)-40);
+
+                /* Register the function. */
+                if (luaCreateFunction(NULL,server.lua,funcname,body) == C_ERR) {
+                    rdbExitReportCorruptRDB(
+                        "Can't load Lua script from RDB file! "
+                        "Script SHA1: %.42s BODY: %s",
+                            combo, combo+42);
+                }
+                decrRefCount(body);
             } else {
                 /* We ignore fields we don't understand, as by AUX field
                  * contract. */
diff --git a/src/server.h b/src/server.h
index e3b56075a..a0c65355b 100644
--- a/src/server.h
+++ b/src/server.h
@@ -1781,6 +1781,7 @@ void scriptingInit(int setup);
 int ldbRemoveChild(pid_t pid);
 void ldbKillForkedSessions(void);
 int ldbPendingChildren(void);
+int luaCreateFunction(client *c, lua_State *lua, char *funcname, robj *body);
 
 /* Blocked clients */
 void processUnblockedClients(void);