ในฐานะ Senior Backend Engineer ที่เคยสร้างระบบ AI customer service ขนาดใหญ่สำหรับแพลตฟอร์ม e-commerce ที่มี TPS สูงสุดถึง 5,000 requests/second ผมเจอปัญหา latency ที่ทำให้ผู้ใช้รู้สึกว่าระบบ "ค้าง" โดยเฉพาะเมื่อ streaming response มีข้อมูลขนาดใหญ่ ในบทความนี้ผมจะแชร์ patterns ที่ใช้จริงใน production พร้อมโค้ดที่รันได้ทันทีผ่าน HolySheep AI ซึ่งให้บริการ Claude Sonnet 4.5 streaming ด้วยความหน่วงต่ำกว่า 50ms

ทำไมต้อง Streaming และทำไมมันถึงยาก

Streaming ช่วยให้ผู้ใช้เห็นผลลัพธ์เร็วขึ้น 50-70% ในแง่ perceived latency แต่ปัญหาจริงคือ buffer management, memory leak จาก SSE connections ที่ไม่ปิด, และ backpressure ที่ทำให้ server ล่มเมื่อมี concurrent connections สูงๆ

Pattern 1: Streaming Handler พื้นฐาน

เริ่มจาก implementation ที่ง่ายที่สุดแต่ใช้งานได้จริง สำหรับ use case การสร้าง AI chat ของ e-commerce platform

// streaming-handler.ts - Basic streaming pattern with HolySheep Claude
import { EventEmitter } from 'events';

interface StreamConfig {
  apiKey: string;
  baseUrl: string;
  model: string;
  maxTokens: number;
  temperature: number;
}

interface StreamChunk {
  type: 'content' | 'error' | 'done';
  content?: string;
  delta?: string;
  error?: string;
}

class ClaudeStreamHandler extends EventEmitter {
  private config: StreamConfig;
  private abortController: AbortController | null = null;
  private connectionCount = 0;

  constructor(config: StreamConfig) {
    super();
    this.config = {
      baseUrl: 'https://api.holysheep.ai/v1',
      model: 'claude-sonnet-4.5',
      maxTokens: 4096,
      temperature: 0.7,
      ...config,
    };
  }

  async streamChat(
    messages: Array<{ role: string; content: string }>,
    onChunk: (chunk: StreamChunk) => void
  ): Promise<string> {
    this.abortController = new AbortController();
    this.connectionCount++;

    const startTime = Date.now();
    let fullResponse = '';

    try {
      const response = await fetch(${this.config.baseUrl}/chat/completions, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': Bearer ${this.config.apiKey},
        },
        body: JSON.stringify({
          model: this.config.model,
          messages,
          max_tokens: this.config.maxTokens,
          temperature: this.config.temperature,
          stream: true,
        }),
        signal: this.abortController.signal,
      });

      if (!response.ok) {
        throw new Error(HTTP ${response.status}: ${response.statusText});
      }

      const reader = response.body?.getReader();
      if (!reader) throw new Error('No response body');

      const decoder = new TextDecoder();
      let buffer = '';

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\n');
        buffer = lines.pop() || '';

        for (const line of lines) {
          if (!line.startsWith('data: ')) continue;
          
          const data = line.slice(6);
          if (data === '[DONE]') continue;

          try {
            const parsed = JSON.parse(data);
            const content = parsed.choices?.[0]?.delta?.content;
            
            if (content) {
              fullResponse += content;
              onChunk({
                type: 'content',
                content,
                delta: content,
              });
            }
          } catch (e) {
            // Skip malformed JSON
          }
        }
      }

      const elapsed = Date.now() - startTime;
      console.log(Stream completed in ${elapsed}ms, ${fullResponse.length} chars);
      
      onChunk({ type: 'done' });
      return fullResponse;

    } catch (error: any) {
      if (error.name === 'AbortError') {
        onChunk({ type: 'error', error: 'Request cancelled' });
      } else {
        onChunk({ type: 'error', error: error.message });
      }
      throw error;
    } finally {
      this.connectionCount--;
    }
  }

  cancel(): void {
    this.abortController?.abort();
  }

  getConnectionCount(): number {
    return this.connectionCount;
  }
}

