Single-Flight 기반 캐시 미스 병합

Single-Flight
= 같은 Key의 동시 작업 N개
→ 실제 Loader 실행 1개
→ 결과를 N명의 Waiter가 공유

In-Flight Registry
= Key → 현재 실행 중인 Promise / Task / Future

Cache
= 완료된 값을 일정 기간 보존

Single-Flight
= 완료 전 작업만 일시적으로 공유

불변식:
동일 key에는 동시에 loader가 하나만 실행된다.
다른 key의 loader는 서로 독립적으로 실행된다.
loader가 끝나면 성공과 실패에 관계없이 entry를 제거한다.
waiter 하나가 떠나도 다른 waiter가 쓰는 loader는 계속 실행된다.

상품 조회 요청
→ cache 조회
→ cache hit이면 즉시 반환
→ cache miss이면 DB 조회
→ cache 저장
→ 결과 반환

요청 1   cache miss → DB 조회
요청 2   cache miss → DB 조회
요청 3   cache miss → DB 조회
...
요청 100 cache miss → DB 조회

요청 1
→ product-42 loader 실행권 획득
→ DB 조회와 cache 저장 수행

요청 2~100
→ 같은 product-42 작업이 진행 중임을 확인
→ 별도 DB 조회 없이 기존 작업 대기
→ 동일 결과 공유

product-42 → loader 1개
product-77 → loader 1개

두 key의 loader는 동시에 실행 가능

SingleFlight.run(key, loader, callerCancellation)

key
= 병합 단위

loader
= 실제 DB·API 조회와 cache 저장을 수행하는 공유 작업

callerCancellation
= 개별 호출자의 대기만 중단

serviceShutdown
= 공유 loader 자체를 중단할 수 있는 상위 수명 신호

using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

public sealed class SingleFlight<TKey, TValue>
    where TKey : notnull
{
    private readonly ConcurrentDictionary<
        TKey,
        Lazy<Task<TValue>>> inFlight = new();

    private readonly CancellationToken shutdownToken;

    public SingleFlight(
        CancellationToken shutdownToken = default)
    {
        this.shutdownToken = shutdownToken;
    }

    public async Task<TValue> RunAsync(
        TKey key,
        Func<CancellationToken, Task<TValue>> loaderAsync,
        CancellationToken callerToken = default)
    {
        ArgumentNullException.ThrowIfNull(loaderAsync);

        Lazy<Task<TValue>>? candidate = null;

        candidate = new Lazy<Task<TValue>>(
            () => this.ExecuteAndReleaseAsync(
                key,
                candidate!,
                loaderAsync),
            LazyThreadSafetyMode.ExecutionAndPublication);

        Lazy<Task<TValue>> selected =
            this.inFlight.GetOrAdd(key, candidate);

        return await selected.Value
            .WaitAsync(callerToken)
            .ConfigureAwait(false);
    }

    public int GetInFlightCount()
    {
        return this.inFlight.Count;
    }

    private async Task<TValue> ExecuteAndReleaseAsync(
        TKey key,
        Lazy<Task<TValue>> entry,
        Func<CancellationToken, Task<TValue>> loaderAsync)
    {
        try
        {
            return await loaderAsync(this.shutdownToken)
                .ConfigureAwait(false);
        }
        finally
        {
            this.RemoveIfCurrent(key, entry);
        }
    }

    private void RemoveIfCurrent(
        TKey key,
        Lazy<Task<TValue>> entry)
    {
        // key만 보고 지우면, 교체된 새 entry까지 제거할 수 있다.
        // key와 value가 모두 같은 경우에만 현재 entry를 삭제한다.
        this.inFlight.TryRemove(new KeyValuePair<
            TKey,
            Lazy<Task<TValue>>>(key, entry));
    }
}

export type AsyncLoader<TValue> = (
  shutdownSignal?: AbortSignal,
) => Promise<TValue>;

export class SingleFlight<TValue>
{
  readonly #inFlight =
    new Map<string, Promise<TValue>>();

  readonly #shutdownSignal: AbortSignal | undefined;

  public constructor(
    shutdownSignal?: AbortSignal,
  )
  {
    this.#shutdownSignal = shutdownSignal;
  }

