A chunk routed by the demux inside the adopt_channel install→drain window was delivered ahead of earlier-parked early arrivals: the mpsc send order was inverted relative to wire-arrival order, misordering the channel's byte stream (C-01, critical — downstream alkgit observed two desync shapes under multi-process load). A related loss variant is also closed: a chunk whose map miss parked it after the drain had removed the queue sat parked forever. The window closes by serialization, not by widening locks across awaits — the resolution is fully synchronous: - route_payload holds the channels-map lock across the miss → park decision (park_early_arrival runs nested; lock order channels → early_arrivals): a park can never straddle the install, and the installer that caused the miss always swaps the queue out. - install_locked (new, shared by open_channel / adopt_channel / install_channel_zero) swaps the parked queue, inserts the state, and try_sends the parked queue into the fresh, empty receiver before releasing the channels lock — a direct route can only begin after release, so parked chunks always precede it in send order. No await runs inside the section: the effective park cap is now min(EARLY_ARRIVAL_CAP, buffer_cap), bounding the parked queue by the receiver's capacity so the drain never blocks. drain_early_arrivals remains as the defensive full-buffer fallback and is now called by all three install paths. Regression tests: adopt_window_does_not_reorder_chunks (200 iterations on a 2-worker runtime — a racer routing C the moment the install is visible, receiver byte stream asserted byte-identical to the producer's; catches the old window under genuine preemption, the alkgit repro shape), adopt_drains_parked_and_raced_chunks_in_fifo, and park_cap_bounded_by_receiver_buffer. rt-multi-thread added as a dev-dependency for the runtime test. Semver: patch (0.10.1) — private fix surface; no API, wire-format, or documented-cap behavior change; behavior only strengthens. Verification: cargo test 700/724 (--all-features) passed; clippy --all-targets -D warnings clean; fmt --check clean; cargo doc --no-deps generated; fuzz corpus replay passed; wasm32-unknown-unknown check passed.
alkcall
Call + channels RPC: structured JSON operations, streaming subscriptions, service discovery, and N-channel multiplexing over one transport stream.
This crate unifies the call protocol and the channels protocol, plus the
vendored core types formerly in alknet-core. It is a pure protocol crate
— no networking, no transport dependencies. Downstream crates (alktty,
alktunnels, alktrader) compose on top of it.
Quick start
Producer — register an operation and run a dispatcher
use std::sync::Arc;
use alkcall::core::{Capabilities, IdentityProvider};
use alkcall::protocol::{CallAdapter, connection::CallConnection};
use alkcall::registry::{
registration::{HandlerRegistration, HandlerKind, OperationRegistry, make_handler},
spec::{OperationSpec, OperationType, Visibility, AccessControl},
};
let registry = OperationRegistry::new();
registry.register(HandlerRegistration::new(
OperationSpec::new(
"echo/run",
OperationType::Query,
Visibility::External,
serde_json::json!({}),
serde_json::json!({}),
vec![],
AccessControl::default(),
None,
),
HandlerKind::Once(make_handler(|input, ctx| async move {
alkcall::protocol::wire::ResponseEnvelope::ok(ctx.request_id, input)
})),
alkcall::registry::registration::OperationProvenance::Local,
None,
None,
Capabilities::new(),
)).unwrap();
let registry = Arc::new(registry);
let provider: Arc<dyn IdentityProvider> = /* your identity provider */;
let adapter = CallAdapter::new(registry, provider);
// adapter implements ProtocolHandler — call adapter.handle(connection, &auth).await
Consumer — call an operation
use alkcall::core::Connection;
use alkcall::protocol::connection::CallConnection;
let connection = Connection::from_bidi(
transport_stream,
b"alk/call".to_vec(),
Some(remote_addr),
);
let conn = CallConnection::new(connection);
let response = conn.call("echo/run", serde_json::json!({"msg": "hello"})).await;
assert!(response.result.is_ok());
Channels — open a channel and call through channel 0
use alkcall::channels::client::ChannelClient;
use alkcall::core::Connection;
let connection = Connection::from_bidi(
transport_stream,
b"alk/channels".to_vec(),
Some(remote_addr),
);
let client = ChannelClient::from_connection(connection).await?;
let response = client.call_open_op(
"echo/run",
serde_json::json!({"msg": "hello"}),
).await;
from_call — discover and import remote operations
use alkcall::client::{from_call, FromCallConfig};
let registrations = from_call(&conn, FromCallConfig::new()).await?;
for reg in registrations {
conn.register_imported(reg);
}
// now call remote ops as if they were local
let response = conn.call("remote/status", serde_json::json!({})).await;
Serving your own ops as a connected consumer
The call protocol is symmetric — both sides of a connection can serve
ops. A ChannelClient built with from_connection is a pure consumer
(inbound call.requested frames are dropped); pass a ServingConfig
to also serve your registry to the peer, and use op/register to
announce which ops you serve:
use std::sync::Arc;
use alkcall::channels::client::{ChannelClient, ServingConfig};
use alkcall::registry::discovery::install_bootstrap_discovery;
let registry = Arc::new(OperationRegistry::new());
// ... register your ops on the registry, then:
install_bootstrap_discovery(®istry)?;
let client = ChannelClient::from_connection_with_serving(
connection,
Some(ServingConfig {
registry: Arc::clone(®istry),
identity_provider: provider,
identity: None, // peer identity: transport `Connection::set_identity` propagates
}),
).await?;
// peer-callable ops resolve against `registry` on channel 0;
// `client.call_open_op` still works — both directions share the pump
Architecture
alkcall is a pure protocol crate — no networking, no transport dependencies. It provides the call and channels protocols as a library. Downstream crates compose on top of it in a layered dependency chain.
| Role | Call protocol | Channels protocol |
|---|---|---|
| Producer | Registers ops on an OperationRegistry, runs a Dispatcher |
Runs a ChannelsAdapter, registers openable ALPNs via ChannelCore::register_openable |
| Consumer | Uses CallConnection to call ops, uses from_call to discover/import remote ops; may also serve its own ops (from_connection_with_serving) |
Uses ChannelClient to open channels via call_open_op + open_channel |
| Hub | Both: runs a Dispatcher for ops it produces, holds CallConnections to spokes for ops it consumes |
Both: runs a ChannelsAdapter for inbound connections, holds ChannelClients to spokes |
| Spoke / Worker | Both: produces ops (its own services), consumes hub ops | Both: produces channels (TTY, tunnel), may consume hub channels |
A single process can be a producer of some ops, a consumer of others, a
channel opener for TTY, and a channel acceptor for tunnels — all on the
same alk/channels connection.
Documentation
- Architecture docs — the authoritative spec: ADRs, wire formats, protocol contracts, and composition patterns.
- API docs — full crate documentation on docs.rs.
- Open questions — tracked deferred decisions and feature gaps.
License
MIT OR Apache-2.0