// Usage Example
async function main() {
  const handler = new ClaudeStreamHandler({
    apiKey: 'YOUR_HOLYSHEEP_API_KEY',
  });

  const messages = [
    { role: 'system', content: 'คุณเป็นผู้ช่วยแนะนำสินค้าออนไลน์' },
    { role: 'user', content: 'แนะนำหูฟังไร้สายราคาดีให้ทำงาน Home Office' },
  ];

  process.stdout.write('AI: ');

  await handler.streamChat(messages, (chunk) => {
    if (chunk.type === 'content' && chunk.delta) {
      process.stdout.write(chunk.delta);
    } else if (chunk.type === 'done') {
      console.log('\n--- Stream Complete ---');
    }
  });
}

main().catch(console.error);

Pattern 2: Backpressure และ Connection Pool Management

ปัญหาที่พบบ่อยใน production คือ memory leak เมื่อมี thousands of concurrent streaming connections ดังนั้นต้อง implement backpressure และ connection limiting อย่างเคร่งครัด

// backpressure-manager.ts - Connection management with backpressure
import { EventEmitter } from 'events';

interface BackpressureConfig {
  maxConcurrentStreams: number;
  maxQueueSize: number;
  queueTimeoutMs: number;
  bufferHighWatermark: number;
  bufferLowWatermark: number;
}

interface QueuedRequest {
  id: string;
  messages: any[];
  resolve: (value: string) => void;
  reject: (error: Error) => void;
  createdAt: number;
}

class BackpressureManager {
  private config: BackpressureConfig;
  private activeStreams = 0;
  private requestQueue: QueuedRequest[] = [];
  private bufferUsage = new Map<string, number>();
  private streamHandler: any;

  constructor(config: BackpressureConfig, streamHandler: any) {
    this.config = config;
    this.streamHandler = streamHandler;
  }

  async streamWithBackpressure(
    requestId: string,
    messages: any[],
    onProgress: (text: string) => void
  ): Promise<string> {
    return new Promise((resolve, reject) => {
      const queuedRequest: QueuedRequest = {
        id: requestId,
        messages,
        resolve,
        reject,
        createdAt: Date.now(),
      };

      this.requestQueue.push(queuedRequest);
      this.processQueue();

      // Queue timeout
      setTimeout(() => {
        const index = this.requestQueue.findIndex(r => r.id === requestId);
        if (index !== -1) {
          this.requestQueue.splice(index, 1);
          reject(new Error('Request timeout in queue'));
        }
      }, this.config.queueTimeoutMs);
    });
  }

  private async processQueue(): Promise<void> {
    if (this.activeStreams >= this.config.maxConcurrentStreams) {
      return; // Wait for active streams to complete
    }

    if (this.requestQueue.length === 0) {
      return;
    }

    const request = this.requestQueue.shift()!;
    this.activeStreams++;

    try {
      const result = await this.executeStream(request);
      request.resolve(result);
    } catch (error) {
      request.reject(error as Error);
    } finally {
      this.activeStreams--;
      this.processQueue(); // Process next in queue
    }
  }

  private async executeStream(request: QueuedRequest): Promise<string> {
    let fullResponse = '';

    await this.streamHandler.streamChat(request.messages, (chunk: any) => {
      if (chunk.type === 'content') {
        // Check buffer pressure
        const currentBuffer = this.bufferUsage.get(request.id) || 0;
        
        if (currentBuffer >= this.config.bufferHighWatermark) {
          // Apply backpressure: slow down processing
          this.streamHandler.cancel();
          throw new Error('Buffer overflow: backpressure triggered');
        }

        fullResponse += chunk.delta || '';
        this.bufferUsage.set(request.id, fullResponse.length);
        
        onProgress(chunk.delta || '');
      }
    });

    this.bufferUsage.delete(request.id);
    return fullResponse;
  }