  public async runAsync(
    key: string,
    loaderAsync: AsyncLoader<TValue>,
    callerSignal?: AbortSignal,
  ): Promise<TValue>
  {
    validateKey(key);
    callerSignal?.throwIfAborted();

    const sharedPromise = this.#getOrCreate(
      key,
      loaderAsync,
    );

    return await waitForSharedAsync(
      sharedPromise,
      callerSignal,
    );
  }

  public getInFlightCount(): number
  {
    return this.#inFlight.size;
  }

  #getOrCreate(
    key: string,
    loaderAsync: AsyncLoader<TValue>,
  ): Promise<TValue>
  {
    const existing = this.#inFlight.get(key);

    if (existing !== undefined) {
      return existing;
    }

    const created = Promise.resolve().then(() =>
    {
      return loaderAsync(this.#shutdownSignal);
    });

    this.#inFlight.set(key, created);
    this.#registerCleanup(key, created);

    return created;
  }

  #registerCleanup(
    key: string,
    promise: Promise<TValue>,
  ): void
  {
    const cleanup = (): void =>
    {
      if (this.#inFlight.get(key) === promise) {
        this.#inFlight.delete(key);
      }
    };

    void promise.then(cleanup, cleanup);
  }
}

function waitForSharedAsync<TValue>(
  sharedPromise: Promise<TValue>,
  callerSignal?: AbortSignal,
): Promise<TValue>
{
  if (callerSignal === undefined) {
    return sharedPromise;
  }

  return new Promise<TValue>((resolve, reject) =>
  {
    let settled = false;

    const cleanup = (): void =>
    {
      callerSignal.removeEventListener(
        "abort",
        onAbort,
      );
    };

    const onAbort = (): void =>
    {
      if (settled) {
        return;
      }

      settled = true;
      cleanup();
      reject(callerSignal.reason);
    };

    callerSignal.addEventListener(
      "abort",
      onAbort,
      { once: true },
    );

    // 사전 throwIfAborted()와 listener 등록 사이에 abort가 발생할 수 있다.
    // listener를 붙인 뒤 상태를 다시 확인해 그 틈을 닫는다.
    if (callerSignal.aborted) {
      onAbort();
      return;
    }

    sharedPromise.then(
      (value) =>
      {
        if (settled) {
          return;
        }

        settled = true;
        cleanup();
        resolve(value);
      },
      (error: unknown) =>
      {
        if (settled) {
          return;
        }

        settled = true;
        cleanup();
        reject(error);
      },
    );
  });
}

function validateKey(key: string): void
{
  if (key.trim().length === 0) {
    throw new RangeError(
      "Single-Flight key는 비어 있을 수 없습니다.",
    );
  }
}

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Callable
from typing import Generic, TypeVar


TValue = TypeVar("TValue")
AsyncLoader = Callable[[], Awaitable[TValue]]


class SingleFlight(Generic[TValue]):
    def __init__(self) -> None:
        self._in_flight: dict[
            str,
            asyncio.Task[TValue],
        ] = {}

        self._lock = asyncio.Lock()

    async def run(
        self,
        key: str,
        loader_async: AsyncLoader[TValue],
    ) -> TValue:
        _validate_key(key)

        task = await self._get_or_create(
            key,
            loader_async,
        )

        return await asyncio.shield(task)

    def get_in_flight_count(self) -> int:
        return len(self._in_flight)

    async def _get_or_create(
        self,
        key: str,
        loader_async: AsyncLoader[TValue],
    ) -> asyncio.Task[TValue]:
        async with self._lock:
            existing = self._in_flight.get(key)

            if existing is not None:
                return existing

            task = asyncio.create_task(
                self._execute_and_release(
                    key,
                    loader_async,
                ),
                name=f"single-flight:{key}",
            )

            self._in_flight[key] = task
            task.add_done_callback(
                _observe_task_completion
            )

            return task

    async def _execute_and_release(
        self,
        key: str,
        loader_async: AsyncLoader[TValue],
    ) -> TValue:
        try:
            return await loader_async()
        finally:
            current_task = asyncio.current_task()

            async with self._lock:
                current = self._in_flight.get(key)

                if current is current_task:
                    self._in_flight.pop(key, None)


