iovalkey

A robust, performance-focused and full-featured Valkey client for Node.js. This is a friendly fork of ioredis after this commit.

Supports Valkey >= 2.6.12. Completely compatible with Valkey 7.x.

Features

iovalkey is a robust, full-featured Valkey client that is used in the world's biggest online commerce company Alibaba and many other awesome companies.

0. Full-featured. It supports Cluster, Sentinel, Streams, Pipelining, and of course Lua scripting, Valkey Functions, Pub/Sub (with the support of binary messages).
1. High performance 🚀.
2. Delightful API 😄. It works with Node callbacks and Native promises.
3. Transformation of command arguments and replies.
4. Transparent key prefixing.
5. Abstraction for Lua scripting, allowing you to define custom commands.
6. Supports binary data.
7. Supports TLS 🔒.
8. Supports offline queue and ready checking.
9. Supports ES6 types, such as Map and Set.
10. Supports GEO commands 📍.
11. Supports Valkey ACL.
12. Sophisticated error handling strategy.
13. Supports NAT mapping.
14. Supports autopipelining.

100% written in TypeScript and official declarations are provided:

---

Quick Start

Install

1npm install iovalkey

In a TypeScript project, you may want to add TypeScript declarations for Node.js:

1npm install --save-dev @types/node

Basic Usage

1// Import iovalkey.
2// You can also use `import { Valkey } from "iovalkey"`
3// if your project is a TypeScript project,
4// Note that `import Valkey from "iovalkey"` is still supported,
5// but will be deprecated in the next major version.
6const Valkey = require("iovalkey");
7
8// Create a Valkey instance.
9// By default, it will connect to localhost:6379.
10// We are going to cover how to specify connection options soon.
11const valkey = new Valkey();
12
13valkey.set("mykey", "value"); // Returns a promise which resolves to "OK" when the command succeeds.
14
15// iovalkey supports the node.js callback style
16valkey.get("mykey", (err, result) => {
17  if (err) {
18    console.error(err);
19  } else {
20    console.log(result); // Prints "value"
21  }
22});
23
24// Or iovalkey returns a promise if the last argument isn't a function
25valkey.get("mykey").then((result) => {
26  console.log(result); // Prints "value"
27});
28
29valkey.zadd("sortedSet", 1, "one", 2, "dos", 4, "quatro", 3, "three");
30valkey.zrange("sortedSet", 0, 2, "WITHSCORES").then((elements) => {
31  // ["one", "1", "dos", "2", "three", "3"] as if the command was `ZRANGE sortedSet 0 2 WITHSCORES`
32  console.log(elements);
33});
34
35// All arguments are passed directly to the valkey/valkey server,
36// so technically iovalkey supports all Valkey commands.
37// The format is: valkey[SOME_VALKEY_COMMAND_IN_LOWERCASE](ARGUMENTS_ARE_JOINED_INTO_COMMAND_STRING)
38// so the following statement is equivalent to the CLI: `SET mykey hello EX 10`
39valkey.set("mykey", "hello", "EX", 10);

See the examples/ folder for more examples. For example:

• TTL
• Strings
• Hashes
• Lists
• Sets
• Sorted Sets
• Streams
• Redis Modules e.g. RedisJSON

All Valkey commands are supported. See the documentation for details.

Connect to Valkey

When a new Valkey instance is created, a connection to Valkey will be created at the same time. You can specify which Valkey to connect to by:

1new Valkey(); // Connect to 127.0.0.1:6379
2new Valkey(6380); // 127.0.0.1:6380
3new Valkey(6379, "192.168.1.1"); // 192.168.1.1:6379
4new Valkey("/tmp/valkey.sock");
5new Valkey({
6  port: 6379, // Valkey port
7  host: "127.0.0.1", // Valkey host
8  username: "default", // needs Valkey >= 6
9  password: "my-top-secret",
10  db: 0, // Defaults to 0
11});

You can also specify connection options as a redis:// URL or rediss:// URL when using TLS encryption:

1// Connect to 127.0.0.1:6380, db 4, using password "authpassword":
2new Valkey("redis://:authpassword@127.0.0.1:6380/4");
3
4// Username can also be passed via URI.
5new Valkey("redis://username:authpassword@127.0.0.1:6380/4");

See API Documentation for all available options.

Pub/Sub

Valkey provides several commands for developers to implement the Publish–subscribe pattern. There are two roles in this pattern: publisher and subscriber. Publishers are not programmed to send their messages to specific subscribers. Rather, published messages are characterized into channels, without knowledge of what (if any) subscribers there may be.

By leveraging Node.js's built-in events module, iovalkey makes pub/sub very straightforward to use. Below is a simple example that consists of two files, one is publisher.js that publishes messages to a channel, the other is subscriber.js that listens for messages on specific channels.

1// publisher.js
2
3const Valkey = require("iovalkey");
4const valkey = new Valkey();
5
6setInterval(() => {
7  const message = { foo: Math.random() };
8  // Publish to my-channel-1 or my-channel-2 randomly.
9  const channel = `my-channel-${1 + Math.round(Math.random())}`;
10
11  // Message can be either a string or a buffer
12  valkey.publish(channel, JSON.stringify(message));
13  console.log("Published %s to %s", message, channel);
14}, 1000);

1// subscriber.js
2
3const Valkey = require("iovalkey");
4const valkey = new Valkey();
5
6valkey.subscribe("my-channel-1", "my-channel-2", (err, count) => {
7  if (err) {
8    // Just like other commands, subscribe() can fail for some reasons,
9    // ex network issues.
10    console.error("Failed to subscribe: %s", err.message);
11  } else {
12    // `count` represents the number of channels this client are currently subscribed to.
13    console.log(
14      `Subscribed successfully! This client is currently subscribed to ${count} channels.`
15    );
16  }
17});
18
19valkey.on("message", (channel, message) => {
20  console.log(`Received ${message} from ${channel}`);
21});
22
23// There's also an event called 'messageBuffer', which is the same as 'message' except
24// it returns buffers instead of strings.
25// It's useful when the messages are binary data.
26valkey.on("messageBuffer", (channel, message) => {
27  // Both `channel` and `message` are buffers.
28  console.log(channel, message);
29});

