Postgres와 Qdrant를 동기화 상태로 유지하기

Postgres와 Qdrant를 동기화 상태로 유지하기 (data-synchronization-with-postgres)

벡터를 Qdrant로 마이그레이션했지만 여전히 Postgres를 데이터의 근원(source of truth)으로 쓴다면, 다음 과제는 데이터가 바뀔 때 두 시스템을 동기화 상태로 유지하는 거예요.

이 가이드는 단순한 애플리케이션 레벨의 이중 쓰기(dual-write)부터 프로덕션급 변경 데이터 캡처(Change Data Capture, CDC)까지, 점점 더 견고해지는 세 가지 동기화 아키텍처를 다뤄요. 동작하는 코드, 실패 모드 분석, 그리고 각각을 언제 써야 하는지에 대한 명확한 안내까지 함께 담고 있어요.

Qdrant와 나란히 전용 벡터 저장소가 정말 필요한지 확신이 없다면, pgvector tradeoffs 블로그 포스트를 읽어보세요. pgvector로 충분한 여섯 가지 조건과, 언제 그 한계를 넘어서야 하는지 설명해 주거든요.

출처: Qdrant 공식 문서

세 가지 동기화 티어 (Three Tiers of Sync)

티어 패턴 가장 적합한 곳 동기화 지연
Tier 1: Dual-Write 요청 핸들러에서 두 곳 모두에 쓰기 프로토타입, 1만 건 미만 레코드 없음 (동기식)
Tier 2: Transactional Outbox Outbox 테이블 + 백그라운드 워커 대부분의 프로덕션 앱 수 초
Tier 3: Change Data Capture Debezium + Redpanda/Kafka 고처리량(high-throughput), 멀티 컨슈머 수 초

티어 선택하기 (Choosing Your Tier)

동기화 티어 선택

이 티어들은 영구적인 결정이 아니에요. Tier 1에서 시작하세요. 한계에 부딪히면 — 예를 들어 Qdrant 중단으로 드리프트(drift)가 너무 많이 쌓이거나, 쓰기 지연이 눈에 띄게 되면 — Tier 2로 넘어가세요. Tier 2가 병목이 되거나 재생(replay) 능력이 필요해져야 비로소 Tier 3에 투자하세요.

Tier 1: 애플리케이션 레벨 이중 쓰기 (Application-Level Dual-Write)

"그냥 앱 코드에서 해라"

아키텍처 (Architecture)

모든 CRUD 엔드포인트가 같은 요청 핸들러 안에서 Postgres에 먼저 쓰고, 그다음 Qdrant에 써요. Qdrant 쓰기가 실패하면 오류는 로그로 남고 요청은 성공해요 — Postgres가 데이터의 근원이므로, 조정(reconciliation) 작업이 나중에 드리프트를 고칠 수 있기 때문이에요.

Tier 1: 이중 쓰기 아키텍처

코드 (The Code)

라우트 핸들러는 예상한 그대로예요. 하나씩 쓰고, Qdrant 호출 주변에 오류 처리를 두면 돼요.

@router.post("/products", response_model=ProductResponse, status_code=201)
async def create_product(product: ProductCreate):
    # 1. Postgres에 먼저 쓴다 — 여기가 source of truth
    row = await insert_product(product.model_dump())
    # 2. Qdrant에 쓴다 — 실패해도 비차단; reconcile이 드리프트 처리
    try:
        await upsert_product(row)
    except Exception as exc:
        logger.error("Qdrant upsert failed for %s: %s", product.article_id, exc)
    return row

같은 패턴이 모든 변경 연산에 적용돼요. Postgres를 먼저, Qdrant를 두 번째로 쓰고, 예외는 잡아서 기록만 하고 다시 발생시키지 않아요.

실패 모드 (Failure Modes)

실패 결과 완화
Qdrant 중단 Postgres 쓰기 성공; Qdrant 쓰기 조용히 건너뜀 로그 남김; reconcile가 드리프트 수정
Qdrant 느림 요청 대기 시간 급증(Qdrant 호출에 블로킹) 클라이언트의 구성 가능한 타임아웃
Qdrant 쓰기 후 Postgres 실패 Qdrant에 고아(orphaned) 포인트 Postgres를 먼저 씀; reconcile --fix가 정리
네트워크 분할 부분 쓰기 발생 조정 스크립트

이 접근 방식이 잘하는 점 (What This Approach Gets Right)

이중 쓰기의 가장 강력한 장점은 **정확성이 아니라 인지적 단순함(cognitive simplicity)**이에요. 새 엔지니어가 이 코드를 읽으면 전체 동기화 스토리를 즉시 이해해요. 백그라운드 워커도, 큐도, 별도 프로세스도 없어요. 요청 핸들러 자체가 동기화 메커니즘인 거죠.

