diff --git a/src/Makefile b/src/Makefile index 504f35592a1..54c46f56761 100644 --- a/src/Makefile +++ b/src/Makefile @@ -604,6 +604,7 @@ ENGINE_SERVER_OBJ = \ crdt/hlc.o \ crdt/merge.o \ crdt/peer.o \ + rdb_forkless.o \ rdma.o \ release.o \ replication.o \ diff --git a/src/config.c b/src/config.c index 67107c63894..4492828aee8 100644 --- a/src/config.c +++ b/src/config.c @@ -189,6 +189,10 @@ configEnum bgsave_method_enum[] = {{"fork", RDB_BGSAVE_TYPE_FORK}, {"forkless", RDB_BGSAVE_TYPE_FORKLESS}, {NULL, 0}}; +configEnum snapshot_mode_enum[] = {{"fork", SNAPSHOT_MODE_FORK}, + {"forkless", SNAPSHOT_MODE_FORKLESS}, + {NULL, 0}}; + configEnum cluster_replica_no_failover_enum[] = {{"no", CLUSTER_REPLICA_NO_FAILOVER_NO}, {"yes", CLUSTER_REPLICA_NO_FAILOVER_YES}, {"if-empty", CLUSTER_REPLICA_NO_FAILOVER_IF_EMPTY}, @@ -2475,8 +2479,8 @@ static void numericConfigRewrite(standardConfig *config, const char *name, struc } #define createSpecialConfig(name, alias, modifiable, setfn, getfn, rewritefn, applyfn) \ - {.type = SPECIAL_CONFIG, \ - embedCommonConfig(name, alias, modifiable) embedConfigInterface(NULL, setfn, getfn, rewritefn, applyfn)} + { .type = SPECIAL_CONFIG, \ + embedCommonConfig(name, alias, modifiable) embedConfigInterface(NULL, setfn, getfn, rewritefn, applyfn) } static int isValidBgsaveDefaultMethod(int val, const char **err) { /* During startup config parsing the directives are applied one by one, so @@ -3449,6 +3453,8 @@ standardConfig static_configs[] = { createBoolConfig("activerehashing", NULL, MODIFIABLE_CONFIG, server.activerehashing, 1, NULL, NULL), createBoolConfig("stop-writes-on-bgsave-error", NULL, MODIFIABLE_CONFIG, server.stop_writes_on_bgsave_err, 1, NULL, NULL), createEnumConfig("bgsave-default-method", NULL, MODIFIABLE_CONFIG, bgsave_method_enum, server.bgsave_default_method, RDB_BGSAVE_TYPE_FORK, isValidBgsaveDefaultMethod, NULL), + createEnumConfig("snapshot-mode", NULL, MODIFIABLE_CONFIG, snapshot_mode_enum, server.snapshot_mode, SNAPSHOT_MODE_FORK, NULL, NULL), + createIntConfig("snapshot-forkless-slice-keys", NULL, MODIFIABLE_CONFIG, 1, 1000000, server.snapshot_forkless_slice_keys, 64, INTEGER_CONFIG, NULL, NULL), createBoolConfig("set-proc-title", NULL, IMMUTABLE_CONFIG, server.set_proc_title, 1, NULL, NULL), /* Should setproctitle be used? */ createBoolConfig("lazyfree-lazy-eviction", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.lazyfree_lazy_eviction, 1, NULL, NULL), createBoolConfig("lazyfree-lazy-expire", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.lazyfree_lazy_expire, 1, NULL, NULL), diff --git a/src/db.c b/src/db.c index 4551fed9a43..e403b172825 100644 --- a/src/db.c +++ b/src/db.c @@ -29,6 +29,7 @@ #include "server.h" #include "crdt/peer.h" +#include "rdb_forkless.h" #include "listpack.h" #include "hotkeys.h" #include "ordered_index.h" @@ -86,6 +87,8 @@ static robj *dbFindWithDictIndex(serverDb *db, sds key, int dict_index); * expired on replicas even if the primary is lagging expiring our key via DELs * in the replication link. */ robj *lookupKey(serverDb *db, robj *key, int flags) { + /* Capture the pre-image before expiry or the caller mutates the value. */ + if ((flags & LOOKUP_WRITE) && snapshotForklessInProgress()) snapshotForklessBeforeWrite(db, key); int dict_index = getKVStoreIndexUsingCachedSlot(objectGetVal(key)); robj *val = dbFindWithDictIndex(db, objectGetVal(key), dict_index); if (val) { @@ -546,6 +549,7 @@ int dbGenericDeleteWithDictIndex(serverDb *db, robj *key, int async, int flags, /* Helper for sync and async delete. */ int dbGenericDelete(serverDb *db, robj *key, int async, int flags) { + snapshotForklessBeforeWrite(db, key); int dict_index = getKVStoreIndexForKey(objectGetVal(key)); return dbGenericDeleteWithDictIndex(db, key, async, flags, dict_index); } @@ -858,6 +862,7 @@ void flushAllDataAndResetRDB(int flags) { server.dirty += emptyData(-1, flags, NULL); if (isForkBgsaveInProgress()) killRDBChild(); if (isForklessSaveInProgress()) forklessSaveCancel(); + if (snapshotForklessInProgress()) snapshotForklessCancel(); if (server.child_type == CHILD_TYPE_SLOT_MIGRATION) killSlotMigrationChild(); if (server.saveparamslen > 0) { rdbSaveInfo rsi, *rsiptr; diff --git a/src/rdb.c b/src/rdb.c index 2fd3cba7d7c..7e7941f3b00 100644 --- a/src/rdb.c +++ b/src/rdb.c @@ -50,6 +50,7 @@ #include "compression.h" #include "compression_stream.h" #include "forkless.h" +#include "rdb_forkless.h" #include #include @@ -1813,6 +1814,10 @@ int rdbSaveToFile(const char *filename) { /* Save the DB on disk. Return C_ERR on error, C_OK on success. */ int rdbSave(int req, char *filename, rdbSaveInfo *rsi, int rdbflags) { char tmpfile[256]; + /* A synchronous save replaces an in-progress incremental snapshot. + * The incremental file has not been renamed yet, so the previous dump + * is still the live file until this save finishes. */ + if (snapshotForklessInProgress()) snapshotForklessCancel(); char cwd[MAXPATHLEN]; /* Current working dir path for error messages. */ startSaving(rdbflags); @@ -1855,7 +1860,7 @@ int isForkBgsaveInProgress(void) { } int isSaveInProgress(void) { - return isForkBgsaveInProgress() || isForklessSaveInProgress(); + return isForkBgsaveInProgress() || isForklessSaveInProgress() || snapshotForklessInProgress(); } /* Choose the background save method based on configuration. Returns forkless @@ -1879,6 +1884,7 @@ int resolveBgsaveType(void) { /* Start a background save, choosing fork or forkless based on bgsave_type. */ int rdbStartBgsave(int bgsave_type) { + if (server.snapshot_mode == SNAPSHOT_MODE_FORKLESS) return snapshotForklessStart(server.rdb_filename); if (bgsave_type == RDB_BGSAVE_TYPE_FORKLESS) { return forklessSaveToDisk(server.rdb_filename); } else { @@ -4541,6 +4547,10 @@ void bgsaveCommand(client *c) { serverLog(LL_NOTICE, "Background saving (forkless) will be aborted due to user request"); forklessSaveCancel(); addReplyStatus(c, "Background saving cancelled"); + } else if (snapshotForklessInProgress()) { + serverLog(LL_NOTICE, "Background saving (forkless snapshot) will be aborted due to user request"); + snapshotForklessCancel(); + addReplyStatus(c, "Background saving cancelled"); } else if (server.rdb_bgsave_scheduled != RDB_BGSAVE_TYPE_NONE) { serverLog(LL_NOTICE, "Scheduled background saving will be cancelled due to user request"); server.rdb_bgsave_scheduled = RDB_BGSAVE_TYPE_NONE; @@ -4576,6 +4586,12 @@ void bgsaveCommand(client *c) { "Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenever " "possible."); } + } else if (server.snapshot_mode == SNAPSHOT_MODE_FORKLESS) { + if (snapshotForklessStart(server.rdb_filename) == C_OK) { + addReplyStatus(c, "Background saving started"); + } else { + addReplyErrorObject(c, shared.err); + } } else if (chosen_save_type == RDB_BGSAVE_TYPE_FORKLESS) { if (forklessSaveToDisk(server.rdb_filename) == C_OK) { addReplyStatus(c, "Background saving started"); diff --git a/src/rdb_forkless.c b/src/rdb_forkless.c new file mode 100644 index 00000000000..cd8a63668bf --- /dev/null +++ b/src/rdb_forkless.c @@ -0,0 +1,334 @@ +/* Incremental RDB snapshot on the main thread. + * + * The snapshot is consistent with the moment it starts. Keys are copied into + * the temp file a slice at a time. The first time a not-yet-saved key is + * written or deleted, its pre-image is appended immediately (key-level + * copy-on-write) and the live mutation proceeds. The previous dump.rdb is + * replaced only by rename(2) after the footer and checksum are durable, so a + * crash mid-snapshot leaves the last good file in place. + * + * This is not the background-thread forkless saver and does not use fork(). + * + * Copyright (c) 2026, Cache Me Outside contributors. + * Licensed under the BSD 3-Clause License, the same terms as the server. + */ +#include "server.h" +#include "rdb.h" +#include "rdb_forkless.h" +#include "forkless.h" + +#include +#include +#include +#include +#include + +extern int clusterRDBSaveSlotImports(rio *rdb, int rdbver); + +typedef struct snapKey { + int dbid; + sds key; +} snapKey; + +static int active = 0; +static int in_save = 0; +static FILE *fp = NULL; +static rio rdb; +static char temp_path[256]; +static char finalfile[256]; +static int last_dbid = -1; +static hashtable *pending = NULL; +static long long cow_keys = 0; +static long long slice_us_max = 0; +static long long dirty_before = 0; + +static const void *snapEntryKey(const void *entry) { + return entry; +} + +static uint64_t snapHash(const void *key) { + const snapKey *s = key; + return dictSdsHash(s->key) ^ ((uint64_t)s->dbid * 0x9E3779B97F4A7C15ULL); +} + +/* Hashtable compare returns non-zero when the keys are equal. */ +static int snapCompare(const void *a, const void *b) { + const snapKey *sa = a; + const snapKey *sb = b; + if (sa->dbid != sb->dbid) return 0; + if (sdslen(sa->key) != sdslen(sb->key)) return 0; + return memcmp(sa->key, sb->key, sdslen(sa->key)) == 0; +} + +static void snapDestructor(void *entry) { + snapKey *s = entry; + sdsfree(s->key); + zfree(s); +} + +static hashtableType snapType = { + .entryGetKey = snapEntryKey, + .hashFunction = snapHash, + .keyCompare = snapCompare, + .entryDestructor = snapDestructor, +}; + +static void snapshotReset(void) { + active = 0; + in_save = 0; + if (fp) { + fclose(fp); + fp = NULL; + } + if (temp_path[0]) { + unlink(temp_path); + temp_path[0] = '\0'; + } + if (pending) { + hashtableRelease(pending); + pending = NULL; + } + last_dbid = -1; + memset(&rdb, 0, sizeof(rdb)); +} + +static int snapshotWriteKey(int dbid, sds keystr) { + serverDb *db; + robj *val; + robj keyobj; + long long expire; + + if (dbid < 0 || dbid >= server.dbnum) return C_OK; + db = server.db[dbid]; + val = db ? dbFind(db, keystr) : NULL; + /* Deleted without a pre-image. The delete hook writes the pre-image + * first, so this is only a missing key that was never present. */ + if (!val) return C_OK; + + if (last_dbid != dbid) { + if (rdbSaveType(&rdb, RDB_OPCODE_SELECTDB) == -1) return C_ERR; + if (rdbSaveLen(&rdb, dbid) == -1) return C_ERR; + last_dbid = dbid; + } + initStaticStringObject(keyobj, keystr); + expire = objectGetExpire(val); + if (rdbSaveKeyValuePair(&rdb, &keyobj, val, expire, dbid, RDB_VERSION) == -1) return C_ERR; + return C_OK; +} + +static int snapshotPopOne(void) { + void *found = NULL; + snapKey *owned; + sds keycopy; + int dbid; + int rc; + + if (!hashtableRandomEntry(pending, &found)) return C_OK; + owned = found; + keycopy = sdsdup(owned->key); + dbid = owned->dbid; + hashtableDelete(pending, owned); + in_save = 1; + rc = snapshotWriteKey(dbid, keycopy); + in_save = 0; + sdsfree(keycopy); + return rc; +} + +static int snapshotAddPending(int dbid, sds key) { + snapKey *s = zcalloc(sizeof(*s)); + s->dbid = dbid; + s->key = sdsdup(key); + if (!hashtableAdd(pending, s)) { + snapDestructor(s); + return C_ERR; + } + return C_OK; +} + +static int snapshotBuildPending(void) { + int dbid; + pending = hashtableCreate(&snapType); + if (!pending) return C_ERR; + for (dbid = 0; dbid < server.dbnum; dbid++) { + serverDb *db = server.db[dbid]; + kvstoreIterator *it; + void *next; + if (!db || kvstoreSize(db->keys) == 0) continue; + it = kvstoreIteratorInit(db->keys, HASHTABLE_ITER_SAFE); + while (kvstoreIteratorNext(it, &next)) { + robj *o = next; + if (snapshotAddPending(dbid, objectGetKey(o)) != C_OK) { + kvstoreIteratorRelease(it); + return C_ERR; + } + } + kvstoreIteratorRelease(it); + } + return C_OK; +} + +int snapshotForklessInProgress(void) { + return active; +} + +long long snapshotForklessKeysRemaining(void) { + if (!active || !pending) return 0; + return (long long)hashtableSize(pending); +} + +long long snapshotForklessCowKeys(void) { + return cow_keys; +} + +long long snapshotForklessSliceUsMax(void) { + return slice_us_max; +} + +void snapshotForklessBeforeWrite(serverDb *db, robj *key) { + snapKey probe; + void *found = NULL; + snapKey *owned; + sds keycopy; + int dbid; + int rc; + + if (!active || in_save || !db || !key || !pending) return; + probe.dbid = db->id; + probe.key = objectGetVal(key); + if (!probe.key) return; + if (!hashtableFind(pending, &probe, &found)) return; + + owned = found; + keycopy = sdsdup(owned->key); + dbid = owned->dbid; + hashtableDelete(pending, owned); + cow_keys++; + in_save = 1; + rc = snapshotWriteKey(dbid, keycopy); + in_save = 0; + sdsfree(keycopy); + if (rc != C_OK) { + serverLog(LL_WARNING, "Forkless snapshot failed while copying a key pre-image"); + snapshotForklessCancel(); + } +} + +int snapshotForklessStart(const char *filename) { + rdbSaveInfo rsi, *rsiptr; + + if (active || isForkBgsaveInProgress() || isForklessSaveInProgress()) return C_ERR; + if (!filename) return C_ERR; + + snprintf(temp_path, sizeof(temp_path), "temp-forkless-%d.rdb", (int)getpid()); + snprintf(finalfile, sizeof(finalfile), "%s", filename); + fp = fopen(temp_path, "w"); + if (!fp) { + serverLog(LL_WARNING, "Forkless snapshot failed to open %s: %s", temp_path, strerror(errno)); + temp_path[0] = '\0'; + return C_ERR; + } + rioInitWithFile(&rdb, fp); + if (server.rdb_save_incremental_fsync) rioSetAutoSync(&rdb, REDIS_AUTOSYNC_BYTES); + + rsiptr = rdbPopulateSaveInfo(&rsi); + if (rdbWriteHeader(&rdb, REPLICA_REQ_NONE, RDB_VERSION, RDBFLAGS_NONE, rsiptr) == C_ERR) goto err; + if (server.cluster_enabled && clusterRDBSaveSlotImports(&rdb, RDB_VERSION) == C_ERR) goto err; + /* Pending is built before `active` so the scan cannot race a mutation. + * The server is single-threaded: no command runs until we return. */ + if (snapshotBuildPending() != C_OK) goto err; + + dirty_before = server.dirty; + cow_keys = 0; + slice_us_max = 0; + last_dbid = -1; + server.dirty_before_bgsave = server.dirty; + server.lastbgsave_try = time(NULL); + server.rdb_save_time_start = time(NULL); + server.stat_rdb_saves++; + active = 1; + serverLog(LL_NOTICE, "Forkless snapshot started, %lld keys queued into %s", + pending ? (long long)hashtableSize(pending) : 0, temp_path); + return C_OK; + +err: + serverLog(LL_WARNING, "Forkless snapshot failed to start"); + snapshotReset(); + server.lastbgsave_status = C_ERR; + return C_ERR; +} + +static int snapshotFinish(void) { + if (rdbWriteFooter(&rdb, REPLICA_REQ_NONE) == C_ERR) return C_ERR; + if (fflush(fp) != 0) return C_ERR; + if (fsync(fileno(fp)) != 0) return C_ERR; + if (fclose(fp) != 0) { + fp = NULL; + return C_ERR; + } + fp = NULL; + /* Readers keep seeing the previous dump until this rename. */ + if (rename(temp_path, finalfile) == -1) { + serverLog(LL_WARNING, "Forkless snapshot rename %s -> %s failed: %s", temp_path, finalfile, strerror(errno)); + unlink(temp_path); + temp_path[0] = '\0'; + return C_ERR; + } + temp_path[0] = '\0'; + if (fsyncFileDir(finalfile) != 0) { + serverLog(LL_WARNING, "Forkless snapshot fsync dir failed: %s", strerror(errno)); + } + if (pending) { + hashtableRelease(pending); + pending = NULL; + } + active = 0; + server.dirty -= dirty_before; + if (server.dirty < 0) server.dirty = 0; + server.lastsave = time(NULL); + server.lastbgsave_status = C_OK; + server.lastbgsave_type = RDB_BGSAVE_TYPE_FORKLESS; + server.rdb_save_time_last = time(NULL) - server.rdb_save_time_start; + server.rdb_save_time_start = -1; + serverLog(LL_NOTICE, "Forkless snapshot saved on disk (%lld keys copied on write)", cow_keys); + return C_OK; +} + +void snapshotForklessCron(void) { + int budget; + int wrote = 0; + long long start; + long long took; + + if (!active) return; + budget = server.snapshot_forkless_slice_keys; + if (budget < 1) budget = 1; + start = ustime(); + while (pending && hashtableSize(pending) > 0 && wrote < budget) { + if (wrote > 0 && ustime() - start > 1000) break; + if (snapshotPopOne() != C_OK) { + serverLog(LL_WARNING, "Forkless snapshot write failed"); + snapshotForklessCancel(); + server.lastbgsave_status = C_ERR; + return; + } + wrote++; + } + took = ustime() - start; + if (took > slice_us_max) slice_us_max = took; + if (active && (!pending || hashtableSize(pending) == 0)) { + if (snapshotFinish() != C_OK) { + snapshotReset(); + server.lastbgsave_status = C_ERR; + server.rdb_save_time_start = -1; + } + } +} + +void snapshotForklessCancel(void) { + if (!active && !fp && !pending) return; + serverLog(LL_NOTICE, "Forkless snapshot cancelled; previous dump left in place"); + snapshotReset(); + server.rdb_save_time_start = -1; + server.lastbgsave_status = C_ERR; +} diff --git a/src/rdb_forkless.h b/src/rdb_forkless.h new file mode 100644 index 00000000000..763bc9b4fe8 --- /dev/null +++ b/src/rdb_forkless.h @@ -0,0 +1,18 @@ +#ifndef __RDB_FORKLESS_H +#define __RDB_FORKLESS_H + +#include "server.h" + +/* Main-thread incremental snapshot. Distinct from the background-thread + * forkless saver (forkless.c): this one never forks and never samples the + * live keyspace after a key has been mutated. */ +int snapshotForklessStart(const char *filename); +void snapshotForklessCancel(void); +int snapshotForklessInProgress(void); +void snapshotForklessCron(void); +void snapshotForklessBeforeWrite(serverDb *db, robj *key); +long long snapshotForklessKeysRemaining(void); +long long snapshotForklessCowKeys(void); +long long snapshotForklessSliceUsMax(void); + +#endif diff --git a/src/server.c b/src/server.c index bec1545bfca..2583ddc3a17 100644 --- a/src/server.c +++ b/src/server.c @@ -60,6 +60,7 @@ #include "util.h" #include "forkless.h" #include "crdt/peer.h" +#include "rdb_forkless.h" #include "eval.h" #include "bgiteration.h" @@ -1694,6 +1695,7 @@ long long serverCron(struct aeEventLoop *eventLoop, long long id, void *clientDa /* Handle background operations on databases. */ databasesCron(); crdtCron(); + snapshotForklessCron(); /* Start a scheduled AOF rewrite if this was requested by the user while * a BGSAVE was in progress. We don't start the rewrite if there is an @@ -5249,6 +5251,10 @@ int finishShutdown(void) { serverLog(LL_WARNING, "There is a thread saving an .rdb. Cancelling it!"); forklessSaveCancel(); } + if (snapshotForklessInProgress()) { + serverLog(LL_WARNING, "There is a forkless snapshot in progress. Cancelling it!"); + snapshotForklessCancel(); + } /* Kill module child if there is one. */ if (server.child_type == CHILD_TYPE_MODULE) { @@ -6786,6 +6792,10 @@ sds genValkeyInfoString(dict *section_dict, int all_sections, int everything) { "rdb_last_bgsave_time_sec:%jd\r\n", (intmax_t)server.rdb_save_time_last, "rdb_current_bgsave_time_sec:%jd\r\n", (intmax_t)(isSaveInProgress() ? time(NULL) - server.rdb_save_time_start : -1), "rdb_saves:%lld\r\n", server.stat_rdb_saves, + "forkless_snapshot_in_progress:%d\r\n", snapshotForklessInProgress(), + "forkless_snapshot_keys_remaining:%lld\r\n", snapshotForklessKeysRemaining(), + "forkless_snapshot_cow_keys:%lld\r\n", snapshotForklessCowKeys(), + "forkless_snapshot_slice_us_max:%lld\r\n", snapshotForklessSliceUsMax(), "rdb_last_cow_size:%zu\r\n", server.stat_rdb_cow_bytes, "rdb_last_load_keys_expired:%lld\r\n", server.rdb_last_load_keys_expired, "rdb_last_load_keys_loaded:%lld\r\n", server.rdb_last_load_keys_loaded, diff --git a/src/server.h b/src/server.h index 15fe783a628..84501eedcdb 100644 --- a/src/server.h +++ b/src/server.h @@ -170,7 +170,7 @@ struct ValkeyModule; #define CONFIG_DEFAULT_PID_FILE "/var/run/valkey.pid" #define CONFIG_DEFAULT_BINDADDR_COUNT 2 #define CONFIG_DEFAULT_BINDADDR \ - {"*", "-::*"} + { "*", "-::*" } #define CONFIG_BINDADDR_MAX 16 #define CONFIG_MIN_RESERVED_FDS 32 #define CONFIG_DEFAULT_PROC_TITLE_TEMPLATE "{title} {listen-addr} {server-mode}" @@ -696,6 +696,10 @@ typedef enum { RDB_BGSAVE_TYPE_FORKLESS = 2 /* Forkless bgsave. */ } rdbBgsaveType; +/* Main-thread incremental snapshot. Independent of bgsave-default-method. */ +#define SNAPSHOT_MODE_FORK 0 +#define SNAPSHOT_MODE_FORKLESS 1 + /* Replica failover policy for server.cluster_replica_no_failover. */ typedef enum { CLUSTER_REPLICA_NO_FAILOVER_NO = 0, /* Allow automatic failover (default). */ @@ -1733,7 +1737,8 @@ typedef struct rdbSaveInfo { bool loaded_compressed; /* True if rdbLoad() read a streaming-compressed file. */ } rdbSaveInfo; -#define RDB_SAVE_INFO_INIT {-1, 0, "0000000000000000000000000000000000000000", -1, false} +#define RDB_SAVE_INFO_INIT \ + { -1, 0, "0000000000000000000000000000000000000000", -1, false } struct malloc_stats { size_t zmalloc_used; @@ -2216,6 +2221,8 @@ struct valkeyServer { int crdt_region_id; char *crdt_peers; long long crdt_tombstone_gc_ms; + int snapshot_mode; /* SNAPSHOT_MODE_FORK or SNAPSHOT_MODE_FORKLESS */ + int snapshot_forkless_slice_keys; /* Max keys written per cron tick in forkless mode */ int key_load_delay; /* Delay in microseconds between keys while * loading aof or rdb. (for testings). negative * value means fractions of microseconds (on average). */ diff --git a/tests/unit/forkless-snapshot.tcl b/tests/unit/forkless-snapshot.tcl new file mode 100644 index 00000000000..563ea18287d --- /dev/null +++ b/tests/unit/forkless-snapshot.tcl @@ -0,0 +1,129 @@ +# Main-thread forkless snapshots. The snapshot must match the dataset at +# BGSAVE start, stay under 1.2x memory, and leave the previous dump in place +# if the process dies before the atomic rename. + +start_server [list tags {external:skip} overrides [list save "" snapshot-mode forkless snapshot-forkless-slice-keys 8]] { + test {forkless snapshot matches the dataset at start under concurrent writes} { + for {set i 0} {$i < 200} {incr i} { + r set k$i v0 + } + set before [r debug digest] + r bgsave + for {set i 0} {$i < 200} {incr i} { + r set k$i v1 + } + wait_for_condition 100 50 { + [string match {*forkless_snapshot_in_progress:0*} [r info persistence]] + } else { + fail "forkless snapshot did not finish" + } + assert {[r debug reload nosave] eq {OK}} + assert_equal [r debug digest] $before + } {} {needs:debug} + + test {forkless snapshot memory stays under 1.2x the dataset} { + r flushall + set payload [string repeat x 100000] + for {set i 0} {$i < 20} {incr i} { + r set big$i $payload + } + set base [status r used_memory] + r config set snapshot-forkless-slice-keys 1 + r bgsave + set peak $base + for {set i 0} {$i < 30} {incr i} { + set m [status r used_memory] + if {$m > $peak} {set peak $m} + after 10 + } + wait_for_condition 100 50 { + [string match {*forkless_snapshot_in_progress:0*} [r info persistence]] + } else { + fail "forkless snapshot did not finish" + } + set m [status r used_memory] + if {$m > $peak} {set peak $m} + assert {$peak < $base * 1.2} + } + + test {forkless snapshot keeps command latency bounded} { + r config set snapshot-forkless-slice-keys 1 + for {set i 0} {$i < 400} {incr i} { + r set lat$i [string repeat y 64] + } + r bgsave + set lats {} + for {set i 0} {$i < 100} {incr i} { + set t [clock microseconds] + r ping + lappend lats [expr {[clock microseconds] - $t}] + } + set ordered [lsort -integer $lats] + set p99 [lindex $ordered [expr {int(([llength $ordered] - 1) * 0.99)}]] + # Drain the rest of the snapshot in one cron tick. + r config set snapshot-forkless-slice-keys 100000 + wait_for_condition 100 50 { + [string match {*forkless_snapshot_in_progress:0*} [r info persistence]] + } else { + fail "forkless snapshot did not finish" + } + # The cron slice is 1ms plus one key. A 50ms client p99 leaves room + # for a loaded CI host without hiding a stalled event loop. + assert {$p99 < 50000} + } + + test {crash mid-snapshot leaves the previous dump valid} { + r flushall + r config set snapshot-forkless-slice-keys 64 + for {set i 0} {$i < 50} {incr i} { + r set old$i before + } + r bgsave + wait_for_condition 100 50 { + [string match {*forkless_snapshot_in_progress:0*} [r info persistence]] + } else { + fail "initial snapshot did not finish" + } + set digest [r debug digest] + for {set i 0} {$i < 50} {incr i} { + r set old$i after + } + for {set i 0} {$i < 2000} {incr i} { + r set extra$i extra + } + r config set snapshot-forkless-slice-keys 1 + r bgsave + assert {[string match {*forkless_snapshot_in_progress:1*} [r info persistence]]} + exec kill -9 [srv 0 pid] + set pid [srv 0 pid] + set spins 0 + while {[is_alive $pid]} { + after 20 + incr spins + if {$spins > 250} {fail "server did not die"} + } + restart_server 0 true false + assert_equal [r debug digest] $digest + assert_equal [r get old0] before + } {} {needs:debug} +} + +start_cluster 1 0 {tags {external:skip cluster}} { + test {forkless snapshot runs on a cluster primary} { + R 0 config set save {} + R 0 config set snapshot-mode forkless + R 0 config set snapshot-forkless-slice-keys 4 + R 0 set clusterkey v0 + set before [R 0 debug digest] + R 0 bgsave + R 0 set clusterkey v1 + wait_for_condition 100 50 { + [string match {*forkless_snapshot_in_progress:0*} [R 0 info persistence]] + } else { + fail "cluster forkless snapshot did not finish" + } + assert {[R 0 debug reload nosave] eq {OK}} + assert_equal [R 0 debug digest] $before + assert_equal [R 0 get clusterkey] v0 + } {} {needs:debug} +} diff --git a/valkey.conf b/valkey.conf index 2454856e5f3..e3b129812d7 100644 --- a/valkey.conf +++ b/valkey.conf @@ -623,6 +623,15 @@ stop-writes-on-bgsave-error yes # # bgsave-default-method fork +# Main-thread incremental snapshot. `forkless` writes the RDB a slice per +# event-loop tick and, before a not-yet-saved key is mutated, copies that +# key's pre-image into the snapshot. The previous dump is replaced only by +# rename, so a crash leaves the last complete file. Default is fork, which +# is the historical BGSAVE child. This is separate from bgsave-default-method. +# +# snapshot-mode fork +# snapshot-forkless-slice-keys 64 + # Control compression when dumping .rdb databases. # Supported values: # yes - use the default compression algorithm (currently lzf)