지금까지는 전부 todo를 open으로 놔뒀어. 행도, 실행도, 탐색도, 등록도, 초대장도. 관리되는 배정의 일이 실제로 주인을 바꾸는 곳은 Engine.accept 하나뿐이고, 그것도 엔진 커맨드 하나 안에서 일어나. BEGIN IMMEDIATE 한 번, 그리고 커밋 한 번이든 롤백 한 번이든.
트랜잭션이 열리기 전
어댑터는 올바른 논스와 함께 COOP_ACCEPT가 실린 완료된 최종 답변일 때만 accept를 불러. 그래도 accept는 영수증부터 봐. 예라고 말한 바로 그 네이티브 메시지를 영수증이 짚지 않으면, 아무것도 열기 전에 거절해. "A correlated native receipt is required." 메시지 id 없는 예는 소문이야.
안에서: 단계를 순서대로
세션. 어댑터가 쥔 세대 그대로여야 하고 끝난 세션이면 안 돼. "Stale or exited session generation."
디스패치. 세션과 세대는 이 세션, 이 세대여야 하고, 논스는 방금 답한 그 논스, 상태는 invited여야 해. 취소됐든, 만료됐든, 이미 수락됐든, 남의 초대든, 거절은 하나로 다 덮어. "Invitation is stale or belongs to another receiver."
항목마다, 하나씩:
디스패치에 적어 둔 리비전 그대로 아직 open인지 봐. 아니면 Work changed before acceptance로 실패해.
초대장에 답글 리뷰가 실렸던 todo라면, 바로 그 답글 id들을 세션의 행위자 이름으로 읽음 처리해.
도메인 자체의 take 연산으로, 세션의 행위자 이름을 달고 가져가.
바인딩을 넣어. 항목, 세션, 세대, 디스패치를 담아서.
결과 의무를 넣어. 기한은 두 시간 뒤야(OUTCOME_REVIEW_SECONDS).
바인딩에 work_revision을 적어. take 이후의 항목 리비전이야.
디스패치를 accepted로 바꾸고, 영수증도 같이 저장해.
작업 봉투를 꾸려(디스패치 id, 바인딩, 쓰기 의도, shared_checkout, 작업 자체, 범위가 있으면 범위까지) 아웃박스에 work 행으로 올려. 이것도 같은 트랜잭션 안이야.
항목마다 확인과 take가 번갈아 일어난다는 걸 눈여겨봐. 항목 2의 리비전 검사가 실패하기 전에 항목 1은 이미 take됐을 수 있어. 괜찮아. 마지막 단계가 성공할 때까지 커밋은 없으니까. 항목 2가 실패하면 롤백이 항목 1의 take, 바인딩, 의무까지 같이 걷어가.
왜 꼭 하나여야 해?
아키텍처는 이 규칙을 디스패치의 네 번째 단계로 적어 뒀어. 트랜잭션 하나 안에서 고른 행과 리비전을 전부 검증하고, 시도를 바인딩하고, 정확히 그 선택만 가져가고, 이벤트와 결과 의무를 덧붙이라고. "Any conflict claims none of the selection." 이걸 쪼개 놓고 단계 사이마다 한 번씩 죽어 봐. 바인딩 없이 커밋된 take는 어느 세션에도 안 묶인 소유 작업이야. 의무 없는 바인딩은 아무도 결과를 빚지지 않는 소유 작업이고, 이 트랙 전체가 막으려는 바로 그 원래 실패야. 커밋 전에 나간 작업 봉투는 장부가 아직 거절할 수 있는 일을 워커가 시작하게 만들고. 트랜잭션 하나면 어디서 죽든 전부 남거나 아무것도 안 남아.
봉투는 따로 짚을 만해. accept 안에서 보내는 게 아니라 거기서 적혀. 아웃박스 행으로. 어댑터가 커밋 뒤에 그걸 배달해. 커밋이 실패하면 봉투는 존재한 적도 없는 거지. 이게 트랜잭셔널 아웃박스 패턴이고, 초대장이 "The engine will commit the claim before sending the task"라고 약속할 수 있었던 이유야.
옛 도우미, 빌린 트랜잭션
take 자체는 Coop 원래의 도메인 연산이야. CLI의 todo take가 부르는 바로 그 take_todo. 행 수로 확인하는 보호된 UPDATE 하나고, 읽고 나서 쓰는 식이 아니야. 자기 트랜잭션은 따로 안 열어. 엔진 밖이라면 connect()가 autocommit 모드 연결을 새로 열어. 그래서 갱신도 이벤트 행도 실행되는 즉시 하나씩 커밋되고, 끝나면 그 연결을 닫아. 엔진 안에선 connect()가 대신 빌린 연결을 건네. 이미 accept의 BEGIN IMMEDIATE 안에 있는 엔진 자신의 연결이야. 그래서 take는 그 트랜잭션 하나에 같이 들어가고, close()는 아무것도 안 해. 자기 트랜잭션을 직접 여는 도우미도 같은 포장으로 끼어. take_scope 같은 거. 걔들의 BEGIN IMMEDIATE는 무해한 SELECT 1이 되고, 커밋은 아무것도 안 하고, 롤백하려 들면 엔진이 커맨드 전체를 거절해. context 모듈의 독스트링이 한 줄로 말해. Legacy helpers may close/commit; the enclosing engine owns the transaction. 혼자 쓰라고 만든 도우미가 고쳐 쓰지 않고도 원자적 수락에 끼는 방법이 이거야.
저장소가 뒤를 받쳐
코드가 틀려도 중요한 불변식은 유일 인덱스 둘이 지켜. engine_active_binding은 작업 항목 하나에 활성 바인딩을 최대 하나만 허락하고, engine_binding_obligation은 바인딩 하나에 종류별 의무를 하나만 허락해. 전체를 멱등으로 만드는 건 커맨드 키야. accept: 뒤에 초대장의 아웃박스 id를 붙인 값이지. 커밋 직후에 죽었다가 다시 묻는 어댑터는 두 번째 take가 아니라 저장된 결과를 돌려받아.
실행도, 소켓 쓰기도, 큐 영수증도, 워커의 문장도 기록을 가져가지 못해. 배정 경로에서 가져가는 건 이 트랜잭션뿐이고, 받은 걸 전부 가져가거나 아무것도 안 가져가.
Code
engine.py 원문 그대로: Engine.accept·python
def accept(self,key,dispatch_id,session_id,generation,nonce,receipt,metadata):
if not receipt or not receipt.get('native_message_id'):
raise ValueError('A correlated native receipt is required')
def run(conn):
session=self._session(conn,session_id,generation)
dispatch=self.row(conn,'engine_dispatches',dispatch_id)
if (dispatch['session_id'],dispatch['generation'],dispatch['nonce'],dispatch['status']) != (session_id,generation,nonce,'invited'):
raise Conflict('Invitation is stale or belongs to another receiver')
bindings=[]
for item in json.loads(dispatch['items']):
table=KINDS[item['kind']]
row=self.row(conn,table,item['id'])
if row['status'] != 'open' or self.revision(conn,table,item['id']) != item['revision']:
raise Conflict('Work changed before acceptance')
if item['kind'] == 'todo':
reviewed = next((r for r in json.loads(dispatch.get('reply_review') or '[]') if r['id']==item['id']), None)
if reviewed:
domain.OPERATIONS['ack_todo_replies'](item['id'], [m['id'] for m in reviewed['messages']], who=session['actor'])
fn=domain.OPERATIONS['take_'+item['kind']]
fn(item['id'],who=session['actor'])
bid=uid()
conn.execute('INSERT INTO engine_bindings(id,kind,item_id,session_id,generation,dispatch_id,created) VALUES (?,?,?,?,?,?,?)',(bid,item['kind'],item['id'],session_id,generation,dispatch_id,time.time()))
conn.execute('INSERT INTO engine_obligations(id,binding_id,kind,due,created) VALUES (?,?,?,?,?)',(uid(),bid,'outcome',time.time()+OUTCOME_REVIEW_SECONDS,time.time()))
conn.execute('UPDATE engine_bindings SET work_revision=? WHERE id=?',(self.revision(conn,table,item['id']),bid))
bindings.append(self.row(conn,'engine_bindings',bid))
conn.execute("UPDATE engine_dispatches SET status='accepted',receipt=? WHERE id=?",(dump(receipt),dispatch_id))
payload={'dispatch_id':dispatch_id,'bindings':bindings,'write_intents':json.loads(dispatch['intents']),
'shared_checkout':self.shared_checkout(conn,dispatch),
'work':[domain.OPERATIONS['get_todo'](i['id']) if i['kind']=='todo' else self.row(conn,KINDS[i['kind']],i['id']) for i in json.loads(dispatch['items'])]}
if dispatch['scope']:
payload['scope']=json.loads(dispatch['scope'])
self._enqueue(conn,session_id,generation,'work',payload)
return payload
return self.command(key,'accept',[dispatch_id,session_id,generation,nonce,receipt],{**metadata,'session':session_id},run)
직접 돌려볼 수 있는 최소 스케치: take, 바인딩, 의무를 원자적으로 넣고 리비전이 어긋나면 롤백·python
# A minimal sketch, not Coop's code: accept takes, binds and obligates in one
# SQLite transaction, and any conflict claims nothing at all.
import json
import sqlite3
import time
import uuid
OUTCOME_REVIEW_SECONDS = 2 * 3600
db = sqlite3.connect(':memory:', isolation_level=None) # we issue BEGIN ourselves
db.executescript('''
CREATE TABLE todos(id INTEGER PRIMARY KEY, status TEXT NOT NULL, taken_by TEXT,
revision INTEGER NOT NULL DEFAULT 1);
CREATE TABLE dispatches(id TEXT PRIMARY KEY, session TEXT NOT NULL, nonce TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'invited', items TEXT NOT NULL);
CREATE TABLE bindings(id TEXT PRIMARY KEY, todo_id INTEGER NOT NULL, session TEXT NOT NULL,
state TEXT NOT NULL DEFAULT 'active');
CREATE UNIQUE INDEX one_owner ON bindings(todo_id) WHERE state='active';
CREATE TABLE obligations(id TEXT PRIMARY KEY, binding_id TEXT NOT NULL,
kind TEXT NOT NULL, due REAL NOT NULL);
''')
class Conflict(Exception):
pass
def prepare(session, todo_ids):
"""Invite a session: record expected revisions and a nonce. Nothing is taken."""
items = [[i, db.execute('SELECT revision FROM todos WHERE id=?', (i,)).fetchone()[0]]
for i in todo_ids]
did, nonce = str(uuid.uuid4()), str(uuid.uuid4())
db.execute('INSERT INTO dispatches(id, session, nonce, items) VALUES (?,?,?,?)',
(did, session, nonce, json.dumps(items)))
return did, nonce
def accept(dispatch_id, session, nonce):
db.execute('BEGIN IMMEDIATE') # take the write lock before the first check
try:
row = db.execute('SELECT session, nonce, status, items FROM dispatches WHERE id=?',
(dispatch_id,)).fetchone()
if row is None or row[:3] != (session, nonce, 'invited'):
raise Conflict('Invitation is stale or belongs to another receiver')
for todo_id, expected in json.loads(row[3]):
status, revision = db.execute('SELECT status, revision FROM todos WHERE id=?',
(todo_id,)).fetchone()
if status != 'open' or revision != expected:
raise Conflict(f'Work changed before acceptance (todo {todo_id})')
db.execute("UPDATE todos SET status='taken', taken_by=?, revision=revision+1 WHERE id=?",
(session, todo_id))
bid = str(uuid.uuid4())
db.execute('INSERT INTO bindings(id, todo_id, session) VALUES (?,?,?)',
(bid, todo_id, session))
db.execute("INSERT INTO obligations VALUES (?,?,'outcome',?)",
(str(uuid.uuid4()), bid, time.time() + OUTCOME_REVIEW_SECONDS))
db.execute("UPDATE dispatches SET status='accepted' WHERE id=?", (dispatch_id,))
db.execute('COMMIT')
except BaseException:
db.execute('ROLLBACK') # todo 1's take, binding and obligation vanish too
raise
def ledger():
todos = db.execute('SELECT id, status, taken_by FROM todos ORDER BY id').fetchall()
bindings = db.execute('SELECT count(*) FROM bindings').fetchone()[0]
owed = db.execute('SELECT count(*) FROM obligations').fetchone()[0]
return f'{todos} bindings={bindings} obligations={owed}'
db.executemany('INSERT INTO todos(id, status) VALUES (?, ?)', [(1, 'open'), (2, 'open')])
did, nonce = prepare('session-A', [1, 2])
# Dad edits todo 2 after the invitation went out: its revision moves on.
db.execute('UPDATE todos SET revision=revision+1 WHERE id=2')
try:
accept(did, 'session-A', nonce)
except Conflict as exc:
print('refused:', exc)
print('after refusal:', ledger())
# Retire that invitation and send a fresh one at the current revisions.
db.execute("UPDATE dispatches SET status='cancelled' WHERE id=?", (did,))
did, nonce = prepare('session-A', [1, 2])
try:
accept(did, 'session-B', nonce) # the wrong receiver presents the nonce
except Conflict as exc:
print('refused:', exc)
accept(did, 'session-A', nonce)
print('after accept: ', ledger())
# refused: Work changed before acceptance (todo 2)
# after refusal: [(1, 'open', None), (2, 'open', None)] bindings=0 obligations=0
# refused: Invitation is stale or belongs to another receiver
# after accept: [(1, 'taken', 'session-A'), (2, 'taken', 'session-A')] bindings=2 obligations=2
스케치를 돌려보고, 일부러 망가뜨려봐. 먼저 마지막에 한 번 커밋하는 대신 todo마다 커밋하게(COMMIT 하고 다시 BEGIN IMMEDIATE) 바꾸고 다시 돌려. 거절된 수락이 남긴 바인딩이랑 의무를 세봐. 그다음 네 프로젝트에서 여러 단계짜리 상태 변경 하나를 찾아(가져가기 + 배정 + 알림 예약 같은 거) 스케치처럼 감싸. 두 단계 사이에 예외를 넣고, 반쪽짜리가 하나도 안 남는지 확인해.
Hint
todo마다 커밋하면 디스패치가 거절됐는데도 todo 1은 바인딩이랑 의무를 단 채 taken으로 남아. Coop의 accept가 절대 만들 수 없는 반쯤 가져간 상태가 바로 그거야. 네 코드에선 안에서 스스로 커밋하는 도우미를 찾아봐. 더 큰 트랜잭션에 끼려면 빌린 연결을 쓰든 고쳐 쓰든 해야 하는 게 걔네야.
Progress
Progress is local-only — sign in to sync across devices.