2nd/10_Wiki/Topics/Coding/CS_Backpressure_Deep.md

id, title, category, status, source_trust_level, verification_status, created_at, updated_at, tags, tech_stack, applied_in, aliases

id	title	category	status	source_trust_level	verification_status	created_at	updated_at	tags	tech_stack	applied_in	aliases
cs-backpressure-deep	Backpressure 깊이 — 큐 / Reactive / Token Bucket	Coding	draft	B	conceptual	2026-05-09	2026-05-09	cs backpressure queue vibe-coding	language applicable_to TS Backend		backpressure flow control drop bounded queue reactive streams

Backpressure Deep

빠른 producer + 느린 consumer = 메모리 폭발 / latency. Bounded queue + drop / block / shed. Reactive streams 가 표준 framework.

📖 핵심 개념

• 큐: producer-consumer buffer.
• Bounded: 한도 초과 시 drop / block.
• Shed: 우선순위 낮은 거 먼저 drop.
• Reactive Streams: Pull-based + demand signal.

💻 코드 패턴

Unbounded queue (위험)

// ❌ 메모리 폭발
const queue: Job[] = [];
producer.on('event', (job) => queue.push(job));
// consumer 가 느리면 queue 무한 자라남

Bounded — drop

class BoundedDropQueue<T> {
  private q: T[] = [];
  constructor(private max: number) {}
  
  push(item: T): boolean {
    if (this.q.length >= this.max) {
      // Drop oldest 또는 newest
      this.q.shift();  // drop oldest
    }
    this.q.push(item);
    return true;
  }
}

Bounded — block

class BoundedBlockingQueue<T> {
  private q: T[] = [];
  private waiters: ((v: T) => void)[] = [];
  
  constructor(private max: number) {}
  
  async push(item: T): Promise<void> {
    while (this.q.length >= this.max) {
      await new Promise(r => setTimeout(r, 10));  // backoff
    }
    if (this.waiters.length > 0) {
      this.waiters.shift()!(item);
    } else {
      this.q.push(item);
    }
  }
  
  async pop(): Promise<T> {
    if (this.q.length > 0) return this.q.shift()!;
    return new Promise<T>(r => this.waiters.push(r));
  }
}

Backoff producer (자체 throttle)

async function producerLoop() {
  while (true) {
    if (queue.length > THRESHOLD) {
      const wait = Math.min(100 * (queue.length / THRESHOLD), 5000);
      await sleep(wait);
      continue;
    }
    const job = await fetchJob();
    queue.push(job);
  }
}

Reactive Streams (Pull-based demand)

// RxJS / Most.js
import { from, interval } from 'rxjs';
import { mergeMap, map } from 'rxjs/operators';

interval(10)  // producer
  .pipe(
    mergeMap(i => slowAsync(i), 5),  // 동시 5개만 — backpressure
    map(x => x * 2),
  )
  .subscribe(console.log);

→ mergeMap concurrency = backpressure.

Node Streams (자동)

import { pipeline } from 'node:stream/promises';

await pipeline(
  source,
  transform,
  destination,
);
// 각 stream 의 highWaterMark 가 backpressure
// readable 가 빠르고 writable 가 느리면 — readable pause

const writable = new Writable({
  highWaterMark: 16384,  // 16KB buffer
  write(chunk, _, cb) {
    slowProcess(chunk).then(() => cb());
  },
});

source.pipe(writable);
// pipe 가 자동 backpressure

Drop priority (load shedding)

class PriorityDropQueue {
  private q: Job[] = [];
  
  push(job: Job) {
    if (this.q.length >= MAX) {
      // 낮은 priority drop
      const lowest = this.q.findIndex(j => j.priority < job.priority);
      if (lowest >= 0) {
        this.q.splice(lowest, 1);
        this.q.push(job);
      }
      // Job 의 priority 도 낮으면 — drop new job
    } else {
      this.q.push(job);
    }
    this.q.sort((a, b) => b.priority - a.priority);
  }
}

→ 결제 = high, log = low. Overload 시 log drop.

Adaptive concurrency

