DEV Community: Munawwar

Concurrent Redis writes and correctness (Take 2)

Munawwar — Tue, 13 Jul 2021 10:01:27 +0000

Last year I wrote a blog post on concurrency control and ensuring data consistency while caching things to redis from multiple processes.

Recently I revisited the blog post again and I couldn't help wonder how over-complicated my final solution was. Custom lua script and stuff!?

Here is my attempt to fix what I started. Going back to the initial requirements:

Congestion Control - We need to try our best to call fetchFreshValues() as few times as possible, across all server processes, since it is expensive (how expensive? e.g. it was a super slow end-point that internally fires 100 of MySQL queries, which if it runs too many times, will slow the database down to a halt).
Data Consistency - We need to make sure that all processes returns a consistent value for a key. i.e. let's say our system calls fetchFreshValues() independently & parallelly from two processes, then only one should successfully write to redis whereas the other must fail.

Last blog post I established that redlock (a distributed lock algorithm from folks at redis) is great at congestion control, even though there could be edge cases where two processes manage to get a lock.

However for ensuring data consistency in the case two processes simultaneously gets hold of a lock, I did not have to go with a "versioned" solution. All I needed is redis to not overwrite a key if it already has a value. A simple SET NX command would have done the trick.

> SET myKey "myValue" EX 60 NX
"OK"
> SET myKey "myValue" EX 60 NX
(nil)

When a key expires, and if, two processes manage to get a lock, then only the first write would succeed. Which is what we wanted and is also how the old solution effectively worked. And with the return value, we can detect if a write succeeded or not.

Love the simplicity of this solution over the previous one. That's all for this post. Thanks for reading.

This post was reposted from https://www.codepasta.com/distributed-systems/2021/07/13/concurrent-redis-writes-and-correctness-take-2.html. Follow my blog at codepasta.com for long-form articles.

Queues with Redis without duplicate items

Munawwar — Mon, 28 Sep 2020 15:32:19 +0000

I work on an ecommerce store (Shopify), that has got webhooks* to auto-fix product issues like wrong product details, auto-fix assignment to the wrong inventory location etc. The way the system worked previously was that any update to a product immediately does all the checks and computations to try and auto-fix that product's details.

(* webhook is basically a trigger.. a notification. That is, I can set Shopify to notify me about all "product updates" or "order creation" etc to a URL on my server. The body of the http call is a JSON that I can process)

The Problem

The way merchandisers update products are occasionally through bulk actions. That is, for example first they change multiple products' pricing and then the same products' inventory etc. And they do this in a space of couple of minutes. And this causes lots of hits to the same webhook. Note that many of the updates are to the same products, so they are "duplicates".

Furthermore the webhook internally uses shopify APIs that are rate limited to around 2 requests per second. My webhooks were hitting this limit a lot especially when bulk actions are done. Which then requires retries, exponential backoff etc.

Overall this means my system was doing a lot of duplicated work and was hitting some hard limits.

Solution

The solution I imagined is that the webhook adds these product ids to a queue and every 30 minutes or so, take the ids from the queue, find unique ids and process them in one go.

Implementation 1 - RabbitMQ

One of the often used queuing system is RabbitMQ. However RabbitMQ cannot de-duplicate the messages of the queue "nicely".

What does "nicely" mean? Well.. there is a way, which involves reading all the messages in the queue, and then finding unique ones through code. That is, deduplication done by the consumer. I would rather not have duplicates in the queue in the first place, if possible.

Besides I already have redis running. If redis can be used, I could avoid adding yet another system for me to maintain.

Implementation 2 - Redis

Redis has several data structures in its feature list. It has regular queues. But that is just like RabbitMQ in that messages will have to deduplicated by my code. Redis has got another interesting data structure called "sorted set".

A sorted set is a data structure where each item in the set also has a numeric score, so that items can be fetched in sorted order of the scores.

This data structure may not look very intuitive at first glance. A queue is a just a list.. pretty easy to understand. You can push and pop items to and from it. But how to use a sorted set as a queue?

To build the intuition, note that even with queues each item in a queue have got a numeric association. If you are told that "you are the 10th person in the queue" means there are 9 people ahead of you. And the next person joining the queue will be the 11th person. If you think of position as a "score", then a queue's scoring logic for a new item is always the number of items previously in the queue + 1.

