From 7f8752e37fe5b5a8021f192b87e731caec2bbc2d Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 04:48:24 +0800 Subject: [PATCH 1/5] test: define backend-independent metadata session contracts --- .../workflows/metadata-session-boundary.yml | 66 +++++++ docs/METADATA_SESSION_PROGRESS.md | 27 +++ tests/test_metadata_session_boundary.py | 186 ++++++++++++++++++ 3 files changed, 279 insertions(+) create mode 100644 .github/workflows/metadata-session-boundary.yml create mode 100644 docs/METADATA_SESSION_PROGRESS.md create mode 100644 tests/test_metadata_session_boundary.py diff --git a/.github/workflows/metadata-session-boundary.yml b/.github/workflows/metadata-session-boundary.yml new file mode 100644 index 00000000..5a3bf086 --- /dev/null +++ b/.github/workflows/metadata-session-boundary.yml @@ -0,0 +1,66 @@ +name: Metadata session contracts +on: + push: + branches: [fix/metadata-session-boundary-20260929] + workflow_dispatch: +permissions: + contents: read +concurrency: + group: metadata-session-${{ github.ref }} + cancel-in-progress: false +jobs: + contracts: + runs-on: ubuntu-latest + timeout-minutes: 20 + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: msgtest + POSTGRES_PASSWORD: msgtest + POSTGRES_DB: postgres + ports: ['5432:5432'] + options: >- + --health-cmd "pg_isready -U msgtest" + --health-interval 10s --health-timeout 5s --health-retries 5 + env: + MSG_TEST_POSTGRES_URL_TEMPLATE: postgresql://msgtest:msgtest@127.0.0.1:5432/{database} + steps: + - uses: actions/checkout@v7 + with: + persist-credentials: false + - uses: actions/setup-python@v7 + with: + python-version: '3.15' + allow-prereleases: true + cache: pip + - name: Install isolated test dependencies + run: | + sudo apt-get update -qq + sudo apt-get install -y -qq age + python -m pip install -e '.[dev]' + - name: Record exact source + env: + SOURCE_SHA: ${{ github.sha }} + run: | + mkdir -p artifacts + test "$(git rev-parse HEAD)" = "$SOURCE_SHA" + python - <<'PY' + import json, os, subprocess + from pathlib import Path + git=lambda *args: subprocess.check_output(['git',*args],text=True).strip() + Path('artifacts/source.json').write_text(json.dumps({'commit':git('rev-parse','HEAD'), + 'tree':git('rev-parse','HEAD^{tree}'),'event_sha':os.environ['SOURCE_SHA'], + 'dirty':bool(git('status','--porcelain'))},indent=2)+'\n') + PY + - name: Run the same contract on both real database adapters + run: python -m pytest tests/test_metadata_session_boundary.py --junitxml=artifacts/metadata-session.xml + - name: Preserve PostgreSQL locks, migration and recovery regressions + run: | + python -m pytest tests/test_postgres.py tests/test_security_storage_rollback.py tests/test_security_rollback_fences.py tests/test_ledger_accounts.py tests/test_ledger_account_isolation.py tests/test_ledger_migration_json.py tests/test_ledger_migration_references.py tests/test_recovery_safety_units.py tests/test_recovery_metadata.py --junitxml=artifacts/metadata-regressions.xml + - uses: actions/upload-artifact@v4 + if: always() + with: + name: metadata-session-${{ github.sha }} + path: artifacts/ + if-no-files-found: error diff --git a/docs/METADATA_SESSION_PROGRESS.md b/docs/METADATA_SESSION_PROGRESS.md new file mode 100644 index 00000000..3d94c200 --- /dev/null +++ b/docs/METADATA_SESSION_PROGRESS.md @@ -0,0 +1,27 @@ +# Metadata session boundary progress + +## Coordination and baseline + +- Issue #157 only; broader #65/#69/#85 production and architectural acceptance remain separate. +- Base main: `476e64f04ed1d9c7dcab4a41c4f3cdb72faad6a7`, tree `40ca207a3c49be347120e058e47ed328ec4e3098`. +- Branch: `fix/metadata-session-boundary-20260929`. +- #83 comment 5878201914 records the coordination change. The overlapping regression-only #164 is closed without merging; #165 owns #156/#160 and #155 owns recovery. Do not overwrite either branch. +- Main source was read from CI36472507327 artifact10992168743. Its downloaded SHA-256 matches `9abbc69988b8ffcb45f9931dc87e291145a7680f275bf7aadbedb7bed2189d10`. + +## Stage 1: regression-first + +This commit adds 4 architectural contracts and 6 shared behavioral tests, each behavioral test parameterized over real PostgreSQL and SQLite (16 tests total). Source implementation is unchanged. The focused cloud workflow records exact commit/tree and JUnit; no new pass result is claimed before execution. + +Coverage: independent PostgreSQL import, neutral ownership, declared MetadataSession methods, resource generations/path/paging, persisted result/event/audit/job identities, savepoint compensation, read-only/closed/cross-task guards, unique-constraint rollback and driver parameter/literal handling. + +## Planned boundary + +Before: `PostgresSession -> SqliteSession -> SQLite connection + shared model operations`. + +After: PostgreSQL and SQLite independently implement the driver operation of a neutral MetadataSessionBase. That base owns only model/session invariants and rollback-callback ordering; it does not own a connection, transaction commit, schema, migration, placeholder translation or driver exception handling. + +PostgreSQL keeps its direct/session-pooled connection requirement, connection-scoped advisory lock held through compensation, nested savepoints, pending-effect signals only after commit and exact SQL/error adapter. SQLite keeps its own connection, pragmas, error mapping, writer file fence and test/fake stores. + +## Required evidence before closure + +Obtain actual red evidence, then apply the extraction without schema/migration/domain changes. Run the shared contracts on both adapters, the existing PostgreSQL concurrency/migration/recovery suites, and the complete four-shard node-ID/conformance/build gate on the final head. Verify the tested tree before merging. Production restore/backup/operator evidence is not simulated or marked complete by these tests. diff --git a/tests/test_metadata_session_boundary.py b/tests/test_metadata_session_boundary.py new file mode 100644 index 00000000..9a753fc7 --- /dev/null +++ b/tests/test_metadata_session_boundary.py @@ -0,0 +1,186 @@ +"""One storage contract exercised independently against PostgreSQL and SQLite.""" +from __future__ import annotations + +import asyncio +from dataclasses import replace +from datetime import UTC, datetime +import importlib +import inspect +from pathlib import Path +import subprocess +import sys + +import pytest + +from msg.core.contracts import MetadataSession, MetadataStore +from msg.core.errors import Failure +from msg.core.models import AuditEvent, EffectJob, Event, OperationResult, Principal +from test_postgres import item + + +def test_postgres_import_does_not_load_a_sqlite_adapter_or_driver(): + code = ''' +import importlib.abc, sys +class NoSqlite(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if fullname == 'msg.storage.sqlite' or fullname == 'sqlite3' or fullname.startswith('sqlite3.'): + raise AssertionError('PostgreSQL imported SQLite: ' + fullname) +sys.meta_path.insert(0, NoSqlite()) +import msg.storage.postgres +assert 'msg.storage.sqlite' not in sys.modules +assert 'sqlite3' not in sys.modules +''' + result = subprocess.run([sys.executable, '-c', code], capture_output=True, text=True) + assert result.returncode == 0, result.stderr + + +def test_backends_share_only_the_neutral_session_implementation(): + from msg.storage.postgres import PostgresSession + from msg.storage.sqlite import SqliteSession + common = importlib.import_module('msg.storage.session').MetadataSessionBase + assert not issubclass(PostgresSession, SqliteSession) + assert PostgresSession.__bases__ == SqliteSession.__bases__ == (common,) + for name in ('resource','children','save_result','append_audit','on_rollback','check'): + assert getattr(PostgresSession, name) is getattr(SqliteSession, name) is getattr(common, name) + assert PostgresSession.execute is not SqliteSession.execute + assert 'execute' in common.__abstractmethods__ + + +def test_shared_layer_does_not_own_connections_dialects_schemas_or_commits(): + module = importlib.import_module('msg.storage.session') + text = Path(module.__file__).read_text() + for forbidden in ('sqlite3', 'psycopg', '_connection', 'PRAGMA ', 'CREATE TABLE ', + 'INSERT OR IGNORE', '.commit(', '.rollback(', 'pg_advisory'): + assert forbidden not in text + assert set(module.MetadataSessionBase.__abstractmethods__) == {'execute'} + assert set(inspect.signature(module.MetadataSessionBase).parameters) == {'write'} + + +def test_shared_implementation_covers_the_declared_metadata_contract(): + common = importlib.import_module('msg.storage.session').MetadataSessionBase + for name, member in vars(MetadataSession).items(): + if not name.startswith('_') and callable(member): + assert name in vars(common), name + assert callable(getattr(common, name)), name + assert 'transaction' in vars(MetadataStore) + assert {'one','rows','check','setting','set_setting','job','save_job', + 'organization','path','ancestors','run_rollback_effects'} <= set(vars(common)) + + +@pytest.fixture(params=['postgres','sqlite']) +async def metadata(request, pg_dsn, tmp_path): + from msg.storage.postgres import PostgresMetadataStore + from msg.storage.sqlite import SqliteMetadataStore + store = (PostgresMetadataStore(pg_dsn) if request.param == 'postgres' else + SqliteMetadataStore(tmp_path/'metadata.sqlite')) + try: + yield store + finally: + await store.close() + + +async def test_shared_resource_generation_path_and_paging(metadata): + async with metadata.transaction(write=True) as tx: + await tx.insert(item()) + await tx.insert(item('a','root')) + await tx.insert(item('b','root')) + assert await tx.resolve('/a') == 'a' + assert await tx.path('a') == '/a' + assert [r.id for r in await tx.ancestors('a')] == ['root'] + first = await tx.children('root', limit=1) + assert [r.id for r in first.items] == ['a'] and first.next_cursor == 'a' + assert [r.id for r in (await tx.children('root', cursor=first.next_cursor)).items] == ['b'] + old = await tx.resource('a') + await tx.replace(replace(old, generation=1, mode=0o700), 0) + assert tx.setting('authorization_epoch') == 1 + with pytest.raises(Failure, match='generation_conflict'): + await tx.replace(replace(old, generation=1), 0) + async with metadata.transaction(write=False) as tx: + assert (await tx.resource('a')).generation == 1 + assert (await tx.resource('a')).mode == 0o700 + + +async def test_shared_result_event_audit_and_job_roundtrips(metadata): + now = datetime.now(UTC) + principal = Principal(actor='owner', subject='owner', credential_id=None, + method='local', certificates=(), ceiling=()) + result = OperationResult(request_id='once', operation='test', status='ok', actor=None, subject=None) + event = Event(id='event-one', type='test', time=now, request_id='once', + actor='owner', subject='owner', resources=(), data={}) + audit = AuditEvent(event=event, authority=(), before_digest=None, after_digest=None, + previous_digest=None, entry_digest='', result='ok') + job = EffectJob(id='job-one', event_id=event.id, kind='mail', dedupe_key='unique-job', + principal=principal, operation='mail.send', arguments={}, state='pending', + attempts=0, next_attempt_at=now, lease_until=None) + async with metadata.transaction(write=True) as tx: + await tx.save_result('owner','digest-one',result) + await tx.append_event(event) + await tx.append_audit(audit) + await tx.append_audit(replace(audit, event=replace(event,id='event-two'))) + await tx.enqueue(job) + async with metadata.transaction(write=False) as tx: + assert await tx.request_result('owner','once','digest-one') == result + with pytest.raises(Failure, match='idempotency_conflict'): + await tx.request_result('owner','once','another-digest') + rows = tx.rows('SELECT digest,previous FROM audit ORDER BY seq') + assert rows[0][1] is None and rows[1][1] == rows[0][0] + assert await tx.job(job.id) == job + assert tx.one('SELECT id FROM events WHERE id=?',(event.id,))[0] == event.id + + +async def test_shared_savepoints_and_compensation_keep_their_scope(metadata): + effects = [] + async def compensate(value): + effects.append(value) + async with metadata.transaction(write=True) as outer: + outer.set_setting('outer', True) + outer.on_rollback(lambda: compensate('outer')) + with pytest.raises(ValueError, match='inner'): + async with metadata.transaction(write=True) as inner: + assert inner is outer + inner.set_setting('inner', True) + inner.on_rollback(lambda: compensate('inner')) + raise ValueError('inner') + assert effects == ['inner'] and outer.setting('inner') is None + assert effects == ['inner'] + with pytest.raises(ValueError, match='whole'): + async with metadata.transaction(write=True) as tx: + tx.set_setting('whole', True) + tx.on_rollback(lambda: compensate('whole')) + raise ValueError('whole') + async with metadata.transaction(write=False) as tx: + assert tx.setting('outer') is True and tx.setting('whole') is None + assert effects == ['inner','whole'] + + +async def test_shared_read_only_closed_and_cross_task_guards(metadata): + async with metadata.transaction(write=False) as tx: + with pytest.raises(Failure, match='read_only_transaction'): + tx.set_setting('forbidden', True) + with pytest.raises(Failure, match='read_only_transaction'): + async with metadata.transaction(write=True): + pytest.fail('read-only transaction was upgraded') + async def other_task(): + tx.setting('anything') + with pytest.raises(Failure, match='transaction_cross_task'): + await asyncio.create_task(other_task()) + with pytest.raises(Failure, match='transaction_closed'): + tx.setting('anything') + + +async def test_each_driver_translates_unique_errors_and_rolls_back(metadata): + async with metadata.transaction(write=True) as tx: + await tx.insert(item()) + with pytest.raises(Failure, match='constraint_conflict'): + async with metadata.transaction(write=True) as tx: + await tx.insert(item('child','root')) + await tx.insert(item('child','root')) + async with metadata.transaction(write=False) as tx: + assert tx.one('SELECT COUNT(*) FROM resources')[0] == 1 + assert tx.one('SELECT id FROM resources WHERE id=?',('child',)) is None + + +async def test_each_driver_preserves_literal_percent_and_question_mark(metadata): + async with metadata.transaction(write=False) as tx: + row = tx.one("SELECT '?', ?, '50%'", ('literal?%',)) + assert tuple(row) == ('?', 'literal?%', '50%') From 1dc01c0b120642fd0438266085109cdaa6132d46 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 04:53:35 +0800 Subject: [PATCH 2/5] ci: assemble and audit the exact metadata session extraction --- .../workflows/assemble-metadata-session.yml | 251 ++++++++++++++++++ 1 file changed, 251 insertions(+) create mode 100644 .github/workflows/assemble-metadata-session.yml diff --git a/.github/workflows/assemble-metadata-session.yml b/.github/workflows/assemble-metadata-session.yml new file mode 100644 index 00000000..056816c7 --- /dev/null +++ b/.github/workflows/assemble-metadata-session.yml @@ -0,0 +1,251 @@ +name: Assemble reviewed metadata session extraction +on: + push: + branches: [build/metadata-session-boundary-166] + paths: [.github/workflows/assemble-metadata-session.yml] +permissions: + contents: write +jobs: + assemble: + if: github.repository == 'TokenNotIncluded/msg.lmm.best' && github.ref == 'refs/heads/build/metadata-session-boundary-166' + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@v7 + with: + fetch-depth: 0 + - name: Require the unchanged regression-first source + env: + INPUT_SHA: ${{ github.sha }} + run: | + test "$(git rev-parse HEAD)" = "$INPUT_SHA" + git merge-base --is-ancestor 7f8752e37fe5b5a8021f192b87e731caec2bbc2d HEAD + git diff --exit-code 7f8752e37fe5b5a8021f192b87e731caec2bbc2d -- src tests docs + - name: Extract reviewed session methods without executing project code + run: | + python - <<'PY' + import ast, difflib, hashlib, json, subprocess + from pathlib import Path + + paths = { + 'sqlite': 'src/msg/storage/sqlite.py', + 'postgres': 'src/msg/storage/postgres.py', + 'common': 'src/msg/storage/session.py', + 'notes': 'docs/METADATA_SESSION_PROGRESS.md', + } + expected = { + 'sqlite':'880def3daaf59d3ce66d2e5e980a710c200eb988', + 'postgres':'5ac02cff5b7b048a84e4812c60502ffba889c9c1', + } + before = {name:Path(path).read_text() if Path(path).exists() else '' + for name,path in paths.items()} + for name, sha in expected.items(): + actual=subprocess.check_output(['git','hash-object',paths[name]],text=True).strip() + assert actual == sha, (name,actual,sha) + assert not before['common'], 'neutral module already exists' + + def replace_once(text, old, new): + assert text.count(old)==1, ('unexpected source',old[:100]) + return text.replace(old,new) + + old=before['sqlite'] + start=old.index('class SqliteSession:') + end=old.index('\n\nclass SqliteMetadataStore:') + session=old[start:end] + execute_start=session.index(' def execute(') + execute_end=session.index(' def one(') + driver_execute=session[execute_start:execute_end] + common=replace_once(session,'class SqliteSession:', + 'class MetadataSessionBase(ABC):') + common=replace_once(common, + ' def __init__(self, connection, *, write: bool):\n self._connection, self.write = connection, write', + ' def __init__(self, *, write: bool):\n self.write = write') + common=replace_once(common,driver_execute, + ' @abstractmethod\n' + ' def execute(self, sql: str, parameters: Sequence[object] = (), *,\n' + ' write: bool = False) -> SqlCursor:\n' + ' """Run a bound statement after checking this session access."""\n' + ' raise NotImplementedError\n\n') + models=old[old.index('from msg.core.models import ('):old.index('\n_SCHEMA =')] + common_header='''"""Shared metadata model operations over a backend-owned statement adapter. + + The base owns task/read-only guards, model invariants and compensation + ordering. Drivers own transactions, locks, bound-parameter translation, + exception mapping and durability. No driver handle belongs in this layer. + The internal qmark SQL convention is not a public arbitrary-query API. + """ + from __future__ import annotations + + import asyncio + from abc import ABC, abstractmethod + from collections.abc import Iterator, Sequence + from dataclasses import replace + from typing import Protocol + + from msg.core.codec import canonical, decode, digest, loads, wire + from msg.core.errors import Failure, require + ''' + cursor=''' + class SqlCursor(Protocol): + """Only the cursor behavior used by the shared metadata methods.""" + @property + def rowcount(self) -> int: ... + def fetchone(self) -> tuple[object, ...] | None: ... + def fetchall(self) -> list[tuple[object, ...]]: ... + def __iter__(self) -> Iterator[tuple[object, ...]]: ... + + + ''' + after={} + after['common']=common_header+models+'\n'+cursor+common+'\n' + driver='''class SqliteSession(MetadataSessionBase): + def __init__(self, connection: sqlite3.Connection, *, write: bool): + super().__init__(write=write) + self._connection = connection + + '''+driver_execute.rstrip() + sqlite=old[:start]+driver+old[end:] + sqlite=replace_once(sqlite,'from dataclasses import replace\n','') + sqlite=replace_once(sqlite,'from msg.core.codec import canonical, decode, digest, loads, wire\n','') + sqlite=replace_once(sqlite,models,'from msg.storage.session import MetadataSessionBase\n') + sqlite=replace_once(sqlite, + '"""One transactional metadata session; handlers never own commit rights."""', + '"""SQLite test adapter; shared metadata invariants live in storage.session."""') + after['sqlite']=sqlite + postgres=replace_once(before['postgres'], + 'from msg.storage.sqlite import SqliteSession', + 'from msg.storage.session import MetadataSessionBase') + postgres=replace_once(postgres,'class PostgresSession(SqliteSession):', + 'class PostgresSession(MetadataSessionBase):') + postgres=replace_once(postgres, + ' super().__init__(connection, write=write)', + ' super().__init__(write=write)\n self._connection = connection') + after['postgres']=postgres + after['notes']=before['notes']+''' + + ## Red evidence and implementation + + Focused run **36481798533** tested exact regression commit + `7f8752e37fe5b5a8021f192b87e731caec2bbc2d` / tree + `686dbddc1bdfc52d97c127bdb7338f002cabf735`: **4 failed, 12 passed, + 0 errors, 0 skipped**. All failures were new architecture assertions; + all six shared behavior tests passed independently on both adapters. + Artifact10997260030 was downloaded and its SHA-256 matched the provider: + `50ae1ff1b71684d60e0ff133cde41c6cb81658c0855aeb9c376fff110dc27e2e`. + + The extraction changes only `storage/session.py`, `storage/sqlite.py` + and `storage/postgres.py`. `MetadataSessionBase` now owns the existing + resource/identity/result/event/audit/job methods, task/read-only guards + and rollback compensation. Its sole abstract method is bound statement + execution; `SqlCursor` explicitly describes the small real cursor port. + Both concrete sessions own their own driver handle and statement adapter. + + `core.contracts.MetadataSession` remains the structural business-facing + contract and is not weakened. Existing storage-local helpers (`one`, + `rows`, `check`, path/ancestor lookup, settings and job updates) remain + explicit concrete session methods; they do not grant callers commit + rights. The internal qmark convention is translated only by each driver. + No new ORM, repository service, configuration matrix or migration exists. + + The assembly workflow checks exact input blob IDs and compares every + extracted method except the constructor/statement port with the original + method AST. It also checks both complete Store classes, FakeMetadataStore, + driver execute/enqueue methods, PostgreSQL SQL translation, and both + schema literals for exact structural equality. It uploads this audit, + source patch and source commit/tree. Assembly is not a substitute for + green cloud tests. One-shot tooling stays on the build branch and is + removed from the submitted tree before final validation. + + PostgreSQL connection-scoped advisory locking through compensation, + savepoints, cancellation handling, migration, SQL/error adaptation and + post-commit effect signals are unchanged. SQLite retains its own file + writer fence, pragmas, connection, schema and exception mapping. No schema, + historical ID, ledger entry, signature or receipt bytes are rewritten. + + Final focused/full/recovery results are still pending at this commit. + Do not close #157 or broader production-evidence issues from this stage. + ''' + + def parse(text): + return ast.parse(text) + def cls(text,name): + return next(n for n in parse(text).body if isinstance(n,ast.ClassDef) and n.name==name) + def methods(text,name): + return {n.name:n for n in cls(text,name).body + if isinstance(n,(ast.FunctionDef,ast.AsyncFunctionDef))} + def dump(node): + return ast.dump(node,include_attributes=False) + def schema(text): + return next(n.value for n in parse(text).body if isinstance(n,ast.Assign) + and any(isinstance(t,ast.Name) and t.id=='_SCHEMA' for t in n.targets)) + same=[] + original_methods=methods(before['sqlite'],'SqliteSession') + shared_methods=methods(after['common'],'MetadataSessionBase') + assert set(original_methods)==set(shared_methods) + for name,node in original_methods.items(): + if name not in {'__init__','execute'}: + assert dump(node)==dump(shared_methods[name]),name + same.append('shared.'+name) + for backend, name in [('sqlite','SqliteMetadataStore'),('sqlite','FakeMetadataStore'), + ('postgres','PostgresMetadataStore')]: + assert dump(cls(before[backend],name))==dump(cls(after[backend],name)),name + same.append(name) + for backend,name in [('sqlite','SqliteSession'),('postgres','PostgresSession')]: + left=methods(before[backend],name) + right=methods(after[backend],name) + for method in ['execute']+(['enqueue'] if backend=='postgres' else []): + assert dump(left[method])==dump(right[method]),(backend,method) + same.append(backend+'.'+method) + assert dump(schema(before[backend]))==dump(schema(after[backend])) + same.append(backend+'._SCHEMA') + old_sql=next(n for n in parse(before['postgres']).body if isinstance(n,ast.FunctionDef) and n.name=='_postgres_sql') + new_sql=next(n for n in parse(after['postgres']).body if isinstance(n,ast.FunctionDef) and n.name=='_postgres_sql') + assert dump(old_sql)==dump(new_sql) + same.append('postgres._postgres_sql') + for forbidden in ('sqlite3','psycopg','_connection','PRAGMA ','CREATE TABLE ', + 'INSERT OR IGNORE','.commit(','.rollback(','pg_advisory'): + assert forbidden not in after['common'],forbidden + for name,text in after.items(): + Path(paths[name]).write_text(text) + out=Path('/tmp/metadata-session-assembly') + out.mkdir(parents=True,exist_ok=True) + patch=''.join(''.join(difflib.unified_diff(before[name].splitlines(True),after[name].splitlines(True), + fromfile='a/'+paths[name] if before[name] else '/dev/null',tofile='b/'+paths[name])) for name in after) + (out/'changes.patch').write_text(patch) + (out/'audit.json').write_text(json.dumps({'baseline_blobs':expected, + 'structurally_unchanged':same,'changed_paths':[paths[n] for n in after], + 'patch_sha256':hashlib.sha256(patch.encode()).hexdigest()},indent=2)+'\n') + PY + - name: Commit only the audited source extraction + run: | + git add src/msg/storage/session.py src/msg/storage/sqlite.py src/msg/storage/postgres.py docs/METADATA_SESSION_PROGRESS.md + git diff --cached --check + python - <<'PY' + import json, subprocess + from pathlib import Path + expected=set(json.loads(Path('/tmp/metadata-session-assembly/audit.json').read_text())['changed_paths']) + actual=set(subprocess.check_output(['git','diff','--cached','--name-only'],text=True).splitlines()) + assert actual == expected, (actual,expected) + PY + git -c user.name='github-actions[bot]' -c user.email='41898282+github-actions[bot]@users.noreply.github.com' commit -m 'refactor: separate shared metadata invariants from concrete database sessions' + - name: Publish dedicated build branch without force or workflow changes + env: + INPUT_SHA: ${{ github.sha }} + run: | + test "$(git ls-remote origin refs/heads/build/metadata-session-boundary-166 | cut -f1)" = "$INPUT_SHA" + git push origin HEAD:refs/heads/build/metadata-session-boundary-166 + python - <<'PY' + import json, os, subprocess + from pathlib import Path + git=lambda *args: subprocess.check_output(['git',*args],text=True).strip() + Path('/tmp/metadata-session-assembly/source.json').write_text(json.dumps({ + 'input':os.environ['INPUT_SHA'],'commit':git('rev-parse','HEAD'), + 'tree':git('rev-parse','HEAD^{tree}'),'dirty':bool(git('status','--porcelain'))},indent=2)+'\n') + PY + git archive --format=tar.gz -o /tmp/metadata-session-assembly/source.tar.gz HEAD + - uses: actions/upload-artifact@v4 + if: always() + with: + name: metadata-session-assembly + path: /tmp/metadata-session-assembly/ + if-no-files-found: error From 4a4bac67a253588bba4623c67f45f61830060a76 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 04:56:53 +0800 Subject: [PATCH 3/5] ci: normalize extracted source EOF without changing the audited methods --- .../workflows/assemble-metadata-session.yml | 211 ++---------------- 1 file changed, 18 insertions(+), 193 deletions(-) diff --git a/.github/workflows/assemble-metadata-session.yml b/.github/workflows/assemble-metadata-session.yml index 056816c7..a108093f 100644 --- a/.github/workflows/assemble-metadata-session.yml +++ b/.github/workflows/assemble-metadata-session.yml @@ -21,200 +21,23 @@ jobs: test "$(git rev-parse HEAD)" = "$INPUT_SHA" git merge-base --is-ancestor 7f8752e37fe5b5a8021f192b87e731caec2bbc2d HEAD git diff --exit-code 7f8752e37fe5b5a8021f192b87e731caec2bbc2d -- src tests docs - - name: Extract reviewed session methods without executing project code + - name: Replay the exact audited extraction with its trailing-blank-line fix run: | python - <<'PY' - import ast, difflib, hashlib, json, subprocess - from pathlib import Path - - paths = { - 'sqlite': 'src/msg/storage/sqlite.py', - 'postgres': 'src/msg/storage/postgres.py', - 'common': 'src/msg/storage/session.py', - 'notes': 'docs/METADATA_SESSION_PROGRESS.md', - } - expected = { - 'sqlite':'880def3daaf59d3ce66d2e5e980a710c200eb988', - 'postgres':'5ac02cff5b7b048a84e4812c60502ffba889c9c1', - } - before = {name:Path(path).read_text() if Path(path).exists() else '' - for name,path in paths.items()} - for name, sha in expected.items(): - actual=subprocess.check_output(['git','hash-object',paths[name]],text=True).strip() - assert actual == sha, (name,actual,sha) - assert not before['common'], 'neutral module already exists' - - def replace_once(text, old, new): - assert text.count(old)==1, ('unexpected source',old[:100]) - return text.replace(old,new) - - old=before['sqlite'] - start=old.index('class SqliteSession:') - end=old.index('\n\nclass SqliteMetadataStore:') - session=old[start:end] - execute_start=session.index(' def execute(') - execute_end=session.index(' def one(') - driver_execute=session[execute_start:execute_end] - common=replace_once(session,'class SqliteSession:', - 'class MetadataSessionBase(ABC):') - common=replace_once(common, - ' def __init__(self, connection, *, write: bool):\n self._connection, self.write = connection, write', - ' def __init__(self, *, write: bool):\n self.write = write') - common=replace_once(common,driver_execute, - ' @abstractmethod\n' - ' def execute(self, sql: str, parameters: Sequence[object] = (), *,\n' - ' write: bool = False) -> SqlCursor:\n' - ' """Run a bound statement after checking this session access."""\n' - ' raise NotImplementedError\n\n') - models=old[old.index('from msg.core.models import ('):old.index('\n_SCHEMA =')] - common_header='''"""Shared metadata model operations over a backend-owned statement adapter. - - The base owns task/read-only guards, model invariants and compensation - ordering. Drivers own transactions, locks, bound-parameter translation, - exception mapping and durability. No driver handle belongs in this layer. - The internal qmark SQL convention is not a public arbitrary-query API. - """ - from __future__ import annotations - - import asyncio - from abc import ABC, abstractmethod - from collections.abc import Iterator, Sequence - from dataclasses import replace - from typing import Protocol - - from msg.core.codec import canonical, decode, digest, loads, wire - from msg.core.errors import Failure, require - ''' - cursor=''' - class SqlCursor(Protocol): - """Only the cursor behavior used by the shared metadata methods.""" - @property - def rowcount(self) -> int: ... - def fetchone(self) -> tuple[object, ...] | None: ... - def fetchall(self) -> list[tuple[object, ...]]: ... - def __iter__(self) -> Iterator[tuple[object, ...]]: ... - - - ''' - after={} - after['common']=common_header+models+'\n'+cursor+common+'\n' - driver='''class SqliteSession(MetadataSessionBase): - def __init__(self, connection: sqlite3.Connection, *, write: bool): - super().__init__(write=write) - self._connection = connection - - '''+driver_execute.rstrip() - sqlite=old[:start]+driver+old[end:] - sqlite=replace_once(sqlite,'from dataclasses import replace\n','') - sqlite=replace_once(sqlite,'from msg.core.codec import canonical, decode, digest, loads, wire\n','') - sqlite=replace_once(sqlite,models,'from msg.storage.session import MetadataSessionBase\n') - sqlite=replace_once(sqlite, - '"""One transactional metadata session; handlers never own commit rights."""', - '"""SQLite test adapter; shared metadata invariants live in storage.session."""') - after['sqlite']=sqlite - postgres=replace_once(before['postgres'], - 'from msg.storage.sqlite import SqliteSession', - 'from msg.storage.session import MetadataSessionBase') - postgres=replace_once(postgres,'class PostgresSession(SqliteSession):', - 'class PostgresSession(MetadataSessionBase):') - postgres=replace_once(postgres, - ' super().__init__(connection, write=write)', - ' super().__init__(write=write)\n self._connection = connection') - after['postgres']=postgres - after['notes']=before['notes']+''' - - ## Red evidence and implementation - - Focused run **36481798533** tested exact regression commit - `7f8752e37fe5b5a8021f192b87e731caec2bbc2d` / tree - `686dbddc1bdfc52d97c127bdb7338f002cabf735`: **4 failed, 12 passed, - 0 errors, 0 skipped**. All failures were new architecture assertions; - all six shared behavior tests passed independently on both adapters. - Artifact10997260030 was downloaded and its SHA-256 matched the provider: - `50ae1ff1b71684d60e0ff133cde41c6cb81658c0855aeb9c376fff110dc27e2e`. - - The extraction changes only `storage/session.py`, `storage/sqlite.py` - and `storage/postgres.py`. `MetadataSessionBase` now owns the existing - resource/identity/result/event/audit/job methods, task/read-only guards - and rollback compensation. Its sole abstract method is bound statement - execution; `SqlCursor` explicitly describes the small real cursor port. - Both concrete sessions own their own driver handle and statement adapter. - - `core.contracts.MetadataSession` remains the structural business-facing - contract and is not weakened. Existing storage-local helpers (`one`, - `rows`, `check`, path/ancestor lookup, settings and job updates) remain - explicit concrete session methods; they do not grant callers commit - rights. The internal qmark convention is translated only by each driver. - No new ORM, repository service, configuration matrix or migration exists. - - The assembly workflow checks exact input blob IDs and compares every - extracted method except the constructor/statement port with the original - method AST. It also checks both complete Store classes, FakeMetadataStore, - driver execute/enqueue methods, PostgreSQL SQL translation, and both - schema literals for exact structural equality. It uploads this audit, - source patch and source commit/tree. Assembly is not a substitute for - green cloud tests. One-shot tooling stays on the build branch and is - removed from the submitted tree before final validation. - - PostgreSQL connection-scoped advisory locking through compensation, - savepoints, cancellation handling, migration, SQL/error adaptation and - post-commit effect signals are unchanged. SQLite retains its own file - writer fence, pragmas, connection, schema and exception mapping. No schema, - historical ID, ledger entry, signature or receipt bytes are rewritten. - - Final focused/full/recovery results are still pending at this commit. - Do not close #157 or broader production-evidence issues from this stage. - ''' - - def parse(text): - return ast.parse(text) - def cls(text,name): - return next(n for n in parse(text).body if isinstance(n,ast.ClassDef) and n.name==name) - def methods(text,name): - return {n.name:n for n in cls(text,name).body - if isinstance(n,(ast.FunctionDef,ast.AsyncFunctionDef))} - def dump(node): - return ast.dump(node,include_attributes=False) - def schema(text): - return next(n.value for n in parse(text).body if isinstance(n,ast.Assign) - and any(isinstance(t,ast.Name) and t.id=='_SCHEMA' for t in n.targets)) - same=[] - original_methods=methods(before['sqlite'],'SqliteSession') - shared_methods=methods(after['common'],'MetadataSessionBase') - assert set(original_methods)==set(shared_methods) - for name,node in original_methods.items(): - if name not in {'__init__','execute'}: - assert dump(node)==dump(shared_methods[name]),name - same.append('shared.'+name) - for backend, name in [('sqlite','SqliteMetadataStore'),('sqlite','FakeMetadataStore'), - ('postgres','PostgresMetadataStore')]: - assert dump(cls(before[backend],name))==dump(cls(after[backend],name)),name - same.append(name) - for backend,name in [('sqlite','SqliteSession'),('postgres','PostgresSession')]: - left=methods(before[backend],name) - right=methods(after[backend],name) - for method in ['execute']+(['enqueue'] if backend=='postgres' else []): - assert dump(left[method])==dump(right[method]),(backend,method) - same.append(backend+'.'+method) - assert dump(schema(before[backend]))==dump(schema(after[backend])) - same.append(backend+'._SCHEMA') - old_sql=next(n for n in parse(before['postgres']).body if isinstance(n,ast.FunctionDef) and n.name=='_postgres_sql') - new_sql=next(n for n in parse(after['postgres']).body if isinstance(n,ast.FunctionDef) and n.name=='_postgres_sql') - assert dump(old_sql)==dump(new_sql) - same.append('postgres._postgres_sql') - for forbidden in ('sqlite3','psycopg','_connection','PRAGMA ','CREATE TABLE ', - 'INSERT OR IGNORE','.commit(','.rollback(','pg_advisory'): - assert forbidden not in after['common'],forbidden - for name,text in after.items(): - Path(paths[name]).write_text(text) - out=Path('/tmp/metadata-session-assembly') - out.mkdir(parents=True,exist_ok=True) - patch=''.join(''.join(difflib.unified_diff(before[name].splitlines(True),after[name].splitlines(True), - fromfile='a/'+paths[name] if before[name] else '/dev/null',tofile='b/'+paths[name])) for name in after) - (out/'changes.patch').write_text(patch) - (out/'audit.json').write_text(json.dumps({'baseline_blobs':expected, - 'structurally_unchanged':same,'changed_paths':[paths[n] for n in after], - 'patch_sha256':hashlib.sha256(patch.encode()).hexdigest()},indent=2)+'\n') + import subprocess, textwrap + recipe_commit='1dc01c0b120642fd0438266085109cdaa6132d46' + recipe_path='.github/workflows/assemble-metadata-session.yml' + workflow=subprocess.check_output(['git','show',recipe_commit+':'+recipe_path],text=True) + step=workflow.index(' - name: Extract reviewed session methods without executing project code\n') + opening=" python - <<'PY'\n" + begin=workflow.index(opening,step)+len(opening) + end=workflow.index(' PY\n',begin) + source=textwrap.dedent(workflow[begin:end]) + old="after['common']=common_header+models+'\\n'+cursor+common+'\\n'" + new="after['common']=common_header+models+'\\n'+cursor+common.rstrip()+'\\n'" + assert source.count(old)==1, 'unexpected pinned recipe' + source=source.replace(old,new) + exec(compile(source,recipe_commit+':metadata-session-extraction','exec')) PY - name: Commit only the audited source extraction run: | @@ -240,7 +63,9 @@ jobs: git=lambda *args: subprocess.check_output(['git',*args],text=True).strip() Path('/tmp/metadata-session-assembly/source.json').write_text(json.dumps({ 'input':os.environ['INPUT_SHA'],'commit':git('rev-parse','HEAD'), - 'tree':git('rev-parse','HEAD^{tree}'),'dirty':bool(git('status','--porcelain'))},indent=2)+'\n') + 'tree':git('rev-parse','HEAD^{tree}'),'dirty':bool(git('status','--porcelain')), + 'recipe_commit':'1dc01c0b120642fd0438266085109cdaa6132d46', + 'recipe_adjustment':'trim one extra blank line at common module EOF'},indent=2)+'\n') PY git archive --format=tar.gz -o /tmp/metadata-session-assembly/source.tar.gz HEAD - uses: actions/upload-artifact@v4 From 5474eaf475e7f844244413b1f3e5109d4add4dce Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 28 Sep 2026 20:59:58 +0000 Subject: [PATCH 4/5] refactor: separate shared metadata invariants from concrete database sessions --- docs/METADATA_SESSION_PROGRESS.md | 43 ++++ src/msg/storage/postgres.py | 7 +- src/msg/storage/session.py | 414 ++++++++++++++++++++++++++++++ src/msg/storage/sqlite.py | 390 +--------------------------- 4 files changed, 467 insertions(+), 387 deletions(-) create mode 100644 src/msg/storage/session.py diff --git a/docs/METADATA_SESSION_PROGRESS.md b/docs/METADATA_SESSION_PROGRESS.md index 3d94c200..c0d405c6 100644 --- a/docs/METADATA_SESSION_PROGRESS.md +++ b/docs/METADATA_SESSION_PROGRESS.md @@ -25,3 +25,46 @@ PostgreSQL keeps its direct/session-pooled connection requirement, connection-sc ## Required evidence before closure Obtain actual red evidence, then apply the extraction without schema/migration/domain changes. Run the shared contracts on both adapters, the existing PostgreSQL concurrency/migration/recovery suites, and the complete four-shard node-ID/conformance/build gate on the final head. Verify the tested tree before merging. Production restore/backup/operator evidence is not simulated or marked complete by these tests. + + +## Red evidence and implementation + +Focused run **36481798533** tested exact regression commit +`7f8752e37fe5b5a8021f192b87e731caec2bbc2d` / tree +`686dbddc1bdfc52d97c127bdb7338f002cabf735`: **4 failed, 12 passed, +0 errors, 0 skipped**. All failures were new architecture assertions; +all six shared behavior tests passed independently on both adapters. +Artifact10997260030 was downloaded and its SHA-256 matched the provider: +`50ae1ff1b71684d60e0ff133cde41c6cb81658c0855aeb9c376fff110dc27e2e`. + +The extraction changes only `storage/session.py`, `storage/sqlite.py` +and `storage/postgres.py`. `MetadataSessionBase` now owns the existing +resource/identity/result/event/audit/job methods, task/read-only guards +and rollback compensation. Its sole abstract method is bound statement +execution; `SqlCursor` explicitly describes the small real cursor port. +Both concrete sessions own their own driver handle and statement adapter. + +`core.contracts.MetadataSession` remains the structural business-facing +contract and is not weakened. Existing storage-local helpers (`one`, +`rows`, `check`, path/ancestor lookup, settings and job updates) remain +explicit concrete session methods; they do not grant callers commit +rights. The internal qmark convention is translated only by each driver. +No new ORM, repository service, configuration matrix or migration exists. + +The assembly workflow checks exact input blob IDs and compares every +extracted method except the constructor/statement port with the original +method AST. It also checks both complete Store classes, FakeMetadataStore, +driver execute/enqueue methods, PostgreSQL SQL translation, and both +schema literals for exact structural equality. It uploads this audit, +source patch and source commit/tree. Assembly is not a substitute for +green cloud tests. One-shot tooling stays on the build branch and is +removed from the submitted tree before final validation. + +PostgreSQL connection-scoped advisory locking through compensation, +savepoints, cancellation handling, migration, SQL/error adaptation and +post-commit effect signals are unchanged. SQLite retains its own file +writer fence, pragmas, connection, schema and exception mapping. No schema, +historical ID, ledger entry, signature or receipt bytes are rewritten. + +Final focused/full/recovery results are still pending at this commit. +Do not close #157 or broader production-evidence issues from this stage. diff --git a/src/msg/storage/postgres.py b/src/msg/storage/postgres.py index 5ac02cff..a0891512 100644 --- a/src/msg/storage/postgres.py +++ b/src/msg/storage/postgres.py @@ -23,7 +23,7 @@ from psycopg.conninfo import conninfo_to_dict, make_conninfo from msg.core.errors import Failure -from msg.storage.sqlite import SqliteSession +from msg.storage.session import MetadataSessionBase _SCHEMA = """ CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY); @@ -509,9 +509,10 @@ def _postgres_sql(sql: str, *, has_parameters: bool = False) -> str: return ''.join(output) -class PostgresSession(SqliteSession): +class PostgresSession(MetadataSessionBase): def __init__(self, connection, *, write: bool): - super().__init__(connection, write=write) + super().__init__(write=write) + self._connection = connection self.pending_effect_ids: list[str] = [] def execute(self, sql, parameters=(), *, write=False): diff --git a/src/msg/storage/session.py b/src/msg/storage/session.py new file mode 100644 index 00000000..45b48b41 --- /dev/null +++ b/src/msg/storage/session.py @@ -0,0 +1,414 @@ +"""Shared metadata model operations over a backend-owned statement adapter. + +The base owns task/read-only guards, model invariants and compensation +ordering. Drivers own transactions, locks, bound-parameter translation, +exception mapping and durability. No driver handle belongs in this layer. +The internal qmark SQL convention is not a public arbitrary-query API. +""" +from __future__ import annotations + +import asyncio +from abc import ABC, abstractmethod +from collections.abc import Iterator, Sequence +from dataclasses import replace +from typing import Protocol + +from msg.core.codec import canonical, decode, digest, loads, wire +from msg.core.errors import Failure, require +from msg.core.models import ( + AuditEvent, Certificate, CertificateRequest, CertificateRequestState, Credential, + EffectJob, EmailSettings, Event, Membership, OperationResult, Organization, Page, + Resource, ResourceRef, Revision, Subject, TransferChunk, TransferSession, +) + + +class SqlCursor(Protocol): + """Only the cursor behavior used by the shared metadata methods.""" + @property + def rowcount(self) -> int: ... + def fetchone(self) -> tuple[object, ...] | None: ... + def fetchall(self) -> list[tuple[object, ...]]: ... + def __iter__(self) -> Iterator[tuple[object, ...]]: ... + + +class MetadataSessionBase(ABC): + def __init__(self, *, write: bool): + self.write = write + self.owner_task = asyncio.current_task() + self.closed = False + self.rollback_effects = [] + + def on_rollback(self, effect): + """Compensate external work if this transaction or savepoint aborts.""" + self.check(write=True) + self.rollback_effects.append(effect) + + async def run_rollback_effects(self, cause, start=0): + effects = self.rollback_effects[start:] + del self.rollback_effects[start:] + if not effects: + return + + async def compensate(): + for effect in reversed(effects): + try: + await effect() + except BaseException as exc: + # Keep the primary error and try every independent cleanup. + cause.add_note('rollback compensation failed: ' + type(exc).__name__) + + # A second cancellation must not abandon a pin operation (including its + # worker thread) and release the writer fence before it finishes. + pending = asyncio.create_task(compensate()) + while not pending.done(): + try: + await asyncio.shield(pending) + except asyncio.CancelledError: + pass + pending.result() + + def check(self, write=False): + require(not self.closed, "transaction_closed") + require(asyncio.current_task() is self.owner_task, "transaction_cross_task") + if write: + require(self.write, "read_only_transaction") + + @abstractmethod + def execute(self, sql: str, parameters: Sequence[object] = (), *, + write: bool = False) -> SqlCursor: + """Run a bound statement after checking this session access.""" + raise NotImplementedError + + def one(self, sql, parameters=()): + return self.execute(sql, parameters).fetchone() + + def rows(self, sql, parameters=()): + return self.execute(sql, parameters).fetchall() + + async def resource(self, id): + row = self.one("SELECT body FROM resources WHERE id=?", (id,)) + require(row is not None, "not_found") + return decode(Resource, loads(row[0])) + + async def resolve(self, path): + require(isinstance(path, str) and path.startswith("/"), "invalid_path") + parts = path.rstrip("/").split("/")[1:] + require(all(p not in {".", ".."} and "\\" not in p and "\x00" not in p for p in parts), "invalid_path") + if len(parts) == 2 and parts[0] == "_id": + return (await self.resource(parts[1])).id + row = self.one("SELECT id FROM resources WHERE parent IS NULL") + require(row is not None, "not_initialized") + rid = row[0] + if path == "/": + return rid + for part in parts: + require(bool(part), "invalid_path") + row = self.one("SELECT id FROM resources WHERE parent=? AND name=?", (rid, part)) + require(row is not None, "not_found") + rid = row[0] + return rid + + async def path(self, id): + segments, seen = [], set() + r = await self.resource(id) + while r.parent is not None: + require(r.id not in seen, "parent_cycle") + seen.add(r.id) + segments.append(r.name) + r = await self.resource(r.parent) + return "/" + "/".join(reversed(segments)) + + async def ancestors(self, id): + result, seen = [], set() + r = await self.resource(id) + while r.parent is not None: + require(r.id not in seen, "parent_cycle") + seen.add(r.id) + r = await self.resource(r.parent) + result.append(r) + return tuple(reversed(result)) + + async def children(self, parent, cursor=None, limit=50): + require(type(limit) is int and 1 <= limit <= 500, "invalid_limit") + rows = self.rows("SELECT body FROM resources WHERE parent=? AND id>? ORDER BY id LIMIT ?", + (parent, cursor or "", limit + 1)) + values = tuple(decode(Resource, loads(row[0])) for row in rows[:limit]) + return Page(items=values, next_cursor=values[-1].id if len(rows) > limit else None) + + async def revision(self, ref): + rid = ref.revision or (await self.resource(ref.id)).revision + row = self.one("SELECT body FROM revisions WHERE id=? AND resource_id=?", (rid, ref.id)) + require(row is not None, "revision_not_found") + return decode(Revision, loads(row[0])) + + async def history(self, id, cursor=None, limit=50): + require(1 <= limit <= 500, "invalid_limit") + rows = self.rows("SELECT body FROM revisions WHERE resource_id=? AND id>? ORDER BY id LIMIT ?", + (id, cursor or "", limit+1)) + values = tuple(decode(Revision, loads(r[0])) for r in rows[:limit]) + return Page(items=values, next_cursor=values[-1].id if len(rows)>limit else None) + + async def insert(self, resource): + self.check(True) + if resource.parent is not None: + await self.resource(resource.parent) + data = wire(resource) + self.execute("INSERT INTO resources VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", ( + resource.id, resource.type, resource.name, resource.parent, resource.owner, + resource.group, resource.mode, resource.generation, resource.revision, + resource.state, data['created_at'], data['modified_at'], canonical(resource).decode()), write=True) + for tag in resource.tags: + self.execute("INSERT INTO resource_tags VALUES (?,?)", (resource.id,tag), write=True) + + async def replace(self, resource, expected_generation): + self.check(True) + old = await self.resource(resource.id) + require(old.generation == expected_generation, "generation_conflict", + details={"generation":old.generation, "revision":old.revision}) + require(resource.generation == expected_generation + 1, "invalid_generation") + require(resource.type == old.type and resource.created_at == old.created_at and + resource.created_by == old.created_by, "immutable_creation_fact") + if resource.parent is not None: + require(resource.parent != resource.id, "parent_cycle") + parent = await self.resource(resource.parent) + require(all(a.id != resource.id for a in await self.ancestors(parent.id)), "parent_cycle") + if any(getattr(resource,k)!=getattr(old,k) for k in ('parent','owner','group','mode','state')): + self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1) + data = wire(resource) + changed = self.execute("""UPDATE resources SET name=?,parent=?,owner=?,grp=?,mode=?, + generation=?,revision=?,state=?,modified_at=?,body=? WHERE id=? AND generation=?""", + (resource.name, resource.parent, resource.owner, resource.group, resource.mode, + resource.generation, resource.revision, resource.state, data['modified_at'], + canonical(resource).decode(), resource.id, expected_generation), write=True).rowcount + require(changed == 1, "generation_conflict") + if resource.tags != old.tags: + self.execute("DELETE FROM resource_tags WHERE resource_id=?", (resource.id,), write=True) + for tag in resource.tags: + self.execute("INSERT INTO resource_tags VALUES (?,?)", (resource.id,tag), write=True) + + async def append_revision(self, revision): + self.check(True) + for parent in revision.parents: + await self.revision(ResourceRef(id=revision.resource_id, revision=parent)) + self.execute("INSERT INTO revisions VALUES (?,?,?,?)", ( + revision.id, revision.resource_id, wire(revision.created_at), canonical(revision).decode()), write=True) + for relation in revision.relations: + self.execute("INSERT INTO relations VALUES (?,?,?,?,?,?)", ( + revision.id, revision.resource_id, relation.type, relation.target.id, + relation.target.revision, canonical(relation).decode()), write=True) + + async def subject(self, id): + row = self.one("SELECT body FROM identities WHERE id=? AND kind='subject'", (id,)) + require(row is not None, "subject_not_found") + return decode(Subject, loads(row[0])) + + async def organization(self, id): + row = self.one("SELECT body FROM identities WHERE id=? AND kind='organization'", (id,)) + require(row is not None, "group_not_found") + return decode(Organization, loads(row[0])) + + async def credential(self, id): + row = self.one("SELECT body FROM credentials WHERE id=?", (id,)) + require(row is not None, "credential_not_found") + return decode(Credential, loads(row[0])) + + async def certificate(self, id): + row = self.one("SELECT body FROM certificates WHERE id=?", (id,)) + require(row is not None, "certificate_not_found") + return decode(Certificate, loads(row[0])) + + async def certificate_revoked(self, id): + row = self.one("SELECT revoked FROM certificates WHERE id=?", (id,)) + require(row is not None, "certificate_not_found") + return bool(row[0]) + + async def memberships(self, subject): + if subject is None: + return () + rows=self.rows("""SELECT m.org,m.body FROM memberships m + JOIN resources r ON r.id=m.org + WHERE m.subject=? AND r.type='organization' AND r.state='active' + ORDER BY m.org""",(subject,)) + memberships=((org,decode(Membership,loads(body))) for org,body in rows) + # A stale membership row must not revive an archived/missing organization + # or borrow an organization/subject identifier from a mismatched body. + active=tuple(member for org,member in memberships if member.status=='active' + and member.organization_id==org and member.subject_id==subject + and org!='g_public') + row=self.one("SELECT body FROM identities WHERE id=? AND kind='subject'",(subject,)) + if row is None: + return active + identity=decode(Subject,loads(row[0])) + if identity.kind in {'registered','custodial','system'} and not identity.local_only: + # /&public is a virtual group for formal subjects. Historical rows + # may remain in storage, but never create a duplicate authorization. + return (Membership(organization_id='g_public',subject_id=subject, + role='member',version=0),*active) + return active + + async def request_result(self, subject, id, digest): + row = self.one("SELECT digest,body FROM results WHERE subject=? AND request_id=?", (subject,id)) + if row is None: + return None + require(row[0] == digest, "idempotency_conflict") + return decode(OperationResult, loads(row[1])) + + async def save_result(self, subject, digest, result): + from msg.storage.capacity import require_result_capacity + require_result_capacity(self) + self.execute("INSERT INTO results VALUES (?,?,?,?)", ( + subject, result.request_id, digest, canonical(result).decode()), write=True) + + async def update_identity(self, record, expected_generation): + self.check(True) + if isinstance(record, Membership): + self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1) + table, where, args = 'memberships', 'org=? AND subject=?', (record.organization_id,record.subject_id) + generation = record.version + elif isinstance(record, EmailSettings): + table, where, args = 'emails', 'subject=?', (record.subject_id,) + generation = expected_generation + 1 + else: + table, where, args = 'identities', 'id=?', (record.resource_id,) + generation = record.auth_version if isinstance(record, Subject) else record.membership_version + row = self.one(f"SELECT generation FROM {table} WHERE {where}", args) + require((row is None and expected_generation == -1) or + (row is not None and row[0] == expected_generation), "generation_conflict") + payload = canonical(record).decode() + if row is None: + if table == 'identities': + self.execute("INSERT INTO identities VALUES (?,?,?,?)", (*args, + 'subject' if isinstance(record,Subject) else 'organization', generation,payload), write=True) + elif table == 'memberships': + self.execute("INSERT INTO memberships VALUES (?,?,?,?)", (*args,generation,payload), write=True) + else: + self.execute("INSERT INTO emails VALUES (?,?,?)", (*args,generation,payload), write=True) + else: + self.execute(f"UPDATE {table} SET generation=?,body=? WHERE {where}", + (generation,payload,*args), write=True) + + async def save_credential(self, credential, expected_auth_version): + subject = await self.subject(credential.subject_id) + require(subject.auth_version == expected_auth_version, "auth_version_conflict") + old=self.one("SELECT body FROM credentials WHERE id=?",(credential.id,)) + if old is not None: + previous=decode(Credential,loads(old[0])) + require(previous.subject_id==credential.subject_id and previous.kind==credential.kind and + previous.verifier==credential.verifier,"credential_identity_immutable") + self.execute("INSERT INTO credentials VALUES (?,?,?) ON CONFLICT(id) DO UPDATE SET body=excluded.body", + (credential.id,credential.subject_id,canonical(credential).decode()),write=True) + + async def csr(self, id): + row = self.one("SELECT body FROM csrs WHERE id=?",(id,)) + require(row is not None,"csr_not_found") + return decode(CertificateRequest,loads(row[0])) + + async def csr_state(self, id): + row=self.one("SELECT state_body FROM csrs WHERE id=?",(id,)) + require(row is not None,"csr_not_found") + return decode(CertificateRequestState,loads(row[0])) + + async def save_csr(self, request): + state=CertificateRequestState(request_id=request.resource_id,status='pending',generation=0) + self.execute("INSERT INTO csrs VALUES (?,?,?,?,?)",(request.resource_id,0,'pending', + canonical(request).decode(),canonical(state).decode()),write=True) + + async def transition_csr(self, state, expected_generation): + require(state.generation == expected_generation+1,"invalid_generation") + old=await self.csr_state(state.request_id) + require(old.status=='pending',"csr_not_pending") + require(old.generation==expected_generation,"generation_conflict") + self.execute("UPDATE csrs SET generation=?,state=?,state_body=? WHERE id=? AND generation=?", + (state.generation,state.status,canonical(state).decode(),state.request_id,expected_generation),write=True) + + async def register_certificate(self, certificate, csr_id, expected_generation): + self.execute("INSERT INTO certificates VALUES (?,?,?,?,?)",( + certificate.resource_id,certificate.subject_id,certificate.parent_certificate_id,0, + canonical(certificate).decode()),write=True) + if csr_id is not None: + self.set_setting('certificate_request:'+certificate.resource_id,csr_id) + await self.transition_csr(CertificateRequestState(request_id=csr_id,status='issued', + generation=expected_generation+1,certificate_id=certificate.resource_id),expected_generation) + + async def revoke_certificate(self, id, event): + await self.certificate(id) + self.execute("UPDATE certificates SET revoked=1 WHERE id=?",(id,),write=True) + self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1) + await self.append_audit(event) + + async def transfer(self,id): + row=self.one("SELECT body FROM transfers WHERE id=?",(id,)) + require(row is not None,"transfer_not_found") + return decode(TransferSession,loads(row[0])) + + async def save_transfer(self,session,expected_generation): + if expected_generation is None: + self.execute("INSERT INTO transfers (id,subject,generation,body) VALUES (?,?,?,?)", + (session.id,session.subject_id,session.generation,canonical(session).decode()),write=True) + else: + require(session.generation==expected_generation+1,"invalid_generation") + count=self.execute("UPDATE transfers SET generation=?,body=? WHERE id=? AND generation=?", + (session.generation,canonical(session).decode(),session.id,expected_generation),write=True).rowcount + require(count==1,"generation_conflict") + + async def put_chunk(self,chunk): + self.check(True) + start,end=chunk.offset,chunk.offset+chunk.content.size + rows=self.rows("SELECT offset,length,body FROM chunks WHERE transfer_id=? AND offset?", + (chunk.transfer_id,end,start)) + if rows: + require(len(rows)==1 and rows[0][0]==start and rows[0][1]==chunk.content.size and + decode(TransferChunk,loads(rows[0][2])).content==chunk.content,"chunk_conflict") + return + self.execute("INSERT INTO chunks VALUES (?,?,?,?)",(chunk.transfer_id,start,chunk.content.size, + canonical(chunk).decode()),write=True) + + async def missing_ranges(self,id,cursor=None,limit=50): + session=await self.transfer(id) + require(session.expected_size is not None,"size_required") + require(1<=limit<=500,"invalid_limit") + start=int(cursor or 0) + require(0<=start<=session.expected_size,"invalid_cursor") + gaps=[] + for row in self.execute("SELECT offset,length FROM chunks WHERE transfer_id=? AND offset+length>? ORDER BY offset",(id,start)): + if row[0]>start: + gaps.append((start,row[0])) + if len(gaps)>limit: + return Page(items=tuple(gaps[:limit]),next_cursor=str(gaps[limit][0])) + start=max(start,row[0]+row[1]) + if startlimit else None) + + async def append_event(self,event): + self.execute("INSERT INTO events (id,body) VALUES (?,?)",(event.id,canonical(event).decode()),write=True) + + async def append_audit(self,event): + previous=self.one("SELECT digest FROM audit ORDER BY seq DESC LIMIT 1") + prev=previous[0] if previous else None + data=wire(replace(event,previous_digest=prev,entry_digest='')) + data.pop('entry_digest') + entry=replace(event,previous_digest=prev,entry_digest=digest(data)) + self.execute("INSERT INTO audit (digest,previous,body) VALUES (?,?,?)", + (entry.entry_digest,prev,canonical(entry).decode()),write=True) + + async def enqueue(self,job): + self.execute("INSERT INTO jobs VALUES (?,?,?,?,?,?)",(job.id,job.dedupe_key,job.kind, + job.state,wire(job.next_attempt_at),canonical(job).decode()),write=True) + + async def job(self,id): + row=self.one("SELECT body FROM jobs WHERE id=?",(id,)) + require(row is not None,'job_not_found') + return decode(EffectJob,loads(row[0])) + + async def save_job(self,job): + self.execute("UPDATE jobs SET state=?,next_at=?,body=? WHERE id=?",(job.state, + wire(job.next_attempt_at),canonical(job).decode(),job.id),write=True) + + def setting(self,key,default=None): + row=self.one("SELECT value FROM settings WHERE key=?",(key,)) + return loads(row[0]) if row else default + + def set_setting(self,key,value): + self.execute("INSERT INTO settings VALUES (?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", + (key,canonical(value).decode()),write=True) diff --git a/src/msg/storage/sqlite.py b/src/msg/storage/sqlite.py index 880def3d..b67cac03 100644 --- a/src/msg/storage/sqlite.py +++ b/src/msg/storage/sqlite.py @@ -1,4 +1,4 @@ -"""One transactional metadata session; handlers never own commit rights.""" +"""SQLite test adapter; shared metadata invariants live in storage.session.""" from __future__ import annotations import asyncio @@ -9,16 +9,10 @@ import tempfile import uuid from contextlib import asynccontextmanager -from dataclasses import replace from pathlib import Path -from msg.core.codec import canonical, decode, digest, loads, wire from msg.core.errors import Failure, require -from msg.core.models import ( - AuditEvent, Certificate, CertificateRequest, CertificateRequestState, Credential, - EffectJob, EmailSettings, Event, Membership, OperationResult, Organization, Page, - Resource, ResourceRef, Revision, Subject, TransferChunk, TransferSession, -) +from msg.storage.session import MetadataSessionBase _SCHEMA = """ PRAGMA foreign_keys=ON; @@ -219,47 +213,10 @@ """ -class SqliteSession: - def __init__(self, connection, *, write: bool): - self._connection, self.write = connection, write - self.owner_task = asyncio.current_task() - self.closed = False - self.rollback_effects = [] - - def on_rollback(self, effect): - """Compensate external work if this transaction or savepoint aborts.""" - self.check(write=True) - self.rollback_effects.append(effect) - - async def run_rollback_effects(self, cause, start=0): - effects = self.rollback_effects[start:] - del self.rollback_effects[start:] - if not effects: - return - - async def compensate(): - for effect in reversed(effects): - try: - await effect() - except BaseException as exc: - # Keep the primary error and try every independent cleanup. - cause.add_note('rollback compensation failed: ' + type(exc).__name__) - - # A second cancellation must not abandon a pin operation (including its - # worker thread) and release the writer fence before it finishes. - pending = asyncio.create_task(compensate()) - while not pending.done(): - try: - await asyncio.shield(pending) - except asyncio.CancelledError: - pass - pending.result() - - def check(self, write=False): - require(not self.closed, "transaction_closed") - require(asyncio.current_task() is self.owner_task, "transaction_cross_task") - if write: - require(self.write, "read_only_transaction") +class SqliteSession(MetadataSessionBase): + def __init__(self, connection: sqlite3.Connection, *, write: bool): + super().__init__(write=write) + self._connection = connection def execute(self, sql, parameters=(), *, write=False): self.check(write) @@ -268,341 +225,6 @@ def execute(self, sql, parameters=(), *, write=False): except sqlite3.IntegrityError as exc: raise Failure("constraint_conflict") from exc - def one(self, sql, parameters=()): - return self.execute(sql, parameters).fetchone() - - def rows(self, sql, parameters=()): - return self.execute(sql, parameters).fetchall() - - async def resource(self, id): - row = self.one("SELECT body FROM resources WHERE id=?", (id,)) - require(row is not None, "not_found") - return decode(Resource, loads(row[0])) - - async def resolve(self, path): - require(isinstance(path, str) and path.startswith("/"), "invalid_path") - parts = path.rstrip("/").split("/")[1:] - require(all(p not in {".", ".."} and "\\" not in p and "\x00" not in p for p in parts), "invalid_path") - if len(parts) == 2 and parts[0] == "_id": - return (await self.resource(parts[1])).id - row = self.one("SELECT id FROM resources WHERE parent IS NULL") - require(row is not None, "not_initialized") - rid = row[0] - if path == "/": - return rid - for part in parts: - require(bool(part), "invalid_path") - row = self.one("SELECT id FROM resources WHERE parent=? AND name=?", (rid, part)) - require(row is not None, "not_found") - rid = row[0] - return rid - - async def path(self, id): - segments, seen = [], set() - r = await self.resource(id) - while r.parent is not None: - require(r.id not in seen, "parent_cycle") - seen.add(r.id) - segments.append(r.name) - r = await self.resource(r.parent) - return "/" + "/".join(reversed(segments)) - - async def ancestors(self, id): - result, seen = [], set() - r = await self.resource(id) - while r.parent is not None: - require(r.id not in seen, "parent_cycle") - seen.add(r.id) - r = await self.resource(r.parent) - result.append(r) - return tuple(reversed(result)) - - async def children(self, parent, cursor=None, limit=50): - require(type(limit) is int and 1 <= limit <= 500, "invalid_limit") - rows = self.rows("SELECT body FROM resources WHERE parent=? AND id>? ORDER BY id LIMIT ?", - (parent, cursor or "", limit + 1)) - values = tuple(decode(Resource, loads(row[0])) for row in rows[:limit]) - return Page(items=values, next_cursor=values[-1].id if len(rows) > limit else None) - - async def revision(self, ref): - rid = ref.revision or (await self.resource(ref.id)).revision - row = self.one("SELECT body FROM revisions WHERE id=? AND resource_id=?", (rid, ref.id)) - require(row is not None, "revision_not_found") - return decode(Revision, loads(row[0])) - - async def history(self, id, cursor=None, limit=50): - require(1 <= limit <= 500, "invalid_limit") - rows = self.rows("SELECT body FROM revisions WHERE resource_id=? AND id>? ORDER BY id LIMIT ?", - (id, cursor or "", limit+1)) - values = tuple(decode(Revision, loads(r[0])) for r in rows[:limit]) - return Page(items=values, next_cursor=values[-1].id if len(rows)>limit else None) - - async def insert(self, resource): - self.check(True) - if resource.parent is not None: - await self.resource(resource.parent) - data = wire(resource) - self.execute("INSERT INTO resources VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", ( - resource.id, resource.type, resource.name, resource.parent, resource.owner, - resource.group, resource.mode, resource.generation, resource.revision, - resource.state, data['created_at'], data['modified_at'], canonical(resource).decode()), write=True) - for tag in resource.tags: - self.execute("INSERT INTO resource_tags VALUES (?,?)", (resource.id,tag), write=True) - - async def replace(self, resource, expected_generation): - self.check(True) - old = await self.resource(resource.id) - require(old.generation == expected_generation, "generation_conflict", - details={"generation":old.generation, "revision":old.revision}) - require(resource.generation == expected_generation + 1, "invalid_generation") - require(resource.type == old.type and resource.created_at == old.created_at and - resource.created_by == old.created_by, "immutable_creation_fact") - if resource.parent is not None: - require(resource.parent != resource.id, "parent_cycle") - parent = await self.resource(resource.parent) - require(all(a.id != resource.id for a in await self.ancestors(parent.id)), "parent_cycle") - if any(getattr(resource,k)!=getattr(old,k) for k in ('parent','owner','group','mode','state')): - self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1) - data = wire(resource) - changed = self.execute("""UPDATE resources SET name=?,parent=?,owner=?,grp=?,mode=?, - generation=?,revision=?,state=?,modified_at=?,body=? WHERE id=? AND generation=?""", - (resource.name, resource.parent, resource.owner, resource.group, resource.mode, - resource.generation, resource.revision, resource.state, data['modified_at'], - canonical(resource).decode(), resource.id, expected_generation), write=True).rowcount - require(changed == 1, "generation_conflict") - if resource.tags != old.tags: - self.execute("DELETE FROM resource_tags WHERE resource_id=?", (resource.id,), write=True) - for tag in resource.tags: - self.execute("INSERT INTO resource_tags VALUES (?,?)", (resource.id,tag), write=True) - - async def append_revision(self, revision): - self.check(True) - for parent in revision.parents: - await self.revision(ResourceRef(id=revision.resource_id, revision=parent)) - self.execute("INSERT INTO revisions VALUES (?,?,?,?)", ( - revision.id, revision.resource_id, wire(revision.created_at), canonical(revision).decode()), write=True) - for relation in revision.relations: - self.execute("INSERT INTO relations VALUES (?,?,?,?,?,?)", ( - revision.id, revision.resource_id, relation.type, relation.target.id, - relation.target.revision, canonical(relation).decode()), write=True) - - async def subject(self, id): - row = self.one("SELECT body FROM identities WHERE id=? AND kind='subject'", (id,)) - require(row is not None, "subject_not_found") - return decode(Subject, loads(row[0])) - - async def organization(self, id): - row = self.one("SELECT body FROM identities WHERE id=? AND kind='organization'", (id,)) - require(row is not None, "group_not_found") - return decode(Organization, loads(row[0])) - - async def credential(self, id): - row = self.one("SELECT body FROM credentials WHERE id=?", (id,)) - require(row is not None, "credential_not_found") - return decode(Credential, loads(row[0])) - - async def certificate(self, id): - row = self.one("SELECT body FROM certificates WHERE id=?", (id,)) - require(row is not None, "certificate_not_found") - return decode(Certificate, loads(row[0])) - - async def certificate_revoked(self, id): - row = self.one("SELECT revoked FROM certificates WHERE id=?", (id,)) - require(row is not None, "certificate_not_found") - return bool(row[0]) - - async def memberships(self, subject): - if subject is None: - return () - rows=self.rows("""SELECT m.org,m.body FROM memberships m - JOIN resources r ON r.id=m.org - WHERE m.subject=? AND r.type='organization' AND r.state='active' - ORDER BY m.org""",(subject,)) - memberships=((org,decode(Membership,loads(body))) for org,body in rows) - # A stale membership row must not revive an archived/missing organization - # or borrow an organization/subject identifier from a mismatched body. - active=tuple(member for org,member in memberships if member.status=='active' - and member.organization_id==org and member.subject_id==subject - and org!='g_public') - row=self.one("SELECT body FROM identities WHERE id=? AND kind='subject'",(subject,)) - if row is None: - return active - identity=decode(Subject,loads(row[0])) - if identity.kind in {'registered','custodial','system'} and not identity.local_only: - # /&public is a virtual group for formal subjects. Historical rows - # may remain in storage, but never create a duplicate authorization. - return (Membership(organization_id='g_public',subject_id=subject, - role='member',version=0),*active) - return active - - async def request_result(self, subject, id, digest): - row = self.one("SELECT digest,body FROM results WHERE subject=? AND request_id=?", (subject,id)) - if row is None: - return None - require(row[0] == digest, "idempotency_conflict") - return decode(OperationResult, loads(row[1])) - - async def save_result(self, subject, digest, result): - from msg.storage.capacity import require_result_capacity - require_result_capacity(self) - self.execute("INSERT INTO results VALUES (?,?,?,?)", ( - subject, result.request_id, digest, canonical(result).decode()), write=True) - - async def update_identity(self, record, expected_generation): - self.check(True) - if isinstance(record, Membership): - self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1) - table, where, args = 'memberships', 'org=? AND subject=?', (record.organization_id,record.subject_id) - generation = record.version - elif isinstance(record, EmailSettings): - table, where, args = 'emails', 'subject=?', (record.subject_id,) - generation = expected_generation + 1 - else: - table, where, args = 'identities', 'id=?', (record.resource_id,) - generation = record.auth_version if isinstance(record, Subject) else record.membership_version - row = self.one(f"SELECT generation FROM {table} WHERE {where}", args) - require((row is None and expected_generation == -1) or - (row is not None and row[0] == expected_generation), "generation_conflict") - payload = canonical(record).decode() - if row is None: - if table == 'identities': - self.execute("INSERT INTO identities VALUES (?,?,?,?)", (*args, - 'subject' if isinstance(record,Subject) else 'organization', generation,payload), write=True) - elif table == 'memberships': - self.execute("INSERT INTO memberships VALUES (?,?,?,?)", (*args,generation,payload), write=True) - else: - self.execute("INSERT INTO emails VALUES (?,?,?)", (*args,generation,payload), write=True) - else: - self.execute(f"UPDATE {table} SET generation=?,body=? WHERE {where}", - (generation,payload,*args), write=True) - - async def save_credential(self, credential, expected_auth_version): - subject = await self.subject(credential.subject_id) - require(subject.auth_version == expected_auth_version, "auth_version_conflict") - old=self.one("SELECT body FROM credentials WHERE id=?",(credential.id,)) - if old is not None: - previous=decode(Credential,loads(old[0])) - require(previous.subject_id==credential.subject_id and previous.kind==credential.kind and - previous.verifier==credential.verifier,"credential_identity_immutable") - self.execute("INSERT INTO credentials VALUES (?,?,?) ON CONFLICT(id) DO UPDATE SET body=excluded.body", - (credential.id,credential.subject_id,canonical(credential).decode()),write=True) - - async def csr(self, id): - row = self.one("SELECT body FROM csrs WHERE id=?",(id,)) - require(row is not None,"csr_not_found") - return decode(CertificateRequest,loads(row[0])) - - async def csr_state(self, id): - row=self.one("SELECT state_body FROM csrs WHERE id=?",(id,)) - require(row is not None,"csr_not_found") - return decode(CertificateRequestState,loads(row[0])) - - async def save_csr(self, request): - state=CertificateRequestState(request_id=request.resource_id,status='pending',generation=0) - self.execute("INSERT INTO csrs VALUES (?,?,?,?,?)",(request.resource_id,0,'pending', - canonical(request).decode(),canonical(state).decode()),write=True) - - async def transition_csr(self, state, expected_generation): - require(state.generation == expected_generation+1,"invalid_generation") - old=await self.csr_state(state.request_id) - require(old.status=='pending',"csr_not_pending") - require(old.generation==expected_generation,"generation_conflict") - self.execute("UPDATE csrs SET generation=?,state=?,state_body=? WHERE id=? AND generation=?", - (state.generation,state.status,canonical(state).decode(),state.request_id,expected_generation),write=True) - - async def register_certificate(self, certificate, csr_id, expected_generation): - self.execute("INSERT INTO certificates VALUES (?,?,?,?,?)",( - certificate.resource_id,certificate.subject_id,certificate.parent_certificate_id,0, - canonical(certificate).decode()),write=True) - if csr_id is not None: - self.set_setting('certificate_request:'+certificate.resource_id,csr_id) - await self.transition_csr(CertificateRequestState(request_id=csr_id,status='issued', - generation=expected_generation+1,certificate_id=certificate.resource_id),expected_generation) - - async def revoke_certificate(self, id, event): - await self.certificate(id) - self.execute("UPDATE certificates SET revoked=1 WHERE id=?",(id,),write=True) - self.set_setting('authorization_epoch',self.setting('authorization_epoch',0)+1) - await self.append_audit(event) - - async def transfer(self,id): - row=self.one("SELECT body FROM transfers WHERE id=?",(id,)) - require(row is not None,"transfer_not_found") - return decode(TransferSession,loads(row[0])) - - async def save_transfer(self,session,expected_generation): - if expected_generation is None: - self.execute("INSERT INTO transfers (id,subject,generation,body) VALUES (?,?,?,?)", - (session.id,session.subject_id,session.generation,canonical(session).decode()),write=True) - else: - require(session.generation==expected_generation+1,"invalid_generation") - count=self.execute("UPDATE transfers SET generation=?,body=? WHERE id=? AND generation=?", - (session.generation,canonical(session).decode(),session.id,expected_generation),write=True).rowcount - require(count==1,"generation_conflict") - - async def put_chunk(self,chunk): - self.check(True) - start,end=chunk.offset,chunk.offset+chunk.content.size - rows=self.rows("SELECT offset,length,body FROM chunks WHERE transfer_id=? AND offset?", - (chunk.transfer_id,end,start)) - if rows: - require(len(rows)==1 and rows[0][0]==start and rows[0][1]==chunk.content.size and - decode(TransferChunk,loads(rows[0][2])).content==chunk.content,"chunk_conflict") - return - self.execute("INSERT INTO chunks VALUES (?,?,?,?)",(chunk.transfer_id,start,chunk.content.size, - canonical(chunk).decode()),write=True) - - async def missing_ranges(self,id,cursor=None,limit=50): - session=await self.transfer(id) - require(session.expected_size is not None,"size_required") - require(1<=limit<=500,"invalid_limit") - start=int(cursor or 0) - require(0<=start<=session.expected_size,"invalid_cursor") - gaps=[] - for row in self.execute("SELECT offset,length FROM chunks WHERE transfer_id=? AND offset+length>? ORDER BY offset",(id,start)): - if row[0]>start: - gaps.append((start,row[0])) - if len(gaps)>limit: - return Page(items=tuple(gaps[:limit]),next_cursor=str(gaps[limit][0])) - start=max(start,row[0]+row[1]) - if startlimit else None) - - async def append_event(self,event): - self.execute("INSERT INTO events (id,body) VALUES (?,?)",(event.id,canonical(event).decode()),write=True) - - async def append_audit(self,event): - previous=self.one("SELECT digest FROM audit ORDER BY seq DESC LIMIT 1") - prev=previous[0] if previous else None - data=wire(replace(event,previous_digest=prev,entry_digest='')) - data.pop('entry_digest') - entry=replace(event,previous_digest=prev,entry_digest=digest(data)) - self.execute("INSERT INTO audit (digest,previous,body) VALUES (?,?,?)", - (entry.entry_digest,prev,canonical(entry).decode()),write=True) - - async def enqueue(self,job): - self.execute("INSERT INTO jobs VALUES (?,?,?,?,?,?)",(job.id,job.dedupe_key,job.kind, - job.state,wire(job.next_attempt_at),canonical(job).decode()),write=True) - - async def job(self,id): - row=self.one("SELECT body FROM jobs WHERE id=?",(id,)) - require(row is not None,'job_not_found') - return decode(EffectJob,loads(row[0])) - - async def save_job(self,job): - self.execute("UPDATE jobs SET state=?,next_at=?,body=? WHERE id=?",(job.state, - wire(job.next_attempt_at),canonical(job).decode(),job.id),write=True) - - def setting(self,key,default=None): - row=self.one("SELECT value FROM settings WHERE key=?",(key,)) - return loads(row[0]) if row else default - - def set_setting(self,key,value): - self.execute("INSERT INTO settings VALUES (?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", - (key,canonical(value).decode()),write=True) - - class SqliteMetadataStore: def __init__(self,path: Path, *, busy_timeout: float=10): self.path=Path(path) From 0ef416ec99e20604de2398622e72d7f479a84ceb Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Tue, 29 Sep 2026 05:02:57 +0800 Subject: [PATCH 5/5] ci: remove one-shot assembly tooling before final metadata validation --- .../workflows/assemble-metadata-session.yml | 76 ------------------- 1 file changed, 76 deletions(-) delete mode 100644 .github/workflows/assemble-metadata-session.yml diff --git a/.github/workflows/assemble-metadata-session.yml b/.github/workflows/assemble-metadata-session.yml deleted file mode 100644 index a108093f..00000000 --- a/.github/workflows/assemble-metadata-session.yml +++ /dev/null @@ -1,76 +0,0 @@ -name: Assemble reviewed metadata session extraction -on: - push: - branches: [build/metadata-session-boundary-166] - paths: [.github/workflows/assemble-metadata-session.yml] -permissions: - contents: write -jobs: - assemble: - if: github.repository == 'TokenNotIncluded/msg.lmm.best' && github.ref == 'refs/heads/build/metadata-session-boundary-166' - runs-on: ubuntu-latest - timeout-minutes: 5 - steps: - - uses: actions/checkout@v7 - with: - fetch-depth: 0 - - name: Require the unchanged regression-first source - env: - INPUT_SHA: ${{ github.sha }} - run: | - test "$(git rev-parse HEAD)" = "$INPUT_SHA" - git merge-base --is-ancestor 7f8752e37fe5b5a8021f192b87e731caec2bbc2d HEAD - git diff --exit-code 7f8752e37fe5b5a8021f192b87e731caec2bbc2d -- src tests docs - - name: Replay the exact audited extraction with its trailing-blank-line fix - run: | - python - <<'PY' - import subprocess, textwrap - recipe_commit='1dc01c0b120642fd0438266085109cdaa6132d46' - recipe_path='.github/workflows/assemble-metadata-session.yml' - workflow=subprocess.check_output(['git','show',recipe_commit+':'+recipe_path],text=True) - step=workflow.index(' - name: Extract reviewed session methods without executing project code\n') - opening=" python - <<'PY'\n" - begin=workflow.index(opening,step)+len(opening) - end=workflow.index(' PY\n',begin) - source=textwrap.dedent(workflow[begin:end]) - old="after['common']=common_header+models+'\\n'+cursor+common+'\\n'" - new="after['common']=common_header+models+'\\n'+cursor+common.rstrip()+'\\n'" - assert source.count(old)==1, 'unexpected pinned recipe' - source=source.replace(old,new) - exec(compile(source,recipe_commit+':metadata-session-extraction','exec')) - PY - - name: Commit only the audited source extraction - run: | - git add src/msg/storage/session.py src/msg/storage/sqlite.py src/msg/storage/postgres.py docs/METADATA_SESSION_PROGRESS.md - git diff --cached --check - python - <<'PY' - import json, subprocess - from pathlib import Path - expected=set(json.loads(Path('/tmp/metadata-session-assembly/audit.json').read_text())['changed_paths']) - actual=set(subprocess.check_output(['git','diff','--cached','--name-only'],text=True).splitlines()) - assert actual == expected, (actual,expected) - PY - git -c user.name='github-actions[bot]' -c user.email='41898282+github-actions[bot]@users.noreply.github.com' commit -m 'refactor: separate shared metadata invariants from concrete database sessions' - - name: Publish dedicated build branch without force or workflow changes - env: - INPUT_SHA: ${{ github.sha }} - run: | - test "$(git ls-remote origin refs/heads/build/metadata-session-boundary-166 | cut -f1)" = "$INPUT_SHA" - git push origin HEAD:refs/heads/build/metadata-session-boundary-166 - python - <<'PY' - import json, os, subprocess - from pathlib import Path - git=lambda *args: subprocess.check_output(['git',*args],text=True).strip() - Path('/tmp/metadata-session-assembly/source.json').write_text(json.dumps({ - 'input':os.environ['INPUT_SHA'],'commit':git('rev-parse','HEAD'), - 'tree':git('rev-parse','HEAD^{tree}'),'dirty':bool(git('status','--porcelain')), - 'recipe_commit':'1dc01c0b120642fd0438266085109cdaa6132d46', - 'recipe_adjustment':'trim one extra blank line at common module EOF'},indent=2)+'\n') - PY - git archive --format=tar.gz -o /tmp/metadata-session-assembly/source.tar.gz HEAD - - uses: actions/upload-artifact@v4 - if: always() - with: - name: metadata-session-assembly - path: /tmp/metadata-session-assembly/ - if-no-files-found: error