It's worth noticing that a connection (aka a Valkey instance) can't play both roles at the same time. More specifically, when a client issues subscribe() or psubscribe(), it enters the "subscriber" mode. From that point, only commands that modify the subscription set are valid. Namely, they are: subscribe, psubscribe, unsubscribe, punsubscribe, ping, and quit. When the subscription set is empty (via unsubscribe/punsubscribe), the connection is put back into the regular mode.

If you want to do pub/sub in the same file/process, you should create a separate connection:

1const Valkey = require("iovalkey");
2const sub = new Valkey();
3const pub = new Valkey();
4
5sub.subscribe(/* ... */); // From now, `sub` enters the subscriber mode.
6sub.on("message" /* ... */);
7
8setInterval(() => {
9  // `pub` can be used to publish messages, or send other regular commands (e.g. `hgetall`)
10  // because it's not in the subscriber mode.
11  pub.publish(/* ... */);
12}, 1000);

PSUBSCRIBE is also supported in a similar way when you want to subscribe all channels whose name matches a pattern:

1valkey.psubscribe("pat?ern", (err, count) => {});
2
3// Event names are "pmessage"/"pmessageBuffer" instead of "message/messageBuffer".
4valkey.on("pmessage", (pattern, channel, message) => {});
5valkey.on("pmessageBuffer", (pattern, channel, message) => {});

Streams

Valkey v5 introduces a new data type called streams. It doubles as a communication channel for building streaming architectures and as a log-like data structure for persisting data. With iovalkey, the usage can be pretty straightforward. Say we have a producer publishes messages to a stream with valkey.xadd("mystream", "*", "randomValue", Math.random()) (You may find the official documentation of Streams as a starter to understand the parameters used), to consume the messages, we'll have a consumer with the following code:

1const Valkey = require("iovalkey");
2const valkey = new Valkey();
3
4const processMessage = (message) => {
5  console.log("Id: %s. Data: %O", message[0], message[1]);
6};
7
8async function listenForMessage(lastId = "$") {
9  // `results` is an array, each element of which corresponds to a key.
10  // Because we only listen to one key (mystream) here, `results` only contains
11  // a single element. See more: https://valkey.io/commands/xread/
12  const results = await valkey.xread("block", 0, "STREAMS", "mystream", lastId);
13  const [key, messages] = results[0]; // `key` equals to "mystream"
14
15  messages.forEach(processMessage);
16
17  // Pass the last id of the results to the next round.
18  await listenForMessage(messages[messages.length - 1][0]);
19}
20
21listenForMessage();

Expiration

Valkey can set a timeout to expire your key, after the timeout has expired the key will be automatically deleted. (You can find the official Expire documentation to understand better the different parameters you can use), to set your key to expire in 60 seconds, we will have the following code:

1valkey.set("key", "data", "EX", 60);
2// Equivalent to valkey command "SET key data EX 60", because on iovalkey set method,
3// all arguments are passed directly to the valkey server.

Handle Binary Data

Binary data support is out of the box. Pass buffers to send binary data:

1valkey.set("foo", Buffer.from([0x62, 0x75, 0x66]));

Every command that returns a bulk string has a variant command with a Buffer suffix. The variant command returns a buffer instead of a UTF-8 string:

1const result = await valkey.getBuffer("foo");
2// result is `<Buffer 62 75 66>`

It's worth noticing that you don't need the Buffer suffix variant in order to send binary data. That means in most case you should just use valkey.set() instead of valkey.setBuffer() unless you want to get the old value with the GET parameter:

1const result = await valkey.setBuffer("foo", "new value", "GET");
2// result is `<Buffer 62 75 66>` as `GET` indicates returning the old value.

Pipelining

If you want to send a batch of commands (e.g. > 5), you can use pipelining to queue the commands in memory and then send them to Valkey all at once. This way the performance improves by 50%~300% (See benchmark section).

valkey.pipeline() creates a Pipeline instance. You can call any Valkey commands on it just like the Valkey instance. The commands are queued in memory and flushed to Valkey by calling the exec method:

1const pipeline = valkey.pipeline();
2pipeline.set("foo", "bar");
3pipeline.del("cc");
4pipeline.exec((err, results) => {
5  // `err` is always null, and `results` is an array of responses
6  // corresponding to the sequence of queued commands.
7  // Each response follows the format `[err, result]`.
8});
9
10// You can even chain the commands:
11valkey
12  .pipeline()
13  .set("foo", "bar")
14  .del("cc")
15  .exec((err, results) => {});
16
17// `exec` also returns a Promise:
18const promise = valkey.pipeline().set("foo", "bar").get("foo").exec();
19promise.then((result) => {
20  // result === [[null, 'OK'], [null, 'bar']]
21});

Each chained command can also have a callback, which will be invoked when the command gets a reply:

1valkey
2  .pipeline()
3  .set("foo", "bar")
4  .get("foo", (err, result) => {
5    // result === 'bar'
6  })
7  .exec((err, result) => {
8    // result[1][1] === 'bar'
9  });

In addition to adding commands to the pipeline queue individually, you can also pass an array of commands and arguments to the constructor:

1valkey
2  .pipeline([
3    ["set", "foo", "bar"],
4    ["get", "foo"],
5  ])
6  .exec(() => {
7    /* ... */
8  });

#length property shows how many commands in the pipeline:

1const length = valkey.pipeline().set("foo", "bar").get("foo").length;
2// length === 2

Transaction

Most of the time, the transaction commands multi & exec are used together with pipeline. Therefore, when multi is called, a Pipeline instance is created automatically by default, so you can use multi just like pipeline:

1valkey
2  .multi()
3  .set("foo", "bar")
4  .get("foo")
5  .exec((err, results) => {
6    // results === [[null, 'OK'], [null, 'bar']]
7  });

