Forem: Abubakar Hassan

How to Prevent Race Conditions in a Node.js System Using Mutexes and RabbitMQ

Abubakar Hassan — Fri, 20 Dec 2024 21:39:31 +0000

In distributed systems, ensuring data consistency across multiple concurrent processes is critical. When dealing with sensitive operations like account withdrawals, race conditions can result in corrupted data or financial discrepancies. This article explains how to use mutexes (mutual exclusion locks) with RabbitMQ and Node.js to safely handle concurrent operations while maintaining efficient memory management.

The Problem

Imagine a scenario where multiple RabbitMQ workers process withdrawal requests from the same account simultaneously. Without proper synchronization, two workers might read the same initial balance, subtract different amounts, and then update the account, leading to an incorrect balance.

Race condition example:

Initial balance: $1000
Worker A reads the balance: $1000
Worker B reads the balance: $1000
Worker A subtracts $300 and updates the balance: $700
Worker B subtracts $200 and updates the balance: $800 (overwriting Worker A’s update)

Final balance: $800 (should have been $500).

Solution: Using Mutexes

A mutex ensures that only one worker can process operations on a specific account at any time. Here’s how we can implement a mutex-based solution in a RabbitMQ-driven Node.js system.

Step-by-Step Implementation

Prerequisites

Install the necessary packages:

npm install async-mutex amqplib

async-mutex: Provides an easy-to-use mutex for asynchronous operations.
amqplib: Enables interaction with RabbitMQ.

Code Implementation

Below is the complete implementation, including robust memory management for mutex cleanup.

a. Mutex Management

We use a Map to maintain mutexes per account. Mutexes are created on demand and removed when they are no longer needed.

b. Memory Optimization

To avoid memory bloat, we implement:

Idle Timeouts: Automatically remove mutexes for inactive accounts after 5 minutes.
Periodic Cleanup: A background process ensures stale mutexes are removed every 1 minute.

c. Full Implementation

const { Mutex } = require('async-mutex');
const amqp = require('amqplib');

// Mutex for account operations with automatic cleanup
const accountMutexes = new Map(); // Store one mutex per account
const accountTimeouts = new Map(); // Store timeout references for cleanup
const CLEANUP_INTERVAL_MS = 60000; // 1-minute cleanup interval
const IDLE_TIMEOUT_MS = 300000; // 5-minute idle timeout per account

// Function to get or create a mutex for a specific account
function getAccountMutex(accountId) {
    if (!accountMutexes.has(accountId)) {
        const mutex = new Mutex();
        accountMutexes.set(accountId, mutex);
        resetAccountTimeout(accountId); // Start idle timeout cleanup
    }
    return accountMutexes.get(accountId);
}

// Function to reset idle timeout for an account
function resetAccountTimeout(accountId) {
    if (accountTimeouts.has(accountId)) {
        clearTimeout(accountTimeouts.get(accountId));
    }

    const timeout = setTimeout(() => {
        accountMutexes.delete(accountId);
        accountTimeouts.delete(accountId);
        console.log(`Mutex for account ${accountId} removed due to inactivity.`);
    }, IDLE_TIMEOUT_MS);

    accountTimeouts.set(accountId, timeout);
}

// Periodic cleanup process
function startPeriodicCleanup() {
    setInterval(() => {
        accountTimeouts.forEach((_, accountId) => {
            if (!accountMutexes.has(accountId)) {
                accountTimeouts.delete(accountId);
            }
        });
    }, CLEANUP_INTERVAL_MS);

    console.log(`Periodic cleanup started: checking every ${CLEANUP_INTERVAL_MS / 1000} seconds.`);
}

// Simulated database of accounts
const accounts = {
    "123": { balance: 1000 },
    "456": { balance: 2000 },
};