def _observe_task_completion(
    task: asyncio.Task[object],
) -> None:
    if task.cancelled():
        return

    task.exception()


def _validate_key(key: str) -> None:
    if key.strip() == "":
        raise ValueError(
            "Single-Flight key는 비어 있을 수 없습니다."
        )

use std::collections::HashMap;
use std::future::Future;
use std::hash::Hash;
use std::sync::atomic::{
    AtomicU64,
    Ordering,
};
use std::sync::Arc;

use tokio::sync::{
    Mutex,
    Notify,
    OnceCell,
};

#[derive(Clone, Debug)]
pub enum SingleFlightError<TError> {
    Loader(TError),
    LoaderTaskFailed(String),
}

struct Entry<TValue, TError> {
    id: u64,
    result: OnceCell<
        Result<TValue, SingleFlightError<TError>>
    >,
    notify: Notify,
}

impl<TValue, TError> Entry<TValue, TError> {
    fn new(id: u64) -> Self {
        Self {
            id,
            result: OnceCell::new(),
            notify: Notify::new(),
        }
    }
}

pub struct SingleFlight<TKey, TValue, TError> {
    in_flight:
        Mutex<HashMap<TKey, Arc<Entry<TValue, TError>>>>,

    next_id: AtomicU64,
}

impl<TKey, TValue, TError>
    SingleFlight<TKey, TValue, TError>
where
    TKey:
        Eq
        + Hash
        + Clone
        + Send
        + 'static,

    TValue:
        Clone
        + Send
        + Sync
        + 'static,

    TError:
        Clone
        + Send
        + Sync
        + 'static,
{
    pub fn new() -> Arc<Self> {
        Arc::new(Self {
            in_flight: Mutex::new(HashMap::new()),
            next_id: AtomicU64::new(1),
        })
    }

    pub async fn run<TLoader, TFuture>(
        self: &Arc<Self>,
        key: TKey,
        loader: TLoader,
    ) -> Result<TValue, SingleFlightError<TError>>
    where
        TLoader:
            FnOnce() -> TFuture
            + Send
            + 'static,

        TFuture:
            Future<Output = Result<TValue, TError>>
            + Send
            + 'static,
    {
        let (entry, is_leader) =
            self.get_or_create(&key).await;

        if is_leader {
            self.spawn_loader(
                key,
                Arc::clone(&entry),
                loader,
            );
        }

        wait_for_result(entry).await
    }

    pub async fn get_in_flight_count(&self) -> usize {
        self.in_flight.lock().await.len()
    }

    async fn get_or_create(
        &self,
        key: &TKey,
    ) -> (Arc<Entry<TValue, TError>>, bool) {
        let mut in_flight = self.in_flight.lock().await;

        if let Some(entry) = in_flight.get(key) {
            return (Arc::clone(entry), false);
        }

        let id = self.next_id.fetch_add(
            1,
            Ordering::Relaxed,
        );

        let entry = Arc::new(Entry::new(id));

        in_flight.insert(
            key.clone(),
            Arc::clone(&entry),
        );

        (entry, true)
    }

    fn spawn_loader<TLoader, TFuture>(
        self: &Arc<Self>,
        key: TKey,
        entry: Arc<Entry<TValue, TError>>,
        loader: TLoader,
    )
    where
        TLoader:
            FnOnce() -> TFuture
            + Send
            + 'static,

        TFuture:
            Future<Output = Result<TValue, TError>>
            + Send
            + 'static,
    {
        let owner = Arc::clone(self);

        tokio::spawn(async move
        {
            // loader를 한 단계 안쪽 task로 실행한다.
            // loader 내부 panic!이나 task abort는 JoinError가 된다.
            // 이를 terminal error로 공유해야 waiter가 영구 대기하지 않는다.
            let loader_task = tokio::spawn(async move {
                loader().await
            });

            let result = match loader_task.await {
                Ok(Ok(value)) => Ok(value),
                Ok(Err(error)) => {
                    Err(SingleFlightError::Loader(error))
                }
                Err(join_error) => {
                    Err(SingleFlightError::LoaderTaskFailed(
                        join_error.to_string(),
                    ))
                }
            };

            let _ = entry.result.set(result);
            entry.notify.notify_waiters();

            owner.remove_if_current(
                &key,
                entry.id,
            ).await;
        });
    }

    async fn remove_if_current(
        &self,
        key: &TKey,
        entry_id: u64,
    ) {
        let mut in_flight = self.in_flight.lock().await;

        let should_remove = in_flight
            .get(key)
            .is_some_and(|entry| entry.id == entry_id);

        if should_remove {
            in_flight.remove(key);
        }
    }
}

