diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e5ccecb9..c71d1edd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -48,6 +48,8 @@ jobs: run: npm run build - name: test:unit run: npm run test:unit + - name: test:incoming-helpers + run: npm run test:incoming-helpers - name: test:ci-lifecycle run: npm run test:ci-lifecycle integration: diff --git a/docs/incoming-migration.md b/docs/incoming-migration.md index 4837537d..0e0cb873 100644 --- a/docs/incoming-migration.md +++ b/docs/incoming-migration.md @@ -92,7 +92,10 @@ sequenceDiagram end Browser->>Entryway: 1. Verify email with OTP and prepare arrival - Entryway->>Destination: Reserve repository signing key + Entryway->>Entryway: Persist arrival, DID and selected PDS allocation intent + Entryway->>Destination: POST reserveSigningKey with incoming DID (no auth) + Destination-->>Entryway: Public signingKey; private key stays on PDS + Entryway->>Entryway: Persist validated key with same arrival, DID and PDS Entryway-->>Browser: Arrival code and public preparation facts Browser->>Tool: Explicit user handoff of arrival code @@ -107,6 +110,7 @@ sequenceDiagram Tool->>Source: Obtain fresh createAccount-scoped service JWT Source-->>Tool: Creation authorization Tool->>Entryway: 3. Create account with arrival code and admission fields + Note over Entryway,Destination: Entryway rotation authority signs move containing PDS public key Entryway->>Destination: Create with Entryway-signed destination move Destination->>PLC: Publish destination move Note over Destination,PLC: Identity cutover happens BEFORE data import @@ -190,10 +194,30 @@ handle, and confirms the migration. Entryway checks email uniqueness and atomica records the verified user ID, email/handle reservations, selected PDS, source PLC head and key facts, approved custody configuration and operation ID. -Entryway journals the allocation obligation before reserving a repository signing -key on the selected PDS through its existing contract. The preparation response -contains arrival authorization, the Entryway service DID, explicit source and -destination endpoints, the expected PLC predecessor and public handoff configuration. +The storage work in plan 009 must persist the allocation intent, including the +operation/arrival ID, incoming DID and selected PDS identity and endpoint. Only then +does plan 010 dispatch, through the plan 012 outbound wire adapter, the +unauthenticated public request whose result plan 009 then persists: + +```http +POST /xrpc/com.atproto.server.reserveSigningKey +Content-Type: application/json + +{"did":""} +``` + +The response is `{"signingKey":""}`. Although the upstream DID +field is optional, Entryway must supply the admitted DID and validate the public +key. Plan 009 must persist that result against the same immutable tuple; a changed +DID, PDS or key cannot overwrite it. Plan 010 uses that saved key for destination +creation and plan 011 checks the expected public identity and readiness. These +consumer interfaces are specified in [C2](#c2-persisted-arrival-contract-accepted) +and not implemented. + +The private repository key remains on the selected PDS. Reservation is not source +ownership proof. The preparation response contains arrival authorization, the +Entryway service DID, explicit source and destination endpoints, the expected PLC +predecessor and public handoff configuration. A supplied DID or reservation does not establish account ownership. Source proof is required separately, and admission rate limits bound reservation abuse. @@ -248,9 +272,11 @@ when applicable, and arrival-code header. A submit-scoped JWT cannot authorize creation. Entryway matches the supplied facts to the saved arrival and applies invitation -policy before backend creation. It signs and journals the destination PLC operation -with the reserved repository key, target endpoint and approved destination custody. -The selected PDS receives that operation in its stock creation request, publishes +policy before backend creation. It journals a destination PLC operation containing +the reserved public repository key in `verificationMethods.atproto`, the target +endpoint and approved destination custody. Entryway signs that operation with its +PLC rotation authority, not the PDS repository private key. The selected PDS consumes +its local private reservation during stock creation, publishes the identity move and creates the account active. ### 4. Deactivation, binding and migration access @@ -306,6 +332,8 @@ Counts, a client completion boolean, or zero missing blobs on an empty destinati are insufficient evidence. Valid empty and blob-free source repositories are accepted when integrity checks agree. The exact public source/destination evidence mechanism remains unresolved; synthetic snapshot assertions do not establish it. +[C4](#c4-completeness-contract-blocked) records the qualified behavior and its +limits, and the decision that stays open. All activation paths for pending arrivals share this readiness gate. Entryway journals activation, invokes the lifecycle hook, observes the outcome and atomically @@ -343,12 +371,31 @@ After cutover, recovery no longer requires a source-key JWT because that key is longer the current repository verification key. Replacement credentials recover a lost response without repeating account provisioning. +### Reservation uncertainty + +The inspected stock PDS implementation can return the same saved public key for +sequential calls with the same DID on the same PDS. Its read/generate/write sequence +is not an atomic concurrent-reservation guarantee, nor a promise of indefinite +retention. A lost response or failed local persistence requires bounded +same-DID/same-PDS reconciliation under the current operation owner; an uncertain +remote write cannot be made safe merely by lease expiry. Do not switch placement, +blindly allocate again after creation, or overwrite a different recorded key. +The upstream reservation contract exposes no expiry or cancel API. Successful +creation attempts remote reservation cleanup on a best-effort, nonfatal basis; +a subsequent reservation is not a safe way to recover the used key. Local +cancellation does not remove remote private material. The exact retry and fencing +contract is specified in [C2](#c2-persisted-arrival-contract-accepted) +and [C5](#c5-dispatch-and-recovery-contract-accepted). + ### Cancellation -Abandoned arrivals are not automatically deleted. Before handoff dispatch, -cancellation can release reservations only after checking for remote obligations -and changed PLC state. Once publication is possible, observed or uncertain, the -arrival and reservations remain available for reconciliation. +Abandoned arrivals are not automatically deleted. Cancellation can release local +state only before any allocation intent is persisted, and even then only after +checking for remote obligations and changed PLC state. Preparation persists the +allocation intent before it asks the destination for a signing key, so an arrival +that reached preparation already has a reservation row and cannot be released. Once +publication is possible, observed or uncertain, the arrival and reservations remain +available for reconciliation. Deleting local state cannot remove published Entryway authority. Reversal requires an authorized PLC operation; it is not an expiry side effect. The @@ -368,6 +415,400 @@ and release of conflicting work. Recovery records retain public custody evidence and safe operation identifiers. Reports exclude OTPs, arrival codes, bearer tokens and private keys. +## Contract qualification gates for implementers + +These are delivery contracts for implementers, not implemented APIs. Where a contract +names a callable signature, no such callable exists in the repository yet; the owning +plan must produce it. Where this guide says a behavior is qualified, that means a +bounded committed test observed the stated stock behavior. It does not mean Entryway +exposes, accepts or gates it. + +Read each gate as a producer/consumer handoff. A contract becomes usable by +downstream code only when this section states both the exact call shape and the +failure behavior, and a maintainer accepts it per case. C1, C2, C3 and C5 are +**ACCEPTED**; the maintainer decisions behind them are recorded inline as M1 to M4. +C4 remains **BLOCKED** on the named maintainer decision described below. + +Source and destination versions stay distinct. The upstream facts here are +inspection of pinned PDS 0.5.36, and the standalone qualification source is the +separately pinned 0.5.34 image. No contract may treat the two as interchangeable. + +### Existing policy these contracts reuse + +Contracts below cite existing repository policy rather than inventing limits. A +consumer must use these named producers, not restate the numbers. + +| Policy | Existing producer | Limit | +| --- | --- | --- | +| Destination OTP proof | `src/authentication/better-auth.mjs:22-27` | Eight digits, 600-second code lifetime, five attempts, digest-only storage | +| Extra challenge proof and budget | `src/accounts/challenges.mjs:22-29,36-45` | Default five requests per 600-second bucket per purpose and subject; code digest expires after 600 seconds | +| Login rate limit | `src/features/email-login/xrpc-routes.mjs:28` | 60 requests per IP | +| Signup and create rate limits | `src/features/account-registration/xrpc-routes.mjs:61,68` | 30 requests per IP each | +| Reservation rate limit | `src/plc/xrpc-routes.mjs:7` | 30 requests per IP | +| Password-reset rate limit | `src/features/account-recovery/xrpc-routes.mjs:7` | 30 requests per IP | +| Pending-operation block | `src/accounts/primitives.mjs:78-87` | 409 `MigrationPending` while any migration is not complete | + +The accepted 24-hour arrival code lifetime is a reservation lifetime for one +arrival. It is not an OTP freshness window and does not extend any proof above. + +### C1 wire and admission contract (ACCEPTED) + +Owner: plan 010 prepares and manages arrival lifecycle; plan 012 owns the wire +adapter. Consumers: the browser pages in plans 010 and 012, and the migration tool +through the standard XRPC bodies. + +Proposed Entryway-only POST surfaces. These are Entryway extensions, not AT Protocol +methods. They never accept or export browser cookies on behalf of a tool. + +| Path | Request body | Response | +| --- | --- | --- | +| `/incoming/prepare` | `{did, handle}` | `{operationId, version, arrivalCode, expiresAt, entrywayDid, sourceEndpoint, destinationEndpoint, expectedPlcHead, handoff}` | +| `/incoming/renew` | `{operationId, expectedVersion}` | `{operationId, version, arrivalCode, expiresAt}` | +| `/incoming/cancel` | `{operationId, expectedVersion}` | `{operationId, version, state}` | +| `/incoming/recover` | `{operationId, expectedVersion}` | Standard tracked credential fields, only after C3 prerequisites | + +Authentication per surface: + +- All four require a normalized verified browser principal. `prepare` additionally + requires a fresh destination OTP proof and CSRF protection. +- `prepare` derives the saved user and email from that verified proof. The body never + supplies an email or a user. +- `renew`, `cancel` and `recover` select only the saved arrival bound to that user. A + different user receives the same not-found shape as an unknown operation, so the + response cannot confirm that another user's operation exists. +- `recover` is refused until C3 prerequisites hold. Before cutover it must not accept + source authority as a substitute for the saved binding. + +Secret response handling. `arrivalCode` is returned once to the browser and only its +hash is stored. Secret responses set `Cache-Control: no-store` and are excluded from +logs, URLs, metric labels and error messages. Renewal issues a new code and +generation and invalidates the previous code for authentication; the previous code +must fail with the expired error rather than a distinct replay signal. + +A lost preparation response leaves the caller holding nothing usable, because the +saved hash cannot reproduce the code. Recovery is `renew` from a fresh destination +proof: it issues a new code and generation for the same arrival and invalidates the +previous code, so no second arrival is created. Renewal is refused once the arrival is +provisioned, and the caller resumes through `recover` from C3 instead. + +Errors. Each code is stable and mapped explicitly, and no message carries a secret. + +| Condition | Error code | HTTP status | +| --- | --- | --- | +| Body fails schema | `InvalidRequest` | 400 | +| Missing or stale OTP proof | `ArrivalProofRequired` | 403 | +| Unknown arrival or verified principal does not own it | `AccountNotFound` | 404 | +| Saved arrival is no longer actionable | `ArrivalExpired` | 409 | +| `expectedVersion` differs from the saved version | `OperationConflict` | 409 | +| Concurrent work for the DID is admitted | `OperationPending` | 409 | +| Per-IP or per-subject budget exhausted | `RateLimitExceeded` | 429 | +| Dispatch state makes cancellation unsafe | `OperationRecoveryRequired` | 409 | + +Replay. `prepare` with the same verified user, DID and handle is idempotent and +returns the same saved operation while that operation is still actionable. A changed +DID or handle for the user is a conflict, not a second arrival. Cancellation is safe +only at the internal pre-intent seam, before any persisted allocation or signed +handoff and with unchanged canonical PLC state and no remote obligation. A cancelled +operation is retained as history; `cancel` is idempotent for that state. Preparation +already persists an allocation intent, so a prepared arrival cannot be cancelled to +release reservations or start a replacement arrival. It returns +`OperationRecoveryRequired` with the arrival still saved. No abandonment, +remote-release or reversal path is defined here. + +Standard XRPC bodies keep their shapes and carry the arrival code in the proposed +`Entryway-Arrival-Code` header at `com.atproto.identity.submitPlcOperation` and +`com.atproto.server.createAccount`. The arrival code never substitutes for +`inviteCode`, and invitation policy is unchanged. + +Callables. Plan 010 owns the first three and plan 011 owns the last: + +```text +prepareArrival(principal, proof, {did, handle}) -> prepared arrival +renewArrival(principal, proof, {operationId, expectedVersion}) -> renewed arrival +cancelArrival(principal, proof, {operationId, expectedVersion}) + -> cancelled or reconciliation-required +recoverArrival(principal, proof, {operationId, expectedVersion}) + -> standard tracked credential fields +``` + +Accepted decision M1 (budgets). Arrival wiring reuses the existing limits in +[existing policy](#existing-policy-these-contracts-reuse) and adds no new numeric +budget. Reservation requests keep the existing 30-per-IP limit in +`src/plc/xrpc-routes.mjs:7`. Preparation and renewal use the existing challenge +producer in `src/accounts/challenges.mjs:22-29`, which is five requests per +600-second bucket per purpose and subject. A consumer must call that producer rather +than restate or retune these numbers. + +### C2 persisted arrival contract (ACCEPTED) + +Owner: plan 010 produces the reservation result and plan 009 owns conditional +persistence. Consumers: 010 for destination creation signing and 011 for readiness. +Both consume the immutable tuple already defined in +[arrival preparation](#1-arrival-preparation). + +Required persistence order. Intent is durable before any network call, and the +validated result is durable against that same tuple: + +1. Persist allocation intent: operation/arrival ID, incoming DID, selected PDS ID and + endpoint, saved verified user ID and email. +2. Plan 010 dispatches the unauthenticated `reserveSigningKey` request with the + incoming DID and validates a public `did:key`. +3. Persist the validated public key against the same operation, DID and PDS. + +Callables, all owned by plan 009. Each returns a new version or a stable +outcome, and none overwrites an already-saved tuple member. + +```text +recordAllocationIntent(tx, owner, expectedVersion, {operationId, did, pdsId, endpoint}) +recordReservationResult(tx, owner, expectedVersion, {operationId, did, pdsId, endpoint, signingKey}) +retainUncertainAttempt(tx, owner, expectedVersion, attemptBinding) +``` + +Binding rules these calls must enforce: + +- The tuple `{operationId, did, pdsId, endpoint}` is immutable after intent. A + mismatched DID, PDS or host is a conflict and must not overwrite it. +- An already-saved `signingKey` cannot change. A second reservation returning a + different key is `OperationRecoveryRequired`, not an update. +- Every call runs under the current claim and expected version. A stale owner, + attempt or fence returns a stable conflict and writes nothing. + +Prohibitions the implementation must preserve: + +- No network call belongs inside a database transaction. Intent and result are + separate fenced transactions around the external call. +- Reconciliation may not race a live or possibly delayed reservation. Sequential + same-DID retry observed in the bounded run is not concurrency fencing. +- Do not switch placement, reallocate blindly after creation, or treat a new public + key as proof that the earlier allocation was unapplied. + +Existing seams to extend rather than replace: the checkpoint and external-ready +assertions at `src/accounts/operation-ownership.ts:436-450` (`store.checkpoint`, +`store.assertExternalReady`, `store.assertActive`) and the validated +`signingKeyResult` projection at `src/accounts/operation-ownership.ts:471-491`. The +working producer pattern is `src/pds/migration-client.ts:203-227`, which journals +step `external-target-key` with intent `{did}` and projects only the validated public +key. Its `observe` reports `replay-safe` for an isolated and drained allocation and +never claims that the earlier allocation was unapplied. + +Reconciliation after response loss. The same-DID, same-PDS case is the only supported +path: the current operation owner re-reads the saved attempt, and an isolated and +drained allocation may allocate again. Any uncertain or divergent result retains +`OperationRecoveryRequired` until the operator recovery in +[shared operations](shared-operations.md#supported-operator-recovery) applies. + +Accepted decision M2 (retry). Existing operator isolation and drain evidence is +sufficient for an arrival reservation. Automatic retry is permitted only for the +worker that still owns the live operation and whose fence is current, and only after +it has re-read the saved attempt and confirmed that no other dispatch is in flight. +Every other case retains `OperationRecoveryRequired`. A worker may not retry after +lease expiry, after losing ownership, or on behalf of another worker, and no caller +may promote the sequential same-DID observation into a concurrency guarantee. + +### C3 lifecycle and credential contract (ACCEPTED) + +Owner: plan 009 owns the atomic association, binding, claims and checkpoint writes. +Plans 011 and 012 consume the projection, guard and credential interfaces below. + +Order: create active, observe deactivation, atomically bind as pending, then issue +tracked credentials. Pending is an operation condition, not an account class. It is +derived from the current operation and never stored as a permanent imported-account +category. Account status values remain `provisioning`, `active`, `deactivated` and +`deleted` as enforced at `src/accounts/rules.ts:19-39`; no new status value is added. + +Pending projection. Existing evidence is `hasPendingExternalMigration` at +`src/database/drizzle/account-storage.ts:631-641`, true while any migration +reservation is not complete. The proposed richer projection is owned by plan 009: + +```text +readArrivalAccess(did) -> {operationId, version, did, userId, pdsId, provisioned, + deactivationObserved, bound, complete, unresolvedAttempt} +``` + +Every field derives from saved records. `complete` is true only when the activation +checkpoint is complete and is never inferred from account status alone. + +Shared guard. Proposed predicate, owned jointly by 009 and the composition root: + +```text +requireArrivalAction(projection, action) -> allowed, or a stable refusal +``` + +It must reject ordinary OAuth authorization, account management and unguarded +activation while an arrival is pending, and it must not block read-only diagnosis. +The existing partial precedent is `assertNoMigration` at +`src/accounts/primitives.mjs:78-87`, consumed by handle change +(`src/features/handle-change/change-handle.ts:73,127`), account deletion +(`src/features/account-deletion/delete-account.mjs:16`) and account status change +(`src/features/account-settings/change-status.mjs:16`). It is not applied to OAuth +authorization today. + +Activation must invoke the same readiness operation on every path: the browser +action, the Entryway XRPC route and the stock backend callback. The current Entryway +route at `src/features/account-settings/xrpc-routes.mjs:29-48` calls +`accounts.setStatus` and does not run arrival readiness. + +Stock backend behavior the guard must account for, from the pinned 0.5.36 inspection. +Package-relative paths below are under +`node_modules/@atproto/pds/dist/api/com/atproto/`: + +- Stock activation delegates to Entryway when configured. The installed + `@atproto/pds` `server/activateAccount.js:14-24` forwards the Entryway passthrough + headers and performs no local activation, so delegation inherits Entryway's + readiness gate rather than bypassing it. +- Stock import and blob upload check takedown, not deactivation. The installed + `repo/importRepo.js:13-19` and `repo/uploadBlob.js:7` use `checkTakedown: true` + with the `ACCESS_FULL` access scope. A deactivated destination therefore stays + writable to a valid backend bearer, which is intentional for transfer. +- A standard backend bearer is not a migration-only credential. Access tokens use + the backend audience with stock PDS scope (`com.atproto.access` and app-password + scopes) as issued by `src/oauth/legacy-credentials.mjs:9-14,233-250`. There is no + universal migration-only permission, and this contract must not promise one. + +Credential interfaces. Proposed extensions of the tracked legacy credentials in +`src/oauth/legacy-credentials.mjs:307-326,406-439`, owned by plan 011: + +```text +issueArrivalSession(owner, expectedVersion, arrivalId) +refreshArrivalSession(refreshCredential) +recoverArrivalSession(principal, freshProof, arrivalId, expectedVersion) +``` + +Rules: no issuance before observed provisioning, observed deactivation and saved-user +binding. Token return follows the committed association and never precedes it. +Existing refresh replay and revocation semantics remain: a replayed refresh +invalidates the family (`src/oauth/legacy-credentials.mjs:400-405`), and a family +bound to another PDS DID is rejected (`src/oauth/legacy-credentials.mjs:466-467`), +without revoking that family. +After cutover, recovery does not require a source-key JWT, and no contract may +require one. + +Saved-user recovery after cutover is distinct from source JWT authorization before +it. Recovery uses the saved verified user binding and never an email match alone. + +Accepted decision M3 (recovery families). `recoverArrivalSession` revokes every prior +refresh family for that arrival in the same committed association that issues the new +one. Earlier families cannot be refreshed afterwards. Revoking local family state +does not revoke stateless stock-PDS access tokens already issued, so already-issued +access tokens remain usable until they expire. Entryway must state that gap wherever +it describes recovery rather than implying full revocation. + +The destination permission and callback runtime proof also remains unrun; authoring +the guard contract does not qualify it. + +### C4 completeness contract (BLOCKED) + +Plan 011 readiness and plan 012 exposure stay blocked. Named blocker: no supported +mechanism yet provides authenticated historical source key and head binding, a +complete source and destination inventory, all missing-blob pages, and a barrier that +prevents writes between verification and activation. The named maintainer decision is +which finality and mutation-serialization guarantee to adopt. + +Historical qualification and its limits. An earlier bounded run, recorded as +`d29515b` in the original repair history rather than an ancestor of this stack, +reported the following stock source and inactive destination behaviors within its +explicit byte and phase budgets. These are historical observations, not a current +verification result or full migration qualification; reproof requires a fresh +managed sandbox run: + +- Source CAR exports verified against the observed repository key, including a + nonempty snapshot. Two export requests were started before a write was dispatched, + which shows request ordering only and does not establish server-side overlap. The + case asserts a CAR size limit and per-phase budgets. The withheld final client + byte belongs to the separate destination import case below, not source export. +- Source snapshot capture, signature verification and record inventory comparison are + qualified only within those budgets for that snapshot. +- Inactive destination: an Entryway-hosted inactive destination accepted a streamed + import and destination readback, within the observed limits. Withholding the final + client byte does not establish server receipt of the prefix or universal finality. +- Reservation: sequential same-DID retry and placement isolation only. + +What remains unqualified: blob retention across pages, pagination of a mutable +missing-blob set, a consistent export under concurrent writes, and any write barrier +between verification and activation. The stock export reads one root and then pages +blocks (`node_modules/@atproto/pds/dist/actor-store/repo/sql-repo-reader.js:68-90`), so +a read is not a consistent snapshot under writes, and the missing-blob set can change +between pages. A callback guard alone does not fence an in-flight import or write. + +Forbidden resolutions: a private freeze, trusting the client's completion summary, an +arbitrary quiet period, a PDS patch, or a data-plane proxy. Absence of a convenience +endpoint is not proof of impossibility. + +Inputs C4 must still define once the decision is made: how the final source snapshot +is agreed, how later source changes invalidate prior readiness, and whether source +availability or source-supplied authorization is required after cutover. Empty and +blob-free repositories remain valid inputs. + +### C5 dispatch and recovery contract (ACCEPTED) + +Owner: plans 009 to 012 share one owner/fence and observation contract. The only +callable seam today is the generic dispatch at +`src/accounts/operation-ownership.ts:265-328`, consumed as `ownership.dispatch` by the +migration client above and by account primitives at +`src/accounts/primitives.mjs:20-29`. + +Owner value. Every physical write transaction checks this owner with the lease and +expected version: + +```text +owner = {operationId, workerId, executionAttemptId, fence} +``` + +Callables: + +```text +beginExternalAttempt(tx, owner, expectedVersion, {target, method, publicIntent}) -> attemptId +recordExternalObservation(tx, owner, expectedVersion, {attemptId, outcome, evidence}) +reconcileAcceptedOperation(owner, {operationId, requestDigest}) -> exact saved facts +``` + +These callables extend the existing `ownership.dispatch` seam rather than creating a +second dispatch path, and plans 009 to 012 reconcile its `beginExternal` and +`acknowledgeExternal` store calls instead of duplicating them. + +Rules: + +- `beginExternalAttempt` returns an immutable attempt ID. The persisted descriptor is + step, target, method and public intent digest only. It never stores credentials, + proof codes or raw response bodies, matching `ExternalAttemptInput` at + `src/database/operation-ownership.port.ts:27-32`. +- `recordExternalObservation` accepts typed public evidence only. Outcomes distinguish + acknowledged, definitively rejected, uncertain and divergent. A status match does + not drain an earlier request. +- `reconcileAcceptedOperation` resumes the exact saved facts. It never signs or sends + a new operation, and a changed request digest is a conflict. +- Caller authorization is separate from continuation. Exact durable accepted replay + after JWT expiry or key cutover is continuation, not fresh authorization. A lost + credential uses saved-user recovery from C3. + +Uncertainty and recovery. Any interruption after the dispatch marker requires +dispatch-source isolation and drain, which is why lease expiry alone may not permit a +new remote write. The existing operator path is +[shared operations](shared-operations.md#supported-operator-recovery): trusted +isolation and drain attestations, an `observe` to `retry-if-safe` upgrade, and a fresh +fenced execution that observes remote state before acknowledging. It is a candidate +reuse for public arrival, not automatic coverage. + +Cancellation safe point. Cancellation releases local reservations only before any +allocation intent is persisted, and only when no remote obligation exists and the +canonical PLC state is unchanged. Because preparation persists the allocation intent +before dispatching the reservation, a prepared arrival always carries a reservation +row and stays pending for reconciliation rather than being released. Once publication +is possible, observed or uncertain, custody and admission are retained. + +Partial outcomes. A partial PLC publication with an absent backend stays pending +unless an explicitly reviewed repair applies; the managed migration exact-head repair +must not be extrapolated to public arrival. Activation response loss retains its +attempt until exact completion or drain evidence permits reconciliation. + +Accepted decision M4 (repair permission). The affected user may resume the exact +accepted operation at any time; that continuation never creates new authority and +always replays the saved signed facts. An operator may additionally reconcile a +partial outcome, but only with canonical observation evidence recorded for that +attempt. Lease expiry alone authorizes no repair and permits no new remote write, and +no automatic repair exists. A repair whose evidence is missing, uncertain or +divergent retains `OperationRecoveryRequired`. + ## Related documentation - [Data custody](data-custody.md): identity, key roles and atomic authority boundaries. diff --git a/docs/reuse-assessment.md b/docs/reuse-assessment.md index d9fc1e61..83ff2f03 100644 --- a/docs/reuse-assessment.md +++ b/docs/reuse-assessment.md @@ -74,6 +74,30 @@ moderation, credential lifetime and production qualification. 5. **Email coverage:** sending login OTP through SMTP does not prove every XRPC email challenge is delivered through the same transport. +## Incoming public mechanism provenance + +For maintainers evaluating incoming migration, the reference PDS 0.5.36 +implementation exposes unauthenticated DID-keyed repository-key reservation and +forwards activation to Entryway when configured. Its repository export captures a +root before paging blocks; import applies a verified diff, and missing-blob +pagination operates on a mutable set. These implementation observations do not +establish source finality or a destination write barrier. The managed standalone +source image is separately pinned to PDS 0.5.34; inspect its version-specific +behavior rather than assuming full equivalence. The synthetic migration fixture +and repository inventory comparison are reuse candidates, not public end-to-end +qualification or proof of a historical source signature. + +One bounded committed qualification run observed a real stock source CAR +verification, a nonempty snapshot, two export requests started before a write was +dispatched (request ordering only, not server-side overlap), an +Entryway-hosted inactive destination accepting a streamed import, and sequential +same-DID reservation with placement isolation. That run is bounded by its own +committed byte and phase budgets and does not establish finality, blob pagination +or an activation barrier. C4 therefore stays blocked, while C1, C2, C3 and C5 are +accepted contracts whose callables the downstream plans must still produce; see the +contract gates in +[incoming migration](incoming-migration.md#contract-qualification-gates-for-implementers). + ## Custody design reference The approved model follows [Rust Entryway in atproto-crates](https://tangled.org/ngerakines.me/atproto-crates) diff --git a/docs/testing.md b/docs/testing.md index cfd32fdd..70b96318 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -48,6 +48,7 @@ and requires no host dependency install: | `./tests/local.sh resilience-profiles` | Run all three fresh deployment profiles and require complete bounded receipts; see [profile contract](deployment-profiles.md) | | `./tests/local.sh contracts` | Run imported contract scenarios in the test container after up | | `./tests/local.sh browser` | Run browser scenarios after up | +| `./tests/local.sh incoming-public-mechanisms` | Bounded stock public-mechanism contract test after up: standalone signup, Mailpit, PLC signing, inactive destination and reservation; 12 offline plus 5 live cases, not a migration qualification | | `./tests/local.sh migration` | Fresh synthetic external-source migration, including two Entryway restarts | | `./tests/local.sh migration-resume` | Resume a reviewed interrupted fixture; inspect state before using | | `./tests/local.sh reverify` | Check an already completed migration fixture | @@ -77,6 +78,60 @@ Review `resilience.json` and per-phase reports if interrupted. See | `SANDBOX_PROJECT` | Explicit owned project name | | `ACCEPTANCE_REPORT_DIR` | Report directory override | +For operators running browser scenarios or the public-mechanism qualification phase, +the controller refreshes and validates sandbox access and writes a private allowlisted +`incoming-access.json` in the owned sandbox. It mounts that file read-only at +`/run/incoming-access.json` in the browser container and sets `INCOMING_ACCESS_PATH`. +The qualification phase additionally sets `INCOMING_PROJECT`; ordinary browser runs +do not. `all` and `fresh` inherit the projection through the browser phase; they do +not run the qualification phase. + +`incoming-public-mechanisms` runs the contract test directly in the browser container +with `node --test`, not a Playwright spec: + +```sh +./tests/local.sh incoming-public-mechanisms +``` + +After `up`, that one command runs every case in +`tests/contracts/incoming-public-mechanisms.test.mjs`. The live cases create a unique +account on the standalone PDS, request a PLC confirmation through its public API, +read its recipient-bound message from Mailpit, use the token for public signing, and +observe the inactive destination and reservation behavior. The controller applies the +repository-owned SMTP overlay to pds3 after startup. + +Case counts and what they mean. One live-gated test runs two further subcases, so a +full opt-in run reports more cases than the file declares. The file declares fifteen +top-level tests: twelve offline and three live-gated. A full opt-in run reports 17 +cases, which is 12 offline plus 5 live. Only those 5 live cases form the public +qualification. + +`tests/local.sh incoming-public-mechanisms` always exports +`INCOMING_PUBLIC_MECHANISMS=1` for the container, so unsetting the variable does not +turn that command into an offline run. To run only the offline cases natively, use: + +```sh +env -u INCOMING_PUBLIC_MECHANISMS -u INCOMING_ACCESS_PATH node --test \ + tests/contracts/incoming-public-mechanisms.test.mjs +``` + +That run reports the three live-gated tests skipped; those skips are not +qualification. At the time of writing, that native run reports 12 passed and 3 +skipped, which is the offline subset only. A live count is a claim about a sandbox run +and must not be reported from this document. + +The offline cases cover the managed request boundary and phase budgets. They do not +prove any public behavior, and no Entryway endpoint or route implements this +migration: the contract test consumes stock PDS APIs, not Entryway handlers. + +The preflight verifies the signed PLC operation against the source's observed +rotation keys. The remaining contracts are still open in the +[contract gates](incoming-migration.md#contract-qualification-gates-for-implementers): +C4 finality and activation-race closure stay BLOCKED; C1, C2, C3 and C5 are accepted +contracts whose callables the downstream plans must still produce. + +Use the owned-project cleanup procedure below, including after failures. + Runtime configuration uses `SERVICE_CONFIG_PATH` (default `./.runtime/config.json`). `DATABASE_BACKEND` selects SQLite or PostgreSQL and `DEPLOYMENT_MODE` selects the single-node or multi-node profile. SQLite uses `account-authority.sqlite` under @@ -254,13 +309,31 @@ profile controllers are not filtered a second time. Raw reports are never upload private reports locally for detailed failures. Lifecycle fixtures use fake Docker natively; they do not establish hosted-runner compatibility. +### Development: native incoming helper checks + +Contributors changing incoming test infrastructure run the stock mail poller, +managed I/O helper and access-projection tests explicitly: + +```sh +npm run test:incoming-helpers +``` + +This script runs `tests/support/helpers/stock-plc-mailpit.test.mjs`, +`tests/support/helpers/incoming-public-io.test.mjs` and +`tests/atmosphere/project-incoming-access.test.mjs`. Native CI runs the script in +its own step. Contract discovery does not include these helper files, and the +live stock-mechanism suite is not part of the native unit allowlist. + ## Feature tests and operational evidence Contract discovery executes both `tests/contracts/*.test.mjs` and recursively co-located `src/features/**/*.test.mjs`, once and sorted, with a shared JUnit report. -Run browser before contracts in a fresh project: real contracts require the -verified account created by the browser journey. Pure diagnostic unit runs use -`npm run test:unit` natively and do not need that fixture. +It does not discover helper tests under `tests/support`, so run those explicitly with +`node --test` when a helper changes. See the rule in +`tests/support/discover-contracts.mjs`. Run browser before contracts in a fresh +project: real contracts require the verified account created by the browser journey. +Pure diagnostic unit runs use `npm run test:unit` natively and do not need that +fixture. Browser email-proof tests must obtain the recipient's visible code from Mailpit message content; reading the database outbox's full token hides missing challenge context. Verify the hidden selector is retained and that wrong codes or selectors diff --git a/package.json b/package.json index 1a286dce..c9865703 100644 --- a/package.json +++ b/package.json @@ -30,6 +30,7 @@ "changeset": "changeset", "changeset:check": "changeset status", "version-packages": "changeset version && npm install --package-lock-only --ignore-scripts --offline --no-audit --no-fund", + "test:incoming-helpers": "node --test tests/support/helpers/stock-plc-mailpit.test.mjs tests/support/helpers/incoming-public-io.test.mjs tests/atmosphere/project-incoming-access.test.mjs", "test:ci-lifecycle": "python3 tests/support/ci-lifecycle.test.py" }, "dependencies": { diff --git a/tests/atmosphere/project-incoming-access.mjs b/tests/atmosphere/project-incoming-access.mjs new file mode 100644 index 00000000..30db947f --- /dev/null +++ b/tests/atmosphere/project-incoming-access.mjs @@ -0,0 +1,112 @@ +import { readFile } from "node:fs/promises"; +import { pathToFileURL } from "node:url"; + +function origin(value) { + if (typeof value !== "string") throw new Error("Invalid HTTPS origin"); + let parsed; + try { + parsed = new URL(value); + } catch { + throw new Error("Invalid HTTPS origin"); + } + if ( + parsed.protocol !== "https:" || + parsed.username || + parsed.password || + parsed.pathname !== "/" || + parsed.search || + parsed.hash || + (value !== parsed.origin && value !== `${parsed.origin}/`) + ) { + throw new Error("Invalid HTTPS origin"); + } + return parsed.origin; +} + +export function projectIncomingAccess(access, project, routes) { + if ( + typeof project !== "string" || + !/^hypercerts-entryway(-[a-z0-9_-]+)?$/.test(project) || + access?.schemaVersion !== 1 || + access.project !== project || + access.network?.internal !== true || + access.network?.physicalName !== `${project}_atmosinabox` || + access.observations?.configuration !== "current" || + access.trust?.certificateExists !== true || + !Array.isArray(access.endpoints) + ) { + throw new Error("Invalid or stale private sandbox access"); + } + if ( + !Array.isArray(routes?.destinationUrls) || + routes.destinationUrls.length !== 2 + ) { + throw new Error("Expected two configured destination origins"); + } + const expected = [ + ["pds", "pds3", origin(routes.sourceUrl)], + ["application", "entryway", origin(routes.entrywayUrl)], + ["application", "pds1", origin(routes.destinationUrls[0])], + ["application", "pds2", origin(routes.destinationUrls[1])], + ["plc", "plc", origin(routes.plcUrl)], + ["application", "mail-capture", origin(routes.mailpitUrl)], + ]; + if (new Set(expected.map((item) => item[2])).size !== expected.length) { + throw new Error("Configured origins must be distinct"); + } + for (const [serviceType, service, expectedOrigin] of expected) { + const hostname = new URL(expectedOrigin).hostname; + const matching = access.endpoints.filter( + (endpoint) => endpoint?.hostname === hostname, + ); + if ( + matching.length !== 1 || + matching[0].serviceType !== serviceType || + matching[0].service !== service || + matching[0].kind !== "exact" || + origin(matching[0].url) !== expectedOrigin + ) { + throw new Error("Missing, ambiguous or inconsistent sandbox endpoint"); + } + } + return { + schemaVersion: 1, + project, + sourceUrl: expected[0][2], + entrywayUrl: expected[1][2], + destinationUrls: [expected[2][2], expected[3][2]], + plcUrl: expected[4][2], + mailpitUrl: expected[5][2], + }; +} + +if ( + process.argv[1] && + import.meta.url === pathToFileURL(process.argv[1]).href +) { + const [ + accessPath, + project, + sourceUrl, + entrywayUrl, + firstDestination, + secondDestination, + plcUrl, + mailpitUrl, + ...extra + ] = process.argv.slice(2); + if (!mailpitUrl || extra.length) { + throw new Error( + "usage: project-incoming-access.mjs ", + ); + } + const access = JSON.parse(await readFile(accessPath, "utf8")); + const projection = projectIncomingAccess(access, project, { + sourceUrl, + entrywayUrl, + destinationUrls: [firstDestination, secondDestination], + plcUrl, + mailpitUrl, + }); + process.stdout.write(`${JSON.stringify(projection)}\n`); +} diff --git a/tests/atmosphere/project-incoming-access.test.mjs b/tests/atmosphere/project-incoming-access.test.mjs new file mode 100644 index 00000000..225e2c44 --- /dev/null +++ b/tests/atmosphere/project-incoming-access.test.mjs @@ -0,0 +1,171 @@ +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { test } from "node:test"; +import { projectIncomingAccess } from "./project-incoming-access.mjs"; + +const project = "hypercerts-entryway-test"; +const routes = { + sourceUrl: "https://pds3.atmosbox.test", + entrywayUrl: "https://entryway.atmosbox.test", + destinationUrls: [ + "https://cluster1.atmosbox.test", + "https://cluster2.atmosbox.test", + ], + plcUrl: "https://plc.atmosbox.test", + mailpitUrl: "https://mailpit.atmosbox.test", +}; +const fixture = () => ({ + schemaVersion: 1, + project, + network: { internal: true, physicalName: `${project}_atmosinabox` }, + observations: { configuration: "current" }, + trust: { certificateExists: true, certificatePath: "/private/secret" }, + secret: "never-project-this", + endpoints: [ + ["pds", "pds3", routes.sourceUrl], + ["application", "entryway", routes.entrywayUrl], + ["application", "pds1", routes.destinationUrls[0]], + ["application", "pds2", routes.destinationUrls[1]], + ["plc", "plc", routes.plcUrl], + ["application", "mail-capture", routes.mailpitUrl], + ].map(([serviceType, service, url]) => ({ + serviceType, + service, + url, + hostname: new URL(url).hostname, + kind: "exact", + token: "never-project-this", + })), +}); + +test("projects only configured public origins", () => { + const result = projectIncomingAccess(fixture(), project, routes); + assert.deepEqual(result, { schemaVersion: 1, project, ...routes }); + assert.doesNotMatch(JSON.stringify(result), /secret|private|token/); +}); + +test("rejects invalid private sandbox metadata", () => { + for (const mutate of [ + (access) => { + access.project = "other"; + }, + (access) => { + access.schemaVersion = 2; + }, + (access) => { + access.network.internal = false; + }, + (access) => { + access.network.physicalName = "other_atmosinabox"; + }, + (access) => { + access.observations.configuration = "stale"; + }, + (access) => { + access.trust.certificateExists = false; + }, + (access) => { + delete access.endpoints; + }, + ]) { + const access = fixture(); + mutate(access); + assert.throws(() => projectIncomingAccess(access, project, routes)); + } +}); + +test("rejects absent, ambiguous and inconsistent endpoints", () => { + for (const mutate of [ + (access) => { + access.endpoints.shift(); + }, + (access) => { + access.endpoints.push({ ...access.endpoints[0] }); + }, + (access) => { + access.endpoints[0].service = "pds1"; + }, + (access) => { + access.endpoints[0].serviceType = "application"; + }, + (access) => { + access.endpoints[0].kind = "wildcard"; + }, + (access) => { + access.endpoints[0].url = routes.plcUrl; + }, + (access) => { + access.endpoints[0].url += "/private"; + }, + (access) => { + access.endpoints[0].url = "https://user:password@pds3.atmosbox.test"; + }, + ]) { + const access = fixture(); + mutate(access); + assert.throws(() => projectIncomingAccess(access, project, routes)); + } +}); + +test("rejects unsafe or inconsistent expected routes", () => { + for (const overrides of [ + { sourceUrl: "http://pds3.atmosbox.test" }, + { sourceUrl: "https://user@pds3.atmosbox.test" }, + { sourceUrl: `${routes.sourceUrl}/path` }, + { sourceUrl: `${routes.sourceUrl}?token=secret` }, + { sourceUrl: routes.plcUrl }, + { destinationUrls: [routes.destinationUrls[0]] }, + { entrywayUrl: "https://wrong.atmosbox.test" }, + ]) { + assert.throws(() => + projectIncomingAccess(fixture(), project, { ...routes, ...overrides }), + ); + } +}); + +test("CLI projects a real JSON file without exposing extra fields", () => { + const directory = mkdtempSync(join(tmpdir(), "incoming-access-")); + try { + const accessPath = join(directory, "access.json"); + writeFileSync(accessPath, JSON.stringify(fixture()), { mode: 0o600 }); + const result = spawnSync( + process.execPath, + [ + fileURLToPath( + new URL("./project-incoming-access.mjs", import.meta.url), + ), + accessPath, + project, + routes.sourceUrl, + routes.entrywayUrl, + ...routes.destinationUrls, + routes.plcUrl, + routes.mailpitUrl, + ], + { encoding: "utf8" }, + ); + assert.equal(result.status, 0, result.stderr); + assert.deepEqual(JSON.parse(result.stdout), { + schemaVersion: 1, + project, + ...routes, + }); + assert.doesNotMatch(result.stdout, /secret|private|token/); + } finally { + rmSync(directory, { recursive: true, force: true }); + } +}); + +test("CLI requires explicit arguments and fails without them", () => { + const result = spawnSync( + process.execPath, + [fileURLToPath(new URL("./project-incoming-access.mjs", import.meta.url))], + { encoding: "utf8" }, + ); + assert.notEqual(result.status, 0); + assert.match(result.stderr, /usage:/); +}); diff --git a/tests/atmosphere/stacks/incoming-source-mail.yaml b/tests/atmosphere/stacks/incoming-source-mail.yaml new file mode 100644 index 00000000..2e07486e --- /dev/null +++ b/tests/atmosphere/stacks/incoming-source-mail.yaml @@ -0,0 +1,5 @@ +services: + pds3: + environment: + PDS_EMAIL_SMTP_URL: smtp://mail-capture:2525 + PDS_EMAIL_FROM_ADDRESS: pds@atmosbox.test diff --git a/tests/contracts/incoming-public-mechanisms.test.mjs b/tests/contracts/incoming-public-mechanisms.test.mjs new file mode 100644 index 00000000..87ebedb2 --- /dev/null +++ b/tests/contracts/incoming-public-mechanisms.test.mjs @@ -0,0 +1,1200 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { readFile } from "node:fs/promises"; +import { randomUUID } from "node:crypto"; +import * as plc from "@did-plc/lib"; +import { + MemoryBlockstore, + def, + ensureV3Commit, + readCarWithRoot, + verifyCommitSig, + verifyRepoCar, +} from "@atproto/repo"; +import { waitForStockPlcMessage } from "../support/helpers/stock-plc-mailpit.mjs"; +import { + ResponseStatusError, + createIncomingPublicIo, + runHeldStreamedCall, +} from "../support/helpers/incoming-public-io.mjs"; + +const requestTimeoutMs = 15_000; +const settleTimeoutMs = 5_000; +const carLimitBytes = 4 * 1024 * 1024; +const preflightBudgetMs = 30_000; +const exportBudgetMs = 30_000; +const destinationBudgetMs = 60_000; +const reservationBudgetMs = 20_000; + +const within = async (signal, limitMs) => + Promise.race([ + new Promise((resolve) => + signal.addEventListener( + "abort", + () => resolve({ error: signal.reason }), + { once: true }, + ), + ), + new Promise((resolve) => + setTimeout(() => resolve({ pending: true }), limitMs), + ), + ]); + +const assertBudgetDeadline = (deadlineMs) => { + if (!Number.isSafeInteger(deadlineMs) || deadlineMs < 1) + throw new RangeError("Invalid phase budget"); +}; + +const createPhaseBudget = (t, { deadlineMs }) => { + if (!Number.isSafeInteger(deadlineMs) || deadlineMs < 1) + throw new RangeError("Invalid phase budget"); + if (!t.signal) throw new TypeError("A node test signal is required"); + const expired = new Error("Phase budget exhausted"); + const controller = new AbortController(); + const timers = new Set(); + const children = new Set(); + const expire = () => { + if (!controller.signal.aborted) controller.abort(expired); + }; + const timer = setTimeout(expire, deadlineMs); + timers.add(timer); + const signal = AbortSignal.any([t.signal, controller.signal]); + const onParentAbort = () => { + for (const child of children) child.controller.abort(t.signal.reason); + }; + t.signal.addEventListener("abort", onParentAbort, { once: true }); + const child = ({ deadlineMs: childBudgetMs } = {}) => { + if (controller.signal.aborted) throw new Error("Phase budget exhausted"); + if (childBudgetMs !== undefined) assertBudgetDeadline(childBudgetMs); + const childController = new AbortController(); + const childTimer = + childBudgetMs === undefined + ? undefined + : setTimeout(() => childController.abort(expired), childBudgetMs); + if (childTimer !== undefined) timers.add(childTimer); + const entry = { controller: childController, timer: childTimer }; + children.add(entry); + const childSignal = AbortSignal.any([ + t.signal, + controller.signal, + childController.signal, + ]); + const finish = () => { + children.delete(entry); + if (entry.timer !== undefined) { + clearTimeout(entry.timer); + timers.delete(entry.timer); + } + }; + childSignal.addEventListener("abort", finish, { once: true }); + return { controller: childController, signal: childSignal }; + }; + const settle = async () => { + clearTimeout(timer); + timers.delete(timer); + t.signal.removeEventListener("abort", onParentAbort); + for (const entry of children) { + entry.controller.abort(new Error("Phase budget retired")); + if (entry.timer !== undefined) { + clearTimeout(entry.timer); + timers.delete(entry.timer); + } + } + children.clear(); + for (const handle of timers) clearTimeout(handle); + timers.clear(); + }; + return { signal, child, settle, expired }; +}; + +const isDidKey = (value) => + typeof value === "string" && value.startsWith("did:key:"); + +const requireObservedOperation = (response) => { + if (!Array.isArray(response?.rotationKeys)) + throw new Error("Invalid PLC rotation keys"); + for (const key of response.rotationKeys) + if (!isDidKey(key)) throw new Error("Invalid PLC rotation keys"); + const { verificationMethods } = response; + if ( + typeof verificationMethods !== "object" || + verificationMethods === null || + Array.isArray(verificationMethods) + ) + throw new Error("Invalid PLC verification methods"); + for (const key of Object.values(verificationMethods)) + if (!isDidKey(key)) throw new Error("Invalid PLC verification methods"); + if (!isDidKey(verificationMethods.atproto)) + throw new Error("Invalid PLC verification methods"); + return { + rotationKeys: [...response.rotationKeys], + verificationMethods: { ...verificationMethods }, + }; +}; + +const createManagedObservation = ({ fetch, signal }) => { + if (typeof fetch !== "function") throw new TypeError("Invalid fetch"); + if (!signal) throw new TypeError("A managed boundary requires a signal"); + const io = createIncomingPublicIo({ fetch, signal, timeoutMs: 5_000 }); + return { + async readLastOperation(endpoint, did, childSignal) { + const origin = new URL(endpoint); + const response = await io.call( + new URL(`/${encodeURIComponent(did)}/log/last`, origin).href, + { verb: "GET", signal: childSignal }, + ); + return requireObservedOperation(response); + }, + readJson: (url, childSignal) => + io.call(url, { verb: "GET", signal: childSignal }), + pollStockConfirmation: (options, child) => + waitForStockPlcMessage({ ...options, fetch, signal: child.signal }), + close: (options) => io.close(options), + }; +}; + +test( + "managed boundary cancellation reaches the injected transport", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const budget = createPhaseBudget(t, { deadlineMs: 5_000 }); + const seen = []; + const observation = createManagedObservation({ + signal: budget.signal, + fetch: (_url, init) => + new Promise((_resolve, reject) => { + seen.push(init.signal); + init.signal.addEventListener("abort", () => + reject(init.signal.reason), + ); + }), + }); + t.after(() => observation.close({ settleTimeoutMs })); + const phase = budget.child(); + const pending = observation.readLastOperation( + "https://plc.example", + "did:plc:example", + phase.signal, + ); + const early = await Promise.race([ + pending.then( + () => "settled", + () => "settled", + ), + new Promise((resolve) => setTimeout(() => resolve("pending"), 50)), + ]); + assert.equal(early, "pending", "The request must still be in flight"); + phase.controller.abort(new Error("phase cancelled by test")); + const outcome = await pending.then( + (value) => ({ value }), + (error) => ({ error }), + ); + assert.equal(outcome.error?.message, "phase cancelled by test"); + assert.equal(seen.length, 1, "The transport saw exactly one dispatch"); + assert.equal( + seen[0].aborted, + true, + "Cancellation must reach the transport signal, not only the caller", + ); + }, +); + +test( + "managed boundary reports a PLC non-2xx as a typed status error", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const budget = createPhaseBudget(t, { deadlineMs: 5_000 }); + const observation = createManagedObservation({ + signal: budget.signal, + fetch: async () => + new Response(new Uint8Array([1]), { + status: 404, + headers: { "content-type": "application/json" }, + }), + }); + t.after(() => observation.close({ settleTimeoutMs })); + const outcome = await observation + .readLastOperation("https://plc.example", "did:plc:example") + .then( + (value) => ({ value }), + (error) => ({ error }), + ); + assert.ok(outcome.error instanceof ResponseStatusError); + assert.equal(outcome.error.status, 404); + }, +); + +test( + "managed boundary propagates the phase deadline as its own failure", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const budget = createPhaseBudget(t, { deadlineMs: 25 }); + let aborted = false; + const observation = createManagedObservation({ + signal: budget.signal, + fetch: (_url, init) => + new Promise((_resolve, reject) => { + init.signal.addEventListener("abort", () => { + aborted = true; + reject(init.signal.reason); + }); + }), + }); + t.after(() => observation.close({ settleTimeoutMs })); + const outcome = await observation + .readLastOperation("https://plc.example", "did:plc:example") + .then( + (value) => ({ value }), + (error) => ({ error }), + ); + assert.equal(outcome.error?.message, "Phase budget exhausted"); + assert.equal(aborted, true, "The deadline must reach the transport"); + }, +); +test( + "managed observation returns the observed PLC verification methods it validated", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const budget = createPhaseBudget(t, { deadlineMs: 5_000 }); + const rotationKeys = [ + "did:key:zQ3shP2mWsZYWK55LZTotwS4VBDQk7ztNzESpMDDfBQ8VVNsY", + ]; + const verificationMethods = { + atproto: "did:key:zQ3shP2mWsZYWK55LZTotwS4VBDQk7ztNzESpMDDfBQ8VVNsY", + }; + const observation = createManagedObservation({ + signal: budget.signal, + fetch: async () => + new Response( + JSON.stringify({ + did: "did:plc:example", + rotationKeys, + verificationMethods, + alsoKnownAs: [], + services: {}, + }), + { + status: 200, + headers: { "content-type": "application/json" }, + }, + ), + }); + t.after(async () => { + await observation.close({ settleTimeoutMs }); + await budget.settle(); + }); + const initial = await observation.readLastOperation( + "https://plc.example", + "did:plc:example", + budget.child().signal, + ); + assert.deepEqual( + initial.rotationKeys, + rotationKeys, + "The rotation keys the caller verifies against survive the boundary", + ); + assert.equal( + initial.verificationMethods?.atproto, + verificationMethods.atproto, + "The repository signature key used by the caller must survive the boundary, not be projected away", + ); + }, +); + +test( + "managed observation rejects a PLC response missing or malforming verification methods", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const budget = createPhaseBudget(t, { deadlineMs: 5_000 }); + const respondWith = (payload) => + createManagedObservation({ + signal: budget.signal, + fetch: async () => + new Response(JSON.stringify(payload), { + status: 200, + headers: { "content-type": "application/json" }, + }), + }); + t.after(() => budget.settle()); + const rotationKeys = ["did:key:zQ3shRotationKeyForTheBoundary"]; + const cases = [ + { + name: "omitted verificationMethods", + payload: { rotationKeys }, + message: "Invalid PLC verification methods", + }, + { + name: "null verificationMethods", + payload: { rotationKeys, verificationMethods: null }, + message: "Invalid PLC verification methods", + }, + { + name: "non-string atproto method", + payload: { rotationKeys, verificationMethods: { atproto: 42 } }, + message: "Invalid PLC verification methods", + }, + { + name: "empty atproto method", + payload: { rotationKeys, verificationMethods: { atproto: "" } }, + message: "Invalid PLC verification methods", + }, + { + name: "omitted rotationKeys", + payload: { verificationMethods: { atproto: "did:key:zQ3shAtproto" } }, + message: "Invalid PLC rotation keys", + }, + ]; + for (const { name, payload, message } of cases) { + const observation = respondWith(payload); + try { + const outcome = await observation + .readLastOperation( + "https://plc.example", + "did:plc:example", + budget.child().signal, + ) + .then( + (value) => ({ value }), + (error) => ({ error }), + ); + assert.equal( + outcome.error?.message, + message, + `${name}: a malformed PLC operation must be rejected at the boundary`, + ); + } finally { + await observation.close({ settleTimeoutMs }); + } + } + }, +); + +const createPhasedClient = (t, { deadlineMs }) => { + if (!t.signal) throw new TypeError("A node test signal is required"); + assertBudgetDeadline(deadlineMs); + const budget = createPhaseBudget(t, { deadlineMs }); + const clients = new Set(); + let phaseCount = 0; + const startPhase = (options = {}) => { + const phase = budget.child(); + const client = createIncomingPublicIo({ ...options, signal: phase.signal }); + clients.add(client); + phaseCount += 1; + return { client, signal: phase.signal, controller: phase.controller }; + }; + const endPhase = async ({ client }) => { + await client.close({ settleTimeoutMs }); + clients.delete(client); + }; + return { + budget, + startPhase, + endPhase, + phaseCount: () => phaseCount, + clientCount: () => clients.size, + async retire({ settleTimeoutMs: settleMs = settleTimeoutMs } = {}) { + for (const client of clients) + await client.close({ settleTimeoutMs: settleMs }); + clients.clear(); + await budget.settle(); + }, + }; +}; + +test( + "each source phase gets a fresh budget after the preflight retires", + { timeout: preflightBudgetMs * 3 + settleTimeoutMs }, + async (t) => { + const dispatch = []; + const transport = (_url, init) => + new Promise((_resolve, reject) => { + dispatch.push(init.signal); + init.signal.addEventListener("abort", () => reject(init.signal.reason)); + }); + const settled = (promise) => + Promise.race([ + promise.then(() => "settled"), + new Promise((resolve) => setTimeout(() => resolve("pending"), 20)), + ]); + + const preflight = createPhasedClient(t, { deadlineMs: 40 }); + const prerequisite = preflight.startPhase({ fetch: transport }); + const prerequisites = prerequisite.client + .call("https://source.example/xrpc/prerequisite", { verb: "GET" }) + .catch((error) => error); + assert.equal( + (await prerequisites)?.message, + "Phase budget exhausted", + "The preflight budget expires on its own", + ); + await preflight.endPhase(prerequisite); + await preflight.retire(); + + const firstPhase = createPhasedClient(t, { deadlineMs: exportBudgetMs }); + const first = firstPhase.startPhase({ fetch: transport }); + const firstOutcome = first.client + .call("https://source.example/xrpc/getRepo", { verb: "GET" }) + .catch((error) => error); + assert.equal( + await settled(firstOutcome), + "pending", + "An expired preflight must not abort a later phase's fresh budget", + ); + first.controller.abort(new Error("first phase cancelled by test")); + assert.equal( + (await firstOutcome)?.message, + "first phase cancelled by test", + "The phase's own cancellation reason reaches the actual transport", + ); + assert.equal( + dispatch.at(-1).aborted, + true, + "Cancellation reaches the transport signal of that phase's own client", + ); + await firstPhase.endPhase(first); + await firstPhase.retire(); + + const secondPhase = createPhasedClient(t, { deadlineMs: exportBudgetMs }); + const second = secondPhase.startPhase({ fetch: transport }); + const secondOutcome = second.client + .call("https://source.example/xrpc/getRepo", { verb: "GET" }) + .catch((error) => error); + assert.equal( + await settled(secondOutcome), + "pending", + "The next phase starts with a fresh budget, not the retired one", + ); + second.controller.abort(new Error("second phase cancelled by test")); + assert.equal( + (await secondOutcome)?.message, + "second phase cancelled by test", + "Each new export gets its own fresh budget", + ); + await secondPhase.endPhase(second); + await secondPhase.retire(); + + assert.equal( + dispatch.length, + 3, + "Each phase dispatched exactly once through its own client", + ); + assert.equal(firstPhase.phaseCount(), 1); + assert.equal(secondPhase.phaseCount(), 1); + }, +); + +test( + "a cancelled parent phase reaches the request through its own client", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + let transportSignal; + const phase = createPhasedClient(t, { deadlineMs: exportBudgetMs }); + const bound = phase.startPhase({ + fetch: (_url, init) => { + transportSignal = init.signal; + return new Promise((_resolve, reject) => { + init.signal.addEventListener("abort", () => + reject(init.signal.reason), + ); + }); + }, + }); + const outcome = bound.client + .call("https://source.example/xrpc/getRepo", { verb: "GET" }) + .catch((error) => error); + await new Promise((resolve) => setTimeout(resolve, 20)); + bound.controller.abort(new Error("phase cancelled by test")); + assert.equal( + (await outcome)?.message, + "phase cancelled by test", + "The cancellation reaches the request through that client", + ); + assert.equal(transportSignal.aborted, true); + await phase.endPhase(bound); + await phase.retire(); + assert.equal(phase.clientCount(), 0, "No client remains tracked"); + }, +); + +const readProjection = async () => { + if (!process.env.INCOMING_ACCESS_PATH || !process.env.INCOMING_PROJECT) + throw new Error("Missing managed projection or project"); + const access = JSON.parse( + await readFile(process.env.INCOMING_ACCESS_PATH, "utf8"), + ); + if ( + access.schemaVersion !== 1 || + access.project !== process.env.INCOMING_PROJECT + ) + throw new Error("Projection project mismatch"); + for (const value of [ + access.sourceUrl, + access.entrywayUrl, + access.plcUrl, + access.mailpitUrl, + ...(access.destinationUrls ?? []), + ]) { + const url = new URL(value); + if ( + url.protocol !== "https:" || + url.username || + url.password || + url.origin !== value + ) + throw new Error("Invalid projection origin"); + } + if ( + !Array.isArray(access.destinationUrls) || + access.destinationUrls.length !== 2 + ) + throw new Error("Missing destination projection"); + return access; +}; + +test( + "phase budget composes an independent deadline with the node test signal", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const budget = createPhaseBudget(t, { deadlineMs: 5_000 }); + const signals = budget.signal; + assert.ok(signals, "A composed phase signal is required"); + assert.notEqual( + signals, + t.signal, + "The phase deadline must not replace the node test signal", + ); + assert.equal(t.signal.aborted, false); + assert.equal( + signals.aborted, + false, + "The composed phase signal is live while within budget", + ); + const child = budget.child(); + assert.ok(child.signal instanceof AbortSignal); + assert.notEqual( + child.signal, + signals, + "Each call gets its own child signal", + ); + assert.notEqual( + budget.child().signal, + child.signal, + "Child signals are never shared between calls", + ); + await budget.settle(); + }, +); + +test( + "an exhausted phase budget aborts the parent and its live children", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const budget = createPhaseBudget(t, { deadlineMs: 25 }); + const first = budget.child(); + const second = budget.child(); + const exhausted = await within(budget.signal, 1_000); + assert.ok(exhausted.error, "An exhausted budget must abort"); + assert.equal(exhausted.error.message, "Phase budget exhausted"); + assert.equal( + first.signal.aborted, + true, + "Child cancellation follows the parent", + ); + assert.equal(second.signal.aborted, true, "Every live child is cancelled"); + assert.equal(t.signal.aborted, false, "The test signal is left untouched"); + await budget.settle(); + }, +); + +test( + "an aggregate parent ceiling outlives its child phases", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const aggregate = createPhaseBudget(t, { deadlineMs: 2_000 }); + const phase = aggregate.child({ deadlineMs: 25 }); + assert.equal(phase.signal.aborted, false); + await within(phase.signal, 25).then(({ error }) => { + assert.equal(error?.message, "Phase budget exhausted"); + }); + assert.equal( + phase.signal.aborted, + true, + "The child phase expires on its own", + ); + assert.equal( + aggregate.signal.aborted, + false, + "A child expiry must not exhaust the aggregate ceiling", + ); + await aggregate.settle(); + }, +); + +test( + "cancelling one phase leaves the aggregate and other phases usable", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const aggregate = createPhaseBudget(t, { deadlineMs: 2_000 }); + const cancelled = aggregate.child({ deadlineMs: 2_000 }); + const survivor = aggregate.child({ deadlineMs: 2_000 }); + cancelled.controller.abort(new Error("phase cancelled by test")); + assert.equal(cancelled.signal.aborted, true); + assert.equal(cancelled.signal.reason.message, "phase cancelled by test"); + assert.equal( + aggregate.signal.aborted, + false, + "Independent phases are not cancelled with their sibling", + ); + assert.equal(survivor.signal.aborted, false); + await survivor.controller.abort(new Error("retired by test")); + await aggregate.settle(); + }, +); + +test( + "a parent ceiling expiry reaches composed children without touching the test signal", + { timeout: preflightBudgetMs + settleTimeoutMs }, + async (t) => { + const ceiling = createPhaseBudget(t, { deadlineMs: 25 }); + const child = ceiling.child(); + assert.equal(child.signal.aborted, false); + const exhausted = await within(ceiling.signal, 1_000); + assert.equal(exhausted.error?.message, "Phase budget exhausted"); + assert.equal( + child.signal.aborted, + true, + "The ceiling must reach the composed child signal", + ); + assert.equal( + child.signal.reason?.message, + "Phase budget exhausted", + "The child keeps the ceiling reason, not a rewritten one", + ); + assert.equal(t.signal.aborted, false, "The node signal is never assigned"); + await ceiling.settle(); + }, +); + +test( + "standalone public signup and PLC confirmation preflight", + { skip: !process.env.INCOMING_PUBLIC_MECHANISMS }, + async (t) => { + const access = await readProjection(); + const preflight = createPhasedClient(t, { deadlineMs: preflightBudgetMs }); + t.after(() => preflight.retire()); + const xrpc = (method) => `${access.sourceUrl}/xrpc/${method}`; + + const prerequisite = preflight.startPhase({ + timeoutMs: requestTimeoutMs, + }); + const source = prerequisite.client; + const prereqSignal = prerequisite.signal; + const description = await source.call( + xrpc("com.atproto.server.describeServer"), + { verb: "GET" }, + ); + if ( + description.inviteCodeRequired || + !Array.isArray(description.availableUserDomains) || + !description.availableUserDomains[0] + ) + throw new Error("Standalone signup unavailable"); + const unique = randomUUID().replaceAll("-", ""); + const email = `public-${unique}@example.com`; + const password = `Public-${randomUUID()}!`; + const handle = `u${unique.slice(0, 16)}${description.availableUserDomains[0]}`; + const account = await source.call( + xrpc("com.atproto.server.createAccount"), + { + json: { email, password, handle }, + }, + ); + if (typeof account.did !== "string") throw new Error("Signup omitted DID"); + const session = await source.call( + xrpc("com.atproto.server.createSession"), + { + json: { identifier: account.did, password }, + }, + ); + if (session.did !== account.did || typeof session.accessJwt !== "string") + throw new Error("Session did not match account"); + + const observation = createManagedObservation({ + signal: prereqSignal, + fetch: (url, init) => globalThis.fetch(url, init), + }); + t.after(() => observation.close({ settleTimeoutMs })); + const search = new URL("/api/v1/search", access.mailpitUrl); + search.searchParams.set("query", `to:${email}`); + search.searchParams.set("limit", "50"); + const prior = await observation.readJson(search.href, prereqSignal); + if (!Array.isArray(prior.messages)) + throw new Error("Invalid Mailpit search"); + const since = Date.now(); + const excludeIds = prior.messages.map((item) => item.ID); + await source.call( + xrpc("com.atproto.identity.requestPlcOperationSignature"), + { + token: session.accessJwt, + signal: prereqSignal, + }, + ); + const confirmation = await observation.pollStockConfirmation( + { recipient: email, since, excludeIds, baseUrl: access.mailpitUrl }, + prerequisite, + ); + const initial = await observation.readLastOperation( + access.plcUrl, + account.did, + prereqSignal, + ); + const signed = await source.call( + xrpc("com.atproto.identity.signPlcOperation"), + { token: session.accessJwt, json: { token: confirmation.token } }, + ); + if (!signed.operation) throw new Error("PLC signing omitted operation"); + await plc.assureValidSig(initial.rotationKeys, signed.operation); + assert.ok(true, "Public signing verified"); + + const exportUrl = new URL( + "/xrpc/com.atproto.sync.getRepo", + access.sourceUrl, + ); + exportUrl.searchParams.set("did", account.did); + await preflight.endPhase(prerequisite); + await preflight.retire(); + + const phaseFor = (bound) => { + const exportRepo = () => + bound.client.call(exportUrl.href, { + token: session.accessJwt, + verb: "GET", + expect: "car", + }); + const write = (writes) => + bound.client.call(xrpc("com.atproto.repo.applyWrites"), { + token: session.accessJwt, + json: { repo: account.did, writes }, + expect: "status", + }); + return { ...bound, exportRepo, write }; + }; + const collection = "app.bsky.feed.post"; + const record = (text) => ({ + $type: collection, + text, + createdAt: "2026-10-06T00:00:00.000Z", + }); + const action = (kind, rkey, text) => ({ + $type: `com.atproto.repo.applyWrites#${kind}`, + collection, + rkey, + ...(text === undefined ? {} : { value: record(text) }), + }); + const [one, two, three] = [ + "3mznesxxhic2e", + "3mznesxxhid2e", + "3mznesxxhie2e", + ]; + + await t.test( + "full source CAR verifies against the observed PLC key", + { timeout: exportBudgetMs + settleTimeoutMs }, + async (child) => { + const owner = createPhasedClient(child, { deadlineMs: exportBudgetMs }); + child.after(() => owner.retire()); + const phase = phaseFor( + owner.startPhase({ timeoutMs: requestTimeoutMs }), + ); + try { + const bytes = await phase.exportRepo(); + assert.ok(bytes.byteLength <= carLimitBytes); + const verified = await verifyRepoCar(bytes, account.did); + const car = await readCarWithRoot(bytes); + const commit = ensureV3Commit( + await new MemoryBlockstore(car.blocks).readObj( + car.root, + def.versionedCommit, + ), + ); + assert.equal(verified.commit.cid.toString(), car.root.toString()); + assert.equal(commit.did, account.did); + assert.equal(verified.creates.length, 0); + assert.equal( + await verifyCommitSig(commit, initial.verificationMethods.atproto), + true, + ); + } finally { + await owner.retire(); + } + }, + ); + + await t.test( + "nonempty source snapshots and export-before-write request ordering", + { timeout: exportBudgetMs + settleTimeoutMs }, + async (child) => { + const owner = createPhasedClient(child, { deadlineMs: exportBudgetMs }); + child.after(() => owner.retire()); + const phase = phaseFor( + owner.startPhase({ timeoutMs: requestTimeoutMs }), + ); + const { exportRepo, write } = phase; + try { + const verifyExport = async (bytes) => { + const verified = await verifyRepoCar( + bytes, + account.did, + initial.verificationMethods.atproto, + ); + const car = await readCarWithRoot(bytes); + assert.equal(verified.commit.cid.toString(), car.root.toString()); + return { + root: car.root.toString(), + inventory: verified.creates + .map( + ({ collection: name, rkey, cid }) => `${name}/${rkey}/${cid}`, + ) + .sort(), + }; + }; + assert.equal( + await write([ + action("create", one, "original"), + action("create", two, "removed"), + ]), + 200, + ); + const before = await verifyExport(await exportRepo()); + assert.equal(before.inventory.length, 2); + assert.equal( + await write([ + action("update", one, "updated"), + action("delete", two), + action("create", three, "added"), + ]), + 200, + ); + const after = await verifyExport(await exportRepo()); + assert.equal(after.inventory.length, before.inventory.length); + assert.notDeepEqual(after.inventory, before.inventory); + assert.notEqual(after.root, before.root); + assert.deepEqual( + after.inventory.map((entry) => entry.split("/")[1]), + [one, three], + ); + const pendingExports = [exportRepo(), exportRepo()]; + const overlapWrite = write([action("update", one, "overlap")]); + const captures = await Promise.all( + (await Promise.all(pendingExports)).map(verifyExport), + ); + assert.equal(await overlapWrite, 200); + const final = await verifyExport(await exportRepo()); + assert.equal(final.inventory.length, 2); + assert.notDeepEqual(final.inventory, after.inventory); + for (const capture of captures) { + const matching = [after, final].find( + (snapshot) => snapshot.root === capture.root, + ); + assert.ok( + matching, + "Captured root must be a known complete state; the export was in flight when the write was dispatched and the server may already have buffered its response, so overlap is not a finality proof", + ); + assert.deepEqual(capture.inventory, matching.inventory); + } + } finally { + await owner.retire(); + } + }, + ); + }, +); + +test( + "Entryway-hosted inactive destination and streamed import", + { + skip: !process.env.INCOMING_PUBLIC_MECHANISMS, + timeout: destinationBudgetMs + settleTimeoutMs, + }, + async (t) => { + const access = await readProjection(); + const { chromium } = await import("playwright"); + const { createAccount, config } = + await import("../support/helpers/browser-oauth.mjs"); + const phaseBudget = createPhaseBudget(t, { + deadlineMs: destinationBudgetMs, + }); + const browser = await chromium.launch({ headless: true }); + let destination; + t.after(async () => { + await destination?.close({ settleTimeoutMs }); + await browser.close(); + await phaseBudget.settle(); + }); + const page = await browser.newPage(); + const identity = await createAccount(page, "destination-probe"); + const password = `Public qualification ${randomUUID()}!`; + const form = page.locator('form[action="/account/password-set"]'); + await form.getByLabel("New account password").fill(password); + await form.getByRole("button").click(); + const origin = config.pds[0].url; + assert.ok( + access.destinationUrls.includes(origin), + "Hosted PDS must be projected destination", + ); + destination = createIncomingPublicIo({ + timeoutMs: requestTimeoutMs, + carLimitBytes, + signal: phaseBudget.signal, + }); + const xrpc = (method) => `${origin}/xrpc/${method}`; + const attempt = async (method, options = {}) => { + try { + return { + status: 200, + body: await destination.call(xrpc(method), options), + }; + } catch (error) { + if (!(error instanceof ResponseStatusError)) throw error; + return { status: error.status, body: error.body }; + } + }; + const login = await attempt("com.atproto.server.createSession", { + json: { identifier: identity.did, password }, + }); + assert.equal(login.status, 200); + const session = login.body; + assert.equal(session.did, identity.did); + assert.equal(typeof session.accessJwt, "string"); + assert.equal(typeof session.refreshJwt, "string"); + const token = session.accessJwt; + const query = `?did=${encodeURIComponent(identity.did)}`; + const exportRepo = () => + destination.call(`${xrpc("com.atproto.sync.getRepo")}${query}`, { + token, + verb: "GET", + expect: "car", + }); + const write = (writes) => + attempt("com.atproto.repo.applyWrites", { + token, + json: { repo: identity.did, writes }, + }); + const inventoryOf = (bytes) => + verifyRepoCar(bytes, identity.did).then(({ creates }) => + creates + .map(({ collection: name, rkey, cid }) => `${name}/${rkey}/${cid}`) + .sort(), + ); + const collection = "app.bsky.feed.post"; + const rkey = "3mznesxxhic2e"; + const createAction = { + $type: "com.atproto.repo.applyWrites#create", + collection, + rkey, + value: { + $type: collection, + text: "destination snapshot", + createdAt: "2026-10-06T00:00:00.000Z", + }, + }; + const deleteAction = { + $type: "com.atproto.repo.applyWrites#delete", + collection, + rkey, + }; + assert.equal((await write([createAction])).status, 200); + const originalCar = await exportRepo(); + const originalInventory = await inventoryOf(originalCar); + assert.deepEqual( + originalInventory.map((entry) => entry.split("/")[1]), + [rkey], + "Destination observation compares a full sorted inventory, never a count or presence claim", + ); + assert.equal((await write([deleteAction])).status, 200); + assert.deepEqual(await inventoryOf(await exportRepo()), []); + + const deactivated = await attempt("com.atproto.server.deactivateAccount", { + token, + json: {}, + }); + t.diagnostic(`deactivate backend HTTP ${deactivated.status}`); + assert.equal(deactivated.status, 200, "public deactivation prerequisite"); + + const status = async () => { + const result = await attempt("com.atproto.server.checkAccountStatus", { + token, + verb: "GET", + }); + assert.equal(result.status, 200); + return result.body; + }; + assert.equal((await status()).activated, false); + assert.deepEqual(await inventoryOf(await exportRepo()), []); + + const missing = await attempt("com.atproto.repo.listMissingBlobs", { + token, + verb: "GET", + }); + assert.equal(missing.status, 200); + assert.deepEqual(missing.body.blobs, []); + assert.equal( + missing.body.cursor, + undefined, + "A blob-free repository reports no cursor; this is not pagination", + ); + + const inactiveWrite = await write([createAction]); + assert.equal(inactiveWrite.status, 401); + assert.equal(inactiveWrite.body?.error, "AccountDeactivated"); + assert.deepEqual(await inventoryOf(await exportRepo()), []); + + assert.equal( + ( + await attempt("com.atproto.server.refreshSession", { + token: session.refreshJwt, + }) + ).status, + 400, + "Deactivation also refuses the local refresh family", + ); + + const firstImport = await attempt("com.atproto.repo.importRepo", { + token, + bytes: originalCar, + expect: "text", + }); + assert.equal(firstImport.status, 200); + assert.equal( + firstImport.body.length, + 0, + "A successful import returns an empty body; it is not data", + ); + const importedInventory = await inventoryOf(await exportRepo()); + assert.deepEqual(importedInventory, originalInventory); + + const repeatedImport = await attempt("com.atproto.repo.importRepo", { + token, + bytes: originalCar, + expect: "text", + }); + assert.equal(repeatedImport.status, 200); + assert.equal(repeatedImport.body.length, 0); + assert.deepEqual( + await inventoryOf(await exportRepo()), + importedInventory, + "A repeated same-DID import is idempotent", + ); + + const upload = await attempt("com.atproto.repo.uploadBlob", { + token, + bytes: new Uint8Array([137, 80, 78, 71, 13, 10, 26, 10]), + headers: { "content-type": "image/png" }, + }); + assert.equal(upload.status, 200); + assert.equal(typeof upload.body.blob?.ref?.$link, "string"); + assert.equal(upload.body.blob.size, 8); + assert.equal(upload.body.blob.mimeType, "image/png"); + + let release; + const gate = new Promise((resolve) => { + release = resolve; + }); + let started; + const sent = new Promise((resolve) => { + started = resolve; + }); + const stream = new ReadableStream({ + async start(controller) { + controller.enqueue(originalCar.subarray(0, originalCar.length - 1)); + started(); + await gate; + try { + controller.enqueue(originalCar.subarray(originalCar.length - 1)); + controller.close(); + } catch { + release(); + } + }, + cancel() { + release(); + }, + }); + const held = runHeldStreamedCall( + destination, + xrpc("com.atproto.repo.importRepo"), + { token, bytes: stream, expect: "text", deadlineMs: requestTimeoutMs }, + ); + t.after(() => held.retireNow({ settleTimeoutMs })); + try { + await sent; + const held = await status(); + assert.equal(held.activated, false); + assert.equal(typeof held.repoCommit, "string"); + assert.equal(typeof held.repoRev, "string"); + assert.equal(typeof held.indexedRecords, "number"); + assert.equal(typeof held.repoBlocks, "number"); + assert.equal( + ( + await attempt("com.atproto.server.activateAccount", { + token, + }) + ).status, + 401, + "Delegated activation is refused and the account stays inactive: deactivation revoked the local session family, the Entryway hop passes the backend's stable refusal through, and this is not evidence of a pending-arrival guard", + ); + } finally { + release(); + } + await held.consume({ expect: "text" }); + + const finalStatus = await status(); + assert.equal(finalStatus.activated, false); + assert.deepEqual( + await inventoryOf(await exportRepo()), + importedInventory, + "The released import preserves the exact completed inventory; this equality is a separate observation and does not by itself prove the held stream succeeded", + ); + t.diagnostic( + "Withholding the final client byte does not prove the server received the prefix or establish universal finality; blob retention and pagination remain unqualified.", + ); + }, +); + +test( + "public reservation sequential retry and placement isolation", + { + skip: !process.env.INCOMING_PUBLIC_MECHANISMS, + timeout: reservationBudgetMs + settleTimeoutMs, + }, + async (t) => { + const access = await readProjection(); + const phaseBudget = createPhaseBudget(t, { + deadlineMs: reservationBudgetMs, + }); + const reservation = createIncomingPublicIo({ + timeoutMs: requestTimeoutMs, + signal: phaseBudget.signal, + }); + t.after(async () => { + await reservation.close({ settleTimeoutMs }); + await phaseBudget.settle(); + }); + const reserve = async (baseUrl, did) => { + const result = await reservation.call( + `${baseUrl}/xrpc/com.atproto.server.reserveSigningKey`, + { json: { did }, signal: phaseBudget.child().signal }, + ); + assert.match(result.signingKey, /^did:key:/); + return result.signingKey; + }; + const did = `did:plc:${"a".repeat(24)}`; + const first = await reserve(access.destinationUrls[0], did); + assert.equal(await reserve(access.destinationUrls[0], did), first); + assert.notEqual(await reserve(access.destinationUrls[1], did), first); + assert.notEqual( + await reserve(access.destinationUrls[0], `did:plc:${"b".repeat(24)}`), + first, + ); + }, +); diff --git a/tests/local.sh b/tests/local.sh index e1922da4..c9c6c984 100755 --- a/tests/local.sh +++ b/tests/local.sh @@ -9,17 +9,19 @@ export SANDBOX_CHECKOUT="$sandbox" SANDBOX_PROJECT="$project" ACCEPTANCE_REPORT_ [[ "$project" =~ ^hypercerts-entryway(-[a-z0-9_-]+)?$ ]] || { echo "Invalid Entryway project name" >&2; exit 2; } mkdir -p "$sandbox" cd "$sandbox" +sandbox=$PWD +export SANDBOX_CHECKOUT="$sandbox" command=${1:-help} mkdir -p "$artifacts" # Phase identity comes from the controller, never from a command or request body. case "$command" in - fresh|fresh-group|prepare|up|contracts|database-contracts|database-dialects|resilience-profiles|database-profile|browser|migration|migration-resume|resilience|reverify|interop-profile|interop-acceptance|plc-recovery|process-crash|managed-recovery|authority-drills|all|down|status|help) ;; + fresh|fresh-group|prepare|up|contracts|database-contracts|database-dialects|resilience-profiles|database-profile|browser|incoming-public-mechanisms|migration|migration-resume|resilience|reverify|interop-profile|interop-acceptance|plc-recovery|process-crash|managed-recovery|authority-drills|all|down|status|help) ;; *) echo 'code=UNKNOWN_PHASE'; exit 2 ;; esac printf 'phase=%s status=started\n' "$command" trap 'result=$?; printf "phase=%s status=finished exit=%s\n" "$command" "$result"' EXIT source "$root/tests/support/rootless-docker.sh" -compose() { docker --context "$DOCKER_CONTEXT" compose --project-name "$project" -f "$sandbox/compose.yaml" "$@"; } +compose() { docker --context "$DOCKER_CONTEXT" compose --project-name "$project" -f "$sandbox/compose.yaml" -f "$root/tests/atmosphere/stacks/incoming-source-mail.yaml" "$@"; } guard_project() { node --input-type=module - "$sandbox/state/manifest.json" "$project" <<'JS' import { readFileSync } from 'node:fs' @@ -32,7 +34,33 @@ require_state() { test -f state/manifest.json && test -f compose.yaml || { echo 'Run ./tests/local.sh prepare first' >&2; exit 2; } guard_project } -browser() { compose run --rm --no-deps browser npx --no-install playwright test --config tests/playwright.config.mjs "$@"; } +prepare_incoming_access() { + local routes temporary + routes=$(node --input-type=module - "$root/tests/sandbox.json" "$root/tests/atmosphere/stacks/hypercerts-entryway.definition.json" "$root/tests/atmosphere/stacks/entryway-fixtures.definition.json" <<'JS' +import { readFileSync } from 'node:fs' +const [manifest, entryway, fixtures] = process.argv.slice(2).map(path => JSON.parse(readFileSync(path, 'utf8'))) +const source = manifest.pds.filter(pds => pds.id === 'pds3') +if (source.length !== 1 || source[0].hostname !== new URL(source[0].url).hostname) throw new Error('Invalid configured standalone PDS') +function route(definition, id, service) { + const matches = definition.routes.filter(item => item.id === id && item.service === service) + if (matches.length !== 1 || !matches[0].host || matches[0].hostname) throw new Error('Invalid configured route') + return `https://${matches[0].host}.${manifest.domain}` +} +console.log([source[0].url, route(entryway, 'entryway', 'entryway'), route(entryway, 'cluster1', 'pds1'), route(entryway, 'cluster2', 'pds2'), manifest.plc, route(fixtures, 'mailpit', 'mail-capture')].join(' ')) +JS + ) || return + local -a origins + read -r -a origins <<< "$routes" + deno task sandbox access --json > "$sandbox/access.json" + node "$root/tests/atmosphere/validate-access.mjs" "$sandbox/access.json" "$project" + temporary=$(mktemp "$sandbox/.incoming-access.XXXXXX") + if ! node "$root/tests/atmosphere/project-incoming-access.mjs" "$sandbox/access.json" "$project" "${origins[@]}" > "$temporary"; then + rm -f -- "$temporary" + return 1 + fi + mv -f -- "$temporary" "$sandbox/incoming-access.json" +} +browser() { compose run --rm --no-deps --volume "$sandbox/incoming-access.json:/run/incoming-access.json:ro" -e INCOMING_ACCESS_PATH=/run/incoming-access.json browser npx --no-install playwright test --config tests/playwright.config.mjs "$@"; } external_browser() { local phase=$1; compose run --rm --no-deps -e "EXTERNAL_PHASE=$phase" browser npx --no-install playwright test --config tests/external.playwright.config.mjs 2>&1 | tee "$artifacts/external-browser-$phase.log" | python3 "$root/scripts/ci-summary.py"; } migration_command() { compose exec -T entryway node tests/support/external-migration.mjs "$@"; } case "$command" in @@ -101,7 +129,7 @@ case "$command" in echo "Preserved project $SANDBOX_PROJECT at $temp" >&2 else if [[ -f "$SANDBOX_CHECKOUT/compose.yaml" ]] \ - && ! docker --context "$DOCKER_CONTEXT" compose --project-name "$SANDBOX_PROJECT" -f "$SANDBOX_CHECKOUT/compose.yaml" down --volumes --remove-orphans; then + && ! docker --context "$DOCKER_CONTEXT" compose --project-name "$SANDBOX_PROJECT" -f "$SANDBOX_CHECKOUT/compose.yaml" -f "$root/tests/atmosphere/stacks/incoming-source-mail.yaml" down --volumes --remove-orphans; then echo "Scoped cleanup failed; preserved $temp for project $SANDBOX_PROJECT" >&2 (( result != 0 )) || result=1 echo "Group exit: $result; reports: $ACCEPTANCE_REPORT_DIR" @@ -141,6 +169,7 @@ case "$command" in compose up -d --wait entryway-postgres fi deno task sandbox up --build + compose up -d --wait --no-deps pds3 compose build test browser deno task sandbox access --json > "$sandbox/access.json" node "$root/tests/atmosphere/validate-access.mjs" "$sandbox/access.json" "$project" @@ -169,8 +198,14 @@ case "$command" in ;; browser) require_state + prepare_incoming_access browser 2>&1 | tee "$artifacts/browser.log" | python3 "$root/scripts/ci-summary.py" ;; + incoming-public-mechanisms) + require_state + prepare_incoming_access + compose run --rm --no-deps --volume "$sandbox/incoming-access.json:/run/incoming-access.json:ro" -e INCOMING_ACCESS_PATH=/run/incoming-access.json -e "INCOMING_PROJECT=$project" -e INCOMING_PUBLIC_MECHANISMS=1 browser node --test tests/contracts/incoming-public-mechanisms.test.mjs + ;; migration) require_state exec 9> "$artifacts/migration.lock" @@ -249,7 +284,7 @@ case "$command" in deno task sandbox status ;; *) - echo 'Usage: ./tests/local.sh fresh|fresh-group |prepare|up|browser|contracts|database-contracts|database-profile|resilience-profiles|migration|migration-resume|reverify|interop-profile|interop-acceptance|plc-recovery|process-crash|managed-recovery|authority-drills|resilience|all|status|down' + echo 'Usage: ./tests/local.sh fresh|fresh-group |prepare|up|browser|incoming-public-mechanisms|contracts|database-contracts|database-profile|resilience-profiles|migration|migration-resume|reverify|interop-profile|interop-acceptance|plc-recovery|process-crash|managed-recovery|authority-drills|resilience|all|status|down' echo 'all requires a fresh project fixture; down retains state and volumes. No implicit reset.' ;; esac diff --git a/tests/support/ci-lifecycle.test.py b/tests/support/ci-lifecycle.test.py index 89a5b0d9..f09e8182 100644 --- a/tests/support/ci-lifecycle.test.py +++ b/tests/support/ci-lifecycle.test.py @@ -128,6 +128,26 @@ def test_explicit_default_guard(self): self.assertNotEqual(self.guard(ENTRYWAY_DOCKER_CONTEXT="default").returncode, 0) self.assertNotEqual(self.guard(ENTRYWAY_DOCKER_CONTEXT="foreign").returncode, 0) + def test_incoming_helpers_have_explicit_native_coverage(self): + source = Path(__file__).resolve().parents[2] + package = json.loads((source / "package.json").read_text()) + helpers = ["tests/support/helpers/stock-plc-mailpit.test.mjs", + "tests/support/helpers/incoming-public-io.test.mjs", + "tests/atmosphere/project-incoming-access.test.mjs"] + self.assertEqual(package["scripts"]["test:incoming-helpers"].split(), + ["node", "--test", *helpers]) + checks = (source / ".github/workflows/ci.yml").read_text().split(" integration:", 1)[0] + self.assertEqual(checks.count("run: npm run test:incoming-helpers\n"), 1) + discovery = subprocess.run(["node", "--input-type=module", "-e", + "import {discoverContracts} from './tests/support/discover-contracts.mjs';" + "console.log(JSON.stringify(discoverContracts(undefined,{unit:true})))"], + cwd=source, capture_output=True, text=True, check=True) + files = json.loads(discovery.stdout) + self.assertEqual(len(files), len(set(files))) + for helper in helpers: + self.assertNotIn(helper, files) + self.assertNotIn("tests/contracts/incoming-public-mechanisms.test.mjs", files) + def test_live_controller_shell_syntax(self): source = Path(__file__).resolve().parents[2] for script in [source / "tests/local.sh", *sorted((source / "tests/support").glob("*.sh"))]: diff --git a/tests/support/helpers/incoming-public-io.mjs b/tests/support/helpers/incoming-public-io.mjs new file mode 100644 index 00000000..38f08829 --- /dev/null +++ b/tests/support/helpers/incoming-public-io.mjs @@ -0,0 +1,428 @@ +const defaultCarLimitBytes = 4 * 1024 * 1024; +const defaultSettleTimeoutMs = 5_000; + +const assertTimeout = (timeoutMs) => { + if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 1) + throw new RangeError("Invalid request timeout"); +}; + +const assertSignal = (signal, label) => { + if ( + signal !== undefined && + !( + signal instanceof AbortSignal || + (typeof signal === "object" && + signal !== null && + typeof signal.aborted === "boolean" && + typeof signal.addEventListener === "function" && + typeof signal.removeEventListener === "function") + ) + ) + throw new TypeError(`Invalid ${label} signal`); +}; + +const assertExternalSignal = (signal) => { + if (signal === undefined) return; + assertSignal(signal, "external abort"); + if (signal.aborted) throw new Error("Test aborted"); +}; + +export class ResponseStatusError extends Error { + constructor(status, body) { + super("Unexpected response status"); + this.name = "ResponseStatusError"; + this.status = status; + this.body = body; + } +} + +export class ResponseBodyError extends Error { + constructor(code) { + super(code); + this.name = "ResponseBodyError"; + this.code = code; + } +} + +const abortReason = (signal) => + signal.reason ?? new DOMException("The operation was aborted.", "AbortError"); + +const withSignal = (promise, signal) => + new Promise((resolve, reject) => { + if (signal.aborted) { + reject(abortReason(signal)); + return; + } + const onAbort = () => reject(abortReason(signal)); + signal.addEventListener("abort", onAbort, { once: true }); + promise.then( + (value) => { + signal.removeEventListener("abort", onAbort); + resolve(value); + }, + (error) => { + signal.removeEventListener("abort", onAbort); + reject(error); + }, + ); + }); + +const discard = (promise) => { + promise.catch(() => {}); +}; + +const isReadableStream = (body) => + typeof ReadableStream !== "undefined" && body instanceof ReadableStream; + +const disposeResponseBody = (response, signal) => { + if (!response.body) return Promise.resolve(); + return withSignal( + Promise.resolve(response.body.cancel(abortReason(signal))), + signal, + ); +}; + +const decodeError = (bytes) => { + try { + const text = new TextDecoder("utf-8", { fatal: true }).decode(bytes); + return text ? JSON.parse(text) : undefined; + } catch { + return undefined; + } +}; + +const decodeText = async (bytes) => { + try { + return new TextDecoder("utf-8", { fatal: true }).decode(bytes); + } catch { + throw new Error("Malformed response body"); + } +}; + +export const createIncomingPublicIo = ({ + fetch: fetchImpl = globalThis.fetch, + timeoutMs = 15_000, + carLimitBytes = defaultCarLimitBytes, + signal: factorySignal, +} = {}) => { + if (typeof fetchImpl !== "function") throw new TypeError("Invalid fetch"); + assertTimeout(timeoutMs); + assertExternalSignal(factorySignal); + if (!Number.isSafeInteger(carLimitBytes) || carLimitBytes < 1) + throw new RangeError("Invalid CAR limit"); + const inflight = new Set(); + const deadlines = new Map(); + const requests = new Set(); + let closed = false; + + const later = (fn, ms) => { + if (closed) throw new Error("Incoming public I/O is closed"); + const handle = setTimeout(() => { + deadlines.delete(handle); + fn(); + }, ms); + deadlines.set(handle, fn); + return handle; + }; + + const clearDeadline = (handle) => { + clearTimeout(handle); + deadlines.delete(handle); + }; + + const clearDeadlines = () => { + for (const handle of deadlines.keys()) clearTimeout(handle); + deadlines.clear(); + }; + + const closeRequest = ({ controller, deadline }) => { + requests.delete(controller); + clearDeadline(deadline); + }; + + const openRequest = async ( + url, + { method = "GET", headers = {}, body, signal, redirect } = {}, + ) => { + assertExternalSignal(signal); + const controller = new AbortController(); + const external = [factorySignal, signal].filter(Boolean); + const composed = external.length + ? AbortSignal.any([...external, controller.signal]) + : controller.signal; + const entry = { + controller, + deadline: later(() => controller.abort(), timeoutMs), + composed, + }; + requests.add(controller); + const fetched = fetchImpl(url, { + method, + headers, + ...(body === undefined + ? {} + : { body, ...(isReadableStream(body) ? { duplex: "half" } : {}) }), + signal: composed, + ...(redirect === undefined ? {} : { redirect }), + }); + try { + const response = await withSignal(fetched, composed); + return { ...entry, response }; + } catch (error) { + discard( + Promise.resolve(fetched) + .then((response) => disposeResponseBody(response, composed)) + .catch(() => {}), + ); + closeRequest(entry); + throw error; + } + }; + + const disposeBody = (entry) => { + const { body } = entry.response; + if (!body || body.locked) return Promise.resolve(); + return withSignal( + Promise.resolve(body.cancel(abortReason(entry.composed))), + entry.composed, + ); + }; + + const consumeBody = async (entry, { read, dispose = true }) => { + try { + const value = await withSignal( + Promise.resolve(read(entry.response, entry.composed)), + entry.composed, + ); + if (dispose) await disposeBody(entry); + return value; + } catch (error) { + discard(disposeBody(entry).catch(() => {})); + throw error; + } finally { + closeRequest(entry); + } + }; + + const readStatus = async (options) => { + const entry = await openRequest(options.url, options); + return consumeBody(entry, { + read: (response) => response.status, + dispose: entry.response.body !== null, + }); + }; + + const readJson = async (options) => { + const entry = await openRequest(options.url, options); + return consumeBody(entry, { + read: async (response) => { + const bytes = await response.arrayBuffer(); + if (!response.ok) + throw new ResponseStatusError(response.status, decodeError(bytes)); + const type = response.headers.get("content-type") ?? ""; + if (!bytes.byteLength) return undefined; + if (!/^application\/json\b/i.test(type)) + throw new Error("Unexpected response content type"); + return JSON.parse(await decodeText(bytes)); + }, + }); + }; + + const readText = async (options) => { + const entry = await openRequest(options.url, options); + return consumeBody(entry, { + read: async (response) => { + const bytes = await response.arrayBuffer(); + if (!response.ok) + throw new ResponseStatusError(response.status, decodeError(bytes)); + return decodeText(bytes); + }, + }); + }; + + const assertCarReadiness = (response, composed) => { + if (!response.ok) { + discard(disposeResponseBody(response, composed)); + throw new ResponseStatusError(response.status); + } + const declared = response.headers.get("content-length"); + if (declared !== null) { + if (!/^\d+$/.test(declared.trim())) { + discard(disposeResponseBody(response, composed)); + throw new ResponseBodyError("Malformed response content length"); + } + if (Number(declared) > carLimitBytes) { + discard(disposeResponseBody(response, composed)); + throw new ResponseBodyError("Response exceeds CAR limit"); + } + } + if (!response.body) throw new ResponseBodyError("Response omitted body"); + }; + + const streamCar = async (response, composed) => { + assertCarReadiness(response, composed); + const reader = response.body.getReader(); + const onAbort = () => { + discard(reader.cancel(abortReason(composed))); + }; + composed.addEventListener("abort", onAbort, { once: true }); + const chunks = []; + let size = 0; + let failure; + try { + for (;;) { + const { done, value } = await withSignal(reader.read(), composed); + if (done) break; + size += value.byteLength; + if (size > carLimitBytes) { + discard(reader.cancel(abortReason(composed))); + throw new ResponseBodyError("Response exceeds CAR limit"); + } + chunks.push(value); + } + } catch (error) { + failure = error; + discard(reader.cancel(abortReason(composed))); + } finally { + composed.removeEventListener("abort", onAbort); + } + if (failure) throw failure; + const bytes = new Uint8Array(size); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; + } + return bytes; + }; + + const readCar = async (options) => { + const entry = await openRequest(options.url, options); + return consumeBody(entry, { + read: (response, composed) => streamCar(response, composed), + dispose: false, + }); + }; + + const consumers = { + status: readStatus, + json: readJson, + text: readText, + car: readCar, + }; + + const call = async ( + url, + { + token, + json, + bytes, + verb = "POST", + headers: extra = {}, + signal, + expect = "json", + redirect, + } = {}, + ) => { + if (closed) throw new Error("Incoming public I/O is closed"); + const readResponse = consumers[expect]; + if (!readResponse) throw new TypeError("Invalid response consumption"); + const headers = { + ...(token ? { authorization: `Bearer ${token}` } : {}), + ...(json !== undefined ? { "content-type": "application/json" } : {}), + ...(bytes !== undefined + ? { "content-type": "application/vnd.ipld.car" } + : {}), + ...extra, + }; + const body = json !== undefined ? JSON.stringify(json) : bytes; + return await track( + readResponse({ url, method: verb, headers, body, signal, redirect }), + ); + }; + + const track = (promise) => { + inflight.add(promise); + promise.then( + () => inflight.delete(promise), + () => inflight.delete(promise), + ); + return promise; + }; + + const close = async ({ settleTimeoutMs = defaultSettleTimeoutMs } = {}) => { + if (closed) return; + assertTimeout(settleTimeoutMs); + closed = true; + for (const abort of deadlines.values()) abort(); + clearDeadlines(); + const pending = [...inflight]; + for (const controller of requests) controller.abort(); + if (!pending.length) return; + let settleTimer; + const bounded = new Promise((resolve) => { + settleTimer = setTimeout(() => resolve("unsettled"), settleTimeoutMs); + }); + const outcome = await Promise.race([ + Promise.allSettled(pending).then(() => "settled"), + bounded, + ]); + clearTimeout(settleTimer); + if (outcome === "unsettled") throw new Error("Requests did not settle"); + }; + + return { call, close, inflight }; +}; + +const defaultHeldDeadlineMs = 15_000; + +const assertHeldDeadline = (deadlineMs) => { + if (!Number.isSafeInteger(deadlineMs) || deadlineMs < 1) + throw new RangeError("Invalid held stream deadline"); +}; + +export const runHeldStreamedCall = (io, url, options = {}) => { + const { + deadlineMs = defaultHeldDeadlineMs, + signal, + ...callOptions + } = options; + assertHeldDeadline(deadlineMs); + if (typeof io?.call !== "function") throw new TypeError("Invalid client"); + const controller = new AbortController(); + const composed = signal + ? AbortSignal.any([signal, controller.signal]) + : controller.signal; + const timer = setTimeout( + () => controller.abort(new Error("Held stream deadline exhausted")), + deadlineMs, + ); + const retire = () => { + clearTimeout(timer); + if (!controller.signal.aborted) + controller.abort(new Error("Held stream retired")); + }; + const settlement = withSignal( + io.call(url, { ...callOptions, signal: composed }), + composed, + ); + const tagged = settlement.then( + (body) => ({ outcome: "success", body }), + (error) => ({ outcome: "failure", error }), + ); + return { + tagged, + async consume({ expect } = {}) { + const result = await tagged; + if (result.outcome === "failure") throw result.error; + if (expect === "text" && result.body !== "") + throw new Error("Held stream settlement is not an empty body"); + return result.body; + }, + async retireNow({ settleTimeoutMs = defaultSettleTimeoutMs } = {}) { + assertTimeout(settleTimeoutMs); + retire(); + return await withSignal(tagged, AbortSignal.timeout(settleTimeoutMs)); + }, + }; +}; diff --git a/tests/support/helpers/incoming-public-io.test.mjs b/tests/support/helpers/incoming-public-io.test.mjs new file mode 100644 index 00000000..214f5ab0 --- /dev/null +++ b/tests/support/helpers/incoming-public-io.test.mjs @@ -0,0 +1,1097 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { + ResponseBodyError, + ResponseStatusError, + createIncomingPublicIo, + runHeldStreamedCall, +} from "./incoming-public-io.mjs"; + +const jsonResponse = (body, status = 200) => { + const bytes = new TextEncoder().encode(JSON.stringify(body)); + return new Response(bytes, { + status, + headers: { + "content-type": "application/json", + "content-length": String(bytes.byteLength), + }, + }); +}; + +const carResponse = (chunks, { declared } = {}) => { + const stream = new ReadableStream({ + async start(controller) { + for (const chunk of chunks) { + controller.enqueue(chunk); + await Promise.resolve(); + } + controller.close(); + }, + }); + const headers = { "content-type": "application/vnd.ipld.car" }; + if (declared !== undefined) headers["content-length"] = String(declared); + return new Response(stream, { status: 200, headers }); +}; + +const hangingResponse = () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); + +const within = async (promise, limitMs) => { + const outcome = await Promise.race([ + promise.then( + (value) => ({ value }), + (error) => ({ error }), + ), + new Promise((resolve) => + setTimeout(() => resolve({ pending: true }), limitMs), + ), + ]); + assert.equal(outcome.pending, undefined, "Operation must settle in time"); + return outcome; +}; + +const disposalOrderingResponse = (disposed) => ({ + status: 204, + ok: true, + headers: new Headers(), + body: { + getReader: () => ({ + read: () => Promise.resolve({ done: true, value: undefined }), + cancel: () => { + disposed.push("reader"); + return Promise.resolve(); + }, + }), + cancel: () => { + disposed.push("body"); + return Promise.resolve(); + }, + }, +}); + +const guardedStreamResponse = ({ ready = {} } = {}) => { + const guard = (name) => { + if (!ready[name]) throw new Error(`Stream ${name} is not ready`); + }; + const state = { cancelled: false }; + const response = { + status: 200, + ok: true, + headers: new Headers({ "content-type": "application/vnd.ipld.car" }), + body: { + getReader: () => { + guard("getReader"); + return { + read: () => { + guard("read"); + return new Promise(() => {}); + }, + cancel: (reason) => { + guard("cancel"); + state.cancelled = true; + state.reason = reason; + return Promise.resolve(); + }, + }; + }, + cancel: (reason) => { + guard("cancel"); + state.cancelled = true; + state.reason = reason; + return Promise.resolve(); + }, + }, + }; + return { response, state }; +}; + +test("per-call redirect policy reaches the transport without changing its default", async () => { + const requests = []; + const io = createIncomingPublicIo({ + fetch: async (url, init) => { + requests.push({ init, request: new Request(url, init) }); + return jsonResponse({ ok: true }); + }, + }); + try { + for (const redirect of [undefined, "error", "manual", "follow"]) { + assert.deepEqual( + await io.call("https://pds.example/xrpc/a", { redirect }), + { ok: true }, + ); + const received = requests.at(-1); + assert.equal(received.request.redirect, redirect ?? "follow"); + assert.equal( + Object.hasOwn(received.init, "redirect"), + redirect !== undefined, + ); + } + } finally { + await io.close(); + } +}); + +test("injected response headers overrun the phase deadline", async () => { + const stalls = []; + const io = createIncomingPublicIo({ + timeoutMs: 40, + fetch: (_url, init) => { + init.signal.addEventListener("abort", () => stalls.push(Date.now())); + return new Promise(() => {}); + }, + }); + const started = Date.now(); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "status" }), + 1_000, + ); + assert.ok(outcome.error, "A stalled header read must reject"); + assert.ok(Date.now() - started < 1_000, "Rejection must be prompt"); + assert.equal(stalls.length, 1, "The deadline must abort the fetch"); + await io.close(); +}); + +test("status consumption stays pending until the body is disposed", async () => { + const disposed = []; + let dispose; + const disposalGate = new Promise((resolve) => { + dispose = resolve; + }); + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => { + const response = disposalOrderingResponse(disposed); + response.body.cancel = () => { + disposed.push("body"); + return disposalGate; + }; + return response; + }, + }); + const pending = io.call("https://pds.example/xrpc/a", { expect: "status" }); + const settledEarly = await Promise.race([ + pending.then( + () => "settled", + () => "settled", + ), + new Promise((resolve) => setTimeout(() => resolve("pending"), 50)), + ]); + assert.equal( + settledEarly, + "pending", + "The call must not resolve before the body is disposed", + ); + assert.deepEqual(disposed, ["body"], "Disposal must be under way"); + dispose(); + assert.equal(await pending, 204); + await io.close(); +}); + +test("a null status body succeeds without disposal", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => new Response(null, { status: 204 }), + }); + assert.equal( + await io.call("https://pds.example/xrpc/a", { expect: "status" }), + 204, + ); + await io.close(); +}); + +test("a status body disposed under deadline cannot stall the caller", async () => { + let disposed = false; + const io = createIncomingPublicIo({ + timeoutMs: 40, + fetch: async () => { + const response = disposalOrderingResponse([]); + response.body.cancel = () => { + disposed = true; + return new Promise(() => {}); + }; + return response; + }, + }); + const started = Date.now(); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "status" }), + 1_000, + ); + assert.ok(outcome.error, "An undisposable body must reject, not hang"); + assert.ok(Date.now() - started < 1_000, "Rejection must stay bounded"); + assert.equal(disposed, true, "Disposal must still have been attempted"); + await io.close(); +}); + +test("cancelling a status call rejects within its deadline and disposes", async () => { + const disposed = []; + const controller = new AbortController(); + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => disposalOrderingResponse(disposed), + }); + const pending = io.call("https://pds.example/xrpc/a", { + expect: "status", + signal: controller.signal, + }); + controller.abort(new Error("status cancelled")); + const outcome = await within(pending, 1_000); + assert.equal(outcome.error?.message, "status cancelled"); + await within(io.close(), 1_000); + assert.deepEqual( + disposed, + ["body"], + "A cancelled call must still release its body", + ); +}); + +test("deadline stays active across a stalled JSON body", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 40, + fetch: async () => hangingResponse(), + }); + const started = Date.now(); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "json" }), + 1_000, + ); + assert.ok(outcome.error, "A stalled JSON body must reject"); + assert.ok(Date.now() - started < 1_000, "Rejection must be prompt"); + await io.close(); +}); + +test("deadline stays active across a stalled CAR body", async () => { + let cancelled = false; + const io = createIncomingPublicIo({ + timeoutMs: 40, + fetch: async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + }, + cancel() { + cancelled = true; + }, + }), + { + status: 200, + headers: { "content-type": "application/vnd.ipld.car" }, + }, + ), + }); + const started = Date.now(); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "car" }), + 1_000, + ); + assert.ok(outcome.error, "A stalled CAR body must reject"); + assert.ok(Date.now() - started < 1_000, "Rejection must be prompt"); + assert.equal(cancelled, true, "The stalled stream must be cancelled"); + await io.close(); +}); + +test("phase deadline releases a CAR reader that never completes", async () => { + let cancelled = false; + const io = createIncomingPublicIo({ + timeoutMs: 40, + fetch: async (_url, init) => { + init.signal.addEventListener("abort", () => { + cancelled = true; + }); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1])); + }, + cancel() { + cancelled = true; + }, + }), + { + status: 200, + headers: { "content-type": "application/vnd.ipld.car" }, + }, + ); + }, + }); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "car" }), + 1_000, + ); + assert.ok(outcome.error); + assert.equal(cancelled, true, "The abort path must release the reader"); + await io.close(); +}); + +test("tracking covers response consumption until settlement", async () => { + let settles; + const gate = new Promise((resolve) => { + settles = resolve; + }); + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => { + await gate; + return jsonResponse({ ok: true }); + }, + }); + const pending = io.call("https://pds.example/xrpc/a", { expect: "json" }); + assert.equal(io.inflight.size, 1, "In-flight work is tracked"); + settles(); + assert.deepEqual(await pending, { ok: true }); + await Promise.resolve(); + assert.equal(io.inflight.size, 0, "Settled work is untracked"); + await io.close(); +}); + +test("teardown aborts in-flight work before bounded settlement", async () => { + let abortedAtTeardown = false; + let transportSignal; + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: (_url, init) => + new Promise((_resolve, reject) => { + transportSignal = init.signal; + transportSignal.addEventListener("abort", () => + reject(init.signal.reason), + ); + }), + }); + const pending = io.call("https://pds.example/xrpc/a", { + expect: "status", + }); + const observed = []; + const closing = io.close({ settleTimeoutMs: 1_000 }).then(() => { + abortedAtTeardown = transportSignal.aborted; + observed.push("settled"); + }); + assert.equal( + transportSignal.aborted, + true, + "Teardown must abort the transport before it waits for settlement", + ); + await assert.rejects(pending); + await closing; + assert.equal(abortedAtTeardown, true); + assert.deepEqual(observed, ["settled"]); + assert.equal(io.inflight.size, 0); +}); + +test("teardown waits on request settlement, not only on abort", async () => { + let releaseTransport; + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: () => + new Promise((_resolve, reject) => { + releaseTransport = () => reject(new Error("still draining")); + }), + }); + const pending = io.call("https://pds.example/xrpc/a", { + expect: "status", + }); + const order = []; + const closing = io.close({ settleTimeoutMs: 1_000 }).then(() => { + order.push("close resolved"); + }); + for (let turn = 0; turn < 5; turn += 1) await Promise.resolve(); + order.push("close still waiting"); + releaseTransport(); + await assert.rejects(pending); + await closing; + assert.deepEqual(order, ["close still waiting", "close resolved"]); +}); + +test("teardown settles within its budget when the transport ignores abort", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: () => new Promise(() => {}), + }); + const ignored = io.call("https://pds.example/xrpc/a", { + expect: "status", + }); + const started = Date.now(); + const outcome = await within(io.close({ settleTimeoutMs: 60 }), 1_000); + assert.equal(outcome.value, undefined, "Bounded teardown completes"); + assert.ok(Date.now() - started < 1_000, "Cleanup wait is bounded"); + await assert.rejects(ignored, { name: "AbortError" }); +}); + +test("an external test abort settles and cleans up the call", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: () => new Promise(() => {}), + }); + const controller = new AbortController(); + const pending = io.call("https://pds.example/xrpc/a", { + expect: "json", + signal: controller.signal, + }); + controller.abort(new Error("test aborted")); + const outcome = await within(pending, 1_000); + assert.equal(outcome.error?.message, "test aborted"); + await io.close(); +}); + +test("a pre-aborted external signal never dispatches", async () => { + let calls = 0; + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => { + calls += 1; + return jsonResponse({}); + }, + }); + const controller = new AbortController(); + controller.abort(); + await assert.rejects( + io.call("https://pds.example/xrpc/a", { + signal: controller.signal, + }), + ); + assert.equal(calls, 0, "Aborted work must not reach the transport"); + await io.close(); +}); + +test("a streamed request body is dispatched with its duplex requirement met", async () => { + const seen = []; + let reads = 0; + const body = new ReadableStream({ + pull(controller) { + reads += 1; + controller.enqueue(new Uint8Array([reads])); + if (reads >= 2) controller.close(); + }, + }); + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async (url, init) => { + const request = new Request(url, init); + seen.push({ method: request.method, duplex: init.duplex }); + const received = new Uint8Array(await request.arrayBuffer()); + return jsonResponse({ received: received.byteLength }); + }, + }); + const result = await io.call("https://pds.example/xrpc/a", { + bytes: body, + expect: "json", + }); + assert.equal(result.received, 2); + assert.equal(seen.length, 1); + assert.equal(seen[0].method, "POST"); + assert.equal(seen[0].duplex, "half", "A stream body needs duplex half"); + await io.close(); +}); + +test("a streamed request body keeps its own content type", async () => { + const seen = []; + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async (url, init) => { + const request = new Request(url, init); + seen.push({ + duplex: init.duplex, + contentType: init.headers["content-type"], + length: (await request.arrayBuffer()).byteLength, + }); + return jsonResponse({ ok: true }); + }, + }); + await io.call("https://pds.example/xrpc/a", { + bytes: new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + controller.close(); + }, + }), + headers: { "content-type": "application/vnd.ipld.car" }, + }); + assert.deepEqual(seen, [ + { duplex: "half", contentType: "application/vnd.ipld.car", length: 3 }, + ]); + await io.close(); +}); + +test("a non-stream body is dispatched without duplex", async () => { + const seen = []; + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async (_url, init) => { + seen.push({ duplex: init.duplex, body: init.body }); + return jsonResponse({ ok: true }); + }, + }); + await io.call("https://pds.example/xrpc/a", { bytes: new Uint8Array([1]) }); + assert.deepEqual(seen, [{ duplex: undefined, body: new Uint8Array([1]) }]); + await io.close(); +}); + +test("a prefix outside readiness failures never starts its stream", async () => { + const cases = [ + [ + "non-2xx", + { status: 500, ok: false }, + (error) => error instanceof ResponseStatusError && error.status === 500, + ], + [ + "malformed declared length", + { headers: new Headers({ "content-length": "not-a-number" }) }, + (error) => + error instanceof ResponseBodyError && + error.message === "Malformed response content length", + ], + [ + "declared over the limit", + { headers: new Headers({ "content-length": "9" }) }, + (error) => + error instanceof ResponseBodyError && + error.message === "Response exceeds CAR limit", + ], + [ + "omitted body", + { omitBody: true }, + (error) => + error instanceof ResponseBodyError && + error.message === "Response omitted body", + ], + ]; + for (const [name, shape, matches] of cases) { + const { response, state } = guardedStreamResponse({ + ready: { cancel: true }, + }); + const candidate = { + ...response, + ...shape, + headers: + shape.headers ?? + new Headers({ "content-type": "application/vnd.ipld.car" }), + }; + if (shape.omitBody) candidate.body = undefined; + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + carLimitBytes: 8, + fetch: async () => candidate, + }); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "car" }), + 1_000, + ); + assert.ok(matches(outcome.error), `${name}: expected a typed rejection`); + assert.equal( + state.cancelled, + name === "omitted body" ? false : true, + `${name}: an unread body must be released without starting it`, + ); + await io.close(); + } +}); + +test("a network TypeError is reported, not absorbed", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => { + throw new TypeError("fetch failed"); + }, + }); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "car" }), + 1_000, + ); + assert.ok(outcome.error instanceof TypeError); + assert.equal(outcome.error.message, "fetch failed"); + await io.close(); +}); + +test("a non-2xx CAR rejection is not a body-limit diagnosis", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => + new Response(new Uint8Array([1]), { + status: 404, + headers: { "content-type": "application/vnd.ipld.car" }, + }), + }); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "car" }), + 1_000, + ); + assert.ok(outcome.error instanceof ResponseStatusError); + assert.equal(outcome.error.status, 404); + await io.close(); +}); + +test("a phase timeout reports its own reason, not a body error", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 30, + fetch: async () => { + const { response } = guardedStreamResponse(); + response.body.getReader = () => ({ + read: () => new Promise(() => {}), + cancel: () => Promise.resolve(), + }); + return response; + }, + }); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "car" }), + 1_000, + ); + assert.ok(outcome.error instanceof Error); + assert.equal( + outcome.error instanceof ResponseBodyError, + false, + "A timeout must not masquerade as a body diagnosis", + ); + assert.equal(outcome.error instanceof ResponseStatusError, false); + await io.close(); +}); + +test("oversized declared CAR rejects before reading the body", async () => { + let reads = 0; + let cancelled = false; + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + carLimitBytes: 8, + fetch: async () => ({ + status: 200, + ok: true, + headers: new Headers({ + "content-type": "application/vnd.ipld.car", + "content-length": "16", + }), + body: { + getReader: () => ({ + read: () => { + reads += 1; + return Promise.resolve({ done: true, value: undefined }); + }, + cancel: () => { + cancelled = true; + return Promise.resolve(); + }, + }), + cancel: () => { + cancelled = true; + return Promise.resolve(); + }, + }, + }), + }); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "car" }), + 1_000, + ); + assert.equal(outcome.error?.message, "Response exceeds CAR limit"); + assert.equal(reads, 0, "No body chunk may be read to decide the bound"); + assert.equal(cancelled, true, "The unread body is released"); + await io.close(); +}); + +test("undeclared oversized CAR rejects without retaining excess", async () => { + let reads = 0; + let cancelled = false; + const chunk = new Uint8Array(6); + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + carLimitBytes: 8, + fetch: async () => ({ + status: 200, + ok: true, + headers: new Headers({ "content-type": "application/vnd.ipld.car" }), + body: { + getReader: () => ({ + read: () => { + reads += 1; + return Promise.resolve({ done: false, value: chunk }); + }, + cancel: () => { + cancelled = true; + return Promise.resolve(); + }, + }), + cancel: () => Promise.resolve(), + }, + }), + }); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "car" }), + 1_000, + ); + assert.equal(outcome.error?.message, "Response exceeds CAR limit"); + assert.equal(reads, 2, "The overrun chunk is detected, not aggregated"); + assert.equal(cancelled, true, "The overrun reader is released"); + await io.close(); +}); + +test("a CAR exactly at the limit succeeds", async () => { + const chunk = new Uint8Array(6); + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + carLimitBytes: 8, + fetch: async () => carResponse([chunk, new Uint8Array(2)]), + }); + const bytes = await io.call("https://pds.example/xrpc/a", { + expect: "car", + }); + assert.equal(bytes.byteLength, 8); + await io.close(); +}); + +test("a limit of one byte accepts exactly one byte", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + carLimitBytes: 1, + fetch: async () => carResponse([new Uint8Array(1)]), + }); + const bytes = await io.call("https://pds.example/xrpc/a", { + expect: "car", + }); + assert.deepEqual([...bytes], [0]); + await io.close(); +}); + +test("malformed declared CAR length rejects", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + carLimitBytes: 8, + fetch: async () => + carResponse([new Uint8Array(1)], { declared: "not-a-number" }), + }); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "car" }), + 1_000, + ); + assert.equal(outcome.error?.message, "Malformed response content length"); + await io.close(); +}); + +test("request bodies and headers are composed without escaping", async () => { + const seen = []; + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async (url, init) => { + seen.push({ url, init }); + return jsonResponse({ ok: true }); + }, + }); + await io.call("https://pds.example/xrpc/a", { + verb: "POST", + token: "token-value", + json: { did: "did:plc:example" }, + }); + await io.call("https://pds.example/xrpc/b", { + verb: "POST", + token: "token-value", + bytes: new Uint8Array([1]), + headers: { "content-type": "image/png" }, + expect: "text", + }); + assert.equal(seen[0].init.method, "POST"); + assert.equal(seen[0].init.headers.authorization, "Bearer token-value"); + assert.equal(seen[0].init.headers["content-type"], "application/json"); + assert.equal(seen[0].init.body, JSON.stringify({ did: "did:plc:example" })); + assert.equal(seen[1].init.headers["content-type"], "image/png"); + assert.deepEqual([...seen[1].init.body], [1]); + await io.close(); +}); + +test("non-JSON, failed and malformed bodies reject", async () => { + const cases = [ + [ + "unexpected content type", + () => new Response("text", { status: 200 }), + "Unexpected response content type", + ], + [ + "unexpected status", + () => jsonResponse({ error: "InvalidRequest" }, 400), + "Unexpected response status", + ], + [ + "malformed body", + () => + new Response(new Uint8Array([0xff]), { + status: 200, + headers: { "content-type": "application/json" }, + }), + "Malformed response body", + ], + ]; + for (const [name, respond, message] of cases) { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => respond(), + }); + const outcome = await within( + io.call("https://pds.example/xrpc/a", { expect: "json" }), + 1_000, + ); + assert.equal(outcome.error?.message, message, name); + await io.close(); + } +}); + +test("a failed response exposes its machine error code", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => + jsonResponse( + { error: "AccountDeactivated", message: "deactivated" }, + 401, + ), + }); + const outcome = await within(io.call("https://pds.example/xrpc/a"), 1_000); + assert.equal(outcome.error.status, 401); + assert.equal(outcome.error.body.error, "AccountDeactivated"); + await io.close(); +}); + +test("status consumption reports the code synchronously", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => new Response(null, { status: 204 }), + }); + assert.equal( + await io.call("https://pds.example/xrpc/a", { expect: "status" }), + 204, + ); + await io.close(); +}); + +test("an empty successful body decodes to no value", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => new Response(null, { status: 200 }), + }); + assert.equal(await io.call("https://pds.example/xrpc/a"), undefined); + assert.equal( + await io.call("https://pds.example/xrpc/a", { expect: "text" }), + "", + ); + await io.close(); +}); + +test("a closed client refuses further work", async () => { + const io = createIncomingPublicIo({ + timeoutMs: 5_000, + fetch: async () => jsonResponse({}), + }); + await io.close(); + await assert.rejects(io.call("https://pds.example/xrpc/a"), { + message: "Incoming public I/O is closed", + }); +}); + +test("invalid construction is refused", () => { + assert.throws(() => createIncomingPublicIo({ fetch: 1 }), TypeError); + assert.throws(() => createIncomingPublicIo({ timeoutMs: 0 }), RangeError); + assert.throws(() => createIncomingPublicIo({ carLimitBytes: 0 }), RangeError); +}); + +const createHeldStream = (body) => { + let release; + const gate = new Promise((resolve) => { + release = resolve; + }); + let started; + const sent = new Promise((resolve) => { + started = resolve; + }); + const stream = new ReadableStream({ + async start(controller) { + controller.enqueue(body.subarray(0, body.length - 1)); + started(); + await gate; + try { + controller.enqueue(body.subarray(body.length - 1)); + controller.close(); + } catch { + release(); + } + }, + cancel() { + release(); + }, + }); + return { stream, sent, release }; +}; + +const heldSeam = (fetch, options = {}) => { + const io = createIncomingPublicIo({ timeoutMs: 5_000, fetch }); + const heldStreams = []; + const run = (url, runOptions = {}) => { + const held = runHeldStreamedCall(io, url, { + ...options, + ...runOptions, + }); + heldStreams.push(held); + return held; + }; + return { + run, + close: async () => { + for (const held of heldStreams) await held.retireNow(); + await io.close(); + }, + }; +}; + +test("a released held stream reports success only with its empty text body", async () => { + const body = new Uint8Array([1, 2, 3, 4]); + const { stream, sent, release } = createHeldStream(body); + const { run, close } = heldSeam( + async () => new Response(null, { status: 200 }), + { deadlineMs: 5_000 }, + ); + const held = run("https://pds.example/xrpc/import", { + bytes: stream, + expect: "text", + }); + await sent; + release(); + const settled = await within(held.tagged, 1_000); + assert.equal(settled.value.outcome, "success"); + assert.equal( + await held.consume({ expect: "text" }), + "", + "Only the empty text body attests this stream's success", + ); + await close(); +}); + +test("a consumed held stream rethrows the original HTTP failure", async () => { + const body = new Uint8Array([1, 2, 3, 4]); + const { stream, sent } = createHeldStream(body); + const { run, close } = heldSeam( + async () => + new Response(new TextEncoder().encode('{"error":"InvalidRequest"}'), { + status: 400, + headers: { "content-type": "application/json" }, + }), + { deadlineMs: 5_000 }, + ); + const held = run("https://pds.example/xrpc/import", { + bytes: stream, + expect: "text", + }); + await sent; + const settled = await within(held.tagged, 1_000); + assert.equal(settled.value.outcome, "failure"); + await assert.rejects(held.consume({ expect: "text" }), (error) => { + assert.ok(error instanceof ResponseStatusError); + assert.equal(error.status, 400); + return true; + }); + await close(); +}); + +test("a consumed held stream rethrows the original transport failure", async () => { + const body = new Uint8Array([1, 2, 3, 4]); + const { stream, sent } = createHeldStream(body); + const failure = new TypeError("fetch failed"); + const { run, close } = heldSeam( + async () => { + throw failure; + }, + { deadlineMs: 5_000 }, + ); + const held = run("https://pds.example/xrpc/import", { + bytes: stream, + expect: "text", + }); + await sent; + const settled = await within(held.tagged, 1_000); + assert.equal(settled.value.outcome, "failure"); + assert.equal( + settled.value.error, + failure, + "The exact transport failure instance must surface, not a wrapper", + ); + await assert.rejects( + held.consume({ expect: "text" }), + (error) => error === failure, + ); + await close(); +}); + +test("a held stream still delivers its deadline abort as the original failure", async () => { + const body = new Uint8Array([1, 2, 3, 4]); + const { stream } = createHeldStream(body); + const { run, close } = heldSeam( + (_url, init) => + new Promise((_resolve, reject) => { + init.signal.addEventListener("abort", () => reject(init.signal.reason)); + }), + { deadlineMs: 30 }, + ); + const held = run("https://pds.example/xrpc/import", { + bytes: stream, + expect: "text", + }); + const settled = await within(held.tagged, 1_000); + assert.equal(settled.value.outcome, "failure"); + assert.equal( + settled.value.error?.message, + "Held stream deadline exhausted", + "A deadline abort must reject with its own reason even if the stream never completes", + ); + await close(); +}); + +test("a held stream failure tags its settlement without an unhandled rejection", async () => { + const body = new Uint8Array([1, 2, 3, 4]); + const { stream } = createHeldStream(body); + const unhandled = []; + const onUnhandled = (error) => unhandled.push(error); + process.on("unhandledRejection", onUnhandled); + const { run, close } = heldSeam( + async () => new Response(null, { status: 500 }), + { deadlineMs: 30 }, + ); + const held = run("https://pds.example/xrpc/import", { + bytes: stream, + expect: "text", + }); + const settled = await within(held.tagged, 1_000); + assert.equal( + settled.value.outcome, + "failure", + "The settlement is attached early and tagged as a failure", + ); + await new Promise((resolve) => setTimeout(resolve, 20)); + process.off("unhandledRejection", onUnhandled); + assert.deepEqual( + unhandled, + [], + "An unread held failure must not surface as an unhandled rejection", + ); + await close(); +}); + +test("an inventory check is retained beside, never instead of, stream success", async () => { + const body = new Uint8Array([1, 2, 3, 4]); + const { stream, release } = createHeldStream(body); + const { run, close } = heldSeam( + async () => new Response(null, { status: 200 }), + { deadlineMs: 5_000 }, + ); + const held = run("https://pds.example/xrpc/import", { + bytes: stream, + expect: "text", + }); + release(); + assert.equal(await held.consume({ expect: "text" }), ""); + assert.equal( + (await held.tagged).outcome, + "success", + "Inventory equality is a separate observation; only the empty body attests this stream's success", + ); + await close(); +}); diff --git a/tests/support/helpers/stock-plc-mailpit.mjs b/tests/support/helpers/stock-plc-mailpit.mjs new file mode 100644 index 00000000..1576baf8 --- /dev/null +++ b/tests/support/helpers/stock-plc-mailpit.mjs @@ -0,0 +1,133 @@ +import { setTimeout as delay } from "node:timers/promises"; +import { createIncomingPublicIo } from "./incoming-public-io.mjs"; + +const subject = "PLC Update Operation Requested"; +const phrase = + "We received a request to update your PLC identity. Your confirmation code is:"; + +export function parseStockPlcMessage(message) { + if (message?.Subject !== subject) throw new Error("Not a stock PLC message"); + const tokens = []; + if (message.Text) { + const text = message.Text.replace(/\s+/g, " "); + const matches = [ + ...text.matchAll( + /We received a request to update your PLC identity\. Your confirmation code is:\s*([A-Z2-7]{5}-[A-Z2-7]{5})\b/g, + ), + ]; + if (matches.length !== 1) throw new Error("Malformed stock PLC text"); + tokens.push(matches[0][1]); + } + if (message.HTML) { + const html = message.HTML.replace(/\s+/g, " "); + const matches = [ + ...html.matchAll( + /We received a request to update your PLC identity\. Your confirmation code is:<\/p>\s*]*>\s*([A-Z2-7]{5}-[A-Z2-7]{5})\s*<\/code>/g, + ), + ]; + if (matches.length !== 1 || !html.includes(phrase)) + throw new Error("Malformed stock PLC HTML"); + tokens.push(matches[0][1]); + } + if (!tokens.length || new Set(tokens).size !== 1) + throw new Error("Ambiguous stock PLC message"); + return tokens[0]; +} + +export async function waitForStockPlcMessage({ + recipient, + since, + excludeIds, + baseUrl, + timeoutMs = 20_000, + fetch: fetchImpl = globalThis.fetch, + signal, + intervalMs = 250, +}) { + if (typeof recipient !== "string" || !/^[^\s@]+@[^\s@]+$/.test(recipient)) + throw new TypeError("Invalid recipient"); + if ( + !Number.isFinite(since) || + !Array.isArray(excludeIds) || + excludeIds.some((id) => typeof id !== "string" || !id) + ) + throw new TypeError("Invalid message boundary"); + if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 1 || timeoutMs > 60_000) + throw new RangeError("Invalid timeout"); + if (!Number.isSafeInteger(intervalMs) || intervalMs < 1 || intervalMs > 1000) + throw new RangeError("Invalid polling interval"); + if (typeof fetchImpl !== "function") throw new TypeError("Invalid fetch"); + if (signal !== undefined && !(signal instanceof AbortSignal)) + throw new TypeError("Invalid poll signal"); + signal?.throwIfAborted(); + const origin = new URL(baseUrl); + if (origin.protocol !== "https:" || origin.username || origin.password) + throw new Error("Mailpit requires HTTPS without credentials"); + const endpoint = new URL("/api/v1/search", origin); + endpoint.searchParams.set("query", `to:${recipient}`); + endpoint.searchParams.set("limit", "50"); + const controller = new AbortController(); + const timer = setTimeout( + () => + controller.abort(new Error("Timed out waiting for stock PLC message")), + timeoutMs, + ); + const pollSignal = signal + ? AbortSignal.any([signal, controller.signal]) + : controller.signal; + const io = createIncomingPublicIo({ + fetch: fetchImpl, + signal: pollSignal, + timeoutMs: Math.min(5000, timeoutMs), + carLimitBytes: 1024 * 1024, + }); + const hasRecipient = (message) => + Array.isArray(message.To) && + message.To.some( + (item) => item.Address?.toLowerCase() === recipient.toLowerCase(), + ); + // Reuse managed cancellation/settlement and bounded streaming consumption. + // Mailpit timestamps in search are server Created times; message Date is not. + const read = async (url) => { + const bytes = await io.call(url.href, { + verb: "GET", + expect: "car", + redirect: "error", + }); + return JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(bytes)); + }; + try { + for (;;) { + pollSignal.throwIfAborted(); + const result = await read(endpoint); + if (!Array.isArray(result.messages)) + throw new Error("Invalid Mailpit response"); + const candidates = result.messages.filter( + (message) => + typeof message.ID === "string" && + !excludeIds.includes(message.ID) && + hasRecipient(message) && + Date.parse(message.Created) >= since && + message.Subject === subject, + ); + if (candidates.length > 1) + throw new Error("Ambiguous stock PLC messages"); + if (candidates.length === 1) { + const message = candidates[0]; + const detail = await read( + new URL(`/api/v1/message/${encodeURIComponent(message.ID)}`, origin), + ); + if (!hasRecipient(detail) || detail.ID !== message.ID) + throw new Error("Invalid Mailpit message boundary"); + return { token: parseStockPlcMessage(detail), messageId: message.ID }; + } + await delay(intervalMs, undefined, { signal: pollSignal }); + } + } catch (error) { + if (pollSignal.aborted) throw pollSignal.reason; + throw error; + } finally { + clearTimeout(timer); + await io.close(); + } +} diff --git a/tests/support/helpers/stock-plc-mailpit.test.mjs b/tests/support/helpers/stock-plc-mailpit.test.mjs new file mode 100644 index 00000000..dd5d6721 --- /dev/null +++ b/tests/support/helpers/stock-plc-mailpit.test.mjs @@ -0,0 +1,322 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { + parseStockPlcMessage, + waitForStockPlcMessage, +} from "./stock-plc-mailpit.mjs"; + +const phrase = + "We received a request to update your PLC identity. Your confirmation code is:"; +const message = { + Subject: "PLC Update Operation Requested", + Text: `${phrase}\nABCDE-FG234`, + HTML: `

${phrase}

ABCDE-FG234`, +}; + +test("parses stock PLC text and HTML and rejects mismatched content", () => { + assert.equal(parseStockPlcMessage(message), "ABCDE-FG234"); + assert.equal(parseStockPlcMessage({ ...message, HTML: "" }), "ABCDE-FG234"); + assert.equal(parseStockPlcMessage({ ...message, Text: "" }), "ABCDE-FG234"); + for (const bad of [ + { ...message, Subject: "Other" }, + { ...message, Text: "Use 12345678 to login", HTML: "" }, + { ...message, Text: `${message.Text} ${message.Text}` }, + { ...message, HTML: message.HTML.replace("ABCDE", "ZZZZZ") }, + { ...message, Text: "", HTML: "" }, + ]) + assert.throws(() => parseStockPlcMessage(bad)); +}); + +test("poll binds recipient, time and excluded IDs", async () => { + let fetchImpl; + const since = Date.now() - 1000; + const candidate = { + Subject: message.Subject, + ID: "new", + To: [{ Address: "user@example.com" }], + Created: new Date().toISOString(), + }; + const detail = { + ...message, + ID: candidate.ID, + To: candidate.To, + Date: new Date(since - 60_000).toISOString(), + }; + { + fetchImpl = async (url) => + Response.json( + String(url).includes("/search") + ? { + messages: [ + { ...candidate, ID: "excluded" }, + { + ...candidate, + ID: "old", + Created: new Date(since - 1).toISOString(), + }, + { + ...candidate, + ID: "wrong", + To: [{ Address: "other@example.com" }], + }, + candidate, + ], + } + : detail, + ); + const result = await waitForStockPlcMessage({ + recipient: "user@example.com", + since, + excludeIds: ["excluded"], + baseUrl: "https://mail.example", + fetch: (...args) => fetchImpl(...args), + timeoutMs: 1000, + }); + assert.equal(result.messageId, "new"); + const options = { + recipient: "user@example.com", + since, + excludeIds: [], + baseUrl: "https://mail.example", + fetch: (...args) => fetchImpl(...args), + timeoutMs: 50, + }; + for (const created of [ + undefined, + "invalid", + new Date(since - 1).toISOString(), + ]) { + fetchImpl = async () => + Response.json({ messages: [{ ...candidate, Created: created }] }); + await assert.rejects(waitForStockPlcMessage(options), /Timed out/); + } + for (const badDetail of [ + { ...detail, ID: "other" }, + { ...detail, To: [{ Address: "other@example.com" }] }, + { ...detail, Subject: "Other" }, + ]) { + fetchImpl = async (url) => + Response.json( + String(url).includes("/search") + ? { messages: [candidate] } + : badDetail, + ); + await assert.rejects(waitForStockPlcMessage(options)); + } + fetchImpl = async () => + Response.json({ + messages: [candidate, { ...candidate, ID: "another" }], + }); + await assert.rejects( + waitForStockPlcMessage({ + recipient: "user@example.com", + since, + excludeIds: [], + baseUrl: "https://mail.example", + fetch: (...args) => fetchImpl(...args), + timeoutMs: 1000, + }), + /Ambiguous/, + ); + await assert.rejects( + waitForStockPlcMessage({ + recipient: "user@example.com", + since, + excludeIds: [], + baseUrl: "http://mail.example", + fetch: (...args) => fetchImpl(...args), + timeoutMs: 1000, + }), + /HTTPS/, + ); + } +}); + +const pollOptions = { + recipient: "user@example.com", + since: Date.now() - 1000, + excludeIds: [], + baseUrl: "https://mail.example", + timeoutMs: 1000, + intervalMs: 1, +}; + +test("poll rejects redirects at search and detail without following HTTP downgrades", async () => { + const candidate = { + ...message, + ID: "matching", + To: [{ Address: pollOptions.recipient }], + Created: new Date().toISOString(), + }; + for (const redirectAt of ["search", "detail"]) { + const requests = []; + const failure = new TypeError("unexpected redirect"); + await assert.rejects( + waitForStockPlcMessage({ + ...pollOptions, + fetch: async (url, init) => { + const request = new Request(url, init); + requests.push(request); + const phase = request.url.includes("/search") ? "search" : "detail"; + if (phase === redirectAt) { + const response = Response.redirect("http://mail.example/downgrade"); + if (request.redirect === "error") throw failure; + requests.push(new Request(response.headers.get("location"))); + return Response.json( + phase === "search" ? { messages: [candidate] } : candidate, + ); + } + return Response.json({ messages: [candidate] }); + }, + }), + (error) => error === failure, + ); + assert.equal(requests.length, redirectAt === "search" ? 1 : 2); + for (const request of requests) { + assert.equal(new URL(request.url).protocol, "https:"); + assert.equal(request.redirect, "error"); + } + } +}); + +test("poll accepts direct HTTPS search and detail with redirects disabled", async () => { + const candidate = { + ...message, + ID: "matching", + To: [{ Address: pollOptions.recipient }], + Created: new Date().toISOString(), + }; + const requests = []; + const result = await waitForStockPlcMessage({ + ...pollOptions, + fetch: async (url, init) => { + const request = new Request(url, init); + requests.push(request); + return Response.json( + request.url.includes("/search") ? { messages: [candidate] } : candidate, + ); + }, + }); + assert.deepEqual(result, { token: "ABCDE-FG234", messageId: "matching" }); + assert.equal(requests.length, 2); + for (const request of requests) { + assert.equal(new URL(request.url).protocol, "https:"); + assert.equal(request.method, "GET"); + assert.equal(request.redirect, "error"); + } +}); + +test("poll cancellation reaches the injected transport and settles", async () => { + const controller = new AbortController(); + let requestSignal; + const pending = waitForStockPlcMessage({ + ...pollOptions, + signal: controller.signal, + fetch: async (_url, { signal }) => { + requestSignal = signal; + return new Promise((_, reject) => { + signal.addEventListener("abort", () => reject(signal.reason), { + once: true, + }); + }); + }, + }); + const reason = new Error("poll cancelled"); + controller.abort(reason); + await assert.rejects(pending, (error) => error === reason); + assert.equal(requestSignal.aborted, true); +}); + +test("poll deadline bounds held response bodies and cancels the reader", async () => { + let cancelled = false; + await assert.rejects( + waitForStockPlcMessage({ + ...pollOptions, + timeoutMs: 30, + fetch: async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("{")); + }, + cancel() { + cancelled = true; + }, + }), + ), + }), + /Timed out/, + ); + assert.equal(cancelled, true); +}); + +test("poll rejects malformed matching content, invalid search and oversized bodies", async () => { + const candidate = { + ...message, + ID: "matching", + To: [{ Address: pollOptions.recipient }], + Created: new Date().toISOString(), + }; + for (const detail of [ + { ...candidate, Text: "unrelated code", HTML: "" }, + { ...candidate, Text: null, HTML: 42 }, + { ...candidate, ID: "changed" }, + { ...candidate, To: [] }, + { ...candidate, Subject: "login" }, + ]) { + await assert.rejects( + waitForStockPlcMessage({ + ...pollOptions, + fetch: async (url) => + Response.json( + String(url).includes("/search") + ? { messages: [candidate] } + : detail, + ), + }), + ); + } + for (const value of [null, {}, { messages: null }]) { + await assert.rejects( + waitForStockPlcMessage({ + ...pollOptions, + fetch: async () => Response.json(value), + }), + ); + } + await assert.rejects( + waitForStockPlcMessage({ + ...pollOptions, + fetch: async () => new Response("x".repeat(1024 * 1024 + 1)), + }), + /limit/, + ); + await assert.rejects( + waitForStockPlcMessage({ + ...pollOptions, + fetch: async () => new Response("{broken"), + }), + SyntaxError, + ); +}); + +test("poll remains sequential and cancellation interrupts the retry delay", async () => { + const controller = new AbortController(); + let active = 0; + let calls = 0; + const pending = waitForStockPlcMessage({ + ...pollOptions, + intervalMs: 1000, + signal: controller.signal, + fetch: async () => { + calls++; + assert.equal(++active, 1); + await new Promise((resolve) => setTimeout(resolve, 5)); + active--; + setTimeout(() => controller.abort(new Error("stop delay")), 5); + return Response.json({ messages: [] }); + }, + }); + await assert.rejects(pending, /stop delay/); + assert.equal(calls, 1); +});