async def sweep_expired(service, caller, session_id, *, limit=100):
"""No background timer or ambient transaction; retain tombstones forever.
Selection has no row locks: CORE-324 retirement acquires the tenant,
node-row, advisory endpoint, and guard locks in its existing order. The
entire transaction retries only a body-origin 40P01, with bounded attempts.
"""
if type(limit) is not int or not 1 <= limit <= 500:
raise ValueError('Working TTL sweep limit must be in 1..500')
if bound_connection(service.graph.pool) is not None:
raise WorkingWriteUnavailable('Working TTL sweep refuses ambient transaction')
await service.owner.check(caller, session_id)
scope = memory_scope(caller.scope)
owned = tenant_scope(scope)
operation = str(uuid4())
attempted = ()
def record(outcome, retired=()):
return AuditRecord(
action='memory.expire.working', outcome=outcome,
principal_id=caller.principal_id, scope_key=scope.path,
resource_type='memory_ttl_sweep', resource_id=operation,
metadata={'session_id': session_id, 'limit': limit, 'retired_ids': tuple(retired),
'attempted_ids': attempted},
)
phase = 'before'
retired = ()
await service._audit_record(record('initiated'), mandatory=True)
try:
async def write(conn):
nonlocal phase, attempted
await conn.execute('SET TRANSACTION ISOLATION LEVEL READ COMMITTED')
await conn.execute("SET LOCAL lock_timeout = '5s'")
await conn.execute("SET LOCAL statement_timeout = '10s'")
await service.owner.claim(caller, session_id)
from symfonic.memory.backends.postgres_working_expiry_reads import expiry_function
if await expiry_function(conn) is None:
raise WorkingWriteUnavailable('Working TTL sweep requires offline expiry migration')
await conn.execute(
"SELECT pg_advisory_xact_lock("
"hashtextextended('symfonic-working-retire:' || $1,0))",
owned.tenant_id,
)
schema = service.guard.quoted_schema
rows = await conn.fetch(
f'SELECT n.id FROM {schema}.memory_nodes n '
f'JOIN {schema}.memory_working_endpoint_guard g '
'ON g.tenant_id COLLATE "C"=n.tenant_id COLLATE "C" '
'AND g.record_id COLLATE "C"=n.id COLLATE "C" '
f'JOIN {schema}.memory_working_expiry x '
'ON x.tenant_id COLLATE "C"=g.tenant_id COLLATE "C" '
'AND x.record_id COLLATE "C"=g.record_id COLLATE "C" '
'WHERE n.tenant_id COLLATE "C"=$1 AND n.scope_path COLLATE "C"=$2 '
'AND g.scope_path COLLATE "C"=$2 AND x.scope_path COLLATE "C"=$2 '
'AND NOT g.retired AND x.expires_at<=clock_timestamp() '
"AND n.layer='working' AND n.properties->>'session_id' COLLATE \"C\"=$3 "
"AND n.properties->'__symfonic_working_post_v1__'='true'::jsonb "
'ORDER BY x.expires_at,n.id COLLATE "C" LIMIT $4',
owned.tenant_id, owned.scope_path, session_id, limit,
)
identifiers = tuple(row['id'] for row in rows)
attempted = identifiers
for identifier in identifiers:
await retire_working_endpoint(service.guard, owned, identifier)
phase = 'commit_attempt'
return identifiers
retired = await run_working_transaction(service.graph.pool, write)
phase = 'committed'
await service._audit_record(record('completed', retired), mandatory=True)
return {'operation_id': operation, 'retired_ids': retired, 'limit': limit}
except asyncio.CancelledError:
await service._audit_record(record('cancelled' if phase == 'before' else 'uncertain'))
raise
except Exception as exc:
await service._audit_record(record('failed' if phase == 'before' else 'uncertain', retired))
if phase != 'before':
raise WorkingCommitUncertain(operation) from exc
raise WorkingWriteUnavailable('Working TTL sweep was not acknowledged') from exc