8 Commits
Author SHA1 Message Date
glm-5.3-flash a99cc67e76 chore(release): 0.10.1 — the adoption-window ordering fix
Changelog entry for review 013's fix (4dbd4bc); version bump 0.10.0 →
0.10.1 (patch — private fix surface, no API or wire-format change).

Verification: semver-checks 196 pass (no semver update required);
cargo test 700 + 724 (--all-features) pass; clippy --all-targets
--all-features -D warnings, fmt --check, wasm32-unknown-unknown check +
clippy, fuzz corpus replay (5/5), and cargo publish --dry-run clean.
Downstream gate: alkgit's flaking row re-verified green under
multi-process load against 4dbd4bc.
2026-10-22 05:28:12 +00:00
glm-5.3-flash 4dbd4bc583 fix(channels): close the adopt-window ordering race (review 013) — atomic install+park-drain
A chunk routed by the demux inside the adopt_channel install→drain
window was delivered ahead of earlier-parked early arrivals: the mpsc
send order was inverted relative to wire-arrival order, misordering the
channel's byte stream (C-01, critical — downstream alkgit observed two
desync shapes under multi-process load). A related loss variant is also
closed: a chunk whose map miss parked it after the drain had removed
the queue sat parked forever.

The window closes by serialization, not by widening locks across
awaits — the resolution is fully synchronous:

- route_payload holds the channels-map lock across the miss → park
  decision (park_early_arrival runs nested; lock order channels →
  early_arrivals): a park can never straddle the install, and the
  installer that caused the miss always swaps the queue out.
- install_locked (new, shared by open_channel / adopt_channel /
  install_channel_zero) swaps the parked queue, inserts the state, and
  try_sends the parked queue into the fresh, empty receiver before
  releasing the channels lock — a direct route can only begin after
  release, so parked chunks always precede it in send order. No await
  runs inside the section: the effective park cap is now
  min(EARLY_ARRIVAL_CAP, buffer_cap), bounding the parked queue by the
  receiver's capacity so the drain never blocks. drain_early_arrivals
  remains as the defensive full-buffer fallback and is now called by
  all three install paths.

Regression tests: adopt_window_does_not_reorder_chunks (200 iterations
on a 2-worker runtime — a racer routing C the moment the install is
visible, receiver byte stream asserted byte-identical to the producer's;
catches the old window under genuine preemption, the alkgit repro
shape), adopt_drains_parked_and_raced_chunks_in_fifo, and
park_cap_bounded_by_receiver_buffer. rt-multi-thread added as a
dev-dependency for the runtime test.

Semver: patch (0.10.1) — private fix surface; no API, wire-format, or
documented-cap behavior change; behavior only strengthens.

Verification: cargo test 700/724 (--all-features) passed; clippy
--all-targets -D warnings clean; fmt --check clean; cargo doc --no-deps
generated; fuzz corpus replay passed; wasm32-unknown-unknown check
passed.
2026-10-22 04:47:42 +00:00
glm-5.3-flash 56e0f135d7 docs(review): find the adoption-window ordering race — routed from alkgit's wave-2 gate
Review 013 [critical]: a chunk routed in the adopt_channel
install->drain window overtakes earlier-parked early arrivals,
misordering the reassembled byte stream (and, combined with the
EARLY_ARRIVAL_CAP overflow branch, can drop chunks) on a freshly
adopted channel. Diagnosed downstream: alkgit's wave-2 gate caught its
channels open-op test flaking under load and traced both observed
desync shapes (mid-advertisement 'fetc' after 0 lines, '00ev' after 1)
to the window. Includes the fix-shape sketch (per-channel ordered
mailbox recommended), the chunk-granularity regression gate, and the
downstream re-verification requirement. The ordering half of review
006 E-04's adopt-race family; alkgit carries no workaround by design.
2026-10-21 13:51:09 +00:00
glm-5.3-flash 1a1e4eb7c0 chore(release): 0.10.0 — the gateway ownership seam
The review-012 batch rides this tag: the dispatch-spine ownership
provider threading (with_ownership_provider) and the loud default
(FORBIDDEN, pre-dispatch, for resource_type-gated ops on an unwired
spine) with the with_static_ownership_fallback escape hatch.

Verified: 698 tests (incl. stub guard), clippy --all-targets
--all-features -D warnings, fmt --check, cargo doc, wasm32
check+clippy, fuzz corpus replay 5/5, semver-checks (additive API diff;
two new builder methods only), publish --dry-run clean.
2026-10-17 14:28:57 +00:00
glm-5.3-flash 1fda0d8587 fix(gateway): review-012 regression-pass cleanup on the loud ownership default
- deduplicate the thrice-repeated refusal block into
  GatewayDispatch::ownership_wiring_refusal(&self, op)
- drop the dead unwrap_or("") branch in the refusal predicate
  (let-else: the internal-visibility exemption and the resource_type
  predicate are now separate lets)
- pin the predicate refinement review 012 ruled on: a list-case op
  (resource_type set, resource_id_path None) is refused on an unwired
  spine and served through owns_any when a provider is wired

Verification: cargo test (697), cargo test --all-features (721),
clippy all-targets both configs, fmt --check
2026-10-17 14:17:06 +00:00
glm-5.3-flash 9899b37a82 feat(gateway): the loud ownership-wiring default (review 012 ruling)
- GatewayDispatch now refuses (FORBIDDEN, pre-dispatch) any op whose
  spec gates on access_control.resource_type when no ownership
  provider is wired — invoke, invoke_streaming, and invoke_sink alike
- Predicate is resource_type, not resource_id_path as review 012
  first drafted: AccessControl::check consults the provider whenever
  the resource type is set, including the list/owns_any case; the
  refusal message is operator-directed and names both opt-ins
- Exemptions: resource_action-only ACLs (no provider is consulted on
  the wire either) and Internal-visibility ops (spec-404, no
  acknowledgment leak)
- with_static_ownership_fallback(): declared opt-out to the ADR-011
  static fallback for deployments with no runtime ownership model —
  the door contract becomes enforceable, not just documented
- Superseded the pre-ruling pin
  gateway_without_provider_admits_only_the_static_claim with refusal/
  opt-out/exemption tests (all three entry points covered)
- ADR-048 amendment records the ruling; review 012 status updated

Blast-radius survey: alkhttp is the only live GatewayDispatch
consumer (unpublished, three unwired sites); alkblobs the only live
resource_type op producer (serves native-wire, providers wired);
no published 0.9.0 consumer relies on the silent fallback for such ops.
Ships in the same unreleased batch as the review-012 amendment.

Verification: cargo test (697 pass), cargo test --features gateway
(720 pass), cargo clippy --all-features --all-targets -- -D warnings,
cargo fmt --check, fuzz corpus replay (5 pass)
2026-10-17 14:09:16 +00:00
glm-5.3-flash d0b8124064 feat(gateway): thread the ownership provider through the dispatch spine (review 012)
- GatewayDispatch carries ownership_provider: Option<Arc<dyn
  OwnershipProvider>> + with_ownership_provider (dispatcher-identical
  builder); the shared root-context builder threads it, covering
  invoke / invoke_streaming / invoke_sink. Default stays None (the
  ADR-011 static fallback) — additive and opt-in, no context shape
  change, no behavior change for existing consumers.
- Door contract documented in the ADR-048 amendment: doors serving ops
  whose authorization truth is a revocable grant table must wire the
  same provider the wire path consults (visible-surface =
  authorized-surface). Anchored downstream by alkgit's LFS POC.
- The loud default (refuse resource_id_path ops with no provider
  wired) is deferred, not decided — tracked in review 012, ruled on
  after the door blast-radius survey; it belongs in the same release
  batch.
