창이 사라져도 남겨야 할 작업 상태와 파일이 화면에서 빠져도 남을 수 있는 참조를 함께 읽고 싶었다. 그래서 진행 메타데이터와 자원 참조를 보관하는 작은 SQLite 실험을 만들었다.
실제 회사 파일이나 작업 실행은 없다. 테스트가 만드는 임시 DB에서 재개·중복·오래된 작업자·참조 보호를 확인하고 코드와 실행 기록을 공유한다.
진행 위치와 작업자 세대의 계약
작업에는 회사, 작업 ID, 전체 범위, cursor, generation, worker와 상태가 있다. create_job으로 만든 뒤 claim이 세대를 올리고 실행 중으로 바꾼다. advance는 예상 위치와 세대가 맞을 때만 진행한다. 실패 뒤 다른 작업자가 새 세대로 가져가면 이전 작업자의 요청은 거절된다. 같은 묶음 token과 내용은 이미 계산한 위치를 다시 돌려준다.
이것은 실행기나 분산 임대 서비스가 아니다. 진행 중 작업을 자동으로 회수하는 시간 제한, heartbeat, 실제 업무 행의 저장과 인증을 구현하지 않았다. 처리 위치가 남았다고 회사 자료까지 그만큼 확정됐다는 뜻도 아니다. 운영 연결에서는 업무 행 쓰기와 진행 갱신의 원자적 경계를 별도로 설계해야 한다.
파일 내용과 문서 관계를 다른 기록으로 둔다
register_resource는 합성 바이트의 지문과 길이만 보관한다. 원본 바이트의 객체 저장은 없다. 같은 회사 안의 같은 내용은 기존 메타데이터 ID를 돌려주고 다른 회사의 등록은 분리한다. 검토 전에는 연결할 수 없으며 current 참조가 있는 자원의 검토 해제도 거절한다. 파일 중복과 사용 가능 상태가 서로 다른 질문이라는 계약이다.
refs의 current·history·backup은 각각 유지할 이유다. 현재 문서에서 연결을 빼도 나머지 참조가 있으면 보관 대상으로 남는다. 같은 연결을 반복해도 고유 조건으로 행이 늘지 않는다. 실제 서버는 문서 소유 회사와 접근 권한을 별도로 확인해야 한다. 이 코드가 받는 식별자는 신뢰할 합성 문맥이지 외부 사용자의 인증 결과가 아니다.
미참조는 삭제 명령이 아니라 수동 검토 후보
resource_preview는 알려진 참조가 있으면 retain, 없으면 manual_review_only를 돌려주며 deleted는 항상 false다. 원본 삭제 함수는 없다. 모델 밖의 참조를 모를 수 있기 때문에 0이라는 숫자만으로 모든 사용처가 사라졌다고 결론 내리지 않는다. 보호해야 할 것과 추가로 확인할 것을 나누는 목록이 결과다.
SQLite는 이 실험의 메타데이터 저장소이며 기존에 모르는 DB를 넣으면 쓰기 전에 거절한다. 실제 회사 ERP의 형식과 호환되는 도구는 아니다. 회사·작업자·문서의 입력 범위, 원본 보관, 객체 저장·복구·보관 기간과 실제 정리는 운영 요구에 맞춰 추가해야 한다. 합성 예제의 작은 편의를 전체 파일 관리 서비스로 부르지 않는다.
코드와 구현 살펴보기
독립 합성 전체 구현
"""Synthetic jobs and resource-reference metadata. No upload, sending or deletion.
Company and worker are trusted test context, not authenticated request fields.
Resource bytes are hashed but NOT stored: this is not an object-storage service.
"""
import sqlite3,json,hashlib,re
from pathlib import Path
from contextlib import closing,contextmanager
def ident(v):return isinstance(v,str) and re.fullmatch('[A-Za-z0-9_-]{1,40}',v) is not None
class WorkLab:
def __init__(self,path):
self.path=str(path);exists=Path(path).exists() and Path(path).stat().st_size>0
with closing(self.connect()) as db:
if exists:
if not db.execute("SELECT name FROM sqlite_master WHERE name='synthetic_work_marker'").fetchone() or db.execute('SELECT version FROM synthetic_work_marker').fetchall()[0][0]!='synthetic-work-v1':raise ValueError('Unrecognized synthetic store')
else:
db.executescript('''CREATE TABLE synthetic_work_marker(version TEXT);INSERT INTO synthetic_work_marker VALUES('synthetic-work-v1');
CREATE TABLE jobs(company_id TEXT,id TEXT,total INTEGER,cursor INTEGER,generation INTEGER,worker TEXT,state TEXT,error_code TEXT,PRIMARY KEY(company_id,id));
CREATE TABLE chunks(company_id TEXT,job_id TEXT,token TEXT,fingerprint TEXT,result_cursor INTEGER,PRIMARY KEY(company_id,job_id,token));
CREATE TABLE resources(company_id TEXT,id TEXT,digest TEXT,byte_size INTEGER,reviewed INTEGER,PRIMARY KEY(company_id,id),UNIQUE(company_id,digest));
CREATE TABLE refs(company_id TEXT,resource_id TEXT,document_id TEXT,kind TEXT,PRIMARY KEY(company_id,resource_id,document_id,kind));''')
def connect(self):
db=sqlite3.connect(self.path,timeout=5,isolation_level=None);db.row_factory=sqlite3.Row;return db
@contextmanager
def transaction(self):
with closing(self.connect()) as db:
db.execute('BEGIN IMMEDIATE')
try:yield db;db.commit()
except BaseException:db.rollback();raise
def create_job(self,company,job,total):
if not ident(company) or not ident(job) or type(total) is not int or not 0<=total<=1000:raise ValueError('Invalid job')
with self.transaction() as db:db.execute('INSERT INTO jobs VALUES(?,?,?,0,0,NULL,?,NULL)',[company,job,total,'done' if total==0 else 'queued'])
def read_job(self,company,job):
if not ident(company) or not ident(job):return None
with closing(self.connect()) as db:
row=db.execute('SELECT * FROM jobs WHERE company_id=? AND id=?',[company,job]).fetchone();return dict(row) if row else None
def claim(self,company,job,worker,expected_generation):
if not all(ident(v) for v in [company,job,worker]) or type(expected_generation) is not int:return None
with self.transaction() as db:
row=db.execute('SELECT * FROM jobs WHERE company_id=? AND id=?',[company,job]).fetchone()
if not row or row['state'] not in ['queued','failed'] or row['generation']!=expected_generation:return None
generation=row['generation']+1;db.execute("UPDATE jobs SET generation=?,worker=?,state='running',error_code=NULL WHERE company_id=? AND id=?",[generation,worker,company,job]);return generation
def advance(self,company,job,worker,generation,expected_cursor,token,count):
if not all(ident(v) for v in [company,job,worker,token]) or any(type(v) is not int for v in [generation,expected_cursor,count]) or count<1:return {'ok':False,'reason':'input'}
fingerprint=hashlib.sha256(json.dumps([expected_cursor,count]).encode()).hexdigest()
with self.transaction() as db:
row=db.execute('SELECT * FROM jobs WHERE company_id=? AND id=?',[company,job]).fetchone()
if not row or row['worker']!=worker or row['generation']!=generation:return {'ok':False,'reason':'stale_worker'}
old=db.execute('SELECT * FROM chunks WHERE company_id=? AND job_id=? AND token=?',[company,job,token]).fetchone()
if old:return {'ok':True,'cursor':old['result_cursor'],'replayed':True} if old['fingerprint']==fingerprint else {'ok':False,'reason':'token_reused'}
if row['state']!='running' or row['cursor']!=expected_cursor or row['cursor']+count>row['total']:return {'ok':False,'reason':'cursor_conflict'}
next_cursor=row['cursor']+count;db.execute('INSERT INTO chunks VALUES(?,?,?,?,?)',[company,job,token,fingerprint,next_cursor]);db.execute('UPDATE jobs SET cursor=?,state=? WHERE company_id=? AND id=?',[next_cursor,'done' if next_cursor==row['total'] else 'running',company,job]);return {'ok':True,'cursor':next_cursor,'replayed':False}
def fail(self,company,job,worker,generation):
with self.transaction() as db:
return db.execute("UPDATE jobs SET state='failed',error_code='synthetic_failure' WHERE company_id=? AND id=? AND worker=? AND generation=? AND state='running'",[company,job,worker,generation]).rowcount==1
def register_resource(self,company,resource_id,data):
if not ident(company) or not ident(resource_id) or not isinstance(data,bytes) or not 0<len(data)<=100000:raise ValueError('Invalid resource metadata input')
digest=hashlib.sha256(data).hexdigest()
with self.transaction() as db:
old=db.execute('SELECT id FROM resources WHERE company_id=? AND digest=?',[company,digest]).fetchone()
if old:return {'id':old['id'],'reused':True}
db.execute('INSERT INTO resources VALUES(?,?,?,?,0)',[company,resource_id,digest,len(data)]);return {'id':resource_id,'reused':False}
def review_resource(self,company,resource_id,reviewed):
if type(reviewed) is not bool:raise ValueError('Review flag required')
with self.transaction() as db:
if not reviewed and db.execute("SELECT 1 FROM refs WHERE company_id=? AND resource_id=? AND kind='current'",[company,resource_id]).fetchone():return False
return db.execute('UPDATE resources SET reviewed=? WHERE company_id=? AND id=?',[int(reviewed),company,resource_id]).rowcount==1
def link(self,company,resource_id,document_id,kind):
if not all(ident(v) for v in [company,resource_id,document_id]) or kind not in ['current','history','backup']:return False
with self.transaction() as db:
row=db.execute('SELECT reviewed FROM resources WHERE company_id=? AND id=?',[company,resource_id]).fetchone()
if not row or not row['reviewed']:return False
db.execute('INSERT OR IGNORE INTO refs VALUES(?,?,?,?)',[company,resource_id,document_id,kind]);return True
def unlink(self,company,resource_id,document_id,kind):
with self.transaction() as db:return db.execute('DELETE FROM refs WHERE company_id=? AND resource_id=? AND document_id=? AND kind=?',[company,resource_id,document_id,kind]).rowcount==1
def resource_preview(self,company):
with closing(self.connect()) as db:
rows=db.execute('SELECT r.*,count(f.resource_id) AS ref_count FROM resources r LEFT JOIN refs f ON f.company_id=r.company_id AND f.resource_id=r.id WHERE r.company_id=? GROUP BY r.company_id,r.id ORDER BY r.id',[company]).fetchall()
return [{**dict(row),'disposition':'retain' if row['ref_count'] else 'manual_review_only','deleted':False} for row in rows]기록을 읽고 같은 결과를 다시 만든다
아래 뷰어는 재개 결과, 회사별 같은 내용의 등록, 이력과 백업 참조의 보호를 보여 준다. 자료는 run_examples.py가 실제 임시 DB에서 생성한 JSON이며 선택할 때 DB 명령을 실행하지 않는다. 브라우저에서 Python을 실행하는 기능도 아니다. 자료가 어떻게 나왔는지는 패키지의 실행기를 읽고 직접 재현할 수 있다.
ZIP에는 구현·테스트·실행기·결과 JSON과 범위 문서가 있다. 압축을 풀고 python test_work.py, python run_examples.py를 실행한다. 실제 회사 DB나 파일을 입력할 필요가 없고 외부 라이브러리도 설치하지 않는다. 수정·재배포·상업적 사용이 가능하며 무보증이다. 새로운 독립 합성 코드로 작성했으며 고객 프로그램을 그대로 추출한 패키지가 아니다.
확인한 조건과 연결해야 할 조건
테스트15개는 재개 뒤 위치 보존, 경쟁하는 작업자, 중복 묶음, 잘못된 위치와 초과 진행, 세대 교체 뒤 차단을 확인한다. 같은 회사의 내용 재사용과 다른 회사의 분리, 검토 전 연결 거절, 반복 연결과 이력·백업 참조 보호도 검사했다. 압축에서 꺼낸 파일로 같은 테스트를 실행했으며 실제 운영 자료나 파일 삭제는 없다.
실제 브라우저·기기 검수나 운영 보안 감사를 수행한 결과는 아니다. 실무에 연결할 때는 신원·권한, 실제 작업의 원자성, 실행기와 재개 규칙, 객체 원본과 참조 목록의 일치를 확인해야 한다. 나는 구현을 작은 경계로 나누고 그 경계가 지키는 것과 지키지 않는 것을 코드·검증 기록과 함께 넘기는 방식으로 공유한다.