If there's a syntax error in the transaction's command chain (e.g. wrong number of arguments, wrong command name, etc), then none of the commands would be executed, and an error is returned:

1valkey
2  .multi()
3  .set("foo")
4  .set("foo", "new value")
5  .exec((err, results) => {
6    // err:
7    //  { [ReplyError: EXECABORT Transaction discarded because of previous errors.]
8    //    name: 'ReplyError',
9    //    message: 'EXECABORT Transaction discarded because of previous errors.',
10    //    command: { name: 'exec', args: [] },
11    //    previousErrors:
12    //     [ { [ReplyError: ERR wrong number of arguments for 'set' command]
13    //         name: 'ReplyError',
14    //         message: 'ERR wrong number of arguments for \'set\' command',
15    //         command: [Object] } ] }
16  });

In terms of the interface, multi differs from pipeline in that when specifying a callback to each chained command, the queueing state is passed to the callback instead of the result of the command:

1valkey
2  .multi()
3  .set("foo", "bar", (err, result) => {
4    // result === 'QUEUED'
5  })
6  .exec(/* ... */);

If you want to use transaction without pipeline, pass { pipeline: false } to multi, and every command will be sent to Valkey immediately without waiting for an exec invocation:

1valkey.multi({ pipeline: false });
2valkey.set("foo", "bar");
3valkey.get("foo");
4valkey.exec((err, result) => {
5  // result === [[null, 'OK'], [null, 'bar']]
6});

The constructor of multi also accepts a batch of commands:

1valkey
2  .multi([
3    ["set", "foo", "bar"],
4    ["get", "foo"],
5  ])
6  .exec(() => {
7    /* ... */
8  });

Inline transactions are supported by pipeline, which means you can group a subset of commands in the pipeline into a transaction:

1valkey
2  .pipeline()
3  .get("foo")
4  .multi()
5  .set("foo", "bar")
6  .get("foo")
7  .exec()
8  .get("foo")
9  .exec();

Lua Scripting

iovalkey supports all of the scripting commands such as EVAL, EVALSHA and SCRIPT. However, it's tedious to use in real world scenarios since developers have to take care of script caching and to detect when to use EVAL and when to use EVALSHA. iovalkey exposes a defineCommand method to make scripting much easier to use:

1const valkey = new Valkey();
2
3// This will define a command myecho:
4valkey.defineCommand("myecho", {
5  numberOfKeys: 2,
6  lua: "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}",
7});
8
9// Now `myecho` can be used just like any other ordinary command,
10// and iovalkey will try to use `EVALSHA` internally when possible for better performance.
11valkey.myecho("k1", "k2", "a1", "a2", (err, result) => {
12  // result === ['k1', 'k2', 'a1', 'a2']
13});
14
15// `myechoBuffer` is also defined automatically to return buffers instead of strings:
16valkey.myechoBuffer("k1", "k2", "a1", "a2", (err, result) => {
17  // result[0] equals to Buffer.from('k1');
18});
19
20// And of course it works with pipeline:
21valkey.pipeline().set("foo", "bar").myecho("k1", "k2", "a1", "a2").exec();

Dynamic Keys

If the number of keys can't be determined when defining a command, you can omit the numberOfKeys property and pass the number of keys as the first argument when you call the command:

1valkey.defineCommand("echoDynamicKeyNumber", {
2  lua: "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}",
3});
4
5// Now you have to pass the number of keys as the first argument every time
6// you invoke the `echoDynamicKeyNumber` command:
7valkey.echoDynamicKeyNumber(2, "k1", "k2", "a1", "a2", (err, result) => {
8  // result === ['k1', 'k2', 'a1', 'a2']
9});

As Constructor Options

Besides defineCommand(), you can also define custom commands with the scripts constructor option:

1const valkey = new Valkey({
2  scripts: {
3    myecho: {
4      numberOfKeys: 2,
5      lua: "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}",
6    },
7  },
8});

TypeScript Usages

You can refer to the example for how to declare your custom commands.

Transparent Key Prefixing

This feature allows you to specify a string that will automatically be prepended to all the keys in a command, which makes it easier to manage your key namespaces.