  getStats() {
    return {
      activeStreams: this.activeStreams,
      queuedRequests: this.requestQueue.length,
      bufferUsage: Object.fromEntries(this.bufferUsage),
      backpressureActive: this.activeStreams >= this.config.maxConcurrentStreams,
    };
  }
}

// Example: Production config for e-commerce with 5000 TPS target
const config: BackpressureConfig = {
  maxConcurrentStreams: 1000, // Limit concurrent streams
  maxQueueSize: 5000,         // Max requests waiting
  queueTimeoutMs: 30000,       // 30s max wait time
  bufferHighWatermark: 10000,  // 10KB per stream
  bufferLowWatermark: 5000,    // Resume at 5KB
};

const backpressure = new BackpressureManager(config, null);

// Simulate load test
async function loadTest() {
  console.log('Starting load test...');
  
  const startTime = Date.now();
  const totalRequests = 100;
  let completed = 0;

  for (let i = 0; i < totalRequests; i++) {
    const requestId = req_${i};
    
    backpressure.streamWithBackpressure(
      requestId,
      [{ role: 'user', content: Query ${i} }],
      (text) => {} // Progress callback
    ).then(() => {
      completed++;
      if (completed % 10 === 0) {
        console.log(Progress: ${completed}/${totalRequests});
        console.log('Stats:', backpressure.getStats());
      }
    }).catch(console.error);
  }

  // Wait for completion
  await new Promise(resolve => setTimeout(resolve, 5000));
  const elapsed = Date.now() - startTime;
  
  console.log(Load test completed in ${elapsed}ms);
  console.log('Final stats:', backpressure.getStats());
}

Pattern 3: SSE Reconnection อัตโนมัติสำหรับ Enterprise RAG

ในระบบ RAG ขององค์กร การ streaming ข้อมูลจาก document retrieval ต้องมีความทนทานต่อ network disruption ผมใช้ exponential backoff กับ circuit breaker pattern

// sse-reconnection.ts - Auto-reconnect with circuit breaker for RAG
interface RetryConfig {
  maxRetries: number;
  baseDelayMs: number;
  maxDelayMs: number;
  backoffMultiplier: number;
}

interface CircuitBreakerState {
  failures: number;
  lastFailure: number;
  state: 'closed' | 'open' | 'half-open';
}

class SseReconnectionHandler {
  private retryConfig: RetryConfig;
  private circuitBreaker: CircuitBreakerState = {
    failures: 0,
    lastFailure: 0,
    state: 'closed',
  };
  private readonly CIRCUIT_THRESHOLD = 5;
  private readonly CIRCUIT_RESET_MS = 60000;

  constructor(retryConfig: Partial<RetryConfig> = {}) {
    this.retryConfig = {
      maxRetries: 5,
      baseDelayMs: 1000,
      maxDelayMs: 30000,
      backoffMultiplier: 2,
      ...retryConfig,
    };
  }

  async streamWithRetry(
    baseUrl: string,
    apiKey: string,
    payload: any,
    onChunk: (data: any) => void,
    signal?: AbortSignal
  ): Promise<void> {
    let lastError: Error | null = null;

    for (let attempt = 0; attempt <= this.retryConfig.maxRetries; attempt++) {
      // Check circuit breaker
      if (this.circuitBreaker.state === 'open') {
        const timeSinceFailure = Date.now() - this.circuitBreaker.lastFailure;
        if (timeSinceFailure < this.CIRCUIT_RESET_MS) {
          throw new Error(Circuit breaker open. Retry after ${this.CIRCUIT_RESET_MS - timeSinceFailure}ms);
        }
        this.circuitBreaker.state = 'half-open';
      }

      try {
        await this.executeStream(baseUrl, apiKey, payload, onChunk, signal);
        this.onSuccess();
        return;
      } catch (error: any) {
        lastError = error;
        this.onFailure();

        if (signal?.aborted) {
          throw new Error('Request cancelled');
        }

        if (attempt < this.retryConfig.maxRetries) {
          const delay = this.calculateBackoff(attempt);
          console.log(Attempt ${attempt + 1} failed, retrying in ${delay}ms...);
          await this.sleep(delay);
        }
      }
    }

    throw lastError || new Error('Max retries exceeded');
  }