// Process withdrawal
async function processWithdrawal(accountId, amount) {
    const mutex = getAccountMutex(accountId);
    const release = await mutex.acquire();

    try {
        console.log(`Processing withdrawal for account ${accountId}`);
        const account = accounts[accountId];

        if (!account) {
            throw new Error('Account not found');
        }

        if (account.balance < amount) {
            throw new Error('Insufficient funds');
        }

        account.balance -= amount;
        console.log(`Withdrawal successful! New balance for account ${accountId}: ${account.balance}`);
    } catch (error) {
        console.error(`Error processing withdrawal for account ${accountId}:`, error.message);
    } finally {
        release();
        resetAccountTimeout(accountId);
    }
}

// RabbitMQ message handler
async function handleMessage(message) {
    const { accountId, amount } = JSON.parse(message.content.toString());
    await processWithdrawal(accountId, amount);
}

// Connect to RabbitMQ and consume messages
(async () => {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    const queueName = 'withdrawals';
    await channel.assertQueue(queueName, { durable: true });

    console.log(`Waiting for messages in queue: ${queueName}`);

    channel.consume(queueName, async (msg) => {
        if (msg) {
            await handleMessage(msg);
            channel.ack(msg); // Acknowledge message after processing
        }
    });

    startPeriodicCleanup(); // Start periodic cleanup
})();

How It Works

Account-Specific Mutex:
- Each account has its own mutex (accountMutexes), allowing safe concurrency for different accounts.
- Mutexes are dynamically created when accessed for the first time.
Critical Section:
- The processWithdrawal function locks the mutex to ensure that only one worker can modify the account’s balance.
Memory Management:
- Idle Timeout: Removes mutexes after 5 minutes of inactivity.
- Periodic Cleanup: A background process runs every minute to clean up stale or unreferenced mutexes.

Advantages

Race Condition Prevention:
- Ensures that only one worker processes withdrawals for a given account at a time.
Efficient Memory Management:
- Automatically removes mutexes for inactive accounts, preventing memory bloat.
High Throughput:
- Concurrent processing of different accounts is unaffected, maintaining system scalability.
Robust Error Handling:
- Proper handling of account errors and lock release in the finally block ensures the system stays consistent.

Sample Output

Input Queue Messages:

{ "accountId": "123", "amount": 100 }
{ "accountId": "456", "amount": 200 }
{ "accountId": "123", "amount": 300 }

Console Output:

Waiting for messages in queue: withdrawals
Periodic cleanup started: checking every 60 seconds.
Processing withdrawal for account 123
Withdrawal successful! New balance for account 123: 900
Processing withdrawal for account 456
Withdrawal successful! New balance for account 456: 1800
Processing withdrawal for account 123
Withdrawal successful! New balance for account 123: 600
Mutex for account 123 removed due to inactivity.
Mutex for account 456 removed due to inactivity.

Conclusion

By combining mutexes with RabbitMQ, you can safely handle concurrent operations in Node.js systems.

Adding idle timeouts and periodic cleanup ensures efficient memory management, making this solution scalable and robust for real-world use cases.

How to end a telnet session (Linux, MAC, Windows)

Abubakar Hassan — Tue, 13 Dec 2022 16:33:17 +0000

Hello there 👋

Like me, you have started a telnet connection through the terminal and have no idea how to close it 🤣

If you currently don't have an active telnet connection, you can start one on the terminal by typing:

$ telnet address port

To end your current telnet session you must reach the telnet prompt and type quit.

Here are the steps for doing so:

Open the telnet prompt by holding down the 'Ctrl' key and push the ']' key. (prompt: telnet>)
Type quit.
Push the Enter key.

I hope this helped.

You're welcome 💪

How to build a minimal production-ready infrastructure with Terraform on DigitalOcean

Abubakar Hassan — Sun, 11 Dec 2022 23:21:32 +0000

This how-to will help you deploy a production-ready infrastructure on Digital Ocean using Terraform.

Pre-requisites

Install Terraform
Create a Digital Ocean account if you don't already have one (Use this link to get $100 credit)
Generate a Personal Access Token for your Digital Ocean account to access the DigitalOcean API. Go to API => Tokens/Keys => Generate New Token. Save the string generated.
Create a domain for your project. Go to Networking => Domains => Add domain

Initial setup terraform files

Open your terminal and create a new project directory, and open with you code editor.