But in a sorted set, the score for a new item doesn't have to be strictly the position of the item. It can be any increasing number.. like time. Time is easy to access as well. Though time can have duplicates if two items are added in the same millisecond. However in our case that is not a problem as sorted sets allows duplicate scores.. and few duplicate scores are ok, as long as they do not majorly alter the overall order of insertions. And we are more concerned about keys being unique.

With that said, I am going to write down the redis command that will add a key to the sorted set if and only if it doesn't exist already in the set.

ZADD product-updates NX <currentTimeInMilliseconds> product-id-1

The NX in the command prevents duplicate product ids from getting into the "queue". Now from the consumer side, I can fetch all the oldest product ids that were added into the set and limit the result to 100. Note that I am not removing the items from the set.. just reading them.

ZRANGEBYSCORE product-updates 0 <currentTimeInMilliseconds> LIMIT 0 100

I can process each item (product id) and then remove the items either one by one as I process them, or remove all the 100 items in single command.

ZREM product-updates product-id-1 product-id-2 ...

Workers and load-balancing

For me a single process/thread is sufficient to do the job as I would hit shopify's rate limits if I try to parallelize the work. However, maybe for another use case that doesn't have such limits, you would probably like to have multiple workers doing the work. How would we do this?

My solution would be with the following assumptions:

we need to know how many workers are going to run upfront
each worker will have a queue dedicated for itself

The main problem here is ensuring de-duplication works now. How can we de-duplicate when there are multiple queues?

The trick I am going to pull is by using a hash function.

hash("product-id") % numberOfQueues

A hash function takes a string and returns a number. That's how the remainder operator % can work. It will always return the same number for the same string. The hash function output must be pretty random for this to work reliably.

There are several hash functions out there. One of my favorite is fnv1a, since it is tiny in implementation (literally 10 lines of code).

So let's take an example. Let's say we have 4 workers/queues.

fnv1a("product-id-1") % 4 // = 0 always, if one uses https://www.npmjs.com/package/fnv1a

You will always get the same queue number if you pass the the same product id. So producers don't need to handle duplicates, rather they try to consistently add a given product id to the same queue number as long as the number of queues is constant. And our ZADD NX command will handle duplicates within a queue.

(By the way, this approach is also known as "sticky session load balancing")

That's all for this post. The post was mostly about how to implement a queue with Redis without containing any duplicates.

Shorter Unique IDs

Munawwar — Wed, 09 Sep 2020 13:38:45 +0000

Unique IDs

If you have dealt with any form data & database, you have had the need to assign unique IDs to data saved. Many a time these IDs are internal, but sometimes IDs need to be used in other ways by your users (e.g. to be emailed or pasted into some form etc). And the common two technique devs use for assigning an ID are either:

a sequential number - an auto incrementing ID
a random number generator - like UUID v4

However they both have at least one downside. The downside of a sequential number is that (obviously) the IDs are too predictable. It is too easy to run some scraper iterating over your URLs.
And the downside of uuid v4 is that, well, it's long and ugly.

For internal IDs, that users don't need to deal with, both auto incrementing ID or uuid v4 seems fine. But when they got to paste it, email it or speak it over a phone or communicate it in any way, then it would be nicer to have shorter IDs. I have this use-case now, and I am thinking.. are any another alternatives where the IDs generated are shorter?

What does Stack overflow say?

A stack overflow answer gives me the math behind calculating the probability of generating a colliding ID. The math is called the "Birthday problem"

Given n number of IDs to generate and c number of characters of b character set (like hexadecimal), the probability of collision of IDs, approximately is:

p (approximate) = (n^2) / (2 * b^c)

(Academic note: the equation approximation assumes that p is less than 0.5)

Cool! So let's decide on n, b and a "safe" probability of collision.

n - the number of IDs we are going to generate. I am taking this number to be as 10 million. My data is not in the "big data" space so this is big enough.

b - character set. uuid v4 uses hex. But it's better if we can use a larger character set, since that would make the ID shorter.
For my requirement, I need to be able to use it in a URL part (and also be printable on paper). So I am going for upper case alphanumeric characters. Which gives us 36 characters. I would take a tip from Base58 encoding and avoid visually similar characters (namely 0,O,I and L), but.. for this post let me stick to upper case alphanumeric characters.

p - and finally a "safe" margin for collisions. I have run simulation tests on various p values, and figured out 0.001 is sufficient to ensure IDs are unique, given that you are using a reasonably good pseudo random number generator. I am using node.js Math.random() function (Math.random() is "crackable"/predictable though, so if you want more security use a secure random number generator).

We can now compute how many characters that is required:

0.001 = (10^7)^2 / (2 * 36^c)

