서로 다른 설명용 함수가 각각 맞아도 한 업무를 함께 처리할 때 어긋날 수 있다. 나는 승인과 변경 이력, 저장 충돌과 요청 재응답을 같은 작은 DB에서 시험하는 모듈을 만들었다.
실제 회사 자료와 계정, 외부 발송은 없다. 파이썬 표준 라이브러리만 사용하는 독립 합성 실험이며, 아래 뷰어는 실제 로컬 실행 기록을 보여 준다. 브라우저에서 DB를 실행하는 기능은 아니다.
두 종류의 버전과 승인 회차
문서에는 모든 저장 변화의 version과 내용 변경의 content_revision을 따로 둔다. 승인 회차는 어느 내용 버전을 누가 제출했는지 가리킨다. 제출·승인으로 저장 번호가 올라가도 내용 번호는 유지되고, 수량 수정은 내용 번호를 올린다. 과거 승인 기록을 현재 수량의 승인인 것처럼 재사용하지 않게 한 구조다.
합성 역할에서 직원은 자기 담당 문서를 수정·제출하고 담당자는 같은 회사 범위에서 확인·반려할 수 있다. 자기 승인은 거절하며 대표는 읽기만 한다. 이 정책은 조직별로 바꿀 수 있는 예시다. Actor는 서버가 인증한 문맥을 가정한 테스트 값이므로 외부 입력의 역할 필드를 믿고 이 모듈을 보안 계층으로 배포해서는 안 된다.
함께 남아야 할 기록을 같은 경계에 둔다
문서 갱신, 승인 변화, 변경 전후 사건, 후속 알림 대기 기록과 성공 요청 결과가 한 SQLite 쓰기 트랜잭션 안에 있다. BEGIN IMMEDIATE 뒤에 현재 버전을 읽어 예상 번호를 비교하고, 중간 오류가 나면 모두 되돌린다. 문서만 바뀌거나 이력만 남는 상태를 작은 합성 DB에서 재현하지 않도록 시험했다.
같은 회사·사용자·요청 키에는 명령 지문을 대조한다. 같으면 확정된 결과를 다시 주고 다르면 거절한다. 두 연결에서 같은 요청을 경쟁시켜 사건과 대기 행이 하나만 생기는지도 확인했다. 이 보장은 로컬 DB 안의 성공한 변경에 대한 것이다. 원격 전송의 처리 중 복구, 여러 서버의 분산 잠금이나 모든 외부 작업의 단 한 번 실행을 구현한 것은 아니다.
패키지의 함수와 실패 결과
WorkflowLab의 read와 history는 회사·담당 범위의 자료를 읽고 inbox는 담당자의 승인 대기를 보여 준다. execute는 edit·submit·approve·reject·execute 명령을 처리한다. 마지막 execute도 실제 전송이 아니라 합성 실행 상태다. 결과의 version_conflict, approval_stale, self_approval, request_key_reused는 서로 다른 이유로 다음 판단을 멈추게 한다.
알림 대기 기록은 실제 발송을 하지 않는다. delivery_result는 합성 전달 결과만 남기며 전달 여부 미확인은 자동 재시도하지 않는다. 보관된 요청 결과는 자동 만료되지 않고 잠금 오류도 무한 재시도하지 않는다. 운영 로그·신원 확인·보관 정책·외부 결과 조회·DB 형식 변경은 별도로 연결해야 할 책임이다. 이력은 이 API에서 덧붙일 뿐 DB 관리자에 대한 변조 방지 보장은 아니다.
코드와 구현 살펴보기
독립 합성 SQLite 전체 구현
"""Independent synthetic SQLite lab; no HTTP, real accounts, messages or ERP data.
Actor is a trusted test context, NOT authentication. Never trust client-supplied roles.
Results, document change, approval, event and outbox row commit in one local transaction.
SQLite's writer lock serializes this example; no distributed-system guarantee is implied.
"""
from dataclasses import dataclass
from pathlib import Path
import sqlite3,json,hashlib,re
from contextlib import closing
class ClosingConnection(sqlite3.Connection):
def __exit__(self,*args):
try:return super().__exit__(*args)
finally:self.close()
def packed(value):return json.dumps(value,sort_keys=True,separators=(',',':'),ensure_ascii=False,allow_nan=False)
def identifier(value):return isinstance(value,str) and re.fullmatch(r'[A-Za-z0-9_-]{1,40}',value) is not None
@dataclass(frozen=True)
class Actor:
company_id:str
user_id:str
role:str
active:bool=True
class WorkflowLab:
def __init__(self,path):
self.path=str(Path(path))
if Path(path).exists() and Path(path).stat().st_size:
with closing(sqlite3.connect(path)) as db:
marker=db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='synthetic_lab_marker'").fetchone()
if not marker or db.execute('SELECT version FROM synthetic_lab_marker').fetchall()!=[('synthetic-atomic-v1',)]:raise ValueError('Not a recognized synthetic lab')
else:
with self.connect() as db:
db.executescript('''
CREATE TABLE synthetic_lab_marker(version TEXT NOT NULL);
INSERT INTO synthetic_lab_marker VALUES('synthetic-atomic-v1');
CREATE TABLE documents(company_id TEXT,id TEXT,assigned_to TEXT,version INTEGER,content_revision INTEGER,quantity INTEGER,state TEXT,PRIMARY KEY(company_id,id));
CREATE TABLE approvals(company_id TEXT,document_id TEXT,round INTEGER,target_revision INTEGER,requester TEXT,status TEXT,reviewer TEXT,reason TEXT,PRIMARY KEY(company_id,document_id,round));
CREATE TABLE events(sequence INTEGER PRIMARY KEY,company_id TEXT,document_id TEXT,actor TEXT,action TEXT,before_json TEXT,after_json TEXT,reason TEXT);
CREATE TABLE requests(company_id TEXT,user_id TEXT,request_key TEXT,fingerprint TEXT,result_json TEXT,PRIMARY KEY(company_id,user_id,request_key));
CREATE TABLE outbox(id INTEGER PRIMARY KEY,company_id TEXT,event_sequence INTEGER,target TEXT,state TEXT,attempts INTEGER DEFAULT 0,UNIQUE(company_id,event_sequence,target));
''')
db.execute("INSERT INTO documents VALUES('A','S01','USER_A',0,0,6,'draft')")
def connect(self):
db=sqlite3.connect(self.path,timeout=10,isolation_level=None,factory=ClosingConnection);db.row_factory=sqlite3.Row
return db
def _context(self,actor):
return isinstance(actor,Actor) and actor.active is True and identifier(actor.company_id) and identifier(actor.user_id) and actor.role in ['staff','coordinator','owner']
def read(self,actor):
if not self._context(actor):return []
with self.connect() as db:
rows=db.execute('SELECT * FROM documents WHERE company_id=? ORDER BY id',[actor.company_id]).fetchall()
return [dict(r) for r in rows if actor.role!='staff' or r['assigned_to']==actor.user_id]
def history(self,actor):
ids={r['id'] for r in self.read(actor)}
with self.connect() as db:
return [dict(r) for r in db.execute('SELECT * FROM events WHERE company_id=? ORDER BY sequence',[actor.company_id]) if r['document_id'] in ids]
def inbox(self,actor):
if not self._context(actor) or actor.role!='coordinator':return []
with self.connect() as db:
return [dict(r) for r in db.execute('SELECT * FROM approvals WHERE company_id=? AND status=? AND requester<>? ORDER BY document_id,round',[actor.company_id,'pending',actor.user_id])]
def execute(self,actor,request_key,command,*,fail_before_commit=False):
if not self._context(actor) or not identifier(request_key) or not isinstance(command,dict):return {'ok':False,'reason':'denied'}
action=command.get('action');keys={'action','document_id','expected_version','reason'}|({'quantity'} if action=='edit' else set())
if action not in ['edit','submit','approve','reject','execute'] or set(command)!=keys or not identifier(command.get('document_id')) or type(command.get('expected_version')) is not int or command['expected_version']<0 or not isinstance(command.get('reason'),str) or not 1<=len(command['reason'])<=160:return {'ok':False,'reason':'invalid_command'}
if action=='edit' and (type(command['quantity']) is not int or not 1<=command['quantity']<=99):return {'ok':False,'reason':'invalid_quantity'}
if actor.role=='owner' or (action in ['approve','reject','execute'] and actor.role!='coordinator'):return {'ok':False,'reason':'denied'}
fingerprint=hashlib.sha256(packed(command).encode('utf-8')).hexdigest()
db=self.connect()
try:
db.execute('BEGIN IMMEDIATE')
def stop(reason):db.rollback();return {'ok':False,'reason':reason}
row=db.execute('SELECT * FROM documents WHERE company_id=? AND id=?',[actor.company_id,command['document_id']]).fetchone()
if not row or (actor.role=='staff' and row['assigned_to']!=actor.user_id):return stop('denied')
previous=db.execute('SELECT * FROM requests WHERE company_id=? AND user_id=? AND request_key=?',[actor.company_id,actor.user_id,request_key]).fetchone()
if previous:
if previous['fingerprint']!=fingerprint:return stop('request_key_reused')
result=json.loads(previous['result_json']);db.commit();return {**result,'replayed':True}
before=dict(row)
if before['version']!=command['expected_version']:return stop('version_conflict')
after={**before,'version':before['version']+1}
if after['version']>9007199254740991:return stop('version_exhausted')
latest=db.execute('SELECT * FROM approvals WHERE company_id=? AND document_id=? ORDER BY round DESC LIMIT 1',[actor.company_id,row['id']]).fetchone()
if action=='edit':
if row['state']=='executed':return stop('state_conflict')
after.update(quantity=command['quantity'],content_revision=row['content_revision']+1,state='draft')
db.execute("UPDATE approvals SET status='superseded' WHERE company_id=? AND document_id=? AND status='pending'",[actor.company_id,row['id']])
elif action=='submit':
if row['state'] not in ['draft','rejected']:return stop('state_conflict')
round_number=(latest['round'] if latest else 0)+1
db.execute('INSERT INTO approvals VALUES(?,?,?,?,?,?,?,?)',[actor.company_id,row['id'],round_number,row['content_revision'],actor.user_id,'pending',None,command['reason']]);after['state']='pending'
elif action in ['approve','reject']:
if row['state']!='pending' or not latest or latest['status']!='pending' or latest['target_revision']!=row['content_revision']:return stop('approval_stale')
if latest['requester']==actor.user_id:return stop('self_approval')
state='approved' if action=='approve' else 'rejected'
db.execute('UPDATE approvals SET status=?,reviewer=?,reason=? WHERE company_id=? AND document_id=? AND round=?',[state,actor.user_id,command['reason'],actor.company_id,row['id'],latest['round']]);after['state']=state
elif action=='execute':
if row['state']!='approved' or not latest or latest['status']!='approved' or latest['target_revision']!=row['content_revision']:return stop('not_approved')
after['state']='executed' # Simulation marker only. No network operation.
db.execute('UPDATE documents SET version=?,content_revision=?,quantity=?,state=? WHERE company_id=? AND id=? AND version=?',[after['version'],after['content_revision'],after['quantity'],after['state'],actor.company_id,row['id'],before['version']])
event=db.execute('INSERT INTO events(company_id,document_id,actor,action,before_json,after_json,reason) VALUES(?,?,?,?,?,?,?)',[actor.company_id,row['id'],actor.user_id,action,packed(before),packed(after),command['reason']]).lastrowid
db.execute("INSERT INTO outbox(company_id,event_sequence,target,state) VALUES(?,?,?,'queued')",[actor.company_id,event,'synthetic-reviewer'])
result={'ok':True,'document':after,'event_sequence':event,'replayed':False}
db.execute('INSERT INTO requests VALUES(?,?,?,?,?)',[actor.company_id,actor.user_id,request_key,fingerprint,packed(result)])
if fail_before_commit:raise RuntimeError('Synthetic injected transaction failure')
db.commit();return result
except BaseException:
db.rollback();raise
finally:db.close()
def delivery_result(self,actor,outbox_id,outcome):
if not self._context(actor) or actor.role!='coordinator' or outcome not in ['delivered','unknown','failed']:return False
with self.connect() as db:
db.execute('BEGIN IMMEDIATE')
row=db.execute('SELECT * FROM outbox WHERE company_id=? AND id=?',[actor.company_id,outbox_id]).fetchone()
if not row or row['state'] in ['delivered','unknown']:db.rollback();return False
db.execute('UPDATE outbox SET state=?,attempts=attempts+1 WHERE id=?',[outcome,outbox_id]);db.commit();return True
def outbox(self,actor):
if not self._context(actor) or actor.role!='coordinator':return []
with self.connect() as db:return [dict(r) for r in db.execute('SELECT * FROM outbox WHERE company_id=? ORDER BY id',[actor.company_id])]
def command(action,version,**values):
return {'action':action,'document_id':'S01','expected_version':version,'reason':'Synthetic review',**values}실행 기록을 읽고 로컬에서 다시 시험한다
아래에서 승인 뒤 수정, 변경 이력, 버전 충돌, 같은 요청의 재응답, 반려 뒤 재제출, 대기함, 실행 분리와 전달 실패 기록을 골라 볼 수 있다. 화면은 패키지의 run_examples.py가 실제로 만든 JSON을 읽어 보여 준다. 선택할 때 새로운 DB 명령을 보내거나 회사 프로그램을 실행하지 않는다. 기록을 확인하는 뷰어라는 범위를 제목과 설명에도 남겼다.
ZIP을 풀고 python test_lab.py와 python run_examples.py를 실행하면 자신의 임시 합성 DB로 다시 확인할 수 있다. 실제 업무 DB 경로를 넣지 않는다. 모르는 기존 DB는 쓰기 전에 거절하는 검사도 있다. 코드·테스트·실행기·기록 JSON·범위 문서가 포함되며 수정·재배포·상업적 사용을 허용하고 무보증으로 제공한다.
검증 결과와 실제 서비스로 연결할 조건
독립 테스트17개는 승인 대상, 자기 승인, 재제출 회차, 대표의 읽기 역할, 승인과 실행의 분리, 두 연결의 수정 경쟁과 중복 요청, 재개 뒤 결과 읽기, 전체 되돌림과 회사 경계를 다룬다. 압축에서 꺼낸 파일로도 실행했다. 처음 Windows 임시 파일 정리에서 열린 DB 연결 문제가 드러나 명시적으로 닫도록 고친 뒤 통과한 기록도 남겼다.
실제 브라우저·물리 기기나 회사의 운영 보안을 검수한 결과가 아니다. 도입하려면 인증된 사용자, 서버의 명령 검증, 실제 저장소의 트랜잭션 조건, 초안 보존과 충돌 화면, 외부 결과 확인을 더 구현해야 한다. 나는 테스트를 통과한 범위와 그 밖의 책임을 함께 넘겨 이 모듈이 무엇을 설명하는지 다음 작업자도 판단할 수 있게 했다.