src/migrate.c

#include "redis.h"
#include "endianconv.h"

#include <sys/stat.h>
#include <lmdb.h>

/* -----------------------------------------------------------------------------
 * DUMP, RESTORE and MIGRATE commands
 * -------------------------------------------------------------------------- */

MDB_env *env;
MDB_dbi dbi;

/* Generates a DUMP-format representation of the object 'o', adding it to the
 * io stream pointed by 'rio'. This function can't fail. */
void createDumpPayload(rio *payload, robj *o) {
    unsigned char buf[2];
    uint64_t crc;

    /* Serialize the object in a RDB-like format. It consist of an object type
     * byte followed by the serialized object. This is understood by RESTORE. */
    rioInitWithBuffer(payload,sdsempty());
    redisAssert(rdbSaveObjectType(payload,o));
    redisAssert(rdbSaveObject(payload,o));

    /* Write the footer, this is how it looks like:
     * ----------------+---------------------+---------------+
     * ... RDB payload | 2 bytes RDB version | 8 bytes CRC64 |
     * ----------------+---------------------+---------------+
     * RDB version and CRC are both in little endian.
     */

    /* RDB version */
    buf[0] = REDIS_RDB_VERSION & 0xff;
    buf[1] = (REDIS_RDB_VERSION >> 8) & 0xff;
    payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,buf,2);

    /* CRC64 */
    crc = crc64(0,(unsigned char*)payload->io.buffer.ptr,
                sdslen(payload->io.buffer.ptr));
    memrev64ifbe(&crc);
    payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,&crc,8);
}

/* Verify that the RDB version of the dump payload matches the one of this Redis
 * instance and that the checksum is ok.
 * If the DUMP payload looks valid REDIS_OK is returned, otherwise REDIS_ERR
 * is returned. */
int verifyDumpPayload(unsigned char *p, size_t len) {
    unsigned char *footer;
    uint16_t rdbver;
    uint64_t crc;

    /* At least 2 bytes of RDB version and 8 of CRC64 should be present. */
    if (len < 10) return REDIS_ERR;
    footer = p+(len-10);

    /* Verify RDB version */
    rdbver = (footer[1] << 8) | footer[0];
    if (rdbver != REDIS_RDB_VERSION) return REDIS_ERR;

    /* Verify CRC64 */
    crc = crc64(0,p,len-8);
    memrev64ifbe(&crc);
    return (memcmp(&crc,footer+2,8) == 0) ? REDIS_OK : REDIS_ERR;
}

/* DUMP keyname
 * DUMP is actually not used by Redis Cluster but it is the obvious
 * complement of RESTORE and can be useful for different applications. */
void dumpCommand(redisClient *c) {
    robj *o, *dumpobj;
    rio payload;

    /* Check if the key is here. */
    if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
        addReply(c,shared.nullbulk);
        return;
    }

    /* Create the DUMP encoded representation. */
    createDumpPayload(&payload,o);

    /* Transfer to the client */
    dumpobj = createObject(REDIS_STRING,payload.io.buffer.ptr);
    addReplyBulk(c,dumpobj);
    decrRefCount(dumpobj);
    return;
}

/* RESTORE key ttl serialized-value */
void restoreCommand(redisClient *c) {
    long ttl;
    rio payload;
    int type;
    robj *obj;

    /* Make sure this key does not already exist here... */
    if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
        addReplyError(c,"Target key name is busy.");
        return;
    }

    /* Check if the TTL value makes sense */
    if (getLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != REDIS_OK) {
        return;
    } else if (ttl < 0) {
        addReplyError(c,"Invalid TTL value, must be >= 0");
        return;
    }

    /* Verify RDB version and data checksum. */
    if (verifyDumpPayload(c->argv[3]->ptr,sdslen(c->argv[3]->ptr)) == REDIS_ERR) {
        addReplyError(c,"DUMP payload version or checksum are wrong");
        return;
    }

    rioInitWithBuffer(&payload,c->argv[3]->ptr);
    if (((type = rdbLoadObjectType(&payload)) == -1) ||
        ((obj = rdbLoadObject(type,&payload)) == NULL))
    {
        addReplyError(c,"Bad data format");
        return;
    }

    /* Create the key and set the TTL if any */
    dbAdd(c->db,c->argv[1],obj);
    if (ttl) setExpire(c->db,c->argv[1],mstime()+ttl);
    signalModifiedKey(c->db,c->argv[1]);
    addReply(c,shared.ok);
    server.dirty++;
}