And the answer is approximately 11 character length.

So for 10m items, a 11 character alphanumeric ID seems sufficient to prevent ID collisions.

Sortability

The random IDs I have suggested so far are not sortable by creation date. Being able to sort the IDs, also makes sequential access (from disk) for databases faster. So some databases like MongoDB use a timestamp prefix and random suffix to generate their IDs. This would keep the ID pretty random looking, and yet sortable by string prefix.

One such convention would be = "YYSSSSSRRRR"
where
YY = year in base 36, "zero" padded to 2 character length
SSSSS = seconds from start of year in base 36, "zero" padded to 5 character length,
RRRR = 4 random base 36 characters,
giving total of 11 character length like the math I calculated before

Additional reference on this topic:

That's all folks. You can use the math and adjust the character length for your use case. The point of this post was to come up with an educated guess on how many characters you really need.

Implementation with JS

If I go for the 32 glyph character set, the character length I need is still 11 characters. So I am going to implement that.

const characterSet = '123456789ABCDEFGHJKMNPQRSTUVWXYZ';
function generateUniqueId() {
  return 'x'
    .repeat(11)
    .replace(/x/g, () => characterSet[Math.trunc(Math.random() * 32)]);
}

generateUniqueId() // e.g. YMSATXVFM94AG

Blog Post Revision 2021

In the initial version of this blog post:

I used p value in percentage. Thats the way I was thinking at that point in time. But in mathematic literature, everyone use p as a value between 0 and 1. Also saying "0.1%" is no more intuitive than saying "0.001", so I changed equation back to value between 0 and 1.
I tried to make the IDs "unguessable" to brute force attacks over an HTTP end-point. I took a arbitrary number to multiply the p value with, to try and protect from such attacks, but in reality a brute force attack's success totally depends on how many requests the attacker can make to the end-point. I don't want to get deeper into that topic for now. So the value of p is now based on n, the max number of IDs that one intended to generate, which I used in my collision simulation tests as well.

Lodash chaining alternative

Munawwar — Wed, 15 Jul 2020 21:04:33 +0000

Those of you dealing with data transforms/manipulations for charting/dashboards/whatever, need no introduction to Lodash library, and have been using it happily on the backend, frontend etc

The problem

However there is one method Lodash has that's so useful but has its performance implications on the frontend.. namely chain().
On the frontend if you naively import chain the entire lodash library will end up in your bundle.. and the entire lodash library isn't small. Backend code doesn't care about some extra bloat.

That's sad. chain is very useful and I would like to use chaining on the frontend without having to take a performance hit. So.. what is the solution?

What does Google say?

Googling around you would see many suggestions to use lodash/fp's flow() method. You can see the code from this 2016 post

import map from "lodash/fp/map";
import flatten from "lodash/fp/flatten";
import sortBy from "lodash/fp/sortBy";
import flow from "lodash/fp/flow";
flow(
  map(x => [x, x*2]),
  flatten,
  sortBy(x => x) 
)([1,2,3]);

It works.. it keeps the bundle size small and gives you chaining capability.

But there is something bothersome about that code...

_.chain begins with the data you need to manipulate and then you call the transforms.. whereas flow() begins with the transforms and ends with the data you want to manipulate. This isn't natural to read. It needs to be flipped around.