이 단순함은 반복 속도가 운영적 엄격함보다 중요한 프로토타입, 내부 도구, 초기 단계 제품에서 실질적인 가치가 있어요.

어디서 무너지는가 (Where It Falls Apart)

쓰기 경로가 Qdrant 가용성에 결합돼 있어요. Qdrant가 잠깐이라도 문제가 생기면 — 짧은 한 번조차 — 나중에 정리해야 할 드리프트가 생겨요. 모든 쓰기가 Qdrant에 도달한다는 보장은 없으니, 궁극적으로는 조정(reconciliation)에 의존해서 바로잡게 돼요.

더 미묘한 점: 요청 대기 시간에 Qdrant 왕복(round-trip)이 포함돼요. 쓰기가 많은 워크로드에서는 이게 병목이 돼요.

언제 쓰면 좋을까 (When to Use This)

  • 프로토타입과 MVP
  • 가끔의 비일관성을 용인할 수 있는 내부 도구
  • 낮은 쓰기 처리량(제품 1만 개 미만, 하루 수백 건 미만)
  • 빠르게 출시하고 운영 문제는 나중에 다시 보려는 팀

Tier 2: 트랜잭션 아웃박스 패턴 (Transactional Outbox Pattern)

"동기화 이벤트를 절대 잃지 않는다"

아키텍처 (Architecture)

요청 핸들러에서 직접 Qdrant에 쓰는 대신, 제품 쓰기와 같은 Postgres 트랜잭션 안에서 sync_outbox 테이블에 이벤트를 써요. 백그라운드 워커가 이 이벤트들을 가져와서 Qdrant에 비동기로 동기화해요.

아웃박스 이벤트는 제품 쓰기가 성공한 경우에만 존재해요. 둘 사이에 갭이 없고 원자적(atomically)으로 커밋돼요.

Tier 2: 트랜잭션 아웃박스 아키텍처

아웃박스 테이블 (The Outbox Table)

CREATE TABLE sync_outbox (
  id BIGSERIAL PRIMARY KEY,
  entity_id VARCHAR(20) NOT NULL,  -- article_id
  operation VARCHAR(10) NOT NULL,  -- 'upsert' | 'delete'
  payload JSONB,                   -- product snapshot at write time
  status VARCHAR(20) DEFAULT 'pending',
  attempts INT DEFAULT 0,
  max_attempts INT DEFAULT 5,
  last_error TEXT,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  processed_at TIMESTAMPTZ
);

payload 컬럼을 주목하세요. 쓰기 시점의 전체 제품 데이터를 저장해요. 워커가 products 테이블을 다시 조회하지 않아요 — 워커가 이벤트를 처리할 무렵엔 제품이 다시 업데이트됐을 수 있으니까요. payload는 이벤트가 만들어진 시점에 유효했던 스냅샷이에요.

쓰기 경로 (The Write Path)

모든 CRUD 엔드포인트는 제품과 아웃박스 이벤트 둘 다를 쓰는 단일 Postgres 트랜잭션을 열어요.

@router.post("/products", response_model=ProductResponse, status_code=201)
async def create_product(product: ProductCreate):
    pool = await get_pool()
    async with pool.acquire() as conn:
        async with conn.transaction():
            row = await conn.fetchrow(INSERT_SQL, *product_values())
            await enqueue_upsert(conn, product.article_id, dict(row))
    return dict(row)

enqueue_upsert는 같은 연결, 같은 트랜잭션 안에서 sync_outbox에 행을 삽입해요. 뭐 하나라도 실패하면 두 쓰기가 함께 롤백돼요.

Postgres 트랜잭션이 커밋되자마자 요청이 반환돼요. 요청 처리 중에는 Qdrant가 전혀 건드려지지 않아요.

워커 (The Worker)

백그라운드 비동기 작업이 아웃박스 이벤트를 **배치(batch)**로 처리해요. 핵심 동시성 기본 요소는 FOR UPDATE SKIP LOCKED예요. 이게 여러 워커가 서로 충돌 없이 안전하게 이벤트를 가져오게 해줘요.

async def process_batch() -> int:
    pool = await get_pool()
    async with pool.acquire() as conn:
        rows = await conn.fetch("""
            UPDATE sync_outbox
            SET status = 'processing', attempts = attempts + 1
            WHERE id IN (
                SELECT id FROM sync_outbox
                WHERE status IN ('pending', 'failed')
                  AND attempts < max_attempts
                ORDER BY created_at ASC
                LIMIT $1
                FOR UPDATE SKIP LOCKED
            )
            RETURNING *
        """, BATCH_SIZE)
        for row in rows:
            await _process_event(dict(row))
    return len(rows)