/* MIGRATE host port key dbid timeout */
void migrateCommand(redisClient *c) {
    int fd;
    long timeout;
    long dbid;
    long long ttl = 0, expireat;
    robj *o;
    rio cmd, payload;

    /* Sanity check */
    if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK)
        return;
    if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK)
        return;
    if (timeout <= 0) timeout = 1000;

    /* Check if the key is here. If not we reply with success as there is
     * nothing to migrate (for instance the key expired in the meantime), but
     * we include such information in the reply string. */
    if ((o = lookupKeyRead(c->db,c->argv[3])) == NULL) {
        addReplySds(c,sdsnew("+NOKEY\r\n"));
        return;
    }

    /* Connect */
    fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
                atoi(c->argv[2]->ptr));
    if (fd == -1) {
        addReplyErrorFormat(c,"Can't connect to target node: %s",
            server.neterr);
        return;
    }
    if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) {
        close(fd);
        addReplySds(c,sdsnew("-IOERR error or timeout connecting to the client\r\n"));
        return;
    }

    /* Create RESTORE payload and generate the protocol to call the command. */
    rioInitWithBuffer(&cmd,sdsempty());
    redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',2));
    redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"SELECT",6));
    redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid));

    expireat = getExpire(c->db,c->argv[3]);
    if (expireat != -1) {
        ttl = expireat-mstime();
        if (ttl < 1) ttl = 1;
    }
    redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',4));
    redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7));
    redisAssertWithInfo(c,NULL,c->argv[3]->encoding == REDIS_ENCODING_RAW);
    redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr)));
    redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,ttl));

    /* Finally the last argument that is the serailized object payload
     * in the DUMP format. */
    createDumpPayload(&payload,o);
    redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,payload.io.buffer.ptr,
                                sdslen(payload.io.buffer.ptr)));
    sdsfree(payload.io.buffer.ptr);

    /* Tranfer the query to the other node in 64K chunks. */
    {
        sds buf = cmd.io.buffer.ptr;
        size_t pos = 0, towrite;
        int nwritten = 0;

        while ((towrite = sdslen(buf)-pos) > 0) {
            towrite = (towrite > (64*1024) ? (64*1024) : towrite);
            nwritten = syncWrite(fd,buf+pos,towrite,timeout);
            if (nwritten != (signed)towrite) goto socket_wr_err;
            pos += nwritten;
        }
    }

    /* Read back the reply. */
    {
        char buf1[1024];
        char buf2[1024];

        /* Read the two replies */
        if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0)
            goto socket_rd_err;
        if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0)
            goto socket_rd_err;
        if (buf1[0] == '-' || buf2[0] == '-') {
            addReplyErrorFormat(c,"Target instance replied with error: %s",
                (buf1[0] == '-') ? buf1+1 : buf2+1);
        } else {
            robj *aux;

            dbDelete(c->db,c->argv[3]);
            signalModifiedKey(c->db,c->argv[3]);
            addReply(c,shared.ok);
            server.dirty++;

            /* Translate MIGRATE as DEL for replication/AOF. */
            aux = createStringObject("DEL",3);
            rewriteClientCommandVector(c,2,aux,c->argv[3]);
            decrRefCount(aux);
        }
    }

    sdsfree(cmd.io.buffer.ptr);
    close(fd);
    return;

socket_wr_err:
    addReplySds(c,sdsnew("-IOERR error or timeout writing to target instance\r\n"));
    sdsfree(cmd.io.buffer.ptr);
    close(fd);
    return;

socket_rd_err:
    addReplySds(c,sdsnew("-IOERR error or timeout reading from target node\r\n"));
    sdsfree(cmd.io.buffer.ptr);
    close(fd);
    return;
}

void stopKeyArchive(void) {
    redisAssert(env != NULL);

    mdb_dbi_close(env, dbi);
    mdb_env_close(env);
    env = NULL;

    server.mdb_state = REDIS_MDB_OFF;
}