$ mkdir minimal-prod`
$ cd minimal-prod`
$ code .

Create the following terraform files:

$ touch versions.tf
$ touch main.tf
$ touch variables.tf

In versions.tf, specify the Digital Ocean terraform provider as follows:

terraform {
  required_providers {
    digitalocean = {
      source  = "digitalocean/digitalocean"
      version = "2.25.2"
    }
  }
}

In main.tf, add the token required by the provider like this:

provider "digitalocean" {
  token = var.do_token
}

In variables.tf, create a new variable called do_token.

variable "do_token" {
  type        = string
  description = "Digital Ocean personal access token"
  default     = "<token_string>"
}

If you wish to commit variables.tf to your version control system, you might want to use a different file for more sensitive information, such as your Digital Ocean's personal access token. Create a new file terraform.tfvars and save the following to it:

do_token = "<digital_ocean_token>"

Replace <digital_ocean_token> with the actual value generated.
⚠️ Make sure *.tfvars is in your .gitignore file.

Now prepare your working directory for other commands by running the following command:

$ terraform init

Architecture Diagram

Below is a diagram representing the architecture that will be produced by executing the Terraform files at the end of this tutorial.

                       |                                                   
                     https                                                
                       |                                                   
                       v                                                    
              +--------------------+                                        
              |    Load Balancer   |                                        
 +--------------------------------------------------+                          
 |            |                    |                |                     
 |            +--------------------+                |                     
 |                     |                            |                     
 |                     |                            |                     
 |                    http              +---------+ |                     
 |                     |                |         | |                     
 |                     |                |         | |                     
 |                     +-------SSH------| Bastion |<---SSH---
 |                     |                |         | |                     
 |                     |                |         | |                     
 |                     |                +---------+ |                     
 |          +----------+---------+                  |                     
 |          |          |         |                  |                     
 |          v          v         v                  |                     
 |      +-------+  +-------+  +-------+             |                     
 |      |  web  |  |  web  |  |  web  |             |                     
 |      +---+---+  +---+---+  +---+---+             |                     
 |          |          |          |                 |                     
 |          |          v          |                 |                     
 |          |     +----------+    |                 |                     
 |          |     |          |    |                 |                     
 |          +---->| database |<---+                 |                     
 |                |          |                      |                     
 |                +----------+                      |                     
 +--------------------------------------------------+

Virtual Private Cloud (VPC) setup

Create a file network.tf and add the following to build the VPC

resource "digitalocean_vpc" "web" {
  name     = "${var.name}-vpc"
  region   = var.region
  ip_range = var.ip_range
}

Add new variables to variables.tf

...

variable "name" {
  type        = string
  description = "Infrastructure project name"
  default     = "minimal-prod"
}

variable "region" {
  type    = string
  default = "ams2"
}

variable "ip_range" {
  type        = string
  description = "IP range for VPC"
  default     = "192.168.22.0/24"
}

Run the command below to see the execution plan so far

$ terraform plan

Web Servers setup

Add a few more variables to variables.tf to be used in the future

...

variable "droplet_count" {
  type    = number
  default = 1
}

variable "image" {
  type        = string
  description = "OS to install on the servers"
  default     = "ubuntu-20-04-x64"
}

variable "droplet_size" {
  type    = string
  default = "s-1vcpu-1gb"
}

variable "ssh_key" {
  type = string
}

variable "subdomain" {
  type = string
}

variable "domain_name" {
  type = string
}

Create a new file data.tf and add a data resource for our ssh key which will be pulled from Digital Ocean directly:

data "digitalocean_ssh_key" "main" {
  name = var.ssh_key
}

Create a file servers.tf hold all the resources we'll be creating for our web servers as follows:

resource "digitalocean_droplet" "web" {
  count     = var.droplet_count
  image     = var.image
  name      = "web-${var.name}-${var.region}-${count.index + 1}"
  region    = var.region
  size      = var.droplet_size
  ssh_keys  = [data.digitalocean_ssh_key.main.id]
  vpc_uuid  = digitalocean_vpc.web.id
  tags      = ["${var.name}-webserver"]
  user_data = <<EOF
  #cloud-config
  packages:
    - nginx
    - postgresql
    - postgresql-contrib
  runcmd:
    - [ sh, -xc, "echo '<h1>web-${var.region}-${count.index + 1}</h1>' >> /var/www/html/index.html"]
  EOF
  lifecycle {
    create_before_destroy = true
  }
}

