docs(arch): hub-wiring cluster — aggregated env, peer_operations, from_call cleanup, alknet-hub crate spec
Three ADRs addressing gaps surfaced by the first hub consumer (alkapi): - ADR-067: Aggregated peer-env wiring — Dispatcher::with_aggregated_env hook so compose_root_env reads a shared PeerCompositeEnv across all calls, not a fresh per-call one. The hub-defining gap (alkapi OQ-08 / G.1). - ADR-068: PeerCompositeEnv::peer_operations override — adds list_operation_names() to OperationEnv, overrides on OverlayOperationEnv and PeerCompositeEnv. Fixes services/list-peers returning empty operation lists for non-local peers (alkapi G.6). - ADR-069: from_call is a manual free function, not auto-wired — reverses the aspirational OQ-27 resolution to match the implementation. Cleans the 'v1 default' hedging language in ADR-017 and client-and-adapters.md (alkapi G.4). New crate spec: alknet-hub — reusable hub pattern (aggregated env, connection lifecycle, worker supervision with backoff, service discovery). Spec fixes: builder API drift (with_local/with_local_streaming separation), ScopedOperationEnv → ScopedPeerEnv type name, OQ count 51→54. Three new OQs: OQ-52 (wait_for_close), OQ-53 (backoff defaults), OQ-54 (inbound hook placement).
This commit is contained in:
glm-5.2 committed
1 parent
2282647f8b
commit
87b2c2a5ef
15 files changed
+1052
-47
No files matched your search
@@ -58,7 +58,7 @@ Structured RPC: operations, request/response, streaming subscriptions, and servi
|
||||
| OQ-19 | Session-scoped operation registries | resolved | Agent-written operations overlaid on curated registry via `OperationEnv` trait layering. Protocol doesn't need changes; `OperationEnv` must remain a trait. Generalized by ADR-024 to cover connection-scoped overlays. |
|
||||
| OQ-25 | ~~Remote-safe marking shape~~ | **dissolved** (ADR-029) | `remote_safe`/`trusted_peer` retired; peer authorization is `AccessControl::check(peer_identity)` |
|
||||
| OQ-26 | OperationAdapter error type (AdapterError variants) | **resolved** | `DiscoveryFailed`, `SchemaParse`, `Transport`, `Unauthorized`, `SamePeerCollision`; `#[non_exhaustive]` |
|
||||
| OQ-27 | from_call re-import trigger | **resolved** | Auto-re-import on connection establishment; `refresh()` is a feature addition |
|
||||
| OQ-27 | from_call re-import trigger | **resolved** | `from_call` is a manual free function; the assembly layer calls it after `connect()`. `refresh()` is a genuine feature addition. See ADR-069. |
|
||||
| OQ-28 | from_call namespace collision | **resolved** | Same-peer collision = error; cross-peer dissolved by ADR-029 (separate sub-overlays) |
|
||||
| OQ-29 | CallClient TLS client-auth | **resolved** | Wire quinn client-auth; key-type-aware server cert verification; fingerprint normalization |
|
||||
| OQ-30 | `PeerRef::Any` routing policy | **resolved** | Insertion-order first-match; richer routing is a feature extension |
|
||||
|
||||
@@ -474,7 +474,7 @@ fn build_root_context(
|
||||
metadata: HashMap::new(), // fresh per request
|
||||
deadline: Some(Instant::now() + self.default_timeout), // root deadline (W7)
|
||||
scoped_env: registration.scoped_env.clone()
|
||||
.unwrap_or_else(ScopedOperationEnv::empty), // from the bundle, empty for leaves
|
||||
.unwrap_or_else(ScopedPeerEnv::empty), // from the bundle, empty for leaves
|
||||
// Per-call env composition (ADR-024 + ADR-029): the root env is a
|
||||
// PeerCompositeEnv — the curated base + this connection's imported-
|
||||
// ops overlay (peer-keyed in the head's aggregation env, ADR-029 §1)
|
||||
@@ -590,8 +590,9 @@ See [open-questions.md](../../open-questions.md) for full details.
|
||||
variants (`DiscoveryFailed`, `SchemaParse`, `Transport`, `Unauthorized`,
|
||||
`SamePeerCollision`); `#[non_exhaustive]`. See
|
||||
[client-and-adapters.md](client-and-adapters.md).
|
||||
- **OQ-27** (resolved): `from_call` re-import trigger — auto-re-import on
|
||||
connection establishment. See [client-and-adapters.md](client-and-adapters.md).
|
||||
- **OQ-27** (resolved): `from_call` re-import trigger — `from_call` is a manual
|
||||
free function; the assembly layer calls it after `connect()`. See
|
||||
[ADR-069](../../decisions/069-from-call-manual-free-function.md).
|
||||
- **OQ-28** (resolved): `from_call` namespace collision — same-peer collision
|
||||
= error; cross-peer dissolved by ADR-029 (separate sub-overlays). See
|
||||
[client-and-adapters.md](client-and-adapters.md).
|
||||
|
||||
@@ -352,11 +352,14 @@ The flow (ADR-017 §3):
|
||||
4. The caller registers the bundles via
|
||||
`CallConnection::register_imported_all()`.
|
||||
|
||||
**Re-import on reconnection** (DC-2, OQ-27): `from_call` runs automatically on
|
||||
connection establishment. The overlay is per-connection (Layer 2, ADR-024), so
|
||||
a stale overlay dies with the connection; re-import on reconnect is naturally
|
||||
scoped to the new connection. This is the v1 default; explicit re-import via a
|
||||
future `CallConnection::refresh()` is additive.
|
||||
**Re-import on reconnection** (DC-2, OQ-27): `from_call` is a free function;
|
||||
the assembly layer calls it after `connect()`. The overlay is per-connection
|
||||
(Layer 2, ADR-024), so a stale overlay dies with the connection; re-import on
|
||||
reconnect is naturally scoped to the new connection. A
|
||||
`CallConnection::refresh()` method for mid-connection re-discovery is a
|
||||
genuine feature addition — non-breaking, additive — if a deployment needs
|
||||
manual re-discovery without drop-and-reconnect. See
|
||||
[ADR-069](../../decisions/069-from-call-manual-free-function.md).
|
||||
|
||||
**Namespace collision** (DC-3, OQ-28): under the peer-graph model (ADR-029),
|
||||
cross-peer collision dissolves — same name on different peers is fine (they
|
||||
@@ -649,8 +652,10 @@ Based on the gap analysis and the downstream unblock chain:
|
||||
holds a `PeerCompositeEnv` with `connections: HashMap<PeerId, Arc<dyn OperationEnv>>`,
|
||||
not a singular connection overlay. `invoke_peer()` routes to the right peer
|
||||
via `PeerRef::Specific` / `PeerRef::Any` (ADR-029 §1-2).
|
||||
- **`from_call` re-import is auto-on-reconnect.** v1 default; the overlay is
|
||||
per-connection so re-import is naturally scoped (DC-2, OQ-27).
|
||||
- **`from_call` is a manual free function.** The assembly layer calls it after
|
||||
`connect()`. The overlay is per-connection so re-import on reconnect is
|
||||
naturally scoped (DC-2, OQ-27). See
|
||||
[ADR-069](../../decisions/069-from-call-manual-free-function.md).
|
||||
- **`from_call` namespace collision is same-peer only.** Cross-peer collision
|
||||
dissolves (same name on different peers is fine — separate sub-overlays,
|
||||
ADR-029 §5). Same-peer collision stays an error. `namespace_prefix` is
|
||||
@@ -713,9 +718,10 @@ See [open-questions.md](../../open-questions.md) for full details.
|
||||
- **OQ-26** (resolved): `AdapterError` variants — `DiscoveryFailed`,
|
||||
`SchemaParse`, `Transport`, `Unauthorized`, `SamePeerCollision`
|
||||
(replaces flat `Conflict`). `#[non_exhaustive]`.
|
||||
- **OQ-27** (resolved): `from_call` re-import trigger — auto-re-import on
|
||||
connection establishment. `CallConnection::refresh()` is a feature
|
||||
addition, not an unmade decision.
|
||||
- **OQ-27** (resolved): `from_call` re-import trigger — `from_call` is a manual
|
||||
free function; the assembly layer calls it after `connect()`. A
|
||||
`CallConnection::refresh()` method is a genuine feature addition —
|
||||
non-breaking, additive. See [ADR-069](../../decisions/069-from-call-manual-free-function.md).
|
||||
- **OQ-28** (resolved): `from_call` namespace collision — same-peer
|
||||
collision = error; cross-peer dissolved by ADR-029 (separate sub-overlays).
|
||||
`namespace_prefix` is optional local-naming sugar.
|
||||
|
||||
@@ -243,7 +243,7 @@ pub struct OperationContext {
|
||||
/// Populated from the registration bundle's `scoped_env` (ADR-022).
|
||||
/// The reachability check in `OperationEnv::invoke()` consults
|
||||
/// `scoped_env.allows(&name)`. This is data, not a dispatch trait.
|
||||
pub scoped_env: ScopedOperationEnv,
|
||||
pub scoped_env: ScopedPeerEnv,
|
||||
/// Composition dispatch trait. A handler calls `env.invoke(...)` to
|
||||
/// compose child operations. This is `Arc<dyn OperationEnv>` (a trait
|
||||
/// object), not a concrete struct — the trait-object design is what
|
||||
@@ -303,7 +303,7 @@ impl OperationContext {
|
||||
- `forwarded_for`: The original caller when this call was forwarded by a `from_call` handler (ADR-032). **Metadata only** — `AccessControl::check` never reads it; the ACL always authorizes the direct caller's `identity`. Handlers may read it for logging, auditing, per-user rate limiting, or application context. Populated from `call.requested.forwarded_for` by the dispatch path; set to `None` for composed children (wire-ingress only). The forwarder's claim, not a verified identity — a malicious hub can lie (same property as HTTP `X-Forwarded-For`). See ADR-032.
|
||||
- `capabilities`: Outbound credentials the handler may use (decrypted API keys, scoped vault access) — see [Capability Injection](#capability-injection) below
|
||||
- `metadata`: Request-scoped context (tracing IDs, connection info). **Must not hold secret material** — see ADR-014. **Does not propagate through `OperationEnv::invoke()`** — nested calls get fresh metadata. The tracing link between parent and child is `parent_request_id`, not metadata propagation. Anything a handler needs to pass to a child goes in the call `input`.
|
||||
- `scoped_env`: The reachability set — the operations this handler may compose. Populated from the registration bundle's `scoped_env` (ADR-022). The reachability check in `OperationEnv::invoke()` consults `scoped_env.allows(&name)`. This is *data* (a `ScopedOperationEnv` struct), not a dispatch trait. `None`/empty for leaves.
|
||||
- `scoped_env`: The reachability set — the operations this handler may compose. Populated from the registration bundle's `scoped_env` (ADR-022). The reachability check in `OperationEnv::invoke()` consults `scoped_env.allows(&name)`. This is *data* (a `ScopedPeerEnv` struct), not a dispatch trait. `None`/empty for leaves.
|
||||
- `env`: The composition dispatch trait (`Arc<dyn OperationEnv + Send + Sync>`). A handler calls `context.env.invoke(...)` to compose child operations. This is a trait object, not a concrete struct — the trait-object design enables registry layering (ADR-024): the CallAdapter composes the root env per call from the active layers (curated base + connection overlay + session overlay), and overlays wrap the base via trait layering. Same pattern as `IdentityProvider` (ADR-004). See ADR-024.
|
||||
- `internal`: When `true`, this call originated from composition (a handler calling another operation via `OperationEnv`), not from a wire request. This switches the authority context: ACL runs against `handler_identity`, not `identity`. The `internal` field uses module-private construction — handlers construct `OperationContext` through `OperationEnv::invoke()` which sets `internal: true`, or through the `CallAdapter` dispatch path which sets `internal: false`. The field is not `pub` for writes; only `pub fn is_internal(&self) -> bool` is exposed for reads. See ADR-015.
|
||||
|
||||
@@ -433,24 +433,37 @@ impl CompositionAuthority {
|
||||
- `scoped_env`: The set of operations this handler may reach via `env.invoke()`. `None` for leaves (empty env). The reachability control from ADR-015.
|
||||
- `capabilities`: Outbound credentials (decrypted API keys, signing keys). Populated by the assembly layer from the vault at registration time. See [Capability Injection](#capability-injection).
|
||||
|
||||
The `OperationRegistryBuilder` provides a fluent API with convenience methods for common cases. The builder absorbs the `HandlerKind` wrapping internally — `.with_local()` and `.with_leaf()` take the raw `Handler` (or `StreamingHandler`) and wrap it in the right `HandlerKind` based on `spec.op_type` (ADR-049):
|
||||
The `OperationRegistryBuilder` provides a fluent API with convenience methods for common cases. The builder validates handler kind against `spec.op_type` at registration time — `with_local` / `with_leaf` accept `Handler` (for `Query`/`Mutation` ops), `with_local_streaming` / `with_leaf_streaming` accept `StreamingHandler` (for `Subscription` ops). Passing a `StreamingHandler` to `with_local` or a `Handler` to `with_local_streaming` is a registration-time error:
|
||||
|
||||
```rust
|
||||
// with_local: Local provenance, full bundle — all 5 args required.
|
||||
// Accepts Handler (for Query/Mutation ops). Validates op_type at registration.
|
||||
// with_local(spec, handler, composition_authority, scoped_env, capabilities)
|
||||
// The builder inspects spec.op_type and wraps in HandlerKind::Once
|
||||
// (Query/Mutation) or HandlerKind::Stream (Subscription) automatically.
|
||||
|
||||
// with_local_streaming: Local provenance, full bundle — all 5 args required.
|
||||
// Accepts StreamingHandler (for Subscription ops). Validates op_type at registration.
|
||||
// with_local_streaming(spec, streaming_handler, composition_authority, scoped_env, capabilities)
|
||||
|
||||
// with_leaf: Leaf provenance (default FromOpenAPI), no composition authority.
|
||||
// Accepts Handler (for Query/Mutation ops).
|
||||
// with_leaf(spec, handler, capabilities)
|
||||
|
||||
// with_leaf_streaming: Leaf provenance (default FromOpenAPI), no composition authority.
|
||||
// Accepts StreamingHandler (for Subscription ops).
|
||||
// with_leaf_streaming(spec, streaming_handler, capabilities)
|
||||
|
||||
// with_leaf_provenance / with_leaf_streaming_provenance: explicit provenance variant.
|
||||
let registry = OperationRegistryBuilder::new()
|
||||
// Built-in service discovery (Local, no composition — empty authority, empty env, empty caps)
|
||||
.with_local(services_list_spec(), Arc::new(services_list_handler),
|
||||
CompositionAuthority::none(), ScopedOperationEnv::empty(), Capabilities::new())
|
||||
CompositionAuthority::none(), ScopedPeerEnv::empty(), Capabilities::new())
|
||||
.with_local(services_schema_spec(), Arc::new(schema_handler),
|
||||
CompositionAuthority::none(), ScopedOperationEnv::empty(), Capabilities::new())
|
||||
CompositionAuthority::none(), ScopedPeerEnv::empty(), Capabilities::new())
|
||||
// Agent handler (Local, Subscription — streams call.responded as the
|
||||
// LLM generates tokens; builder wraps in HandlerKind::Stream)
|
||||
.with_local(agent_chat_spec(), Arc::new(agent_chat_streaming_handler),
|
||||
// LLM generates tokens; uses with_local_streaming for the StreamingHandler)
|
||||
.with_local_streaming(agent_chat_spec(), Arc::new(agent_chat_streaming_handler),
|
||||
CompositionAuthority::new("agent-chat", ["llm:call", "fs:read", "vastai:query"]),
|
||||
ScopedOperationEnv::new(["fs/readFile", "vastai/listMachines", "llm/generate"]),
|
||||
ScopedPeerEnv::new(["fs/readFile", "vastai/listMachines", "llm/generate"]),
|
||||
Capabilities::new().with_api_key("google", google_api_key))
|
||||
// Imported ops (leaves — no authority, no scoped env; capabilities for outbound HTTP)
|
||||
.with_leaf(vastai_listMachines_spec(), Arc::new(vastai_handler), vastai_credentials)
|
||||
@@ -621,7 +634,7 @@ impl OperationEnv for LocalOperationEnv {
|
||||
abort_policy: policy, // Explicit policy (from invoke() default or invoke_with_policy)
|
||||
deadline: parent.deadline, // Inherit parent's deadline (children don't get a fresh 30s)
|
||||
scoped_env: registration.scoped_env.clone()
|
||||
.unwrap_or_else(ScopedOperationEnv::empty), // Child's own scoped env (empty for leaves)
|
||||
.unwrap_or_else(ScopedPeerEnv::empty), // Child's own scoped env (empty for leaves)
|
||||
// Dispatch trait: the child inherits the parent's env (the same
|
||||
// composite of curated base + active overlays). See ADR-024.
|
||||
env: parent.env.clone(),
|
||||
@@ -820,9 +833,9 @@ let vastai_credentials = Capabilities::new().with_http_token("vastai", vastai_to
|
||||
let registry = OperationRegistryBuilder::new()
|
||||
// Built-in service discovery (Local, no composition — empty caps)
|
||||
.with_local(services_list_spec(), Arc::new(services_list_handler),
|
||||
CompositionAuthority::none(), ScopedOperationEnv::empty(), Capabilities::new())
|
||||
CompositionAuthority::none(), ScopedPeerEnv::empty(), Capabilities::new())
|
||||
.with_local(services_schema_spec(), Arc::new(schema_handler),
|
||||
CompositionAuthority::none(), ScopedOperationEnv::empty(), Capabilities::new())
|
||||
CompositionAuthority::none(), ScopedPeerEnv::empty(), Capabilities::new())
|
||||
// Agent handler (Local, Subscription — composes; streaming handler
|
||||
// wrapped in HandlerKind::Stream by the builder per ADR-049)
|
||||
.with(HandlerRegistration {
|
||||
@@ -831,7 +844,7 @@ let registry = OperationRegistryBuilder::new()
|
||||
provenance: OperationProvenance::Local,
|
||||
composition_authority: Some(CompositionAuthority::new(
|
||||
"agent-chat", ["llm:call", "fs:read", "vastai:query"])),
|
||||
scoped_env: Some(ScopedOperationEnv::new(
|
||||
scoped_env: Some(ScopedPeerEnv::new(
|
||||
["fs/readFile", "vastai/listMachines", "llm/generate"])),
|
||||
capabilities: Capabilities::new().with_api_key("google", google_api_key),
|
||||
})
|
||||
@@ -940,9 +953,10 @@ See [open-questions.md](../../open-questions.md) for full details.
|
||||
variants: `DiscoveryFailed`, `SchemaParse`, `Transport`, `Unauthorized`,
|
||||
`SamePeerCollision` (replaces flat `Conflict`). `#[non_exhaustive]`. See
|
||||
[client-and-adapters.md](client-and-adapters.md).
|
||||
- **OQ-27** (resolved): `from_call` re-import trigger — auto-re-import on
|
||||
connection establishment. `CallConnection::refresh()` is a feature
|
||||
addition, not an unmade decision. See [client-and-adapters.md](client-and-adapters.md).
|
||||
- **OQ-27** (resolved): `from_call` re-import trigger — `from_call` is a manual
|
||||
free function; the assembly layer calls it after `connect()`. A
|
||||
`CallConnection::refresh()` method is a genuine feature addition —
|
||||
non-breaking, additive. See [ADR-069](../../decisions/069-from-call-manual-free-function.md).
|
||||
- **OQ-28** (resolved): `from_call` namespace collision — same-peer
|
||||
collision = error; cross-peer dissolved by ADR-029 (separate sub-overlays).
|
||||
`namespace_prefix` is optional local-naming sugar. See
|
||||
|
||||
@@ -0,0 +1,408 @@
|
||||
---
|
||||
status: draft
|
||||
last_updated: 2026-07-09
|
||||
---
|
||||
|
||||
# alknet-hub
|
||||
|
||||
The hub pattern as a reusable crate: aggregated peer environment, connection
|
||||
lifecycle hooks, worker supervision, and service discovery across connected
|
||||
peers. Consumes `alknet-call` types; provides the assembly-layer wiring every
|
||||
hub needs.
|
||||
|
||||
## What
|
||||
|
||||
`alknet-hub` is a thin crate that provides the head/worker (hub-spoke) pattern
|
||||
as a reusable library. It depends on `alknet-call` and `alknet-core`. It does
|
||||
not introduce new types — it wires the existing `PeerCompositeEnv`,
|
||||
`CallClient`, `CallAdapter`, `from_call`, and `Dispatcher` types into a
|
||||
coherent hub runtime.
|
||||
|
||||
A hub is a call-protocol node that:
|
||||
|
||||
1. **Aggregates multiple worker connections** into a single
|
||||
`PeerCompositeEnv` shared across all calls.
|
||||
2. **Discovers worker operations** via `from_call` on connection establishment.
|
||||
3. **Supervises worker connections** — reconnects on drop with backoff,
|
||||
re-discovers ops on reconnect, detaches on disconnect.
|
||||
4. **Exposes service discovery** — `services/list-peers` returns each
|
||||
connected worker's operation list.
|
||||
|
||||
The crate provides these four capabilities. A downstream consumer (alkapi,
|
||||
any future hub) constructs a `Hub`, registers its curated operations, and
|
||||
starts accepting/dialing workers. The hub handles the rest.
|
||||
|
||||
## Why
|
||||
|
||||
The alkapi project identified that the hub pattern requires wiring that
|
||||
alknet-call does not provide out of the box:
|
||||
|
||||
- `Dispatcher::compose_root_env` builds a fresh `PeerCompositeEnv` per call
|
||||
with only the current connection — multi-worker aggregation is not wired
|
||||
(ADR-067).
|
||||
- `PeerCompositeEnv::peer_operations` is not overridden — `services/list-peers`
|
||||
returns empty operation lists for non-local peers (ADR-068).
|
||||
- `from_call` is a free function, not wired into `CallClient::connect` — the
|
||||
assembly layer must call it manually after every connect (ADR-069).
|
||||
- There is no worker supervision loop — reconnection, backoff, and
|
||||
re-discovery are assembly-layer concerns.
|
||||
|
||||
These are not design flaws; they are the correct separation of concerns.
|
||||
`alknet-call` provides the types and the routing logic; the hub wiring is a
|
||||
consumer concern. But it is a concern *every* hub consumer shares. Rather than
|
||||
each downstream project (alkapi, future hubs) building the same wiring
|
||||
independently, `alknet-hub` provides it once, as a reusable crate.
|
||||
|
||||
## Architecture
|
||||
|
||||
### Hub struct
|
||||
|
||||
The `Hub` is the central type. It owns the aggregated `PeerCompositeEnv`, the
|
||||
`OperationRegistry`, and the `Dispatcher`:
|
||||
|
||||
```rust
|
||||
pub struct Hub {
|
||||
registry: Arc<OperationRegistry>,
|
||||
aggregated_env: Arc<RwLock<PeerCompositeEnv>>,
|
||||
dispatcher: Dispatcher,
|
||||
identity_provider: Arc<dyn IdentityProvider>,
|
||||
}
|
||||
```
|
||||
|
||||
Construction:
|
||||
|
||||
```rust
|
||||
impl Hub {
|
||||
pub fn new(
|
||||
registry: Arc<OperationRegistry>,
|
||||
identity_provider: Arc<dyn IdentityProvider>,
|
||||
) -> Self {
|
||||
let base: Arc<dyn OperationEnv + Send + Sync> =
|
||||
Arc::new(LocalOperationEnv::new(Arc::clone(®istry)));
|
||||
let aggregated_env = Arc::new(RwLock::new(PeerCompositeEnv::new(base)));
|
||||
let dispatcher = Dispatcher::new(Arc::clone(®istry), Arc::clone(&identity_provider))
|
||||
.with_aggregated_env(Arc::clone(&aggregated_env));
|
||||
|
||||
Self {
|
||||
registry,
|
||||
aggregated_env,
|
||||
dispatcher,
|
||||
identity_provider,
|
||||
}
|
||||
}
|
||||
|
||||
/// The shared aggregated PeerCompositeEnv. The assembly layer wires
|
||||
/// this into CallAdapter::with_aggregated_env so every call's
|
||||
/// compose_root_env sees all connected workers.
|
||||
pub fn aggregated_env(&self) -> &Arc<RwLock<PeerCompositeEnv>> {
|
||||
&self.aggregated_env
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The `Hub` exposes builder methods for optional hooks (`with_session_source`,
|
||||
`with_ownership_provider`, `with_timeout`) that delegate to the `Dispatcher`.
|
||||
|
||||
### Worker connection lifecycle
|
||||
|
||||
The hub provides two paths for worker connections:
|
||||
|
||||
#### Inbound workers (workers dial the hub)
|
||||
|
||||
The hub's `CallAdapter` accepts inbound connections. The hub provides an
|
||||
`on_worker_connected` hook that the assembly layer wires into the accept path:
|
||||
|
||||
```rust
|
||||
impl Hub {
|
||||
/// Called by the assembly layer when a worker connects inbound (via
|
||||
/// CallAdapter::handle). Runs from_call, registers the bundles in the
|
||||
/// connection's overlay, and attaches the peer to the aggregated env.
|
||||
pub async fn on_worker_connected(
|
||||
&self,
|
||||
connection: &CallConnection,
|
||||
config: FromCallConfig,
|
||||
) -> Result<PeerId, HubError> {
|
||||
let peer_id = connection.identity()
|
||||
.map(|id| id.id.clone())
|
||||
.ok_or(HubError::NoPeerIdentity)?;
|
||||
|
||||
let bundles = from_call(connection, config).await?;
|
||||
connection.register_imported_all(bundles);
|
||||
|
||||
self.aggregated_env
|
||||
.write()
|
||||
.unwrap()
|
||||
.attach_peer(peer_id.clone(), connection.overlay_env());
|
||||
|
||||
Ok(peer_id)
|
||||
}
|
||||
|
||||
/// Called by the assembly layer when a worker disconnects (run_loop exits).
|
||||
pub fn on_worker_disconnected(&self, peer_id: &PeerId) {
|
||||
self.aggregated_env.write().unwrap().detach_peer(peer_id);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Outbound workers (hub dials workers)
|
||||
|
||||
The hub provides a `dial_worker` method that combines `CallClient::connect`,
|
||||
`from_call`, and `attach_peer` into a single operation. It also provides a
|
||||
`supervise_worker` method that wraps `dial_worker` in a reconnect loop with
|
||||
configurable backoff:
|
||||
|
||||
```rust
|
||||
impl Hub {
|
||||
/// Dial a worker, discover its operations, and attach it to the
|
||||
/// aggregated env. Returns the worker's PeerId and the live
|
||||
/// CallConnection.
|
||||
pub async fn dial_worker(
|
||||
&self,
|
||||
addr: SocketAddr,
|
||||
credentials: CallCredentials,
|
||||
config: FromCallConfig,
|
||||
) -> Result<(PeerId, CallConnection), HubError> {
|
||||
let client = CallClient::new(
|
||||
Arc::clone(&self.registry),
|
||||
Arc::clone(&self.identity_provider),
|
||||
);
|
||||
let connection = client.connect(addr, credentials).await?;
|
||||
|
||||
let peer_id = connection.identity()
|
||||
.map(|id| id.id.clone())
|
||||
.ok_or(HubError::NoPeerIdentity)?;
|
||||
|
||||
let bundles = from_call(&connection, config).await?;
|
||||
connection.register_imported_all(bundles);
|
||||
|
||||
self.aggregated_env
|
||||
.write()
|
||||
.unwrap()
|
||||
.attach_peer(peer_id.clone(), connection.overlay_env());
|
||||
|
||||
Ok((peer_id, connection))
|
||||
}
|
||||
|
||||
/// Supervise an outbound worker connection: dial, discover, attach.
|
||||
/// On disconnect, detach and retry with backoff. Runs until the
|
||||
/// Hub is dropped (the returned JoinHandle can be aborted).
|
||||
pub fn supervise_worker(
|
||||
self: &Arc<Self>,
|
||||
addr: SocketAddr,
|
||||
credentials: CallCredentials,
|
||||
config: FromCallConfig,
|
||||
backoff: BackoffConfig,
|
||||
) -> tokio::task::JoinHandle<()> {
|
||||
let hub = Arc::clone(self);
|
||||
tokio::spawn(async move {
|
||||
let mut retries = 0;
|
||||
loop {
|
||||
match hub.dial_worker(addr, credentials.clone(), config.clone()).await {
|
||||
Ok((peer_id, connection)) => {
|
||||
retries = 0;
|
||||
// Wait for the connection to drop (run_loop exits).
|
||||
// Committed interim (OQ-52): poll the underlying
|
||||
// Connection's accept_bi() until it returns
|
||||
// ConnectionClosed. A CallConnection::closed()
|
||||
// method is the target resolution.
|
||||
loop {
|
||||
match connection.connection() {
|
||||
Some(conn) => {
|
||||
match conn.accept_bi().await {
|
||||
Err(StreamError::ConnectionClosed)
|
||||
| Err(StreamError::StreamClosed)
|
||||
| Err(StreamError::Timeout) => break,
|
||||
_ => continue,
|
||||
}
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
hub.on_worker_disconnected(&peer_id);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(?e, retries, "worker dial failed; retrying");
|
||||
}
|
||||
}
|
||||
let delay = backoff.delay_for(retries);
|
||||
tokio::time::sleep(delay).await;
|
||||
retries += 1;
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Backoff configuration
|
||||
|
||||
```rust
|
||||
pub struct BackoffConfig {
|
||||
pub initial: Duration,
|
||||
pub max: Duration,
|
||||
pub multiplier: f64,
|
||||
}
|
||||
|
||||
impl BackoffConfig {
|
||||
pub fn delay_for(&self, retries: u32) -> Duration {
|
||||
let delay = self.initial.as_millis() as f64
|
||||
* self.multiplier.powi(retries as i32);
|
||||
let delay_ms = delay.min(self.max.as_millis() as f64) as u64;
|
||||
Duration::from_millis(delay_ms)
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for BackoffConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
initial: Duration::from_secs(1),
|
||||
max: Duration::from_secs(60),
|
||||
multiplier: 2.0,
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Service discovery
|
||||
|
||||
The hub registers the built-in service discovery operations (`services/list`,
|
||||
`services/schema`, `services/list-peers`) automatically. The
|
||||
`services/list-peers` handler returns each connected worker's operation list
|
||||
via `PeerCompositeEnv::peer_operations` (ADR-068).
|
||||
|
||||
### HubError
|
||||
|
||||
```rust
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
#[non_exhaustive]
|
||||
pub enum HubError {
|
||||
#[error("worker connection has no resolved peer identity")]
|
||||
NoPeerIdentity,
|
||||
#[error("from_call discovery failed: {0}")]
|
||||
Discovery(#[from] AdapterError),
|
||||
#[error("call client error: {0}")]
|
||||
Client(#[from] ClientError),
|
||||
}
|
||||
```
|
||||
|
||||
## What the hub does NOT do
|
||||
|
||||
- **Worker authentication policy.** The hub resolves the worker's identity
|
||||
via `IdentityProvider` (the existing mechanism). Whether a given identity is
|
||||
*allowed* to connect as a worker is an `AccessControl` decision on the
|
||||
assembly layer's curated ops — the hub does not add a separate worker-auth
|
||||
layer.
|
||||
- **Worker-specific routing policy.** `PeerRef::Any` uses insertion-order
|
||||
first-match (ADR-029 §2). A richer `RoutingPolicy` (round-robin,
|
||||
least-loaded) is a future extension behind the same `PeerRef` enum.
|
||||
- **Multi-hop federation.** The hub is one-hop: workers connect to the hub,
|
||||
the hub composes their ops. Worker A does not transitively see worker B's
|
||||
ops through the hub unless the hub explicitly re-exports them (ADR-029
|
||||
Assumption 5).
|
||||
- **Ownership reaping.** Stale ownership entries from autonomously-dead
|
||||
containers are tolerated (ADR-050 §4b). The hub does not add a reaper.
|
||||
- **Worker provisioning.** The hub does not spawn workers, configure them, or
|
||||
manage their lifecycle beyond connection supervision. Worker provisioning
|
||||
is an assembly-layer concern.
|
||||
|
||||
## Crate dependencies
|
||||
|
||||
```
|
||||
alknet-hub
|
||||
├── alknet-call (CallClient, CallAdapter, Dispatcher, PeerCompositeEnv,
|
||||
│ from_call, FromCallConfig, AdapterError, ClientError)
|
||||
├── alknet-core (IdentityProvider, Connection, OperationRegistry)
|
||||
├── tokio (spawn, time::sleep)
|
||||
└── tracing (logging)
|
||||
```
|
||||
|
||||
`alknet-hub` depends on `alknet-call`, which depends on `alknet-core`. No new
|
||||
dependency edges. The crate is a consumer of the call protocol types, not a
|
||||
new protocol handler.
|
||||
|
||||
## Assembly layer integration
|
||||
|
||||
A downstream hub (alkapi) uses `alknet-hub` like this:
|
||||
|
||||
```rust
|
||||
// 1. Build the curated registry (Layer 0)
|
||||
let registry = OperationRegistryBuilder::new()
|
||||
.with_local(agent_chat_spec(), agent_chat_handler, ...)
|
||||
.with_local(services_list_spec(), services_list_handler, ...)
|
||||
// ... other curated ops
|
||||
.build();
|
||||
let registry = Arc::new(registry);
|
||||
|
||||
// 2. Create the hub
|
||||
let hub = Arc::new(Hub::new(
|
||||
Arc::clone(®istry),
|
||||
Arc::clone(&identity_provider),
|
||||
).with_ownership_provider(ownership_provider));
|
||||
|
||||
// 3. Start the inbound call-protocol listener (workers dial the hub)
|
||||
let adapter = CallAdapter::new(Arc::clone(®istry), Arc::clone(&identity_provider))
|
||||
.with_aggregated_env(hub.aggregated_env().clone());
|
||||
// ... register adapter on the QUIC endpoint
|
||||
|
||||
// 4. Dial outbound workers (hub dials workers)
|
||||
hub.supervise_worker(
|
||||
dev1_addr,
|
||||
dev1_credentials,
|
||||
FromCallConfig::new(),
|
||||
BackoffConfig::default(),
|
||||
);
|
||||
|
||||
// 5. Start the HTTP listener (clients dial the hub)
|
||||
// ... HttpAdapter with the same registry and identity_provider
|
||||
```
|
||||
|
||||
The hub's `aggregated_env()` accessor returns the shared `Arc<RwLock<PeerCompositeEnv>>`
|
||||
so the assembly layer can wire it into `CallAdapter::with_aggregated_env`.
|
||||
|
||||
## Design Decisions
|
||||
|
||||
| Decision | ADR | Summary |
|
||||
|----------|-----|---------|
|
||||
| Aggregated peer-env wiring | [ADR-067](../../decisions/067-aggregated-peer-env-wiring.md) | `Dispatcher::with_aggregated_env` hook; `compose_root_env` reads shared env |
|
||||
| PeerCompositeEnv::peer_operations | [ADR-068](../../decisions/068-peer-composite-env-peer-operations.md) | `list_operation_names` trait method; `PeerCompositeEnv` override |
|
||||
| from_call is manual | [ADR-069](../../decisions/069-from-call-manual-free-function.md) | `from_call` is a free function; the hub calls it after connect |
|
||||
| Peer-graph routing model | [ADR-029](../../decisions/029-peer-graph-routing-model.md) | Peer-keyed overlays, `PeerRef` routing, `AccessControl`-based peer auth |
|
||||
| PeerEntry and Identity.id | [ADR-030](../../decisions/030-peerentry-and-identity-id-decoupling.md) | `PeerId` = `Identity.id` = `PeerEntry.peer_id` (stable) |
|
||||
|
||||
## Open Questions
|
||||
|
||||
See [open-questions.md](../../open-questions.md) for full details.
|
||||
|
||||
- **OQ-52** (open): `CallConnection::wait_for_close()` — the supervision loop
|
||||
needs a way to await connection close. Today `CallConnection` exposes
|
||||
`connection()` (the underlying `Connection`) but not a "wait for run_loop
|
||||
exit" future. The committed interim is polling `connection().accept_bi()`
|
||||
in a loop until it returns `ConnectionClosed`. A `closed()` method on
|
||||
`CallConnection` (signaled by the dispatcher on `run_loop` exit) is the
|
||||
target resolution. See OQ-52.
|
||||
- **OQ-53** (open): `BackoffConfig` defaults — the committed policy is 1s
|
||||
initial, 60s max, 2x multiplier. OQ-53 tracks whether operational
|
||||
experience from the alkapi deployment warrants a change before the first
|
||||
release. The `BackoffConfig` struct shape is committed; the defaults are
|
||||
the starting point.
|
||||
- **OQ-54** (open): Inbound worker `on_worker_connected` hook placement —
|
||||
the committed design is the explicit approach: the assembly layer calls
|
||||
`hub.on_worker_connected()` after `CallAdapter::handle` accepts the
|
||||
connection. A `HubCallAdapter` wrapper that calls the hook automatically
|
||||
is additive if needed. See OQ-54.
|
||||
|
||||
## References
|
||||
|
||||
- [client-and-adapters.md](../call/client-and-adapters.md) — `CallClient`,
|
||||
`from_call`, `OperationAdapter`
|
||||
- [call-protocol.md](../call/call-protocol.md) — `CallAdapter`, `Dispatcher`,
|
||||
`CallConnection`
|
||||
- [operation-registry.md](../call/operation-registry.md) — `OperationRegistry`,
|
||||
`OperationRegistryBuilder`, `OperationEnv`
|
||||
- ADR-029: Peer-Graph Routing Model
|
||||
- ADR-067: Aggregated Peer-Environment Wiring
|
||||
- ADR-068: PeerCompositeEnv::peer_operations Override
|
||||
- ADR-069: from_call Is a Manual Free Function
|
||||
- alkapi [hub.md](/workspace/@alkdev/alkapi/docs/architecture/hub.md) — the
|
||||
first hub consumer, the concrete use case that informed this crate
|
||||
- alkapi [ADR-011](/workspace/@alkdev/alkapi/docs/architecture/decisions/011-aggregated-peer-env.md) —
|
||||
the downstream aggregation decision
|
||||
Reference in new issue
Block a user