- Release placement: committed unreleased — additive work rides the
  next breaking batch (0.10.0's class) instead of publishing 0.9.1,
  keeping the six-consumer downstream churn to one coordinated update.

Verification:
- cargo test --all-features (717 + stub_guard); fuzz/shared corpus
  replay (5); clippy --all-features --all-targets -D warnings; fmt;
  cargo doc --no-deps
- in-crate regression tests pin static-fallback behavior, live-table
  override, revocation live on the gateway surface
- POC-L tests/gateway_gap.rs re-run against published 0.9.0 — green
2026-10-17 08:28:44 +00:00
glm-5.3-flash 03a6689c99 docs(review): 012 — gateway ownership seam, consumer input from alkgit's LFS POC
Report-only: GatewayDispatch::invoke hardwires ownership: None, so
External namespace-ACL ops through the HTTP gateway fall back to the
static Identity.resources model — token-carried claims admit without
consulting revocable grant tables (POC-L tests/gateway_gap.rs).
Three resolution shapes recorded; disposition is session-owned.
2026-10-17 07:28:44 +00:00
10 changed files with 1502 additions and 57 deletions

No files matched your search

+125
View File
@@ -4,6 +4,130 @@ All notable changes to this crate are documented here. The format is
based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and
this crate adheres to [Semantic Versioning](https://semver.org/).
## [0.10.1] - 2026-10-22
The adoption-window ordering fix (review 013, routed from alkgit's
wave-2 gate): a chunk routed by the demux inside the
`adopt_channel` install→drain window was delivered **ahead of**
earlier-parked early arrivals, inverting the channel's byte-stream
order (observed downstream as two distinct pkt-line desync shapes in
alkgit's open-op integration test under multi-process load — failure
shape 1, the reader starting mid-advertisement at offset ~16; failure
shape 2, a chunk boundary two bytes into a hex-length header). A
related loss variant is also closed: a chunk whose map miss parked it
after the drain had removed the queue sat parked forever, invisible to
`dropped_unknown_chunks`. No API, wire-format (`EventEnvelope`, the
channels 8-byte chunk header), or documented-cap behavior (review 006
E-04) change — `ChannelState` and the fix surface are private; the
behavior only strengthens.
### Fixed
- **The adopt-window ordering race (review 013 C-01).** The install
and its park-drain are now one atomic section. `route_payload`'s
map-miss → early-arrival-park decision runs under the channels-map
lock (a park can never straddle the install — closing the orphaned
chunk loss variant), and the new private `install_locked` helper
(shared by `open_channel` / `adopt_channel` /
`install_channel_zero`) swaps the parked queue out, inserts the
state, and `try_send`s the parked queue into the fresh, empty
receiver **before releasing the lock** — a direct route (map hit)
can only begin after release, so parked chunks always precede it in
mpsc send order and the reassembled byte stream preserves
wire-arrival order through the adoption window. No await runs inside
the section: the effective park cap is
`min(EARLY_ARRIVAL_CAP, buffer_cap)`, bounding the parked queue by
the receiver's capacity so the non-blocking drain always fits
(smaller `buffer_cap` values tighten the park cap; the default
64/64 pairing is unchanged). `drain_early_arrivals` remains as the
defensive full-buffer fallback, now called by all three install
paths. Regression tests:
`adopt_window_does_not_reorder_chunks` (200 iterations on a real
2-worker runtime — a racer routing a chunk the moment the install
becomes visible; the receiver's byte stream must be byte-identical
to the producer's every time, the alkgit repro shape),
`adopt_drains_parked_and_raced_chunks_in_fifo`, and
`park_cap_bounded_by_receiver_buffer`. Downstream gate: alkgit's
flaking row re-verified green under multi-process load against this
fix before the 0.10.1 publish.
### Verified
- 700 tests pass (including the stub guard; 724 with `--all-features`);
`clippy --all-targets -- -D warnings`, `fmt --check`, `cargo doc`,
wasm32-unknown-unknown check, fuzz corpus replay
(`cargo test --manifest-path fuzz/shared/Cargo.toml`),
`cargo semver-checks`, and `cargo publish --dry-run` clean.
## [0.10.0] - 2026-10-17
The gateway ownership seam (review 012, downstream evidence from
alkgit's LFS POC — POC-L): the gateway dispatch spine can now consult a
runtime ownership provider, and the previously-silent static fallback
becomes loud — `FORBIDDEN` rather than admission-by-claim — unless the
deployment wires the provider or declares the static fallback
explicitly. This closes the two-authorization-paths drift: a
namespace-ACL-gated op served through a door ran against the
token-carried static claim, while the same op on the native wire
consulted the revocable grant table. One behavior break (the loud
default, below); the wire surface and `OperationContext` shape are
unchanged.
### Added
- **`GatewayDispatch::with_ownership_provider` (ADR-048 amendment
"Ownership provider threading").** An `Option<Arc<dyn
OwnershipProvider>>` on the spine, wired by the dispatcher-identical
builder and read by the shared root-context builder so all three
dispatch entry points (`invoke`, `invoke_streaming`, `invoke_sink`)
thread it into `OperationContext.ownership`. `None` (the default)
previously hardwired the static fallback; now it activates the loud
default below. Regression tests:
`gateway_with_provider_consults_the_live_grant_table` (revocation is
live on the gateway surface) and
`gateway_streaming_with_provider_denies_a_static_claim_only_identity`.
- **`GatewayDispatch::with_static_ownership_fallback`** — the declared
opt-in to the ADR-011 static `Identity.resources` fallback, for
deployments that genuinely have no runtime ownership model. Turns the
door contract from a doc line into an enforceable construct: the
static fallback survives only as a code-review-visible declaration,
never as the accidental default. Regression test:
`gateway_static_fallback_opt_out_serves_resource_gated_ops`.
### Changed
- **The loud ownership-wiring default (review 012 decision 2; ADR-048
"The loud default").** The spine refuses (an error envelope,
`FORBIDDEN`, pre-dispatch, all three entry points) an op whose spec
carries `access_control.resource_type` when no ownership provider is
wired and no static fallback is declared. **Behavior break for any
deployment serving `resource_type`-gated ops through an unwired
spine** — the survey (review 012) found no published consumer relies
on the silent fallback (alkhttp, the only live spine consumer, is
unpublished; alkblobs, the only live `resource_type` op producer,
serves native-wire with providers wired). The predicate is
`resource_type` (covering the `list`/`owns_any` case where
`resource_id_path` is `None`), not `resource_id_path`;
`resource_action`-only ACLs and Internal-visibility ops are exempt
(no provider is consulted for those even on the native wire — static
resolution there is family-wide semantics). The refusal message is
operator-directed (names the wired opt-ins) and indistinguishable in
code from a normal ACL denial. Regression tests:
`gateway_refuses_a_resource_gated_op_without_a_provider`,
`gateway_streaming_and_sink_refuse_a_resource_gated_op_without_a_provider`,
and `gateway_refusal_spares_action_only_internal_and_unknown_ops`.
Specs: ADR-048's "Ownership provider threading" and "The loud
default" sections; the finding is
`docs/reviews/012-gateway-ownership-seam-consumer-input.md`.
### Verified
- 698 tests pass (including the stub guard); `clippy --all-targets
--all-features -- -D warnings`, `fmt --check`, `cargo doc`,
wasm32-unknown-unknown check + clippy, fuzz corpus replay
(`cargo test --manifest-path fuzz/shared/Cargo.toml`, 5/5),
`cargo semver-checks`, and `cargo publish --dry-run` clean.
## [0.9.0] - 2026-10-14
The stub sweep: four registered-but-always-failing stubs shipped across
@@ -834,6 +958,7 @@ Vendored core types (`Connection`, `ProtocolHandler`, `BiStream`,
(ADR-046), the channels protocol with openable-ALPNs-as-operations
(ADR-047), and the `ChannelClient` transport-agnostic client.
[0.10.0]: https://git.alk.dev/alkdev/alkcall/releases/tag/v0.10.0
[0.9.0]: https://git.alk.dev/alkdev/alkcall/releases/tag/v0.9.0
[0.8.1]: https://git.alk.dev/alkdev/alkcall/releases/tag/v0.8.1
[0.8.0]: https://git.alk.dev/alkdev/alkcall/releases/tag/v0.8.0
Generated
+1 -1
View File
@@ -27,7 +27,7 @@ dependencies = [
[[package]]
name = "alkcall"
version = "0.9.0"
version = "0.10.1"
dependencies = [
"async-trait",
"bytes",
+5 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "alkcall"
version = "0.9.0"
version = "0.10.1"
edition = "2021"
rust-version = "1.88"
license = "MIT OR Apache-2.0"
@@ -35,4 +35,7 @@ getrandom = { version = "0.4", features = ["wasm_js"] }
futures = "0.3"
parking_lot = "0.12"
bytes = "1"
zeroize = { version = "1", features = ["alloc", "derive"] }
zeroize = { version = "1", features = ["alloc", "derive"] }
[dev-dependencies]
tokio = { version = "1", features = ["rt-multi-thread"] }
+15 -5
View File
@@ -119,11 +119,12 @@ backend's banner, a sub protocol's greeting) must not lose its first
chunks.
The park is bounded: **up to 64 chunks per channel** are parked
(`EARLY_ARRIVAL_CAP`, ADR-040's memory bounds); chunks arriving past
the cap are dropped silently — at the consumer this presents as a
truncated stream with clean framing everywhere else, not as an error.
Consumer-facing consequences (review 006 E-04, noted for tunnel-style
consumers):
(`EARLY_ARRIVAL_CAP`, ADR-040's memory bounds — bounded further by the
receiver's buffer capacity when the connection is constructed with a
smaller `buffer_cap`); chunks arriving past the cap are dropped
silently — at the consumer this presents as a truncated stream with
clean framing everywhere else, not as an error. Consumer-facing
consequences (review 006 E-04, noted for tunnel-style consumers):
- Size any pre-adopt buffering math (e.g. a UDP POC's MTU-vs-buffer
sizing) against **64 parked chunks as the observable bound**, and
@@ -137,6 +138,15 @@ consumers):
adopter was too slow for the producer's burst, and the affected
streams were truncated at the park boundary.
**Adoption-window ordering (review 013):** the install and its drain of
parked chunks are one atomic section inside `adopt_channel`, and
`route_payload` parks only while that section's lock is free. A parked
chunk is therefore always pushed into the new receiver before any
direct route can begin, and a park can never straddle the install —
the reassembled byte stream preserves wire-arrival order through the
adoption window, whatever scheduling interleaves the demux and the
adopter.
## Transport-agnostic by construction
`ChannelClient` is the client side of the channels protocol. The channels
@@ -2,7 +2,7 @@
## Status
Accepted
Accepted (amended 2026-10-17 — ownership provider threading; review 012)
## Context
@@ -101,6 +101,102 @@ identity-aware refinement. When an unauthenticated caller hits both,
copy; until then the two implementations coexist (byte-identical in
behavior, one in each crate).
### Ownership provider threading (amendment 2026-10-17 — review 012)
ADR-048's enumeration of what the spine resolves into the root context
omitted the ownership provider: `GatewayDispatch`'s context builder
hardwired `ownership: None` (review 012, anchor
`src/gateway/dispatch.rs`). The promoted spine predates ADR-011's
`Dispatcher::with_ownership_provider` threading on the protocol
dispatch path, so the two surfaces drifted: the same op, same caller,
ran through the dynamic (revocable) grant table on the wire and
through the static `Identity.resources` fallback at a door —
admission-by-claim where the family's posture is admission-by-live-grant.
Downstream evidence: alkgit's LFS POC (POC-L), whose gateway-gap test
pinned the behavior with a real `git-lfs` client against a
namespace-ACL-gated `blobs/stat` (alkblobs).
Amended:
- `GatewayDispatch` carries `ownership_provider: Option<Arc<dyn
OwnershipProvider>>` — private field, wired by
`with_ownership_provider` (the dispatcher-identical builder), read
by the shared root-context builder so all three dispatch entry
points (`invoke`, `invoke_streaming`, `invoke_sink`) thread it.
`None` (the default) keeps the ADR-011 static fallback. No context
shape change (`OperationContext.ownership` stays `Option`-typed,
unchanged) and no behavior change for existing consumers — the
threading is opt-in, additive.
- **The door contract** (review 012's compensation pattern, blessed
as a doc-contract line and — since the loud default below ruled —
enforced by the spine): the ADR-011 static fallback is correct
*only* where a deployment genuinely has no runtime ownership model.
A door (alkhttp feature doors, the LFS and `git` doors, hub/relay
surfaces) serving an op whose authorization truth is a revocable
grant table MUST wire the same provider the native wire path
consults — visible-surface = authorized-surface.
- **The loud default (ruled 2026-10-17 — review 012):** the spine
refuses (an error envelope, `FORBIDDEN`) a dispatched op whose spec
carries `access_control.resource_type` when no ownership provider is
wired — `invoke`, `invoke_streaming`, and `invoke_sink` alike,
pre-dispatch. The survey (this session) found the blast radius
minimal: alkhttp is the only live `GatewayDispatch` consumer in the
family and it is unpublished, its three construction sites wire no
provider, and alkblobs is the only live op producer with
`resource_type` ACLs (its own serving sides are native-wire and
correctly wired). The refusal message is operator-directed (names
the wired opt-ins); fail-closed and indistinguishable in code from a
normal ACL denial.
**Predicate refinement:** the refusal predicate is
`access_control.resource_type.is_some()`, not
`resource_id_path.is_some()` as review 012 first drafted.
`AccessControl::check` consults the provider whenever the resource
type is set — including the `list` case (`owns_any`) where
`resource_id_path` is `None`, which silently admits any
token-carried static claim too. `resource_action`-only ACLs are
exempt: no provider is consulted for those even on the native wire,
so static resolution there is family-wide semantics, not gateway
drift. Internal-visibility ops are also exempt (the registry answers
the spec-404; the refusal must not acknowledge a never-visible op).
**The escape hatch:** `with_static_ownership_fallback()` — a
declared opt-in to the ADR-011 static `Identity.resources` fallback
for deployments that genuinely have no runtime ownership model. This
turns the door contract from a doc line into an enforceable
construct: the static fallback survives only as a code-review-visible
declaration, never as the accidental default. The default spine is
loud for `resource_type`-gated ops; wiring
`with_ownership_provider` keeps the live grant table.
Verification: the in-crate regression tests
(`gateway_with_provider_consults_the_live_grant_table`,
`gateway_streaming_with_provider_denies_a_static_claim_only_identity`)
pin the live-table override and revocation live on the gateway
surface; `gateway_refuses_a_resource_gated_op_without_a_provider`,
`gateway_streaming_and_sink_refuse_a_resource_gated_op_without_a_provider`,
and `gateway_static_fallback_opt_out_serves_resource_gated_ops` pin
the refusal (all three entry points), the opt-out, and the preserved
static-fallback mechanics under the declaration;
`gateway_refusal_spares_action_only_internal_and_unknown_ops` pins the
predicate exemptions (the pre-ruling pin
`gateway_without_provider_admits_only_the_static_claim` is superseded —
its unwired-spine-admits behavior is now the refusal and the opt-out
tests). POC-L's `tests/gateway_gap.rs`, written against published
0.9.0, pinned the pre-ruling gap: under this ruling its
unwired-spine second branch now answers `FORBIDDEN` pre-dispatch —
the POC is archival evidence of the hole, not a living conformance
suite; re-run it against the wired or opt-out spine if it is revived.
Semver note: additive — committed to main without a release (no
0.9.x publish); it rides the next release batch (0.10.0's class),
keeping the per-alkcall-release downstream churn to one coordinated
update. The loud default ships in the same unreleased batch as the
review-012 amendment, so no published alkcall version ever carried the
silent unwired-spine answer past 0.9.0.
### What the spine deliberately does NOT include
- **No HTTP mapping.** `CallError` → status/body is the consumer's
@@ -0,0 +1,139 @@
# Review 012 — The gateway dispatch's hardwired `ownership: None`: downstream evidence from alkgit's LFS POC
## Status
Resolved — filed 2026-10-17, resolved in two sessions. Report-only
consumer input; resolution shape 1 is implemented (ADR-048 amendment
"Ownership provider threading"; the threading, the in-crate regression
tests, and the re-run of POC-L's `tests/gateway_gap.rs` against
published 0.9.0 are listed there). The deferred loud-default ruling
(shape below, decision 2) was ruled in the follow-up session, after
the blast-radius survey: **adopt, with a refined predicate and an
explicit opt-out** — the spine refuses (`FORBIDDEN`, pre-dispatch, all
three entry points) an op whose spec gates on `access_control.resource_type`
when no provider is wired and no static fallback is declared; the
predicate is `resource_type` (covering the `list`/`owns_any` case the
draft wording missed), and `with_static_ownership_fallback()` is the
declared escape hatch for genuinely-static deployments. Implemented in
the same unreleased batch (ADR-048 "The loud default" section; commit
rides the 0.10.0 release class), keeping the six-consumer downstream
churn to one coordinated update.
Two decisions recorded here as required by the resolution session:
1. **Adopted:** shape 1 — thread an ownership provider through the
dispatch path (`GatewayDispatch::with_ownership_provider`, default
`None` = ADR-011 static fallback), plus the door-contract doc line
from shape 2 folded into the ADR-048 amendment. No context shape
change (`OperationContext.ownership` stays `Option`-typed,
unchanged); the `alkauth`-style shared-vocabulary question for the
record-carried-grants pattern (alkgit ADR-011, alkfs, alkdocker)
is explicitly **not** an alkcall item — pattern-level now,
micro-crate if a third consumer shows up.
2. **Adopted (ruled in the follow-up session, superseding the
original deferral):** the loud default — refusing (an error
envelope) an op whose spec gates on `resource_type` when no
provider is wired inside the dispatch spine. Survey outcome: the
blast radius is minimal (alkhttp is the only live spine consumer
and is unpublished; alkblobs is the only live `resource_type` op
producer and serves those ops native-wire with its providers
wired; no published consumer of 0.9.0's spine relies on the
silent fallback for such ops). Two refinements against the draft
wording: the predicate is `resource_type` not `resource_id_path`
(the `list` case silently falls back too), and the loud default
carries an explicit `with_static_ownership_fallback()` opt-out so
the door contract is enforceable, not just documented. Full ruling
in ADR-048's "The loud default" section.
Input basis:
- **POC-L** (alkgit, 2026-10-16; findings at
`/workspace/@alkdev/alkgit/docs/research/pocl-findings.md`; the
evidence test is `tests/gateway_gap.rs` in `/workspace/lfs-poc`) —
a real `git-lfs` client served through an alkhttp extras door
mapping the LFS batch API onto alkblobs' `ops` (`blobs/stat`,
namespace-ACL-gated, External visibility).
- The ask ledger: alkgit's `substrate-recomposition.md`
§Upstream-push candidates #6 (the "alkhttp gateway ownership seam"
entry — filed here under the crate that owns the dispatch spine).
---
## 1. The finding
`GatewayDispatch::invoke`'s `OperationContext` builder hardwires
`ownership: None` when it assembles an external identity's context
(`src/gateway/dispatch.rs`, the builder around lines 215–249; the
`ownership: None` field assignment verified this session). Downstream
consequence, demonstrated live in POC-L:
- `AccessControl::check` for a namespace-ACL-gated op (`resource_id`
extracted from the payload — alkblobs' `blobs/stat` with
`resource_id_path: $.namespace`, External visibility) falls back
to the **static** `Identity.resources` model when no ownership
provider is present in the context.
- Therefore a token-**carried static claim** — e.g. an LFS-style
bearer JWT embedding `{"blob-namespace": ["read"]}` in its
resources map — **is admitted through the gateway with no runtime
provider consulted**, while the same op served through the native
call protocol consults the dynamic (revocable) grant table and
admits only live grants.
- The dynamic grant table is *unreachable* from this dispatch spine:
no thread of the gateway path carries an ownership provider, so a
deployment whose authorization truth is a revocable table gets
stale-at-birth answers on the HTTP-gateway surface — revocation
and grant changes never apply there.
The security frame is the family's standing invariant
(visible-surface = authorized-surface): an External op must not have
two authorization paths with different revocation semantics
depending on which door the caller walked through. The static
fallback is admission-by-claim at the exact point where the family's
posture wants admission-by-live-grant.
## 2. What it means for the consumer (and what they did about it)
The POC-L door cannot use the gateway dispatch spine for these ops.
Its compensation: build the `OperationContext` itself (identity from
the resolved HTTP identity + the ownership provider wired
explicitly) and call `OperationRegistry::invoke("blobs/stat", …)`
directly — after which the namespace ACL runs exactly as it would on
the native wire, against live grant sets (POC-L finding 2, the batch
binding section). The compensation is correct and verified, but it
is per-door hand-assembly of what the dispatch spine exists to do —
every future door (alkhttp's `git` feature, the LFS door proper,
alkssh's surfaces) inherits the same manual step and the same trap of
forgetting it (the spine *works* with the fallback, silently).
## 3. Resolution shapes (recommendation only — own session)
1. **Thread an ownership provider through the dispatch path** — an
`Option` on the gateway dispatch builder (per-dispatch config or
a connection-establishment property), `None` only where a
deployment genuinely has no ownership model. Preserves the
single-spine posture; needs an ADR if the context shape is
frozen-published (alkcall is 0.9, published — the context builder
surface's semver class decides).
2. **Bless the compensation**: document the door-side pattern (door
builds its own context with ownership wired; never route
External namespace-ACL ops through `GatewayDispatch::invoke`
without it) as a *door contract* line in the gateway docs, plus
a loud static-fallback warning at the hardwired site. Cheap, but
the trap stays structural — silent-by-default.
3. **Hybrid**: doc contract now (2), seam threading as the ADR'd
follow-up (1). Probably where the session lands, given 0.9 is
published and the ADR round for the context shape is not open.
**Verification status**: the `ownership: None` anchor and the
builder's context assembly were re-read this session; the admission
behavior (static `Identity.resources` fallback, token-carried claim
admitted) is POC-L's `tests/gateway_gap.rs` result, witnessed in the
findings doc. Re-verify the behavior during resolution by re-running
that test against a current alkcall.
## 4. Non-goal note
Not asking for anything on the native call-protocol path: the
channel-path dispatch carries ownership correctly (the ops wire path
is what the whole family gates through). The finding is scoped to
the gateway dispatch spine used by HTTP-door routing.
@@ -0,0 +1,264 @@
# Review 013 — The adoption-window ordering race: a chunk routed in the `adopt_channel` install→drain window overtakes earlier-parked early arrivals and misorders the channel's byte stream
## Status
Resolved — filed 2026-10-21, resolved 2026-10-22 (same session). C-01
was re-verified (code walk plus a latch-based repro: holding the old
install→drain window open via a test-only gate makes a receiver read
`CP1P2` where the wire order was P1, P2, C — the deterministically
observed inversion), then closed by making the install and its park
drain **one atomic section** (the re-review's fix direction 1+3
hybrid: `route_payload`'s map-miss-and-park is now also one lock
section, so nothing can straddle either edge of the window). The
resolution is fully synchronous — no settle flag, no extra task, no
`pending` queue per channel:
- `route_payload` holds the channels-map lock across the miss → park
decision (`park_early_arrival` runs nested under it) — a park can
never straddle the install, closing the orphaned-chunk loss variant
the re-review added to the finding (a chunk that missed the map
pre-install but parked after the post-install drain would sit
parked forever, invisible to `dropped_unknown_chunks`).
- `install_locked` (new, shared by `open_channel` /
`adopt_channel` / `install_channel_zero`) swaps the parked queue
out, inserts the state, and `try_send`s the parked queue into the
fresh, empty receiver **before releasing the channels lock** — a
direct route (map hit) can only begin after that release, so parked
chunks always precede it in mpsc send order. No await runs inside
the section: the effective park cap is
`min(EARLY_ARRIVAL_CAP, buffer_cap)` so the parked queue always fits
the fresh receiver. `drain_early_arrivals` remains as the defensive
full-buffer fallback (previously-unreachable-by-invariant) and is
now called by all three install paths.
- **Semver: patch release of the 0.10 line** (0.10.1) — `ChannelState`
and the fix surface are private; the API surface, the wire formats,
and the documented cap behavior (review 006 E-04) are unchanged;
behavior only strengthens (in-window drops and the orphan loss
become unreachable).
Regression tests (src/channels/manager.rs): the chunk-granularity
ordering test the fix sketch called for
(`adopt_window_does_not_reorder_chunks` — 200 iterations on a real
2-worker runtime, a racer routing C the moment the install becomes
visible, receiver byte stream asserted byte-identical to the
producer's — catches the old window probabilistically under genuine
preemption, exactly the alkgit repro shape), the pre-install park
FIFO shape (`adopt_drains_parked_and_raced_chunks_in_fifo`), and the
park-cap normalization (`park_cap_bounded_by_receiver_buffer`).
**Downstream re-verification gate still applies**: alkgit's flaking
row (`tests/native_transport.rs …` under multi-process load, finding
F-2) must pass against the fixed crate before publishing 0.10.1.
Triggered by a downstream finding: alkgit's
wave-2 review gate (`/workspace/@alkdev/alkgit/` —
`core/review-wave-2`) caught its channels open-op integration test
flaking under multi-process load, instrumented the failing byte stream,
and traced every failure to the window analyzed below. The downstream
diagnostic data (two distinct desync shapes at two distinct offsets)
is reproduced here in full; alkgit carries the panic improvement
(alkgit `b6db513`) that makes the shape observable.
As filed — before the resolution — this review contained no alkcall
code change: it was the finding record, the repro evidence, and the
fix-shape sketch the implementing patch builds from. **Recommended
urgency: patch release of the published 0.10 line** (the ruling the
Status section implements), because the failure mode is silent
byte-stream corruption on a freshly adopted channel and the documented
downstream consumers (the git producer half in alkgit; any
dial-then-pump tunnel shape) sit exactly on the affected window.
## Scope
`src/channels/manager.rs` — the adopt/adopt-equivalent paths and the
demux receive side they race:
- `ChannelManager::adopt_channel` (`manager.rs:338-402`): the
install (`channels.insert`, `manager.rs:389-394`) and the
post-install drain (`drain_early_arrivals`, `manager.rs:402`).
- `ChannelManager::route_payload` (`manager.rs:515-533`): the
channels-map lookup → direct `send` path, and the not-yet-installed
→ `park_early_arrival` fallback.
- `ChannelManager::park_early_arrival` (`manager.rs:539-561`).
- `ChannelManager::drain_early_arrivals` (`manager.rs:567+`).
The open-reply adoption path (`ChannelClient`/open-op consumers
calling `adopt_channel` after the open response) and the
`open_channel` path share the same shape; the observed failures are on
the `adopt_channel` path (the consumer side of the open op).
This is the ordering half of the adopt-race family review 006 E-04
already bounded ("early-arrival park cap (64) interacts with
push-first producers under slow adopters"): E-04 covered the *cap drop*
of chunks arriving before adopt and classed it not-a-correctness-bug
because the cap is documented behavior. This review covers the
*ordering* of chunks around the moment routing state becomes visible —
nothing documented bounds it, and nothing in the implementation
preserves it.
## Severity legend note
Same scale as reviews 004/005/006. Tagged
**[critical]** by E-01's second clause ("corrupts data"): the channel's
reassembled byte stream is delivered with misordered bytes — which, at
the layer every real consumer sits on (a framing codec), is corrupted
data. The mitigating fact (the window is narrow and adoption-time-only)
bounds frequency, not impact; a consumer cannot detect or recover from
a misordered chunk, and the REQ-CH series pins delivery as a byte
stream on a "single ordered, reliable" substrate
(`docs/architecture/channels-wire.md:9`) — reassembly delivering
out-of-order violates the one property the demux exists to preserve.
## The finding
### C-01 [critical] — a chunk arriving between install and drain is delivered ahead of the chunks parked before install
**Sequence** (single channel id, one demux):
1. The demux receives chunk P1 for `channel_id`; `route_payload`'s
map lookup misses (the adopting side has not run
`adopt_channel` yet) → P1 is parked in `early_arrivals` (FIFO).
Possibly P2…
2. The adopting side's `adopt_channel` runs: `channels.insert`
stores the state carrying the new `demux_sender`. **The map
lock is released here.**
3. The demux receives chunk C for the same `channel_id` *now*, inside
the window between the insert (step 2) and the drain
(`drain_early_arrivals`, step 4): `route_payload`'s lookup **hits**
→ C is sent **directly** into the mpsc.
4. `drain_early_arrivals` then pushes P1 (and any other parked
chunks) into the **same** mpsc.
Delivery order on the byte stream: **C, then P1** — with P1 having
arrived on the wire *earlier*. The mpsc preserves send order, and the
send order is inverted relative to wire-arrival order. The receiver
(`MpscRecvStream`) has no way to know; it reassembles faithfully in
send order = wrong order.
The window is the wall-clock span between `manager.rs:394` (insert
completes) and the drain's first take — normally nanoseconds apart,
which is why isolated runs never observe it; under scheduling pressure
(spawned adoption task descheduled immediately after the insert, or
the pending demux `send` completing faster than the adoption task's
next lock acquisition) the window is real and reproducible.
**Observed evidence** (alkgit `tests/native_transport.rs`
`the_channels_open_op_serves_a_fetch_session_on_the_opened_channel`;
11 tests, both processes multi-threaded, ~2 failures in ~10
suite-adjacent runs, 0 in ~15 isolated single-test runs). The test
reads the server's V2 capability advertisement back from the channel
as a raw pkt-line stream (`read_exact` 4-byte hex headers). The
advertisement is a fixed byte sequence (4+10 bytes `version 2`, hex
headers + `ls-refs=unborn` / `fetch=wait-for-done` /
`object-format=sha1` lines), so any desync is attributable to
delivery, not content:
- Failure shape 1 — after **0** lines, the first 4-byte header read
came out `fetc` = `[0x66,0x65,0x74,0x63]`, followed by
`h=wait-for-done`. That is offset ~16 bytes into the advertisement
(inside the `fetch=wait-for-done` line, four bytes past its
header): **the leading chunks were delivered late**; the reader
started mid-stream.
- Failure shape 2 — after **1** line (`version 2` parsed correctly),
the next header read came out `00ev` = `[0x30,0x30,0x65,0x76]`
(the `00` prefix of a hex length straddled with text): a **chunk
boundary two bytes into the second line's header** — the second
chunk was delivered ahead of (or its tail interleaved against) the
chunk carrying the line it belongs to.
Both shapes are exactly the C-befor-P1 inversion at different chunk
granularities (chunk sizes vary with scheduling).
A further aggravation this review flags on the same paths: the window
also interacts with `park_early_arrival`'s
`EARLY_ARRIVAL_CAP` overflow branch (`manager.rs:542-551`) — a burst
that parks *while* the window delays the drain can additionally
**drop** chunks. That combination is loss, not just misorder; any fix
must be ordering-exact *and* keep the drop path unreachable while the
channel is installed-but-undrained.
**Why the consumer cannot fix it** (and why no downstream workaround
exists or should): the reassembled `BiStream` is the promised byte
stream; a framing codec (alkgit's pkt-line reader, a tunnel's inner
protocol) consumes bytes with no awareness that chunk reassembly
happened beneath it. Alkgit carries no compensating logic and must
not — the fix belongs here, in the one place that owns delivery order.
### Fix-shape sketch (for the implementing patch)
The window closes when *all* chunks of a channel are serialized
through one ordering decision. Candidate directions, in increasing
invasiveness:
1. **Drain under the channels-map lock.** Hold the map lock across
the insert *and* the parked-queue swap-out, then push the swapped
queue into the mpsc (the direct `route_payload` paths arriving
behind the lock block until the drain's queue is out; they then
send directly — after the parked bytes). The push of the swapped
queue must not await while the lock is held: `try_send` first,
draining backpressure by holding the *queue*, not the lock, gets
complicated — see direction 3 for a clean cut.
2. **Re-check in `route_payload`'s park branch.** After parking,
re-check the map; if the channel now exists, take the queue back
out (FIFO) and push it through the sender, then the just-parked
chunk. This closes the "parked-after-install" half but leaves the
window's other half (a direct send overtaking parked chunks whose
drain hasn't run yet) — a direct route still sends C before the
adopter pushes P1. Not sufficient alone.
3. **Per-channel ordered mailbox.** Make every chunk for an installed
channel take the channel's own queue-lock, append, and hand off to
a single per-channel drainer (or drain under the queue lock with
`try_send` + a parked-overflow flag the drainer resumes). This
serializes park, install-drain, and direct route on one lock per
channel and closes every interleave, including the push-first
producer shapes of 006 E-04.
Whichever shape lands must add a **chunk-granularity ordering
regression test**: a producer that emits many small chunks with an
open-reply-then-immediate-pump shape, the adopter parked mid-window
under injected scheduling (e.g. a `park` latch between insert and
drain — a test-only seam), asserting the receiver's byte stream is
byte-identical to the producer's. The two observed desync shapes above
are the acceptance shapes. The `route_payload_to_open_channel_succeeds`
single-chunk test does not exercise the ordering and must not be
counted as covering this.
**Downstream re-verification gate**: alkcall's fix must be
re-verified against alkgit's flaking row under multi-process load
(the condition that reproduced it) — the alkgit review task
(`core/review-wave-2`, finding F-2) records this; wave 8's
end-to-end `git push`/`git clone` rows ride channel adoption with
push-first producers and would exercise the window hard without the
fix.
## Non-findings (verified, recorded to bound the re-review)
- The demux/mux wire format is not implicated: the 8-byte chunk header
is per-chunk framing only, and the misorder below it is a manager
scheduling property (ADR-034's wire surface untouched; no ADR-level
wire change is needed for the fix).
- `MpscRecvStream`'s pending-chunk continuation is correct — the
observed corruptions are all *order* corruptions, never
*content* corruption; the reassembly layer faithfully delivered
what it was handed.
- The install-side duplicate/collision checks in `adopt_channel`
(`ChannelExists` + the 0.8.1 regression pin) are unaffected by this
finding; the fix must preserve them and the adopt-parity rules.
## References
- alkgit wave-2 gate review:
`/workspace/@alkdev/alkgit/tasks/implementation/core/review-wave-2.md`
(finding F-2, with the repro data and the fix-urgency routing) and
the diagnostic commit alkgit `b6db513`
(`tests/native_transport.rs` → `read_channel_pkts`'s desync panic).
- Review 006 E-04 (the early-arrival cap half of this family;
`docs/reviews/006-channel-open-establishment-gap-review.md`),
review 007 (the establishment follow-ups).
- `docs/architecture/channels-adapter.md` (the demux invariants
REQ-CH-02/04 this finding sits between),
`docs/architecture/channels-wire.md` (the ordered-substrate premise,
line 9; REQ-CH-01..05 wire invariants),
`docs/architecture/decisions/039-channelsadapter-and-channelmanager.md`
(the substrate-agnostic demux loop contract).
- alkcall `a22b2b8` — the original park-early-arrival fix (the
open/first-data race this review's window is the residue of).
+1 -1
View File
@@ -27,7 +27,7 @@ dependencies = [
[[package]]
name = "alkcall"
version = "0.8.1"
version = "0.10.1"
dependencies = [
"async-trait",
"bytes",
+320 -43
View File
@@ -23,6 +23,7 @@ use serde_json::Value;
use super::mux::{MuxHandle, OpenerLedger};
use super::reassembly::{MpscRecvStream, MpscSendStream, DEFAULT_BUFFER_CAP};
use tokio::sync::mpsc::error::TrySendError;
/// Which side of a channels connection this manager belongs to.
/// Determines the ID allocation range to prevent collisions when both
@@ -126,21 +127,22 @@ struct Inner {
control_handles: Mutex<HashMap<u32, ChannelControl>>,
}
/// The per-channel cap on parked early-arrival chunks (`route_payload`
/// parks chunks for not-yet-adopted channels instead of dropping them —
/// the open-op response/first-data race). A channel whose adopter never
/// arrives leaks its parked chunks until `clear_all`; the cap bounds
/// that leak per channel.
/// The nominal per-channel cap on parked early-arrival chunks
/// (`route_payload` parks chunks for not-yet-adopted channels instead
/// of dropping them — the open-op response/first-data race). A channel
/// whose adopter never arrives leaks its parked chunks until
/// `clear_all`; the cap bounds that leak per channel.
///
/// Consumer-facing bound (review 006 E-04): for a push-first producer
/// (dial-then-pump is the common tunnel shape), up to `EARLY_ARRIVAL_CAP`
/// chunks per channel can be parked before the consumer's
/// `adopt_channel` runs — and chunks past the cap drop silently at the
/// consumer (data loss presents as a truncated stream with clean
/// framing elsewhere; `dropped_unknown_chunks` /
/// `early_arrival_count` are the producer-side observability). A
/// tunnel crate's sizing (e.g. a UDP POC's MTU-vs-buffer math) should
/// treat 64 parked chunks as the observable bound and adopt promptly.
/// (dial-then-pump is the common tunnel shape), up to the effective
/// park cap (see `ChannelManager::early_arrival_cap`) chunks per
/// channel can be parked before the consumer's `adopt_channel` runs —
/// and chunks past the cap drop silently at the consumer (data loss
/// presents as a truncated stream with clean framing elsewhere;
/// `dropped_unknown_chunks` / `early_arrival_count` are the
/// producer-side observability). A tunnel crate's sizing (e.g. a UDP
/// POC's MTU-vs-buffer math) should treat the effective park cap as
/// the observable bound and adopt promptly.
const EARLY_ARRIVAL_CAP: usize = 64;
impl ChannelManager {
@@ -311,11 +313,21 @@ impl ChannelManager {
if channels.contains_key(&channel_id) {
return Err(ManagerError::ChannelExists(channel_id));
}
channels.insert(channel_id, state);
Self::install_locked(
&mut channels,
&self.inner.early_arrivals,
channel_id,
state,
&demux_sender,
);
}
self.inner.opener_ledger.record(channel_id, opener);
// Route any parked chunks that survived the install (the
// defensive re-park path — normally nothing lands here).
self.drain_early_arrivals(channel_id, &demux_sender).await;
Ok((channel_id, send, recv))
}
@@ -335,6 +347,14 @@ impl ChannelManager {
/// (collision — the remote side allocated an ID this side already
/// uses). Returns `TooManyChannels` if the per-connection cap is
/// reached.
///
/// **Adoption-window ordering (review 013):** the install and its
/// drain of parked early arrivals share one channels-lock section
/// (`install_locked`), and `route_payload` parks only while that
/// lock is free. A parked chunk is therefore always pushed into
/// the new receiver before any direct route can begin, and a park
/// can never straddle the install — the channel's byte stream
/// preserves wire-arrival order through the adoption window.
pub async fn adopt_channel(
&self,
channel_id: u32,
@@ -394,11 +414,20 @@ impl ChannelManager {
if channels.contains_key(&channel_id) {
return Err(ManagerError::ChannelExists(channel_id));
}
channels.insert(channel_id, state);
Self::install_locked(
&mut channels,
&self.inner.early_arrivals,
channel_id,
state,
&demux_sender,
);
}
// Route any chunks that arrived between the open-op response and
// this adopt (the producer may have started pumping already).
// Route any parked chunks that survived the install (the
// defensive re-park path — normally the install already pushed
// them). Chunks parked between the open-op response and this
// adopt were pushed under the lock above (review 013); the
// receiver was fresh and empty, so nothing ever lands here.
self.drain_early_arrivals(channel_id, &demux_sender).await;
Ok((send, recv))
@@ -474,19 +503,27 @@ impl ChannelManager {
.map_err(|_| ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO))?;
let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap);
let state = ChannelState {
demux_sender,
demux_sender: demux_sender.clone(),
handler_task,
alpn: "alk/call".to_string(),
};
{
let mut channels = self.inner.channels.lock();
if channels
.insert(super::wire::CHANNEL_ID_ZERO, state)
.is_some()
{
if channels.contains_key(&super::wire::CHANNEL_ID_ZERO) {
return Err(ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO));
}
Self::install_locked(
&mut channels,
&self.inner.early_arrivals,
super::wire::CHANNEL_ID_ZERO,
state,
&demux_sender,
);
}
// Route any parked chunks that survived the install (the
// defensive re-park path — normally nothing lands here).
self.drain_early_arrivals(super::wire::CHANNEL_ID_ZERO, &demux_sender)
.await;
// Channel 0 is opened by the accept side itself; the opener
// ledger records the local peer (or is left empty for channel
// 0 — the cap doesn't apply to the pre-negotiated control
@@ -498,7 +535,7 @@ impl ChannelManager {
/// `channel_id` (the demux's per-chunk route). A zero-length
/// payload is the EOF sentinel — the reassembled stream interprets
/// it as EOF (REQ-CH-01). An unknown `channel_id` parks the payload
/// in the early-arrival buffer (up to `EARLY_ARRIVAL_CAP` payloads
/// in the early-arrival buffer (up to `early_arrival_cap` payloads
/// per channel) — the open-op response carrying `channel_id` and
/// the producer's first data-plane write race, and a push-first
/// producer (a TTY backend's banner, a sub protocol's greeting)
@@ -507,6 +544,22 @@ impl ChannelManager {
/// drop with a debug log (the adopted-side REQ-CH-04 leniency for
/// genuinely unknown channels).
///
/// The park decision is atomic with the channels-map state: the
/// map miss and the park happen under the same channels-lock
/// acquisition, so an installing `adopt_channel` can never straddle
/// them (review 013 — a chunk parked after the install drained the
/// queue would otherwise be orphaned). A park can only be observed
/// by an installer that has not inserted yet, and that installer
/// swaps the queue out when it inserts.
///
/// **Ordering (review 013):** when the install and its drain of
/// parked chunks are one lock section, a direct route (map hit)
/// can only begin after that section released — the parked chunks
/// are already in the receiver's mpsc by then. Send order on the
/// mpsc therefore matches wire-arrival order for every chunk of a
/// freshly adopted channel; a direct route can never overtake an
/// earlier-parked chunk.
///
/// Awaits the bounded channel sender — if the handler's read half
/// is slow, the demux stalls here (ADR-040 REQ-CH-05: lossless
/// bounded-buffer backpressure). The demux loop is the only caller;
@@ -515,31 +568,50 @@ impl ChannelManager {
pub async fn route_payload(&self, channel_id: u32, payload: Bytes) {
let sender = {
let channels = self.inner.channels.lock();
channels.get(&channel_id).map(|s| s.demux_sender.clone())
match channels.get(&channel_id) {
Some(state) => state.demux_sender.clone(),
None => return self.park_early_arrival(channel_id, payload),
}
};
match sender {
Some(sender) => {
if sender.send(payload).await.is_err() {
debug!(
channel_id,
"demux: channel receiver dropped, dropping chunk"
);
}
}
None => {
self.park_early_arrival(channel_id, payload);
}
if sender.send(payload).await.is_err() {
debug!(
channel_id,
"demux: channel receiver dropped, dropping chunk"
);
}
}
/// The effective per-channel park cap: the nominal
/// `EARLY_ARRIVAL_CAP` bounded by the receiver's mpsc buffer
/// capacity. The install paths drain the parked queue with
/// `try_send` inside their channels-lock section (review 013), and
/// the drain target is the fresh, empty receiver — bounding the
/// park cap by the buffer capacity guarantees the drain never
/// needs to await (which it cannot, holding a `parking_lot`
/// lock). With the default constructor settings both are 64 and
/// the bound is inactive; smaller `buffer_cap` values tighten the
/// park cap instead of turning the drain into a stall.
fn early_arrival_cap(&self) -> usize {
EARLY_ARRIVAL_CAP.min(self.inner.buffer_cap)
}
/// Park an early-arrival chunk for `channel_id` (the adopter has
/// not run yet). FIFO per channel, bounded by `EARLY_ARRIVAL_CAP`;
/// beyond the cap the chunk is dropped with a debug log and the
/// dropped counter increments (observability only).
/// not run yet). FIFO per channel, bounded by
/// `early_arrival_cap`; beyond the cap the chunk is dropped with a
/// debug log and the dropped counter increments (observability
/// only).
///
/// Called with the channels lock already held (review 013): the
/// caller (`route_payload`) holds it across the map miss and this
/// park so the decision is atomic with an in-flight install — a
/// parked chunk is always visible to the installer that caused the
/// miss. Lock order is channels → early_arrivals; no other path
/// nests them the other way.
fn park_early_arrival(&self, channel_id: u32, payload: Bytes) {
let cap = self.early_arrival_cap();
let mut arrivals = self.inner.early_arrivals.lock();
let queue = arrivals.entry(channel_id).or_default();
if queue.len() >= EARLY_ARRIVAL_CAP {
if queue.len() >= cap {
drop(arrivals);
self.inner
.dropped_unknown_chunks
@@ -561,9 +633,21 @@ impl ChannelManager {
}
/// Drain the early-arrival buffer for `channel_id` into `recv`'s
/// sender (called by `adopt_channel` after the channel state is
/// installed). Returns the drained payload count. Payloads past the
/// receiver's buffer capacity await normally (backpressure).
/// sender (called after the channel state is installed). Returns
/// the drained payload count. Payloads past the receiver's buffer
/// capacity await normally (backpressure).
///
/// This is the defensive leftover path only: since review 013 the
/// install paths (`open_channel` / `adopt_channel` /
/// `install_channel_zero`) drain the parked queue under the
/// channels lock via `install_locked`, and the effective park cap
/// is bounded by the receiver's buffer capacity so that drain
/// never needs this. Only chunks re-parked by `install_locked`'s
/// full-buffer fallback (violating the buffer-cap invariant) or
/// chunks parked for an id with no installer pending at all reach
/// this method after the fact. Chunks parked for an id that is
/// never adopted stay parked until `clear_all` (the documented
/// leak bound).
async fn drain_early_arrivals(
&self,
channel_id: u32,
@@ -632,6 +716,63 @@ impl ChannelManager {
self.inner.opener_ledger.drain()
}
/// Core install, called with the channels lock already held: swap
/// out any parked early-arrival queue for `channel_id` and push it
/// into the fresh receiver's mpsc **before the lock is released**
/// (review 013 — the install and its park-drain are one atomic
/// decision, so a direct route (a map hit after release) can never
/// overtake an earlier-parked chunk).
///
/// The parked queue never exceeds `early_arrival_cap` (bounded by
/// the receiver's buffer capacity) and the receiver is fresh and
/// empty, so each push is a non-blocking `try_send`. Chunks
/// bouncing off a full buffer (possible only if the buffer-cap
/// invariant above is violated) are re-parked for the caller's
/// post-release async drain — the pre-fix behavior, kept as a
/// defensive fallback rather than a correctness path; chunks hit by
/// a dropped receiver are dropped as everywhere else.
fn install_locked(
channels: &mut HashMap<u32, ChannelState>,
early_arrivals: &Mutex<HashMap<u32, VecDeque<Bytes>>>,
channel_id: u32,
state: ChannelState,
demux_sender: &mpsc::Sender<Bytes>,
) {
let parked = early_arrivals
.lock()
.remove(&channel_id)
.unwrap_or_default();
channels.insert(channel_id, state);
let mut leftovers = VecDeque::new();
for payload in parked {
if let Err(error) = demux_sender.try_send(payload) {
match error {
TrySendError::Full(bounced) => {
debug!(
channel_id,
"install: receiver buffer full draining early arrivals"
);
leftovers.push_back(bounced);
break;
}
TrySendError::Closed(_) => {
debug!(
channel_id,
"install: receiver dropped, dropping early arrival"
);
}
}
}
}
if !leftovers.is_empty() {
let mut arrivals = early_arrivals.lock();
let queue = arrivals.entry(channel_id).or_default();
for payload in leftovers {
queue.push_back(payload);
}
}
}
/// `true` if `channel_id` is currently open.
pub fn has_channel(&self, channel_id: u32) -> bool {
self.inner.channels.lock().contains_key(&channel_id)
@@ -1134,4 +1275,140 @@ mod tests {
Ok(_) => panic!("expected TooManyChannels, got Ok"),
}
}
/// Review 013 regression: a chunk routed through the adoption
/// window must not overtake earlier-parked chunks on the receiver's
/// byte stream. The racer routes `C` as soon as the install is
/// visible — under the pre-fix code and real preemption that could
/// land in the install→drain window and be delivered first (with
/// P1/P2 parked); the install and its park-drain being one lock
/// section makes P1, P2, C the only reachable order, whichever
/// side of the install the racer's route lands on (pre-install →
/// parked and drained FIFO; post-install → sent after the park
/// queue already went out). 200 iterations to give a real
/// scheduler the chance to catch the old window; the receiver's
/// byte stream must be byte-identical to the producer's every time.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn adopt_window_does_not_reorder_chunks() {
use std::time::Duration;
use tokio::io::AsyncReadExt;
for iteration in 0..200u32 {
let manager = make_manager_with_runner().await;
let channel_id = 7 + iteration * 2;
manager
.route_payload(channel_id, Bytes::from_static(b"P1"))
.await;
manager
.route_payload(channel_id, Bytes::from_static(b"P2"))
.await;
let adopt_manager = manager.clone();
let adopter = tokio::spawn(async move {
adopt_manager
.adopt_channel(channel_id, "alk/tty", None)
.await
.expect("adopt")
});
let race_manager = manager.clone();
let racer = tokio::spawn(async move {
loop {
if race_manager.has_channel(channel_id) {
race_manager
.route_payload(channel_id, Bytes::from_static(b"C"))
.await;
break;
}
tokio::task::yield_now().await;
}
});
let (_send, mut recv) = adopter.await.expect("adopt task join");
racer.await.expect("race task join");
let mut buf = [0u8; 5];
tokio::time::timeout(Duration::from_secs(10), recv.read_exact(&mut buf))
.await
.expect("read within timeout")
.expect("read all");
assert_eq!(
&buf, b"P1P2C",
"iteration {iteration}: the receiver's byte stream diverged from
the producer's wire order (C overtook parked chunks)"
);
}
}
/// Parked chunks routed while the install is in flight (map still
/// empty) are drained in FIFO order after the parked prefix — the
/// park decision is atomic with the map state, so a chunk can only
/// miss the map while the adopter has not inserted yet, and that
/// adopter swaps the whole queue out when it does.
#[tokio::test]
async fn adopt_drains_parked_and_raced_chunks_in_fifo() {
let manager = make_manager_with_runner().await;
manager.route_payload(7, Bytes::from_static(b"P1")).await;
manager.route_payload(7, Bytes::from_static(b"P2")).await;
let adopt_manager = manager.clone();
let adopter = tokio::spawn(async move {
adopt_manager
.adopt_channel(7, "alk/tty", None)
.await
.expect("adopt")
});
while !manager.has_channel(7) {
tokio::task::yield_now().await;
}
let (_send, mut recv) = adopter.await.expect("adopt task join");
manager.route_payload(7, Bytes::from_static(b"C")).await;
use tokio::io::AsyncReadExt;
let mut buf = [0u8; 5];
recv.read_exact(&mut buf).await.expect("read all");
assert_eq!(&buf, b"P1P2C", "raced chunk must follow the parked prefix");
}
/// The effective park cap is the nominal EARLY_ARRIVAL_CAP bounded
/// by the receiver's buffer capacity (review 013 — the install
/// drains the parked queue with non-blocking sends into the fresh
/// receiver, which must always fit). With buffer_cap 4 the fifth
/// parked chunk drops at park time (the documented E-04 cap
/// behavior, tightened) and the first four arrive on adopt.
#[tokio::test]
async fn park_cap_bounded_by_receiver_buffer() {
let (_client, server) = duplex(1024);
let (_reader, writer) = tokio::io::split(server);
let (handle, runner) = MuxRunner::new(Box::new(writer));
tokio::spawn(async move {
let _ = runner.run().await;
});
let manager = ChannelManager::new(handle, 256, 4, None, ChannelSide::Accept);
for chunk in b"abcde" {
manager
.route_payload(7, Bytes::copy_from_slice(&[*chunk]))
.await;
}
assert_eq!(
manager.dropped_unknown_chunks(),
1,
"the fifth parked chunk dropped (park cap = min(64, buffer_cap=4))"
);
let (_send, mut recv) = manager
.adopt_channel(7, "alk/tty", None)
.await
.expect("adopt");
use tokio::io::AsyncReadExt;
let mut buf = [0u8; 4];
recv.read_exact(&mut buf)
.await
.expect("read parked quartet");
assert_eq!(&buf, b"abcd", "parked chunks arrive in FIFO order");
}
}
+535 -4
View File
@@ -8,8 +8,12 @@
//! every transport (`internal: false` — ACL runs against the caller's
//! identity, not a handler's composition authority; `forwarded_for:
//! None` — wire-ingress only), resolves the registration's
//! composition authority / capabilities / scoped env into it, and
//! bounds Once-ops and sink dispatch with a configurable deadline
//! composition authority / capabilities / scoped env into it, carries
//! the spine-wide ownership provider when one is wired (ADR-011; the
//! ADR-048 amendment — review 012's door contract), refuses a
//! `resource_type`-gated op served without a provider or a declared
//! static fallback (review 012's loud-default ruling), and bounds Once-ops
//! and sink dispatch with a configurable deadline
//! while leaving streaming subscriptions unbounded (ADR-021:
//! subscriptions are long-lived).
//!
@@ -26,6 +30,7 @@ use std::sync::Arc;
use std::time::{Duration, Instant};
use crate::core::auth::Identity;
use crate::core::ownership::OwnershipProvider;
use crate::core::types::Capabilities;
use crate::protocol::wire::{CallError, ResponseEnvelope};
use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
@@ -46,12 +51,17 @@ pub const DEFAULT_DEADLINE: Duration = Duration::from_secs(30);
/// invokes operations for the neutral `ResponseEnvelope` result shape
/// any transport maps to its own wire format. Identity arrives
/// per-call as `Option<Identity>` — bearer resolution, transport
/// framing, and error presentation happen upstream in the consumer.
/// framing, and error presentation happen upstream in the consumer;
/// the ownership provider (review 012 / ADR-048 amendment) arrives
/// once per spine via [`GatewayDispatch::with_ownership_provider`] and
/// is carried by every root context this spine builds.
///
/// See the [module docs](crate::gateway) and ADR-048.
pub struct GatewayDispatch {
registry: Arc<OperationRegistry>,
deadline: Option<Duration>,
ownership_provider: Option<Arc<dyn OwnershipProvider>>,
static_ownership_fallback: bool,
invoke_count: AtomicUsize,
}
@@ -62,10 +72,45 @@ impl GatewayDispatch {
Self {
registry,
deadline: Some(DEFAULT_DEADLINE),
ownership_provider: None,
static_ownership_fallback: false,
invoke_count: AtomicUsize::new(0),
}
}
/// Wire the ownership provider the spine's root contexts carry
/// (ADR-011): an op whose `access_control` gates on a
/// `resource_type` then consults this provider with the
/// `resource_id_path`-extracted resource ID at check time (the
/// live, revocable grant table) — or `owns_any` for the `list`
/// case — instead of the static `Identity.resources` fallback.
/// The door contract: any door serving ops whose authorization
/// truth is a dynamic grant table MUST wire the same provider the
/// native wire path consults — visible-surface =
/// authorized-surface (review 012 / ADR-048 amendment). The
/// default ([`GatewayDispatch::new`]) is `None`, and the default
/// policy is loud: without a provider or a declared
/// [`GatewayDispatch::with_static_ownership_fallback`], a
/// `resource_type`-gated op is refused (an error envelope,
/// review 012's ruling).
pub fn with_ownership_provider(mut self, provider: Arc<dyn OwnershipProvider>) -> Self {
self.ownership_provider = Some(provider);
self
}
/// Declare this deployment's static `Identity.resources` fallback
/// as a deliberate choice: ops whose `access_control` gates on a
/// `resource_type` are served through the ADR-011 static-fallback
/// check (token-carried claims), not refused for the missing
/// provider wiring. Use this only where the deployment genuinely
/// has no runtime ownership model — review 012's door contract
/// makes this a code-review-visible decision, not the accidental
/// default.
pub fn with_static_ownership_fallback(mut self) -> Self {
self.static_ownership_fallback = true;
self
}
/// Override the Once/sink deadline: `Some(duration)` bounds every
/// [`GatewayDispatch::invoke`] and [`GatewayDispatch::invoke_sink`]
/// dispatch (a hung handler surfaces as a retryable `TIMEOUT` error
@@ -106,6 +151,9 @@ impl GatewayDispatch {
return ResponseEnvelope::error(uuid::Uuid::new_v4().to_string(), error);
}
let request_id = uuid::Uuid::new_v4().to_string();
if let Some(error) = self.ownership_wiring_refusal(&operation_name) {
return ResponseEnvelope::error(request_id, error);
}
let context = self.build_root_context(&request_id, &operation_name, identity);
let fut = self.registry.invoke(&operation_name, input, context);
match self.deadline {
@@ -141,6 +189,11 @@ impl GatewayDispatch {
}));
}
let request_id = uuid::Uuid::new_v4().to_string();
if let Some(error) = self.ownership_wiring_refusal(&operation_name) {
return Box::pin(futures::stream::once(async move {
ResponseEnvelope::error(request_id, error)
}));
}
let context = self.build_root_context_streaming(&request_id, &operation_name, identity);
self.registry
.invoke_streaming(&operation_name, input, context)
@@ -161,6 +214,9 @@ impl GatewayDispatch {
) -> ResponseEnvelope {
let operation_name = strip_leading_slash(op).to_string();
let request_id = uuid::Uuid::new_v4().to_string();
if let Some(error) = self.ownership_wiring_refusal(&operation_name) {
return ResponseEnvelope::error(request_id, error);
}
let context = self.build_root_context_sink(&request_id, &operation_name, identity);
let fut = self
.registry
@@ -179,6 +235,15 @@ impl GatewayDispatch {
}
}
fn ownership_wiring_refusal(&self, operation: &str) -> Option<CallError> {
ownership_wiring_refusal(
&self.registry,
operation,
self.ownership_provider.as_ref(),
self.static_ownership_fallback,
)
}
fn build_root_context_sink(
&self,
request_id: &str,
@@ -242,7 +307,7 @@ impl GatewayDispatch {
env,
abort_policy: AbortPolicy::default(),
internal: false,
ownership: None,
ownership: self.ownership_provider.clone(),
}
}
}
@@ -272,6 +337,43 @@ fn schema_via_call_denial(
schema_disclosure_denial(registry, name, identity)
}
/// The loud ownership-wiring default (review 012's ruling, ADR-048
/// amendment): when the dispatched op's `access_control` gates on a
/// `resource_type` and no ownership provider is wired on the spine,
/// refuse with `FORBIDDEN` rather than silently applying the ADR-011
/// static `Identity.resources` fallback — admission-by-claim at a door
/// is exactly the visible-surface ≠ authorized-surface drift. The
/// predicate is `resource_type` (not `resource_id_path`): `AccessControl::check`
/// consults the provider whenever the resource type is set, including
/// the `list` case (`owns_any`) where `resource_id_path` is `None`.
/// `resource_action`-only ACLs are exempt — no provider is consulted
/// for those even on the native wire, so static resolution there is
/// family-wide semantics, not gateway drift. Internal-visibility ops
/// are exempt too (the registry's dispatch answers `NOT_FOUND`; the
/// refusal must not acknowledge a never-visible op). The
/// [`GatewayDispatch::with_static_ownership_fallback`] declaration is
/// the deliberate escape hatch.
fn ownership_wiring_refusal(
registry: &OperationRegistry,
operation: &str,
ownership_provider: Option<&Arc<dyn OwnershipProvider>>,
static_ownership_fallback: bool,
) -> Option<CallError> {
if ownership_provider.is_some() || static_ownership_fallback {
return None;
}
let spec = &registry.registration(operation)?.spec;
if spec.visibility == Visibility::Internal {
return None;
}
let resource_type = spec.access_control.resource_type.as_deref()?;
Some(CallError::forbidden(format!(
"operation '{operation}' gates on resource ownership ({resource_type}), but no ownership \
provider is wired on this dispatch spine; the deployment must wire \
GatewayDispatch::with_ownership_provider or declare with_static_ownership_fallback"
)))
}
/// The schema-disclosure check shared by every transport surface:
/// Internal ops are invisible (`NOT_FOUND` regardless of caller);
/// ACL-forbidden ops are denied with `FORBIDDEN`. One implementation so
@@ -784,4 +886,433 @@ mod tests {
"default-ACL ops are fetchable unauthenticated"
);
}
// --- Review 012 / ADR-048 amendment: ownership threading ------------
use crate::core::ownership::{InMemoryOwnershipStore, OwnershipStore};
fn namespace_acl() -> AccessControl {
AccessControl {
resource_type: Some("blob-namespace".to_string()),
resource_action: Some("read".to_string()),
..Default::default()
}
}
fn namespace_registry() -> Arc<OperationRegistry> {
let registry = OperationRegistry::new();
registry
.register(HandlerRegistration::new(
namespace_spec("ns/stat"),
HandlerKind::Once(make_handler(|input, ctx| async move {
ResponseEnvelope::ok(ctx.request_id, input)
})),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
Arc::new(registry)
}
fn namespace_spec(name: &str) -> OperationSpec {
let mut spec = spec(name, Visibility::External, OperationType::Query);
spec.access_control = namespace_acl();
spec.resource_id_path = Some("$.namespace".to_string());
spec
}
fn identity_with_static_claim(id: &str, claimed: bool) -> Identity {
let mut resources = HashMap::new();
if claimed {
resources.insert("blob-namespace".to_string(), vec!["read".to_string()]);
}
Identity {
id: id.to_string(),
scopes: Vec::new(),
resources,
}
}
// Review 012's ruling replaces the pre-ruling pin
// (`gateway_without_provider_admits_only_the_static_claim`): the
// loud default refuses the unwired spine (see
// `gateway_refuses_a_resource_gated_op_without_a_provider`), the
// opt-out keeps the fallback (see
// `gateway_static_fallback_opt_out_serves_resource_gated_ops`),
// and the wired provider keeps the live-table behavior below.
#[tokio::test]
async fn gateway_with_provider_consults_the_live_grant_table() {
let store = Arc::new(InMemoryOwnershipStore::new());
store
.record(
&identity_with_static_claim("tenant-a", false),
"blob-namespace",
"ns-1",
)
.await
.unwrap();
let dispatch = GatewayDispatch::new(namespace_registry())
.with_ownership_provider(Arc::clone(&store) as Arc<dyn OwnershipProvider>);
let input = serde_json::json!({ "namespace": "ns-1" });
let envelope = dispatch
.invoke(
Some(identity_with_static_claim("tenant-a", false)),
"ns/stat",
input.clone(),
)
.await;
assert!(
envelope.result.is_ok(),
"a live grant admits without any static claim on the identity, got {envelope:?}"
);
let envelope = dispatch
.invoke(
Some(identity_with_static_claim("token-carrier", true)),
"ns/stat",
input.clone(),
)
.await;
match envelope.result {
Err(error) => assert_eq!(error.code, "FORBIDDEN"),
Ok(v) => panic!("a wired provider must override the static fallback, got {v:?}"),
}
store.revoke("blob-namespace", "ns-1").await.unwrap();
let envelope = dispatch
.invoke(
Some(identity_with_static_claim("tenant-a", false)),
"ns/stat",
input,
)
.await;
match envelope.result {
Err(error) => {
assert_eq!(
error.code, "FORBIDDEN",
"revocation is live on the gateway surface, got {error:?}"
);
}
Ok(v) => panic!("revocation must apply on the gateway surface, got {v:?}"),
}
}
#[tokio::test]
async fn gateway_streaming_with_provider_denies_a_static_claim_only_identity() {
let registry = OperationRegistry::new();
let mut stream_spec = namespace_spec("ns/stream");
stream_spec.op_type = OperationType::Sub;
registry
.register(HandlerRegistration::new(
stream_spec,
HandlerKind::Stream(make_streaming_handler(|_input, ctx| {
let request_id = ctx.request_id.clone();
futures::stream::once(async move {
ResponseEnvelope::ok(request_id, serde_json::json!({}))
})
})),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let dispatch = GatewayDispatch::new(Arc::new(registry))
.with_ownership_provider(Arc::new(InMemoryOwnershipStore::new()));
let mut stream = dispatch.invoke_streaming(
Some(identity_with_static_claim("token-carrier", true)),
"ns/stream",
serde_json::json!({ "namespace": "ns-1" }),
);
let envelopes: Vec<ResponseEnvelope> = stream.by_ref().collect().await;
assert_eq!(
envelopes.len(),
1,
"the denied op yields one error envelope"
);
match &envelopes[0].result {
Err(error) => assert_eq!(error.code, "FORBIDDEN"),
Ok(v) => panic!("the provider gate applies to the streaming path too, got {v:?}"),
}
}
// --- Review 012 ruling: the loud ownership-wiring default ------------
#[tokio::test]
async fn gateway_refuses_a_resource_gated_op_without_a_provider() {
let dispatch = GatewayDispatch::new(namespace_registry());
let envelope = dispatch
.invoke(
Some(identity_with_static_claim("token-carrier", true)),
"ns/stat",
serde_json::json!({ "namespace": "ns-1" }),
)
.await;
match envelope.result {
Err(error) => {
assert_eq!(error.code, "FORBIDDEN");
assert!(
error.message.contains("with_static_ownership_fallback"),
"the refusal names the deliberate opt-out, got {error:?}"
);
assert!(
error.message.contains("blob-namespace"),
"the refusal names the resource type, got {error:?}"
);
}
Ok(v) => panic!(
"the loud default must refuse a resource-gated op with no wired provider, got {v:?}"
),
}
}
#[tokio::test]
async fn gateway_streaming_and_sink_refuse_a_resource_gated_op_without_a_provider() {
let registry = OperationRegistry::new();
let registry = {
let mut sink_spec = namespace_spec("ns/put");
sink_spec.op_type = OperationType::Pub;
let mut stream_spec = namespace_spec("ns/stream");
stream_spec.op_type = OperationType::Sub;
registry
.register(HandlerRegistration::new(
stream_spec,
HandlerKind::Stream(make_streaming_handler(|_input, ctx| {
let request_id = ctx.request_id.clone();
futures::stream::once(async move {
ResponseEnvelope::ok(request_id, serde_json::json!({}))
})
})),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
registry
.register(HandlerRegistration::new(
sink_spec,
HandlerKind::Sink(make_sink_handler(|_input, ctx, _chunks| async move {
ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
})),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
Arc::new(registry)
};
let dispatch = GatewayDispatch::new(Arc::clone(&registry));
let mut stream = dispatch.invoke_streaming(
Some(identity_with_static_claim("token-carrier", true)),
"ns/stream",
serde_json::json!({ "namespace": "ns-1" }),
);
let envelopes: Vec<ResponseEnvelope> = stream.by_ref().collect().await;
assert_eq!(envelopes.len(), 1, "the refusal is the only envelope");
match &envelopes[0].result {
Err(error) => assert_eq!(error.code, "FORBIDDEN"),
Ok(v) => panic!("the streaming path refuses unwired resource-gated ops, got {v:?}"),
}
let chunks: crate::registry::registration::PublishStream =
Box::pin(futures::stream::empty());
let envelope = dispatch
.invoke_sink(
Some(identity_with_static_claim("token-carrier", true)),
"ns/put",
serde_json::json!({ "namespace": "ns-1" }),
chunks,
)
.await;
match envelope.result {
Err(error) => assert_eq!(error.code, "FORBIDDEN"),
Ok(v) => panic!("the sink path refuses unwired resource-gated ops, got {v:?}"),
}
}
#[tokio::test]
async fn gateway_list_case_resource_gated_ops_are_refused_unwired_and_served_via_owns_any() {
let registry = OperationRegistry::new();
let mut list_spec = spec("ns/list", Visibility::External, OperationType::Query);
list_spec.access_control = namespace_acl();
registry
.register(HandlerRegistration::new(
list_spec,
HandlerKind::Once(make_handler(|input, ctx| async move {
ResponseEnvelope::ok(ctx.request_id, input)
})),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let registry = Arc::new(registry);
let envelope = GatewayDispatch::new(Arc::clone(&registry))
.invoke(
Some(identity_with_static_claim("token-carrier", true)),
"ns/list",
serde_json::json!({}),
)
.await;
match envelope.result {
Err(error) => assert_eq!(
error.code, "FORBIDDEN",
"the list case (resource_id_path None) is refused unwired too, got {error:?}"
),
Ok(v) => panic!(
"a token-carried static claim must not admit the list case on an unwired spine, \
got {v:?}"
),
}
let store = Arc::new(InMemoryOwnershipStore::new());
store
.record(
&identity_with_static_claim("tenant-a", false),
"blob-namespace",
"ns-9",
)
.await
.unwrap();
let envelope = GatewayDispatch::new(Arc::clone(&registry))
.with_ownership_provider(Arc::clone(&store) as Arc<dyn OwnershipProvider>)
.invoke(
Some(identity_with_static_claim("tenant-a", false)),
"ns/list",
serde_json::json!({}),
)
.await;
assert!(
envelope.result.is_ok(),
"the wired spine serves the list case through owns_any, got {envelope:?}"
);
}
#[tokio::test]
async fn gateway_static_fallback_opt_out_serves_resource_gated_ops() {
let dispatch = GatewayDispatch::new(namespace_registry()).with_static_ownership_fallback();
let envelope = dispatch
.invoke(
Some(identity_with_static_claim("token-carrier", true)),
"ns/stat",
serde_json::json!({ "namespace": "ns-1" }),
)
.await;
assert!(
envelope.result.is_ok(),
"the declared static fallback serves the op through the ADR-011 fallback, got {envelope:?}"
);
let envelope = dispatch
.invoke(
Some(identity_with_static_claim("stranger", false)),
"ns/stat",
serde_json::json!({ "namespace": "ns-1" }),
)
.await;
match envelope.result {
Err(error) => assert_eq!(error.code, "FORBIDDEN"),
Ok(v) => panic!("the static fallback still denies a claimless caller, got {v:?}"),
}
}
#[tokio::test]
async fn gateway_refusal_spares_action_only_internal_and_unknown_ops() {
let registry = OperationRegistry::new();
let mut action_only = spec("act/run", Visibility::External, OperationType::Query);
action_only.access_control = AccessControl {
resource_action: Some("read".to_string()),
..Default::default()
};
let mut internal_gated = namespace_spec("internal/hidden");
internal_gated.visibility = Visibility::Internal;
registry
.register(HandlerRegistration::new(
action_only,
HandlerKind::Once(make_handler(|input, ctx| async move {
ResponseEnvelope::ok(ctx.request_id, input)
})),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
registry
.register(HandlerRegistration::new(
internal_gated,
HandlerKind::Once(make_handler(|_input, ctx| async move {
ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
})),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let dispatch = GatewayDispatch::new(Arc::new(registry));
let mut action_identity_resources = HashMap::new();
action_identity_resources.insert("some-resource".to_string(), vec!["read".to_string()]);
let envelope = dispatch
.invoke(
Some(Identity {
id: "action-holder".to_string(),
scopes: Vec::new(),
resources: action_identity_resources,
}),
"act/run",
serde_json::json!({}),
)
.await;
assert!(
envelope.result.is_ok(),
"a resource_action-only ACL never consults a provider on the wire either — no refusal, got {envelope:?}"
);
let envelope = dispatch
.invoke(
Some(identity_with_static_claim("snooping", false)),
"internal/hidden",
serde_json::json!({ "namespace": "ns-1" }),
)
.await;
match envelope.result {
Err(error) => {
assert_eq!(
error.code, "NOT_FOUND",
"an Internal op stays unacknowledged instead of the refusal"
);
assert!(
!error.message.contains("with_static_ownership_fallback")
&& !error.message.contains("blob-namespace"),
"the refusal text must not leak on an Internal op, got {error:?}"
);
}
Ok(v) => panic!("an Internal op must never run, got {v:?}"),
}
let envelope = dispatch
.invoke(
Some(identity_with_static_claim("anyone", false)),
"missing/op",
serde_json::json!({}),
)
.await;
match envelope.result {
Err(error) => assert_eq!(error.code, "NOT_FOUND"),
Ok(v) => panic!("an unknown op keeps its NOT_FOUND, got {v:?}"),
}
}
}