Update terraform.tfvars with the following variables for our environment:

...
region        = "ams3"
droplet_count = 3
ssh_key       = "<ssh_key_name_on_digitalocean>"
domain_name   = "<domain_added_on_digitalocean>"
subdomain     = "app"

Add a value for the domain you want to use in data.tf

...

data "digitalocean_domain" "web" {
  name = var.domain_name
}

Create a lets encrypt certificate to be used by the load balancer. Add the following to the servers.tf file

resource "digitalocean_certificate" "web" {
  name    = "${var.name}-certificate"
  type    = "lets_encrypt"
  domains = ["${var.subdomain}.${data.digitalocean_domain.web.name}"]
  lifecycle {
    create_before_destroy = true
  }
}

You can use the Digital Ocean terraform provider to provide your own certificate if you want to.

Next, we create our load balancer with the correct forwarding rules and the firewall setup in servers.tf. This is to block all inbound traffic directly to the web servers from the internet.

...

resource "digitalocean_loadbalancer" "web" {
  name                   = "web-${var.region}"
  region                 = var.region
  droplet_ids            = digitalocean_droplet.web.*.id
  vpc_uuid               = digitalocean_vpc.web.id
  redirect_http_to_https = true
  forwarding_rule {
    entry_port       = 443
    entry_protocol   = "https"
    target_port      = 80
    target_protocol  = "http"
    certificate_name = digitalocean_certificate.web.name
  }
  forwarding_rule {
    entry_port       = 80
    entry_protocol   = "http"
    target_port      = 80
    target_protocol  = "http"
    certificate_name = digitalocean_certificate.web.name
  }
  lifecycle {
    create_before_destroy = true
  }
}

resource "digitalocean_firewall" "web" {
  name        = "${var.name}-only-vpc-traffic"
  droplet_ids = digitalocean_droplet.web.*.id
  inbound_rule {
    protocol         = "tcp"
    port_range       = "1-65535"
    source_addresses = [digitalocean_vpc.web.ip_range]
  }
  inbound_rule {
    protocol         = "udp"
    port_range       = "1-65535"
    source_addresses = [digitalocean_vpc.web.ip_range]
  }
  inbound_rule {
    protocol         = "icmp"
    source_addresses = [digitalocean_vpc.web.ip_range]
  }
  outbound_rule {
    protocol              = "tcp"
    port_range            = "1-65535"
    destination_addresses = [digitalocean_vpc.web.ip_range]
  }
  outbound_rule {
    protocol              = "udp"
    port_range            = "1-65535"
    destination_addresses = [digitalocean_vpc.web.ip_range]
  }
  outbound_rule {
    protocol              = "icmp"
    destination_addresses = [digitalocean_vpc.web.ip_range]
  }
  outbound_rule {
    protocol              = "udp"
    port_range            = "53"
    destination_addresses = ["0.0.0.0/0", "::/0"]
  }
  outbound_rule {
    protocol              = "tcp"
    port_range            = "80"
    destination_addresses = ["0.0.0.0/0", "::/0"]
  }
  outbound_rule {
    protocol              = "tcp"
    port_range            = "443"
    destination_addresses = ["0.0.0.0/0", "::/0"]
  }
  outbound_rule {
    protocol              = "icmp"
    destination_addresses = ["0.0.0.0/0", "::/0"]
  }
}

Next, we create the record for the subdomain

...

resource "digitalocean_record" "web" {
  domain = data.digitalocean_domain.web.name
  type   = "A"
  name   = var.subdomain
  value  = digitalocean_loadbalancer.web.ip
  ttl    = 30
}

Database resource

Next, we add a few more variables to be used to setup our database in variables.tf as follows:

...