async fn wait_for_result<TValue, TError>(
    entry: Arc<Entry<TValue, TError>>,
) -> Result<TValue, SingleFlightError<TError>>
where
    TValue: Clone,
    TError: Clone,
{
    loop {
        if let Some(result) = entry.result.get() {
            return result.clone();
        }

        let notified = entry.notify.notified();

        if let Some(result) = entry.result.get() {
            return result.clone();
        }

        notified.await;
    }
}

[dependencies]
tokio = {
    version = "1",
    features = [
        "macros",
        "rt-multi-thread",
        "sync"
    ]
}

type Product = Readonly<{
  productId: string;
  displayName: string;
  priceCents: number;
}>;

type ProductCache = Readonly<{
  getAsync: (
    cacheKey: string,
    signal?: AbortSignal,
  ) => Promise<Product | undefined>;

  trySetAsync: (
    cacheKey: string,
    product: Product,
    ttlSeconds: number,
    signal?: AbortSignal,
  ) => Promise<boolean>;
}>;

type ProductRepository = Readonly<{
  getByIdAsync: (
    productId: string,
    signal?: AbortSignal,
  ) => Promise<Product>;
}>;

export class ProductQueryService
{
  readonly #cache: ProductCache;
  readonly #repository: ProductRepository;
  readonly #singleFlight: SingleFlight<Product>;
  readonly #cacheTtlSeconds: number;

  public constructor(
    cache: ProductCache,
    repository: ProductRepository,
    singleFlight: SingleFlight<Product>,
    cacheTtlSeconds: number,
  )
  {
    validateCacheTtl(cacheTtlSeconds);

    this.#cache = cache;
    this.#repository = repository;
    this.#singleFlight = singleFlight;
    this.#cacheTtlSeconds = cacheTtlSeconds;
  }

  public async getByIdAsync(
    productId: string,
    callerSignal?: AbortSignal,
  ): Promise<Product>
  {
    validateProductId(productId);
    const cacheKey = productCacheKey(productId);

    const cached = await this.#cache.getAsync(
      cacheKey,
      callerSignal,
    );

    if (cached !== undefined) {
      return cached;
    }

    return await this.#singleFlight.runAsync(
      cacheKey,
      (sharedSignal) =>
      {
        return this.#loadAndCacheAsync(
          cacheKey,
          productId,
          sharedSignal,
        );
      },
      callerSignal,
    );
  }

  async #loadAndCacheAsync(
    cacheKey: string,
    productId: string,
    sharedSignal?: AbortSignal,
  ): Promise<Product>
  {
    const cached = await this.#cache.getAsync(
      cacheKey,
      sharedSignal,
    );

    if (cached !== undefined) {
      return cached;
    }

    const product = await this.#repository.getByIdAsync(
      productId,
      sharedSignal,
    );

    // 원본 조회가 성공했다면 cache write 실패가
    // 전체 읽기 실패로 번지지 않게 best-effort로 처리한다.
    await this.#cache.trySetAsync(
      cacheKey,
      product,
      this.#cacheTtlSeconds,
      sharedSignal,
    );

    return product;
  }
}

function productCacheKey(productId: string): string
{
  return `product:detail:v1:${productId}`;
}

function validateProductId(productId: string): void
{
  if (!/^[A-Za-z0-9_-]{1,64}$/.test(productId)) {
    throw new RangeError(
      "productId 형식이 유효하지 않습니다.",
    );
  }
}

function validateCacheTtl(cacheTtlSeconds: number): void
{
  if (!Number.isSafeInteger(cacheTtlSeconds)
    || cacheTtlSeconds < 1
    || cacheTtlSeconds > 86_400) {
    throw new RangeError(
      "cache TTL은 1초 이상 86,400초 이하이어야 합니다.",
    );
  }
}

