This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 0270b76b6 refactor(server): make the module graph a DAG and enforce it
(#4023)
0270b76b6 is described below
commit 0270b76b6b23f45d9fe0ab9a0c8a280f2e396d99
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Tue Sep 1 14:59:56 2026 +0200
refactor(server): make the module graph a DAG and enforce it (#4023)
---
Cargo.lock | 16 +-
Cargo.toml | 2 -
core/partitions/src/send_messages2.rs | 17 --
core/server/Cargo.toml | 7 +-
core/server/build.rs | 23 ++
core/server/src/auth.rs | 14 +-
core/server/src/bootstrap.rs | 187 +-----------
core/server/src/consumer_group.rs | 6 +-
core/server/src/dispatch.rs | 58 ++--
core/server/src/dispatch/authz.rs | 2 +-
core/server/src/http.rs | 8 +-
core/server/src/http/forward.rs | 19 +-
core/server/src/http/handlers.rs | 67 +++++
core/server/src/http/metrics.rs | 101 +------
core/server/src/http/reads.rs | 2 +-
core/server/src/http/state.rs | 23 +-
core/server/src/http/submit.rs | 2 +-
core/server/src/{ => http}/web.rs | 0
core/server/src/lib.rs | 53 ++--
core/server/src/partition_helpers.rs | 4 +-
core/server/src/partition_reconciler.rs | 2 +-
core/server/src/pat.rs | 4 +-
core/server/src/personal_access_token_cleaner.rs | 2 +-
core/server/src/responses.rs | 40 +--
core/server/src/segment_cleaner.rs | 2 +-
core/server/src/server_error.rs | 2 +-
.../src/lib.rs => server/src/shard_allocator.rs} | 51 ++--
core/server/src/shell.rs | 202 +++++++++++++
core/server/src/users.rs | 4 +-
core/server/src/wire.rs | 12 +-
core/server/tests/module_graph.rs | 319 +++++++++++++++++++++
core/shard_allocator/Cargo.toml | 41 ---
core/shard_allocator/build.rs | 43 ---
core/simulator/src/replica.rs | 3 +-
34 files changed, 811 insertions(+), 527 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 251f7f309..5a622dab6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -12048,6 +12048,7 @@ dependencies = [
"compio",
"configs",
"consensus",
+ "cpu_allocation",
"crossfire",
"ctrlc",
"cyper",
@@ -12082,7 +12083,9 @@ dependencies = [
"opentelemetry_sdk",
"papaya",
"partitions",
+ "proc-macro2",
"prometheus-client",
+ "quote",
"rand 0.10.2",
"ringbuffer",
"rmp-serde",
@@ -12097,9 +12100,9 @@ dependencies = [
"serde_json",
"server_common",
"shard",
- "shard_allocator",
"socket2 0.6.5",
"strum 0.28.0",
+ "syn 2.0.119",
"sysinfo 0.39.6",
"system_stats",
"tempfile",
@@ -12233,17 +12236,6 @@ dependencies = [
"tracing",
]
-[[package]]
-name = "shard_allocator"
-version = "0.1.0"
-dependencies = [
- "cpu_allocation",
- "hwlocality",
- "nix",
- "thiserror 2.0.19",
- "tracing",
-]
-
[[package]]
name = "sharded-slab"
version = "0.1.7"
diff --git a/Cargo.toml b/Cargo.toml
index 7872f4093..0f339f28c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -64,7 +64,6 @@ members = [
"core/server",
"core/server_common",
"core/shard",
- "core/shard_allocator",
"core/simulator",
"core/system_stats",
"core/tools",
@@ -303,7 +302,6 @@ serial_test = "3.5.0"
server = { path = "core/server" }
server_common = { path = "core/server_common" }
shard = { path = "core/shard" }
-shard_allocator = { path = "core/shard_allocator" }
simd-json = { version = "0.17.3", features = ["serde_impl"] }
smallvec = "1.15"
socket2 = "0.6.5"
diff --git a/core/partitions/src/send_messages2.rs
b/core/partitions/src/send_messages2.rs
deleted file mode 100644
index 5cd17fb5a..000000000
--- a/core/partitions/src/send_messages2.rs
+++ /dev/null
@@ -1,17 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 22d6ba2f0..1d5911488 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -45,12 +45,10 @@ ignored = [
"figlet-rs",
"hash32",
"human-repr",
- "hwlocality",
"jsonwebtoken",
"left-right",
"mimalloc",
"mime_guess",
- "nix",
"opentelemetry",
"opentelemetry-appender-tracing",
"opentelemetry-otlp",
@@ -103,6 +101,7 @@ clap = { workspace = true }
compio = { workspace = true }
configs = { workspace = true }
consensus = { workspace = true }
+cpu_allocation = { workspace = true }
crossfire = { workspace = true }
ctrlc = { workspace = true }
cyper = { workspace = true }
@@ -151,7 +150,6 @@ serde = { workspace = true }
serde_json = { workspace = true }
server_common = { workspace = true }
shard = { workspace = true }
-shard_allocator = { workspace = true }
socket2 = { workspace = true }
strum = { workspace = true }
sysinfo = { workspace = true }
@@ -179,6 +177,8 @@ vergen-git2 = { workspace = true }
[dev-dependencies]
assert_cmd = { workspace = true }
bytemuck = { workspace = true }
+proc-macro2 = { workspace = true }
+quote = { workspace = true }
# Reconciler unit tests assert on `ShardMetrics` snapshots and
# `IggyShard::parked_frame_count`, gated to test/simulator so they cannot grow
# production callers. `shard`'s own `cfg(test)` is false when compiled as our
@@ -186,6 +186,7 @@ bytemuck = { workspace = true }
# keeps a dev-dependency's features out of non-test targets, so a production
# build still links `shard` without `simulator`.
shard = { workspace = true, features = ["simulator"] }
+syn = { workspace = true }
tokio = { workspace = true, features = ["full", "test-util"] }
[lints.clippy]
diff --git a/core/server/build.rs b/core/server/build.rs
index 421dd918a..ad152b723 100644
--- a/core/server/build.rs
+++ b/core/server/build.rs
@@ -23,11 +23,34 @@ const WEB_ASSETS_PATH: &str = "web/build/static";
const WEB_INDEX_FILE: &str = "web/build/static/index.html";
fn main() -> Result<(), Box<dyn error::Error>> {
+ stub_libm_for_musl();
verify_web_assets_if_enabled();
emit_vergen_instructions()?;
Ok(())
}
+// Vendored `hwloc` references `cbrt`, which makes the linker pull in a
+// `libm`. musl folds the math functions into its `libc`, so the Rust
+// musl sysroot ships no `libm.a`. Without one, `-lm` falls through to
+// the host glibc's `libm.a`, whose `cbrt` needs glibc-internal
+// `__frexp`/`__ldexp` symbols that do not exist on musl, and the static
+// link fails. Drop an empty `libm.a` stub on the search path so `-lm`
+// resolves to nothing and `cbrt` is satisfied later by musl's own
+// `libc`. No effect on non-musl targets.
+fn stub_libm_for_musl() {
+ if env::var("CARGO_CFG_TARGET_ENV").as_deref() != Ok("musl") {
+ return;
+ }
+
+ let out_dir = env::var("OUT_DIR").expect("OUT_DIR is set by cargo for
build scripts");
+ let stub = PathBuf::from(&out_dir).join("libm.a");
+
+ // `!<arch>\n` is the canonical header of an empty `ar` archive.
+ std::fs::write(&stub, b"!<arch>\n").expect("write empty libm.a stub");
+
+ println!("cargo:rustc-link-search=native={out_dir}");
+}
+
/// Returns the workspace root (iggy/), two levels up from core/server.
fn workspace_root() -> PathBuf {
PathBuf::from(env::var("CARGO_MANIFEST_DIR").unwrap())
diff --git a/core/server/src/auth.rs b/core/server/src/auth.rs
index 589214852..1851281c6 100644
--- a/core/server/src/auth.rs
+++ b/core/server/src/auth.rs
@@ -22,11 +22,11 @@
//! surfaced as typed `Eviction` frames, transient ones as result-framed
//! replay hints.
-use crate::bootstrap::{ShellBus, ShellShard};
use crate::dispatch::{send_login_eviction, submit_register_on_owner};
use crate::login_register::LoginRegisterError;
use crate::responses::{build_login_register_reply, current_metadata_commit};
use crate::session_manager::{ClientSdkInfo, SessionManager};
+use crate::shell::{ShellBus, ShellShard};
use consensus::{MetadataHandle, build_result_rejection_reply};
use iggy_binary_protocol::PrepareHeader;
use iggy_binary_protocol::{ClientVersionInfo, EvictionReason,
RoutedRequestHeader};
@@ -58,11 +58,11 @@ static DUMMY_PASSWORD_HASH: LazyLock<String> =
/// Pay the one-time Argon2 cost of [`DUMMY_PASSWORD_HASH`] at boot instead of
/// inside the first unknown-username login request.
-pub(crate) fn warm_dummy_password_hash() {
+pub fn warm_dummy_password_hash() {
LazyLock::force(&DUMMY_PASSWORD_HASH);
}
-pub(crate) fn verify_login_credentials<B, MJ, S, SB>(
+pub fn verify_login_credentials<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
username: &str,
password: &str,
@@ -109,7 +109,7 @@ where
})
}
-pub(crate) fn verify_pat_credentials<B, MJ, S, SB>(
+pub fn verify_pat_credentials<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
token: &str,
) -> Result<u32, LoginRegisterError>
@@ -127,7 +127,7 @@ where
/// seconds, `u64::MAX` when the PAT never expires). The HTTP extractor keys a
/// per-token VSR session table on this expiry for lazy eviction; the wire and
/// login paths only need the user id and go through
[`verify_pat_credentials`].
-pub(crate) fn verify_pat_credentials_with_expiry<B, MJ, S, SB>(
+pub fn verify_pat_credentials_with_expiry<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
token: &str,
) -> Result<(u32, u64), LoginRegisterError>
@@ -179,7 +179,7 @@ where
}
#[allow(clippy::future_not_send)]
-pub(crate) async fn complete_login_register<B, MJ, S, SB>(
+pub async fn complete_login_register<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
@@ -294,7 +294,7 @@ where
/// SDK surfaces the real reason (every frame transport decodes
/// `Command::Eviction`) instead of a decode error or a timeout.
#[allow(clippy::future_not_send)]
-pub(crate) async fn surface_login_failure<B, MJ, S, SB>(
+pub async fn surface_login_failure<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
request_header: &RoutedRequestHeader,
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 4320c3bc5..186325ff4 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -33,14 +33,19 @@ use crate::server_error::{
PartitionRecoveryRefusal, ServerError, ShardJoinFailure,
ShardJoinFailureKind,
};
use crate::session_manager::SessionManager;
+use crate::shard_allocator::{ShardAllocator, ShardInfo};
+use crate::shell::{
+ ServerMetadata, ServerMetadataBundle, ServerMuxStateMachine, ServerShard,
ShellBus,
+ ShellHandlers, ShellShardHandle, consensus_timers, repair_retry_ticks,
+};
use compio::runtime::ResumeUnwind;
use configs::server::{ServerConfig, ServerSystemConfig};
use configs::sharding::{
INBOX_CAPACITY_MAX, SHUTDOWN_DRAIN_TIMEOUT_MAX, SHUTDOWN_POLL_INTERVAL_MAX,
};
use consensus::{
- ClientTable, ConsensusTimers, JoinMode, LocalPipeline, MetadataHandle,
PartitionsHandle,
- PipelineEntry, Sequencer, VsrConsensus, VsrRestore,
+ ClientTable, JoinMode, LocalPipeline, MetadataHandle, PartitionsHandle,
PipelineEntry,
+ Sequencer, VsrConsensus, VsrRestore,
};
// `try_send` / `try_recv` resolve through these traits on `MAsyncTx` /
// `MAsyncRx`; the metadata-handoff loops below depend on the
@@ -53,8 +58,7 @@ use iggy_common::defaults::{
MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
};
use iggy_common::{
- Aes256GcmEncryptor, EncryptorKind, IggyByteSize, IggyError, PartitionStats,
- TopicRuntimeOptions, variadic,
+ Aes256GcmEncryptor, EncryptorKind, IggyByteSize, IggyError,
PartitionStats, TopicRuntimeOptions,
};
use journal::prepare_journal::PrepareJournal;
use journal::superblock::{PingPongSuperblock, SuperblockStore};
@@ -65,7 +69,7 @@ use message_bus::installer::conn_info::{ClientConnMeta,
ClientTransportKind};
use message_bus::replica::auth::{self, ReplicaAuth};
use message_bus::replica::handshake::{ReplicaHandshakeCtx, ReplicaTlsCtx};
use message_bus::replica::io as replica_io;
-use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::replica::listener::{self as replica_listener};
use message_bus::transports::quic::server_config_with_cert;
use message_bus::transports::tls::{
AcceptAnyServerCert, REPLICA_ALPN, TlsServerCredentials,
install_default_crypto_provider,
@@ -76,15 +80,11 @@ use message_bus::{
AcceptedWsClientFn, AcceptedWssClientFn, ConnectionInstaller,
DialedReplicaFn, IggyMessageBus,
MAX_INFLIGHT_REPLICA_HANDSHAKES, MessageBus, ReplicaOwnerTable, connector,
};
-use metadata::IggyMetadata;
-use metadata::MuxStateMachine;
use metadata::ReplicaIdentity;
use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
use metadata::impls::recovery::recover;
-use metadata::stm::mux::WithFactory;
use metadata::stm::snapshot::Snapshot;
-use metadata::stm::stream::{Partition, Streams};
-use metadata::stm::user::Users;
+use metadata::stm::stream::Partition;
use partitions::{
FatalCommit, IggyIndexWriter, IggyPartition, IggyPartitions,
MessagesWriter, PartitionsConfig,
};
@@ -100,18 +100,16 @@ use shard::builder::IggyShardBuilder;
use shard::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant};
use shard::shards_table::{PapayaShardsTable, ShardsTable,
calculate_shard_assignment};
use shard::{
- CoordinatorConfig, IggyShard, LifecycleFrame, ListClientsHandler,
MetadataSubmitHandler,
- PartitionConsensusConfig, PartitionReadHandler, Receiver as ShardReceiver,
ShardFrame,
- ShardIdentity, TaggedSender, channel, shard_mesh_channels,
+ CoordinatorConfig, LifecycleFrame, PartitionConsensusConfig, Receiver as
ShardReceiver,
+ ShardFrame, ShardIdentity, TaggedSender, channel, shard_mesh_channels,
};
-use shard_allocator::{ShardAllocator, ShardInfo};
use std::cell::RefCell;
use std::collections::HashMap;
use std::env;
use std::net::{IpAddr, SocketAddr};
use std::panic;
use std::path::{Path, PathBuf};
-use std::rc::{Rc, Weak};
+use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::thread;
@@ -123,76 +121,6 @@ const SHARD_REPLICA_ID: u8 = 0;
pub const IGGY_ROOT_USERNAME_ENV: &str = "IGGY_ROOT_USERNAME";
pub const IGGY_ROOT_PASSWORD_ENV: &str = "IGGY_ROOT_PASSWORD";
-type ServerMuxStateMachine = MuxStateMachine<variadic!(Users, Streams)>;
-
-/// Cross-thread bundle carrying one `ReadHandleFactory` per metadata
-/// state. Shard 0 mints one after `recover()` and broadcasts a clone to
-/// every peer shard; each peer rebuilds a reader-mode
-/// [`ServerMuxStateMachine`] on its own runtime, skipping the WAL.
-type ServerMetadataBundle = <variadic!(Users, Streams) as WithFactory>::Bundle;
-
-pub(crate) type ServerMetadata = IggyMetadata<
- VsrConsensus<Rc<IggyMessageBus>>,
- PrepareJournal,
- IggySnapshot,
- ServerMuxStateMachine,
->;
-
-/// The shard type the dispatch layer is generic over.
-///
-/// `B`/`MJ`/`S`/`SB` are free; the metadata state machine (`M`) and shards
-/// table (`T`) are pinned, being identical in production and the simulator.
-/// Production instantiates it as [`ServerShard`], defaulting `SB` to the
-/// on-disk [`PingPongSuperblock`]; the simulator supplies its own
-/// `B`/`MJ`/`S`/`SB`.
-pub type ShellShard<B, MJ, S, SB = PingPongSuperblock> =
- IggyShard<B, MJ, S, ServerMuxStateMachine, PapayaShardsTable, SB>;
-
-/// Late-bound self-reference the deferred dispatch handlers upgrade per frame.
-pub type ShellShardHandle<B, MJ, S, SB = PingPongSuperblock> =
- Rc<RefCell<Option<Weak<ShellShard<B, MJ, S, SB>>>>>;
-
-/// Bus bounds the dispatch/pump path needs (matches `run_message_pump`).
-/// Blanket-impl'd, so it is only shorthand for the four underlying bounds.
-pub trait ShellBus: MessageBus + ConnectionInstaller + Clone + 'static {}
-impl<B: MessageBus + ConnectionInstaller + Clone + 'static> ShellBus for B {}
-
-/// The five dispatch handlers a shard is built with, plus the
-/// [`SessionManager`] the request-plane pair shares.
-///
-/// Both production (`build_shard_for_thread`) and the simulator's shell
-/// mode construct these through [`wire_shell_handlers`], so the request
-/// plane is wired one way. The simulator's shell-off fast path uses
-/// [`ShellHandlers::noop`] instead.
-pub struct ShellHandlers {
- pub on_replica_message: MessageHandler,
- pub on_client_request: RequestHandler,
- pub on_metadata_submit: MetadataSubmitHandler,
- pub on_list_clients: ListClientsHandler,
- pub on_partition_read: PartitionReadHandler,
- /// Bound by the client-request handler, read by the get-clients
- /// handler; the caller keeps it to reach locally-homed sessions.
- pub sessions: Rc<RefCell<SessionManager>>,
-}
-
-impl ShellHandlers {
- /// Inert handlers for the shell-off fast path: every callback is a
- /// no-op over an empty [`SessionManager`]. Behaviorally identical to
- /// hand-written no-op closures, so a caller can keep one destructure
- /// site across both toggle states.
- #[must_use]
- pub fn noop() -> Self {
- Self {
- on_replica_message: Rc::new(|_, _| {}),
- on_client_request: Rc::new(|_, _| {}),
- on_metadata_submit: Rc::new(|_| {}),
- on_list_clients: Rc::new(|_| {}),
- on_partition_read: Rc::new(|_, _, _| {}),
- sessions: Rc::new(RefCell::new(SessionManager::new())),
- }
- }
-}
-
/// Build the deferred dispatch handlers for `shard_handle` against `bus`.
///
/// They share one fresh [`SessionManager`]. The caller must set the weak
@@ -228,8 +156,6 @@ where
}
}
-pub type ServerShard = ShellShard<Rc<IggyMessageBus>, PrepareJournal,
IggySnapshot>;
-
/// Result of a multi-shard bootstrap.
///
/// Carries the cross-thread shutdown flag and one OS-thread `JoinHandle`
@@ -727,7 +653,7 @@ fn validate_sharding_runtime_knobs(
/// `(senders, inboxes)` channels, and spawns one OS thread per shard.
///
/// Each thread pins itself (`nix::sched::sched_setaffinity` on Linux via
-/// [`ShardInfo::bind_cpu`]), binds memory to its NUMA node when
+/// `ShardInfo::bind_cpu`), binds memory to its NUMA node when
/// configured, builds a fresh `compio::runtime::Runtime` (one
/// `io_uring` instance per shard), and runs `shard_main` inside it.
///
@@ -2255,13 +2181,6 @@ const _: () =
const _: () =
assert!(consensus::DVC_HEADERS_MAX ==
iggy_binary_protocol::consensus::DVC_HEADERS_MAX);
const _: () = assert!(consensus::DVC_HEADERS_MAX == u128::BITS as usize);
-/// Convert a consensus-timer interval to whole ticks, floored at one tick so a
-/// sub-tick value still fires and saturated on overflow.
-fn duration_to_ticks(interval: Duration) -> u64 {
- let ticks = interval.as_millis() /
shard::CONSENSUS_TICK_INTERVAL.as_millis();
- u64::try_from(ticks.max(1)).unwrap_or(u64::MAX)
-}
-
/// `[cluster] superblock_wedged_fatal_timeout` as a consecutive-failure count.
/// Retries pin at the backoff cap after warmup, so the window divided by
/// [`journal::superblock::SUPERBLOCK_RETRY_BACKOFF_MAX_MICROS`] bounds how
@@ -2284,14 +2203,6 @@ fn superblock_window_to_failures(window: Duration) ->
u64 {
u64::try_from((window.as_micros() / cap_micros).max(1)).unwrap_or(u64::MAX)
}
-/// `[cluster] heartbeat_timeout` in consensus ticks. Every consensus group
-/// (metadata and per-partition planes alike) gets the same window: the failure
-/// it guards against - a primary that stopped heartbeating - is host-level,
not
-/// per-plane.
-pub(crate) fn cluster_heartbeat_ticks(config: &ServerConfig) -> u64 {
- duration_to_ticks(config.cluster.heartbeat_timeout.get_duration())
-}
-
/// Floor for the post-restart read-recovery deadline (see
/// [`recovery_barrier_deadline`]). At and below the 5s default heartbeat the
/// worst-case recovery is dominated by the heartbeat-independent term - the
@@ -2327,76 +2238,6 @@ pub(crate) fn recovery_barrier_deadline(
.max(RECOVERY_BARRIER_DEADLINE_FLOOR)
}
-/// `[cluster] commit_broadcast_interval` in consensus ticks: how often the
-/// primary broadcasts its commit point, the cluster's liveness feed. Applied
-/// to every consensus group, matching `cluster_heartbeat_ticks`.
-pub(crate) fn commit_broadcast_ticks(config: &ServerConfig) -> u64 {
- duration_to_ticks(config.cluster.commit_broadcast_interval.get_duration())
-}
-
-/// `[cluster] prepare_retransmit_interval` in consensus ticks: how often the
-/// primary retransmits un-acked prepares. Applied to every consensus group,
-/// matching `cluster_heartbeat_ticks`.
-pub(crate) fn prepare_retransmit_ticks(config: &ServerConfig) -> u64 {
-
duration_to_ticks(config.cluster.prepare_retransmit_interval.get_duration())
-}
-
-/// `[cluster] view_change_retransmit_interval` in consensus ticks: how often a
-/// replica retransmits its `StartViewChange` / `DoViewChange` during a view
-/// change. Applied to every consensus group, matching
`cluster_heartbeat_ticks`.
-pub(crate) fn view_change_retransmit_ticks(config: &ServerConfig) -> u64 {
- duration_to_ticks(
- config
- .cluster
- .view_change_retransmit_interval
- .get_duration(),
- )
-}
-
-/// `[cluster] view_change_status_timeout` in consensus ticks: the stalled
-/// view-change backstop before escalating to a fresh election. Applied to
every
-/// consensus group, matching `cluster_heartbeat_ticks`.
-pub(crate) fn view_change_status_ticks(config: &ServerConfig) -> u64 {
- duration_to_ticks(config.cluster.view_change_status_timeout.get_duration())
-}
-
-/// `[cluster] request_start_view_retransmit_interval` in consensus ticks: how
-/// often a recovering or view-change backup re-requests the current
`StartView`.
-/// Applied to every consensus group, matching `cluster_heartbeat_ticks`.
-pub(crate) fn request_start_view_ticks(config: &ServerConfig) -> u64 {
- duration_to_ticks(
- config
- .cluster
- .request_start_view_retransmit_interval
- .get_duration(),
- )
-}
-
-/// The full `[cluster]` timer set every consensus group boots with, built
-/// once so the planes cannot diverge in what they apply.
-pub(crate) fn consensus_timers(config: &ServerConfig) -> ConsensusTimers {
- ConsensusTimers {
- normal_heartbeat_ticks: cluster_heartbeat_ticks(config),
- commit_message_ticks: commit_broadcast_ticks(config),
- prepare_ticks: prepare_retransmit_ticks(config),
- view_change_retransmit_ticks: view_change_retransmit_ticks(config),
- view_change_status_ticks: view_change_status_ticks(config),
- request_start_view_ticks: request_start_view_ticks(config),
- probe_attempts_max: config.cluster.view_probe_attempts_max,
- }
-}
-
-/// `[cluster] repair_retry_interval` in consensus ticks: how long a stalled
-/// journal-repair stream waits before re-requesting its window. Both planes'
-/// repair loops share it, so it is applied once per shard (not per consensus
-/// group). Clamped to `u32`, the width of the session idle-tick counter.
-pub(crate) fn repair_retry_ticks(config: &ServerConfig) -> u32 {
- u32::try_from(duration_to_ticks(
- config.cluster.repair_retry_interval.get_duration(),
- ))
- .unwrap_or(u32::MAX)
-}
-
/// Shard 0's half of a metadata recovery: everything [`recover`] produced
except the
/// state machine, which every shard receives through the factory bundle.
///
diff --git a/core/server/src/consumer_group.rs
b/core/server/src/consumer_group.rs
index 00a6df558..d4b77eb95 100644
--- a/core/server/src/consumer_group.rs
+++ b/core/server/src/consumer_group.rs
@@ -25,8 +25,8 @@
//! primary enriches the op here before replication, mirroring the PAT mint
//! in [`crate::pat`] and the password hash in [`crate::users`].
-use crate::bootstrap::{ShellBus, ShellShard};
use crate::responses::resolve_partition_namespace;
+use crate::shell::{ShellBus, ShellShard};
use crate::wire::{request_body, rewrite_request_body};
use consensus::MetadataHandle;
use iggy_binary_protocol::PrepareHeader;
@@ -60,7 +60,7 @@ use std::rc::Rc;
/// (via the partition-read mesh), so the cooperative rebalance pending-revokes
/// only those and hands off never-polled/drained partitions synchronously at
/// join. Every other operation passes through.
-pub(crate) async fn maybe_rewrite_consumer_group_request<B, MJ, S, SB>(
+pub async fn maybe_rewrite_consumer_group_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
request: Message<RoutedRequestHeader>,
) -> Result<Message<RoutedRequestHeader>, IggyError>
@@ -205,7 +205,7 @@ where
/// purge agree, and a re-created group (new id) never inherits a stale offset.
/// Individual-consumer ops and every other operation pass through untouched.
#[allow(clippy::cast_possible_truncation)]
-pub(crate) fn maybe_rewrite_consumer_offset_request<B, MJ, S, SB>(
+pub fn maybe_rewrite_consumer_offset_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
request: Message<RoutedRequestHeader>,
) -> Result<Message<RoutedRequestHeader>, IggyError>
diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs
index 0a494f524..784b86bdf 100644
--- a/core/server/src/dispatch.rs
+++ b/core/server/src/dispatch.rs
@@ -28,7 +28,6 @@ use crate::auth::{
complete_login_register, surface_login_failure, verify_login_credentials,
verify_pat_credentials,
};
-use crate::bootstrap::{ShellBus, ShellShard, ShellShardHandle};
use crate::cluster_meta::ClusterRoster;
use crate::consumer_group::{
maybe_rewrite_consumer_group_request,
maybe_rewrite_consumer_offset_request,
@@ -48,6 +47,7 @@ use crate::responses::{
};
use crate::segment_cleaner::UNENFORCEABLE_TOPIC_SIZE_WARN;
use crate::session_manager::SessionManager;
+use crate::shell::{ShellBus, ShellShard, ShellShardHandle};
use crate::snapshot;
use crate::users::maybe_rewrite_user_password_request;
use crate::wire::{request_body, usize_to_u32, verify_request_checksum};
@@ -130,10 +130,10 @@ use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, warn};
-pub(crate) type ClientRequestQueues = Rc<RefCell<HashMap<u128,
VecDeque<Message<GenericHeader>>>>>;
-pub(crate) type ActiveClientRequests = Rc<RefCell<HashSet<u128>>>;
+pub type ClientRequestQueues = Rc<RefCell<HashMap<u128,
VecDeque<Message<GenericHeader>>>>>;
+pub type ActiveClientRequests = Rc<RefCell<HashSet<u128>>>;
-pub(crate) fn make_client_request_handler<B, MJ, S, SB>(
+pub fn make_client_request_handler<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
sessions: &Rc<RefCell<SessionManager>>,
system_config: Arc<ServerSystemConfig>,
@@ -181,9 +181,7 @@ where
/// its `SessionManager` and push them back over the reply sender. The
/// aggregation across all shards happens in
/// [`shard::IggyShard::list_all_clients`].
-pub(crate) fn make_list_clients_handler(
- sessions: &Rc<RefCell<SessionManager>>,
-) -> ListClientsHandler {
+pub fn make_list_clients_handler(sessions: &Rc<RefCell<SessionManager>>) ->
ListClientsHandler {
let sessions = Rc::clone(sessions);
Rc::new(move |reply| {
let clients: Vec<ConnectedClientInfo> =
sessions.borrow().iter_clients().collect();
@@ -198,7 +196,7 @@ pub(crate) fn make_list_clients_handler(
/// against the local partitions plane and push the result back over the
/// carried reply sender. The requesting shard bounds the wait with a
/// timeout, so a dropped reply degrades to a client-visible read failure.
-pub(crate) fn make_partition_read_handler<B, MJ, S, SB>(
+pub fn make_partition_read_handler<B, MJ, S, SB>(
shard_handle: &ShellShardHandle<B, MJ, S, SB>,
) -> PartitionReadHandler
where
@@ -451,7 +449,7 @@ fn build_auto_commit_request(
)
}
-pub(crate) fn make_deferred_replica_message_handler<B, MJ, S, SB>(
+pub fn make_deferred_replica_message_handler<B, MJ, S, SB>(
shard_handle: &ShellShardHandle<B, MJ, S, SB>,
) -> MessageHandler
where
@@ -469,7 +467,7 @@ where
})
}
-pub(crate) fn make_deferred_client_request_handler<B, MJ, S, SB>(
+pub fn make_deferred_client_request_handler<B, MJ, S, SB>(
bus: &B,
shard_handle: &ShellShardHandle<B, MJ, S, SB>,
sessions: &Rc<RefCell<SessionManager>>,
@@ -538,7 +536,7 @@ where
/// proposal. Spawns a task so the awaiting peer is woken once the op
/// commits. Submit failures are returned verbatim so the peer can preserve
/// unknown-outcome retry semantics.
-pub(crate) fn make_metadata_submit_handler<B, MJ, S, SB>(
+pub fn make_metadata_submit_handler<B, MJ, S, SB>(
shard_handle: &ShellShardHandle<B, MJ, S, SB>,
) -> shard::MetadataSubmitHandler
where
@@ -790,7 +788,7 @@ fn pop_next_client_request(
/// Zero passes here because a zero-partition TOPIC is legal (legacy
/// `create_topic` admits `0..=MAX`); the add/remove requests reject it in
/// [`validate_partitions_change_count`].
-pub(crate) const fn validate_partitions_count(partitions_count: u32) ->
Result<(), IggyError> {
+pub const fn validate_partitions_count(partitions_count: u32) -> Result<(),
IggyError> {
if partitions_count > MAX_PARTITIONS_PER_REQUEST {
return Err(IggyError::TooManyPartitions);
}
@@ -803,9 +801,7 @@ pub(crate) const fn
validate_partitions_count(partitions_count: u32) -> Result<(
/// force every shard through a rebalance pass. Legacy rejects it with
/// `TooManyPartitions` in both handlers (`1..=MAX` on create, `== 0` on
/// delete), so the code matches rather than inventing a new one.
-pub(crate) const fn validate_partitions_change_count(
- partitions_count: u32,
-) -> Result<(), IggyError> {
+pub const fn validate_partitions_change_count(partitions_count: u32) ->
Result<(), IggyError> {
if partitions_count == 0 {
return Err(IggyError::TooManyPartitions);
}
@@ -820,7 +816,7 @@ pub(crate) const fn validate_partitions_change_count(
/// `segment_size_bytes` is the topic's RESOLVED segment size (explicit
/// option, else this node's default), so a per-topic segment above the
/// global default still floors the topic cap.
-pub(crate) fn validate_topic_bounds(
+pub fn validate_topic_bounds(
partitions_count: u32,
max_topic_size: MaxTopicSize,
segment_size_bytes: u64,
@@ -832,7 +828,7 @@ pub(crate) fn validate_topic_bounds(
/// A topic cap below one segment can never be enforced: the first segment
/// already exceeds it. Split out of [`validate_topic_bounds`] because update
/// admission checks the cap without a partitions count to check.
-pub(crate) fn validate_topic_size_floor(
+pub fn validate_topic_size_floor(
max_topic_size: MaxTopicSize,
segment_size_bytes: u64,
) -> Result<(), IggyError> {
@@ -858,7 +854,7 @@ pub(crate) fn validate_topic_size_floor(
///
/// Warns rather than rejects: which caps are accepted is client-visible wire
/// behavior, and tightening it would break topics that already exist.
-pub(crate) fn warn_unenforceable_topic_size(
+pub fn warn_unenforceable_topic_size(
max_topic_size: MaxTopicSize,
segment_size_bytes: u64,
max_message_size_bytes: usize,
@@ -888,7 +884,7 @@ pub(crate) fn warn_unenforceable_topic_size(
/// partition shrinks the share: a cap that cleared the floor when the topic
was
/// created can stop clearing it here. The request carries only the delta, so
/// the stored cap, segment size and current partition count come from
metadata.
-pub(crate) fn warn_unenforceable_topic_size_on_partition_add(
+pub fn warn_unenforceable_topic_size_on_partition_add(
streams: &Streams,
stream_id: &WireIdentifier,
topic_id: &WireIdentifier,
@@ -920,7 +916,7 @@ pub(crate) fn
warn_unenforceable_topic_size_on_partition_add(
/// keys are rejected rather than skipped: a silently ignored knob would hand
/// the client server defaults without it ever learning. Streams and users
/// have no catalog keys yet, so `known` is empty for both until one lands.
-pub(crate) fn validate_option_keys(options: &WireOptions, known: &[&str]) ->
Result<(), IggyError> {
+pub fn validate_option_keys(options: &WireOptions, known: &[&str]) ->
Result<(), IggyError> {
for entry in options {
// Wire validation already enforced UTF-8 string keys.
let key = String::from_utf8_lossy(entry.key);
@@ -1468,7 +1464,7 @@ async fn handle_get_me<B, MJ, S, SB>(
/// `vsr_client_id` keys the consumer-group offset fence (the member id),
/// not the transport id stamped into the partition-op header.
#[allow(clippy::future_not_send)]
-pub(crate) async fn dispatch_partition_request<B, MJ, S, SB>(
+pub async fn dispatch_partition_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
request: Message<RoutedRequestHeader>,
vsr_client_id: u128,
@@ -2007,7 +2003,7 @@ async fn send_unauthenticated_eviction<B, MJ, S, SB>(
/// groups + rebalances via the replicated `Logout`) and sends a session-
/// terminal `Eviction(StaleClient)` so the client fails fast and can
reconnect.
#[allow(clippy::future_not_send)]
-pub(crate) async fn run_heartbeat_verifier<B, MJ, S, SB>(
+pub async fn run_heartbeat_verifier<B, MJ, S, SB>(
shard: Rc<ShellShard<B, MJ, S, SB>>,
sessions: Rc<RefCell<SessionManager>>,
interval: std::time::Duration,
@@ -2474,14 +2470,14 @@ fn empty_polled_messages_body(partition_id: u32) ->
Bytes {
Bytes::from(body)
}
-pub(crate) type DecodedPollRequest = (IggyNamespace, u32, PollingConsumer,
PollingArgs);
+pub type DecodedPollRequest = (IggyNamespace, u32, PollingConsumer,
PollingArgs);
/// Resolve a decoded poll request into its owning-shard read: namespace,
/// partition, polling consumer, and args. Shared by the TCP dispatch (client
/// id = the connection's bound VSR client) and the HTTP route (client id 0,
/// which fences group polls closed).
#[allow(clippy::cast_possible_truncation)]
-pub(crate) fn resolve_poll_request<B, MJ, S, SB>(
+pub fn resolve_poll_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
wire: &PollMessagesRequest,
client_id: u128,
@@ -2548,7 +2544,7 @@ where
/// namespace, partition, and polling consumer. Shared by the TCP dispatch and
/// the HTTP route; needs no client id because offset reads are not fenced
/// (any client may read a group's offset, member or not).
-pub(crate) fn resolve_consumer_offset_request<B, MJ, S, SB>(
+pub fn resolve_consumer_offset_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
wire: &GetConsumerOffsetRequest,
) -> Result<(IggyNamespace, u32, PollingConsumer), IggyError>
@@ -3082,7 +3078,7 @@ fn build_forward_logout_result_message(
/// reply (shard-0 inbox full / shutdown) maps to a transient `Canceled`, which
/// the caller wraps so the SDK replays.
#[allow(clippy::future_not_send)]
-pub(crate) async fn submit_register_on_owner<B, MJ, S, SB>(
+pub async fn submit_register_on_owner<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
vsr_client_id: u128,
user_id: u32,
@@ -3112,7 +3108,7 @@ where
/// Logout counterpart of [`submit_register_on_owner`].
#[allow(clippy::future_not_send)]
-pub(crate) async fn submit_logout_on_owner<B, MJ, S, SB>(
+pub async fn submit_logout_on_owner<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
vsr_client_id: u128,
session: u64,
@@ -3283,7 +3279,7 @@ async fn handle_delete_segments_request<B, MJ, S, SB>(
/// the error.
#[allow(clippy::future_not_send)]
#[allow(clippy::cast_possible_truncation)]
-pub(crate) async fn resolve_delete_segments_truncate<B, MJ, S, SB>(
+pub async fn resolve_delete_segments_truncate<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
template: &RoutedRequestHeader,
client_id: u128,
@@ -3445,7 +3441,7 @@ fn submit_disconnect_logout<B, MJ, S, SB>(
/// `client` id (it's the VSR id, not the transport/home-shard-encoding id).
/// `None` = transient submit failure (SDK read-timeout replays).
#[allow(clippy::future_not_send)]
-pub(crate) async fn submit_client_request_on_owner<B, MJ, S, SB>(
+pub async fn submit_client_request_on_owner<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
request: Message<RoutedRequestHeader>,
) -> Option<Message<GenericHeader>>
@@ -3759,7 +3755,7 @@ async fn handle_login_register_request<B, MJ, S, SB>(
/// metadata shard and zeroed elsewhere -- the SDK only reads the reason,
/// plus the protocol window on `IncompatibleProtocol`.
#[allow(clippy::future_not_send)]
-pub(crate) async fn send_login_eviction<B, MJ, S, SB>(
+pub async fn send_login_eviction<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
vsr_client_id: u128,
@@ -3799,7 +3795,7 @@ pub(crate) async fn send_login_eviction<B, MJ, S, SB>(
}
}
-pub(crate) fn upgrade_shard_handle<B, MJ, S, SB>(
+pub fn upgrade_shard_handle<B, MJ, S, SB>(
shard_handle: &ShellShardHandle<B, MJ, S, SB>,
) -> Option<Rc<ShellShard<B, MJ, S, SB>>>
where
diff --git a/core/server/src/dispatch/authz.rs
b/core/server/src/dispatch/authz.rs
index 5e8790e7a..ee113e0d1 100644
--- a/core/server/src/dispatch/authz.rs
+++ b/core/server/src/dispatch/authz.rs
@@ -50,10 +50,10 @@ use metadata::permissioner::Permissioner;
use server_common::Message;
use tracing::warn;
-use crate::bootstrap::{ShellBus, ShellShard};
use crate::responses::{
build_deny_reply, current_metadata_commit, resolve_stream_id,
resolve_topic_id,
};
+use crate::shell::{ShellBus, ShellShard};
/// Authorize a partition-plane op on its resolved (stream, topic) for the
/// acting user, returning the deny status code or `None` to proceed. The
diff --git a/core/server/src/http.rs b/core/server/src/http.rs
index ed670883d..25a91eff5 100644
--- a/core/server/src/http.rs
+++ b/core/server/src/http.rs
@@ -36,6 +36,8 @@ mod session;
mod state;
mod submit;
mod tls;
+#[cfg(feature = "iggy-web")]
+mod web;
mod wire;
use std::cell::{Cell, RefCell};
@@ -64,7 +66,6 @@ use send_wrapper::SendWrapper;
use tower_http::cors::{AllowOrigin, CorsLayer};
use tracing::{error, info, warn};
-use crate::bootstrap::ServerShard;
use crate::cluster_meta::{ClusterRoster, resolved_roster_nodes};
use crate::http::handlers::{
change_password, create_cg, create_partitions, create_pat, create_stream,
create_topic,
@@ -80,6 +81,7 @@ use crate::http::jwt::JwtManager;
use crate::http::session::RegistrationBarrier;
use crate::http::state::{HttpInner, HttpState, insert_view_header};
use crate::server_error::ServerError;
+use crate::shell::ServerShard;
/// Bind the shard-0 HTTP listener and spawn the `cyper-axum` serve loop as a
/// background task on shard 0's compio runtime. Serves HTTPS when
@@ -300,7 +302,7 @@ fn router(
.route("/clients", get(get_clients))
.route("/clients/{client_id}", get(get_client));
let local = match metrics_endpoint {
- Some(endpoint) => local.route(endpoint, get(metrics::get_metrics)),
+ Some(endpoint) => local.route(endpoint, get(handlers::get_metrics)),
None => local,
};
let router = Router::new()
@@ -458,7 +460,7 @@ fn merge_web_ui(router: Router, web_ui: bool) -> Router {
#[cfg(feature = "iggy-web")]
let router = if web_ui {
info!("Web UI enabled at /ui");
- router.merge(crate::web::router())
+ router.merge(web::router())
} else {
router
};
diff --git a/core/server/src/http/forward.rs b/core/server/src/http/forward.rs
index 8c1acadcd..b0ead9be5 100644
--- a/core/server/src/http/forward.rs
+++ b/core/server/src/http/forward.rs
@@ -82,7 +82,7 @@ use crate::http::error::{
CustomError, error_response, gateway_timeout_response,
primary_http_socket, with_retry_after,
};
use crate::http::extractor::{bearer_token, resolve_credential};
-use crate::http::state::{HttpInner, VIEW_HEADER};
+use crate::http::state::{ForwardState, HttpInner, VIEW_HEADER};
use crate::server_error::ServerError;
/// Marker stamped on every forwarded request. Loop guard only: a node that is
@@ -135,23 +135,6 @@ const RESPONSE_CAPACITY_HINT: usize = 64 * 1024;
/// (the view layer only fills the header when absent).
const RELAYED_RESPONSE_HEADERS: [HeaderName; 3] = [CONTENT_TYPE, RETRY_AFTER,
VIEW_HEADER];
-/// Per-node forwarding context hung off `HttpInner`: the outbound client
-/// (pinned-cert TLS when the listener serves HTTPS), the scheme it dials, the
-/// request-body buffer bound, and the in-flight budget.
-pub(in crate::http) struct ForwardState {
- /// False when no cluster-wide bearer key material exists (no configured
- /// JWT secret, no cluster PSK): a forwarded bearer would 401 on the
- /// primary, so the middleware passes through and followers answer with
- /// the transient 503 instead.
- active: bool,
- client: cyper::Client,
- /// Also read by the 307 redirect builder: the primary is assumed to serve
- /// the same scheme as this node (uniform cluster HTTP config).
- pub(in crate::http) scheme: &'static str,
- body_limit: usize,
- in_flight: Cell<u32>,
-}
-
/// Build the [`ForwardState`] at listener startup.
///
/// With `http.tls.enabled` the forward hop dials `https` and verifies the peer
diff --git a/core/server/src/http/handlers.rs b/core/server/src/http/handlers.rs
index c1172a4cc..cdce21b5c 100644
--- a/core/server/src/http/handlers.rs
+++ b/core/server/src/http/handlers.rs
@@ -129,6 +129,7 @@ use crate::http::error::{
ReadError, WriteError,
};
use crate::http::extractor::{Authenticated, Identity};
+use crate::http::metrics::gauge_value;
use crate::http::reads::{
authorize_data_plane, authorize_read, read_local, resolve_gate_stream,
resolve_gate_topic,
resolve_gate_topic_ids, resolve_gate_user,
@@ -600,6 +601,72 @@ pub(in crate::http) async fn get_stats(
Ok(Json(Stats::from(response)))
}
+/// `GET <http.metrics.endpoint>`: the metric set in prometheus text
+/// exposition. Auth-only, like `/stats`: the `Identity` extractor rejects a
+/// missing or invalid bearer with 401, and any authenticated user may scrape
+/// (no RBAC rule guards it). Scrapers present a JWT or a raw PAT the same way
+/// every read route accepts them.
+///
+/// The entity gauges sample the same reads `/stats` serves: the metadata STM
+/// stream and user maps plus the stats-registry rollups, whose partition-plane
+/// increments are relaxed, so scraped values are approximate while writes are
+/// in flight. The clients count scatter-gathers the per-shard session managers
+/// exactly like `GET /clients` and turns partial when a shard misses the reply
+/// deadline.
+pub(in crate::http) async fn get_metrics(
+ State(state): State<HttpState>,
+ _identity: Identity,
+) -> String {
+ let (streams_count, topics_count, partitions_count, segments_count,
messages_count) = state
+ .shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .read(|streams| {
+ let mut topics_count = 0u64;
+ let mut partitions_count = 0u64;
+ let mut segments_count = 0u64;
+ let mut messages_count = 0u64;
+ for (_, stream) in &streams.items {
+ topics_count = topics_count.saturating_add(stream.topics.len()
as u64);
+ segments_count = segments_count
+
.saturating_add(u64::from(stream.stats.segments_count_inconsistent()));
+ messages_count =
+
messages_count.saturating_add(stream.stats.messages_count_inconsistent());
+ for (_, topic) in &stream.topics {
+ partitions_count =
+ partitions_count.saturating_add(topic.partitions.len()
as u64);
+ }
+ }
+ (
+ streams.items.len() as u64,
+ topics_count,
+ partitions_count,
+ segments_count,
+ messages_count,
+ )
+ });
+ let users_count = state
+ .shard
+ .plane
+ .metadata()
+ .mux_stm
+ .users()
+ .read(|users| users.items.len() as u64);
+ let clients_count =
SendWrapper::new(state.shard.list_all_clients()).await.len() as u64;
+
+ let metrics = &state.metrics;
+ metrics.streams.set(gauge_value(streams_count));
+ metrics.topics.set(gauge_value(topics_count));
+ metrics.partitions.set(gauge_value(partitions_count));
+ metrics.segments.set(gauge_value(segments_count));
+ metrics.messages.set(gauge_value(messages_count));
+ metrics.users.set(gauge_value(users_count));
+ metrics.clients.set(gauge_value(clients_count));
+ metrics.formatted_output()
+}
+
/// `POST /snapshot`: collect a diagnostic archive and return it as a ZIP
/// download with the same headers the legacy server sets.
///
diff --git a/core/server/src/http/metrics.rs b/core/server/src/http/metrics.rs
index 3fdb1b423..6546c00c1 100644
--- a/core/server/src/http/metrics.rs
+++ b/core/server/src/http/metrics.rs
@@ -16,40 +16,35 @@
// under the License.
//! The `[http.metrics]` scrape surface: the legacy-parity metric registry
-//! (entity gauges plus the request counter), its public scrape handler, and
-//! the config gate deciding whether the route is mounted.
+//! (entity gauges plus the request counter) and the config gate deciding
+//! whether the route is mounted. The scrape handler itself lives with the
+//! other route handlers so this leaf never imports the state hub.
-use axum::extract::State;
use configs::http::HttpMetricsConfig;
-use consensus::MetadataHandle;
use iggy_common::IggyError;
-use metadata::impls::metadata::StreamsFrontend;
use prometheus_client::encoding::text::encode;
use prometheus_client::metrics::counter::Counter;
use prometheus_client::metrics::gauge::Gauge;
use prometheus_client::registry::Registry;
-use send_wrapper::SendWrapper;
use tracing::error;
-use crate::http::extractor::Identity;
-use crate::http::state::HttpState;
-
/// The legacy server's metric set, registered under the same names and help
/// texts so existing dashboards and alerts keep working unchanged.
///
/// Unlike the legacy server, the entity gauges are not counted at mutation
-/// sites: [`get_metrics`] samples the live state on every scrape, so a gauge
-/// can never drift from the state it describes.
+/// sites: the scrape handler (`http::handlers::get_metrics`) samples the live
+/// state on every scrape, so a gauge can never drift from the state it
+/// describes.
pub(in crate::http) struct HttpMetrics {
registry: Registry,
http_requests: Counter,
- streams: Gauge,
- topics: Gauge,
- partitions: Gauge,
- segments: Gauge,
- messages: Gauge,
- users: Gauge,
- clients: Gauge,
+ pub(in crate::http) streams: Gauge,
+ pub(in crate::http) topics: Gauge,
+ pub(in crate::http) partitions: Gauge,
+ pub(in crate::http) segments: Gauge,
+ pub(in crate::http) messages: Gauge,
+ pub(in crate::http) users: Gauge,
+ pub(in crate::http) clients: Gauge,
}
impl HttpMetrics {
@@ -111,7 +106,7 @@ impl HttpMetrics {
self.http_requests.clone()
}
- fn formatted_output(&self) -> String {
+ pub(in crate::http) fn formatted_output(&self) -> String {
let mut buffer = String::new();
if let Err(error) = encode(&mut buffer, &self.registry) {
error!(%error, "failed to encode metrics");
@@ -147,75 +142,9 @@ pub(in crate::http) fn validated_endpoint(
Ok(Some(config.endpoint.clone()))
}
-/// `GET <http.metrics.endpoint>`: the metric set in prometheus text
-/// exposition. Auth-only, like `/stats`: the `Identity` extractor rejects a
-/// missing or invalid bearer with 401, and any authenticated user may scrape
-/// (no RBAC rule guards it). Scrapers present a JWT or a raw PAT the same way
-/// every read route accepts them.
-///
-/// The entity gauges sample the same reads `/stats` serves: the metadata STM
-/// stream and user maps plus the stats-registry rollups, whose partition-plane
-/// increments are relaxed, so scraped values are approximate while writes are
-/// in flight. The clients count scatter-gathers the per-shard session managers
-/// exactly like `GET /clients` and turns partial when a shard misses the reply
-/// deadline.
-pub(in crate::http) async fn get_metrics(
- State(state): State<HttpState>,
- _identity: Identity,
-) -> String {
- let (streams_count, topics_count, partitions_count, segments_count,
messages_count) = state
- .shard
- .plane
- .metadata()
- .mux_stm
- .streams()
- .read(|streams| {
- let mut topics_count = 0u64;
- let mut partitions_count = 0u64;
- let mut segments_count = 0u64;
- let mut messages_count = 0u64;
- for (_, stream) in &streams.items {
- topics_count = topics_count.saturating_add(stream.topics.len()
as u64);
- segments_count = segments_count
-
.saturating_add(u64::from(stream.stats.segments_count_inconsistent()));
- messages_count =
-
messages_count.saturating_add(stream.stats.messages_count_inconsistent());
- for (_, topic) in &stream.topics {
- partitions_count =
- partitions_count.saturating_add(topic.partitions.len()
as u64);
- }
- }
- (
- streams.items.len() as u64,
- topics_count,
- partitions_count,
- segments_count,
- messages_count,
- )
- });
- let users_count = state
- .shard
- .plane
- .metadata()
- .mux_stm
- .users()
- .read(|users| users.items.len() as u64);
- let clients_count =
SendWrapper::new(state.shard.list_all_clients()).await.len() as u64;
-
- let metrics = &state.metrics;
- metrics.streams.set(gauge_value(streams_count));
- metrics.topics.set(gauge_value(topics_count));
- metrics.partitions.set(gauge_value(partitions_count));
- metrics.segments.set(gauge_value(segments_count));
- metrics.messages.set(gauge_value(messages_count));
- metrics.users.set(gauge_value(users_count));
- metrics.clients.set(gauge_value(clients_count));
- metrics.formatted_output()
-}
-
/// Clamp a count into the gauge's `i64` domain; only `messages` can pass
/// `i64::MAX` even in theory, the rest are bounded far below it.
-fn gauge_value(count: u64) -> i64 {
+pub(in crate::http) fn gauge_value(count: u64) -> i64 {
i64::try_from(count).unwrap_or(i64::MAX)
}
diff --git a/core/server/src/http/reads.rs b/core/server/src/http/reads.rs
index 08e743f47..3859b28ae 100644
--- a/core/server/src/http/reads.rs
+++ b/core/server/src/http/reads.rs
@@ -19,7 +19,7 @@
//! metadata-STM read entry, and the wire/domain identifier resolvers the read
//! and data-plane routes ground their scopes through.
-use crate::bootstrap::ServerShard;
+use crate::shell::ServerShard;
use bytes::Bytes;
use consensus::MetadataHandle;
use iggy_binary_protocol::WireIdentifier;
diff --git a/core/server/src/http/state.rs b/core/server/src/http/state.rs
index 8ea1bb486..c0c67e7de 100644
--- a/core/server/src/http/state.rs
+++ b/core/server/src/http/state.rs
@@ -37,17 +37,17 @@ use send_wrapper::SendWrapper;
use tokio::sync::Mutex;
use tracing::warn;
-use crate::bootstrap::ServerShard;
use crate::cluster_meta::ClusterRoster;
use crate::dispatch::submit_register_on_owner;
use crate::http::error::{AuthError, ReadError, primary_redirect_location};
-use crate::http::forward::ForwardState;
+
use crate::http::jwt::JwtManager;
use crate::http::metrics::HttpMetrics;
use crate::http::session::{
BarrierEntry, FIRST_REQUEST_ID, FRESH_ENTRY_WATERMARK, HttpSession,
RegistrationBarrier,
forget_if_same, live_entry, sweep_expired,
};
+use crate::shell::ServerShard;
/// Response header carrying the current VSR view number. Stamped by
/// `insert_view_header` on success and redirect responses only (never on
@@ -63,6 +63,25 @@ pub(in crate::http) const VIEW_HEADER: HeaderName =
HeaderName::from_static("igg
/// same thread that builds this state. Never touch it off that thread.
pub(in crate::http) type HttpState = SendWrapper<Rc<HttpInner>>;
+/// Per-node forwarding context hung off `HttpInner`: the outbound client
+/// (pinned-cert TLS when the listener serves HTTPS), the scheme it dials, the
+/// request-body buffer bound, and the in-flight budget. Built by
+/// `http::forward::build_forward_state`; lives here so the state hub never
+/// imports the forwarding middleware.
+pub(in crate::http) struct ForwardState {
+ /// False when no cluster-wide bearer key material exists (no configured
+ /// JWT secret, no cluster PSK): a forwarded bearer would 401 on the
+ /// primary, so the middleware passes through and followers answer with
+ /// the transient 503 instead.
+ pub(in crate::http) active: bool,
+ pub(in crate::http) client: cyper::Client,
+ /// Also read by the 307 redirect builder: the primary is assumed to serve
+ /// the same scheme as this node (uniform cluster HTTP config).
+ pub(in crate::http) scheme: &'static str,
+ pub(in crate::http) body_limit: usize,
+ pub(in crate::http) in_flight: Cell<u32>,
+}
+
/// Shared shard-0 HTTP state.
///
/// Groups the shard handle, the JWT issuer/verifier, and the per-credential
VSR
diff --git a/core/server/src/http/submit.rs b/core/server/src/http/submit.rs
index ca763d836..eeaf9dc1a 100644
--- a/core/server/src/http/submit.rs
+++ b/core/server/src/http/submit.rs
@@ -33,7 +33,6 @@ use metadata::impls::metadata::StreamsFrontend;
use server_common::Message;
use tracing::warn;
-use crate::bootstrap::ServerShard;
use crate::dispatch::{
dispatch_partition_request, resolve_delete_segments_truncate,
submit_client_request_on_owner,
submit_logout_on_owner,
@@ -47,6 +46,7 @@ use crate::http::session::HttpSession;
use crate::http::state::HttpInner;
use crate::http::wire::build_request_message;
use crate::pat::rewrite_pat_request_for_user;
+use crate::shell::ServerShard;
use crate::users::maybe_rewrite_user_password_request;
use crate::wire::request_body;
diff --git a/core/server/src/web.rs b/core/server/src/http/web.rs
similarity index 100%
rename from core/server/src/web.rs
rename to core/server/src/http/web.rs
diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs
index 2f5a00932..46ce02fb7 100644
--- a/core/server/src/lib.rs
+++ b/core/server/src/lib.rs
@@ -30,28 +30,43 @@ static GLOBAL: MiMalloc = MiMalloc;
pub const VERSION: &str = env!("CARGO_PKG_VERSION");
pub const SEMANTIC_VERSION: SemanticVersion =
SemanticVersion::parse_const(VERSION);
-pub mod auth;
+// Visibility rule: `pub` = named external consumer. main.rs consumes
+// `bootstrap`, `server_error`, and `systemd`; the simulator consumes `shell`,
+// `bootstrap::wire_shell_handlers`, and (through `ShellHandlers.sessions`)
+// `session_manager`. Everything else is crate-internal.
+
+// boot: process entry, shard threads, recovery orchestration.
pub mod bootstrap;
-pub(crate) mod cluster_meta;
-pub mod config_writer;
-pub mod consumer_group;
-pub mod dispatch;
+pub(crate) mod config_writer;
+pub(crate) mod shard_allocator;
+#[cfg(feature = "systemd")]
+pub mod systemd;
+
+// spine: the request path - shell vocabulary, dispatch funnel, per-domain ops.
+pub(crate) mod auth;
+pub(crate) mod consumer_group;
+pub(crate) mod dispatch;
+pub(crate) mod login_register;
+pub(crate) mod pat;
+pub(crate) mod responses;
+pub mod session_manager;
+pub mod shell;
+pub(crate) mod users;
+pub(crate) mod wire;
+
+// http: the REST spine (role-leaf tree).
pub(crate) mod http;
-pub mod login_register;
-pub(crate) mod offset_recovery;
-pub mod partition_helpers;
-pub mod partition_reconciler;
-pub mod pat;
+
+// background: per-shard maintenance loops.
+pub(crate) mod partition_reconciler;
pub(crate) mod personal_access_token_cleaner;
-pub mod responses;
pub(crate) mod segment_cleaner;
+pub(crate) mod snapshot;
+
+// support: shared plumbing and the crash-recovery readers
+// (the readers move beside their writers in core/partitions later).
+pub(crate) mod cluster_meta;
+pub(crate) mod offset_recovery;
+pub(crate) mod partition_helpers;
pub(crate) mod segment_recovery;
pub mod server_error;
-pub mod session_manager;
-pub(crate) mod snapshot;
-#[cfg(feature = "systemd")]
-pub mod systemd;
-pub mod users;
-#[cfg(feature = "iggy-web")]
-pub(crate) mod web;
-pub mod wire;
diff --git a/core/server/src/partition_helpers.rs
b/core/server/src/partition_helpers.rs
index a6f709880..91610720b 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -432,7 +432,7 @@ pub async fn ensure_initial_segment(
/// [`ServerError::PartitionSuperblockIo`] when the directory or a slot
/// cannot be read; the `VersionUnknown` / `Unverifiable` / `Undecodable` /
/// `IdentityMismatch` variants when a record exists but cannot be trusted.
-pub(crate) async fn open_partition_superblock(
+pub async fn open_partition_superblock(
partition_dir: &str,
identity: ReplicaIdentity,
) -> Result<(Rc<PingPongSuperblock>, Option<VsrState>), ServerError> {
@@ -609,7 +609,7 @@ pub async fn build_partition_fresh(
// Request queue holds 2x the prepare depth (buffered requests drain as
// prepares commit); depth is the per-partition `[partition]` knob.
let prepare_queue_depth = config.partition.prepare_queue_depth;
- let timers = crate::bootstrap::consensus_timers(config);
+ let timers = crate::shell::consensus_timers(config);
let consensus = VsrConsensus::restored(
cluster_id,
self_replica_id,
diff --git a/core/server/src/partition_reconciler.rs
b/core/server/src/partition_reconciler.rs
index dea36c9c0..5cbd5f19a 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -169,8 +169,8 @@
//! discriminator, like `checkpoint_id` on every prepare
//! -- `PrepareHeader.reserved` has room, but it is a `#[repr(C)]` wire change.
-use crate::bootstrap::ServerShard;
use crate::partition_helpers::{build_partition_fresh,
delete_partitions_from_disk};
+use crate::shell::ServerShard;
use ahash::{AHashMap, AHashSet};
use configs::server::ServerConfig;
use consensus::{MetadataHandle, PartitionsHandle};
diff --git a/core/server/src/pat.rs b/core/server/src/pat.rs
index 26c8ea6db..dfaf3f50f 100644
--- a/core/server/src/pat.rs
+++ b/core/server/src/pat.rs
@@ -37,7 +37,7 @@ use server_common::Message;
use std::cell::RefCell;
use std::rc::Rc;
-pub(crate) fn maybe_rewrite_pat_request(
+pub fn maybe_rewrite_pat_request(
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
max_tokens_per_user: u32,
@@ -65,7 +65,7 @@ pub(crate) fn maybe_rewrite_pat_request(
/// form scoped to `user_id` (`only_if_expired` false); every other operation
/// passes through unchanged with `None` (the HTTP write core routes all of
/// its ops here, so the non-PAT arm is a no-op, not `unreachable`).
-pub(crate) fn rewrite_pat_request_for_user(
+pub fn rewrite_pat_request_for_user(
user_id: u32,
max_tokens_per_user: u32,
pat_count_of: impl FnOnce(u32) -> usize,
diff --git a/core/server/src/personal_access_token_cleaner.rs
b/core/server/src/personal_access_token_cleaner.rs
index a55093605..61e7f585a 100644
--- a/core/server/src/personal_access_token_cleaner.rs
+++ b/core/server/src/personal_access_token_cleaner.rs
@@ -22,7 +22,7 @@
//! proposes it once and every replica applies the commit. Backups never
//! propose, so cleanup cannot race across the cluster.
-use crate::bootstrap::ServerShard;
+use crate::shell::ServerShard;
use consensus::MetadataHandle;
use iggy_binary_protocol::WireName;
use iggy_common::IggyTimestamp;
diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs
index 8165bdf63..6575456ae 100644
--- a/core/server/src/responses.rs
+++ b/core/server/src/responses.rs
@@ -23,9 +23,9 @@
//! `NonReplicatedResponse` dispatch shim and the partition-namespace
//! resolvers.
-use crate::bootstrap::{ShellBus, ShellShard};
use crate::cluster_meta::ClusterRoster;
use crate::session_manager::SessionManager;
+use crate::shell::{ShellBus, ShellShard};
use crate::wire::{transport_kind_to_wire, usize_to_u32};
use bytes::{Bytes, BytesMut};
use consensus::{MetadataHandle, VsrConsensus};
@@ -106,7 +106,7 @@ use system_stats::SystemProbe;
/// (`user_id`, transport kind, peer address) comes from the per-shard
/// [`SessionManager`]; the `consumer_groups` list is read from the
/// (replicated) consumer-group STM by the connection's bound VSR client id.
-pub(crate) fn build_get_personal_access_tokens_response<B, MJ, S, SB>(
+pub fn build_get_personal_access_tokens_response<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
@@ -143,7 +143,7 @@ where
})
}
-pub(crate) fn build_get_me_response<B, MJ, S, SB>(
+pub fn build_get_me_response<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
@@ -218,7 +218,7 @@ where
/// `consumer_groups_count` is resolved from the connection's bound VSR client
/// id against the replicated `Streams` STM (memberships are keyed by VSR id,
not
/// transport id). Connections that never bound (pre-register) count 0.
-pub(crate) fn connected_client_to_response<B, MJ, S, SB>(
+pub fn connected_client_to_response<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
info: &ConnectedClientInfo,
) -> ClientResponse
@@ -326,7 +326,7 @@ where
resolve_partition_namespace(shard, stream_id, topic_id, partition_id)
}
-pub(crate) fn resolve_partition_request_namespace<B, MJ, S, SB>(
+pub fn resolve_partition_request_namespace<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
operation: Operation,
body: &[u8],
@@ -431,7 +431,7 @@ where
)
}
-pub(crate) fn resolve_partition_namespace<B, MJ, S, SB>(
+pub fn resolve_partition_namespace<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
stream_id: &WireIdentifier,
topic_id: &WireIdentifier,
@@ -494,7 +494,7 @@ fn wire_identifier_for_display(id: &WireIdentifier) ->
Identifier {
/// connected-client total, used only by the stats read: it comes from the
async
/// `ListClients` scatter-gather, which this sync builder cannot run, so both
/// transport callers gather it up front (0 for every other opcode).
-pub(crate) fn build_non_replicated_response<B, MJ, S, SB>(
+pub fn build_non_replicated_response<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
code: u32,
body: &[u8],
@@ -886,7 +886,7 @@ static STATS_DATA_PATH: OnceLock<PathBuf> = OnceLock::new();
/// Capture the configured data directory for `GetStats` disk reporting.
/// Idempotent: only the first call (process bootstrap) takes effect.
-pub(crate) fn init_stats_data_path(path: PathBuf) {
+pub fn init_stats_data_path(path: PathBuf) {
let _ = STATS_DATA_PATH.set(path);
}
@@ -1279,7 +1279,7 @@ fn topic_not_found(stream_id: &WireIdentifier, topic_id:
&WireIdentifier) -> Igg
)
}
-pub(crate) fn resolve_stream_id(
+pub fn resolve_stream_id(
streams: &metadata::stm::stream::StreamsInner,
identifier: &WireIdentifier,
) -> Option<usize> {
@@ -1292,7 +1292,7 @@ pub(crate) fn resolve_stream_id(
}
}
-pub(crate) fn resolve_topic_id(
+pub fn resolve_topic_id(
streams: &metadata::stm::stream::StreamsInner,
stream_id: usize,
identifier: &WireIdentifier,
@@ -1409,7 +1409,7 @@ fn partition_response(
})
}
-pub(crate) enum NonReplicatedResponse {
+pub enum NonReplicatedResponse {
Empty,
Bytes(Bytes),
}
@@ -1431,7 +1431,7 @@ impl NonReplicatedResponse {
}
}
-pub(crate) fn build_empty_reply(
+pub fn build_empty_reply(
request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
@@ -1447,7 +1447,7 @@ pub(crate) fn build_empty_reply(
/// body, nonzero status); op carries the builder's session argument like every
/// reply, and only the partition primary's pre-pipeline deny pins it to 0,
/// stamped through `consensus::build_deny_reply_from_request`.
-pub(crate) fn build_deny_reply(
+pub fn build_deny_reply(
request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
@@ -1504,7 +1504,7 @@ fn build_result_framed_reply(
)
}
-pub(crate) fn build_login_register_reply(
+pub fn build_login_register_reply(
request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
@@ -1523,7 +1523,7 @@ pub(crate) fn build_login_register_reply(
build_result_framed_reply(request_header, client_id, session, commit,
&payload)
}
-pub(crate) fn build_reply_from_bytes(
+pub fn build_reply_from_bytes(
request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
@@ -1546,7 +1546,7 @@ pub(crate) fn build_reply_from_bytes(
/// reusing the confirmed commit position from the committed reply. Otherwise
/// (no token, a committed business rejection, or an eviction frame) the
/// committed reply passes through unchanged.
-pub(crate) fn build_raw_pat_reply(
+pub fn build_raw_pat_reply(
request_header: &RoutedRequestHeader,
committed: Message<GenericHeader>,
raw_token: Option<String>,
@@ -1600,7 +1600,7 @@ pub(crate) fn build_raw_pat_reply(
Ok(reply.into_generic())
}
-pub(crate) fn build_reply_with_body(
+pub fn build_reply_with_body(
request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
@@ -1636,7 +1636,7 @@ pub(crate) fn build_reply_with_body(
reply
}
-pub(crate) fn current_metadata_commit<B, MJ, S, SB>(shard: &Rc<ShellShard<B,
MJ, S, SB>>) -> u64
+pub fn current_metadata_commit<B, MJ, S, SB>(shard: &Rc<ShellShard<B, MJ, S,
SB>>) -> u64
where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -1663,7 +1663,7 @@ where
/// decrypt point, so encrypted records are rebuilt over the plaintext.
///
/// Body layout: `[partition_id:4][current_offset:8][count:4][batch
records...]`.
-pub(crate) fn build_polled_messages_body(
+pub fn build_polled_messages_body(
partition_id: u32,
current_offset: u64,
fragments: PollFragments,
@@ -1715,7 +1715,7 @@ pub(crate) fn build_polled_messages_body(
/// Build the `ConsumerOffsetResponse` reply body:
/// `[partition_id:4][current_offset:8][stored_offset:8]`.
-pub(crate) fn build_consumer_offset_body(
+pub fn build_consumer_offset_body(
partition_id: u32,
current_offset: u64,
stored_offset: u64,
diff --git a/core/server/src/segment_cleaner.rs
b/core/server/src/segment_cleaner.rs
index eac91c156..724666b24 100644
--- a/core/server/src/segment_cleaner.rs
+++ b/core/server/src/segment_cleaner.rs
@@ -26,7 +26,7 @@
//! with reads. This mirrors the legacy server's `MessagesCleaner` ->
//! message-pump `CleanTopicMessages` path.
-use crate::bootstrap::ServerShard;
+use crate::shell::ServerShard;
use consensus::{MetadataHandle, PartitionsHandle};
use iggy_common::{IggyExpiry, IggyTimestamp, MaxTopicSize};
use metadata::impls::metadata::StreamsFrontend;
diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs
index d5343d137..53e84e46d 100644
--- a/core/server/src/server_error.rs
+++ b/core/server/src/server_error.rs
@@ -15,11 +15,11 @@
// specific language governing permissions and limitations
// under the License.
+use crate::shard_allocator::ShardingError;
use consensus::VsrStateError;
use metadata::impls::recovery::RecoveryError;
use server_common::log::LogError;
use shard::ShardCtorError;
-use shard_allocator::ShardingError;
use std::path::PathBuf;
use thiserror::Error;
diff --git a/core/shard_allocator/src/lib.rs
b/core/server/src/shard_allocator.rs
similarity index 93%
rename from core/shard_allocator/src/lib.rs
rename to core/server/src/shard_allocator.rs
index f0aadae26..e71ddbfac 100644
--- a/core/shard_allocator/src/lib.rs
+++ b/core/server/src/shard_allocator.rs
@@ -15,10 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-//! `shard_allocator`: decide which CPU cores each shard lives on.
+//! Decide which CPU cores each shard lives on.
//!
//! The server makes many shards and wants each one to run on its own
-//! core so they do not fight over CPU time. This crate reads the
+//! core so they do not fight over CPU time. This module reads the
//! operator's choice ([`CpuAllocation`] from the config), looks at the
//! real machine with `hwloc`, and hands back one [`ShardInfo`] per
//! shard. On Linux it also pins each shard's thread to its core and
@@ -97,7 +97,7 @@ pub struct NumaTopology {
impl NumaTopology {
/// Ask `hwloc` to read this machine's NUMA layout right now.
/// Errors if hwloc fails or the machine reports no NUMA nodes.
- pub fn detect() -> Result<NumaTopology, ShardingError> {
+ pub fn detect() -> Result<Self, ShardingError> {
let topology =
Topology::new().map_err(|e| ShardingError::TopologyDetection {
msg: e.to_string() })?;
@@ -113,20 +113,19 @@ impl NumaTopology {
let mut logical_cores_per_node = Vec::new();
for node in numa_nodes {
- let cpuset = node.cpuset().ok_or(ShardingError::TopologyDetection {
- msg: "NUMA node has no CPU set".to_string(),
- })?;
+ let cpuset = node
+ .cpuset()
+ .ok_or_else(|| ShardingError::TopologyDetection {
+ msg: "NUMA node has no CPU set".to_string(),
+ })?;
let logical_cores = cpuset.weight().unwrap_or(0);
let physical_cores = topology
.objects_with_type(ObjectType::Core)
.filter(|core| {
- if let Some(core_cpuset) = core.cpuset() {
- !(cpuset & core_cpuset).is_empty()
- } else {
- false
- }
+ core.cpuset()
+ .is_some_and(|core_cpuset| !(cpuset &
core_cpuset).is_empty())
})
.count();
@@ -153,15 +152,15 @@ impl NumaTopology {
self.logical_cores_per_node.get(node).copied().unwrap_or(0)
}
- fn filter_physical_cores(&self, node_cpuset: CpuSet) -> CpuSet {
+ fn filter_physical_cores(&self, node_cpuset: &CpuSet) -> CpuSet {
let mut physical_cpuset = CpuSet::new();
for core in self.topology.objects_with_type(ObjectType::Core) {
if let Some(core_cpuset) = core.cpuset() {
- let intersection = node_cpuset.clone() & core_cpuset;
+ let intersection = node_cpuset & core_cpuset;
if !intersection.is_empty()
&& let Some(first_cpu) = intersection.iter_set().min()
{
- physical_cpuset.set(first_cpu)
+ physical_cpuset.set(first_cpu);
}
}
}
@@ -182,14 +181,16 @@ impl NumaTopology {
available: self.node_count,
})?;
- let cpuset_ref = node.cpuset().ok_or(ShardingError::TopologyDetection {
- msg: format!("Node {} has no CPU set", node_id),
- })?;
+ let cpuset_ref = node
+ .cpuset()
+ .ok_or_else(|| ShardingError::TopologyDetection {
+ msg: format!("Node {node_id} has no CPU set"),
+ })?;
let cpuset = SpecializedBitmapRef::to_owned(&cpuset_ref);
if avoid_hyperthread {
- Ok(self.filter_physical_cores(cpuset))
+ Ok(self.filter_physical_cores(&cpuset))
} else {
Ok(cpuset)
}
@@ -199,6 +200,8 @@ impl NumaTopology {
/// One shard's home: which CPU cores it may run on, and which NUMA
/// node its memory should sit near (`None` means do not pin memory).
#[derive(Debug, Clone)]
+// The only readers are the Linux-gated bind paths below.
+#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub struct ShardInfo {
pub cpu_set: HashSet<usize>,
pub numa_node: Option<usize>,
@@ -249,7 +252,7 @@ impl ShardInfo {
let node = topology
.objects_with_type(ObjectType::NUMANode)
.nth(node_id)
- .ok_or(ShardingError::InvalidNode {
+ .ok_or_else(|| ShardingError::InvalidNode {
requested: node_id,
available:
topology.objects_with_type(ObjectType::NUMANode).count(),
})?;
@@ -324,10 +327,7 @@ pub struct ShardAllocator {
impl ShardAllocator {
/// Build an allocator for the given choice. Only `NumaAware` reads
/// the machine topology up front; the simpler modes do not.
- pub fn new(
- allocation: &CpuAllocation,
- pin_cores: bool,
- ) -> Result<ShardAllocator, ShardingError> {
+ pub fn new(allocation: &CpuAllocation, pin_cores: bool) -> Result<Self,
ShardingError> {
let topology = if matches!(allocation, CpuAllocation::NumaAware(_)) {
let numa_topology = NumaTopology::detect()?;
@@ -353,7 +353,7 @@ impl ShardAllocator {
// quota-restricted process gets proportionally fewer shards.
let available_cpus = available_parallelism()
.map_err(|err| ShardingError::Other {
- msg: format!("Failed to get available_parallelism:
{:?}", err),
+ msg: format!("Failed to get available_parallelism:
{err:?}"),
})?
.get();
@@ -431,7 +431,7 @@ impl ShardAllocator {
}
CpuAllocation::NumaAware(numa_config) => {
let topology =
self.topology.as_ref().ok_or(ShardingError::NoTopology)?;
- let assignments = self.compute_numa_assignments(topology,
numa_config)?;
+ let assignments = Self::compute_numa_assignments(topology,
numa_config)?;
if !self.pin_cores {
tracing::warn!(
@@ -446,7 +446,6 @@ impl ShardAllocator {
}
fn compute_numa_assignments(
- &self,
topology: &NumaTopology,
numa: &NumaConfig,
) -> Result<Vec<ShardInfo>, ShardingError> {
diff --git a/core/server/src/shell.rs b/core/server/src/shell.rs
new file mode 100644
index 000000000..bc05b40a5
--- /dev/null
+++ b/core/server/src/shell.rs
@@ -0,0 +1,202 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The shell vocabulary.
+//!
+//! The shard/metadata type aliases the dispatch layer is generic over, the
+//! [`ShellBus`] bound, the [`ShellHandlers`] slot struct, and the
+//! `[cluster]` timer-to-tick translation every consensus group boots with.
+//! Everything here is type- and config-level; construction (wiring the
+//! handlers against a live bus) stays in [`crate::bootstrap`].
+
+use crate::session_manager::SessionManager;
+use configs::server::ServerConfig;
+use consensus::{ConsensusTimers, VsrConsensus};
+use iggy_common::variadic;
+use journal::prepare_journal::PrepareJournal;
+use journal::superblock::PingPongSuperblock;
+use message_bus::client_listener::RequestHandler;
+use message_bus::replica::listener::MessageHandler;
+use message_bus::{ConnectionInstaller, IggyMessageBus, MessageBus};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::IggySnapshot;
+use metadata::stm::mux::WithFactory;
+use metadata::stm::stream::Streams;
+use metadata::stm::user::Users;
+use shard::shards_table::PapayaShardsTable;
+use shard::{IggyShard, ListClientsHandler, MetadataSubmitHandler,
PartitionReadHandler};
+use std::cell::RefCell;
+use std::rc::{Rc, Weak};
+use std::time::Duration;
+
+pub(crate) type ServerMuxStateMachine = MuxStateMachine<variadic!(Users,
Streams)>;
+
+/// Cross-thread bundle carrying one `ReadHandleFactory` per metadata
+/// state. Shard 0 mints one after `recover()` and broadcasts a clone to
+/// every peer shard; each peer rebuilds a reader-mode
+/// [`ServerMuxStateMachine`] on its own runtime, skipping the WAL.
+pub(crate) type ServerMetadataBundle = <variadic!(Users, Streams) as
WithFactory>::Bundle;
+
+pub(crate) type ServerMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerMuxStateMachine,
+>;
+
+/// The shard type the dispatch layer is generic over.
+///
+/// `B`/`MJ`/`S`/`SB` are free; the metadata state machine (`M`) and shards
+/// table (`T`) are pinned, being identical in production and the simulator.
+/// Production instantiates it as [`ServerShard`], defaulting `SB` to the
+/// on-disk [`PingPongSuperblock`]; the simulator supplies its own
+/// `B`/`MJ`/`S`/`SB`.
+pub type ShellShard<B, MJ, S, SB = PingPongSuperblock> =
+ IggyShard<B, MJ, S, ServerMuxStateMachine, PapayaShardsTable, SB>;
+
+/// Late-bound self-reference the deferred dispatch handlers upgrade per frame.
+pub type ShellShardHandle<B, MJ, S, SB = PingPongSuperblock> =
+ Rc<RefCell<Option<Weak<ShellShard<B, MJ, S, SB>>>>>;
+
+/// Bus bounds the dispatch/pump path needs (matches `run_message_pump`).
+/// Blanket-impl'd, so it is only shorthand for the four underlying bounds.
+pub trait ShellBus: MessageBus + ConnectionInstaller + Clone + 'static {}
+impl<B: MessageBus + ConnectionInstaller + Clone + 'static> ShellBus for B {}
+
+/// The five dispatch handlers a shard is built with, plus the
+/// [`SessionManager`] the request-plane pair shares.
+///
+/// Both production (`build_shard_for_thread`) and the simulator's shell
+/// mode construct these through [`crate::bootstrap::wire_shell_handlers`],
+/// so the request plane is wired one way. The simulator's shell-off fast
+/// path uses [`ShellHandlers::noop`] instead.
+pub struct ShellHandlers {
+ pub on_replica_message: MessageHandler,
+ pub on_client_request: RequestHandler,
+ pub on_metadata_submit: MetadataSubmitHandler,
+ pub on_list_clients: ListClientsHandler,
+ pub on_partition_read: PartitionReadHandler,
+ /// Bound by the client-request handler, read by the get-clients
+ /// handler; the caller keeps it to reach locally-homed sessions.
+ pub sessions: Rc<RefCell<SessionManager>>,
+}
+
+impl ShellHandlers {
+ /// Inert handlers for the shell-off fast path: every callback is a
+ /// no-op over an empty [`SessionManager`]. Behaviorally identical to
+ /// hand-written no-op closures, so a caller can keep one destructure
+ /// site across both toggle states.
+ #[must_use]
+ pub fn noop() -> Self {
+ Self {
+ on_replica_message: Rc::new(|_, _| {}),
+ on_client_request: Rc::new(|_, _| {}),
+ on_metadata_submit: Rc::new(|_| {}),
+ on_list_clients: Rc::new(|_| {}),
+ on_partition_read: Rc::new(|_, _, _| {}),
+ sessions: Rc::new(RefCell::new(SessionManager::new())),
+ }
+ }
+}
+
+pub type ServerShard = ShellShard<Rc<IggyMessageBus>, PrepareJournal,
IggySnapshot>;
+
+/// Convert a consensus-timer interval to whole ticks, floored at one tick so a
+/// sub-tick value still fires and saturated on overflow.
+fn duration_to_ticks(interval: Duration) -> u64 {
+ let ticks = interval.as_millis() /
shard::CONSENSUS_TICK_INTERVAL.as_millis();
+ u64::try_from(ticks.max(1)).unwrap_or(u64::MAX)
+}
+
+/// `[cluster] heartbeat_timeout` in consensus ticks. Every consensus group
+/// (metadata and per-partition planes alike) gets the same window: the failure
+/// it guards against - a primary that stopped heartbeating - is host-level,
not
+/// per-plane.
+pub(crate) fn cluster_heartbeat_ticks(config: &ServerConfig) -> u64 {
+ duration_to_ticks(config.cluster.heartbeat_timeout.get_duration())
+}
+
+/// `[cluster] commit_broadcast_interval` in consensus ticks: how often the
+/// primary broadcasts its commit point, the cluster's liveness feed. Applied
+/// to every consensus group, matching `cluster_heartbeat_ticks`.
+pub(crate) fn commit_broadcast_ticks(config: &ServerConfig) -> u64 {
+ duration_to_ticks(config.cluster.commit_broadcast_interval.get_duration())
+}
+
+/// `[cluster] prepare_retransmit_interval` in consensus ticks: how often the
+/// primary retransmits un-acked prepares. Applied to every consensus group,
+/// matching `cluster_heartbeat_ticks`.
+pub(crate) fn prepare_retransmit_ticks(config: &ServerConfig) -> u64 {
+
duration_to_ticks(config.cluster.prepare_retransmit_interval.get_duration())
+}
+
+/// `[cluster] view_change_retransmit_interval` in consensus ticks: how often a
+/// replica retransmits its `StartViewChange` / `DoViewChange` during a view
+/// change. Applied to every consensus group, matching
`cluster_heartbeat_ticks`.
+pub(crate) fn view_change_retransmit_ticks(config: &ServerConfig) -> u64 {
+ duration_to_ticks(
+ config
+ .cluster
+ .view_change_retransmit_interval
+ .get_duration(),
+ )
+}
+
+/// `[cluster] view_change_status_timeout` in consensus ticks: the stalled
+/// view-change backstop before escalating to a fresh election. Applied to
every
+/// consensus group, matching `cluster_heartbeat_ticks`.
+pub(crate) fn view_change_status_ticks(config: &ServerConfig) -> u64 {
+ duration_to_ticks(config.cluster.view_change_status_timeout.get_duration())
+}
+
+/// `[cluster] request_start_view_retransmit_interval` in consensus ticks: how
+/// often a recovering or view-change backup re-requests the current
`StartView`.
+/// Applied to every consensus group, matching `cluster_heartbeat_ticks`.
+pub(crate) fn request_start_view_ticks(config: &ServerConfig) -> u64 {
+ duration_to_ticks(
+ config
+ .cluster
+ .request_start_view_retransmit_interval
+ .get_duration(),
+ )
+}
+
+/// The full `[cluster]` timer set every consensus group boots with, built
+/// once so the planes cannot diverge in what they apply.
+pub(crate) fn consensus_timers(config: &ServerConfig) -> ConsensusTimers {
+ ConsensusTimers {
+ normal_heartbeat_ticks: cluster_heartbeat_ticks(config),
+ commit_message_ticks: commit_broadcast_ticks(config),
+ prepare_ticks: prepare_retransmit_ticks(config),
+ view_change_retransmit_ticks: view_change_retransmit_ticks(config),
+ view_change_status_ticks: view_change_status_ticks(config),
+ request_start_view_ticks: request_start_view_ticks(config),
+ probe_attempts_max: config.cluster.view_probe_attempts_max,
+ }
+}
+
+/// `[cluster] repair_retry_interval` in consensus ticks: how long a stalled
+/// journal-repair stream waits before re-requesting its window. Both planes'
+/// repair loops share it, so it is applied once per shard (not per consensus
+/// group). Clamped to `u32`, the width of the session idle-tick counter.
+pub(crate) fn repair_retry_ticks(config: &ServerConfig) -> u32 {
+ u32::try_from(duration_to_ticks(
+ config.cluster.repair_retry_interval.get_duration(),
+ ))
+ .unwrap_or(u32::MAX)
+}
diff --git a/core/server/src/users.rs b/core/server/src/users.rs
index 129bbfca8..3b146cb88 100644
--- a/core/server/src/users.rs
+++ b/core/server/src/users.rs
@@ -35,7 +35,7 @@
//! the caller's request sequence stays contiguous; see
//! `verify_and_rewrite_change_password`.
-use crate::bootstrap::{ShellBus, ShellShard};
+use crate::shell::{ShellBus, ShellShard};
use crate::wire::{request_body, rewrite_request_body};
use bytes::Bytes;
use consensus::MetadataHandle;
@@ -58,7 +58,7 @@ use std::rc::Rc;
/// rejection (see [`verify_and_rewrite_change_password`]). Every other
operation
/// passes through unchanged. Returns [`IggyError::InvalidCommand`] only on an
/// undecodable password body.
-pub(crate) fn maybe_rewrite_user_password_request<B, MJ, S, SB>(
+pub fn maybe_rewrite_user_password_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
request: Message<RoutedRequestHeader>,
) -> Result<Message<RoutedRequestHeader>, IggyError>
diff --git a/core/server/src/wire.rs b/core/server/src/wire.rs
index d869a2155..737d72440 100644
--- a/core/server/src/wire.rs
+++ b/core/server/src/wire.rs
@@ -26,7 +26,7 @@ use iggy_common::IggyError;
use message_bus::installer::conn_info::ClientTransportKind;
use server_common::Message;
-pub(crate) fn request_body(request: &Message<RoutedRequestHeader>) -> &[u8] {
+pub fn request_body(request: &Message<RoutedRequestHeader>) -> &[u8] {
&request.as_slice()[std::mem::size_of::<RoutedRequestHeader>()..request.header().size
as usize]
}
@@ -38,9 +38,7 @@ pub(crate) fn request_body(request:
&Message<RoutedRequestHeader>) -> &[u8] {
///
/// # Errors
/// [`IggyError::InvalidFormat`] when the stamp disagrees with the body.
-pub(crate) fn verify_request_checksum(
- request: &Message<RoutedRequestHeader>,
-) -> Result<(), IggyError> {
+pub fn verify_request_checksum(request: &Message<RoutedRequestHeader>) ->
Result<(), IggyError> {
let stamped = request.header().request_checksum;
if stamped == 0 ||
u128::from(iggy_common::calculate_checksum(request_body(request))) == stamped
{
@@ -53,7 +51,7 @@ pub(crate) fn verify_request_checksum(
/// (`1=TCP, 2=QUIC, 4=WebSocket`); TLS variants report their base
/// transport. `ClientTransportKind` is `#[non_exhaustive]`, so any other
/// (TCP, TCP-TLS, or a future) variant falls back to TCP.
-pub(crate) const fn transport_kind_to_wire(kind: ClientTransportKind) -> u8 {
+pub const fn transport_kind_to_wire(kind: ClientTransportKind) -> u8 {
match kind {
ClientTransportKind::Quic => 2,
ClientTransportKind::Ws | ClientTransportKind::Wss => 4,
@@ -61,7 +59,7 @@ pub(crate) const fn transport_kind_to_wire(kind:
ClientTransportKind) -> u8 {
}
}
-pub(crate) fn usize_to_u32(value: usize) -> Result<u32, IggyError> {
+pub fn usize_to_u32(value: usize) -> Result<u32, IggyError> {
u32::try_from(value).map_err(|_| IggyError::InvalidIdentifier)
}
@@ -69,7 +67,7 @@ pub(crate) fn usize_to_u32(value: usize) -> Result<u32,
IggyError> {
/// preserving the header (and fixing `size`). Used by the primary-side
/// request rewrites that swap a secret-bearing wire body for the
/// hash-carrying replicated body before consensus.
-pub(crate) fn rewrite_request_body(
+pub fn rewrite_request_body(
request: &Message<RoutedRequestHeader>,
body: &Bytes,
) -> Result<Message<RoutedRequestHeader>, IggyError> {
diff --git a/core/server/tests/module_graph.rs
b/core/server/tests/module_graph.rs
new file mode 100644
index 000000000..d9d978a56
--- /dev/null
+++ b/core/server/tests/module_graph.rs
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Module-graph guard: the crate's module dependency graph must stay a DAG.
+//!
+//! The http/ role-leaf pattern rotted into two state-hub cycles while nothing
+//! enforced it, and two of the historical bootstrap cycles were invisible to
+//! `use`-greps (an inline-qualified `crate::...` call and `use super::`
+//! imports). So this test parses every source file with `syn` and scans full
+//! token streams - it sees qualified call sites and macro arguments, not just
+//! use declarations.
+//!
+//! Granularity is the module FILE. Parent<->child edges are exempt: a root
+//! composing its children (and children reaching items the root defines) is
+//! the pattern working as intended. Sibling and cross-tree cycles are the
+//! rot this guard exists to stop.
+//!
+//! `WHITELIST` carries the known survivors. Each entry must still be a live
+//! cycle - a stale entry fails the test, so the list can only shrink.
+
+use proc_macro2::{TokenStream, TokenTree};
+use quote::ToTokens;
+use std::collections::{BTreeMap, BTreeSet};
+use std::path::{Path, PathBuf};
+
+/// Known mutual edges, as unordered pairs of module paths. Burned down per
+/// refactor PR; the auth<->dispatch cycle dies with the dispatch/ login merge.
+const WHITELIST: [(&str, &str); 1] = [("auth", "dispatch")];
+
+type Module = Vec<String>;
+
+#[test]
+fn module_graph_is_a_dag_modulo_whitelist() {
+ let src = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
+ let modules = collect_modules(&src);
+ let module_set: BTreeSet<Module> = modules.iter().map(|(module, _)|
module.clone()).collect();
+
+ let mut edges: BTreeMap<Module, BTreeSet<Module>> = BTreeMap::new();
+ for (module, file) in &modules {
+ let source = std::fs::read_to_string(file)
+ .unwrap_or_else(|error| panic!("cannot read {}: {error}",
file.display()));
+ let ast = syn::parse_file(&source)
+ .unwrap_or_else(|error| panic!("cannot parse {}: {error}",
file.display()));
+ let mut paths = Vec::new();
+ scan_items(&ast.items, &mut paths);
+ let targets = edges.entry(module.clone()).or_default();
+ for raw in paths {
+ if let Some(target) = resolve(module, &raw, &module_set)
+ && target != *module
+ && !is_ancestor(&target, module)
+ && !is_ancestor(module, &target)
+ {
+ targets.insert(target);
+ }
+ }
+ }
+
+ let whitelist: BTreeSet<(Module, Module)> = WHITELIST
+ .iter()
+ .flat_map(|(a, b)| {
+ let a = parse_module(a);
+ let b = parse_module(b);
+ [(a.clone(), b.clone()), (b, a)]
+ })
+ .collect();
+
+ for (from, to) in &whitelist {
+ assert!(
+ edges.get(from).is_some_and(|targets| targets.contains(to)),
+ "stale whitelist entry {} -> {}: the edge is gone, remove it",
+ render(from),
+ render(to),
+ );
+ }
+
+ for (from, targets) in &mut edges {
+ targets.retain(|to| !whitelist.contains(&(from.clone(), to.clone())));
+ }
+
+ if let Some(cycle) = find_cycle(&edges) {
+ let chain = cycle.iter().map(render).collect::<Vec<_>>();
+ panic!(
+ "module cycle: {}\nBreak it by moving the shared vocabulary into \
+ a leaf both sides import, or point the edge one way.",
+ chain.join(" -> "),
+ );
+ }
+}
+
+/// Map every source file under `src/` to its module path. `lib.rs` is the
+/// crate root `[]`; `main.rs`/`args.rs` belong to the bin target and are
+/// skipped (their `crate::` is a different crate).
+fn collect_modules(src: &Path) -> Vec<(Module, PathBuf)> {
+ let mut files = Vec::new();
+ walk(src, &mut files);
+ files.sort();
+ files
+ .into_iter()
+ .filter_map(|file| {
+ let relative = file
+ .strip_prefix(src)
+ .unwrap_or_else(|_| panic!("{} outside src", file.display()));
+ let mut module: Vec<String> = relative
+ .components()
+ .map(|c| c.as_os_str().to_string_lossy().into_owned())
+ .collect();
+ let last = module.pop().unwrap_or_default();
+ match last.as_str() {
+ "main.rs" | "args.rs" => return None,
+ "lib.rs" | "mod.rs" => {}
+ _ => module.push(last.trim_end_matches(".rs").to_owned()),
+ }
+ Some((module, file))
+ })
+ .collect()
+}
+
+fn walk(dir: &Path, files: &mut Vec<PathBuf>) {
+ let entries = std::fs::read_dir(dir)
+ .unwrap_or_else(|error| panic!("cannot read dir {}: {error}",
dir.display()));
+ for entry in entries {
+ let path = entry
+ .unwrap_or_else(|error| panic!("cannot read entry in {}: {error}",
dir.display()))
+ .path();
+ if path.is_dir() {
+ walk(&path, files);
+ } else if path.extension().is_some_and(|ext| ext == "rs") {
+ files.push(path);
+ }
+ }
+}
+
+/// Collect every `crate::`/`super::`-rooted path in the token streams of
+/// `items`, recursing through nested modules and skipping `#[cfg(test)]` ones.
+fn scan_items(items: &[syn::Item], paths: &mut Vec<Vec<String>>) {
+ for item in items {
+ if let syn::Item::Mod(module) = item {
+ if is_cfg_test(&module.attrs) {
+ continue;
+ }
+ if let Some((_, nested)) = &module.content {
+ scan_items(nested, paths);
+ }
+ continue;
+ }
+ scan_tokens(item.to_token_stream(), paths);
+ }
+}
+
+fn is_cfg_test(attrs: &[syn::Attribute]) -> bool {
+ attrs.iter().any(|attr| {
+ attr.path().is_ident("cfg")
+ && attr
+ .parse_args::<TokenStream>()
+ .is_ok_and(tokens_mention_test)
+ })
+}
+
+/// True when the cfg predicate names the bare `test` ident anywhere,
+/// `cfg(all(test, ...))` included.
+fn tokens_mention_test(tokens: TokenStream) -> bool {
+ tokens.into_iter().any(|tree| match tree {
+ TokenTree::Ident(ident) => ident == "test",
+ TokenTree::Group(group) => tokens_mention_test(group.stream()),
+ _ => false,
+ })
+}
+
+/// Token-level scan: catches inline-qualified calls and macro arguments that
+/// an AST `use`-only walk would miss. Doc comments and string literals are
+/// literals, not idents, so they never produce edges.
+fn scan_tokens(tokens: TokenStream, paths: &mut Vec<Vec<String>>) {
+ let trees: Vec<TokenTree> = tokens.into_iter().collect();
+ let mut index = 0;
+ while index < trees.len() {
+ match &trees[index] {
+ TokenTree::Group(group) => {
+ scan_tokens(group.stream(), paths);
+ index += 1;
+ }
+ TokenTree::Ident(ident) => {
+ let name = ident.to_string();
+ if name == "crate" || name == "super" {
+ let (segments, consumed) = read_path(&trees, index);
+ if segments.len() > 1 {
+ paths.push(segments);
+ }
+ index += consumed;
+ } else {
+ index += 1;
+ }
+ }
+ _ => index += 1,
+ }
+ }
+}
+
+/// Read `ident (:: ident)*` starting at `start`, descending into no groups.
+fn read_path(trees: &[TokenTree], start: usize) -> (Vec<String>, usize) {
+ let mut segments = Vec::new();
+ let mut index = start;
+ while let Some(TokenTree::Ident(ident)) = trees.get(index) {
+ segments.push(ident.to_string());
+ index += 1;
+ let double_colon = matches!(trees.get(index),
Some(TokenTree::Punct(p)) if p.as_char() == ':')
+ && matches!(trees.get(index + 1), Some(TokenTree::Punct(p)) if
p.as_char() == ':');
+ if double_colon {
+ index += 2;
+ } else {
+ break;
+ }
+ }
+ (segments, index - start)
+}
+
+/// Resolve a raw `crate::`/`super::` path to the deepest known module it
+/// names. Returns `None` for paths into the crate root's own items.
+fn resolve(current: &Module, raw: &[String], modules: &BTreeSet<Module>) ->
Option<Module> {
+ let (base, rest): (Module, &[String]) = match
raw.first().map(String::as_str) {
+ Some("crate") => (Vec::new(), &raw[1..]),
+ Some("super") => {
+ let supers = raw.iter().take_while(|s| *s == "super").count();
+ if supers > current.len() {
+ return None;
+ }
+ (current[..current.len() - supers].to_vec(), &raw[supers..])
+ }
+ _ => return None,
+ };
+ let mut best: Option<Module> = if base.is_empty() {
+ None
+ } else {
+ Some(base.clone())
+ };
+ let mut candidate = base;
+ for segment in rest {
+ candidate.push(segment.clone());
+ if modules.contains(&candidate) {
+ best = Some(candidate.clone());
+ } else {
+ break;
+ }
+ }
+ best
+}
+
+fn is_ancestor(shorter: &Module, longer: &Module) -> bool {
+ shorter.len() < longer.len() && longer[..shorter.len()] == shorter[..]
+}
+
+fn render(module: &Module) -> String {
+ if module.is_empty() {
+ "crate".to_owned()
+ } else {
+ module.join("::")
+ }
+}
+
+fn parse_module(path: &str) -> Module {
+ path.split("::").map(str::to_owned).collect()
+}
+
+/// DFS three-color cycle search; returns one cycle as a module chain.
+fn find_cycle(edges: &BTreeMap<Module, BTreeSet<Module>>) ->
Option<Vec<Module>> {
+ let mut visiting = BTreeSet::new();
+ let mut done = BTreeSet::new();
+ let mut stack = Vec::new();
+ for start in edges.keys() {
+ if let Some(cycle) = dfs(start, edges, &mut visiting, &mut done, &mut
stack) {
+ return Some(cycle);
+ }
+ }
+ None
+}
+
+fn dfs(
+ node: &Module,
+ edges: &BTreeMap<Module, BTreeSet<Module>>,
+ visiting: &mut BTreeSet<Module>,
+ done: &mut BTreeSet<Module>,
+ stack: &mut Vec<Module>,
+) -> Option<Vec<Module>> {
+ if done.contains(node) {
+ return None;
+ }
+ if visiting.contains(node) {
+ let from = stack.iter().position(|n| n == node).unwrap_or(0);
+ let mut cycle = stack[from..].to_vec();
+ cycle.push(node.clone());
+ return Some(cycle);
+ }
+ visiting.insert(node.clone());
+ stack.push(node.clone());
+ if let Some(targets) = edges.get(node) {
+ for target in targets {
+ if let Some(cycle) = dfs(target, edges, visiting, done, stack) {
+ return Some(cycle);
+ }
+ }
+ }
+ stack.pop();
+ visiting.remove(node);
+ done.insert(node.clone());
+ None
+}
diff --git a/core/shard_allocator/Cargo.toml b/core/shard_allocator/Cargo.toml
deleted file mode 100644
index 1d51d3a8d..000000000
--- a/core/shard_allocator/Cargo.toml
+++ /dev/null
@@ -1,41 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-[package]
-name = "shard_allocator"
-version = "0.1.0"
-description = "CPU and NUMA shard allocation for the iggy server, backed by
hwloc."
-edition = "2024"
-license = "Apache-2.0"
-publish = false
-
-[dependencies]
-cpu_allocation = { workspace = true }
-thiserror = { workspace = true }
-tracing = { workspace = true }
-
-[target.'cfg(not(target_env = "musl"))'.dependencies]
-hwlocality = { workspace = true }
-
-[target.'cfg(target_env = "musl")'.dependencies]
-hwlocality = { workspace = true, features = ["vendored"] }
-
-[target.'cfg(target_os = "linux")'.dependencies]
-nix = { workspace = true }
-
-[lints]
-workspace = true
diff --git a/core/shard_allocator/build.rs b/core/shard_allocator/build.rs
deleted file mode 100644
index d66e93f07..000000000
--- a/core/shard_allocator/build.rs
+++ /dev/null
@@ -1,43 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-// Vendored `hwloc` references `cbrt`, which makes the linker pull in a
-// `libm`. musl folds the math functions into its `libc`, so the Rust
-// musl sysroot ships no `libm.a`. Without one, `-lm` falls through to
-// the host glibc's `libm.a`, whose `cbrt` needs glibc-internal
-// `__frexp`/`__ldexp` symbols that do not exist on musl, and the static
-// link fails. Drop an empty `libm.a` stub on the search path so `-lm`
-// resolves to nothing and `cbrt` is satisfied later by musl's own
-// `libc`. No effect on non-musl targets.
-
-use std::env;
-use std::fs;
-use std::path::Path;
-
-fn main() {
- if env::var("CARGO_CFG_TARGET_ENV").as_deref() != Ok("musl") {
- return;
- }
-
- let out_dir = env::var("OUT_DIR").expect("OUT_DIR is set by cargo for
build scripts");
- let stub = Path::new(&out_dir).join("libm.a");
-
- // `!<arch>\n` is the canonical header of an empty `ar` archive.
- fs::write(&stub, b"!<arch>\n").expect("write empty libm.a stub");
-
- println!("cargo:rustc-link-search=native={out_dir}");
-}
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index 638a61dce..8e57773ea 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -32,7 +32,8 @@ use metadata::stm::stream::{Streams, StreamsInner};
use metadata::stm::user::{Users, UsersInner};
use metadata::{IggyMetadata, apply_committed_prepare};
use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig};
-use server::bootstrap::{ShellHandlers, ShellShardHandle, wire_shell_handlers};
+use server::bootstrap::wire_shell_handlers;
+use server::shell::{ShellHandlers, ShellShardHandle};
use server_common::crypto;
use server_common::sharding::{METADATA_GROUP, ShardId};
use shard::shards_table::PapayaShardsTable;