// 응답 시간 따라 동시 처리 수 조정
class AdaptiveExecutor {
  private concurrency = 10;
  private latencies: number[] = [];
  
  async execute(task: () => Promise<void>) {
    while (this.active >= this.concurrency) await sleep(10);
    
    const t = Date.now();
    await task();
    const ms = Date.now() - t;
    
    this.latencies.push(ms);
    if (this.latencies.length > 100) this.latencies.shift();
    
    const p95 = percentile(this.latencies, 0.95);
    if (p95 > 500 && this.concurrency > 1) this.concurrency--;
    if (p95 < 100) this.concurrency++;
  }
}

→ Latency 가 늘면 concurrency 줄이기.

Token bucket (위 rate limit 문서)

class TokenBucket {
  private tokens: number;
  
  consume(n = 1): boolean {
    this.refill();
    if (this.tokens < n) return false;
    this.tokens -= n;
    return true;
  }
}

// Producer
if (bucket.consume()) {
  await emit(event);
} else {
  drop();
}

Circuit breaker + backpressure

// 다운스트림 fail 시 circuit open → 더 이상 보내지 X
if (circuit.isOpen()) {
  drop();
  return;
}

→ Backend_Circuit_Breaker.

Kafka — partition + lag

Producer 가 빠르면 Kafka partition 의 disk usage 증가.
Consumer lag 모니터링 + scale up consumer.

-- Lag 모니터링
SELECT topic, partition, current_offset, log_end_offset, lag
FROM kafka_consumer_groups WHERE group_id = 'my-group';

Bounded memory 안 큐 정책

1. Drop newest (FIFO loss)
2. Drop oldest (LIFO loss)
3. Drop priority-based
4. Block producer
5. Spill to disk (overflow)
6. Reject (return error)

→ Use case 별 다름.

Feedback loop (사용자에 알리기)

app.post('/api/job', async (req, res) => {
  if (queue.length > MAX) {
    return res.status(503).set('Retry-After', '30').json({
      error: 'Service overloaded, please retry',
    });
  }
  await queue.push(req.body);
  res.status(202).json({ status: 'queued' });
});

→ Client 가 알고 backoff.

Server health-based load shed

app.use((req, res, next) => {
  const cpu = await getCPU();
  if (cpu > 90 && Math.random() < 0.1) {
    return res.status(503).end();
  }
  next();
});

→ 10% 확률 drop — 급격한 cliff 방지.

Queue depth limit (별 service)

Worker 수 N + queue depth M 면 — N + M 가 max in-flight.
M 너무 큼 = latency 증가 (큐 안 오래 wait).
M 너무 적음 = drop / reject 자주.

Little's Law: L = λ × W
- L = 평균 큐 안 작업
- λ = 도착률
- W = 평균 처리 시간

→ p99 latency 목표 고려해서 M 결정.

Async iterator + bounded

async function* boundedProducer<T>(source: AsyncIterable<T>, limit: number) {
  let inflight = 0;
  for await (const item of source) {
    while (inflight >= limit) await sleep(10);
    inflight++;
    yield Promise.resolve(item).finally(() => inflight--);
  }
}

🤔 의사결정 기준

상황	정책
Latency critical	Bounded + drop
Strong consistency	Bounded + block
Bursty traffic	Token bucket
Different priority	Priority drop
분산	Kafka + lag monitoring
Reactive	RxJS / Reactive Streams

❌ 안티패턴

• Unbounded queue: OOM.
• Blocking 무 timeout: deadlock 가능.
• Drop 정책 없음: 무엇이 잃었는지 모름.
• Producer 만 throttle — consumer 안 scale: 큐 늘어남.
• Latency 모니터링 X: 점진적 죽음.
• Block producer + caller blocking: 상위 시스템 다운.
• fast retry after drop: thundering herd.

🤖 LLM 활용 힌트

• Bounded queue + drop / block 정책 명시.
• Circuit breaker + 503 + Retry-After.
• 큰 처리량 = Kafka + lag monitor.
• Adaptive concurrency 가 modern.

🔗 관련 문서

• Backpressure_Patterns
• Backend_Circuit_Breaker
• Backend_Rate_Limiting