실패 후 다시 처리

실패가 절반의 새 데이터가 되는 순간

A의 통행량을 수정한 뒤 C의 새 관측을 추가하는 배치를 생각합니다. 첫 행을 저장하고 두 번째 행에서 예외가 발생하면 보고서는 일부 정정만 반영한 데이터로 계산될 수 있습니다. 담당자가 실패 메시지를 보고도 첫 행이 이미 저장되었는지 알 수 없다면 같은 배치를 다시 실행할 판단이 어려워집니다. 이번 레슨은 예외가 생긴 위치와 저장이 공개되는 경계를 직접 확인합니다. 정상 결과만 확인하는 대신 의도적으로 첫 쓰기 직후 예외를 넣어 복구 계약을 검사합니다.

트랜잭션은 배치의 여러 DB 쓰기를 하나의 확정 단위로 묶습니다. BEGIN 뒤에 변경하고, 모든 행이 성공하면 COMMIT, 중간 실패면 ROLLBACK을 실행합니다. 이번 작은 증분 배치는 입력 전체를 한 트랜잭션으로 처리합니다. 각 행마다 커밋하면 앞 행이 이미 확정되어 뒤 행 실패 때 취소할 수 없습니다. 입력 건수가 커져 나눠 저장해야 한다면 각 조각의 공개 상태와 체크포인트를 설계해야 합니다. 지금은 그런 계약이 없으므로 성능을 이유로 행마다 커밋하지 않습니다.

명시적인 시작과 끝을 코드로 읽습니다

실습은 sqlite3.connect(database, isolation_level=None)로 연결하고 BEGIN IMMEDIATE를 직접 실행합니다. 이는 자동으로 시작되는 트랜잭션과 수동 시작을 섞지 않기 위한 선택입니다. BEGIN IMMEDIATE는 쓰기 트랜잭션을 시작하므로 다른 쓰기 작업이 이미 잠금을 유지하면 대기하거나 database is locked로 실패할 수 있습니다. 이 오류는 불량 날짜나 중복키 오류와 다릅니다. 트랜잭션 범위와 다른 작업의 연결 종료 여부를 확인한 뒤 재시도 여부를 판단합니다. 실습은 단일 실행을 기본 계약으로 둡니다.

try 블록은 before 스냅샷을 읽고 행을 순서대로 UPSERT한 뒤 after를 읽어 COMMIT합니다. except 블록은 ROLLBACK 후 raise로 원래 실패를 호출자에게 전달합니다. 예외를 출력만 하고 정상 반환하면 셸은 성공이라고 판단할 수 있습니다. 데이터 복구와 실패 전달은 둘 다 완료되어야 합니다. finally나 contextlib.closing은 연결을 닫는 역할을 하며 실패를 자동으로 성공으로 바꾸지 않습니다. 명시적 커밋 이전의 값과 이후 다른 연결에서 읽는 값을 구분하여 관찰합니다.

테이블 준비는 배치 밖에서 수행합니다. 신규 DB에서 실패 주입이 일어나면 빈 observations 테이블 자체는 남을 수 있지만 미완성 관측 행은 남지 않습니다. 기존 DB의 실패에서는 이전에 확정된 행이 유지됩니다. 따라서 DB 파일이 생겼다는 사실만으로 데이터가 공개되었다고 판단하지 않습니다. 완료 기준은 논리적 행 상태입니다. 파일 바이트 비교는 SQLite의 메타데이터 변화나 저장 방식 때문에 논리적 복구의 유일한 증거로 쓰지 않습니다. 정렬된 전체 행과 키, 합계를 조회해 전후를 대조합니다.

실패를 재현할 지점을 고릅니다