Warning This feature won't apply to commands like KEYS and SCAN that take patterns rather than actual keys(#239), and this feature also won't apply to the replies of commands even if they are key names (#325).

1const fooValkey = new Valkey({ keyPrefix: "foo:" });
2fooValkey.set("bar", "baz"); // Actually sends SET foo:bar baz
3
4fooValkey.defineCommand("myecho", {
5  numberOfKeys: 2,
6  lua: "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}",
7});
8
9// Works well with pipelining/transaction
10fooValkey
11  .pipeline()
12  // Sends SORT foo:list BY foo:weight_*->fieldname
13  .sort("list", "BY", "weight_*->fieldname")
14  // Supports custom commands
15  // Sends EVALSHA xxx foo:k1 foo:k2 a1 a2
16  .myecho("k1", "k2", "a1", "a2")
17  .exec();

Transforming Arguments & Replies

Most Valkey commands take one or more Strings as arguments, and replies are sent back as a single String or an Array of Strings. However, sometimes you may want something different. For instance, it would be more convenient if the HGETALL command returns a hash (e.g. { key: val1, key2: v2 }) rather than an array of key values (e.g. [key1, val1, key2, val2]).

iovalkey has a flexible system for transforming arguments and replies. There are two types of transformers, argument transformer and reply transformer:

1const Valkey = require("iovalkey");
2
3// Here's the built-in argument transformer converting
4// hmset('key', { k1: 'v1', k2: 'v2' })
5// or
6// hmset('key', new Map([['k1', 'v1'], ['k2', 'v2']]))
7// into
8// hmset('key', 'k1', 'v1', 'k2', 'v2')
9Valkey.Command.setArgumentTransformer("hmset", (args) => {
10  if (args.length === 2) {
11    if (args[1] instanceof Map) {
12      // utils is a internal module of iovalkey
13      return [args[0], ...utils.convertMapToArray(args[1])];
14    }
15    if (typeof args[1] === "object" && args[1] !== null) {
16      return [args[0], ...utils.convertObjectToArray(args[1])];
17    }
18  }
19  return args;
20});
21
22// Here's the built-in reply transformer converting the HGETALL reply
23// ['k1', 'v1', 'k2', 'v2']
24// into
25// { k1: 'v1', 'k2': 'v2' }
26Valkey.Command.setReplyTransformer("hgetall", (result) => {
27  if (Array.isArray(result)) {
28    const obj = {};
29    for (let i = 0; i < result.length; i += 2) {
30      obj[result[i]] = result[i + 1];
31    }
32    return obj;
33  }
34  return result;
35});

There are three built-in transformers, two argument transformers for hmset & mset and a reply transformer for hgetall. Transformers for hmset and hgetall were mentioned above, and the transformer for mset is similar to the one for hmset:

1valkey.mset({ k1: "v1", k2: "v2" });
2valkey.get("k1", (err, result) => {
3  // result === 'v1';
4});
5
6valkey.mset(
7  new Map([
8    ["k3", "v3"],
9    ["k4", "v4"],
10  ])
11);
12valkey.get("k3", (err, result) => {
13  // result === 'v3';
14});

Another useful example of a reply transformer is one that changes hgetall to return array of arrays instead of objects which avoids an unwanted conversation of hash keys to strings when dealing with binary hash keys:

1Valkey.Command.setReplyTransformer("hgetall", (result) => {
2  const arr = [];
3  for (let i = 0; i < result.length; i += 2) {
4    arr.push([result[i], result[i + 1]]);
5  }
6  return arr;
7});
8valkey.hset("h1", Buffer.from([0x01]), Buffer.from([0x02]));
9valkey.hset("h1", Buffer.from([0x03]), Buffer.from([0x04]));
10valkey.hgetallBuffer("h1", (err, result) => {
11  // result === [ [ <Buffer 01>, <Buffer 02> ], [ <Buffer 03>, <Buffer 04> ] ];
12});

Monitor

Valkey supports the MONITOR command, which lets you see all commands received by the Valkey server across all client connections, including from other client libraries and other computers.

The monitor method returns a monitor instance. After you send the MONITOR command, no other commands are valid on that connection. iovalkey will emit a monitor event for every new monitor message that comes across. The callback for the monitor event takes a timestamp from the Valkey server and an array of command arguments.

Here is a simple example:

1valkey.monitor((err, monitor) => {
2  monitor.on("monitor", (time, args, source, database) => {});
3});

Here is another example illustrating an async function and monitor.disconnect():

1async () => {
2  const monitor = await valkey.monitor();
3  monitor.on("monitor", console.log);
4  // Any other tasks
5  monitor.disconnect();
6};

Streamify Scanning

Valkey 2.8 added the SCAN command to incrementally iterate through the keys in the database. It's different from KEYS in that SCAN only returns a small number of elements each call, so it can be used in production without the downside of blocking the server for a long time. However, it requires recording the cursor on the client side each time the SCAN command is called in order to iterate through all the keys correctly. Since it's a relatively common use case, iovalkey provides a streaming interface for the SCAN command to make things much easier. A readable stream can be created by calling scanStream:

1const valkey = new Valkey();
2// Create a readable stream (object mode)
3const stream = valkey.scanStream();
4stream.on("data", (resultKeys) => {
5  // `resultKeys` is an array of strings representing key names.
6  // Note that resultKeys may contain 0 keys, and that it will sometimes
7  // contain duplicates due to SCAN's implementation in Valkey.
8  for (let i = 0; i < resultKeys.length; i++) {
9    console.log(resultKeys[i]);
10  }
11});
12stream.on("end", () => {
13  console.log("all keys have been visited");
14});

scanStream accepts an option, with which you can specify the MATCH pattern, the TYPE filter, and the COUNT argument:

1const stream = valkey.scanStream({
2  // only returns keys following the pattern of `user:*`
3  match: "user:*",
4  // only return objects that match a given type,
5  // (requires Valkey >= 6.0)
6  type: "zset",
7  // returns approximately 100 elements per call
8  count: 100,
9});

Just like other commands, scanStream has a binary version scanBufferStream, which returns an array of buffers. It's useful when the key names are not utf8 strings.

There are also hscanStream, zscanStream and sscanStream to iterate through elements in a hash, zset and set. The interface of each is similar to scanStream except the first argument is the key name:

1const stream = valkey.hscanStream("myhash", {
2  match: "age:??",
3});

You can learn more from the Valkey documentation.

Useful Tips It's pretty common that doing an async task in the data handler. We'd like the scanning process to be paused until the async task to be finished. Stream#pause() and Stream#resume() do the trick. For example if we want to migrate data in Valkey to MySQL:

1const stream = valkey.scanStream();
2stream.on("data", (resultKeys) => {
3  // Pause the stream from scanning more keys until we've migrated the current keys.
4  stream.pause();
5
6  Promise.all(resultKeys.map(migrateKeyToMySQL)).then(() => {
7    // Resume the stream here.
8    stream.resume();
9  });
10});
11
12stream.on("end", () => {
13  console.log("done migration");
14});

Auto-reconnect

By default, iovalkey will try to reconnect when the connection to Valkey is lost except when the connection is closed manually by valkey.disconnect() or valkey.quit().

It's very flexible to control how long to wait to reconnect after disconnection using the retryStrategy option:

1const valkey = new Valkey({
2  // This is the default value of `retryStrategy`
3  retryStrategy(times) {
4    const delay = Math.min(times * 50, 2000);
5    return delay;
6  },
7});

retryStrategy is a function that will be called when the connection is lost. The argument times means this is the nth reconnection being made and the return value represents how long (in ms) to wait to reconnect. When the return value isn't a number, iovalkey will stop trying to reconnect, and the connection will be lost forever if the user doesn't call valkey.connect() manually.

When reconnected, the client will auto subscribe to channels that the previous connection subscribed to. This behavior can be disabled by setting the autoResubscribe option to false.

And if the previous connection has some unfulfilled commands (most likely blocking commands such as brpop and blpop), the client will resend them when reconnected. This behavior can be disabled by setting the autoResendUnfulfilledCommands option to false.

By default, all pending commands will be flushed with an error every 20 retry attempts. That makes sure commands won't wait forever when the connection is down. You can change this behavior by setting maxRetriesPerRequest:

1const valkey = new Valkey({
2  maxRetriesPerRequest: 1,
3});

Set maxRetriesPerRequest to null to disable this behavior, and every command will wait forever until the connection is alive again (which is the default behavior before iovalkey v4).

Reconnect on Error

Besides auto-reconnect when the connection is closed, iovalkey supports reconnecting on certain Valkey errors using the reconnectOnError option. Here's an example that will reconnect when receiving READONLY error:

1const valkey = new Valkey({
2  reconnectOnError(err) {
3    const targetError = "READONLY";
4    if (err.message.includes(targetError)) {
5      // Only reconnect when the error contains "READONLY"
6      return true; // or `return 1;`
7    }
8  },
9});

This feature is useful when using Amazon ElastiCache instances with Auto-failover disabled. On these instances, test your reconnectOnError handler by manually promoting the replica node to the primary role using the AWS console. The following writes fail with the error READONLY. Using reconnectOnError, we can force the connection to reconnect on this error in order to connect to the new master. Furthermore, if the reconnectOnError returns 2, iovalkey will resend the failed command after reconnecting.

On ElastiCache instances with Auto-failover enabled, reconnectOnError does not execute. Instead of returning a Valkey error, AWS closes all connections to the master endpoint until the new primary node is ready. iovalkey reconnects via retryStrategy instead of reconnectOnError after about a minute. On ElastiCache instances with Auto-failover enabled, test failover events with the Failover primary option in the AWS console.

Connection Events

The Valkey instance will emit some events about the state of the connection to the Valkey server.

Event	Description
connect	emits when a connection is established to the Valkey server.
ready	If enableReadyCheck is true, client will emit ready when the server reports that it is ready to receive commands (e.g. finish loading data from disk). Otherwise, ready will be emitted immediately right after the connect event.
error	emits when an error occurs while connecting. However, iovalkey emits all error events silently (only emits when there's at least one listener) so that your application won't crash if you're not listening to the error event.
close	emits when an established Valkey server connection has closed.
reconnecting	emits after close when a reconnection will be made. The argument of the event is the time (in ms) before reconnecting.
end	emits after close when no more reconnections will be made, or the connection is failed to establish.
wait	emits when lazyConnect is set and will wait for the first command to be called before connecting.

You can also check out the Valkey#status property to get the current connection status.

Besides the above connection events, there are several other custom events:

Event	Description
select	emits when the database changed. The argument is the new db number.

Offline Queue

When a command can't be processed by Valkey (being sent before the ready event), by default, it's added to the offline queue and will be executed when it can be processed. You can disable this feature by setting the enableOfflineQueue option to false:

1const valkey = new Valkey({ enableOfflineQueue: false });

TLS Options

Valkey doesn't support TLS natively, however if the valkey server you want to connect to is hosted behind a TLS proxy (e.g. stunnel) or is offered by a PaaS service that supports TLS connection, you can set the tls option:

1const valkey = new Valkey({
2  host: "localhost",
3  tls: {
4    // Refer to `tls.connect()` section in
5    // https://nodejs.org/api/tls.html
6    // for all supported options
7    ca: fs.readFileSync("cert.pem"),
8  },
9});

Alternatively, specify the connection through a rediss:// URL.

1const valkey = new Valkey("rediss://valkey.my-service.com");

If you do not want to use a connection string, you can also specify an empty tls: {} object:

1const valkey = new Valkey({
2  host: "valkey.my-service.com",
3  tls: {},
4});

---

Sentinel

iovalkey supports Sentinel out of the box. It works transparently as all features that work when you connect to a single node also work when you connect to a sentinel group. Make sure to run Valkey >= 2.8.12 if you want to use this feature. Sentinels have a default port of 26379.

To connect using Sentinel, use:

1const valkey = new Valkey({
2  sentinels: [
3    { host: "localhost", port: 26379 },
4    { host: "localhost", port: 26380 },
5  ],
6  name: "mymaster",
7});
8
9valkey.set("foo", "bar");

The arguments passed to the constructor are different from the ones you use to connect to a single node, where:

• name identifies a group of Valkey instances composed of a master and one or more slaves (mymaster in the example);
• sentinelPassword (optional) password for Sentinel instances.
• sentinels are a list of sentinels to connect to. The list does not need to enumerate all your sentinel instances, but a few so that if one is down the client will try the next one.
• role (optional) with a value of slave will return a random slave from the Sentinel group.
• preferredSlaves (optional) can be used to prefer a particular slave or set of slaves based on priority. It accepts a function or array.
• enableTLSForSentinelMode (optional) set to true if connecting to sentinel instances that are encrypted

iovalkey guarantees that the node you connected to is always a master even after a failover. When a failover happens, instead of trying to reconnect to the failed node (which will be demoted to slave when it's available again), iovalkey will ask sentinels for the new master node and connect to it. All commands sent during the failover are queued and will be executed when the new connection is established so that none of the commands will be lost.

It's possible to connect to a slave instead of a master by specifying the option role with the value of slave and iovalkey will try to connect to a random slave of the specified master, with the guarantee that the connected node is always a slave. If the current node is promoted to master due to a failover, iovalkey will disconnect from it and ask the sentinels for another slave node to connect to.

If you specify the option preferredSlaves along with role: 'slave' iovalkey will attempt to use this value when selecting the slave from the pool of available slaves. The value of preferredSlaves should either be a function that accepts an array of available slaves and returns a single result, or an array of slave values priorities by the lowest prio value first with a default value of 1.

1// available slaves format
2const availableSlaves = [{ ip: "127.0.0.1", port: "31231", flags: "slave" }];
3
4// preferredSlaves array format
5let preferredSlaves = [
6  { ip: "127.0.0.1", port: "31231", prio: 1 },
7  { ip: "127.0.0.1", port: "31232", prio: 2 },
8];
9
10// preferredSlaves function format
11preferredSlaves = function (availableSlaves) {
12  for (let i = 0; i < availableSlaves.length; i++) {
13    const slave = availableSlaves[i];
14    if (slave.ip === "127.0.0.1") {
15      if (slave.port === "31234") {
16        return slave;
17      }
18    }
19  }
20  // if no preferred slaves are available a random one is used
21  return false;
22};
23
24const valkey = new Valkey({
25  sentinels: [
26    { host: "127.0.0.1", port: 26379 },
27    { host: "127.0.0.1", port: 26380 },
28  ],
29  name: "mymaster",
30  role: "slave",
31  preferredSlaves: preferredSlaves,
32});

Besides the retryStrategy option, there's also a sentinelRetryStrategy in Sentinel mode which will be invoked when all the sentinel nodes are unreachable during connecting. If sentinelRetryStrategy returns a valid delay time, iovalkey will try to reconnect from scratch. The default value of sentinelRetryStrategy is:

1function (times) {
2  const delay = Math.min(times * 10, 1000);
3  return delay;
4}

Cluster

Valkey Cluster provides a way to run a Valkey installation where data is automatically sharded across multiple Valkey nodes. You can connect to a Valkey Cluster like this:

1const Valkey = require("iovalkey");
2
3const cluster = new Valkey.Cluster([
4  {
5    port: 6380,
6    host: "127.0.0.1",
7  },
8  {
9    port: 6381,
10    host: "127.0.0.1",
11  },
12]);
13
14cluster.set("foo", "bar");
15cluster.get("foo", (err, res) => {
16  // res === 'bar'
17});

Cluster constructor accepts two arguments, where:

0. The first argument is a list of nodes of the cluster you want to connect to. Just like Sentinel, the list does not need to enumerate all your cluster nodes, but a few so that if one is unreachable the client will try the next one, and the client will discover other nodes automatically when at least one node is connected.

1. The second argument is the options, where:

   • clusterRetryStrategy: When none of the startup nodes are reachable, clusterRetryStrategy will be invoked. When a number is returned, iovalkey will try to reconnect to the startup nodes from scratch after the specified delay (in ms). Otherwise, an error of "None of startup nodes is available" will be returned. The default value of this option is:

     1function (times) {
     2  const delay = Math.min(100 + times * 2, 2000);
     3  return delay;
     4}

     It's possible to modify the startupNodes property in order to switch to another set of nodes here:

     1function (times) {
     2  this.startupNodes = [{ port: 6790, host: '127.0.0.1' }];
     3  return Math.min(100 + times * 2, 2000);
     4}

   • dnsLookup: Alternative DNS lookup function (dns.lookup() is used by default). It may be useful to override this in special cases, such as when AWS ElastiCache used with TLS enabled.

   • enableOfflineQueue: Similar to the enableOfflineQueue option of Valkey class.

   • enableReadyCheck: When enabled, "ready" event will only be emitted when CLUSTER INFO command reporting the cluster is ready for handling commands. Otherwise, it will be emitted immediately after "connect" is emitted.

   • scaleReads: Config where to send the read queries. See below for more details.

   • maxRedirections: When a cluster related error (e.g. MOVED, ASK and CLUSTERDOWN etc.) is received, the client will redirect the command to another node. This option limits the max redirections allowed when sending a command. The default value is 16.

   • retryDelayOnFailover: If the target node is disconnected when sending a command, iovalkey will retry after the specified delay. The default value is 100. You should make sure retryDelayOnFailover * maxRedirections > cluster-node-timeout to insure that no command will fail during a failover.

   • retryDelayOnClusterDown: When a cluster is down, all commands will be rejected with the error of CLUSTERDOWN. If this option is a number (by default, it is 100), the client will resend the commands after the specified time (in ms).

   • retryDelayOnTryAgain: If this option is a number (by default, it is 100), the client will resend the commands rejected with TRYAGAIN error after the specified time (in ms).

   • retryDelayOnMoved: By default, this value is 0 (in ms), which means when a MOVED error is received, the client will resend the command instantly to the node returned together with the MOVED error. However, sometimes it takes time for a cluster to become state stabilized after a failover, so adding a delay before resending can prevent a ping pong effect.

   • redisOptions: Default options passed to the constructor of Valkey when connecting to a node.

   • slotsRefreshTimeout: Milliseconds before a timeout occurs while refreshing slots from the cluster (default 1000).

   • slotsRefreshInterval: Milliseconds between every automatic slots refresh (by default, it is disabled).

Read-Write Splitting

A typical valkey cluster contains three or more masters and several slaves for each master. It's possible to scale out valkey cluster by sending read queries to slaves and write queries to masters by setting the scaleReads option.

scaleReads is "master" by default, which means iovalkey will never send any queries to slaves. There are other three available options:

1. "all": Send write queries to masters and read queries to masters or slaves randomly.
2. "slave": Send write queries to masters and read queries to slaves.
3. a custom function(nodes, command): node: Will choose the custom function to select to which node to send read queries (write queries keep being sent to master). The first node in nodes is always the master serving the relevant slots. If the function returns an array of nodes, a random node of that list will be selected.

For example:

1const cluster = new Valkey.Cluster(
2  [
3    /* nodes */
4  ],
5  {
6    scaleReads: "slave",
7  }
8);
9cluster.set("foo", "bar"); // This query will be sent to one of the masters.
10cluster.get("foo", (err, res) => {
11  // This query will be sent to one of the slaves.
12});

NB In the code snippet above, the res may not be equal to "bar" because of the lag of replication between the master and slaves.

Running Commands to Multiple Nodes

Every command will be sent to exactly one node. For commands containing keys, (e.g. GET, SET and HGETALL), iovalkey sends them to the node that serving the keys, and for other commands not containing keys, (e.g. INFO, KEYS and FLUSHDB), iovalkey sends them to a random node.

Sometimes you may want to send a command to multiple nodes (masters or slaves) of the cluster, you can get the nodes via Cluster#nodes() method.

Cluster#nodes() accepts a parameter role, which can be "master", "slave" and "all" (default), and returns an array of Valkey instance. For example:

1// Send `FLUSHDB` command to all slaves:
2const slaves = cluster.nodes("slave");
3Promise.all(slaves.map((node) => node.flushdb()));
4
5// Get keys of all the masters:
6const masters = cluster.nodes("master");
7Promise.all(
8  masters
9    .map((node) => node.keys())
10    .then((keys) => {
11      // keys: [['key1', 'key2'], ['key3', 'key4']]
12    })
13);

NAT Mapping

Sometimes the cluster is hosted within a internal network that can only be accessed via a NAT (Network Address Translation) instance. See Accessing ElastiCache from outside AWS as an example.

You can specify nat mapping rules via natMap option:

1const cluster = new Valkey.Cluster(
2  [
3    {
4      host: "203.0.113.73",
5      port: 30001,
6    },
7  ],
8  {
9    natMap: {
10      "10.0.1.230:30001": { host: "203.0.113.73", port: 30001 },
11      "10.0.1.231:30001": { host: "203.0.113.73", port: 30002 },
12      "10.0.1.232:30001": { host: "203.0.113.73", port: 30003 },
13    },
14  }
15);

Or you can specify this parameter through function:

1const cluster = new Redis.Cluster(
2  [
3    {
4      host: "203.0.113.73",
5      port: 30001,
6    },
7  ],
8  {
9    natMap: (key) => {
10      if(key.indexOf('30001')) {
11        return { host: "203.0.113.73", port: 30001 };
12      }
13
14      return null;
15    },
16  }
17);

When is a dynamic natMap especially needed?

• Dockerized Redis clusters where IPs change frequently.
• Kubernetes-hosted Redis clusters with ephemeral Pods.
• Cloud deployments where private subnets or NAT gateways are used for Redis communication where NAT mappings frequently change.
• Scenarios with Redis node failover where failing nodes get replaced by new replicas and need rebalancing.

Example of problem in a distributed Redis cluster with NAT in a Kubernetes environment:

Your Redis client is configured to connect to 10.0.1.101:6379, but this is only accessible internally. The client uses static natMap to remap 10.0.1.101:6379 to 203.0.113.10:6379. A failure occurs, and the cluster rebalances, replacing 10.0.1.101:6379 with 10.0.1.105:6379. Without a function-based natMap, the static mapping is stale, and your client can no longer connect. With a function-based natMap, you dynamically fetch the new mapping for 10.0.1.105, ensuring continued access.

Specifying through may be useful if you don't know concrete internal host and know only node port.

Transaction and Pipeline in Cluster Mode

Almost all features that are supported by Valkey are also supported by Valkey.Cluster, e.g. custom commands, transaction and pipeline. However there are some differences when using transaction and pipeline in Cluster mode:

0. All keys in a pipeline should belong to slots served by the same node, since iovalkey sends all commands in a pipeline to the same node.
1. You can't use multi without pipeline (aka cluster.multi({ pipeline: false })). This is because when you call cluster.multi({ pipeline: false }), iovalkey doesn't know which node the multi command should be sent to.

When any commands in a pipeline receives a MOVED or ASK error, iovalkey will resend the whole pipeline to the specified node automatically if all of the following conditions are satisfied:

0. All errors received in the pipeline are the same. For example, we won't resend the pipeline if we got two MOVED errors pointing to different nodes.
1. All commands executed successfully are readonly commands. This makes sure that resending the pipeline won't have side effects.

Pub/Sub

Pub/Sub in cluster mode works exactly as the same as in standalone mode. Internally, when a node of the cluster receives a message, it will broadcast the message to the other nodes. iovalkey makes sure that each message will only be received once by strictly subscribing one node at the same time.

1const nodes = [
2  /* nodes */
3];
4const pub = new Valkey.Cluster(nodes);
5const sub = new Valkey.Cluster(nodes);
6sub.on("message", (channel, message) => {
7  console.log(channel, message);
8});
9
10sub.subscribe("news", () => {
11  pub.publish("news", "highlights");
12});

Events

Event	Description
connect	emits when a connection is established to the Valkey server.
ready	emits when CLUSTER INFO reporting the cluster is able to receive commands (if enableReadyCheck is true) or immediately after connect event (if enableReadyCheck is false).
error	emits when an error occurs while connecting with a property of lastNodeError representing the last node error received. This event is emitted silently (only emitting if there's at least one listener).
close	emits when an established Valkey server connection has closed.
reconnecting	emits after close when a reconnection will be made. The argument of the event is the time (in ms) before reconnecting.
end	emits after close when no more reconnections will be made.
+node	emits when a new node is connected.
-node	emits when a node is disconnected.
node error	emits when an error occurs when connecting to a node. The second argument indicates the address of the node.

Password

Setting the password option to access password-protected clusters:

1const Valkey = require("iovalkey");
2const cluster = new Valkey.Cluster(nodes, {
3  redisOptions: {
4    password: "your-cluster-password",
5  },
6});

If some of nodes in the cluster using a different password, you should specify them in the first parameter:

1const Valkey = require("iovalkey");
2const cluster = new Valkey.Cluster(
3  [
4    // Use password "password-for-30001" for 30001
5    { port: 30001, password: "password-for-30001" },
6    // Don't use password when accessing 30002
7    { port: 30002, password: null },
8    // Other nodes will use "fallback-password"
9  ],
10  {
11    redisOptions: {
12      password: "fallback-password",
13    },
14  }
15);

Special Note: Aws Elasticache Clusters with TLS

AWS ElastiCache for Valkey (Clustered Mode) supports TLS encryption. If you use this, you may encounter errors with invalid certificates. To resolve this issue, construct the Cluster with the dnsLookup option as follows:

1const cluster = new Valkey.Cluster(
2  [
3    {
4      host: "clustercfg.myCluster.abcdefg.xyz.cache.amazonaws.com",
5      port: 6379,
6    },
7  ],
8  {
9    dnsLookup: (address, callback) => callback(null, address),
10    redisOptions: {
11      tls: {},
12    },
13  }
14);

---

Autopipelining

In standard mode, when you issue multiple commands, iovalkey sends them to the server one by one. As described in Valkey pipeline documentation, this is a suboptimal use of the network link, especially when such link is not very performant.

The TCP and network overhead negatively affects performance. Commands are stuck in the send queue until the previous ones are correctly delivered to the server. This is a problem known as Head-Of-Line blocking (HOL).

iovalkey supports a feature called “auto pipelining”. It can be enabled by setting the option enableAutoPipelining to true. No other code change is necessary.

In auto pipelining mode, all commands issued during an event loop are enqueued in a pipeline automatically managed by iovalkey. At the end of the iteration, the pipeline is executed and thus all commands are sent to the server at the same time.

This feature can dramatically improve throughput and avoids HOL blocking. In our benchmarks, the improvement was between 35% and 50%.

While an automatic pipeline is executing, all new commands will be enqueued in a new pipeline which will be executed as soon as the previous finishes.

When using Valkey Cluster, one pipeline per node is created. Commands are assigned to pipelines according to which node serves the slot.

A pipeline will thus contain commands using different slots but that ultimately are assigned to the same node.

Note that the same slot limitation within a single command still holds, as it is a Valkey limitation.

Example of Automatic Pipeline Enqueuing

This sample code uses iovalkey with automatic pipeline enabled.

1const Valkey = require("./built");
2const http = require("http");
3
4const db = new Valkey({ enableAutoPipelining: true });
5
6const server = http.createServer((request, response) => {
7  const key = new URL(request.url, "https://localhost:3000/").searchParams.get(
8    "key"
9  );
10
11  db.get(key, (err, value) => {
12    response.writeHead(200, { "Content-Type": "text/plain" });
13    response.end(value);
14  });
15});
16
17server.listen(3000);

When Node receives requests, it schedules them to be processed in one or more iterations of the events loop.

All commands issued by requests processing during one iteration of the loop will be wrapped in a pipeline automatically created by iovalkey.

In the example above, the pipeline will have the following contents:

GET key1
GET key2
GET key3
...
GET keyN

When all events in the current loop have been processed, the pipeline is executed and thus all commands are sent to the server at the same time.

While waiting for pipeline response from Valkey, Node will still be able to process requests. All commands issued by request handler will be enqueued in a new automatically created pipeline. This pipeline will not be sent to the server yet.

As soon as a previous automatic pipeline has received all responses from the server, the new pipeline is immediately sent without waiting for the events loop iteration to finish.

This approach increases the utilization of the network link, reduces the TCP overhead and idle times and therefore improves throughput.

Benchmarks

Here's some of the results of our tests for a single node.

Each iteration of the test runs 1000 random commands on the server.

	Samples	Result	Tolerance
default	1000	174.62 op/sec	± 0.45 %
enableAutoPipelining=true	1500	233.33 op/sec	± 0.88 %

And here's the same test for a cluster of 3 masters and 3 replicas:

	Samples	Result	Tolerance
default	1000	164.05 op/sec	± 0.42 %
enableAutoPipelining=true	3000	235.31 op/sec	± 0.94 %

Error Handling

All the errors returned by the Valkey server are instances of ReplyError, which can be accessed via Valkey:

1const Valkey = require("iovalkey");
2const valkey = new Valkey();
3// This command causes a reply error since the SET command requires two arguments.
4valkey.set("foo", (err) => {
5  err instanceof Valkey.ReplyError;
6});

This is the error stack of the ReplyError:

ReplyError: ERR wrong number of arguments for 'set' command
    at ReplyParser._parseResult (/app/node_modules/iovalkey/lib/parsers/javascript.js:60:14)
    at ReplyParser.execute (/app/node_modules/iovalkey/lib/parsers/javascript.js:178:20)
    at Socket.<anonymous> (/app/node_modules/iovalkey/lib/redis/event_handler.js:99:22)
    at Socket.emit (events.js:97:17)
    at readableAddChunk (_stream_readable.js:143:16)
    at Socket.Readable.push (_stream_readable.js:106:10)
    at TCP.onread (net.js:509:20)

By default, the error stack doesn't make any sense because the whole stack happens in the iovalkey module itself, not in your code. So it's not easy to find out where the error happens in your code. iovalkey provides an option showFriendlyErrorStack to solve the problem. When you enable showFriendlyErrorStack, iovalkey will optimize the error stack for you:

1const Valkey = require("iovalkey");
2const valkey = new Valkey({ showFriendlyErrorStack: true });
3valkey.set("foo");

And the output will be:

ReplyError: ERR wrong number of arguments for 'set' command
    at Object.<anonymous> (/app/index.js:3:7)
    at Module._compile (module.js:446:26)
    at Object.Module._extensions..js (module.js:464:10)
    at Module.load (module.js:341:32)
    at Function.Module._load (module.js:296:12)
    at Function.Module.runMain (module.js:487:10)
    at startup (node.js:111:16)
    at node.js:799:3

This time the stack tells you that the error happens on the third line in your code. Pretty sweet! However, it would decrease the performance significantly to optimize the error stack. So by default, this option is disabled and can only be used for debugging purposes. You shouldn't use this feature in a production environment.

Running tests

Start a Valkey/Valkey server on 127.0.0.1:6379, and then:

1npm test

FLUSH ALL will be invoked after each test, so make sure there's no valuable data in it before running tests.

Debug

You can set the DEBUG env to iovalkey:* to print debug info:

1$ DEBUG=iovalkey:* node app.js

Join in!

I'm happy to receive bug reports, fixes, documentation enhancements, and any other improvements.

And since I'm not a native English speaker, if you find any grammar mistakes in the documentation, please also let me know. :)

License

MIT