int startKeyArchive(void) {
    redisAssert(env == NULL);

    int ret;

    ret = mdb_env_create(&env);
    if (ret != 0) return ret;

    ret = mdb_env_set_mapsize(env, server.mdb_mapsize);
    if (ret != 0) return ret;

    ret = mdb_env_set_maxdbs(env, 1);
    if (ret != 0) return ret;

    mkdir(server.mdb_environment, 0755);

    ret = mdb_env_open(env, server.mdb_environment, MDB_FIXEDMAP | MDB_NOSYNC, 0644);
    if (ret != 0) return ret;

    MDB_txn *txn;
    ret = mdb_txn_begin(env, NULL, 0, &txn);
    if (ret != 0) return ret;

    ret = mdb_open(txn, NULL, 0, &dbi);
    if (ret != 0) return ret;

    mdb_txn_commit(txn);

    server.mdb_state = REDIS_MDB_ON;
    return 0;
}

int archive(redisDb *db, robj *key) {
    if (server.mdb_state == REDIS_MDB_OFF)
        return 1;
    redisAssert(env != NULL);

    MDB_val kval;
    kval.mv_data = key->ptr;
    kval.mv_size = sdslen((sds)key->ptr);

    robj *object;
    object = lookupKey(db, key);
    if (object == NULL)
        return 0;

    if (object->archived != 0)
        return 1;

    rio payload;
    createDumpPayload(&payload, object);

    MDB_val dval;
    dval.mv_size = sdslen(payload.io.buffer.ptr);
    dval.mv_data = payload.io.buffer.ptr;

    int ret;

    MDB_txn *txn;
    ret = mdb_txn_begin(env, NULL, 0, &txn);
    if (ret != 0)
        goto archive_err;

    ret = mdb_put(txn, dbi, &kval, &dval, 0);
    if (ret != 0) {
        mdb_txn_abort(txn);
        goto archive_err;
    }

    mdb_txn_commit(txn);
    sdsfree(payload.io.buffer.ptr);
    return 1;

archive_err:
    sdsfree(payload.io.buffer.ptr);
    redisAssert(0);
    return 0;
}

