Go로 PostgreSQL 프록시 만들기 (60) - Extended Query 커넥션 바인딩과 E2E 전수 검증

Round 1: Parse + Describe + Sync   --> ReadyForQuery
Round 2: Bind + Execute  + Sync   --> ReadyForQuery

Round 1: Parse+Describe+Sync → Reader A에서 실행 → ReadyForQuery → 커넥션 릴리즈
Round 2: Bind+Execute+Sync   → Reader B에 할당됨  → "prepared statement does not exist"

var extHasExecute bool // true if the current batch contains an Execute message

case protocol.MsgExecute:
    extHasExecute = true
    extBuf = append(extBuf, protocol.CopyMessage(msg))

// relayQueries 시작 부분의 상태 변수
var boundReader *pool.Conn
var boundReaderPool *pool.Pool
var boundReaderAddr string

if boundReader != nil {
    // Reuse reader bound from a previous Parse-only batch
    ct.setFromConn(boundReaderAddr, boundReader)
    if err := s.forwardExtBatch(boundReader, extBuf, msg); err != nil {
        boundReaderPool.Discard(boundReader)
        boundReader, boundReaderPool, boundReaderAddr = nil, nil, ""
        return
    }
    // ...
    // Release bound reader after Execute completes
    if extHasExecute {
        boundReaderPool.Release(boundReader)
        boundReader, boundReaderPool, boundReaderAddr = nil, nil, ""
    }
}

} else if !extHasExecute && len(extBuf) > 0 {
    // Parse-only batch: acquire reader and keep connection bound
    readerAddr := dbg.balancer.Next()
    rConn, rPool, usedAddr, err := s.acquireReaderForBind(ctx, readerAddr, dbg)
    // ... forward and relay ...
    // Keep reader bound for next batch
    boundReader = rConn
    boundReaderPool = rPool
    boundReaderAddr = usedAddr
}

} else {
    // Normal reader path with Execute
    readerAddr := dbg.balancer.Next()
    // ... acquire, forward, relay, release ...
}

func (s *Server) acquireReaderForBind(ctx context.Context, readerAddr string, dbg *DatabaseGroup) (*pool.Conn, *pool.Pool, string, error) {
    if readerAddr != "" {
        if cb, ok := dbg.ReaderCB(readerAddr); ok {
            if cb.Allow() != nil {
                readerAddr = "" // circuit breaker open
            }
        }
    }
    if readerAddr != "" {
        if rPool, ok := dbg.ReaderPool(readerAddr); ok {
            if rConn, err := rPool.Acquire(ctx); err == nil {
                return rConn, rPool, readerAddr, nil
            }
            // ... mark unhealthy ...
        }
    }
    // Fallback to writer
    wPool := dbg.writerPool
    wConn, err := wPool.Acquire(ctx)
    if err != nil {
        return nil, nil, "", fmt.Errorf("acquire reader (with writer fallback): %w", err)
    }
    return wConn, wPool, dbg.writerAddr, nil
}

# --- 신규 기능 활성화 ---
auth:
  enabled: true
  users:
    - username: "postgres"
      password: "postgres"
      max_connections: 0
    - username: "limited"
      password: "limited"
      max_connections: 3

firewall:
  enabled: true
  block_delete_without_where: true
  block_update_without_where: true

connection_limits:
  enabled: true
  default_max_connections_per_user: 50
  default_max_connections_per_database: 50

session_compatibility:
  enabled: true
  mode: "warn"

data_api:
  enabled: true
  listen: "0.0.0.0:18080"
  api_keys:
    - "test-api-key-12345"

rate_limit:
  enabled: true
  rate: 10000
  burst: 5000

audit:
  enabled: true
  slow_query_threshold: 100ms

observability:
  sql_redaction: "literals"