백업 파일이 있다는 것만으로 최근 글이 들어 있다고 말할 수는 없었다. K-POP PULSE의 운영 기록은 WAL을 고려한 백업 절차를 남겼다. 나는 이 원리를 새 합성 DB에서 직접 재현하고 다른 사이트에서도 사용할 수 있는 작은 로컬 프로그램을 만들었다.
원본 운영 프로젝트의 백업 스크립트를 복원한 공유물이 아니다. 새 코드는 SQLite 백업 API로 새 파일을 만들고 그 대상의 무결성과 행수를 읽는다. 실제 원본 사이트의 복원 시험과 구별하며, 전체 사이트를 복구하려면 무엇을 더 보존해야 하는지도 함께 남긴다.
살아 있는 DB의 주 파일만 복사한 결과를 비교했다
새 검사 DB에서 기준 행을 저장하고 체크포인트를 완료한 뒤 연결을 열어 둔 채 두 번째 행을 커밋했다. WAL 파일이 실제로 존재하는 상태에서 주 파일만 복사한 사본은 첫 행만 읽혔다. 그 뒤 백업 API로 만든 사본에서는 두 번째 행의 내용까지 다시 읽었다.
이 재현은 새 합성 자료를 사용한 현재 검사다. 실제 K-POP 운영 DB의 최근 글이 유실됐다는 사고로 소개하지 않는다. 같은 DB에서 관측한 두 방식의 결과를 직접 비교하므로 단순히 파일 개수나 크기를 보고 백업 성공으로 판단하는 문제와 구별할 수 있다.
기존 목적지를 덮어쓰지 않는 입력 계약을 둔다
소스는 존재하는 로컬 SQLite 파일이어야 하고 출력은 아직 없는 새 파일이어야 한다. 같은 파일, 이미 있는 출력과 없는 소스는 거부한다. 출력 폴더를 자동으로 만들지 않고 지정한 파일을 독점 생성 방식으로 예약하므로 기존 파일을 지우며 시작하지 않는다.
원본 DB 행은 읽기 전용 연결로 읽지만 SQLite 환경에 따라 WAL과 공유 메모리 보조 파일에 접근하거나 생성할 수 있다. 읽기 전용이라는 설정을 파일시스템의 어떤 변화도 없다는 약속으로 넓히지 않는다. 네트워크 파일시스템이나 확장 DB의 적용 조건은 이 작은 검사로 인증하지 않는다.
새 사본의 연결을 닫고 단독 파일로 다시 읽는다
새 대상만 단독 파일에 맞는 저널 모드로 바꾼다. 원본의 모드와 데이터를 바꾸는 작업이 아니다. 사본의 integrity_check를 실행하고 사용자 테이블의 행수와 파일 바이트 해시를 결과로 남긴다. 실제 검사에서는 연결을 닫은 뒤 사본을 열어 예상 행과 본문을 확인했다.
무결성 검사가 통과해도 업무 내용이 정확하다는 뜻은 아니다. 내가 보존하려던 글과 자료가 있는지는 필요한 테이블과 내용을 따로 대조해야 한다. 행수가 같아도 문장이 바뀔 수 있으므로 서비스를 복구할 때는 주소와 본문, 연결된 자산까지 그 서비스의 계약으로 확인한다.
오류와 완료를 다른 상태로 반환한다
진행 중 출력은 아직 완성되지 않은 파일일 수 있다. 종료 0과 ok=true까지 확인해야 성공으로 읽는다. 입력 오류와 DB 오류는 일반 오류 JSON과 종료 2로 끝나며 실패한 호출이 새로 만든 출력과 보조 파일만 정리한다. 기존 목적지의 바이트는 유지되는지 검사했다.
기본 백업 진행 한도는 30초이며 필요하면 최대 3600초까지 지정한다. 해시와 추가 검사를 포함한 프로그램 전체 시간의 제한으로 주장하지 않는다. 큰 파일이나 특수 테이블의 검사는 자신의 환경에서 따로 맞추며 오류를 없앤다고 검사 조건을 낮추지 않는다.
DB 사본과 전체 홈페이지 복구를 구별한다
백업 진행의 읽기 시점 스냅샷이므로 그 뒤 커밋이나 미커밋 변경을 모두 포함하는 것은 아니다. 외부 이미지와 영상, 프로그램 설정과 기타 파일은 별도 보존 대상이다. 여러 DB를 각각 백업했다고 같은 순간의 원자적 사본이 된다고 가정하지 않는다.
나는 DB 파일 생성을 전체 복원 완료라는 말로 대신하지 않으려 한다. 공유 프로그램을 사용한 사이트에서는 필요한 파일과 서비스 연결을 함께 목록으로 만들고 복원 사본에서 실제 조회와 화면 흐름을 확인한다. 이번 검사는 원본 사이트를 복원하거나 운영 데이터를 변경하지 않았다.
검사와 적용 설명까지 함께 가져간다
공유물은 Python 표준 라이브러리만 사용하는 새 CLI다. --source와 --destination으로 자신의 로컬 파일을 지정한다. 처음에는 본인 테스트 DB와 새 경로로 실행해 권한과 저널 환경을 확인한다. 네트워크로 자료를 전송하거나 클라우드 운영 DB를 복원하지 않는다.
새 합성 DB에서 WAL 활성, 주 파일 복사의 누락, 백업 API 사본 내용, 단독 재열기, 기존 파일 보호와 실제 CLI를 포함한 17개 검사가 통과했다. 아래 구현과 ZIP의 검사 코드·적용 설명을 함께 공유한다. 다른 DB와 미디어를 연결할 때도 보존하려던 내용이 실제로 읽히는지 확인하며 확장한다.
코드와 구현 살펴보기
새 로컬 스냅샷 구현. 자신의 테스트 DB에서 적용 조건을 확인한다.
"""Create a new local SQLite snapshot through its backup API; never overwrite a destination."""
import argparse,hashlib,json,os,sqlite3,sys,time
from contextlib import closing
from pathlib import Path
def sqlite_snapshot(source,destination,*,timeout_seconds=30):
source=Path(source).resolve(strict=True);destination=Path(destination).absolute()
if not source.is_file() or source==destination.resolve():raise ValueError('Distinct database file required')
if isinstance(timeout_seconds,bool) or not isinstance(timeout_seconds,(int,float)) or not 0<timeout_seconds<=3600:
raise ValueError('Bounded positive timeout required')
if destination.exists() or destination.is_symlink():raise FileExistsError('Destination already exists')
if not destination.parent.is_dir():raise ValueError('Destination directory must exist')
# Reserve the exact destination before SQLite opens it, so an existing file is never truncated.
fd=os.open(destination,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600);os.close(fd)
began=time.monotonic();complete=False
try:
with closing(sqlite3.connect(source.as_uri()+'?mode=ro',uri=True,timeout=min(timeout_seconds,5))) as src:
with closing(sqlite3.connect(destination)) as dst:
def progress(status,remaining,total):
if time.monotonic()-began>timeout_seconds:raise TimeoutError('Snapshot time limit')
src.backup(dst,pages=128,progress=progress,sleep=0.05)
# Change only the new destination to standalone rollback-journal mode.
if dst.execute('PRAGMA journal_mode=DELETE').fetchone()[0]!='delete':raise ValueError('Standalone snapshot mode unavailable')
integrity=dst.execute('PRAGMA integrity_check').fetchall()
if integrity!=[('ok',)]:raise ValueError('Snapshot integrity check failed')
tables=[r[0] for r in dst.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")]
counts={name:dst.execute('SELECT count(*) FROM "'+name.replace('"','""')+'"').fetchone()[0] for name in tables}
# Close both handles before treating the new standalone database as complete.
src.close();dst.close()
digest=hashlib.sha256()
with destination.open('rb') as stream:
for chunk in iter(lambda:stream.read(1024*1024),b''):digest.update(chunk)
complete=True
return {'ok':True,'bytes':destination.stat().st_size,'sha256':digest.hexdigest(),'table_rows':counts,'integrity':'ok','sqlite_version':sqlite3.sqlite_version,'scope':'New local database snapshot at its backup reading, not all future commits, external media backup or production restore.'}
finally:
# Handles are closed before cleanup, including exceptional paths.
if 'src' in locals():src.close()
if 'dst' in locals():dst.close()
if not complete:
for suffix in ['', '-journal','-wal','-shm']:
candidate=Path(str(destination)+suffix)
if candidate.exists():candidate.unlink()
def main():
p=argparse.ArgumentParser();p.add_argument('--source',type=Path,required=True);p.add_argument('--destination',type=Path,required=True);p.add_argument('--timeout',type=float,default=30);a=p.parse_args()
try:result=sqlite_snapshot(a.source,a.destination,timeout_seconds=a.timeout)
except (OSError,ValueError,sqlite3.Error,TimeoutError):
print(json.dumps({'ok':False,'error':'Snapshot unavailable; inspect source access, destination and input contract'}));return 2
print(json.dumps(result,ensure_ascii=False));return 0
if __name__=='__main__':raise SystemExit(main())