Field Log · Entry
RAG Session 영속화: PostgreSQL·Redis·Idempotency 설계 (7/10)
오늘의 결론
- PostgreSQL을 session state·event·idempotency result의 source of truth로 두고, Redis는 재생 가능한 cache와 짧은 coordination hint로 제한합니다.
- 동시 append는
SELECT 후 INSERT만으로 막지 않습니다. Session version의 조건부 update와UNIQUE(session_id, sequence)를 DB의 최종 판정자로 사용합니다.- Idempotency key는 tenant·operation 범위로 묶고 canonical request hash를 저장합니다. 같은 key·같은 hash는 같은 완료 응답을 돌려주고, 같은 key·다른 hash는 충돌입니다.
SET NX EXlock은 durable idempotency가 아닙니다. Process crash 뒤에도 완료 응답과 요청 동일성을 판정할 DB record가 필요합니다.- Exactly-once라고 부르지 않습니다. DB commit과 외부 side effect 사이에는 crash window가 남으므로 downstream idempotency key 또는 transactional outbox가 필요합니다.
앞 글에서는 typed state와 bounded tool loop를 만들었습니다. 이번에는 request 재전송·worker crash·동시 실행에도 같은 run을 안전하게 이어갈 durable session boundary를 설계합니다.
이 글의 대상과 학습 시간
- 대상: 대화 session·중복 POST·worker 재시작을 처리해야 하는 Python backend 개발자
- 선수 지식: SQL transaction, unique constraint, cache-aside 기본 개념
- 빠르게 읽기: 약 15분
- SQL과 동시성 test까지 따라 하기: 약 70분
- 산출물: session/event/idempotency schema, optimistic append, durable duplicate response protocol
1. Memory를 Python list로 두면 무엇이 깨지는가
sessions[session_id].append(message)한 process demo에는 충분하지만 service에서는 다음 일이 발생합니다.
- Worker가 재시작하면 state가 사라짐
- 두 request가 같은 sequence에 동시에 append
- Client timeout 뒤 같은 POST를 다시 보냄
- 첫 요청은 성공했지만 응답 전달 전에 연결이 끊김
- Tool side effect가 성공한 뒤 DB 기록 전에 crash
- 여러 worker의 local memory가 서로 다름
- Redis TTL 만료가 run 기록 삭제로 이어짐
Durable session은 “대화 저장”보다 넓습니다. 어떤 요청이 어떤 version의 state를 읽고, 어떤 event를 한 번 commit했으며, 재전송 시 무엇을 반환해야 하는지를 기록합니다.
2. 저장할 것과 저장하지 않을 것을 구분한다
저장할 것
- User·assistant message와 공개 가능한 structured content
- Evidence ID·document version·source span manifest
- Tool proposal의 normalized hash와 validation outcome
- Tool receipt의 safe result·error code
- State transition과 termination reason
- Model·prompt·index·policy version
- Usage·latency·request/provider ID
- Idempotency key의 request hash와 완료 response
기본적으로 저장하지 않을 것
- Raw chain-of-thought
- API key·authorization header·signed credential
- Provider error body 전문
- 불필요한 raw document 전문
- Access token·password가 든 tool argument
Audit에 필요한 것은 숨은 생각이 아니라 입력 provenance, 외부 action, validation decision, outcome입니다.
3. 최소 PostgreSQL schema
CREATE TYPE session_status AS ENUM (
'running', 'completed', 'failed', 'cancelled'
);
CREATE TABLE agent_sessions (
tenant_id text NOT NULL,
session_id uuid NOT NULL,
version bigint NOT NULL DEFAULT 0,
status session_status NOT NULL DEFAULT 'running',
termination text,
state_summary jsonb NOT NULL DEFAULT '{}'::jsonb,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_id, session_id),
CHECK (version >= 0)
);
CREATE TABLE agent_events (
tenant_id text NOT NULL,
session_id uuid NOT NULL,
sequence bigint NOT NULL,
event_type text NOT NULL,
payload jsonb NOT NULL,
request_id text NOT NULL,
created_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_id, session_id, sequence),
UNIQUE (tenant_id, request_id),
FOREIGN KEY (tenant_id, session_id)
REFERENCES agent_sessions (tenant_id, session_id)
);
CREATE INDEX agent_events_session_idx
ON agent_events (tenant_id, session_id, created_at);Tenant ID를 child table과 primary/foreign key에 반복해 query의 security boundary가 눈에 보이게 합니다. PostgreSQL Row-Level Security를 추가할 수 있지만 application query의 tenant 조건을 생략해도 된다는 뜻은 아닙니다.
Snapshot과 event의 역할
agent_events: append-only history와 replay 근거state_summary: 매번 전체 history를 읽지 않게 한 current snapshotversion: snapshot과 마지막 event sequence를 연결하는 optimistic concurrency token
Event payload schema version도 넣어 migration과 replay를 관리할 수 있습니다.
4. Optimistic append를 한 transaction으로 만든다
Client는 자신이 읽은 expected_version을 보냅니다.
BEGIN;
UPDATE agent_sessions
SET version = version + 1,
state_summary = $new_state,
status = $new_status,
termination = $termination,
updated_at = now()
WHERE tenant_id = $tenant_id
AND session_id = $session_id
AND version = $expected_version
RETURNING version;
-- UPDATE가 정확히 1행을 반환했을 때만 실행
INSERT INTO agent_events (
tenant_id, session_id, sequence,
event_type, payload, request_id
) VALUES (
$tenant_id, $session_id, $expected_version + 1,
$event_type, $payload, $request_id
);
COMMIT;Update가 0행이면 session이 없거나 다른 writer가 version을 먼저 바꿨습니다. 두 경우를 다시 조회해 SessionNotFound와 VersionConflict로 구분합니다.
왜 unique constraint도 필요한가
Application bug·재시도·잘못된 isolation 처리로 같은 sequence insert가 들어와도 DB가 최종 거부합니다. request_id unique는 같은 application request가 다른 sequence로 중복 기록되는 것도 막습니다.
5. Python repository가 conflict를 번역한다
from dataclasses import dataclass
@dataclass(frozen=True)
class AppendCommand:
tenant_id: str
session_id: str
expected_version: int
request_id: str
event_type: str
event_payload: dict
new_state_summary: dict
new_status: str
termination: str | None
class PostgresSessionStore:
def __init__(self, pool) -> None:
self._pool = pool
async def append(self, command: AppendCommand) -> int:
async with self._pool.acquire() as connection:
async with connection.transaction():
new_version = await connection.fetchval(
UPDATE_SESSION_SQL,
command.tenant_id,
command.session_id,
command.expected_version,
command.new_state_summary,
command.new_status,
command.termination,
)
if new_version is None:
exists = await connection.fetchval(
SELECT_SESSION_EXISTS_SQL,
command.tenant_id,
command.session_id,
)
if not exists:
raise SessionNotFound(command.session_id)
raise VersionConflict(
command.session_id,
command.expected_version,
)
await connection.execute(
INSERT_EVENT_SQL,
command.tenant_id,
command.session_id,
new_version,
command.event_type,
command.event_payload,
command.request_id,
)
return int(new_version)Transaction 안에서 event insert가 실패하면 session version update도 rollback됩니다. DB exception code를 그대로 API에 노출하지 않고 VersionConflict, DuplicateRequest, PersistenceUnavailable로 번역합니다.
6. Idempotency record는 lock보다 오래 산다
CREATE TYPE idempotency_status AS ENUM (
'in_progress', 'completed', 'failed'
);
CREATE TABLE idempotency_keys (
tenant_id text NOT NULL,
operation text NOT NULL,
key text NOT NULL,
request_hash bytea NOT NULL,
status idempotency_status NOT NULL,
owner_token uuid NOT NULL,
lease_expires_at timestamptz NOT NULL,
response_status integer,
response_body jsonb,
error_code text,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now(),
expires_at timestamptz NOT NULL,
PRIMARY KEY (tenant_id, operation, key),
CHECK (
(status = 'completed' AND response_status IS NOT NULL AND response_body IS NOT NULL)
OR status <> 'completed'
)
);
CREATE INDEX idempotency_expiry_idx ON idempotency_keys (expires_at);Scope가 중요한 이유를 봅니다.
tenant A / create_answer / key abc
tenant A / cancel_run / key abc → 다른 operation
tenant B / create_answer / key abc → 다른 tenantKey 문자열 하나만 global unique로 만들면 tenant 간 충돌이나 operation 오용이 생깁니다.
7. Request hash를 canonical input으로 계산한다
같은 key에 다른 request body가 오면 이전 응답을 돌려줘서는 안 됩니다.
import hashlib
import json
def canonical_request_hash(
*,
tenant_id: str,
operation: str,
body: dict,
) -> bytes:
canonical = json.dumps(
{
"tenant_id": tenant_id,
"operation": operation,
"body": body,
},
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
).encode("utf-8")
return hashlib.sha256(canonical).digest()Body에서 server가 채우는 timestamp·trace ID처럼 매번 달라지는 field는 idempotent business input과 분리합니다. Secret을 canonical JSON이나 log에 넣지 않습니다. File upload는 content digest와 stable metadata를 사용합니다.
Canonical JSON의 한계
Float 표현, Unicode normalization, duplicate JSON key 등은 client·language마다 다를 수 있습니다. API boundary에서 Pydantic model로 parse·normalize한 내부 object를 hash하고 algorithm version을 operation policy와 함께 고정합니다.
8. Claim protocol은 두 statement로 읽는다
첫 worker가 key를 claim합니다.
INSERT INTO idempotency_keys (
tenant_id, operation, key, request_hash,
status, owner_token, lease_expires_at, expires_at
) VALUES (
$1, $2, $3, $4,
'in_progress', $5, now() + $6::interval, now() + $7::interval
)
ON CONFLICT (tenant_id, operation, key) DO NOTHING
RETURNING *;0행이면 다음 statement에서 기존 row를 읽습니다.
SELECT request_hash, status, owner_token,
lease_expires_at, response_status, response_body, error_code,
expires_at
FROM idempotency_keys
WHERE tenant_id = $1 AND operation = $2 AND key = $3;PostgreSQL 기본 Read Committed에서는 각 statement가 새 snapshot을 봅니다. INSERT ... ON CONFLICT DO NOTHING이 concurrent uncommitted row를 기다린 뒤 0행이 되었을 때, 같은 statement 안의 fallback SELECT가 그 row를 항상 볼 것이라고 가정하지 않습니다. 별도 다음 statement로 읽으면 commit된 row를 새 snapshot에서 확인할 수 있습니다.
9. 기존 row를 네 가지로 판정한다
def classify_existing(
row: IdempotencyRow,
*,
request_hash: bytes,
now: datetime,
) -> IdempotencyDecision:
if row.expires_at <= now:
return IdempotencyDecision.expired()
if not hmac.compare_digest(row.request_hash, request_hash):
return IdempotencyDecision.conflict("same_key_different_request")
if row.status == "completed":
return IdempotencyDecision.replay(
status=row.response_status,
body=row.response_body,
)
if row.status == "in_progress" and row.lease_expires_at > now:
return IdempotencyDecision.in_progress(retry_after=1.0)
if row.status in {"in_progress", "failed"}:
return IdempotencyDecision.try_takeover()
raise InvalidIdempotencyState(row.status)Takeover는 lease_expires_at만 보고 무조건 update하지 않습니다. 읽은 owner token과 expiry를 조건으로 compare-and-swap합니다.
expired는 즉시 이전 response를 replay한다는 뜻이 아닙니다. 관측한 expires_at·owner token을 조건으로 row를 삭제하거나 새 generation으로 교체한 뒤 새 요청처럼 claim합니다. Retention window 이후 같은 key를 재사용할 수 있는지는 API contract에 명시합니다.
UPDATE idempotency_keys
SET owner_token = $new_owner,
status = 'in_progress',
lease_expires_at = now() + $lease::interval,
updated_at = now()
WHERE tenant_id = $tenant
AND operation = $operation
AND key = $key
AND owner_token = $observed_owner
AND (
(status = 'in_progress' AND lease_expires_at <= now())
OR status = 'failed'
)
RETURNING *;0행이면 다른 worker가 먼저 takeover했습니다. 다시 읽고 같은 decision protocol을 반복합니다.
10. 완료 응답을 durable하게 저장한다
Business transaction과 idempotency completion을 가능한 한 같은 DB transaction에 묶습니다.
UPDATE idempotency_keys
SET status = 'completed',
response_status = $status,
response_body = $body,
lease_expires_at = now(),
updated_at = now()
WHERE tenant_id = $tenant
AND operation = $operation
AND key = $key
AND owner_token = $owner
AND status = 'in_progress';반환 행이 0이면 lease를 잃은 worker가 늦게 결과를 쓰려는 것입니다. Owner token은 fencing token 역할을 일부 하지만 단순 UUID는 순서를 표현하지 않습니다. 외부 resource까지 fence해야 하면 증가하는 generation을 함께 사용합니다.
Response를 어디까지 저장하는가
- Client가 다시 받아야 하는 stable API response
- HTTP status와 public error code
- Citation의 stable source reference
- Run/session ID
Stream delta 전체를 idempotency row 하나에 계속 append하지 않습니다. Streaming resume가 필요하면 별도 ordered event log와 retention을 설계합니다.
11. Redis는 versioned cache로 사용한다
PostgreSQL
agent_sessions version=17 ← source of truth
Redis key
session:{tenant}:{session_id}:v17
→ serialized snapshot, short TTLVersion을 key에 포함하면 old cache를 덮어쓰는 race가 줄어듭니다.
async def load_session(tenant_id: str, session_id: str) -> Session:
pointer_key = f"session-current:{tenant_id}:{session_id}"
version = await redis.get(pointer_key)
if version is not None:
cached = await redis.get(
f"session:{tenant_id}:{session_id}:v{version.decode()}"
)
if cached is not None:
return decode_session(cached)
session = await postgres.load(tenant_id, session_id)
await redis.set(
f"session:{tenant_id}:{session_id}:v{session.version}",
encode_session(session),
ex=300,
)
await redis.eval(
SET_POINTER_IF_NEWER_LUA,
1,
pointer_key,
session.version,
300,
)
return sessionPointer update는 다음 compare-and-set script처럼 더 높은 version만 허용합니다.
local current = redis.call('GET', KEYS[1])
local incoming = tonumber(ARGV[1])
if (not current) or incoming > tonumber(current) then
redis.call('SET', KEYS[1], ARGV[1], 'EX', ARGV[2])
return 1
end
return 0DB commit 뒤 cache write 전에 crash하면 old pointer가 잠시 남을 수 있습니다. 반대로 늦은 v16 writer가 v17 pointer를 되돌리지 않도록 위 CAS를 사용합니다. Correctness가 중요한 read는 expected version을 확인하거나 DB를 다시 읽습니다. Cache TTL과 invalidation은 성능 policy이지 transaction 보장이 아닙니다.
SET NX EX가 해결하지 못하는 것
SET lock:key owner NX EX 30이는 짧은 lease 후보일 뿐 다음을 저장하지 않습니다.
- 같은 request인지 확인할 hash
- 완료 response
- DB transaction 결과
- Lock 만료 후 이전 owner의 늦은 write를 막는 fence
- Redis 장애·재시작 뒤 durable history
따라서 Redis lock만으로 “exactly once”를 선언하지 않습니다.
12. External side effect에는 crash window가 남는다
tool API 성공
↓ crash
DB에 tool receipt·idempotency completion 없음
↓ retry
tool API 다시 호출해결책은 side effect 성격에 따라 다릅니다.
- Downstream이 idempotency key를 지원하면 같은 stable key 전달
- 내부 event 발행은 DB transaction에 outbox row를 함께 기록
- Consumer는 message ID로 inbox dedup
- Reconciliation job이 external state와 pending run을 대조
- Non-idempotent 고위험 action은 human approval와 status query를 결합
DB transaction 하나로 network 전체의 exactly-once를 만들 수는 없습니다.
13. Transaction isolation을 과신하지 않는다
PostgreSQL Read Committed에서는 같은 transaction의 두 SELECT가 서로 다른 commit 상태를 볼 수 있습니다. 필요한 invariant를 다음으로 명시합니다.
- Unique constraint
- Conditional
UPDATE ... WHERE version = expected - 필요한 row의
SELECT ... FOR UPDATE - Serializable transaction + retry, 정말 필요할 때만
- Application-level idempotency contract
Isolation level 이름만 올리고 serialization failure retry를 구현하지 않으면 가용성 문제가 생길 수 있습니다.
14. 동시성 test를 실제 DB에서 실행한다
In-memory fake는 transaction race를 증명하지 못합니다.
@pytest.mark.integration
@pytest.mark.asyncio
async def test_only_one_append_wins_same_expected_version(store):
session = await store.create(tenant="fab-a")
commands = [
append_command(session, expected_version=0, request_id="req-a"),
append_command(session, expected_version=0, request_id="req-b"),
]
results = await asyncio.gather(
*(store.append(command) for command in commands),
return_exceptions=True,
)
assert sum(result == 1 for result in results) == 1
assert sum(isinstance(result, VersionConflict) for result in results) == 1Idempotency도 두 connection으로 동시에 실행합니다.
@pytest.mark.integration
@pytest.mark.asyncio
async def test_duplicate_key_replays_one_committed_response(service):
request = sample_request(idempotency_key="same-key")
first, second = await asyncio.gather(
service.answer(request),
service.answer(request),
)
assert first.body == second.body
assert first.run_id == second.run_id
assert await count_runs_for_key("same-key") == 1두 번째 호출이 in_progress를 반환하는 API라면 test 기대값을 202/409 + 이후 replay로 고정합니다. 중요한 것은 중복 run이 조용히 둘 생기지 않는 것입니다.
15. 운영 metric과 정리 job
- Session append conflict rate
- Duplicate key replay rate
- Same key/different hash conflict count
- Idempotency in-progress age
- Lease takeover count
- Late owner completion rejected count
- Redis hit rate by session version
- Stale cache detection count
- Outbox oldest pending age
- Expired row cleanup duration
expires_at이 지난 idempotency row는 batch로 정리하되 active lease와 retention·audit policy를 확인합니다. Table이 크면 time partitioning도 고려합니다.
자주 실패하는 구현 8가지
1. Process memory를 durable session으로 부른다
재시작과 multi-worker에서 사라집니다. DB version과 event를 기준으로 복구합니다.
2. SELECT max(sequence)+1만 사용한다
동시 writer가 같은 sequence를 계산합니다. Conditional version update와 unique constraint를 사용합니다.
3. Idempotency key만 저장하고 request hash를 빼먹는다
같은 key의 다른 요청에 이전 응답을 잘못 반환합니다.
4. Redis lock을 exactly-once로 부른다
완료 응답·request 동일성·crash history가 없습니다. Durable DB record가 필요합니다.
5. Lease 만료 뒤 이전 owner의 write를 허용한다
Takeover worker 결과를 덮을 수 있습니다. Owner token·generation 조건으로 late completion을 거부합니다.
6. DB commit 전에 cache를 갱신한다
Rollback된 state가 cache에 노출될 수 있습니다. Commit 이후 versioned cache를 채웁니다.
7. Raw chain-of-thought를 session memory로 저장한다
감사·복구에는 observable event와 decision receipt가 필요합니다. 민감한 숨은 추론 전문은 기본 저장 대상이 아닙니다.
8. External call과 DB write가 원자적이라고 가정한다
Network 사이에는 crash window가 남습니다. Downstream idempotency·outbox·reconciliation을 설계합니다.
직접 적용 체크리스트
- PostgreSQL이 session·event·idempotency의 source of truth다.
- Session에 monotonic version과 conditional update가 있다.
- Event에
(tenant, session, sequence)unique constraint가 있다. - Idempotency scope가 tenant·operation·key로 구성된다.
- Canonical request hash로 key 재사용 충돌을 판정한다.
- In-progress lease와 owner token으로 takeover를 조정한다.
- 완료 API response를 durable하게 저장하고 replay한다.
- Redis cache key에 session version을 포함한다.
- External side effect의 crash window 대응이 있다.
- 실제 PostgreSQL 두 connection으로 race test를 실행한다.
스스로 확인하기
UNIQUE(session_id, sequence)만 있고 expected version update가 없으면 어떤 문제가 남는가?- 같은 idempotency key와 다른 request hash를 왜 거부해야 하는가?
- Redis
SET NX EX가 완료 응답 replay를 제공하지 못하는 이유는 무엇인가? - Lease takeover 뒤 old owner의 late write를 어떻게 막는가?
- External API 성공과 DB commit 사이 crash window를 어떤 방식으로 줄일 수 있는가?
자주 묻는 질문
Q1. 모든 message를 event sourcing으로 재생해야 하나요?
아닙니다. Append-only event와 current snapshot을 함께 두면 audit·recovery와 read 성능을 절충할 수 있습니다. Event replay가 business requirement인지 먼저 정합니다.
Q2. Serializable isolation을 쓰면 idempotency key가 필요 없나요?
필요합니다. Isolation은 DB transaction 간 충돌을 다루고, idempotency는 client 재전송에 같은 business outcome을 제공하는 API contract입니다.
Q3. In-progress 중복 요청은 기다리게 해야 하나요?
짧은 작업이면 제한된 wait·poll을 할 수 있고, 긴 Agent run이면 202와 run status URL을 반환하는 편이 낫습니다. Unbounded connection wait는 피합니다.
Q4. Redis를 빼도 되나요?
가능합니다. 먼저 PostgreSQL만으로 correctness를 완성하고 실제 read load·latency가 필요할 때 cache를 추가하는 편이 단순합니다.
Q5. Session data retention은 어떻게 정하나요?
제품 기능, incident audit, 개인정보·계약 요구를 함께 봅니다. Message·evidence manifest·idempotency response마다 목적과 TTL이 다를 수 있으며 삭제 전파도 설계해야 합니다.
마무리
Durable Agent session은 대화 history 저장소가 아니라 다음 네 계약의 결합입니다.
- Version conflict가 있는 state transition
- Unique sequence를 가진 append-only event
- Request hash와 완료 response를 가진 idempotency record
- DB commit 이후 재생 가능한 versioned cache
다음 글에서는 이 application을 FastAPI endpoint로 열고, bounded queue와 SSE event contract로 delta·citation·done·error를 전달하며 client disconnect 정책과 재연결을 설계합니다.
검증 정보
- 글의 성격: PostgreSQL·Redis 공식 문서 해설 + 작성자 persistence 설계 제안 + 가상 SQL/Python 예시
- 마지막 검증일: 2026-07-16
- 검증 환경: PostgreSQL current 문서의 Read Committed·ON CONFLICT 개념, Redis SET·TTL 개념
- 실행 주의: Driver placeholder 문법·interval binding·JSON serialization은 선택한 async PostgreSQL library에 맞춰 조정
- 수치 표기: lease·TTL·cache 300초는 설명용이며 task duration·recovery SLO·retention으로 결정
- 경험 표기: tenant·session·run ID와 response는 실제 운영 정보가 아닌 가상 예시