  private async executeStream(
    baseUrl: string,
    apiKey: string,
    payload: any,
    onChunk: (data: any) => void,
    signal?: AbortSignal
  ): Promise<void> {
    const response = await fetch(${baseUrl}/chat/completions, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': Bearer ${apiKey},
      },
      body: JSON.stringify({
        model: 'claude-sonnet-4.5',
        ...payload,
        stream: true,
      }),
      signal,
    });

    if (!response.ok) {
      throw new Error(HTTP Error: ${response.status});
    }

    const reader = response.body?.getReader();
    if (!reader) throw new Error('No response body');

    const decoder = new TextDecoder();

    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        const text = decoder.decode(value, { stream: true });
        const lines = text.split('\n').filter(line => line.trim());

        for (const line of lines) {
          if (line.startsWith('data: ')) {
            const data = line.slice(6);
            if (data !== '[DONE]') {
              try {
                const parsed = JSON.parse(data);
                onChunk(parsed);
              } catch (e) {
                // Skip malformed data
              }
            }
          }
        }
      }
    } finally {
      reader.releaseLock();
    }
  }

  private calculateBackoff(attempt: number): number {
    const delay = this.retryConfig.baseDelayMs * 
      Math.pow(this.retryConfig.backoffMultiplier, attempt);
    return Math.min(delay, this.retryConfig.maxDelayMs);
  }

  private onSuccess(): void {
    this.circuitBreaker.failures = 0;
    this.circuitBreaker.state = 'closed';
  }

  private onFailure(): void {
    this.circuitBreaker.failures++;
    this.circuitBreaker.lastFailure = Date.now();

    if (this.circuitBreaker.failures >= this.CIRCUIT_THRESHOLD) {
      this.circuitBreaker.state = 'open';
      console.warn(Circuit breaker opened after ${this.circuitBreaker.failures} failures);
    }
  }

  private sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  getCircuitState(): string {
    return this.circuitBreaker.state;
  }
}

// Example: Enterprise RAG streaming with document chunks
async function streamRAGResponse(
  query: string,
  documentIds: string[],
  onProgress: (chunk: string, source: string) => void
) {
  const handler = new SseReconnectionHandler({
    maxRetries: 5,
    baseDelayMs: 1000,
  });

  const payload = {
    messages: [
      {
        role: 'system',
        content: คุณเป็น AI assistant สำหรับองค์กร ตอบคำถามจากเอกสารที่ให้มา
      },
      {
        role: 'user',
        content: Query: ${query}\nDocument IDs: ${documentIds.join(', ')}
      }
    ],
    max_tokens: 8192,
    temperature: 0.3,
  };

  await handler.streamWithRetry(
    'https://api.holysheep.ai/v1',
    'YOUR_HOLYSHEEP_API_KEY',
    payload,
    (data) => {
      const content = data.choices?.[0]?.delta?.content;
      if (content) {
        const source = data.choices?.[0]?.delta?.source || 'unknown';
        onProgress(content, source);
      }
    }
  );

  console.log('RAG stream completed. Circuit state:', handler.getCircuitState());
}

Performance Benchmark: HolySheep vs OpenAI

จากการทดสอบจริงบน production workload ด้วย 1,000 concurrent streaming connections:

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. Error: "No response body" หรือ Response is null

สาเหตุ: ปัญหานี้เกิดจากการที่ streaming request ถูก cancel ก่อนที่ response จะมาถึง หรือ AbortController ถูก abort ก่อนเวลาอันควร นอกจากนี้ยังอาจเกิดจาก CORS policy หรือ network timeout ฝั่ง server

// ❌ Wrong: Race condition on abort
async function streamChat(messages: any[]) {
  const controller = new AbortController();
  
  // Problem: setTimeout might fire before fetch completes
  setTimeout(() => controller.abort(), 1000);
  
  const response = await fetch(url, {
    signal: controller.signal
  });
  
  return response.json(); // Response might be null!
}

