Accept in One Transaction

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)

# 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