실패한 이벤트는 attempts 카운터가 증가하고 status가 다시 'pending'으로 설정돼요(max_attempts를 초과하면 'failed'). Qdrant 업서트는 포인트 ID 기준으로 자연스럽게 **멱등(idempotent)**이라 중복 처리는 무해해요.

두 가지 워커 모드 (Two Worker Modes)

구현은 두 가지 전달 전략을 지원해요.

폴링(Polling) (기본값): 워커가 N초마다 깨어나 대기 중인 이벤트를 확인하고 처리해요. 단순하고 견고하며 지속적인 DB 연결이 필요 없어요.

LISTEN/NOTIFY: 새 행이 삽입되면 Postgres 트리거가 NOTIFY sync_outbox_insert를 발생시켜요. 워커가 즉시 깨어나 거의 실시간 동기화를 제공해요. 바쁜 폴링 없이요.

async def run_listen_worker() -> None:
    conn = await asyncpg.connect(dsn)
    await conn.execute("LISTEN sync_outbox_insert")
    while True:
        try:
            await asyncio.wait_for(conn.wait_for_notify(), timeout=30.0)
        except asyncio.TimeoutError:
            pass  # fall back to sweep anyway
        await process_batch()

30초 스윕(sweep) 폴백이 중요해요. 알림을 놓치면(예: 워커가 잠깐 다운됐다면) 폴링 폴백이 아무것도 영원히 멈춰 있지 않게 보장해요.

실패 모드 (Failure Modes)

실패 결과 완화
Qdrant 중단 sync_outbox에서 이벤트 큐잉 워커가 복구 시 백오프로 재시도
워커 크래시 이벤트가 pending으로 남음 재시작 시 워커가 처리; SKIP LOCKED가 이중 처리 방지
중복 처리 같은 이벤트 두 번 처리 Qdrant 업서트는 포인트 ID 기준 멱등
아웃박스 테이블 커짐 저장소/성능 영향 N일보다 오래된 completed 이벤트 정리

이 접근 방식이 잘하는 점 (What This Approach Gets Right)

쓰기 경로가 **Qdrant와 완전히 분리(decoupled)**돼 있어요. Qdrant가 한 시간 동안 다운되면 쓰기는 정상적으로 성공하고 이벤트가 큐에 쌓여요. Qdrant가 돌아오면 워커가 큐를 비워요. 애플리케이션이 Qdrant에 절대 블로킹되지 않아요.

전달 의미론이 명확하고 강제 가능해요. 최소 한 번(at-least-once), Postgres에 영속적, 성공하거나 max_attempts까지 재시도. 실패한 이벤트는 DB에 보이고 SQL로 조사할 수 있어요.

자연스러운 동기화 상태 엔드포인트도 얻을 수 있어요.

@router.get("/sync/status")
async def sync_status():
    return await get_sync_status()
    # → {"pending": 0, "failed": 0, "avg_lag_seconds": 1.2}

트레이드오프 (The Trade-off)

최종 일관성(eventual consistency)을 갖게 돼요. 제품이 Postgres에 쓰인 시점부터 Qdrant 검색 결과에 나타나는 시점까지 — 보통 밀리초에서 수 초 — 갭이 있어요. 대부분의 애플리케이션에선 완벽히 용인 가능해요. 쓰기가 즉시 검색 가능해야 하는 애플리케이션이라면 다른 접근이 필요해요.

또 관리할 새 테이블이 생겨요. 아웃박스 테이블은 시간이 지나며 커지므로 완료된 이벤트의 주기적 정리가 필요해요.

언제 쓰면 좋을까 (When to Use This)

  • 적정 쓰기 처리량의 대부분 프로덕션 애플리케이션
  • Qdrant 가용성이 쓰기 경로에 영향을 주면 안 되는 경우
  • 수 초의 동기화 지연을 용인할 수 있는 경우
  • 대부분의 프로덕션 Postgres + Qdrant 배포에서 권장되는 기본값

Tier 3: Debezium + Redpanda를 이용한 변경 데이터 캡처 (Change Data Capture)

"Postgres가 Qdrant에 무슨 일이 바뀌었는지 알려준다"

아키텍처 (Architecture)

CDC는 이전 두 접근 방식과 근본적으로 다르게 구조화돼요. 애플리케이션 코드가 Qdrant를 전혀 인식하지 못해요. FastAPI 라우트는 순수한 Postgres CRUD일 뿐이에요. Qdrant 클라이언트를 import하지 않고, 아웃박스에 쓰지도 않아요. 동기화는 전적으로 인프라스트럭처 계층에서 처리돼요.