[From flow()'s perspective it is built as intended. flow is potentially built for reuse. Fine. However we still miss a closer alternative to chain.]

Alternative solution

My ideal syntax would be the following (for the same code sample from above):

chain([1,2,3])
  (map, x => [x, x*2])
  (flatten)
  (sortBy, x => x)
  ();

However most linter configurations would complain about the indented parentheses. So we need a dummy function and a .value() to break out of the chain (like lodash already does)

chain([1,2,3])
  .fn(map, x => [x, x*2])
  .fn(flatten)
  .fn(sortBy, x => x)
  .value();

Overall if you squint your eyes and ignore the .fn()s, then it looks very similar to lodash's _.chain syntax. And there is a way to implement this. I'll dive straight into the implementation which is small and probably don't need too much explanation:

function chain(value) {
  return {
    fn: (func, ...args) => chain(func(value, ...args)),
    value: () => value,
  };
}

This implementation brings some new opportunity considering how generic the approach is.

The function doesn't know anything about lodash. It takes in any function. So you can write custom functions or use Math.* or Object.* functions

chain({prop: 2, fallback: 1})
  .fn((obj) => obj.prop || obj.fallback)
  .fn(Math.pow, 2)
  .value(); // result = 4

Improvement

With a slight modification, we can make it call any function on result objects.

Which mean for arrays, we can use native array map, filter etc, and we don't really need to use lodash's functions there. We should be able to do something like the following (using same example from before):

chain([1,2,3])
  .fn('map', x => [x, x*2])
  // ... blah
  .value();

Instead of passing the function here we put a name of the method to be invoked from the intermediate result object/array. The implementation of fn will change to the following:

    fn(func, ...args) {
      if (typeof func === 'function') {
        return chain(func(value, ...args));
      }
      return chain(value[func](...args));
    },

We can further improve this to handle promises / async functions:

    fn(func, ...args) {
      if (value instanceof Promise) {
        return chain(value.then((result) => {
          if (typeof func === 'string') {
            return result[func](...args);
          }
          return func(result, ...args);
        }));
      }
      if (typeof func === 'string') {
        return chain(value[func](...args));
      }
      return chain(func(value, ...args));
    },

And then it can be used as follows:

const dates = await chain(Promise.resolve([1, 2, 3]))
  // with method name
  .fn('map', x => x.toString())
  // with functions
  .fn((arr) => arr.map(x => new Date(x)))
  .value();

I believe this is an improvement to the popular approaches suggested out there on the interwebz. Check it out, try it out.. criticism welcome.

That's all folks. Hope you liked my short, hastily written post.

Full code below:

/** @type {import('./chain').default} */
export default function chain(value) {
  return {
    fn(func, ...args) {
      if (value instanceof Promise) {
        return chain(value.then((result) => {
          if (typeof func === 'string') {
            return result[func](...args);
          }
          return func(result, ...args);
        }));
      }
      if (typeof func === 'string') {
        return chain(value[func](...args));
      }
      return chain(func(value, ...args));
    },
    value() {
      return value;
    },
  };
}

chain.d.ts

declare type ParametersExceptFirst<F> = F extends (arg0: any, ...rest: infer P) => any ? P : never;
declare type MethodsOf<T> = {
  [K in keyof T]: T[K] extends ((...arg: any[]) => any) ? T[K]: never;
};

export default function chain<V>(val: V): {
  fn<Func extends (v: Awaited<V>, ...args: any[]) => any>(func: Func, ...args: ParametersExceptFirst<Func>): ReturnType<typeof chain<ReturnType<Func>>>;
  fn<Name extends keyof MethodsOf<Awaited<V>>>(
    func: Name,
    ...args: Awaited<V>[Name] extends ((...args: infer P) => any) ? P : never
  ): ReturnType<typeof chain<ReturnType<MethodsOf<Awaited<V>>[Name]>>>;
  value(): V;
};
export {};

EDIT 2022: Added type definition and improved implementation
EDIT 2024: Added promise / async function handling

Concurrent Redis writes and correctness

Munawwar — Sat, 22 Feb 2020 19:44:41 +0000

Lately I've been trying to re-implement a cache library using redis. The (simplified) requirement is to use cached value for a given key if present else fetch fresh value and cache the value to redis (on the first time or the first time after cache value expires). Also assume that fetching fresh value, even once, is a super expensive operation (it causes heavy load on the database).

Seems like a simple requirement, right? But one does not simply write distributed system code!

So we have couple of challenges here.

Congestion Control - We need to try our best to call fetchFreshValues() as few times as possible, across all server processes, since it is expensive.
Data Consistency - We need to make sure that all processes returns a consistent value for a key. i.e. let's say our system calls fetchFreshValues() independently & parallelly from two processes, then only one should successfully write to redis whereas the other must fail.

The one that failed must fetch whatever was saved and return that instead of the result of fetchFreshValues() it got. You might be asking whether this is necessary. I'd reckon that a system either fails or return a stale value rather than returning potentially different value than the other processes (i.e. correctness. though stale value blows away the "correctness" part of my blog's title I think.. but you can adapt the info from this blog for correctness by returning error in this case).

One way to achieve the above is to acquire a distributed lock on a key, then fetch latest value and write to redis. The recommendation from Redis is to use redlock for achieving distributed lock.
And so for a long time we did use redlock for this, yet there has been criticism about the algorithm much before we started using it -> https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html

The main takeaway from Martin Kleppmann's post is that redlock's dependence on time cannot guarantee correctness on acquiring an exclusive lock nor for the subsequent database/storage operations. There may be situations that two process manages to acquire a lock to the same resource (either due to a system time reset or more likely because of the lock expiring before the DB/storage write began).

