Redis в Node.js: кэширование, пайплайны и распределенные блокировки (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
  ) {}

  /**
   * Вычисление необходимости раннего упреждающего обновления (XFetch)
   */
  private shouldRefreshEarly(
    computedAt: number,
    computeDurationMs: number,
    ttlMs: number,
    beta: number
  ): boolean {
    const expiryTimestamp = computedAt + ttlMs;
    const now = Date.now();

    // Если срок действия уже физически истек
    if (now >= expiryTimestamp) {
      return true;
    }

    // Случайная величина в диапазоне (0, 1]
    const random = Math.max(Math.random(), 0.0001);
    
    // Формула XFetch: now - (beta * delta * ln(random)) > expiry
    const probabilisticThreshold = now - (beta * computeDurationMs * Math.log(random));
    return probabilisticThreshold > expiryTimestamp;
  }

  /**
   * Чтение из кэша с автоматическим вероятностным упреждением
   */
  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) {
          // Асинхронно запускаем фоновое обновление без блокировки ответа клиенту
          this.triggerBackgroundRefresh(key, ttlSeconds, computeFn).catch(() => {
            // Ошибки фонового обновления логируются в telemetry
          });
        }

        return envelope.value;
      } catch (parseError) {
        // При повреждении JSON пересчитываем значение заново
      }
    }

    // Cache Miss: синхронный расчет и запись
    return this.executeAndStore(key, ttlSeconds, computeFn);
  }

  private async triggerBackgroundRefresh<T>(
    key: string,
    ttlSeconds: number,
    computeFn: () => Promise<T>
  ): Promise<void> {
    // Вспомогательный ключ защиты от параллельных фоновых расчетов
    const lockAcquired = await this.redis.set(`refresh_lock:${key}`, "1", "NX", "EX", 15);
    if (!lockAcquired) {
      return; // Другой инстанс уже производит фоновое обновление
    }

    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,
    };

    // Сохраняем в Redis с запасом TTL, чтобы предотвратить исчезновение ключа до XFetch
    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 {
  // Атомарный Lua-скрипт: удалять ключ только в том случае, если токен совпадает с владельцем
  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
  `;

  // Атомарный Lua-скрипт продления блокировки (Watchdog)
  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) {}

  /**
   * Захват распределенной блокировки с защитой токеном и автопродлением
   */
  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") {
        // Запуск фонового сторожевого таймера (Watchdog) для продления срока жизни замка
        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 };
      }

      // Пауза перед следующей попыткой захвата
      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 {
    // Продлеваем замок каждые 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) {
          // Замок был утерян или перехвачен
          clearInterval(timer);
        }
      } catch (err) {
        // Сетевой сбой продления
      }
    }, interval);

    // Unref таймера, чтобы не удерживать процесс Node.js при завершении
    timer.unref();
    return timer;
  }
}