Tier 3: 변경 데이터 캡처 아키텍처

동작 원리 (How It Works)

Postgres의 **WAL(Write-Ahead Log)**은 데이터베이스의 모든 변경을 순차적으로 기록한 로그예요. 크래시 복구와 복제를 위해 존재해요. wal_level = logical로 설정하면 외부 컨슈머가 이 로그를 구조화된 형식으로 읽을 수 있어요.

Debezium은 **논리 복제 슬롯(logical replication slot)**을 통해 Postgres에 연결하고, products 테이블의 모든 INSERT, UPDATE, DELETE를 JSON 이벤트로 캡처해요. 이 이벤트들은 Redpanda 토픽에 게시돼요. Python 컨슈머 서비스가 해당 토픽을 읽고 Qdrant를 호출해요.

Postgres 설정 (Postgres Setup)

-- postgresql.conf(또는 docker-compose command args)에서 활성화:
-- wal_level = logical
-- products 테이블용 publication 생성
CREATE PUBLICATION products_publication FOR TABLE products;

Debezium 커넥터 구성 (Debezium Connector Config)

{
  "name": "products-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.dbname": "fashiondb",
    "database.server.name": "pgserver",
    "table.include.list": "public.products",
    "plugin.name": "pgoutput",
    "publication.name": "products_publication",
    "slot.name": "qdrant_sync",
    "topic.prefix": "pgserver",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.delete.handling.mode": "rewrite"
  }
}

ExtractNewRecordState 변환은 Debezium의 봉투(envelope) 형식을 펼쳐 평평한(flat) payload로 만들고, __op 필드(c = create, u = update, d = delete)를 추가해요. 그래서 컨슈머가 원시 Debezium 스키마를 파싱하지 않아도 돼요.

컨슈머 서비스 (The Consumer Service)

from confluent_kafka import Consumer

def consume_and_sync():
    consumer = Consumer({
        "bootstrap.servers": "redpanda:9092",
        "group.id": "qdrant-sync",
        "auto.offset.reset": "earliest",
        "enable.auto.commit": "false",  # manual commit after successful processing
    })
    consumer.subscribe(["pgserver.public.products"])

    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        event = json.loads(msg.value())
        op = event.get("__op")  # 'c', 'u', or 'd'
        if op in ("c", "u"):
            upsert_to_qdrant(event)
        elif op == "d":
            delete_from_qdrant(event["article_id"])
        consumer.commit()  # commit only after successful processing

수동 오프셋 커밋(각 메시지를 성공적으로 처리한 뒤에만 커밋)은 최소 한 번 의미론을 제공해요. 컨슈머가 메시지 처리 도중 크래시하면, 재시작 시 마지막으로 커밋한 오프셋부터 다시 처리해요.

Redpanda 대 Kafka (Redpanda vs. Kafka)

이 예제는 이벤트 버스로 Redpanda를 사용해요. Redpanda는 Kafka 유선 프로토콜과 호환되지만 단일 바이너리로 출시되고 ZooKeeper 의존성이 없어서 실행이 더 단순해요. 컨슈머 코드는 Redpanda를 가리키든 Apache Kafka를 가리키든 동일해요. bootstrap.servers만 바꾸면 돼요.

킬러 기능: 이력 재생 (The Killer Feature: Replaying History)

CDC의 가장 과소평가된 장점은 **재생 가능성(replayability)**이에요. Qdrant를 처음부터 재구축해야 한다면 — 새 인덱스 구성, 새 클러스터로의 마이그레이션, 재해 복구 — Redpanda 토픽을 처음부터 재생해서 할 수 있어요. products 테이블에서 일어난 모든 변경이 이벤트 스트림에 기록돼 있거든요.

이중 쓰기(Tier 1)나 아웃박스(Tier 2)로는 전체 재구축 시 Postgres에서 대량 내보내기(bulk export)를 실행해야 해요. CDC에서는 이벤트 스트림 자체가 재구축 메커니즘이에요.

실패 모드 (Failure Modes)

실패 결과 완화
Qdrant 중단 컨슈머 일시 중지; Redpanda가 이벤트 보관 컨슈머 재시도; 복구 시 오프셋부터 재개
Redpanda 중단 Debezium 버퍼링; WAL 증가 WAL 크기 모니터링; 프로덕션에서 복제로 Redpanda HA
Debezium 크래시 마지막 체크포인트 이후 변경을 WAL이 보관 재시작 시 Debezium이 복제 슬롯부터 재개
Postgres 스키마 변경 커넥터 재시작 필요할 수 있음 커넥터 상태 모니터링; 스테이징에서 스키마 마이그레이션 테스트