// ✅ Correct: Proper error handling with null check
async function streamChat(messages: any[], timeoutMs = 30000) {
  const controller = new AbortController();
  
  // Set timeout AFTER fetch starts
  const timeoutId = setTimeout(() => {
    controller.abort();
    console.log('Request timed out');
  }, timeoutMs);

  try {
    const response = await fetch(url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ model: 'claude-sonnet-4.5', messages, stream: true }),
      signal: controller.signal,
    });

    clearTimeout(timeoutId);

    // CRITICAL: Always check response before accessing body
    if (!response || !response.ok) {
      const errorText = await response?.text() || 'Unknown error';
      throw new Error(HTTP ${response?.status}: ${errorText});
    }

    if (!response.body) {
      throw new Error('Server returned empty response body');
    }

    return response;

  } catch (error: any) {
    clearTimeout(timeoutId);
    
    if (error.name === 'AbortError') {
      console.log('Request was cancelled by client');
    }
    throw error;
  }
}

2. Memory Leak: Streaming connections ไม่ถูกปิด

สาเหตุ: Reader ถูกสร้างแต่ไม่ถูก release lock เมื่อเกิด error หรือเมื่อ client disconnect ทำให้ event loop ถูก block และ memory เพิ่มขึ้นเรื่อยๆ ปัญหานี้เห็นชัดเจนเมื่อมี thousands of concurrent users

// ❌ Wrong: Memory leak from unclosed reader
async function streamWithLeak(url: string) {
  const response = await fetch(url);
  const reader = response.body!.getReader();
  
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      process.stdout.write(value);
    }
  } catch (error) {
    console.error('Stream error:', error);
    // Reader never released!
  }
  // If error occurs, reader is stuck forever
}

// ✅ Correct: Guaranteed cleanup with try-finally
async function streamWithoutLeak(url: string, onData: (chunk: string) => void) {
  const response = await fetch(url);
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let streamClosed = false;

  const cleanup = () => {
    if (!streamClosed) {
      streamClosed = true;
      reader.releaseLock(); // CRITICAL: Always release lock
      console.log('Stream resources cleaned up');
    }
  };

  try {
    while (true) {
      const { done, value } = await reader.read();
      
      if (done) {
        cleanup();
        break;
      }

      const text = decoder.decode(value, { stream: true });
      onData(text);
    }
  } catch (error) {
    console.error('Stream error:', error);
    cleanup();
    throw error;
  } finally {
    // Double guarantee cleanup
    cleanup();
  }
}

// ✅ Even better: Use AbortController for proper cancellation
async function managedStream(
  url: string,
  options: { signal?: AbortSignal; timeoutMs?: number }
) {
  const controller = new AbortController();
  const timeoutId = options.timeoutMs 
    ? setTimeout(() => controller.abort(), options.timeoutMs)
    : null;

  // Merge external signal with internal
  const mergedSignal = options.signal
    ? anySignalToAbortSignal([options.signal, controller.signal])
    : controller.signal;

  try {
    const response = await fetch(url, { signal: mergedSignal });
    const reader = response.body!.getReader();
    
    return new ReadableStream({
      async start(controller) {
        try {
          while (true) {
            const { done, value } = await reader.read();
            if (done) break;
            controller.enqueue(value);
          }
          controller.close();
        } catch (error) {
          controller.error(error);
        } finally {
          reader.releaseLock();
        }
      },
      cancel() {
        reader.cancel();
        reader.releaseLock();
      }
    });
  } finally {
    if (timeoutId) clearTimeout(timeoutId);
  }
}

3. Backpressure Overflow: Server overwhelmed by slow clients

สาเหตุ: เมื่อ client รับข้อมูลช้ากว่าที่ server ส่งมา (เช่น slow network, slow client processing) buffer ฝั่ง server จะเต็มทำให้ TCP window ปิด และ server ต้อง buffer ทั้งหมดใน memory ส่งผลให้ memory พุ่งและอาจ crash

// ❌ Wrong: No backpressure handling
async function naiveStream(url: string, onData: (data: any) => void) {
  const response = await fetch(url);
  const reader = response.body!.getReader();
  
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    
    // No flow control - process as fast as possible
    onData(value);
  }
}

