diff --git a/.github/workflows/collaboration-list-bounds.yml b/.github/workflows/collaboration-list-bounds.yml new file mode 100644 index 00000000..82f2c388 --- /dev/null +++ b/.github/workflows/collaboration-list-bounds.yml @@ -0,0 +1,72 @@ +name: Bounded collaboration Resource reads +on: + push: + branches: [fix/collaboration-list-bounds-20260929] + pull_request: + paths: + - '.github/workflows/collaboration-list-bounds.yml' + - 'src/msg/plugins/collaboration_resources.py' + - 'tests/test_collaboration_list_bounds.py' + - 'tests/test_collaboration_list_deadline.py' + workflow_dispatch: +permissions: + contents: read +concurrency: + group: work-list-${{ github.event.pull_request.head.repo.full_name || github.repository }}-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true +jobs: + regression: + runs-on: ubuntu-latest + timeout-minutes: 10 + 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 -d postgres" + --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 + - name: Record exact source + run: | + mkdir -p artifacts + python3 - <<'PY' + import json, os, subprocess + from pathlib import Path + def git(*args): + return 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}'), + 'run_id': os.environ['GITHUB_RUN_ID'], 'attempt': os.environ['GITHUB_RUN_ATTEMPT']}, indent=2) + '\n') + PY + - uses: actions/setup-python@v7 + with: + python-version: '3.15' + allow-prereleases: true + cache: pip + - name: Install real dependencies + run: | + sudo apt-get update -qq + sudo env NEEDRESTART_MODE=l apt-get install -y -qq age + python -m pip install -e '.[dev]' + - name: Run real bounded work-record regressions + run: >- + python -m pytest -q + tests/test_collaboration_list_bounds.py tests/test_collaboration_list_deadline.py + tests/test_collaboration_resources.py tests/test_collaboration_resource_surfaces.py + tests/test_proposals.py + --junitxml=artifacts/work-list-regressions.xml + - uses: actions/upload-artifact@v4 + if: always() + with: + name: collaboration-list-evidence + path: artifacts/ + if-no-files-found: error diff --git a/.github/workflows/integration-regressions.yml b/.github/workflows/integration-regressions.yml index 4f519716..6d4d96b7 100644 --- a/.github/workflows/integration-regressions.yml +++ b/.github/workflows/integration-regressions.yml @@ -4,19 +4,42 @@ on: paths: - '.github/workflows/integration-regressions.yml' - 'src/msg/transports/http_routes.py' + - 'src/msg/transports/http.py' + - 'src/msg/plugins/discovery.py' + - 'src/msg/plugins/read_predicates.py' + - 'src/msg/security/quarantine.py' + - 'src/msg/core/executor.py' - 'src/msg/plugins/communication.py' - 'tests/test_http_effect_order.py' + - 'tests/test_read_entry_contract_matrix.py' + - 'tests/test_read_query_keyset_boundary.py' + - 'tests/test_recovery_runtime_generation.py' + - 'tests/read_only_evidence.py' - 'tests/test_legacy_routes_removed.py' - 'tests/test_webhook*.py' pull_request: paths: - '.github/workflows/integration-regressions.yml' - 'src/msg/transports/http_routes.py' + - 'src/msg/transports/http.py' + - 'src/msg/plugins/discovery.py' + - 'src/msg/plugins/read_predicates.py' + - 'src/msg/security/quarantine.py' + - 'src/msg/core/executor.py' - 'src/msg/plugins/communication.py' - 'tests/test_http_effect_order.py' + - 'tests/test_read_entry_contract_matrix.py' + - 'tests/test_read_query_keyset_boundary.py' + - 'tests/test_recovery_runtime_generation.py' + - 'tests/read_only_evidence.py' - 'tests/test_legacy_routes_removed.py' - 'tests/test_webhook*.py' workflow_dispatch: + inputs: + expected_sha: + description: 'Optional full commit SHA; fail before running code if the selected ref moved' + required: false + type: string permissions: contents: read concurrency: @@ -26,6 +49,10 @@ jobs: regression: runs-on: ubuntu-latest timeout-minutes: 10 + strategy: + fail-fast: false + matrix: + locale: ['C', 'en_US.utf8'] services: postgres: image: postgres:16 @@ -33,10 +60,11 @@ jobs: POSTGRES_USER: msgtest POSTGRES_PASSWORD: msgtest POSTGRES_DB: postgres + POSTGRES_INITDB_ARGS: --encoding=UTF8 --locale=${{ matrix.locale }} ports: - 5432:5432 options: >- - --health-cmd "pg_isready -U msgtest" + --health-cmd "pg_isready -U msgtest -d postgres" --health-interval 10s --health-timeout 5s --health-retries 5 env: MSG_TEST_POSTGRES_URL_TEMPLATE: postgresql://msgtest:msgtest@127.0.0.1:5432/{database} @@ -45,7 +73,13 @@ jobs: with: persist-credentials: false - name: Record exact source identity + env: + EXPECTED_SHA: ${{ inputs.expected_sha }} + TEST_DATABASE_LOCALE: ${{ matrix.locale }} run: | + if [ -n "$EXPECTED_SHA" ]; then + test "$(git rev-parse HEAD)" = "$EXPECTED_SHA" + fi mkdir -p artifacts python3 - <<'PY' import json, os, subprocess @@ -54,12 +88,13 @@ jobs: return 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}'), - 'run_id': os.environ['GITHUB_RUN_ID'], 'attempt': os.environ['GITHUB_RUN_ATTEMPT']}, indent=2) + '\n') + 'run_id': os.environ['GITHUB_RUN_ID'], 'attempt': os.environ['GITHUB_RUN_ATTEMPT'], + 'requested_database_locale': os.environ['TEST_DATABASE_LOCALE']}, indent=2) + '\n') PY - name: Preserve reviewed public source even when tests fail uses: actions/upload-artifact@v4 with: - name: integration-regression-source + name: integration-regression-source-${{ matrix.locale }} path: | artifacts/source.json src/ @@ -79,10 +114,39 @@ jobs: sudo apt-get update -qq sudo env NEEDRESTART_MODE=l apt-get install -y -qq age python -m pip install -e '.[dev]' - - name: Run effect-order, path and real webhook regressions + - name: Record and require the actual database locale + env: + EXPECTED_DATABASE_LOCALE: ${{ matrix.locale }} + run: | + python - <<'PY' + import json, os, platform + from pathlib import Path + import psycopg + dsn = os.environ['MSG_TEST_POSTGRES_URL_TEMPLATE'].format(database='postgres') + with psycopg.connect(dsn) as connection: + server_version = connection.execute('SHOW server_version').fetchone()[0] + locale_row = connection.execute( + "SELECT datcollate,datctype,pg_encoding_to_char(encoding) " + "FROM pg_database WHERE datname=current_database()" + ).fetchone() + assert locale_row is not None + values = { + 'server_version': server_version, + 'lc_collate': locale_row[0], + 'lc_ctype': locale_row[1], + 'server_encoding': locale_row[2], + 'python': platform.python_version(), + } + Path('artifacts/database.json').write_text(json.dumps(values, indent=2) + '\n') + assert values['lc_collate'] == os.environ['EXPECTED_DATABASE_LOCALE'], values + assert values['server_encoding'] == 'UTF8', values + PY + - name: Run recovery, keyset, effect-order, path and webhook regressions run: >- python -m pytest -q tests/test_http_effect_order.py tests/test_legacy_routes_removed.py + tests/test_read_query_keyset_boundary.py tests/test_read_entry_contract_matrix.py + tests/test_recovery_runtime_generation.py tests/test_stable_resource_views.py tests/test_canonical_post_paths.py tests/test_webhook_overlap.py tests/test_webhook_domain_events.py tests/test_webhook_delivery.py tests/test_webhook_identity.py @@ -90,6 +154,6 @@ jobs: - uses: actions/upload-artifact@v4 if: always() with: - name: integration-regression-evidence + name: integration-regression-evidence-${{ matrix.locale }} path: artifacts/ if-no-files-found: error diff --git a/.github/workflows/lease-list-bounds.yml b/.github/workflows/lease-list-bounds.yml new file mode 100644 index 00000000..f9bbac6e --- /dev/null +++ b/.github/workflows/lease-list-bounds.yml @@ -0,0 +1,73 @@ +name: Bounded collaboration lease reads +on: + push: + branches: [fix/lease-list-bounds-20260929] + pull_request: + paths: + - '.github/workflows/lease-list-bounds.yml' + - 'src/msg/plugins/collaboration.py' + - 'tests/test_lease_list_bounds.py' + - 'tests/test_lease_list_deadline.py' + workflow_dispatch: +permissions: + contents: read +concurrency: + group: lease-list-${{ github.event.pull_request.head.repo.full_name || github.repository }}-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true +jobs: + regression: + runs-on: ubuntu-latest + timeout-minutes: 10 + 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 -d postgres" + --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 + - name: Record exact source + run: | + mkdir -p artifacts + python3 - <<'PY' + import json, os, subprocess + from pathlib import Path + def git(*args): + return 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}'), + 'run_id': os.environ['GITHUB_RUN_ID'], 'attempt': os.environ['GITHUB_RUN_ATTEMPT']}, indent=2) + '\n') + PY + - uses: actions/setup-python@v7 + with: + python-version: '3.15' + allow-prereleases: true + cache: pip + - name: Install real dependencies + run: | + sudo apt-get update -qq + sudo env NEEDRESTART_MODE=l apt-get install -y -qq age + python -m pip install -e '.[dev]' + - name: Require bounded reads and existing collaboration contracts + run: >- + python -m pytest -q + tests/test_lease_list_bounds.py tests/test_lease_list_deadline.py + tests/test_collaboration_current_visibility.py + tests/test_collaboration_surfaces.py tests/test_collaboration_resources.py + tests/test_collaboration_resource_surfaces.py + --junitxml=artifacts/lease-list-regressions.xml + - uses: actions/upload-artifact@v4 + if: always() + with: + name: lease-list-evidence + path: artifacts/ + if-no-files-found: error diff --git a/docs/INTEGRATION_REGRESSION_PROGRESS.md b/docs/INTEGRATION_REGRESSION_PROGRESS.md index 071f3fc5..2327f4d2 100644 --- a/docs/INTEGRATION_REGRESSION_PROGRESS.md +++ b/docs/INTEGRATION_REGRESSION_PROGRESS.md @@ -1,33 +1,82 @@ # Integration regression closure — progress -Refs #78, #80, #81, #82, #83, #85. Narrow follow-up to PR #176; combines PR #177 -and #178 without changing the active integration branch. No broad closure or -production operation is claimed. +Refs #78, #80, #81, #82, #83, #85. Integration entry: PR #176. +Latest narrow follow-up: PR #180. Live verification, exact final head and merge +state: #176 comment 5890775291 and #180. This note is a timestamped handoff, +not a second completion authority or a claim of production acceptance. -## Sources and completed work +## 2026-09-29 13:05Z: current repair -Integration base: `45470b5a46bccfc4f12ef27acccecc2f3bafce47`, tree -`e75e7cf511a89d72bfdccf5ec39567824b569275`. +Input integration head `de3cd872b71e062d6141916b11ea0bdfa28adca6`. +Actual previous cloud checkout `9b727c16b893959e358811e543233df1608abb7b`, +tree `4936af96eb805c4f90aad002008376425a2effa4`. -The merge commit retains both independent histories: -- #177 head `769c124c83a01a648d3d30ef5d4a19fda98bedb6`: bounded transfer status - cursors. Focused cloud run `36561282522`: 24/24, no failures/errors/skips. -- #178 head `517f7e296fa7ef504d2066980454654ed0634598`: positive integer client - chunk-size checks. RED run `36561926956`: 14 intended failures, 9 passes; - final focused run pending. No protocol/journal/schema changes. +Independently downloaded all four artifacts of run **36564681470**, checked +provider SHA256, each JUnit msg.nodeid against selected/all, common collection, +exact disjoint union and no duplicates: **2407 unique cases, 6 failures, +0 errors, 0 skipped**. Four effect-order failures and two signed descending +read_query@1/@3 failures; the earlier webhook-overlap failures are absent. -## Base full-suite evidence +| Shard | Artifact | ZIP SHA256 | +| --- | --- | --- | +| 0 | 11031896647 | fd8c2110ab4240246af4d3b323b5676b7458af5016b6ed18c940010772a691e0 | +| 1 | 11031762212 | 95bf9181ea1c39acd3d9ea62edc0d194104e116d93a699b38c01a53d704f0dbb | +| 2 | 11031389353 | 7a00d14bea7a34c8d1439f936e8d78813c9d76f973fec0e3e518e5ffd4c52a1d | +| 3 | 11031559533 | d7c9c081408ad09b01be264e3e3ac6df4d84f020a4bab4d9f2d2f3eca5ddc5df | + +### Submitted implementation + +Source commit `12e829a6a7b90c279e9b15cd74cece63978ca387`, tree +`07449e7ae96399ebcbcd1ae33a53de3d9ab5d8f5`, in #180 based on #176: + +- Remove artificial initial empty-string/U+FFFF keyset endpoints. No seek on + the first page; later scans/cursors keep actual SQL keys. Ordering, principal, + query/snapshot binding, authorization and budgets are unchanged. +- Retain the real runtime-generation SQL read in the effect test. Precisely + forbid every other SQL query and executor call, check each request actually + reads generation, and also test stale generations fail closed. Recovery + middleware is not weakened, bypassed or mocked to succeed. +- Add list/search/read_query@1/@2/@3 checks for all three sort keys in both + directions, Unicode names, exact database order, nonoverlapping continuation + and empty results. Existing signed four-transport tests remain unchanged. + +Source-only preparation run 36571871534 published just one reviewed source +blob. Artifact11035120921 ZIP SHA256 +`02f5421eb68c126198dd1fe0f6936918f339470d9350fa201846a5ed0434d47d` was checked +independently, including byte-for-byte source equality. Source blob +`985c0c222220bfc23b0e28b4e54871bef5e7885d`; SHA256 +`2cc2b605e6590d61e0f65e63f613f75e1488b2d40c77224d1b77b4012694ffc0`. +The auxiliary workflow neither ran project code nor updated a branch and is +absent from the product tree. -Run `36559785404` executed 2259 unique test node IDs in four disjoint partitions. -All four manifests and JUnit node IDs were independently compared: exact union, -no missing or duplicate cases. Four failures, no errors or skipped test cases: +### Verification in progress; no GREEN claim yet -- `test_legacy_routes_removed::test_public_views_require_read_effects`. -- `test_webhook_overlap::test_overlap_one_job_per_event_recipient_channel_and_pinned_revocation[endpoint]`. -- The same webhook case with `[unsubscribe]`. -- `test_webhook_overlap::test_different_recipients_and_channels_remain_distinct`. +Initial full run 36572191345 and integration run 36572191005 were triggered. +Focused run 36572191224 failed in the newly added collation diagnostic before +pytest: lc_collate is not a supported current_setting parameter. The follow-up +reads pg_database.datcollate for current_database(), as documented in the +PostgreSQL 16 system catalog, and fixes the health probe's database name. +Runtime source and test assertions are unchanged by that CI correction. +New runs must identify their actual final head/tree; older partial or successful +runs do not satisfy the final combined gate. -Artifacts were independently downloaded and SHA256-verified: +## Preserved earlier history + +Earlier integration base `45470b5a46bccfc4f12ef27acccecc2f3bafce47`, tree +`e75e7cf511a89d72bfdccf5ec39567824b569275`. PR #179 retained both histories +and was merged into #176 as de3cd872; no duplicate implementations are needed: + +- #177 head `769c124c83a01a648d3d30ef5d4a19fda98bedb6`: bounded transfer + status cursors. Historical focused run 36561282522: 24/24, zero failures, + errors or skips. See its original evidence note for full provenance. +- #178 head `517f7e296fa7ef504d2066980454654ed0634598`: positive integer + client chunk sizes. Historical RED36561926956: 14 intended failures, + 9 passes. Original transfer/client evidence remains authoritative for those + narrow changes, not current integrated acceptance. + +Earlier full run36559785404 had 2259 unique cases, four failures, no errors or +skips: one read-effect-before-lookup case and three webhook-overlap cases. +Exact manifests/JUnit union and provider hashes were independently checked: | Shard | Artifact | ZIP SHA256 | | --- | --- | --- | @@ -36,29 +85,19 @@ Artifacts were independently downloaded and SHA256-verified: | 2 | 11029802392 | 18fbb03c4531e3128ffe1e0de45602aede0532132406c1b96388ed5c934b7df2 | | 3 | 11030251180 | e2d4990543cbf937af745f0b9429c260c024a64dee51065561a652e2171c7370 | -The full gate therefore fails; the later conformance/build stages are not -claimed complete. The same read-route failure exists on #177's shard2 run -`36561282628` (577 cases, one failure), independent of its green focused tests. - -## Remaining corrections - -1. `http_routes.py` resolves ordinary resource paths before checking the - selected discovery operation's read effect. A missing path returns 404 - instead of rejecting changed effects with 405. Move the shared effect check - before lookups and retain current request/resource authorization. New tests - cover GET/HEAD, transaction/external effects and metadata/executor guards. -2. `communication.py::_domain_delivery_exists` uses a textual punctuation range - for prefix matching. The real PostgreSQL overlap tests hit jobs_dedupe_key - conflicts. Investigate collation-safe literal prefix matching while retaining - historical scope-key recognition, exact current authorization and pinned - endpoint/subscription generations. Never turn unique conflicts into a blanket - swallowed error or weaken the unique constraint. - -The focused workflow preserves the exact public checkout as an artifact before -running tests, so failures can be reviewed without relying on an older sdist. -It has contents:read only and never pushes commits or touches production data. - -Status: combined histories, regression-first checks and workflow prepared. -Runtime read-route and webhook corrections, exact final-head focused verification, -full four-shard gate, conformance and build are pending. All project execution is -in cloud CI; local work only edits text and inspects/hashes artifacts. +That historical run was RED, with later conformance/build not claimed complete. +#179 moved read-effect checks before resource lookups and corrected literal +webhook prefix matching without removing unique constraints or swallowing +conflicts. The current combined six failures above supersede that old triage. + +## Unchanged completion and execution boundaries + +Require final focused suites, all four shards, exact disjoint node-ID coverage, +conformance/actual tokenizer, wheel/sdist checks and applicable specialist gates. +No removed tests, skips, relaxed authentication or substituted old-head greens. +All project execution happens in cloud CI; local work only edits source text +and independently reads/hashes artifacts. + +#78/#80/#81/#82/#83/#85 broad acceptance and #64/#65/#68–70/#84 field evidence +remain open. No production deployment, Root/PIN access, money movement, real +outbound delivery, backup/key destruction or recovery promotion is performed. diff --git a/src/msg/plugins/collaboration.py b/src/msg/plugins/collaboration.py index 90c20cc8..f978dc61 100644 --- a/src/msg/plugins/collaboration.py +++ b/src/msg/plugins/collaboration.py @@ -2,6 +2,7 @@ from __future__ import annotations from datetime import timedelta +import time from msg.core.codec import canonical, loads, parse_time, wire from msg.core.errors import Failure, require @@ -203,27 +204,37 @@ async def lease_get(ctx, request, tx): async def lease_list(ctx, request, tx): holder = await _self(app, ctx, request, tx) limit = request.arguments.get('limit', 50) - rows = tx.rows('SELECT id,target,status,expires_at,body FROM collaboration_leases ' - 'WHERE holder=? AND id>? ORDER BY id', - (holder, request.arguments.get('after', ''))) + position = request.arguments.get('after', '') records = [] more = False - for _, target, status, expires, raw in rows: - try: - await _safe_ref(app, ctx, request, tx, target) - except Failure as exc: - if exc.code in {'permission_denied', 'credential_ceiling', 'certificate_gate', - 'delegation_scope', 'ancestor_inactive', 'not_found', - 'collaboration_ref_forbidden'}: - continue - raise - record = loads(raw) - record['effective_status'] = ('expired' if status == 'active' and - parse_time(expires) <= ctx.now else status) - if len(records) == limit: - more = True + scanned = 0 + while not more: + require(time.monotonic() < ctx.deadline_monotonic, 'query_cost_exceeded') + rows = tx.rows('SELECT id,target,status,expires_at,body FROM collaboration_leases ' + 'WHERE holder=? AND id>? ORDER BY id LIMIT 128', + (holder, position)) + for lease_id, target, status, expires, raw in rows: + scanned += 1 + require(scanned <= 4096 and time.monotonic() < ctx.deadline_monotonic, + 'query_cost_exceeded') + position = lease_id + try: + await _safe_ref(app, ctx, request, tx, target) + except Failure as exc: + if exc.code in {'permission_denied', 'credential_ceiling', 'certificate_gate', + 'delegation_scope', 'ancestor_inactive', 'not_found', + 'collaboration_ref_forbidden'}: + continue + raise + if len(records) == limit: + more = True + break + record = loads(raw) + record['effective_status'] = ('expired' if status == 'active' and + parse_time(expires) <= ctx.now else status) + records.append(record) + if len(rows) < 128: break - records.append(record) return HandlerOutput(data={'items': records, 'next_after': records[-1]['id'] if more else None}) diff --git a/src/msg/plugins/collaboration_resources.py b/src/msg/plugins/collaboration_resources.py index b41bd177..edad323a 100644 --- a/src/msg/plugins/collaboration_resources.py +++ b/src/msg/plugins/collaboration_resources.py @@ -1,4 +1,6 @@ """Explicit collaboration records backed by ordinary Resource and Revision facts.""" +import time + from msg.core.codec import canonical, loads, parse_time, wire from msg.core.errors import Failure, require from msg.core.models import HandlerOutput, Relation, ResourceRef, ResourceTypeSpec @@ -155,23 +157,33 @@ async def get(ctx, request, tx): async def list_records(ctx, request, tx): subject = await _self(app, ctx, request, tx) limit = request.arguments.get('limit', 50) - rows = tx.rows('SELECT id FROM resources WHERE owner=? AND type=? ' - "AND state='active' AND id>? ORDER BY id", - (subject, KINDS[kind], request.arguments.get('after', ''))) + position = request.arguments.get('after', '') items = [] - for (rid,) in rows: - try: - await check_access(app, ctx, request, tx, rid, 'read') - except Failure as exc: - if exc.code in {'permission_denied', 'credential_ceiling', 'certificate_gate', - 'delegation_scope', 'ancestor_inactive', 'not_found'}: - continue - raise - resource = await tx.resource(rid) - revision = await tx.revision(ResourceRef(id=rid)) - record = loads(await app.contents.read_bytes(revision.content)) - items.append(await _project(app, ctx, request, tx, resource, revision, record)) - if len(items) > limit: + scanned = 0 + while len(items) <= limit: + require(time.monotonic() < ctx.deadline_monotonic, 'query_cost_exceeded') + rows = tx.rows('SELECT id FROM resources WHERE owner=? AND type=? ' + "AND state='active' AND id>? ORDER BY id LIMIT 128", + (subject, KINDS[kind], position)) + for (rid,) in rows: + scanned += 1 + require(scanned <= 4096 and time.monotonic() < ctx.deadline_monotonic, + 'query_cost_exceeded') + position = rid + try: + await check_access(app, ctx, request, tx, rid, 'read') + except Failure as exc: + if exc.code in {'permission_denied', 'credential_ceiling', 'certificate_gate', + 'delegation_scope', 'ancestor_inactive', 'not_found'}: + continue + raise + resource = await tx.resource(rid) + revision = await tx.revision(ResourceRef(id=rid)) + record = loads(await app.contents.read_bytes(revision.content)) + items.append(await _project(app, ctx, request, tx, resource, revision, record)) + if len(items) > limit: + break + if len(rows) < 128: break return HandlerOutput(data={'items': items[:limit], 'next_after': items[limit - 1]['id'] if len(items) > limit else None}) diff --git a/src/msg/plugins/discovery.py b/src/msg/plugins/discovery.py index 56c155c7..985c0c22 100644 --- a/src/msg/plugins/discovery.py +++ b/src/msg/plugins/discovery.py @@ -505,7 +505,7 @@ async def list_items(ctx,request,tx, *, arguments=None,budget=None): position,snapshot=app.cursors.decode_page(a['cursor'],request.operation, query_args,principal,ctx.now) else: - position=app.cursors.decode(a['cursor'],'page',query_hash) if a.get('cursor') else (['\uffff','\uffff'] if descending else ['', '']) + position=app.cursors.decode(a['cursor'],'page',query_hash) if a.get('cursor') else None snapshot=ctx.now from msg.plugins.read_predicates import read_predicates filters,parameters=await read_predicates(app,tx,a,parent) @@ -517,8 +517,11 @@ async def list_items(ctx,request,tx, *, arguments=None,budget=None): scanned=0 while len(values)<=limit: require(time.monotonic() 128: + violations.append('batch size=' + str(len(result))) + raise AssertionError('oversized collaboration Resource batch') + return result + + with monkeypatch.context() as patch: + patch.setattr(tx, 'rows', checked_rows) + yield tx + + with monkeypatch.context() as patch: + patch.setattr(app.metadata, 'transaction', checked_transaction) + yield batches, violations + + +@pytest.mark.asyncio +@pytest.mark.parametrize('kind,count,limit', [ + ('request', 3, 1), ('offer', 3, 1), ('checkpoint', 3, 1), ('proposal', 3, 1), + ('checkpoint', 129, 64), +]) +async def test_work_lists_use_bounded_sql_and_actual_continuations( + installed, monkeypatch, kind, count, limit): + app, _ = installed + key, subject, _ = await register(app, 'work-list-bounds') + ids = await create_records(app, key, subject, kind, count) + other_key, other, _ = await register(app, 'other-work-owner') + other_ids = await create_records(app, other_key, other, kind, 1) + async with app.metadata.transaction(write=False) as tx: + expected = [row[0] for row in tx.rows( + "SELECT id FROM resources WHERE owner=? AND type=? AND state='active' ORDER BY id", + (subject, KINDS[kind]))] + assert set(expected) == set(ids) + async with readonly_evidence(app, monkeypatch): + async with observe_batches(app, monkeypatch) as (batches, violations): + seen, after = [], None + for offset in range(0, count, limit): + args = {'limit': limit, **({'after': after} if after is not None else {})} + page = await call(app, 'communication.' + kind + '_list', args, + key=key, subject=subject) + assert not violations, 'unbounded work-list SQL: ' + repr(violations) + assert page.status == 'ok', str(wire(page)) + ids_on_page = [item['id'] for item in page.data['items']] + assert ids_on_page == expected[offset:offset + limit] + seen.extend(ids_on_page) + more = offset + limit < count + after = page.data['next_after'] + assert after == (ids_on_page[-1] if more else None) + assert all(identifier not in str(wire(page)) for identifier in other_ids) + assert seen == expected and len(set(seen)) == count + assert batches and max(batches) <= 128 + empty = await call(app, 'communication.' + kind + '_list', + {'limit': limit, 'after': expected[-1]}, key=key, subject=subject) + assert not violations, repr(violations) + assert empty.status == 'ok', str(wire(empty)) + assert not empty.data['items'] and empty.data['next_after'] is None diff --git a/tests/test_collaboration_list_deadline.py b/tests/test_collaboration_list_deadline.py new file mode 100644 index 00000000..0535a8fb --- /dev/null +++ b/tests/test_collaboration_list_deadline.py @@ -0,0 +1,35 @@ +"""Work-list deadlines are enforced before scanning and before projecting rows.""" +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from read_only_evidence import readonly_evidence +from test_collaboration_list_bounds import create_records, observe_batches +from test_service import call, register + +from msg.plugins import collaboration_resources + + +@pytest.mark.asyncio +@pytest.mark.parametrize('kind', ['request', 'offer', 'checkpoint', 'proposal']) +@pytest.mark.parametrize('stage', ['before_query', 'after_query']) +async def test_expired_work_list_budget_never_projects_content(installed, monkeypatch, kind, stage): + app, _ = installed + key, subject, _ = await register(app, 'work-list-deadline') + await create_records(app, key, subject, kind, 1) + # Only the work-list handler's clock changes. Authentication, recovery, + # principal selection, database reads and target ownership remain real. + readings = iter([float('inf')] if stage == 'before_query' else [0.0, float('inf')]) + projection = AsyncMock(side_effect=AssertionError('expired request projected work content')) + async with readonly_evidence(app, monkeypatch): + with monkeypatch.context() as patch: + patch.setattr(collaboration_resources, 'time', + SimpleNamespace(monotonic=lambda: next(readings, float('inf')))) + patch.setattr(collaboration_resources, '_project', projection) + async with observe_batches(app, monkeypatch) as (batches, violations): + result = await call(app, 'communication.' + kind + '_list', {'limit': 1}, + key=key, subject=subject) + assert not violations, repr(violations) + assert result.status == 'error' and result.error.code == 'query_cost_exceeded' + assert len(batches) == (0 if stage == 'before_query' else 1) + projection.assert_not_called() diff --git a/tests/test_http_effect_order.py b/tests/test_http_effect_order.py index 2163e57a..08bcdcaa 100644 --- a/tests/test_http_effect_order.py +++ b/tests/test_http_effect_order.py @@ -1,14 +1,61 @@ """Ordinary read views reject changed effects before resolving any resource.""" +from contextlib import asynccontextmanager from dataclasses import replace from unittest.mock import AsyncMock, Mock import httpx import pytest +from msg.security.quarantine import RUNTIME_GENERATION from msg.transports.http import create_app from read_only_evidence import readonly_evidence +@asynccontextmanager +async def runtime_check_only(app, patch): + """Keep the real database-backed freshness guard; forbid business queries. + + Blocking every transaction also blocks the outer HTTP recovery guard. This + narrow observer instead delegates only that guard's generation read to the + actual session. Resource lookups, other settings, SQL and writes still fail. + """ + transaction = app.metadata.transaction + settings_read = [] + transactions = [] + attempts = [] + + class RuntimeSession: + def __init__(self, session): + self._session = session + + def setting(self, key, default=None): + settings_read.append(key) + assert key == RUNTIME_GENERATION, 'effect gate read a business setting' + return self._session.setting(key, default) + + def __getattr__(self, name): + attempts.append(name) + raise AssertionError('effect gate attempted a business query: ' + name) + + @asynccontextmanager + async def guarded_transaction(*, write): + transactions.append(write) + assert write is False, 'effect gate opened a write transaction' + async with transaction(write=False) as session: + yield RuntimeSession(session) + + runtime = AsyncMock(wraps=app.executor.require_current_runtime) + execute = AsyncMock(side_effect=AssertionError('effect gate executed an operation')) + patch.setattr(app.metadata, 'transaction', guarded_transaction) + patch.setattr(app.executor, 'require_current_runtime', runtime) + patch.setattr(app.executor, 'execute', execute) + yield runtime + assert transactions == [False] * runtime.await_count + assert settings_read == [RUNTIME_GENERATION] * runtime.await_count + assert attempts == [], 'effect gate attempted a business query' + execute.assert_not_called() + + @pytest.mark.parametrize('effect', ['transaction', 'external']) @pytest.mark.parametrize('method', ['GET', 'HEAD']) async def test_non_read_views_do_not_resolve_or_execute(installed, monkeypatch, effect, method): @@ -23,16 +70,38 @@ async def test_non_read_views_do_not_resolve_or_execute(installed, monkeypatch, key = (operation, 1) patch.setitem(app.registry._operations, key, replace(app.registry._operations[key], effect=effect)) - lookup = Mock(side_effect=AssertionError('effect gate performed a resource lookup')) - execute = AsyncMock(side_effect=AssertionError('effect gate executed an operation')) - patch.setattr(app.metadata, 'transaction', lookup) - patch.setattr(app.executor, 'execute', execute) - for path in paths: - response = await http.request(method, path) - assert response.status_code == 405, (method, path, response.text) - if method == 'GET': - assert response.json()['error']['code'] == 'effect_mismatch' - else: - assert response.content == b'' - lookup.assert_not_called() - execute.assert_not_called() + async with runtime_check_only(app, patch) as runtime: + for path in paths: + response = await http.request(method, path) + assert response.status_code == 405, (method, path, response.text) + if method == 'GET': + assert response.json()['error']['code'] == 'effect_mismatch' + else: + assert response.content == b'' + assert runtime.await_count == len(paths) + + +@pytest.mark.parametrize('effect', ['transaction', 'external']) +async def test_stale_runtime_is_checked_before_the_public_effect_gate( + installed, monkeypatch, effect): + app, _ = installed + async with app.metadata.transaction(write=True) as tx: + tx.set_setting(RUNTIME_GENERATION, 'effect-order-new-generation') + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=create_app(app)), + base_url=app.settings.service_url) as http: + async with readonly_evidence(app, monkeypatch): + with monkeypatch.context() as patch: + for operation in ('discovery.get', 'discovery.raw'): + key = (operation, 1) + patch.setitem(app.registry._operations, key, + replace(app.registry._operations[key], effect=effect)) + operation_lookup = Mock(side_effect=AssertionError( + 'stale runtime reached the route effect gate')) + patch.setattr(app.registry, 'operation', operation_lookup) + async with runtime_check_only(app, patch) as runtime: + for path in ('/main', '/_id/missing/raw'): + response = await http.get(path) + assert response.status_code == 503 + assert response.json()['error']['code'] == 'recovery_runtime_stale' + assert runtime.await_count == 2 + operation_lookup.assert_not_called() diff --git a/tests/test_lease_list_bounds.py b/tests/test_lease_list_bounds.py new file mode 100644 index 00000000..4a4fcdc3 --- /dev/null +++ b/tests/test_lease_list_bounds.py @@ -0,0 +1,104 @@ +"""Legacy lease pages stay bounded while applying real current permissions.""" +from contextlib import asynccontextmanager +from datetime import timedelta + +import pytest +from read_only_evidence import readonly_evidence +from test_service import NOW, call, register + +from msg.core.codec import canonical, loads, wire + + +@pytest.mark.asyncio +@pytest.mark.parametrize('visible_count', [0, 3]) +async def test_lease_pages_bound_sql_and_cross_hidden_batches(installed, monkeypatch, visible_count): + app, _ = installed + owner_key, owner, _ = await register(app, 'lease-target-owner') + reader_key, reader, _ = await register(app, 'lease-list-reader') + targets, templates = [], [] + for name in ('hidden', 'visible'): + post = await call(app, 'content.post_create', + {'parent': '/main', 'name': name, 'body': name + '-lease-body'}, + key=owner_key, subject=owner) + assert post.status == 'ok', wire(post) + targets.append(post) + acquired = await call(app, 'communication.lease_acquire', + {'target': post.resources[0].id, 'purpose': name + '-lease-purpose', + 'expires_at': wire(NOW + timedelta(hours=1))}, + key=reader_key, subject=reader) + assert acquired.status == 'ok', wire(acquired) + async with app.metadata.transaction(write=False) as tx: + templates.append(loads(tx.one('SELECT body FROM collaboration_leases WHERE id=?', + (acquired.data['lease']['id'],))[0])) + hidden = targets[0] + changed = await call(app, 'content.chmod', + {'id': hidden.resources[0].id, 'mode': '0700'}, key=owner_key, subject=owner, + expected=((hidden.resources[0].id, hidden.data['generation']),)) + assert changed.status == 'ok', wire(changed) + + # Stress fixtures copy records originally created by real signed operations. + # They are test-only authoritative rows, not a production migration claim. + async with app.metadata.transaction(write=True) as tx: + tx.execute('DELETE FROM collaboration_leases WHERE holder=?', (reader,), write=True) + for index in range(129 + visible_count): + record = dict(templates[0 if index < 129 else 1]) + record['id'] = f'lease_page_{index:04d}' + if index == 130: + record.update(acquired_at=wire(NOW - timedelta(hours=2)), + expires_at=wire(NOW - timedelta(hours=1))) + if index == 131: + record.update(status='released', generation=2) + tx.execute('INSERT INTO collaboration_leases VALUES (?,?,?,?,?,?,?)', + (record['id'], reader, record['target'], record['status'], + record['generation'], record['expires_at'], canonical(record).decode()), write=True) + transaction = app.metadata.transaction + batches, unbounded_queries = [], [] + + @asynccontextmanager + async def bounded_transaction(*, write): + assert write is False + async with transaction(write=False) as tx: + query_rows = tx.rows + + def bounded_rows(sql, parameters=()): + lease_query = sql.startswith('SELECT id,target,status,expires_at,body FROM collaboration_leases ') + if lease_query and ' LIMIT ' not in sql.upper(): + unbounded_queries.append(sql) + raise AssertionError('lease_list issued an unbounded SELECT') + rows = query_rows(sql, parameters) + if lease_query: + assert len(rows) <= 128, 'lease_list materialized more than one bounded batch' + batches.append(len(rows)) + return rows + + with monkeypatch.context() as session_patch: + session_patch.setattr(tx, 'rows', bounded_rows) + yield tx + + async with readonly_evidence(app, monkeypatch): + with monkeypatch.context() as patch: + patch.setattr(app.metadata, 'transaction', bounded_transaction) + after = None + seen, statuses = [], [] + for page in range(max(1, visible_count)): + args = {'limit': 1} + if after is not None: + args['after'] = after + result = await call(app, 'communication.lease_list', args, + key=reader_key, subject=reader) + assert not unbounded_queries, 'unbounded lease SQL: ' + repr(unbounded_queries) + assert result.status == 'ok', str(wire(result)) + assert 'hidden-lease-purpose' not in str(wire(result)) + items = result.data['items'] + assert len(items) == (1 if visible_count else 0) + seen.extend(item['id'] for item in items) + statuses.extend(item['effective_status'] for item in items) + if page == 0: + assert len(batches) >= 2, 'must scan past the first fully hidden batch' + after = result.data['next_after'] + if page + 1 < visible_count: + assert after == items[-1]['id'] + else: + assert after is None + assert seen == [f'lease_page_{i:04d}' for i in range(129, 129 + visible_count)] + assert statuses == (['active', 'expired', 'released'] if visible_count else []) diff --git a/tests/test_lease_list_deadline.py b/tests/test_lease_list_deadline.py new file mode 100644 index 00000000..82bc788e --- /dev/null +++ b/tests/test_lease_list_deadline.py @@ -0,0 +1,44 @@ +"""Deadline exhaustion fails closed before legacy lease SQL materialization.""" +from contextlib import asynccontextmanager +from types import SimpleNamespace + +import pytest +from read_only_evidence import readonly_evidence +from test_service import call, register + +from msg.plugins import collaboration + + +@pytest.mark.asyncio +async def test_expired_lease_list_budget_refuses_before_query(installed, monkeypatch): + app, _ = installed + key, subject, _ = await register(app, 'lease-deadline') + transaction = app.metadata.transaction + attempted = [] + + @asynccontextmanager + async def no_lease_scan(*, write): + assert write is False + async with transaction(write=False) as tx: + query_rows = tx.rows + + def guard(sql, parameters=()): + if 'FROM collaboration_leases ' in sql: + attempted.append(sql) + raise AssertionError('expired request issued a lease scan') + return query_rows(sql, parameters) + + with monkeypatch.context() as session_patch: + session_patch.setattr(tx, 'rows', guard) + yield tx + + async with readonly_evidence(app, monkeypatch): + with monkeypatch.context() as patch: + # Only this handler sees an expired clock. Real signing, principal, + # runtime generation and current authorization are not replaced. + patch.setattr(collaboration, 'time', SimpleNamespace(monotonic=lambda: float('inf'))) + patch.setattr(app.metadata, 'transaction', no_lease_scan) + result = await call(app, 'communication.lease_list', {'limit': 1}, + key=key, subject=subject) + assert not attempted, 'lease scan ran after its deadline: ' + repr(attempted) + assert result.status == 'error' and result.error.code == 'query_cost_exceeded' diff --git a/tests/test_read_query_keyset_boundary.py b/tests/test_read_query_keyset_boundary.py new file mode 100644 index 00000000..ec6068b7 --- /dev/null +++ b/tests/test_read_query_keyset_boundary.py @@ -0,0 +1,82 @@ +"""Real first pages must not depend on a fabricated Unicode maximum key.""" +import pytest + +from msg.core.codec import wire +from read_only_evidence import readonly_evidence +from test_service import call, register + + +@pytest.mark.parametrize('operation,version', [ + ('discovery.list', 1), ('discovery.search', 1), + ('discovery.read_query', 1), ('discovery.read_query', 2), + ('discovery.read_query', 3), +]) +@pytest.mark.parametrize('sort', ['id', 'time', 'name']) +@pytest.mark.parametrize('direction', ['asc', 'desc']) +async def test_first_page_and_continuations_use_actual_database_keys( + installed, monkeypatch, operation, version, sort, direction): + app, _ = installed + key, subject, _ = await register(app, 'keyset-owner') + # The supplementary-plane name also exceeds U+FFFF in byte/codepoint + # ordering. Equal creation times exercise the existing ID tie breaker. + names = ['Alpha', 'zebra', 'éclair', '𐐀-tail', 'unmatched'] + expected_ids = [] + for index, name in enumerate(names): + result = await call(app, 'content.post_create', { + 'parent': '/main', 'name': name, + 'body': 'keysetneedle' if index < 4 else 'not a query match', + }, key=key, subject=subject) + assert result.status == 'ok', wire(result) + if index < 4: + expected_ids.append(result.resources[0].id) + column = {'id': 'id', 'time': 'created_at', 'name': 'name'}[sort] + ordering = 'ASC' if direction == 'asc' else 'DESC' + async with app.metadata.transaction(write=False) as tx: + # Independent SQL oracle has no synthetic first-page key. Use the + # database's existing collation, not Python's Unicode sort order. + expected = [row[0] for row in tx.rows( + f'SELECT id FROM resources WHERE id IN (?,?,?,?) ' + f'ORDER BY {column} {ordering},id {ordering}', tuple(expected_ids))] + assert len(expected) == 4 + arguments = {'parent': '/main', 'type': 'post', 'author': subject, + 'query': 'keysetneedle', 'sort': sort, 'direction': direction, + 'fields': ['id'], 'limit': 1} + if version == 3: + arguments['query_version'] = 3 + + async def read(args): + result = await call(app, operation, args, key=key, subject=subject, + contract_version=version) + assert result.status == 'ok', wire(result) + return result.data + + def continuation(cursor): + return ({'cursor': cursor} if operation == 'discovery.read_query' + else {**arguments, 'cursor': cursor}) + + async with readonly_evidence(app, monkeypatch): + query = arguments + found = [] + cursors = [] + for index, expected_id in enumerate(expected): + page = await read(query) + assert list(page['items']) == [{'id': expected_id}] + found.append(page['items'][0]['id']) + if index < len(expected) - 1: + assert page.get('next') + cursor = page['cursor'] + assert cursor not in cursors, 'cursor did not advance' + cursors.append(cursor) + query = continuation(cursor) + else: + assert not page.get('next') + if version == 3: + exhausted = await read(continuation(page['cursor'])) + assert list(exhausted['items']) == [] + assert exhausted['pageInfo']['hasNextPage'] is False + assert found == expected + assert len(found) == len(set(found)) + # Replaying a cursor repeats the same page, not an offset-dependent + # or newly invented boundary, and still goes through authorization. + repeated = await read(continuation(cursors[0])) + assert list(repeated['items']) == [{'id': expected[1]}]