variable "db_count" {
  type    = number
  default = 1
}

variable "database_size" {
  type    = string
  default = "db-s-1vcpu-1gb"
}

Next, create database.tf to build the database. For this example we will be creating a Postgres database.

resource "digitalocean_database_cluster" "postgres-cluster" {
  name                 = "${var.name}-database-cluster"
  engine               = "pg"
  version              = "11"
  size                 = var.database_size
  region               = var.region
  node_count           = var.db_count
  private_network_uuid = digitalocean_vpc.web.id
}

resource "digitalocean_database_firewall" "postgress-cluster-firewall" {
  cluster_id = digitalocean_database_cluster.postgres-cluster.id
  rule {
    type  = "tag"
    value = "${var.name}-webserver"
  }
}

Jump server (Bastion) for accessing our infrastructure

Next, we need to create a jump server (Bastion) to access our infrastructure. Create a bastion.tf with the following:

resource "digitalocean_droplet" "bastion" {
  image    = var.image
  name     = "bastion-${var.name}-${var.region}"
  region   = var.region
  size     = "s-1vcpu-1gb"
  ssh_keys = [data.digitalocean_ssh_key.main.id]
  vpc_uuid = digitalocean_vpc.web.id
  tags     = ["${var.name}-webserver"]
  lifecycle {
    create_before_destroy = true
  }
}

resource "digitalocean_record" "bastion" {
  domain = data.digitalocean_domain.web.name
  type   = "A"
  name   = "bastion-${var.name}-${var.region}"
  value  = digitalocean_droplet.bastion.ipv4_address
  ttl    = 300
}

resource "digitalocean_firewall" "bastion" {
  name        = "${var.name}-only-ssh-bastion"
  droplet_ids = [digitalocean_droplet.bastion.id]
  inbound_rule {
    protocol         = "tcp"
    port_range       = "22"
    source_addresses = ["0.0.0.0/0", "::/0"]
  }
  outbound_rule {
    protocol              = "tcp"
    port_range            = "22"
    destination_addresses = [digitalocean_vpc.web.ip_range]
  }
  outbound_rule {
    protocol              = "icmp"
    destination_addresses = [digitalocean_vpc.web.ip_range]
  }
}

Now you can apply the files and let terraform create the infrastructure on Digital Ocean

$ terraform apply

You can use the --auto-approve flag to let terraform automatically continue without asking you for approval.

Access the server through the bastion host

Copy the domain name of the bastion host and ssh into it from your terminal.

$ ssh -A root@<FQDN of bastion host>

Copy the IP address of any of the web servers and ssh into it from the bastion host

$ ssh root@<private_ip_address_of_server>

Delete infrastructure

This will undo everything. You can delete the infrastructure by running the following command:

$ terraform destroy

That's it.

Thanks for the time reading this!

Let me know if you there's a way you think this infrastructure can be improved.
👍

How to recursively remove .DS_Store

Abubakar Hassan — Sat, 05 Nov 2022 12:58:56 +0000

Lately, I've been doing a lot of development on Mac. With that comes a lot of use of the Finder app.

In order keep your repo clean, you need to remove .DS_Store files from the folders created using the Mac GUI. Doing that by hand is really boring, painful, and repetitive. Enter that awesome thing we call the command line.

Recursively Remove .DS_Store

Open up Terminal

In the command line, type:

cd to/your/directory - then press enter.

Finally, in the command line, type:

find . -name '.DS_Store' -type f -delete - then press enter.

Warning: Never use a wildcard (*) with this command, unless you know what you're doing. Bad things can happen if you don't.

This snippet will remove .DS_Store files from the selected folder, as well as all of the folders that it contains. All of them. At once.

This command can be used for other types of files, as well. Just replace '.DS_Store' with whatever file name, or type, that you want to delete.

Other uses

You can do this for other file types, as well.

For example, you could use the below to delete all JavaScript files within a containing folder.

find . -name '*.js' -type f -delete

Note that the asterisk is a wildcard, meaning the Terminal is looking for any file with an extension of .js... it can be dangerous to use wildcards...

be careful.