Martin further suggests that to fix this, redlock needs to issue incrementing "fencing tokens", and that the storage layer collaborates to accept writes for only those tokens that are greater in number than the saved data's token number (conversely, reject writes that are lesser than or equal to the saved data's token number).

However in my case, I am using Redis for the cache storage. And redis doesn't seem to have this kind of concurrency control out-of-the-box. If two parallel processes tries to write to the same key at the same time, redis will accept both and the last write wins.

The work-around

UPDATE 2021: Read my new post on a simpler solution. You can skip to the last section of this post.

Old solution

The first time I read the fencing tokens solution, it felt strongly similar to multi-version document systems (like CouchDB)... that rejects writes if the version sent doesn't match the one on disk (and the new version gets incremented on save). I think this is called multi-version concurrency control (MVCC).

Redis can't do that out-of-the-box. But searching around, I came across a potential way to implement this. Using Redis scripts - which are custom Lua scripts that redis would run atomically and transactionally + gets rolled back if script fails part-ways. References -> https://redis.io/topics/transactions and https://redis.io/commands/eval

Quoting the relevant sections:

A Redis script is transactional by definition

Also Redis guarantees that a script is executed in an atomic way: no other script or Redis command will be executed while a script is being executed. This semantic is similar to the one of MULTI / EXEC. From the point of view of all the other clients the effects of a script are either still not visible or already completed.

Seems cool. But what does redis do in a multi-node setup? From their docs:

All Redis commands must be analyzed before execution to determine which keys the command will operate on. In order for this to be true for EVAL, keys must be passed explicitly. This is useful in many ways, but especially to make sure Redis Cluster can forward your request to the appropriate cluster node.

Ok. So I presume, a multi-node setup means that a given key only has one master - i.e. keys are sharded across nodes. And that the non-master nodes are replicas. So redis uses the key passed as separate argument, to make the decision about which node the script would run on. Seems cool so far.

Yet another question remains. What about replicas? We know replicas are async in nature, but will the script execute in order on the replicas as the original order? I don't know the answer to this. I am guessing "yes".

Assuming the answer is "yes, order is maintained on replicas", then that does make redis scripts an option to implement a simple version check before every write.

So let's dive in to my potential solution (Lua codez!):

local newPayload = ARGV[1]
local newVersionStr, newData = ARGV[1]:match("^([0-9]+)|(.+)$")
local prevVal = redis.call('get', KEYS[1]) or nil

if prevVal == nil then
    return redis.call('set', KEYS[1], "1|" .. newData)
end

local oldVersionStr, oldData = prevVal:match("^([0-9]+)|(.+)$")
local newVersion = tonumber(newVersionStr)
local oldVersion = tonumber(oldVersionStr)

-- check if version matches before writing
if oldVersion == (newVersion - 1) then
    return redis.call('set', KEYS[1], newPayload)
else
    return nil
end

Redis command:

EVAL "the script as above" 1 "key-to-write" "new-version-number|new data"

What I am doing there is writing a version number for the first write and making sure that for subsequent writes, the application sends the data along with plus-oned version number. If two processes try to use the same version, then the first one would succeed and second would receive null. The one that received null would then know that another process has changed the data, so it can handle the case appropriately.

Two small notes here:

I didn't use JSON for saving version number, cause some of our JSONs are so huge that it did slow down node.js's JSON.parse() in the past (but this may not be true for redis/lua, I didn't measure it).
The version check is pretty rigid oldVersion == (newVersion - 1). Probably making the check as newVersion > oldVersion would suffice and potentially make the writes order insensitive for the replicas.

So I think that prevents unintentional overwrites of data by two parallel processes.

Notes

At this point you might ask, whether there is a point in using redlock anymore? I think it is still valuable... as even though it does not solve for data consistency, it still solves congestion control (from my initial requirements). From a probabilistic perspective, the system most likely would only have one write-lock per resource and the lock acquirer successfully writes to storage before the lock expires. This would work most of the time. And the "most of the time" is what's important for congestion control. So I would implement the data consistency solution, plus also keep redlock.

Hope you enjoyed this piece. I'd like more eyes reviewing that code I wrote there. Here is the code with a small test to play with. Any criticism welcome.

This post was reposted from https://www.codepasta.com/distributed-systems/2020/02/22/concurrent-redis-writes-and-correctness.html. Follow me on twitter for short ideas/tips/tricks on programming and follow my blog at codepasta.com for long-form articles.