Skip to content

Commit

Permalink
fix(cluster): autopipeline when prefix is used
Browse files Browse the repository at this point in the history
Previously the building of pipelines ignored the key prefix.
It was possible that two keys, foo and bar, might be set into
the same pipeline. However, after being prefixed by a configured
"keyPrefix" value, they may no longer belong to the same
pipeline.

This led to the error:
"All keys in the pipeline should belong to the same slots
allocation group"

Amended version of https://github.com/luin/ioredis/pull/1335/files

This may fix #1264 and #1248.
  • Loading branch information
mkemmerer authored and TysonAndre committed Jul 27, 2021
1 parent beefcc1 commit ec6e154
Show file tree
Hide file tree
Showing 2 changed files with 26 additions and 1 deletion.
2 changes: 1 addition & 1 deletion lib/autoPipelining.ts
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,7 @@ export function executeWithAutoPipelining(
}

// If we have slot information, we can improve routing by grouping slots served by the same subset of nodes
const slotKey = client.isCluster ? client.slots[calculateSlot(args[0])].join(",") : 'main';
const slotKey = client.isCluster ? client.slots[calculateSlot(client.options.keyPrefix ? `${client.options.keyPrefix}${args[0]}` : args[0])].join(",") : 'main';

if (!client._autoPipelines.has(slotKey)) {
const pipeline = client.pipeline();
Expand Down
25 changes: 25 additions & 0 deletions test/functional/cluster/autopipelining.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,10 @@ describe("autoPipelining for cluster", function () {
if (argv[0] === "get" && argv[1] === "foo6") {
return "bar6";
}

if (argv[0] === "get" && argv[1] === "baz:foo10") {
return "bar10";
}
});

new MockServer(30002, function (argv) {
Expand Down Expand Up @@ -68,6 +72,10 @@ describe("autoPipelining for cluster", function () {
return "bar5";
}

if (argv[0] === "get" && argv[1] === "baz:foo1") {
return "bar1";
}

if (argv[0] === "evalsha") {
return argv.slice(argv.length - 4);
}
Expand Down Expand Up @@ -177,6 +185,23 @@ describe("autoPipelining for cluster", function () {
cluster.disconnect();
});

it("should support building pipelines when a prefix is used", async () => {
const cluster = new Cluster(hosts, {
enableAutoPipelining: true,
keyPrefix: "baz:",
});
await new Promise((resolve) => cluster.once("connect", resolve));

await cluster.set("foo1", "bar1");
await cluster.set("foo10", "bar10");

expect(
await Promise.all([cluster.get("foo1"), cluster.get("foo10")])
).to.eql(["bar1", "bar10"]);

cluster.disconnect();
});

it("should support commands queued after a pipeline is already queued for execution", (done) => {
const cluster = new Cluster(hosts, { enableAutoPipelining: true });

Expand Down

0 comments on commit ec6e154

Please sign in to comment.