Redis у Node.js: кешування, пайплайни та розподілені блокування (Redlock)

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

[ Рівень додатку Node.js ]
                             |
             +---------------+---------------+
             |                               |
             v                               v
    [ Читання з кешу ]              [ Критична секція ]
             |                               |
             v                               v
     +---------------+               +---------------+
     | Redis Кеш     |               | Розподілений  |
     | (XFetch тест) |               | замок(Redlock)|
     +---------------+               +---------------+
      /             \                        |
  Hit & Свіжий   Hit & Потребує оновлення    v
    /                 \             +-----------------+
   v                   v            | SET key val     |
[ Відповідь ]    [ Фонова задача ]  | NX PX 10000     |
                       |            +-----------------+
                       v                     |
               [ Запит до БД ]        Кворум отримано?
                       |              /              \
                       v            Так               Ні
               [ Оновлення кешу ]    |                 |
                                     v                 v
                              [ Робота секції ]  [ Повтор / Помилка ]
                                     |
                                     v
                              [ Зняття замка ]
                              (Lua скрипт перевірки токена)

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(() => {
            // Помилки фонового оновлення логуються в телеметрію
          });
        }

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