// ✅ Correct: Backpressure with ReadableStream default controller
function createBackpressureStream(
  fetchPromise: Promise<Response>,
  onChunk: (data: string) => void
) {
  return new ReadableStream({
    async start(controller) {
      const response = await fetchPromise;
      const reader = response.body!.getReader();
      const decoder = new TextDecoder();
      let totalProcessed = 0;
      const HIGH_WATERMARK = 64 * 1024; // 64KB

      const processNext = async () => {
        const { done, value } = await reader.read();
        
        if (done) {
          controller.close();
          return;
        }

        // Apply backpressure by checking desired size
        // If controller desires less than our chunk, wait
        while (controller.desiredSize !== null && controller.desiredSize <= 0) {
          // Wait for consumer to catch up
          await waitForDesiredSize(controller);
        }

        const text = decoder.decode(value, { stream: true });
        totalProcessed += value.byteLength;
        
        // Process chunk (can be async)
        await onChunk(text);
        
        // Enqueue to readable stream (applies backpressure automatically)
        controller.enqueue(value);
        
        // Continue reading
        processNext();
      };

      processNext().catch(error => {
        controller.error(error);
        reader.releaseLock();
      });
    },

    cancel() {
      // Handle cancellation
      console.log('Stream cancelled by consumer');
    }
  });
}

// Helper function for backpressure wait
function waitForDesiredSize(controller: ReadableStreamDefaultController): Promise<void> {
  return new Promise(resolve => {
    const checkSize = () => {
      if (controller.desiredSize !== null && controller.desiredSize > 0) {
        resolve();
      } else {
        setTimeout(checkSize, 10);
      }
    };
    checkSize();
  });
}

// ✅ Production example with explicit buffer management
class StreamingBuffer {
  private buffer: string[] = [];
  private highWatermark: number;
  private lowWatermark: number;
  private isPaused = false;
  private pauseResolve: (() => void) | null = null;

  constructor(highWatermark = 100, lowWatermark = 20) {
    this.highWatermark = highWatermark;
    this.lowWatermark = lowWatermark;
  }

  push(item: string): boolean {
    this.buffer.push(item);
    
    if (this.buffer.length >= this.highWatermark && !this.isPaused) {
      this.isPaused = true;
      // Signal upstream to pause
      return false; // Tell producer to pause
    }
    return true;
  }

  async shift(): Promise<string | null> {
    if (this.buffer.length <= this.lowWatermark && this.isPaused) {
      this.isPaused = false;
      this.pauseResolve?.();
    }

    return this.buffer.shift() || null;
  }

  get length(): number {
    return this.buffer.length;
  }

  whenPaused(): Promise<void> {
    if (!this.isPaused) return Promise.resolve();
    
    return new Promise(resolve => {
      this.pauseResolve = resolve;
    });
  }
}

// Usage with backpressure
async function streamWithBuffer(
  url: string,
  onData: (data: string) => void
) {
  const buffer = new StreamingBuffer(50, 10);
  let upstreamPaused = false;

  const response = await fetch(url);
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();

  // Producer loop
  const producer = async () => {
    try {
      while (true) {
        const { done, value } = await reader.read();
        
        if (done) {
          // Flush remaining buffer
          while (buffer.length > 0) {
            const item = await buffer.shift();
            if (item) onData(item);
          }
          break;
        }

        const text = decoder.decode(value, { stream: true });
        
        // Try to push to buffer
        const canContinue = buffer.push(text);
        
        if (!canContinue) {
          upstreamPaused = true;
          // Wait for buffer to drain
          await buffer.whenPaused();
          upstreamPaused = false;
        }
      }
    } finally {
      reader.releaseLock();
    }
  };

  // Consumer loop
  const consumer = async () => {
    try {
      while (true) {
        const item = await buffer.shift();
        if (item === null) {
          if (upstreamPaused) continue;
          break;
        }
        onData(item);
      }
    } catch (error) {
      console.error('Consumer error:', error);
    }
  };

  await Promise.all([producer(), consumer()]);
}

สรุปและ Best Practices