Resume imports from the last committed checkpoint

한국어 원문으로 표시합니다.

목표

원본 지문과 체크포인트를 결합해 다른 파일로 잘못 이어받는 일을 막습니다.

왜 중요한가

수천 행을 가져오던 작업이 중간에 죽었다. 운영자가 파일을 바꿔 같은 작업 id로 다시 실행하자 앞부분은 옛 파일, 뒷부분은 새 파일인 결과가 생겼다. 처리한 행 번호만 저장하면 입력의 정체성을 확인할 수 없다. 원본 바이트의 지문과 마지막으로 커밋한 위치를 함께 보존해야 한다.

단계

/root/work/idem-batch-checkpoint-lab/service.py에서 parse_rows(raw)는 JSON 배열 bytes를 읽습니다. 각 항목은 id(비어 있지 않은 str)와 value(bool 제외 int)를 가지며 id 중복은 금지합니다. 위반은 ValueError입니다. {id,value}만 가진 행 리스트를 반환합니다.

처음 한 번 준비하세요. 기존 파일은 덮어쓰지 않습니다.

mkdir -p /root/work/idem-batch-checkpoint-lab
test -e /root/work/idem-batch-checkpoint-lab/service.py || cp /opt/fixtures/ten_labs/idem-batch-checkpoint-lab/service.py /root/work/idem-batch-checkpoint-lab/service.py
cd /root/work/idem-batch-checkpoint-lab

/root/work/idem-batch-checkpoint-lab/service.py에서 source_digest(raw)는 bytes에 SHA-256을 적용한 hex 문자열입니다. JSON을 정규화하지 않습니다.
/root/work/idem-batch-checkpoint-lab/service.py에서 init_db(path)는 imports(id TEXT PRIMARY KEY,digest TEXT NOT NULL,next_index INTEGER NOT NULL)와 items(batch TEXT NOT NULL,id TEXT NOT NULL,value INTEGER NOT NULL,PRIMARY KEY(batch,id))를 멱등 생성합니다.
/root/work/idem-batch-checkpoint-lab/service.py에서 begin(path,batch,digest)는 새 작업이면 next_index=0을 저장해 0, 기존 같은 지문이면 next_index, 다른 지문이면 ValueError입니다.
/root/work/idem-batch-checkpoint-lab/service.py에서 apply_chunk(path,batch,start,rows,fault=lambda index:None)는 현재 next_index==start일 때만 실행하며 아니면 ValueError입니다. rows를 순서대로 items에 넣고 각 삽입 후 fault(전체인덱스)를 호출합니다. 전부 성공하면 next_index=start+len(rows)를 저장해 반환합니다.
/root/work/idem-batch-checkpoint-lab/service.py에서 checkpoint(path,batch)는 next_index, 없는 작업은 None입니다.
/root/work/idem-batch-checkpoint-lab/service.py에서 values(path,batch)는 해당 batch의 (id,value) 튜플을 id 오름차순으로 반환합니다.
/root/work/idem-batch-checkpoint-lab/service.py에서 import_all(path,batch,raw,size=2,fault=lambda index:None)는 bool 제외 양의 int size를 검증하고 parse_rows·source_digest·begin을 사용합니다. 남은 행을 size씩 apply_chunk로 처리하고 최종 checkpoint를 반환합니다.

참고

인터넷과 패키지 설치 없이 기존 lab-dev 환경에서 수행합니다.
각 단계는 45초 채점 예산 안에서 실행됩니다. 실제 sleep이나 네트워크 호출을 추가하지 마세요.
채점은 제출 모듈을 새로 불러오고 독립 입력과 임시 DB로 검사합니다. 예상값을 상수로 반환하는 대신 계약을 구현하세요.
FastAPI 공식 문서 · pytest 공식 문서 · Python sqlite3
한계: 입력 전체를 메모리에 읽는 작은 데이터용 실습이다. 대용량 파일을 스트리밍 파싱하는 엔진으로 과장하지 않는다. 원본 공백만 달라도 바이트 지문이 달라 재개를 거절한다는 보수적 계약이다. 외부 API 부작용은 이 DB 트랜잭션에 들어가지 않는다.

입력 행의 계약을 확인한다

/root/work/idem-batch-checkpoint-lab/service.py에서 parse_rows(raw)는 JSON 배열 bytes를 읽습니다. 각 항목은 id(비어 있지 않은 str)와 value(bool 제외 int)를 가지며 id 중복은 금지합니다. 위반은 ValueError입니다. {id,value}만 가진 행 리스트를 반환합니다.