ProductCache
= 완료된 값을 TTL 동안 저장

ProductRepository
= 실제 DB·외부 API 조회 I/O

SingleFlight
= 동일 key의 진행 중 loader를 공유

ProductQueryService
= cache hit/miss와 loader 호출 순서 조립

Caller Signal
= 개별 요청의 대기 취소

Shared Shutdown Signal
= 프로세스 종료 시 공유 loader 취소

Cache TTL Policy
= 데이터 신선도와 만료 시점 결정

병합 key는 무엇인가
→ 같은 업무 요청이 정말 같은 key를 사용하는가
→ in-flight entry 등록이 loader 시작보다 먼저인가
→ 동일 key에서 loader가 하나만 실행되는가
→ 다른 key는 병렬 실행 가능한가
→ 완료·실패·취소 후 entry가 제거되는가
→ loader 실패가 모든 waiter에게 동일하게 전달되는가
→ 개별 waiter 취소가 공유 loader를 취소하지 않는가
→ loader 안에서 cache를 다시 확인하는가
→ 이 registry가 process-local인지 distributed인지 명확한가

Cache:
완료된 결과를 TTL 동안 재사용

Single-Flight:
현재 실행 중인 동일 작업만 공유
완료 후 entry 제거

나쁜 key:
productId

실제 결과가 다음에 따라 달라진다면:
- tenantId
- locale
- currency
- permission scope
- API version

권장 key:
tenantId + productId + locale + currency + policyVersion

서버 instance A:
product-42 loader 1개

서버 instance B:
product-42 loader 1개

전체 시스템:
동시에 loader 2개 실행 가능

모든 waiter가 취소됨
→ loader는 계속 실행될 수 있음

동일 key loader가 DB timeout
→ 모든 waiter가 같은 timeout을 받음
→ in-flight entry 제거
→ 다음 요청은 새 loader 실행 가능

loader 1회 실패
→ waiter 500명이 동시에 실패를 받음
→ 상위 계층이 모두 즉시 재시도
→ 다음 single-flight loader에 다시 500명이 몰림

- cache miss마다 loader를 새로 생성함
- in-flight entry 등록 전에 loader가 시작됨
- 완료 후 entry를 제거하지 않아 메모리가 증가함
- 실패한 Promise나 Task를 영구 보존함
- caller cancellation이 공유 loader까지 취소함
- tenant와 권한 정보를 key에서 제외함
- object reference를 key로 써서 의미상 같은 요청이 병합되지 않음
- process-local 병합을 distributed 보장으로 오해함
- loader 내부에서 cache를 다시 확인하지 않음
- Single-Flight가 TTL과 stale 정책까지 해결한다고 오해함
- loader timeout이 없어 entry와 waiter가 끝없이 남음
- 공격자 입력으로 in-flight key cardinality가 계속 증가함
- 공유된 mutable 결과를 waiter가 서로 수정함
- loader 실패 뒤 모든 waiter가 즉시 재시도함
- distributed lock이 결과 공유까지 해 준다고 가정함

async function getProductBadAsync(
  productId: string,
  cache: ProductCache,
  repository: ProductRepository,
  signal?: AbortSignal,
): Promise<Product>
{
  const cacheKey = productCacheKey(productId);

  const cached = await cache.getAsync(
    cacheKey,
    signal,
  );

  if (cached !== undefined) {
    return cached;
  }

  const product = await repository.getByIdAsync(
    productId,
    signal,
  );

  await cache.trySetAsync(
    cacheKey,
    product,
    300,
    signal,
  );

  return product;
}

요청 A cache miss
요청 B cache miss
요청 C cache miss

A DB 조회
B DB 조회
C DB 조회

- cache miss와 DB 조회 사이에 동시성 조정이 없다.
- 인기 key가 만료될 때 backend 요청이 순간적으로 폭증한다.
- cache를 채우는 동일 작업이 여러 번 실행된다.
- 모든 요청이 DB connection과 network slot을 각각 소비한다.
- 평상시 테스트에서는 문제를 발견하기 어렵다.

global lock:
product-42 조회 중
→ product-77 조회도 대기

import assert from "node:assert/strict";
import test from "node:test";