fail_after=1은 첫 번째 UPSERT가 실행된 직후 RuntimeError를 발생시킵니다. 스키마 검사 전에 예외를 넣으면 DB가 아직 쓰이지 않아 트랜잭션이 잘못되어도 테스트가 통과할 수 있습니다. 실제 쓰기 이후에 실패를 주입해야 부분 저장을 잡습니다. 오류 메시지 INJECTED: after row 1은 테스트용 장애임을 표시합니다. 날짜 검증의 SCHEMA 오류와 구분하며, 이 플래그를 사용한 실행을 정상 운영 성공으로 기록하지 않습니다. 실패 주입은 교육용 작업 사본에서 수행합니다.

롤백 테스트는 A=100, B=120의 기존 DB를 준비합니다. 다음 배치는 A=105와 새 C=10입니다. 첫 쓰기 후 실패시키면 A는 다시 100이고 C는 없습니다. 옵션을 제거하고 같은 배치를 재시도하면 A=105, B=120, C=10으로 합계 235가 됩니다. 앞서 처리되지 않았던 B가 보존되는지도 함께 확인합니다. 같은 예제라도 초기 상태가 빈 DB뿐이면 기존 값 복구를 확인할 수 없으므로 두 준비 상태를 별도 테스트로 둡니다.

전후 행 수가 같다는 검사만으로 롤백을 확인하지 않습니다. 첫 변경이 UPDATE라면 행 수는 성공·실패 어느 경우에도 2입니다. 합계와 전체 행을 함께 비교해야 정정된 값이 남았는지 확인됩니다. 합계만 비교해도 두 값이 반대로 바뀌는 실패를 놓칠 수 있으므로 rows의 비교가 최종 근거가 됩니다. 감사 스냅샷은 키 정렬 순서와 NULL 개수를 포함합니다. 미션의 실패 fixture에서는 첫 A 값을 크게 바꿔 부분 반영을 쉽게 드러내지만 테스트는 차이 크기에 의존하지 않습니다.

다시 처리할 수 있는 이유를 설명합니다

재시도는 실패 이전의 입력을 그대로 다시 처리하는 것입니다. 롤백으로 이전 상태를 보존하고 UPSERT가 절대값 대입을 사용하므로, 이전 시도가 어디까지 진행되었든 정상 재실행 후 관측 값은 한 번 성공한 것과 같습니다. 성공 응답이 전달되지 않았을 때도 같은 입력으로 확인할 수 있습니다. 다만 저장 함수 외부에서 보낸 알림이나 다른 파일 쓰기는 DB 롤백으로 취소되지 않습니다. 이번 모듈은 그런 외부 부작용을 미션의 성공 데이터 공개 경로에 넣지 않습니다.

모든 실패를 자동 재시도하면 잘못된 입력도 계속 처리하게 됩니다. DATE, HEADER, KEY_CONFLICT처럼 데이터 계약을 어긴 오류는 원본과 공급자 조건을 확인해 수정한 뒤 실행합니다. 일시적인 잠금 오류는 다른 실행이 끝났는지 확인하고 제한된 횟수로 다시 시도할 수 있습니다. 예외 이름과 메시지, 입력 해시를 함께 남기면 어떤 파일의 어떤 실패인지 찾을 수 있습니다. 단순히 시간만 바꿔 재시작하는 것은 오류 원인을 해결했다는 증거가 아닙니다.

DB 밖의 산출물은 별도 상태입니다

앞 모듈 transform은 etl-out에 정제 CSV와 감사 파일을 씁니다. 이 파일은 DB 트랜잭션의 일부가 아니므로 저장 실패 뒤에도 새로운 변환 결과가 존재할 수 있습니다. 미션은 보고서가 사용하는 성공 자료를 커밋된 observations로 한정합니다. 중간 파일을 최신 보고서 입력으로 직접 읽으면 DB 롤백이 성공해도 실패 데이터가 공개될 수 있습니다. 파일 이름이 out이라는 이유로 성공을 의미한다고 생각하지 않고 각 경로의 역할을 인계서에 적습니다.