처음 한 번 준비하세요. 기존 파일은 덮어쓰지 않습니다.

mkdir -p /root/work/idem-batch-checkpoint-lab
test -e /root/work/idem-batch-checkpoint-lab/service.py || cp /opt/fixtures/ten_labs/idem-batch-checkpoint-lab/service.py /root/work/idem-batch-checkpoint-lab/service.py
cd /root/work/idem-batch-checkpoint-lab

중복 id를 마지막 행으로 조용히 덮으면 가져오기 결과를 예측할 수 없습니다.

저장 후 bash /opt/lab/checks/idem-batch-checkpoint-lab/01-contract.sh로 확인하세요.

원본 바이트의 지문을 고정한다

/root/work/idem-batch-checkpoint-lab/service.py에서 source_digest(raw)는 bytes에 SHA-256을 적용한 hex 문자열입니다. JSON을 정규화하지 않습니다.

재개는 같은 원본에 대해서만 허용하는 계약입니다.

저장 후 bash /opt/lab/checks/idem-batch-checkpoint-lab/02-contract.sh로 확인하세요.

입력과 체크포인트를 저장한다

/root/work/idem-batch-checkpoint-lab/service.py에서 init_db(path)는 imports(id TEXT PRIMARY KEY,digest TEXT NOT NULL,next_index INTEGER NOT NULL)와 items(batch TEXT NOT NULL,id TEXT NOT NULL,value INTEGER NOT NULL,PRIMARY KEY(batch,id))를 멱등 생성합니다.

서로 다른 가져오기 작업의 같은 행 id는 분리해서 저장합니다.

저장 후 bash /opt/lab/checks/idem-batch-checkpoint-lab/03-contract.sh로 확인하세요.

다른 원본으로 재개하지 못하게 한다

/root/work/idem-batch-checkpoint-lab/service.py에서 begin(path,batch,digest)는 새 작업이면 next_index=0을 저장해 0, 기존 같은 지문이면 next_index, 다른 지문이면 ValueError입니다.

작업 id와 행 번호가 같아도 입력 파일이 다를 수 있습니다.

저장 후 bash /opt/lab/checks/idem-batch-checkpoint-lab/04-contract.sh로 확인하세요.

배치와 위치를 원자적으로 커밋한다

/root/work/idem-batch-checkpoint-lab/service.py에서 apply_chunk(path,batch,start,rows,fault=lambda index:None)는 현재 next_index==start일 때만 실행하며 아니면 ValueError입니다. rows를 순서대로 items에 넣고 각 삽입 후 fault(전체인덱스)를 호출합니다. 전부 성공하면 next_index=start+len(rows)를 저장해 반환합니다.

행 하나마다 별도 커밋하면 체크포인트와 행 상태가 어긋납니다.

저장 후 bash /opt/lab/checks/idem-batch-checkpoint-lab/05-contract.sh로 확인하세요.

현재 위치를 조회한다

/root/work/idem-batch-checkpoint-lab/service.py에서 checkpoint(path,batch)는 next_index, 없는 작업은 None입니다.

마지막 처리 시도 위치가 아니라 마지막으로 커밋된 위치를 읽습니다.

저장 후 bash /opt/lab/checks/idem-batch-checkpoint-lab/06-contract.sh로 확인하세요.

작업별 결과를 분리한다

/root/work/idem-batch-checkpoint-lab/service.py에서 values(path,batch)는 해당 batch의 (id,value) 튜플을 id 오름차순으로 반환합니다.

다른 작업의 동일 id 행이 결과에 섞이지 않게 batch를 조건으로 둡니다.

저장 후 bash /opt/lab/checks/idem-batch-checkpoint-lab/07-contract.sh로 확인하세요.

중간 실패 뒤 안전하게 이어간다

/root/work/idem-batch-checkpoint-lab/service.py에서 import_all(path,batch,raw,size=2,fault=lambda index:None)는 bool 제외 양의 int size를 검증하고 parse_rows·source_digest·begin을 사용합니다. 남은 행을 size씩 apply_chunk로 처리하고 최종 checkpoint를 반환합니다.

첫 배치의 성공을 보존한 채 두 번째 배치 실패 후 재개하는 시나리오를 확인합니다.

저장 후 bash /opt/lab/checks/idem-batch-checkpoint-lab/08-contract.sh로 확인하세요.