260 lines
10 KiB
Python
260 lines
10 KiB
Python
"""Offline cancellation/recovery regressions; no HTTP, NAS, secrets or model calls."""
|
|
import asyncio
|
|
import functools
|
|
import sqlite3
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
|
|
def offline_async(function):
|
|
@functools.wraps(function)
|
|
def run(*args, **kwargs):
|
|
return asyncio.run(function(*args, **kwargs))
|
|
return run
|
|
|
|
from control.accounts import Accounts, token_hash
|
|
from control.chat_runs import ChatRuns
|
|
from control.chat_store import ChatConflict, ChatStore
|
|
|
|
|
|
class BlockedAdapter:
|
|
available = True
|
|
approvals = False
|
|
|
|
def __init__(self, *, block_start=False, fail_stop=False):
|
|
self.block_start = block_start
|
|
self.fail_stop = fail_stop
|
|
self.start_entered = asyncio.Event()
|
|
self.start_release = asyncio.Event()
|
|
self.stop_entered = asyncio.Event()
|
|
self.stream_entered = asyncio.Event()
|
|
self.stream_release = asyncio.Event()
|
|
self.starts = []
|
|
self.stops = []
|
|
|
|
async def start(self, owner, rid, payload):
|
|
self.starts.append(rid)
|
|
self.start_entered.set()
|
|
if self.block_start:
|
|
await self.start_release.wait()
|
|
return {'run_id': 'offline-upstream'}
|
|
|
|
async def stop(self, owner, upstream):
|
|
self.stops.append(upstream)
|
|
self.stop_entered.set()
|
|
if self.fail_stop:
|
|
raise RuntimeError('Offline simulated unavailable stop')
|
|
|
|
async def status(self, owner, upstream):
|
|
return {'status': 'cancelled' if self.stops and not self.fail_stop else 'running'}
|
|
|
|
async def events(self, owner, upstream):
|
|
self.stream_entered.set()
|
|
await self.stream_release.wait()
|
|
yield {'event': 'run.cancelled'}
|
|
|
|
|
|
@pytest.fixture
|
|
def execution(tmp_path):
|
|
accounts = Accounts(str(tmp_path / 'offline.sqlite'))
|
|
user = accounts.create('review', 'Offline Review', 'user', 'offline-review-password')
|
|
accounts.update(user['id'], status='ready')
|
|
token, _ = accounts.issue_session(user['id'], 3600, user['password_hash'])
|
|
store = ChatStore(accounts)
|
|
cid = store.create_conversation(user['id'])['id']
|
|
operation = store.send(user['id'], cid, 'Offline request', 'offline-key')
|
|
yield accounts, store, user['id'], token, cid, operation['run']['id']
|
|
accounts.close()
|
|
|
|
|
|
@offline_async
|
|
async def test_cancelled_sqlite_worker_is_joined_by_store_close(execution, monkeypatch, tmp_path):
|
|
"""Task cancellation is not thread completion: close must serialize with SQL."""
|
|
accounts, store, owner, _, _, rid = execution
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
close_attempted = threading.Event()
|
|
close_entered = threading.Event()
|
|
original_lock, original_db = accounts.lock, accounts.db
|
|
original_row = store._row
|
|
closing_thread = threading.local()
|
|
|
|
class ObservedLock:
|
|
def __enter__(self):
|
|
if getattr(closing_thread, 'closing', False):
|
|
close_attempted.set()
|
|
original_lock.acquire()
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
original_lock.release()
|
|
|
|
class ObservedConnection:
|
|
def close(self):
|
|
close_entered.set()
|
|
close_attempted.set()
|
|
# Fail safely on the old implementation, without racing native SQL
|
|
# close in the test process and potentially segfaulting pytest.
|
|
assert original_lock._is_owned(), 'SQLite close must hold the shared store lock'
|
|
original_db.close()
|
|
|
|
def blocked_row(*args):
|
|
entered.set()
|
|
assert release.wait(5), 'Offline worker release timed out'
|
|
return original_row(*args)
|
|
|
|
def close():
|
|
closing_thread.closing = True
|
|
accounts.close()
|
|
|
|
accounts.lock = store.lock = ObservedLock()
|
|
monkeypatch.setattr(accounts, 'db', ObservedConnection())
|
|
monkeypatch.setattr(store, '_row', blocked_row)
|
|
worker = asyncio.create_task(asyncio.to_thread(store.append_event, owner, rid, 'done', {'offline': True}))
|
|
closer = None
|
|
try:
|
|
assert await asyncio.to_thread(entered.wait, 2)
|
|
worker.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await worker
|
|
closer = asyncio.create_task(asyncio.to_thread(close))
|
|
assert await asyncio.to_thread(close_attempted.wait, 2)
|
|
assert not close_entered.is_set(), 'Close must wait for the cancelled task\'s SQLite worker'
|
|
finally:
|
|
release.set()
|
|
try:
|
|
if closer is not None:
|
|
await closer
|
|
finally:
|
|
monkeypatch.setattr(accounts, 'db', original_db)
|
|
accounts.lock = store.lock = original_lock
|
|
assert close_entered.is_set()
|
|
# The worker's terminal replay event committed before close; no test bypass
|
|
# or change to the durable run/event contract is needed.
|
|
with sqlite3.connect(str(tmp_path / 'offline.sqlite')) as reopened:
|
|
assert reopened.execute('SELECT event, data FROM chat_events WHERE run_id=?', (rid,)).fetchall() == [('done', '{"offline": true}')]
|
|
accounts.close() # Teardown close remains idempotent.
|
|
|
|
|
|
def coordinator(execution, adapter, *, upstream=None):
|
|
accounts, store, owner, token, cid, rid = execution
|
|
runs = ChatRuns(accounts, store, adapter=adapter)
|
|
runs.metadata(owner, rid, {'conversation_id': cid, 'payload': {'input': 'Offline request'},
|
|
'session_hash': token_hash(token), 'room': None,
|
|
'upstream_id': upstream, 'assistant_id': None})
|
|
return runs
|
|
|
|
|
|
@offline_async
|
|
async def test_cancel_during_blocked_admission_stops_learned_upstream(execution):
|
|
_, store, owner, _, _, rid = execution
|
|
adapter = BlockedAdapter(block_start=True)
|
|
runs = coordinator(execution, adapter)
|
|
runs.schedule(owner, rid)
|
|
try:
|
|
await asyncio.wait_for(adapter.start_entered.wait(), 2)
|
|
await runs.cancel(owner, rid)
|
|
assert not runs.tasks[rid].done(), 'Do not abort admission before learning its upstream ID'
|
|
adapter.start_release.set()
|
|
await asyncio.wait_for(runs.tasks[rid], 2)
|
|
assert adapter.starts == [rid]
|
|
assert adapter.stops, 'Cancellation must stop the execution admitted during cancellation'
|
|
assert store.get_run(owner, rid)['run']['status'] == 'cancelled'
|
|
finally:
|
|
adapter.start_release.set()
|
|
await runs.stop()
|
|
|
|
|
|
@offline_async
|
|
async def test_stop_failure_keeps_execution_slot_and_durable_intent(execution):
|
|
_, store, owner, _, _, rid = execution
|
|
adapter = BlockedAdapter(fail_stop=True)
|
|
runs = coordinator(execution, adapter, upstream='offline-upstream')
|
|
store.update_run(owner, rid, 'running')
|
|
runs.schedule(owner, rid)
|
|
try:
|
|
await asyncio.wait_for(adapter.stream_entered.wait(), 2)
|
|
await runs.cancel(owner, rid)
|
|
assert runs.metadata(owner, rid).get('stop_requested') is True
|
|
assert store.get_run(owner, rid)['run']['status'] not in runs.store.TERMINAL
|
|
assert runs.locks[owner].locked(), 'Do not admit new account work while old work is unconfirmed'
|
|
assert not runs.tasks[rid].done()
|
|
finally:
|
|
await runs.stop()
|
|
|
|
|
|
@offline_async
|
|
async def test_revoked_session_recovery_stops_remote_before_terminalizing(execution):
|
|
accounts, store, owner, token, _, rid = execution
|
|
adapter = BlockedAdapter()
|
|
runs = coordinator(execution, adapter, upstream='offline-upstream')
|
|
store.update_run(owner, rid, 'running')
|
|
accounts.revoke(token)
|
|
try:
|
|
await asyncio.wait_for(runs.execute(owner, rid), 2)
|
|
assert adapter.starts == [], 'Recovery must never replay a known admission'
|
|
assert adapter.stops, 'A revoked session does not mean its remote execution stopped'
|
|
assert store.get_run(owner, rid)['run']['status'] == 'cancelled'
|
|
finally:
|
|
await runs.stop()
|
|
|
|
|
|
def test_stale_stream_metadata_cannot_erase_cancellation_intent(execution):
|
|
_, _, owner, _, _, rid = execution
|
|
runs = coordinator(execution, BlockedAdapter(), upstream='offline-upstream')
|
|
stale = runs.metadata(owner, rid)
|
|
cancel = runs.metadata(owner, rid)
|
|
cancel['stop_requested'] = True
|
|
runs.metadata(owner, rid, cancel)
|
|
stale['content'] = 'A delta racing with cancellation'
|
|
runs.metadata(owner, rid, stale)
|
|
assert runs.metadata(owner, rid).get('stop_requested') is True
|
|
|
|
|
|
@offline_async
|
|
async def test_restart_after_cancel_during_admission_reconciles_remote(execution):
|
|
_, store, owner, _, _, rid = execution
|
|
adapter = BlockedAdapter(block_start=True, fail_stop=True)
|
|
original = coordinator(execution, adapter)
|
|
original.schedule(owner, rid)
|
|
try:
|
|
await asyncio.wait_for(adapter.start_entered.wait(), 2)
|
|
await original.cancel(owner, rid)
|
|
adapter.start_release.set()
|
|
await asyncio.wait_for(adapter.stop_entered.wait(), 2)
|
|
finally:
|
|
adapter.start_release.set()
|
|
await original.stop()
|
|
# The remote run exists, its stop failed, and the local projection may already
|
|
# be terminal. A restart must still reconcile that durable remote stop intent.
|
|
recovered_adapter = BlockedAdapter()
|
|
recovered = ChatRuns(execution[0], store, adapter=recovered_adapter)
|
|
try:
|
|
await recovered.start()
|
|
await asyncio.wait_for(recovered_adapter.stop_entered.wait(), 2)
|
|
if rid in recovered.tasks:
|
|
await asyncio.wait_for(recovered.tasks[rid], 2)
|
|
assert recovered_adapter.starts == []
|
|
assert store.get_run(owner, rid)['run']['status'] == 'cancelled'
|
|
finally:
|
|
await recovered.stop()
|
|
|
|
|
|
@offline_async
|
|
async def test_delete_cannot_discard_unconfirmed_remote_cancellation(execution):
|
|
_, store, owner, _, cid, rid = execution
|
|
adapter = BlockedAdapter(fail_stop=True)
|
|
runs = coordinator(execution, adapter, upstream='offline-upstream')
|
|
store.update_run(owner, rid, 'running')
|
|
runs.schedule(owner, rid)
|
|
try:
|
|
await asyncio.wait_for(adapter.stream_entered.wait(), 2)
|
|
with pytest.raises(ChatConflict):
|
|
await runs.cancel_conversation(owner, cid)
|
|
store.delete_conversation(owner, cid)
|
|
assert store.get_run(owner, rid)['run']['status'] not in store.TERMINAL
|
|
assert runs.metadata(owner, rid).get('stop_requested') is True
|
|
finally:
|
|
await runs.stop()
|