run-audit.json은 성공 또는 실패 시도에 대한 기록이며 observations의 성공 데이터와 역할이 다릅니다. DB 커밋과 JSON 파일 교체를 하나의 원자적 작업이라고 주장하지 않습니다. 커밋 후 JSON 기록 저장이 실패하면 명령은 실패하지만 DB는 이미 바뀌었을 수 있습니다. 이 경우 DB 상태를 먼저 조회하고 동일 입력을 재처리합니다. 운영으로 확장할 때는 실행 기록을 DB 안에서 함께 커밋하고 파일은 그 기록에서 내보내는 설계를 고려합니다. 이번 실습은 단일 DB 배치 원자성과 그 경계의 한계를 함께 학습합니다.

오류 메시지를 요구와 연결합니다

starter는 자동 커밋 상태에서 각 SQL을 바로 실행하므로 INJECTED 예외가 나도 앞 행이 남습니다. test_mid_failure_and_retry의 AssertionError는 before와 state의 차이를 보여 줍니다. 기대 합계를 바꾸어 테스트를 통과시키는 대신 명시적인 배치 트랜잭션을 복원합니다. test_failed_new_database_has_no_rows는 새 저장소에서 미완성 행이 없는지, test_correction_to_null은 정정된 결측이 남는지 확인합니다. NULL 정정도 배치의 정상 데이터이며 오류로 만들어 제외하지 않습니다.

이 레슨을 마치면 성공 배치, 기존 DB 실패, 신규 DB 실패, 같은 입력 재시도의 상태를 각각 설명할 수 있어야 합니다. 커밋 횟수를 줄였다는 설명만으로 완료하지 않고 실패 전후 전체 행을 증거로 제시합니다. 격리수준과 여러 DB의 차이는 더 읽기의 트랜잭션 장에서 확장합니다. 여기서는 SQLite 예제에서 실제로 시작·확정·취소한 범위와 파일에 남는 중간 산출물의 범위를 분명히 말하는 연습에 집중합니다.

연결과 트랜잭션 설정은 Python sqlite3 공식 문서로 확인합니다.

import sqlite3
from contextlib import closing
with closing(sqlite3.connect(':memory:',isolation_level=None)) as db:
    db.execute('CREATE TABLE traffic(region TEXT PRIMARY KEY,n INTEGER)')
    db.execute("INSERT INTO traffic VALUES('A',100)")
    db.execute('BEGIN IMMEDIATE')
    try:
        db.execute("UPDATE traffic SET n=105 WHERE region='A'")
        raise RuntimeError('INJECTED: after row 1')
    except RuntimeError as e:
        db.execute('ROLLBACK')
        print(str(e))
    print(db.execute('SELECT * FROM traffic ORDER BY region').fetchall())

INJECTED: after row 1
[('A', 100)]

import sqlite3
from contextlib import closing
with closing(sqlite3.connect(':memory:',isolation_level=None)) as db:
    db.execute('CREATE TABLE traffic(region TEXT PRIMARY KEY,n INTEGER)')
    db.executemany('INSERT INTO traffic VALUES(?,?)',[('A',100),('B',120)])
    db.execute('BEGIN IMMEDIATE')
    db.executemany('INSERT INTO traffic VALUES(?,?) ON CONFLICT(region) DO UPDATE SET n=excluded.n',[('A',105),('C',10)])
    db.execute('COMMIT')
    print(db.execute('SELECT * FROM traffic ORDER BY region').fetchall())
    print('sum',db.execute('SELECT SUM(n) FROM traffic').fetchone()[0])

[('A', 105), ('B', 120), ('C', 10)]
sum 235

before=[('A',100),('B',120)]
partial=[('A',105),('B',120)]
print('same_count',len(before)==len(partial))
print('same_rows',before==partial)
print('same_sum',sum(r[1] for r in before)==sum(r[1] for r in partial))

same_count True
same_rows False
same_sum False

python3 -m unittest discover -s tests -v

python3 -m unittest discover -s tests -v