test("같은 key의 동시 요청은 loader를 한 번만 실행한다", async () =>
{
  const singleFlight = new SingleFlight<string>();
  let loaderCallCount = 0;

  const loaderAsync = async (): Promise<string> =>
  {
    loaderCallCount += 1;
    await delayAsync(20);

    return "product-42";
  };

  const results = await Promise.all(
    Array.from(
      { length: 50 },
      () => singleFlight.runAsync(
        "product-42",
        loaderAsync,
      ),
    ),
  );

  assert.equal(loaderCallCount, 1);

  assert.deepEqual(
    new Set(results),
    new Set(["product-42"]),
  );

  assert.equal(
    singleFlight.getInFlightCount(),
    0,
  );
});

test("서로 다른 key는 각각 loader를 실행한다", async () =>
{
  const singleFlight = new SingleFlight<string>();
  let loaderCallCount = 0;

  const loadAsync = async (
    value: string,
  ): Promise<string> =>
  {
    loaderCallCount += 1;
    await delayAsync(10);

    return value;
  };

  const [first, second] = await Promise.all([
    singleFlight.runAsync(
      "product-42",
      () => loadAsync("first"),
    ),
    singleFlight.runAsync(
      "product-77",
      () => loadAsync("second"),
    ),
  ]);

  assert.equal(loaderCallCount, 2);
  assert.equal(first, "first");
  assert.equal(second, "second");
});

test("실패한 entry는 제거되어 다음 요청이 재시도할 수 있다", async () =>
{
  const singleFlight = new SingleFlight<string>();
  let attemptCount = 0;

  await assert.rejects(() =>
  {
    return singleFlight.runAsync(
      "product-42",
      async () =>
      {
        attemptCount += 1;
        throw new Error("temporary failure");
      },
    );
  });

  const result = await singleFlight.runAsync(
    "product-42",
    async () =>
    {
      attemptCount += 1;
      return "recovered";
    },
  );

  assert.equal(attemptCount, 2);
  assert.equal(result, "recovered");
});

test("한 waiter의 취소는 공유 loader를 취소하지 않는다", async () =>
{
  const singleFlight = new SingleFlight<string>();
  const controller = new AbortController();

  let loaderCallCount = 0;

  const loaderAsync = async (): Promise<string> =>
  {
    loaderCallCount += 1;
    await delayAsync(30);

    return "shared-result";
  };

  const cancelledWaiter = singleFlight.runAsync(
    "product-42",
    loaderAsync,
    controller.signal,
  );

  const survivingWaiter = singleFlight.runAsync(
    "product-42",
    loaderAsync,
  );

  controller.abort(
    new Error("caller disconnected"),
  );

  await assert.rejects(
    () => cancelledWaiter,
  );

  assert.equal(
    await survivingWaiter,
    "shared-result",
  );

  assert.equal(loaderCallCount, 1);
});

function delayAsync(
  milliseconds: number,
): Promise<void>
{
  return new Promise((resolve) =>
  {
    setTimeout(resolve, milliseconds);
  });
}

single_flight.request.count
single_flight.loader.started.count
single_flight.waiter.shared.count
single_flight.loader.succeeded.count
single_flight.loader.failed.count
single_flight.in_flight
single_flight.waiter.duration_ms
single_flight.loader.duration_ms

requests per loader
= 전체 Single-Flight 요청 수 / 실제 loader 실행 수

예:
요청 1,000건
loader 50건

requests per loader = 20

loader suppression ratio
= 1 - (실제 loader 실행 수 / 전체 Single-Flight 요청 수)

loader suppression ratio = 95%

- cache hit ratio
- 동일 key 동시 waiter 수
- loader p95/p99 latency
- DB connection pool 사용량
- loader 실패율
- instance별 requests per loader
- in-flight key 수와 key 생성 거부 수
- key당 현재 waiter 수와 waiter 상한 도달 수

동일 key의 진행 중 작업은 하나의 완료 객체로 표현하고,
모든 waiter가 그 동일한 완료 객체를 기다리게 한다.

같은 cache key가 동시에 비었다면,
각자 backend로 내려가지 말고 하나가 읽고 나머지는 그 결과를 기다려라.