Repository navigation
feat: MCP over Fabric follow-up items - #31
anton-i-affinidi wants to merge 7 commits into
Conversation
- move the reader's account.not_found decision and envelope conversion into functions - unit-test them with a problem report built as the mediator sends it
- check each connected connection point's own mediator account every 30 s - restart the listener when the account is gone, re-open a closed receive list - share the account probe and its decision with the gateway ping timeout path - g2g scenario: a gateway that only receives is reachable after a mediator store flush
- per-peer and per-tenant shares of the capability offer, probe and open replay tables - admit opens off the reader loop, holding early frames until admission ends - keep each concurrent first request's own capability agreement - answer full receiver caps and the open rate with capacity_reached (429)
- arrange the mediator store flush as a given, so the scenario passes the bdd lint
- g2g scenario: a modern sender to a legacy-only receiver fails fast with -32022 - one replay proof test for open and capability-query replay through prepare_incoming - document the shared identity pepper and per-node oauth state checks for multiple nodes
- gateway and connection point writes end the streams a narrowed access no longer allows - a peer leaving active or removed ends its streams in both directions - refuse an open admitted while a revocation runs, and admit against the current connection point
mohamed-affinidi
left a comment
There was a problem hiding this comment.
Review of a39b121: two code passes, a design/business pass and a behaviour pass. Requesting changes. Details are inline; one finding sits outside the diff and is listed below.
Blocking
- Held early-frame entries leak when an admission is aborted by a listener restart (inline,
ws_listener.rs:2319). This is the one to fix first. - The revocation-race, admission-timeout, CP re-read and fail-closed store paths have no tests (inline,
ws_listener.rs:2435). - Revocation runs only on this node's own store writes, while the docs and changelog say "every write" (inline,
gateways/filesystem.rs:84). - Outside this diff: the release notes will contradict themselves.
.changelog/wip/fix_fabric-stream-cap-429.yaml(#24, unreleased, same release) still says: "Receivers still refuse a full cap withunavailable(502), because a 1.0.0 sender cannot parsecapacity_reached...". This PR'sbreakingentry says the opposite. Please reword or remove that sentence in this PR.
Non-blocking: the minor comments inline. On your question about the wire change, see the comment on fabric_stream/mod.rs:553. In short, it's acceptable with upgrade-order guidance, or the receiver flip can wait one release.
Looks sound:
- SharedTable accounting on every insert, remove and retain path;
- revocation ordering (store write, then a SeqCst counter bump before the sweep);
- held frames are checked against the sender, bounded, and delivered in order;
- base-DID mediator matching is unchanged, with spoof tests added;
- the Gateway Ping refactor preserves behaviour;
- the watch is idle in Standby;
AG_BDD_FABRIC_RECEIVE_LEGACY_ONLYis honoured only underAG_TEST_MODE=trueand only narrows admission.
Validation:
- Run at a39b121: filtered
cargo test --bin agent-gatewaycovering fabric_stream, revocation, share, account_watch, open_admission and held frames, about 100 tests, all passed. - Not run: the full suite, clippy, fmt, gw-e2e and the g2g BDD scenarios.
- CI on this SHA ran only CodeQL, so the Rust build and tests aren't covered by CI here.
| .from_did | ||
| .as_deref() | ||
| .ok_or("Fabric sender is missing")?; | ||
| registry.hold(stream_id, peer_did)?; |
There was a problem hiding this comment.
[important] Held early-frame entries leak when an admission is aborted
hold() inserts into the process-global ReceiveRegistry.held map, and only release() inside the admission task removes it. The task lives in the listener-local stream_tasks JoinSet. On a listener stop or restart, which the new account watch now triggers unattended, the JoinSet is dropped and the task is aborted mid-admit(). Nothing else cleans up: cancel_listener / cancel_where (registry.rs:858) touch only senders and entries, and held has no expiry.
Each restart during admission therefore orphans up to 16 entries, each with up to 8 cloned frames, for the life of the process, and was_opened() stays true for those ids. A panic in admission leaks the same way.
Fix: hold through a guard that calls release(&stream_id, false) on Drop unless disarmed, or store the listener instance in HeldOpen and clear it in cancel_listener. Add a test that aborts the JoinSet mid-admission and asserts !was_opened(id).
| .revocations() | ||
| != revocations | ||
| { | ||
| return Err("Fabric access changed while the Open was admitted".into()); |
There was a problem hiding this comment.
[important] The revocation-race and admission security paths have no tests
Deleting this guard fails no test. open_admission_runs_off_the_reader_loop_and_is_bounded drives stub closures, so none of these are exercised:
OpenAdmission::admit: this counter mismatch,current_connection_pointre-reading a disabled or removed CP, and the 2 s timeout →release(false);- the outbound counter check at
fabric_forward.rs:330; - the store hooks'
Err(previous)fail-closed arms; - gateway and CP delete revocation;
check_listener'sRestartListener/ReopenReceiveListbranches, which run only in the Docker g2g scenario.
AGENTS.md rule 14 asks for positive and negative cases.
Suggested tests through OpenAdmission with a stub ConnectionPointStore:
get()callsrevoke(...): the Open is refused and nothing stays registered;get()returnsenabled: false: the Open is refused;get()sleeps pastOPEN_ADMISSION_TIMEOUT: the held frames are dropped.
Also add an outbound race test, store tests with unreadable JSON and for delete, and a check_listener test with a stubbed probe outcome.
| /// Saves `gateway` and ends the running Fabric streams the change revokes. | ||
| /// When the previous record cannot be read, every stream with the peer | ||
| /// ends. | ||
| async fn save_and_revoke( |
There was a problem hiding this comment.
[important] Revocation runs only on writes through this node's store, but the docs and changelog say "every write"
revoke() is called only from save_and_revoke and the two delete methods. These paths update the cache without revoking:
refresh_from_disk(cache_refresh_interval_secs). It emitsStorageEvent::{Upsert,Delete}{source: RemoteWrite}, but only the surface store subscribes to those events;- a manual
_storage/edit plusPOST /v1/config/reload(AGENTS.md rule 12); - a restore;
- a write made on another node sharing storage.
On these paths, running streams for a peer that has been disabled or narrowed keep going. Admission re-reads the record, so new Opens are refused.
docs/FABRIC.md says The gateway and Connection Point stores apply it on every write of a record, and the security changelog entry says Every write of a gateway or Connection Point record now ends, at once, ....
Question: can more than one node be Active with Fabric listeners at once? Either subscribe to gateway and CP RemoteWrite events and revoke what the refreshed record no longer admits, or narrow FABRIC.md and the changelog to "writes through this node's management API". In the second case, also say that refreshed, reloaded, restored or other-node changes only block new Opens.
| self.storage | ||
| .delete(id) | ||
| .await?; | ||
| if let Ok(Some(previous)) = &previous |
There was a problem hiding this comment.
[minor] Gateway delete doesn't fail closed when the previous record can't be read
Gateway save, CP save and CP delete all revoke on Err(previous). Gateway delete ignores it, so the delete succeeds and the peer's running streams continue. That contradicts the PR's "fails closed when the previous record can't be read". New Opens are still refused. With only the id at hand there's no DID to revoke by.
Options:
- revoke by a DID read from the cache;
- revoke every inbound stream on
Err; - narrow the claim to exclude gateway delete.
Refusing the delete seems wrong, since operators need to be able to remove a corrupt record. Please add a test for whichever you choose.
| return None; | ||
| } | ||
| match current { | ||
| Some(current) if current.enabled => (current.exposed_channels != previous.exposed_channels) |
There was a problem hiding this comment.
[minor] Any exposure difference, even a widening or a reorder, refuses every in-flight Open on the node
exposed_channels is compared as a Vec, so adding a surface or reordering the list also yields Some. revoke() then bumps the appliance-wide counter. Two things fail as a result:
- every Open being admitted at that moment, for any peer and through any CP;
- every outbound first request inside its probe window (
fabric_forward.rs:330, which returns 502).
Inbound, the replay record is already in opened, so refused_open_binding returns None. The unrelated sender never gets an answer and waits for its deadline. The doc comment "Only a change that narrows its access revokes them" doesn't match this.
Fix: return Some only when the new set drops a surface the old set admitted (an empty list means all), and add widening and reorder cases to only_a_change_that_narrows_access_revokes_streams. The same applies to line 825.
| cap = limit.as_str(), | ||
| "Fabric capability offer cap reached" | ||
| ); | ||
| "Fabric capability offer capacity is unavailable".to_string() |
There was a problem hiding this comment.
[minor] A full per-peer offer share shows up as a 5 s stall and then 502, not 429
The per-peer offer share is MAX_OFFERS / 8 = 32. Offers live for PEER_TTL (300 s) and aren't removed on use. Senders don't single-flight first probes, so a cold burst of more than 32 concurrent first requests to one receiver exhausts the share, and it stays full for up to 5 minutes.
The refused query yields ProcessingResult::Failed, which is only logged, so no reply is sent. The sender waits min(deadline, 5 s) in peers.wait and returns StreamingUnavailable (502). That is not the 429 the PR describes for capacity refusals.
Fix (any of):
- single-flight the sender's probe per
PeerKey; - answer a capacity-refused query so the sender maps it to 429;
- size the share for the expected burst.
| peer_did: binding.peer_did.clone(), | ||
| tenant_id, | ||
| }; | ||
| if opened.admits(&owner).is_err() { |
There was a problem hiding this comment.
[minor] Question: does the per-peer replay share cap a busy single peer?
One peer's eighth of the table is 128 × 512 / 8 = 8,192 records. A record lives up to the offer expiry (≤ 300 s), so a single busy peer reaches its share at about 27–55 Opens/s sustained, below the 50/s rate limit, and then gets 429s.
In a deployment with one paired peer, 7/8 of the table is unreachable. Is that intended, given the OPEN_REPLAY_RECORDS_PER_STREAM comment "bounds memory without capping ordinary traffic"?
If not intended, options:
- raise the per-peer share to at least rate × offer TTL;
- enforce shares only above a fill threshold;
- document the cap.
| fn from(error: registry::RegisterError) -> Self { | ||
| match error { | ||
| registry::RegisterError::CapacityReached(reason) => reason.into(), | ||
| registry::RegisterError::CapacityReached(reason) => { |
There was a problem hiding this comment.
[minor] On the capacity_reached wire change you asked about: acceptable, but please add upgrade guidance
Per-peer negotiation isn't cheap. StreamCapabilities is deny_unknown_fields with version: 1, so a new flag would itself break 1.0.0, and gateway records carry no software version. Released 1.0.0 has no CapacityReached; sender-side parsing arrived in #24, which is unreleased. So the breaking label is accurate.
The cost is fail-slow, not data loss. Under receiver overload, a 1.0.0 sender's callers wait for their deadline instead of getting a fast 502, and hold the sender's own stream slots meanwhile. This covers full caps, the replay share, the new 16-slot admission limit, and the per-peer Open rate. The changelog's "upgrade paired gateways" doesn't describe what happens while a pair runs mixed versions, or what order to upgrade in.
Options:
- (a) Keep the flip, and say in the
breakingentry and docs/FABRIC.md what happens while a pair runs mixed versions and that both sides should be upgraded together. - (b) Ship the shares, revocation and watch now, and flip
From<RegisterError> for OpenRefusalnext release, once every sender parses it.
Either way, consider keeping the rate-limit and admission-slot refusals on unavailable, since they happen before any offer lookup.
| | Opens per peer | 50/s, burst 100, per listener | The Open is refused with `capacity_reached` before any lookup; the caller gets `429` | | ||
| | Opens being admitted | 16 at a time per listener, off the listener's reader loop, each for up to 2 s | The Open is refused with `capacity_reached`; the caller gets `429` | | ||
| | Frames held per Open being admitted | 8, from the Open's sender only | The admitted stream is dropped, and its caller waits for the response deadline | | ||
| | Open replay records | 512 per allowed stream (65,536); one peer holds at most an eighth of them and the peers of one tenant half | The Open is refused with `capacity_reached`; the caller gets `429` | |
There was a problem hiding this comment.
[minor] The tenant share isn't documented for the appliance-global and single-tenant cases
Peers with tenant_id: None count against no tenant share. share.rs says so, but this table and FABRIC.md don't.
The reverse case is a question. If every Remote gateway is created under one tenant-scoped PAT, all peers share one tenant and together get only half of each table, for no fairness gain. Is that real in practice? If so, consider applying the tenant share only when more than one tenant holds entries.
| security: | ||
| - timestamp: '2026-10-09T12:30:00.000Z' | ||
| title: ⭐ Give each Fabric peer and tenant its own share of the shared stream tables | ||
| description: One peer could fill the capability offer, capability probe or Open replay table and get every other peer refused. Each peer may now hold an eighth of a table and the peers of one tenant half of it, so a second peer is still served after one peer sends hundreds of refused requests. The listener also admits Opens off its reader loop, at most 16 at a time, holding up to 8 early frames per stream until its Open is admitted, so a slow admission no longer stalls every other message. The share values are interim constants. |
There was a problem hiding this comment.
[minor] Two changelog nits on this entry
- "The share values are interim constants." is temporary-plan language in a durable file (AGENTS.md rule 7), and it doesn't tell operators anything they can act on. Drop it, or say the values are fixed and not configurable.
- Neither this entry nor the
breakingone mentions that a full capability-probe share or table on the sender now answers the caller 429 withRetry-After: 5instead of 502 (fabric_forward.rs:214,.map_err(registration_failure)?). MCP_METADATA.md documents it. A clause here would cover it.
Problem
Review of modern MCP over Fabric left a set of follow-up items: transport correctness, tests and cleanup around Connection Points and Fabric streams. This PR delivers some of them, one commit per item.
Subtasks
Test mediator problem-report handling with a real envelope
When the mediator reports that it no longer has a Connection Point's account (
e.p.account.not_found), the Connection Point must drop the stale socket and reconnect to re-register. The sender was matched to the mediator by base DID, but no test used a real mediator envelope.account.not_founddecision intomediator_account_loss, and the envelope-to-record conversion intoreceived_message, so both can be unit-tested. Behaviour is unchanged.ProblemReportbody, mediator DID asfrom, session DID asto,pthid):#fragment;e.p.authorization.account.denied), and a non-report message: the socket stays.Reconnect a Connection Point whose account the mediator has lost
When the mediator's store is flushed while the gateways stay connected, mediator v0.33.1 does not send
e.p.account.not_foundto the Connection Points. It recreates an account only when that account sends. It refuses delivery to a missing account withe.p.delivery.refused, which is also its answer to an access-list refusal. A gateway that only receives therefore stayed unreachable, with its connection still shown as healthy, until it sent something, restarted or was reconnected by hand.connection_points::account_watch: every 30 s, each connected Connection Point probes its own mediator account (account_get).e.p.account.not_found: the listener restarts, so it reauthenticates and reregisters.gateways/handlers.rsare removed, and its behaviour is unchanged.docs/FABRIC.md(Runtime health behavior) describes the periodic check.Give each Fabric peer and tenant its own share of the shared stream tables
One peer could fill the capability offer table, the capability probe table or the Open replay table, and every other peer was then refused. Open admission (the peer lookup and route resolution) also ran on the Connection Point reader loop, so one slow admission stalled every other message.
fabric_stream::share::SharedTable: counts each table's entries per peer and per tenant (the tenant owning the peer's Remote gateway record). One peer may hold an eighth of a table and the peers of one tenant half. These are interim constants until the final per-peer limits are set.429.capacity_reached, so the caller gets429withRetry-After: 5instead of502. A 1.0.0 sender cannot parsecapacity_reachedand waits for its response deadline instead, so paired gateways need upgrading.docs/MCP_METADATA.md(refusal codes, Framed Fabric limits, cap logging) anddocs/FABRIC.mddescribe the shares, the admission and the new refusal code.Add Fabric compatibility and replay proof tests and the multi-node note
mcp_over_fabric.feature.AG_BDD_FABRIC_RECEIVE_LEGACY_ONLY=true, which is read only whenAG_TEST_MODE=true, like the otherAG_BDD_*switches. No product setting is added.400and-32022listing only2024-11-05, in under 5 s, and the MCP server is not called.prepare_incoming, with both envelope times set as the current sender sets them.docs/CAPABILITIES.md(High availability) and theAG_IDENTITY_HASH_PEPPERrow ofdocs/CONFIGURATION_REFERENCE.mdsay to set the same pepper on every node. The High availability section also says used OAuthstatenonces are checked per node, not across nodes.End running Fabric streams when a peer's access is revoked
Running framed streams kept going after a peer left active or was removed, a surface was un-exposed, or a Connection Point was disabled. A disabled Connection Point also still admitted new streams until its listener restarted, because admission read the record the listener started with.
fabric_stream::Revocationsays which streams an access change ends:unavailableto each side).cancel_listenernow shares the same registry code (cancel_where).docs/FABRIC.md(Peer exposure, "Revoking access ends running streams") describes it.all, since new peers start withnone.Notes for reviewers
One wire change, on an exceptional path only. Normal traffic never sees it. It applies only when the receiving gateway is out of room:
In those cases the receiver now refuses with
capacity_reachedinstead ofunavailable, so the caller gets429withRetry-After: 5instead of502. A 1.0.0 sender can't parsecapacity_reachedand waits for its response deadline instead. That's why the changelog lists it underbreaking. Please confirm that's acceptable for the 1.0.0 peers still paired.Merge order: fix: per-route opt-in for delegated credentials over Fabric #32 also edits
docs/FABRIC.mdanddocs/MCP_METADATA.md. Merging this PR first leaves the doc merge on fix: per-route opt-in for delegated credentials over Fabric #32's side, which is small.Testing
cargo test --no-fail-fast --all-targets --all-features, with main (fix: make fabric exposure explicit for remote gateways #28, fix: stream connection guard #29) merged in: 4,537 passed, 0 failed (15 ignored).cargo clippy --all-targets --all-features -- -D warnings,cargo fmt --all -- --checkandmake pre-commit-rust: clean.New unit tests:
capacity_reached;Component tests for stream revocation, one per action, through the real gateway and Connection Point stores. Each shows the affected streams end with
unavailablewhile other peers', surfaces' and Connection Points' streams keep running:none;All three fail when revocation is turned off.
g2g scenario for the mediator account loss, against mediator v0.33.1:
e.p.delivery.refused.e.p.account.not_foundand its Connection Points restarted about 30 s after the flush. The next ping round trip took 12 ms.scripts/bdd-gherkin.py --lint: no errors.make gw-e2e(real mediators, 8 scenarios in parallel), with main merged in and stream revocation added: 47 scenarios, 383 steps, all passed, including the legacy-only receiver scenario. In the previous round's run the issuer-DID request ingateway_did_validation.featuretimed out once (504). That feature passed in all 3 re-runs and in this run. Its exchange does not use framed streams.