Go로 PostgreSQL 프록시 만들기 (22) - COPY 프로토콜 교착과 Map 메모리 누수

일반 쿼리:
  Client → Query('Q')
  Server → RowDescription + DataRow... + CommandComplete + ReadyForQuery('Z')

COPY FROM STDIN:
  Client → Query('Q')  "COPY my_table FROM STDIN"
  Server → CopyInResponse('G')     ← 여기서 방향이 전환됨!
  Client → CopyData('d') × N       ← 클라이언트가 데이터를 보냄
  Client → CopyDone('c')
  Server → CommandComplete + ReadyForQuery('Z')

// AS-IS: 백엔드 응답만 읽으며 ReadyForQuery를 기다림
for {
    msg, _ := protocol.ReadMessage(backendConn)  // ← 여기서 영원히 블로킹!
    protocol.WriteMessage(clientConn, msg.Type, msg.Payload)
    if msg.Type == protocol.MsgReadyForQuery {
        return nil
    }
}

switch msg.Type {
case protocol.MsgReadyForQuery:
    return nil
case protocol.MsgCopyInResponse:
    // 클라이언트→백엔드 방향으로 CopyData/CopyDone 릴레이
    if err := s.relayCopyIn(clientConn, backendConn); err != nil {
        return fmt.Errorf("copy in relay: %w", err)
    }
case protocol.MsgCopyOutResponse:
    // 백엔드→클라이언트 방향으로 CopyData/CopyDone 릴레이
    if err := s.relayCopyOut(clientConn, backendConn); err != nil {
        return fmt.Errorf("copy out relay: %w", err)
    }
case protocol.MsgCopyBothResponse:
    // 양방향 릴레이 (스트리밍 복제)
    if err := s.relayCopyBoth(clientConn, backendConn); err != nil {
        return fmt.Errorf("copy both relay: %w", err)
    }
}

func (s *Server) relayCopyIn(clientConn, backendConn net.Conn) error {
    for {
        msg, err := protocol.ReadMessage(clientConn)
        if err != nil {
            return fmt.Errorf("read client copy data: %w", err)
        }
        if err := protocol.WriteMessage(backendConn, msg.Type, msg.Payload); err != nil {
            return fmt.Errorf("forward copy data to backend: %w", err)
        }
        switch msg.Type {
        case protocol.MsgCopyDone, protocol.MsgCopyFail:
            return nil  // COPY 완료, 정상 릴레이 루프로 복귀
        }
    }
}

case protocol.MsgCopyInResponse:
    if err := s.relayCopyIn(clientConn, backendConn); err != nil {
        return nil, fmt.Errorf("copy in relay: %w", err)
    }
    buf = nil        // COPY 결과는 캐시 불가
    oversize = true

func TestRelayUntilReady_CopyIn(t *testing.T) {
    // 프록시 ↔ 클라이언트, 프록시 ↔ 백엔드 파이프
    proxyClientConn, testClientConn := net.Pipe()
    proxyBackendConn, testBackendConn := net.Pipe()

    go func() {
        errCh <- srv.relayUntilReady(proxyClientConn, proxyBackendConn)
    }()

    // 백엔드 시뮬레이션: CopyInResponse → CopyData 수신 → 완료
    go func() {
        protocol.WriteMessage(testBackendConn, 'G', copyInPayload)
        // CopyData/CopyDone 수신 후 CommandComplete + ReadyForQuery 전송
        ...
    }()

    // 클라이언트 시뮬레이션: CopyInResponse 수신 → CopyData 전송 → 완료 수신
    msg, _ := protocol.ReadMessage(testClientConn)
    // msg.Type == 'G' 확인
    protocol.WriteMessage(testClientConn, 'd', data)
    protocol.WriteMessage(testClientConn, 'c', nil)
    // CommandComplete + ReadyForQuery 수신 확인

    select {
    case err := <-errCh:
        // 교착 없이 정상 종료!
    case <-time.After(2 * time.Second):
        t.Fatal("deadlocked!")
    }
}

func (l *Logger) sendWebhook(e Event) {
    queryKey := truncateQuery(e.Query, 100)
    l.lastWebhookMu.Lock()
    if last, ok := l.lastWebhook[queryKey]; ok && time.Since(last) < l.webhookInterval {
        l.lastWebhookMu.Unlock()
        return  // 중복, 건너뜀
    }
    l.lastWebhook[queryKey] = time.Now()  // ← 추가만 하고 삭제 안 함!
    l.lastWebhookMu.Unlock()
    // ... HTTP POST
}

func (l *Logger) cleanupWebhookDedup() {
    defer l.wg.Done()
    ticker := time.NewTicker(l.webhookInterval)
    defer ticker.Stop()

    for {
        select {
        case <-l.stopCh:
            return
        case <-ticker.C:
            l.lastWebhookMu.Lock()
            now := time.Now()
            for key, last := range l.lastWebhook {
                if now.Sub(last) >= l.webhookInterval {
                    delete(l.lastWebhook, key)
                }
            }
            l.lastWebhookMu.Unlock()
        }
    }
}

type WebhookConfig struct {
    Enabled       bool
    URL           string
    Timeout       time.Duration
    DedupInterval time.Duration  // 기본 1분
}

func TestAuditLogger_WebhookDedupCleanup(t *testing.T) {
    logger := New(Config{
        Webhook: WebhookConfig{
            Enabled: true, URL: "http://localhost:9999",
            DedupInterval: 500 * time.Millisecond,
        },
        SlowQueryThreshold: 1 * time.Millisecond,
    })
    defer logger.Close()

    // 50개 고유 쿼리 → map 성장 확인
    for i := 0; i < 50; i++ {
        logger.Log(Event{Query: fmt.Sprintf("SELECT %d", i), DurationMS: 50})
    }
    time.Sleep(300 * time.Millisecond)
    // afterInsert == 50

    // 500ms DedupInterval 경과 후 cleanup 확인
    time.Sleep(800 * time.Millisecond)
    // afterCleanup == 0 ✓
}