robj *recover(redisDb *db, robj *key) {
    if (server.mdb_state == REDIS_MDB_OFF)
        return NULL;

    int ret;

    MDB_val kval;
    kval.mv_data = key->ptr;
    kval.mv_size = sdslen((sds)key->ptr);

    MDB_txn *txn;
    ret = mdb_txn_begin(env, NULL, 0, &txn);
    if (ret != 0)
        return NULL;

    MDB_cursor *cursor;
    ret = mdb_cursor_open(txn, dbi, &cursor);
    if (ret != 0) {
        mdb_txn_abort(txn);
        return NULL;
    }

    MDB_val pval;
    ret = mdb_cursor_get(cursor, &kval, &pval, MDB_SET);
    if (ret != 0) {
        mdb_txn_abort(txn);
        return NULL;
    }

    sds sval = sdsnewlen(pval.mv_data, pval.mv_size);
    mdb_cursor_close(cursor);
    mdb_txn_abort(txn);

    rio payload;
    rioInitWithBuffer(&payload, sval);

    int type = rdbLoadObjectType(&payload);
    if (type == -1)
        goto recover_err;

    robj *object = rdbLoadObject(type, &payload);
    if (object == NULL)
        goto recover_err;

    object->archived = 1;

    dbAdd(db, key, object);
    signalModifiedKey(db, key);
    server.dirty++;

    sdsfree(sval);
    return object;

recover_err:
    sdsfree(sval);
    return NULL;
}
Commit	Line	Data
	1	#include "redis.h"
	2	#include "endianconv.h"
	3
	4	#include <sys/stat.h>
	5	#include <lmdb.h>
	6
	7	/* -----------------------------------------------------------------------------
	8	* DUMP, RESTORE and MIGRATE commands
	9	* -------------------------------------------------------------------------- */
	10
	11	MDB_env *env;
	12	MDB_dbi dbi;
	13
	14	/* Generates a DUMP-format representation of the object 'o', adding it to the
	15	* io stream pointed by 'rio'. This function can't fail. */
	16	void createDumpPayload(rio payload, robj o) {
	17	unsigned char buf[2];
	18	uint64_t crc;
	19
	20	/* Serialize the object in a RDB-like format. It consist of an object type
	21	* byte followed by the serialized object. This is understood by RESTORE. */
	22	rioInitWithBuffer(payload,sdsempty());
	23	redisAssert(rdbSaveObjectType(payload,o));
	24	redisAssert(rdbSaveObject(payload,o));
	25
	26	/* Write the footer, this is how it looks like:
	27	* ----------------+---------------------+---------------+
	28	* ... RDB payload \| 2 bytes RDB version \| 8 bytes CRC64 \|
	29	* ----------------+---------------------+---------------+
	30	* RDB version and CRC are both in little endian.
	31	*/
	32
	33	/* RDB version */
	34	buf[0] = REDIS_RDB_VERSION & 0xff;
	35	buf[1] = (REDIS_RDB_VERSION >> 8) & 0xff;
	36	payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,buf,2);
	37
	38	/* CRC64 */
	39	crc = crc64(0,(unsigned char*)payload->io.buffer.ptr,
	40	sdslen(payload->io.buffer.ptr));
	41	memrev64ifbe(&crc);
	42	payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,&crc,8);
	43	}
	44
	45	/* Verify that the RDB version of the dump payload matches the one of this Redis
	46	* instance and that the checksum is ok.
	47	* If the DUMP payload looks valid REDIS_OK is returned, otherwise REDIS_ERR
	48	* is returned. */
	49	int verifyDumpPayload(unsigned char *p, size_t len) {
	50	unsigned char *footer;
	51	uint16_t rdbver;
	52	uint64_t crc;
	53
	54	/* At least 2 bytes of RDB version and 8 of CRC64 should be present. */
	55	if (len < 10) return REDIS_ERR;
	56	footer = p+(len-10);
	57
	58	/* Verify RDB version */
	59	rdbver = (footer[1] << 8) \| footer[0];
	60	if (rdbver != REDIS_RDB_VERSION) return REDIS_ERR;
	61
	62	/* Verify CRC64 */
	63	crc = crc64(0,p,len-8);
	64	memrev64ifbe(&crc);
	65	return (memcmp(&crc,footer+2,8) == 0) ? REDIS_OK : REDIS_ERR;
	66	}
	67
	68	/* DUMP keyname
	69	* DUMP is actually not used by Redis Cluster but it is the obvious
	70	* complement of RESTORE and can be useful for different applications. */
	71	void dumpCommand(redisClient *c) {
	72	robj o, dumpobj;
	73	rio payload;
	74
	75	/* Check if the key is here. */
	76	if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
	77	addReply(c,shared.nullbulk);
	78	return;
	79	}
	80
	81	/* Create the DUMP encoded representation. */
	82	createDumpPayload(&payload,o);
	83
	84	/* Transfer to the client */
	85	dumpobj = createObject(REDIS_STRING,payload.io.buffer.ptr);
	86	addReplyBulk(c,dumpobj);
	87	decrRefCount(dumpobj);
	88	return;
	89	}
	90
	91	/* RESTORE key ttl serialized-value */
	92	void restoreCommand(redisClient *c) {
	93	long ttl;
	94	rio payload;
	95	int type;
	96	robj *obj;
	97
	98	/* Make sure this key does not already exist here... */
	99	if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
	100	addReplyError(c,"Target key name is busy.");
	101	return;
	102	}
	103
	104	/* Check if the TTL value makes sense */
	105	if (getLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != REDIS_OK) {
	106	return;
	107	} else if (ttl < 0) {
	108	addReplyError(c,"Invalid TTL value, must be >= 0");
	109	return;
	110	}
	111
	112	/* Verify RDB version and data checksum. */
	113	if (verifyDumpPayload(c->argv[3]->ptr,sdslen(c->argv[3]->ptr)) == REDIS_ERR) {
	114	addReplyError(c,"DUMP payload version or checksum are wrong");
	115	return;
	116	}
	117
	118	rioInitWithBuffer(&payload,c->argv[3]->ptr);
	119	if (((type = rdbLoadObjectType(&payload)) == -1) \|\|
	120	((obj = rdbLoadObject(type,&payload)) == NULL))
	121	{
	122	addReplyError(c,"Bad data format");
	123	return;
	124	}
	125
	126	/* Create the key and set the TTL if any */
	127	dbAdd(c->db,c->argv[1],obj);
	128	if (ttl) setExpire(c->db,c->argv[1],mstime()+ttl);
	129	signalModifiedKey(c->db,c->argv[1]);
	130	addReply(c,shared.ok);
	131	server.dirty++;
	132	}
	133
	134	/* MIGRATE host port key dbid timeout */
	135	void migrateCommand(redisClient *c) {
	136	int fd;
	137	long timeout;
	138	long dbid;
	139	long long ttl = 0, expireat;
	140	robj *o;
	141	rio cmd, payload;
	142
	143	/* Sanity check */
	144	if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK)
	145	return;
	146	if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK)
	147	return;
	148	if (timeout <= 0) timeout = 1000;
	149
	150	/* Check if the key is here. If not we reply with success as there is
	151	* nothing to migrate (for instance the key expired in the meantime), but
	152	* we include such information in the reply string. */
	153	if ((o = lookupKeyRead(c->db,c->argv[3])) == NULL) {
	154	addReplySds(c,sdsnew("+NOKEY\r\n"));
	155	return;
	156	}
	157
	158	/* Connect */
	159	fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
	160	atoi(c->argv[2]->ptr));
	161	if (fd == -1) {
	162	addReplyErrorFormat(c,"Can't connect to target node: %s",
	163	server.neterr);
	164	return;
	165	}
	166	if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) {
	167	close(fd);
	168	addReplySds(c,sdsnew("-IOERR error or timeout connecting to the client\r\n"));
	169	return;
	170	}
	171
	172	/* Create RESTORE payload and generate the protocol to call the command. */
	173	rioInitWithBuffer(&cmd,sdsempty());
	174	redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',2));
	175	redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"SELECT",6));
	176	redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid));
	177
	178	expireat = getExpire(c->db,c->argv[3]);
	179	if (expireat != -1) {
	180	ttl = expireat-mstime();
	181	if (ttl < 1) ttl = 1;
	182	}
	183	redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',4));
	184	redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7));
	185	redisAssertWithInfo(c,NULL,c->argv[3]->encoding == REDIS_ENCODING_RAW);
	186	redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr)));
	187	redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,ttl));
	188
	189	/* Finally the last argument that is the serailized object payload
	190	* in the DUMP format. */
	191	createDumpPayload(&payload,o);
	192	redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,payload.io.buffer.ptr,
	193	sdslen(payload.io.buffer.ptr)));
	194	sdsfree(payload.io.buffer.ptr);
	195
	196	/* Tranfer the query to the other node in 64K chunks. */
	197	{
	198	sds buf = cmd.io.buffer.ptr;
	199	size_t pos = 0, towrite;
	200	int nwritten = 0;
	201
	202	while ((towrite = sdslen(buf)-pos) > 0) {
	203	towrite = (towrite > (641024) ? (641024) : towrite);
	204	nwritten = syncWrite(fd,buf+pos,towrite,timeout);
	205	if (nwritten != (signed)towrite) goto socket_wr_err;
	206	pos += nwritten;
	207	}
	208	}
	209
	210	/* Read back the reply. */
	211	{
	212	char buf1[1024];
	213	char buf2[1024];
	214
	215	/* Read the two replies */
	216	if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0)
	217	goto socket_rd_err;
	218	if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0)
	219	goto socket_rd_err;
	220	if (buf1[0] == '-' \|\| buf2[0] == '-') {
	221	addReplyErrorFormat(c,"Target instance replied with error: %s",
	222	(buf1[0] == '-') ? buf1+1 : buf2+1);
	223	} else {
	224	robj *aux;
	225
	226	dbDelete(c->db,c->argv[3]);
	227	signalModifiedKey(c->db,c->argv[3]);
	228	addReply(c,shared.ok);
	229	server.dirty++;
	230
	231	/* Translate MIGRATE as DEL for replication/AOF. */
	232	aux = createStringObject("DEL",3);
	233	rewriteClientCommandVector(c,2,aux,c->argv[3]);
	234	decrRefCount(aux);
	235	}
	236	}
	237
	238	sdsfree(cmd.io.buffer.ptr);
	239	close(fd);
	240	return;
	241
	242	socket_wr_err:
	243	addReplySds(c,sdsnew("-IOERR error or timeout writing to target instance\r\n"));
	244	sdsfree(cmd.io.buffer.ptr);
	245	close(fd);
	246	return;
	247
	248	socket_rd_err:
	249	addReplySds(c,sdsnew("-IOERR error or timeout reading from target node\r\n"));
	250	sdsfree(cmd.io.buffer.ptr);
	251	close(fd);
	252	return;
	253	}
	254
	255	void stopKeyArchive(void) {
	256	redisAssert(env != NULL);
	257
	258	mdb_dbi_close(env, dbi);
	259	mdb_env_close(env);
	260	env = NULL;
	261
	262	server.mdb_state = REDIS_MDB_OFF;
	263	}
	264
	265	int startKeyArchive(void) {
	266	redisAssert(env == NULL);
	267
	268	int ret;
	269
	270	ret = mdb_env_create(&env);
	271	if (ret != 0) return ret;
	272
	273	ret = mdb_env_set_mapsize(env, server.mdb_mapsize);
	274	if (ret != 0) return ret;
	275
	276	ret = mdb_env_set_maxdbs(env, 1);
	277	if (ret != 0) return ret;
	278
	279	mkdir(server.mdb_environment, 0755);
	280
	281	ret = mdb_env_open(env, server.mdb_environment, MDB_FIXEDMAP \| MDB_NOSYNC, 0644);
	282	if (ret != 0) return ret;
	283
	284	MDB_txn *txn;
	285	ret = mdb_txn_begin(env, NULL, 0, &txn);
	286	if (ret != 0) return ret;
	287
	288	ret = mdb_open(txn, NULL, 0, &dbi);
	289	if (ret != 0) return ret;
	290
	291	mdb_txn_commit(txn);
	292
	293	server.mdb_state = REDIS_MDB_ON;
	294	return 0;
	295	}
	296
	297	int archive(redisDb db, robj key) {
	298	if (server.mdb_state == REDIS_MDB_OFF)
	299	return 1;
	300	redisAssert(env != NULL);
	301
	302	MDB_val kval;
	303	kval.mv_data = key->ptr;
	304	kval.mv_size = sdslen((sds)key->ptr);
	305
	306	robj *object;
	307	object = lookupKey(db, key);
	308	if (object == NULL)
	309	return 0;
	310
	311	if (object->archived != 0)
	312	return 1;
	313
	314	rio payload;
	315	createDumpPayload(&payload, object);
	316
	317	MDB_val dval;
	318	dval.mv_size = sdslen(payload.io.buffer.ptr);
	319	dval.mv_data = payload.io.buffer.ptr;
	320
	321	int ret;
	322
	323	MDB_txn *txn;
	324	ret = mdb_txn_begin(env, NULL, 0, &txn);
	325	if (ret != 0)
	326	goto archive_err;
	327
	328	ret = mdb_put(txn, dbi, &kval, &dval, 0);
	329	if (ret != 0) {
	330	mdb_txn_abort(txn);
	331	goto archive_err;
	332	}
	333
	334	mdb_txn_commit(txn);
	335	sdsfree(payload.io.buffer.ptr);
	336	return 1;
	337
	338	archive_err:
	339	sdsfree(payload.io.buffer.ptr);
	340	redisAssert(0);
	341	return 0;
	342	}
	343
	344	robj recover(redisDb db, robj *key) {
	345	if (server.mdb_state == REDIS_MDB_OFF)
	346	return NULL;
	347
	348	int ret;
	349
	350	MDB_val kval;
	351	kval.mv_data = key->ptr;
	352	kval.mv_size = sdslen((sds)key->ptr);
	353
	354	MDB_txn *txn;
	355	ret = mdb_txn_begin(env, NULL, 0, &txn);
	356	if (ret != 0)
	357	return NULL;
	358
	359	MDB_cursor *cursor;
	360	ret = mdb_cursor_open(txn, dbi, &cursor);
	361	if (ret != 0) {
	362	mdb_txn_abort(txn);
	363	return NULL;
	364	}
	365
	366	MDB_val pval;
	367	ret = mdb_cursor_get(cursor, &kval, &pval, MDB_SET);
	368	if (ret != 0) {
	369	mdb_txn_abort(txn);
	370	return NULL;
	371	}
	372
	373	sds sval = sdsnewlen(pval.mv_data, pval.mv_size);
	374	mdb_cursor_close(cursor);
	375	mdb_txn_abort(txn);
	376
	377	rio payload;
	378	rioInitWithBuffer(&payload, sval);
	379
	380	int type = rdbLoadObjectType(&payload);
	381	if (type == -1)
	382	goto recover_err;
	383
	384	robj *object = rdbLoadObject(type, &payload);
	385	if (object == NULL)
	386	goto recover_err;
	387
	388	object->archived = 1;
	389
	390	dbAdd(db, key, object);
	391	signalModifiedKey(db, key);
	392	server.dirty++;
	393
	394	sdsfree(sval);
	395	return object;
	396
	397	recover_err:
	398	sdsfree(sval);
	399	return NULL;
	400	}