WAL 디스크 부풀림 문제 (The WAL Disk Bloat Problem)

CDC에만 특유한 운영 위험이 하나 있어요. Postgres는 복제 슬롯 컨슈머가 WAL 세그먼트를 인정(acknowledge)할 때까지 보관해요. 컨슈머가 오래 다운되면 Postgres가 상당한 디스크 사용량을 쌓을 수 있어요. 복제 슬롯 지연(lag)을 모니터링하세요.

SELECT slot_name,
       pg_size_pretty(
         pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)
       ) AS lag
FROM pg_replication_slots;

이 값이 수 GB를 넘으면 알림을 설정하세요. 극단적인 경우(컨슈머 영구 중단)에는 복제 슬롯을 삭제해야 할 수도 있어요. 그러면 이벤트 스트림이 아니라 **Postgres에서 Qdrant를 다시 시딩(seeding)**하게 돼요.

언제 쓰면 좋을까 (When to Use This)

  • 아웃박스 테이블이 병목이 될 고쓰기 처리량 시스템
  • Qdrant뿐 아니라 여러 다운스트림 컨슈머가 변경에 반응해야 하는 경우
  • 애플리케이션 코드가 동기화 관심사와 완전히 분리돼야 하는 경우
  • 기존 Redpanda/Kafka 인프라가 있는 팀
  • 처음부터 재생(from-scratch replay) 능력이 요구사항인 경우

Tier 3는 정말 강력하지만 실질적인 운영 비용이 따라와요. Redpanda와 Debezium을 배포·구성·모니터링·업그레이드해야 하죠. 이미 스트리밍 인프라를 운영하지 않는다면, 단일 동기화 용도를 위해 이를 추가하기 전에 신중히 생각하세요. Tier 2가 훨씬 적은 복잡성으로 대부분의 프로덕션 시나리오를 처리해요.

보편적인 안전망: 조정 (The Universal Safety Net: Reconciliation)

모든 동기화 아키텍처는 결국 드리프트가 생겨요. 이 잔여물을 잡아내는 게 **조정 스크립트(reconciliation script)**예요.

async def reconcile(fix: bool = False) -> ReconcileResult:
    pg_ids = set(await get_all_article_ids_from_postgres())
    qdrant_ids = set(await get_all_point_ids_from_qdrant())

    missing_in_qdrant = pg_ids - qdrant_ids  # need to sync
    orphaned_in_qdrant = qdrant_ids - pg_ids  # need to delete

    if fix:
        for article_id in missing_in_qdrant:
            product = await get_product(article_id)
            await upsert_product(product)
        if orphaned_in_qdrant:
            await qdrant_client.delete(
                collection_name="products",
                points_selector=orphaned_in_qdrant,
            )

    return ReconcileResult(
        postgres_count=len(pg_ids),
        qdrant_count=len(qdrant_ids),
        missing_in_qdrant=len(missing_in_qdrant),
        orphaned_in_qdrant=len(orphaned_in_qdrant),
        in_sync=len(missing_in_qdrant) == 0 and len(orphaned_in_qdrant) == 0,
    )

Postgres는 데이터의 근원이고, Qdrant는 **파생된 읽기 저장소(derived read store)**예요. 둘이 갈라지면 Postgres가 이겨요. 이걸 스케줄로 — 보통 야간이면 충분해요 — 그리고 드리프트가 의심될 때마다 수동으로 실행하세요.

비교 매트릭스 (Comparison Matrix)

차원 Tier 1: Dual-Write Tier 2: Outbox Tier 3: CDC
동기화 복잡성 낮음 중간 높음
추가 인프라 없음 Outbox 테이블 + 워커 Redpanda + Debezium + 컨슈머
일관성 모델 Best-effort At-least-once, 최종적 At-least-once, 최종적
쓰기 지연 영향 Qdrant 왕복 추가 없음 (비동기) 없음 (비동기)
Qdrant 중단 영향 드리프트 발생 Postgres에서 이벤트 큐잉 Redpanda에서 이벤트 큐잉
직접 SQL 캡처 아님 아님 예 (모든 WAL 변경)
재생 능력 없음 제한적 (아웃박스 보존) 예 (Redpanda 보존)
운영 오버헤드 최소 낮음-중간 높음
가장 적합한 곳 프로토타입, 내부 도구 대부분의 프로덕션 앱 고처리량, 멀티 컨슈머

더 알아보기 (Learn more)