Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a99cc67e76 | ||
|
|
4dbd4bc583 | ||
|
|
56e0f135d7 | ||
|
|
1a1e4eb7c0 | ||
|
|
1fda0d8587 | ||
|
|
9899b37a82 | ||
|
|
d0b8124064 | ||
|
|
03a6689c99 |
No files matched your search
+125
@@ -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
@@ -27,7 +27,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alkcall"
|
||||
version = "0.9.0"
|
||||
version = "0.10.1"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"bytes",
|
||||
|
||||
+5
-2
@@ -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"] }
|
||||
@@ -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).
|
||||
Generated
+1
-1
@@ -27,7 +27,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alkcall"
|
||||
version = "0.8.1"
|
||||
version = "0.10.1"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"bytes",
|
||||
|
||||
+320
-43
@@ -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
@@ -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 = ®istry.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(®istry));
|
||||
|
||||
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(®istry))
|
||||
.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(®istry))
|
||||
.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:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user