Redis in Node.js: Caching, Pipelining, and Distributed Locks (Redlock)

now - (beta * delta * ln(random())) > expiry

[ Node.js Application Layer ]
                             |
             +---------------+---------------+
             |                               |
             v                               v
    [ Cache Read Flow ]             [ Critical Section ]
             |                               |
             v                               v
     +---------------+               +---------------+
     | Redis Cache   |               | Distributed   |
     | (XFetch Check)|               | Lock (Redlock)|
     +---------------+               +---------------+
      /             \                        |
  Hit & Fresh    Hit & Needs Early Refresh   v
    /                 \             +-----------------+
   v                   v            | SET key val     |
[ Return Data ]  [ Async Task ]     | NX PX 10000     |
                       |            +-----------------+
                       v                     |
               [ Query Database ]     Acquired Quorum?
                       |              /              \
                       v            Yes               No
               [ Update Cache ]      |                 |
                                     v                 v
                              [ Execute Work ]   [ Retry / Fail ]
                                     |
                                     v
                              [ Release Lock ]
                              (Lua Script Token Check)

import Redis from "ioredis";

export interface CacheEnvelope<T> {
  value: T;
  computedAt: number;
  computeDurationMs: number;
  ttlMs: number;
}

export class ProbabilisticCacheService {
  constructor(
    private readonly redis: Redis,
    private readonly defaultBeta = 1.0
  ) {}

  /**
   * Evaluates whether probabilistic early expiration should trigger (XFetch)
   */
  private shouldRefreshEarly(
    computedAt: number,
    computeDurationMs: number,
    ttlMs: number,
    beta: number
  ): boolean {
    const expiryTimestamp = computedAt + ttlMs;
    const now = Date.now();

    // Key has physically expired
    if (now >= expiryTimestamp) {
      return true;
    }

    // Uniform random float in range (0, 1]
    const random = Math.max(Math.random(), 0.0001);
    
    // XFetch inequality: now - (beta * delta * ln(random)) > expiry
    const probabilisticThreshold = now - (beta * computeDurationMs * Math.log(random));
    return probabilisticThreshold > expiryTimestamp;
  }

  /**
   * Retrieves an item from cache with automated probabilistic pre-warming
   */
  public async getOrCompute<T>(
    key: string,
    ttlSeconds: number,
    computeFn: () => Promise<T>,
    beta = this.defaultBeta
  ): Promise<T> {
    const rawData = await this.redis.get(key);

    if (rawData) {
      try {
        const envelope: CacheEnvelope<T> = JSON.parse(rawData);
        const needsRefresh = this.shouldRefreshEarly(
          envelope.computedAt,
          envelope.computeDurationMs,
          envelope.ttlMs,
          beta
        );

        if (needsRefresh) {
          // Trigger non-blocking asynchronous background refresh
          this.triggerBackgroundRefresh(key, ttlSeconds, computeFn).catch(() => {
            // Background computation errors logged to telemetry
          });
        }

        return envelope.value;
      } catch (parseError) {
        // Fallback on JSON parse corruption
      }
    }

    // Cache Miss: Synchronous compute and write-through
    return this.executeAndStore(key, ttlSeconds, computeFn);
  }

  private async triggerBackgroundRefresh<T>(
    key: string,
    ttlSeconds: number,
    computeFn: () => Promise<T>
  ): Promise<void> {
    // Acquire a non-blocking mutex to prevent concurrent refresh tasks
    const lockAcquired = await this.redis.set(`refresh_lock:${key}`, "1", "NX", "EX", 15);
    if (!lockAcquired) {
      return; // Another pod is already refreshing this key
    }

    try {
      await this.executeAndStore(key, ttlSeconds, computeFn);
    } finally {
      await this.redis.del(`refresh_lock:${key}`);
    }
  }

  private async executeAndStore<T>(
    key: string,
    ttlSeconds: number,
    computeFn: () => Promise<T>
  ): Promise<T> {
    const start = Date.now();
    const value = await computeFn();
    const duration = Date.now() - start;

    const envelope: CacheEnvelope<T> = {
      value,
      computedAt: Date.now(),
      computeDurationMs: Math.max(duration, 1),
      ttlMs: ttlSeconds * 1000,
    };

    // Store in Redis with safety buffer so key remains available for XFetch reads
    const storageTtl = ttlSeconds * 2;
    await this.redis.set(key, JSON.stringify(envelope), "EX", storageTtl);

    return value;
  }
}

import Redis from "ioredis";
import { randomUUID } from "node:crypto";

export class DistributedLockService {
  // Atomic Lua script: delete key only if token matches holding client
  private static readonly RELEASE_LOCK_LUA = `
    if redis.call("get", KEYS[1]) == ARGV[1] then
      return redis.call("del", KEYS[1])
    else
      return 0
    end
  `;

  // Atomic Lua script: extend TTL only if token matches holding client
  private static readonly EXTEND_LOCK_LUA = `
    if redis.call("get", KEYS[1]) == ARGV[1] then
      return redis.call("pexpire", KEYS[1], ARGV[2])
    else
      return 0
    end
  `;

  constructor(private readonly redis: Redis) {}

  /**
   * Acquires a distributed lock with token validation and background heartbeat renewal
   */
  public async acquire(
    resourceKey: string,
    ttlMs = 5000,
    acquireTimeoutMs = 3000,
    retryIntervalMs = 50
  ): Promise<{ token: string; release: () => Promise<boolean> }> {
    const lockKey = `lock:${resourceKey}`;
    const token = randomUUID();
    const deadline = Date.now() + acquireTimeoutMs;

    while (Date.now() < deadline) {
      // SET resource_key token NX PX ttl
      const acquired = await this.redis.set(lockKey, token, "PX", ttlMs, "NX");
      if (acquired === "OK") {
        // Start background Watchdog timer to prevent premature lock expiration
        const watchdogTimer = this.startWatchdog(lockKey, token, ttlMs);

        const release = async (): Promise<boolean> => {
          clearInterval(watchdogTimer);
          const result = await this.redis.eval(
            DistributedLockService.RELEASE_LOCK_LUA,
            1,
            lockKey,
            token
          );
          return result === 1;
        };

        return { token, release };
      }

      // Backoff delay before retry
      await new Promise((resolve) => setTimeout(resolve, retryIntervalMs));
    }

    throw new Error(`Failed to acquire distributed lock for [${resourceKey}] within ${acquireTimeoutMs}ms`);
  }

  private startWatchdog(lockKey: string, token: string, ttlMs: number): NodeJS.Timeout {
    // Extend lock every ttlMs / 3
    const interval = Math.max(Math.floor(ttlMs / 3), 500);

    const timer = setInterval(async () => {
      try {
        const renewed = await this.redis.eval(
          DistributedLockService.EXTEND_LOCK_LUA,
          1,
          lockKey,
          token,
          ttlMs.toString()
        );
        if (renewed !== 1) {
          // Lock was released or preempted
          clearInterval(timer);
        }
      } catch (err) {
        // Network heartbeat error
      }
    }, interval);

    // Unref timer so it does not block Node.js process termination
    timer.unref();
    return timer;
  }
}