Go로 PostgreSQL 프록시 만들기 (53) - QA 4차: 라우팅 우회와 운영 안전성

-- Simple Query: 정상 차단
INSERT INTO orders VALUES (1, 'test');
→ "cannot execute write query: pgmux is in read-only mode"

-- 1단계: Parse (read-only 전환 전)
Parse("INSERT INTO orders VALUES ($1, $2)")
Bind(params...)
Execute
Sync
→ 성공

-- 2단계: admin에서 read-only 모드 활성화

-- 3단계: 같은 prepared statement 재사용 (Parse 없이)
Bind(params...)     ← Parse가 없으므로 extIsWrite 미설정
Execute
Sync
→ 성공!? ← read-only 우회

// query.go — Parse 때만 설정
case protocol.MsgParse:
    if s.classifyQuery(query) == router.QueryWrite {
        extIsWrite = true   // ← Parse가 있을 때만 실행
    }

// Sync 때 체크
case protocol.MsgSync:
    if s.InReadOnly() && extIsWrite {  // extIsWrite가 false → 통과
        // reject...
    }
    // 배치 끝에서 리셋
    extIsWrite = false

// router.go
type Session struct {
    stmtRoutes map[string]Route
    stmtWrite  map[string]bool  // 추가: statement별 write 분류
}

func (s *Session) RegisterStatement(name, query string) Route {
    route := s.routeLocked(query)
    s.stmtRoutes[name] = route
    // write 분류도 함께 저장
    if s.astParser {
        s.stmtWrite[name] = ClassifyAST(query) == QueryWrite
    } else {
        s.stmtWrite[name] = Classify(query) == QueryWrite
    }
    return route
}

func (s *Session) StatementIsWrite(name string) bool {
    if isWrite, ok := s.stmtWrite[name]; ok {
        return isWrite
    }
    return true  // 모르면 write로 간주 (safe default)
}

case protocol.MsgBind:
    route := session.StatementRoute(stmtName)
    if route == router.RouteWriter {
        extRoute = router.RouteWriter
    }
    // 추가: write 분류 복원
    if session.StatementIsWrite(stmtName) {
        extIsWrite = true
    }

-- 행 잠금 획득 (reader에서 실행 불가)
SELECT * FROM orders WHERE id = 1 FOR UPDATE;

-- 시퀀스 값 증가 (writer에서만 실행해야 함)
SELECT nextval('order_id_seq');

-- 세션 파라미터 변경
SELECT set_config('statement_timeout', '5000', false);

-- Advisory lock 획득
SELECT pg_advisory_lock(12345);

// parser.go
var lockingClauses = []string{
    "FOR UPDATE", "FOR NO KEY UPDATE",
    "FOR SHARE", "FOR KEY SHARE",
}

var sideEffectFuncs = []string{
    "NEXTVAL(", "SETVAL(", "CURRVAL(",
    "SET_CONFIG(", "PG_ADVISORY_LOCK(",
    "PG_ADVISORY_XACT_LOCK(", "PG_ADVISORY_UNLOCK(",
    "PG_TRY_ADVISORY_LOCK(", "PG_NOTIFY(",
    "LO_CREATE(", "LO_UNLINK(",
}

func isSideEffectfulSelect(query string) bool {
    stripped := stripStringLiterals(query)
    upper := strings.ToUpper(stripped)
    for _, lc := range lockingClauses {
        if strings.Contains(upper, lc) { return true }
    }
    return hasSideEffectFunc(upper)
}

if kw == "SELECT" {
    upper := strings.ToUpper(query)
    if strings.Contains(upper, "FOR UPDATE") || ... || hasSideEffectFunc(upper) {
        return 0, false  // full parser로 위임
    }
    return QueryRead, true
}

// parser_ast.go
case *pg_query.Node_SelectStmt:
    s := n.SelectStmt
    // CTE 체크 (기존)
    // ...
    // Locking clause: FOR UPDATE, FOR SHARE 등
    if len(s.GetLockingClause()) > 0 {
        return true
    }
    // 부작용 함수 호출
    return hasSideEffectFuncCalls(node)

/* timeout:5s */ SELECT * FROM large_table;

// query.go — Sync 핸들러
extQueryTimeout := s.getConfig().Pool.QueryTimeout  // global만!

// executeSynthesizedQuery — timer 없이 relay
if err := protocol.WriteMessage(wConn, protocol.MsgQuery, queryPayload); err != nil { ... }
if err := s.relayUntilReady(clientConn, wConn); err != nil { ... }
// ← timeout 없이 무한 대기 가능

var extQueryText string  // 최신 Parse의 query 텍스트

case protocol.MsgParse:
    extQueryText = query  // 저장

case protocol.MsgSync:
    extQueryTimeout := s.resolveQueryTimeout(extQueryText, s.getConfig())

func (s *Server) executeSynthesizedQuery(ctx context.Context, ..., queryTimeout time.Duration, ...) error {
    ct.setFromConn(dbg.writerAddr, wConn)
    stopTimer := s.startQueryTimer(queryTimeout, ct, "writer")
    // ... relay ...
    if stopTimer != nil { stopTimer() }
    ct.clear()
}

// main.go
apiSrv := dataapi.New(..., srv.DefaultDBName(), ...)
//                         ^^^^^^^^^^^^^^^^^ 문자열 한 번 평가

// handler.go
type Server struct {
    defaultDB string  // 정적 필드
}

// handler.go
type Server struct {
    defaultDBFn func() string  // 동적 조회
}

func (s *Server) handleQuery(...) {
    dbName := r.URL.Query().Get("database")
    if dbName == "" {
        dbName = s.defaultDBFn()  // 매 요청마다 최신값
    }
}

// main.go
dataapi.New(..., srv.DefaultDBName, ...)  // 메서드 참조 (호출이 아님)

if c.Pool.MaxConnections < 1 {
    return fmt.Errorf("pool.max_connections must be >= 1")
}

databases:
  mydb:
    pool:
      max_connections: -5   # ← 검증 없이 통과
      idle_timeout: -1s     # ← 검증 없이 통과

for name, db := range c.Databases {
    // 기존: host/port 검증
    // 추가: pool 설정 검증
    if db.Pool.MaxConnections < 1 {
        return fmt.Errorf("databases.%s.pool.max_connections must be >= 1, got %d",
            name, db.Pool.MaxConnections)
    }
    if db.Pool.MinConnections < 0 { ... }
    if db.Pool.MinConnections > db.Pool.MaxConnections { ... }
    if db.Pool.IdleTimeout < 0 { ... }
    if db.Pool.MaxLifetime < 0 { ... }
    if db.Pool.ConnectionTimeout < 0 { ... }
    if db.Pool.QueryTimeout < 0 { ... }
}