You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "hubcio (via GitHub)" <gi...@apache.org> on 2026/05/05 08:32:24 UTC
Re: [PR] feat(cluster): implement replica bootstrap [iggy]
hubcio commented on code in PR #3163:
URL: https://github.com/apache/iggy/pull/3163#discussion_r3186975496
##########
core/server-ng/src/config_writer.rs:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.
+ */
+
+use crate::server_error::ServerNgError;
+use compio::fs::OpenOptions;
+use compio::io::AsyncWriteAtExt;
+use configs::server_ng::ServerNgConfig;
+use std::net::SocketAddr;
+
+/// Write the runtime `current_config.toml` file with the effective bound ports.
+///
+/// # Errors
+///
+/// Returns an error if the config cannot be serialized or if the runtime
+/// config file cannot be written and synced.
+pub async fn write_current_config(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ bound_tcp: Option<SocketAddr>,
+ bound_replica: Option<SocketAddr>,
+ bound_tcp_tls: Option<SocketAddr>,
+ bound_quic: Option<SocketAddr>,
+ bound_websocket: Option<SocketAddr>,
+) -> Result<(), ServerNgError> {
+ let mut current_config = config.clone();
+
+ if let Some(bound_client_tcp) = bound_tcp_tls.or(bound_tcp) {
+ // Keep parity with the current server binary: integration harnesses
+ // read `tcp.address` from `runtime/current_config.toml` to discover
+ // the actual port chosen by the OS when binding to port 0.
+ current_config.tcp.address = bound_client_tcp.to_string();
+ }
+ if let Some(bound_quic) = bound_quic {
+ current_config.quic.address = bound_quic.to_string();
+ }
+ if let Some(bound_websocket) = bound_websocket {
+ current_config.websocket.address = bound_websocket.to_string();
+ }
+
+ if current_config.cluster.enabled
+ && let Some(replica_id) = current_replica_id
+ {
+ let node = current_config
+ .cluster
+ .nodes
+ .iter_mut()
+ .find(|node| node.replica_id == replica_id)
+ .ok_or(ServerNgError::ClusterNodeNotFound { replica_id })?;
+ if let Some(bound_client_tcp) = bound_tcp_tls.or(bound_tcp) {
+ node.ports.tcp = Some(bound_client_tcp.port());
+ }
+ if let Some(bound_replica) = bound_replica {
+ node.ports.tcp_replica = Some(bound_replica.port());
+ }
+ if let Some(bound_quic) = bound_quic {
+ node.ports.quic = Some(bound_quic.port());
+ }
+ if let Some(bound_websocket) = bound_websocket {
+ node.ports.websocket = Some(bound_websocket.port());
+ }
+ }
+
+ let runtime_path = current_config.system.get_runtime_path();
+ let config_path = format!("{runtime_path}/current_config.toml");
+ let content =
+ toml::to_string(¤t_config).map_err(ServerNgError::CurrentConfigSerialize)?;
+
+ let mut file = OpenOptions::new()
Review Comment:
`truncate(true).open` + `write_all_at(0)` + `sync_all` is non-atomic. a crash mid-write leaves a partial TOML, and the integration harness reads the bound port from this file - so a partial write surfaces as garbage or `port=0`. mirror `IggySnapshot::persist` at `core/metadata/src/impls/metadata.rs:91-130`: write to a tmp file, fsync, rename, then fsync the parent dir.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
+ topic_stats.zero_out_all();
+ }
+
+ for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
+ let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
+ let partition = load_partition(
+ config,
+ namespace,
+ topic_stats,
+ &partition_metadata,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?;
+ let local_idx = partitions.insert(namespace, partition);
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ );
+ }
+
+ let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
+ let senders = vec![sender];
+ let shard_handle = Rc::new(RefCell::new(None));
+ let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
+ let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let built = IggyShardBuilder::new(
+ ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ Rc::clone(&bus),
+ on_replica_message,
+ on_client_request,
+ metadata,
+ partitions,
+ senders,
+ inbox,
+ shards_table,
+ PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
+ CoordinatorConfig::default(),
+ bus.token(),
+ )
+ .build();
+ if let Some(refresh_task) = built.refresh_task {
+ bus.track_background(refresh_task);
+ }
+
+ let shard = Rc::new(built.shard);
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ Ok(shard)
+}
+
+const fn validate_recovered_namespace(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ let namespace = &config.extra.namespace;
+ if stream_id < namespace.max_streams
+ && topic_id < namespace.max_topics
+ && partition_id < namespace.max_partitions
+ {
+ return Ok(());
+ }
+
+ Err(ServerNgError::RecoveredNamespaceOutOfBounds {
+ stream_id,
+ topic_id,
+ partition_id,
+ max_streams: namespace.max_streams,
+ max_topics: namespace.max_topics,
+ max_partitions: namespace.max_partitions,
+ })
+}
+
+async fn load_partition(
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ topic_stats: Arc<TopicStats>,
+ partition_metadata: &Partition,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerNgError> {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let stats = Arc::new(PartitionStats::new(topic_stats));
+ let consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ namespace.inner(),
+ bus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+
+ // TODO: decouple the loading logic from the `server` crate and load directly
+ // into the new `partitions` log/runtime types.
+ let loaded_log = server::bootstrap::load_segments(
+ &config.system,
+ stream_id,
+ topic_id,
+ partition_id,
+ config
+ .system
+ .get_partition_path(stream_id, topic_id, partition_id),
+ stats.clone(),
+ )
+ .await
+ .map_err(|source| ServerNgError::PartitionLogLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+
+ let mut partition = IggyPartition::new(stats.clone(), consensus);
+ hydrate_partition_log(
+ &mut partition,
+ config,
+ stream_id,
+ topic_id,
+ partition_id,
+ loaded_log,
+ )
+ .await?;
+
+ let current_offset = partition
Review Comment:
`current_offset` here is derived from `max(end_offset where size > 0)`. `end_offset` comes from the segment metadata index. if the previous run wrote messages but index entries weren't fsynced before a crash, `end_offset` lags actual `messages.log` bytes. next append assigns logical offsets that overlap pre-crash messages already on disk - duplicate-offset corruption visible to all SDK clients. derive `current_offset` from messages-side size + tail-replay, or fail-fast on index/messages length divergence.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
Review Comment:
`restored_op = snapshot.sequence_number()` when `last_applied_op = None` is correct only because `IggySnapshot::persist` (`core/metadata/src/impls/metadata.rs:91-130`) is atomic (tmp + fsync + rename + dir fsync). that's an implicit contract. add a doc-comment naming the invariant so future snapshot persistence changes don't quietly break this fallback path.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
+ topic_stats.zero_out_all();
+ }
+
+ for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
+ let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
+ let partition = load_partition(
+ config,
+ namespace,
+ topic_stats,
+ &partition_metadata,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?;
+ let local_idx = partitions.insert(namespace, partition);
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ );
+ }
+
+ let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
+ let senders = vec![sender];
+ let shard_handle = Rc::new(RefCell::new(None));
+ let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
+ let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let built = IggyShardBuilder::new(
+ ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ Rc::clone(&bus),
+ on_replica_message,
+ on_client_request,
+ metadata,
+ partitions,
+ senders,
+ inbox,
+ shards_table,
+ PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
+ CoordinatorConfig::default(),
+ bus.token(),
+ )
+ .build();
+ if let Some(refresh_task) = built.refresh_task {
+ bus.track_background(refresh_task);
+ }
+
+ let shard = Rc::new(built.shard);
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ Ok(shard)
+}
+
+const fn validate_recovered_namespace(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ let namespace = &config.extra.namespace;
+ if stream_id < namespace.max_streams
+ && topic_id < namespace.max_topics
+ && partition_id < namespace.max_partitions
+ {
+ return Ok(());
+ }
+
+ Err(ServerNgError::RecoveredNamespaceOutOfBounds {
+ stream_id,
+ topic_id,
+ partition_id,
+ max_streams: namespace.max_streams,
+ max_topics: namespace.max_topics,
+ max_partitions: namespace.max_partitions,
+ })
+}
+
+async fn load_partition(
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ topic_stats: Arc<TopicStats>,
+ partition_metadata: &Partition,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerNgError> {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let stats = Arc::new(PartitionStats::new(topic_stats));
+ let consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ namespace.inner(),
+ bus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+
+ // TODO: decouple the loading logic from the `server` crate and load directly
+ // into the new `partitions` log/runtime types.
+ let loaded_log = server::bootstrap::load_segments(
+ &config.system,
+ stream_id,
+ topic_id,
+ partition_id,
+ config
+ .system
+ .get_partition_path(stream_id, topic_id, partition_id),
+ stats.clone(),
+ )
+ .await
+ .map_err(|source| ServerNgError::PartitionLogLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+
+ let mut partition = IggyPartition::new(stats.clone(), consensus);
+ hydrate_partition_log(
+ &mut partition,
+ config,
+ stream_id,
+ topic_id,
+ partition_id,
+ loaded_log,
+ )
+ .await?;
+
+ let current_offset = partition
+ .log
+ .segments()
+ .iter()
+ .filter(|segment| segment.size > IggyByteSize::default())
+ .map(|segment| segment.end_offset)
+ .max()
+ .unwrap_or(0);
+ partition.created_at = partition_metadata.created_at;
+ partition.offset.store(current_offset, Ordering::Release);
+ partition
+ .dirty_offset
+ .store(current_offset, Ordering::Relaxed);
+ partition.should_increment_offset = partition
+ .log
+ .segments()
+ .iter()
+ .any(|segment| segment.size > IggyByteSize::default());
+ partition.stats.set_current_offset(current_offset);
+
+ configure_consumer_offsets(&mut partition, config, namespace, current_offset);
+ ensure_initial_segment(&mut partition, config, stream_id, topic_id, partition_id).await?;
+
+ Ok(partition)
+}
+
+async fn hydrate_partition_log(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ loaded_log: server::streaming::partitions::log::SegmentedLog<
+ server::streaming::partitions::journal::MemoryMessageJournal,
+ >,
+) -> Result<(), ServerNgError> {
+ // TODO: decouple the loading logic from the `server` crate. This currently
+ // adapts the old server segmented log into the new `partitions` log.
+ for (segment_index, (segment, storage)) in loaded_log
+ .segments()
+ .iter()
+ .zip(loaded_log.storages().iter().cloned())
+ .enumerate()
+ {
+ let max_timestamp = match loaded_log
+ .indexes()
+ .get(segment_index)
+ .and_then(|indexes| indexes.as_ref())
+ {
+ Some(indexes) => indexes_max_timestamp(indexes),
+ None => load_segment_max_timestamp(&storage, stream_id, topic_id, partition_id).await?,
+ };
+ partition.log.add_persisted_segment(
+ convert_segment(segment, max_timestamp),
+ storage,
+ None,
+ None,
+ );
+ }
+
+ if let Some(active_index) = partition.log.segments().len().checked_sub(1) {
+ let storage = &partition.log.storages()[active_index];
+ if let (Some(messages_reader), Some(index_reader)) = (
+ storage.messages_reader.as_ref(),
+ storage.index_reader.as_ref(),
+ ) {
+ let index_path = index_reader.path();
+ let index_size = std::fs::metadata(&index_path).map_or(0, |metadata| metadata.len());
+ partition.log.messages_writers_mut()[active_index] = Some(Rc::new(
+ MessagesWriter::new(
+ &messages_reader.path(),
+ Rc::new(AtomicU64::new(u64::from(messages_reader.file_size()))),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::MessagesWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ partition.log.index_writers_mut()[active_index] = Some(Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(AtomicU64::new(index_size)),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::IndexWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ }
+ }
+
+ Ok(())
+}
+
+fn convert_segment(segment: &iggy_common::Segment, max_timestamp: u64) -> Segment {
+ Segment {
+ sealed: segment.sealed,
+ start_timestamp: segment.start_timestamp,
+ end_timestamp: segment.end_timestamp,
+ max_timestamp,
+ current_position: u64::from(segment.current_position),
+ start_offset: segment.start_offset,
+ end_offset: segment.end_offset,
+ size: segment.size,
+ max_size: segment.max_size,
+ }
+}
+
+fn indexes_max_timestamp(indexes: &server::streaming::segments::IggyIndexesMut) -> u64 {
+ let mut max_timestamp = 0;
+ for index in 0..indexes.count() {
+ if let Some(index_view) = indexes.get(index) {
+ max_timestamp = max_timestamp.max(index_view.timestamp());
+ }
+ }
+
+ max_timestamp
+}
+
+async fn load_segment_max_timestamp(
+ storage: &iggy_common::SegmentStorage,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<u64, ServerNgError> {
+ let Some(index_reader) = storage.index_reader.as_ref() else {
+ return Ok(0);
+ };
+
+ let indexes = index_reader
+ .load_all_indexes_from_disk()
+ .await
+ .map_err(|source| ServerNgError::SegmentIndexesLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+ Ok(indexes_max_timestamp(&indexes))
+}
+
+fn configure_consumer_offsets(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ current_offset: u64,
+) {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let consumer_offsets_path =
+ config
+ .system
+ .get_consumer_offsets_path(stream_id, topic_id, partition_id);
+ let consumer_group_offsets_path =
+ config
+ .system
+ .get_consumer_group_offsets_path(stream_id, topic_id, partition_id);
+
+ // TODO: decouple consumer offset loading from the `server` crate.
+ let loaded_consumer_offsets = load_consumer_offsets(&consumer_offsets_path).unwrap_or_default();
+ let consumer_offsets = ConsumerOffsets::with_capacity(loaded_consumer_offsets.len());
+ {
+ let guard = consumer_offsets.pin();
+ for offset in loaded_consumer_offsets {
Review Comment:
also lines 647-650. silent down-clamp of persisted consumer/group offsets to `current_offset`. if recovered `current_offset < persisted consumer offset`, the partition log was truncated/lost relative to consumer state, but the clamp causes the consumer to re-deliver an already-acked range without warning. that breaks at-least-once. fix: detect divergence and either surface an error or require explicit operator opt-in via config.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
+ topic_stats.zero_out_all();
+ }
+
+ for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
+ let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
+ let partition = load_partition(
+ config,
+ namespace,
+ topic_stats,
+ &partition_metadata,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?;
+ let local_idx = partitions.insert(namespace, partition);
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ );
+ }
+
+ let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
+ let senders = vec![sender];
+ let shard_handle = Rc::new(RefCell::new(None));
+ let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
+ let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let built = IggyShardBuilder::new(
+ ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ Rc::clone(&bus),
+ on_replica_message,
+ on_client_request,
+ metadata,
+ partitions,
+ senders,
+ inbox,
+ shards_table,
+ PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
+ CoordinatorConfig::default(),
+ bus.token(),
+ )
+ .build();
+ if let Some(refresh_task) = built.refresh_task {
+ bus.track_background(refresh_task);
+ }
+
+ let shard = Rc::new(built.shard);
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ Ok(shard)
+}
+
+const fn validate_recovered_namespace(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ let namespace = &config.extra.namespace;
+ if stream_id < namespace.max_streams
+ && topic_id < namespace.max_topics
+ && partition_id < namespace.max_partitions
+ {
+ return Ok(());
+ }
+
+ Err(ServerNgError::RecoveredNamespaceOutOfBounds {
+ stream_id,
+ topic_id,
+ partition_id,
+ max_streams: namespace.max_streams,
+ max_topics: namespace.max_topics,
+ max_partitions: namespace.max_partitions,
+ })
+}
+
+async fn load_partition(
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ topic_stats: Arc<TopicStats>,
+ partition_metadata: &Partition,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerNgError> {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let stats = Arc::new(PartitionStats::new(topic_stats));
+ let consensus = VsrConsensus::new(
Review Comment:
per-partition `VsrConsensus::new()` + `init()` boots fresh-zero (commit_min/commit_max/sequencer/view all 0). no `restore_commit_state`, no view restore. today this is vacuous because `MemoryMessageJournal` is volatile, but as soon as a durable partition journal lands, ack-but-not-yet-applied ops are silently lost on restart. either stub a hard-error if the partition journal contains a tail past snapshot, or implement matching restore now so the call shape exists.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
+ topic_stats.zero_out_all();
+ }
+
+ for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
+ let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
+ let partition = load_partition(
+ config,
+ namespace,
+ topic_stats,
+ &partition_metadata,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?;
+ let local_idx = partitions.insert(namespace, partition);
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ );
+ }
+
+ let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
+ let senders = vec![sender];
+ let shard_handle = Rc::new(RefCell::new(None));
+ let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
+ let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let built = IggyShardBuilder::new(
+ ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ Rc::clone(&bus),
+ on_replica_message,
+ on_client_request,
+ metadata,
+ partitions,
+ senders,
+ inbox,
+ shards_table,
+ PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
+ CoordinatorConfig::default(),
+ bus.token(),
+ )
+ .build();
+ if let Some(refresh_task) = built.refresh_task {
+ bus.track_background(refresh_task);
+ }
+
+ let shard = Rc::new(built.shard);
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ Ok(shard)
+}
+
+const fn validate_recovered_namespace(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ let namespace = &config.extra.namespace;
+ if stream_id < namespace.max_streams
+ && topic_id < namespace.max_topics
+ && partition_id < namespace.max_partitions
+ {
+ return Ok(());
+ }
+
+ Err(ServerNgError::RecoveredNamespaceOutOfBounds {
+ stream_id,
+ topic_id,
+ partition_id,
+ max_streams: namespace.max_streams,
+ max_topics: namespace.max_topics,
+ max_partitions: namespace.max_partitions,
+ })
+}
+
+async fn load_partition(
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ topic_stats: Arc<TopicStats>,
+ partition_metadata: &Partition,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerNgError> {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let stats = Arc::new(PartitionStats::new(topic_stats));
+ let consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ namespace.inner(),
+ bus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+
+ // TODO: decouple the loading logic from the `server` crate and load directly
+ // into the new `partitions` log/runtime types.
+ let loaded_log = server::bootstrap::load_segments(
+ &config.system,
+ stream_id,
+ topic_id,
+ partition_id,
+ config
+ .system
+ .get_partition_path(stream_id, topic_id, partition_id),
+ stats.clone(),
+ )
+ .await
+ .map_err(|source| ServerNgError::PartitionLogLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+
+ let mut partition = IggyPartition::new(stats.clone(), consensus);
+ hydrate_partition_log(
+ &mut partition,
+ config,
+ stream_id,
+ topic_id,
+ partition_id,
+ loaded_log,
+ )
+ .await?;
+
+ let current_offset = partition
+ .log
+ .segments()
+ .iter()
+ .filter(|segment| segment.size > IggyByteSize::default())
+ .map(|segment| segment.end_offset)
+ .max()
+ .unwrap_or(0);
+ partition.created_at = partition_metadata.created_at;
+ partition.offset.store(current_offset, Ordering::Release);
+ partition
+ .dirty_offset
+ .store(current_offset, Ordering::Relaxed);
+ partition.should_increment_offset = partition
+ .log
+ .segments()
+ .iter()
+ .any(|segment| segment.size > IggyByteSize::default());
+ partition.stats.set_current_offset(current_offset);
+
+ configure_consumer_offsets(&mut partition, config, namespace, current_offset);
+ ensure_initial_segment(&mut partition, config, stream_id, topic_id, partition_id).await?;
+
+ Ok(partition)
+}
+
+async fn hydrate_partition_log(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ loaded_log: server::streaming::partitions::log::SegmentedLog<
+ server::streaming::partitions::journal::MemoryMessageJournal,
+ >,
+) -> Result<(), ServerNgError> {
+ // TODO: decouple the loading logic from the `server` crate. This currently
+ // adapts the old server segmented log into the new `partitions` log.
+ for (segment_index, (segment, storage)) in loaded_log
+ .segments()
+ .iter()
+ .zip(loaded_log.storages().iter().cloned())
+ .enumerate()
+ {
+ let max_timestamp = match loaded_log
+ .indexes()
+ .get(segment_index)
+ .and_then(|indexes| indexes.as_ref())
+ {
+ Some(indexes) => indexes_max_timestamp(indexes),
+ None => load_segment_max_timestamp(&storage, stream_id, topic_id, partition_id).await?,
+ };
+ partition.log.add_persisted_segment(
+ convert_segment(segment, max_timestamp),
+ storage,
+ None,
+ None,
+ );
+ }
+
+ if let Some(active_index) = partition.log.segments().len().checked_sub(1) {
+ let storage = &partition.log.storages()[active_index];
+ if let (Some(messages_reader), Some(index_reader)) = (
+ storage.messages_reader.as_ref(),
+ storage.index_reader.as_ref(),
+ ) {
+ let index_path = index_reader.path();
+ let index_size = std::fs::metadata(&index_path).map_or(0, |metadata| metadata.len());
+ partition.log.messages_writers_mut()[active_index] = Some(Rc::new(
Review Comment:
`hydrate_partition_log` re-opens active segment files via `OpenOptions::create(true).write(true)` with `file_exists=true` hardcoded at lines 527-541. if the file vanished between `load_segments` and writer init, `metadata().len()` returns 0 and the size atomic resets to 0, but in-memory `segment.end_offset` still holds the stale value - next append writes at file pos 0 with logical offset N+1. file-pos / logical-offset mismatch, segment metadata diverges from file content. open with `create=false` when `file_exists=true` and fail loudly if missing.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
Review Comment:
`restore_metadata_consensus` calls `consensus.set_view(header.view)` BEFORE `consensus.init()`, but `set_log_view` and `set_last_prepare_checksum` (lines 272-273) happen AFTER `init()`. `init()`'s `is_primary` check uses the already-restored view. on a multi-replica boot at non-zero recovered view, primary-side timers can arm in the brief window before `log_view` and the prepare checksum are set. order all view / log_view / checksum sets before `init()`, or post-restore enter `ViewChange` instead of `Normal` so DVC quorum drives state.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
Review Comment:
`zero_out_all` on each topic stat decrements parent stream stats by the topic value (`core/common/src/types/streaming_stats.rs:196-209` does `parent.fetch_sub(prev)`). that assumes `snapshot.stream.size == sum(snapshot.topic.size)`. if the snapshot was filled non-atomically, `fetch_sub` on `AtomicU64` wraps on underflow and the stream counter ends up near `u64::MAX`. cleaner: zero stream stats explicitly via `stream_stats.zero_out_all()` and skip the per-topic decrement.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
Review Comment:
two separate `streams.read(|inner| ...)` traversals at boot - one to count partitions (286-298), one to collect topic_stats + namespaces (314-326). both walk the same nested structure. cold path, but trivially mergeable into a single closure pass.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
Review Comment:
`resolve_tcp_topology` is called here in `RunServerNg::run`, then again at line 213 in `bootstrap()`. `main.rs:69-70` invokes `bootstrap()` then `run()`, with `write_current_config` landing in between. no drift today because `config` isn't reloaded, but any future hot-reload or config mutation between the two calls is a feedback-loop hazard. resolve once, pass topology down.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
+ topic_stats.zero_out_all();
+ }
+
+ for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
+ let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
+ let partition = load_partition(
+ config,
+ namespace,
+ topic_stats,
+ &partition_metadata,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?;
+ let local_idx = partitions.insert(namespace, partition);
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ );
+ }
+
+ let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
+ let senders = vec![sender];
+ let shard_handle = Rc::new(RefCell::new(None));
+ let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
+ let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let built = IggyShardBuilder::new(
+ ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ Rc::clone(&bus),
+ on_replica_message,
+ on_client_request,
+ metadata,
+ partitions,
+ senders,
+ inbox,
+ shards_table,
+ PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
+ CoordinatorConfig::default(),
+ bus.token(),
+ )
+ .build();
+ if let Some(refresh_task) = built.refresh_task {
+ bus.track_background(refresh_task);
+ }
+
+ let shard = Rc::new(built.shard);
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ Ok(shard)
+}
+
+const fn validate_recovered_namespace(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ let namespace = &config.extra.namespace;
+ if stream_id < namespace.max_streams
+ && topic_id < namespace.max_topics
+ && partition_id < namespace.max_partitions
+ {
+ return Ok(());
+ }
+
+ Err(ServerNgError::RecoveredNamespaceOutOfBounds {
+ stream_id,
+ topic_id,
+ partition_id,
+ max_streams: namespace.max_streams,
+ max_topics: namespace.max_topics,
+ max_partitions: namespace.max_partitions,
+ })
+}
+
+async fn load_partition(
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ topic_stats: Arc<TopicStats>,
+ partition_metadata: &Partition,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerNgError> {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let stats = Arc::new(PartitionStats::new(topic_stats));
+ let consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ namespace.inner(),
+ bus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+
+ // TODO: decouple the loading logic from the `server` crate and load directly
+ // into the new `partitions` log/runtime types.
+ let loaded_log = server::bootstrap::load_segments(
+ &config.system,
+ stream_id,
+ topic_id,
+ partition_id,
+ config
+ .system
+ .get_partition_path(stream_id, topic_id, partition_id),
+ stats.clone(),
+ )
+ .await
+ .map_err(|source| ServerNgError::PartitionLogLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+
+ let mut partition = IggyPartition::new(stats.clone(), consensus);
+ hydrate_partition_log(
+ &mut partition,
+ config,
+ stream_id,
+ topic_id,
+ partition_id,
+ loaded_log,
+ )
+ .await?;
+
+ let current_offset = partition
+ .log
+ .segments()
+ .iter()
+ .filter(|segment| segment.size > IggyByteSize::default())
+ .map(|segment| segment.end_offset)
+ .max()
+ .unwrap_or(0);
+ partition.created_at = partition_metadata.created_at;
+ partition.offset.store(current_offset, Ordering::Release);
+ partition
+ .dirty_offset
+ .store(current_offset, Ordering::Relaxed);
+ partition.should_increment_offset = partition
+ .log
+ .segments()
+ .iter()
+ .any(|segment| segment.size > IggyByteSize::default());
+ partition.stats.set_current_offset(current_offset);
+
+ configure_consumer_offsets(&mut partition, config, namespace, current_offset);
+ ensure_initial_segment(&mut partition, config, stream_id, topic_id, partition_id).await?;
+
+ Ok(partition)
+}
+
+async fn hydrate_partition_log(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ loaded_log: server::streaming::partitions::log::SegmentedLog<
+ server::streaming::partitions::journal::MemoryMessageJournal,
+ >,
+) -> Result<(), ServerNgError> {
+ // TODO: decouple the loading logic from the `server` crate. This currently
+ // adapts the old server segmented log into the new `partitions` log.
+ for (segment_index, (segment, storage)) in loaded_log
+ .segments()
+ .iter()
+ .zip(loaded_log.storages().iter().cloned())
+ .enumerate()
+ {
+ let max_timestamp = match loaded_log
+ .indexes()
+ .get(segment_index)
+ .and_then(|indexes| indexes.as_ref())
+ {
+ Some(indexes) => indexes_max_timestamp(indexes),
+ None => load_segment_max_timestamp(&storage, stream_id, topic_id, partition_id).await?,
+ };
+ partition.log.add_persisted_segment(
+ convert_segment(segment, max_timestamp),
+ storage,
+ None,
+ None,
+ );
+ }
+
+ if let Some(active_index) = partition.log.segments().len().checked_sub(1) {
+ let storage = &partition.log.storages()[active_index];
+ if let (Some(messages_reader), Some(index_reader)) = (
+ storage.messages_reader.as_ref(),
+ storage.index_reader.as_ref(),
+ ) {
+ let index_path = index_reader.path();
+ let index_size = std::fs::metadata(&index_path).map_or(0, |metadata| metadata.len());
+ partition.log.messages_writers_mut()[active_index] = Some(Rc::new(
+ MessagesWriter::new(
+ &messages_reader.path(),
+ Rc::new(AtomicU64::new(u64::from(messages_reader.file_size()))),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::MessagesWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ partition.log.index_writers_mut()[active_index] = Some(Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(AtomicU64::new(index_size)),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::IndexWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ }
+ }
+
+ Ok(())
+}
+
+fn convert_segment(segment: &iggy_common::Segment, max_timestamp: u64) -> Segment {
+ Segment {
+ sealed: segment.sealed,
+ start_timestamp: segment.start_timestamp,
+ end_timestamp: segment.end_timestamp,
+ max_timestamp,
+ current_position: u64::from(segment.current_position),
+ start_offset: segment.start_offset,
+ end_offset: segment.end_offset,
+ size: segment.size,
+ max_size: segment.max_size,
+ }
+}
+
+fn indexes_max_timestamp(indexes: &server::streaming::segments::IggyIndexesMut) -> u64 {
+ let mut max_timestamp = 0;
+ for index in 0..indexes.count() {
+ if let Some(index_view) = indexes.get(index) {
+ max_timestamp = max_timestamp.max(index_view.timestamp());
+ }
+ }
+
+ max_timestamp
+}
+
+async fn load_segment_max_timestamp(
+ storage: &iggy_common::SegmentStorage,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<u64, ServerNgError> {
+ let Some(index_reader) = storage.index_reader.as_ref() else {
+ return Ok(0);
+ };
+
+ let indexes = index_reader
+ .load_all_indexes_from_disk()
+ .await
+ .map_err(|source| ServerNgError::SegmentIndexesLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+ Ok(indexes_max_timestamp(&indexes))
+}
+
+fn configure_consumer_offsets(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ current_offset: u64,
+) {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let consumer_offsets_path =
+ config
+ .system
+ .get_consumer_offsets_path(stream_id, topic_id, partition_id);
+ let consumer_group_offsets_path =
+ config
+ .system
+ .get_consumer_group_offsets_path(stream_id, topic_id, partition_id);
+
+ // TODO: decouple consumer offset loading from the `server` crate.
+ let loaded_consumer_offsets = load_consumer_offsets(&consumer_offsets_path).unwrap_or_default();
+ let consumer_offsets = ConsumerOffsets::with_capacity(loaded_consumer_offsets.len());
+ {
+ let guard = consumer_offsets.pin();
+ for offset in loaded_consumer_offsets {
+ if offset.offset.load(Ordering::Relaxed) > current_offset {
+ offset.offset.store(current_offset, Ordering::Relaxed);
+ }
+ guard.insert(offset.consumer_id as usize, offset);
+ }
+ }
+
+ // TODO: decouple consumer group offset loading from the `server` crate.
+ let loaded_group_offsets =
+ load_consumer_group_offsets(&consumer_group_offsets_path).unwrap_or_default();
+ let consumer_group_offsets = ConsumerGroupOffsets::with_capacity(loaded_group_offsets.len());
+ {
+ let guard = consumer_group_offsets.pin();
+ for (group_id, offset) in loaded_group_offsets {
+ if offset.offset.load(Ordering::Relaxed) > current_offset {
+ offset.offset.store(current_offset, Ordering::Relaxed);
+ }
+ guard.insert(group_id, offset);
+ }
+ }
+
+ partition.configure_consumer_offset_storage(
+ consumer_offsets_path,
+ consumer_group_offsets_path,
+ consumer_offsets,
+ consumer_group_offsets,
+ config.system.partition.enforce_fsync,
+ );
+}
+
+async fn ensure_initial_segment(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ if partition.log.has_segments() {
+ return Ok(());
+ }
+
+ // TODO: decouple segment storage creation from the `server` crate.
+ let storage =
+ create_segment_storage(&config.system, stream_id, topic_id, partition_id, 0, 0, 0)
+ .await
+ .map_err(|source| ServerNgError::InitialSegmentStorage {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+ let messages_path = config
+ .system
+ .get_messages_file_path(stream_id, topic_id, partition_id, 0);
+ let index_path = config
+ .system
+ .get_index_path(stream_id, topic_id, partition_id, 0);
+ partition.log.add_persisted_segment(
+ Segment::new(0, config.system.segment.size),
+ storage,
+ Some(Rc::new(
+ MessagesWriter::new(
+ &messages_path,
+ Rc::new(AtomicU64::new(0)),
+ config.system.partition.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|source| ServerNgError::MessagesWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ )),
+ Some(Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(AtomicU64::new(0)),
+ config.system.partition.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|source| ServerNgError::IndexWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ )),
+ );
+ partition.stats.increment_segments_count(1);
+
+ Ok(())
+}
+
+fn resolve_tcp_topology(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<TcpTopology, ServerNgError> {
+ let default_client_addr = parse_socket_addr("tcp.address", &config.tcp.address)?;
+ let default_ws_addr = resolve_optional_listener_addr(
+ config.websocket.enabled,
+ "websocket.address",
+ &config.websocket.address,
+ )?;
+ let default_quic_addr =
+ resolve_optional_listener_addr(config.quic.enabled, "quic.address", &config.quic.address)?;
+ if !config.cluster.enabled {
+ return Ok(TcpTopology {
+ // Keep parity with the current server binary and the integration
+ // harness: `--replica-id` may be passed unconditionally, but in
+ // single-node mode there is only replica 0.
+ self_replica_id: SHARD_REPLICA_ID,
+ replica_count: 1,
+ client_listen_addr: default_client_addr,
+ replica_listen_addr: Some(SocketAddr::new(default_client_addr.ip(), 0)),
Review Comment:
single-node mode (`!config.cluster.enabled`) sets `replica_listen_addr: Some(SocketAddr::new(default_client_addr.ip(), 0))` with `replica_count = 1`. `start_via_replica_io` then unconditionally binds the replica listener (`core/message_bus/src/replica/io.rs:209`). wasted fd, wasted ephemeral port, no consumers. skip the replica listener when `replica_count == 1`.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
Review Comment:
`RunServerNg` trait has exactly one impl (`Rc<ServerNgShard>`) and one caller (`main.rs:26`). single-impl trait used purely for `Rc`-receiver dispatch. a free function `pub async fn run(shard: &Rc<ServerNgShard>, ...)` is cleaner.
##########
core/server-ng/src/server_error.rs:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.
+ */
+
+use metadata::impls::recovery::RecoveryError;
+// TODO: decouple logging errors from the `server` crate.
+use server::server_error::LogError;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum ServerNgError {
+ #[error("failed to load server-ng config")]
+ Config(#[source] configs::ConfigurationError),
+ #[error("failed to prepare server-ng directories")]
+ CreateDirectories(#[source] iggy_common::IggyError),
+ #[error("failed to serialize current server-ng config")]
+ CurrentConfigSerialize(#[source] toml::ser::Error),
+ #[error("failed to write current server-ng config at {path}")]
+ CurrentConfigWrite {
+ path: String,
+ #[source]
+ source: std::io::Error,
+ },
+ #[error("failed to initialize server-ng logging")]
+ Logging(#[source] LogError),
+ #[error("failed to recover metadata snapshot and journal")]
+ MetadataRecovery(#[source] RecoveryError),
Review Comment:
`MetadataRecovery(#[source] RecoveryError)` chain works, but `RecoveryError::StateMachine(IggyError)` (`core/metadata/src/impls/recovery.rs:30-46`) wraps with the prefix `"recovery state machine error: "` on Display, partially burying the underlying `IggyError` context. fine for the CLI, less fine for log scrapers that read top-line Display only.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
+ topic_stats.zero_out_all();
+ }
+
+ for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
+ let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
+ let partition = load_partition(
+ config,
+ namespace,
+ topic_stats,
+ &partition_metadata,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?;
+ let local_idx = partitions.insert(namespace, partition);
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ );
+ }
+
+ let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
+ let senders = vec![sender];
+ let shard_handle = Rc::new(RefCell::new(None));
+ let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
+ let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let built = IggyShardBuilder::new(
+ ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ Rc::clone(&bus),
+ on_replica_message,
+ on_client_request,
+ metadata,
+ partitions,
+ senders,
+ inbox,
+ shards_table,
+ PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
+ CoordinatorConfig::default(),
+ bus.token(),
+ )
+ .build();
+ if let Some(refresh_task) = built.refresh_task {
+ bus.track_background(refresh_task);
+ }
+
+ let shard = Rc::new(built.shard);
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ Ok(shard)
+}
+
+const fn validate_recovered_namespace(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ let namespace = &config.extra.namespace;
+ if stream_id < namespace.max_streams
+ && topic_id < namespace.max_topics
+ && partition_id < namespace.max_partitions
+ {
+ return Ok(());
+ }
+
+ Err(ServerNgError::RecoveredNamespaceOutOfBounds {
+ stream_id,
+ topic_id,
+ partition_id,
+ max_streams: namespace.max_streams,
+ max_topics: namespace.max_topics,
+ max_partitions: namespace.max_partitions,
+ })
+}
+
+async fn load_partition(
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ topic_stats: Arc<TopicStats>,
+ partition_metadata: &Partition,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerNgError> {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let stats = Arc::new(PartitionStats::new(topic_stats));
+ let consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ namespace.inner(),
+ bus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+
+ // TODO: decouple the loading logic from the `server` crate and load directly
+ // into the new `partitions` log/runtime types.
+ let loaded_log = server::bootstrap::load_segments(
+ &config.system,
+ stream_id,
+ topic_id,
+ partition_id,
+ config
+ .system
+ .get_partition_path(stream_id, topic_id, partition_id),
+ stats.clone(),
+ )
+ .await
+ .map_err(|source| ServerNgError::PartitionLogLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+
+ let mut partition = IggyPartition::new(stats.clone(), consensus);
+ hydrate_partition_log(
+ &mut partition,
+ config,
+ stream_id,
+ topic_id,
+ partition_id,
+ loaded_log,
+ )
+ .await?;
+
+ let current_offset = partition
+ .log
+ .segments()
+ .iter()
+ .filter(|segment| segment.size > IggyByteSize::default())
+ .map(|segment| segment.end_offset)
+ .max()
+ .unwrap_or(0);
+ partition.created_at = partition_metadata.created_at;
+ partition.offset.store(current_offset, Ordering::Release);
+ partition
+ .dirty_offset
+ .store(current_offset, Ordering::Relaxed);
+ partition.should_increment_offset = partition
+ .log
+ .segments()
+ .iter()
+ .any(|segment| segment.size > IggyByteSize::default());
+ partition.stats.set_current_offset(current_offset);
+
+ configure_consumer_offsets(&mut partition, config, namespace, current_offset);
+ ensure_initial_segment(&mut partition, config, stream_id, topic_id, partition_id).await?;
+
+ Ok(partition)
+}
+
+async fn hydrate_partition_log(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ loaded_log: server::streaming::partitions::log::SegmentedLog<
+ server::streaming::partitions::journal::MemoryMessageJournal,
+ >,
+) -> Result<(), ServerNgError> {
+ // TODO: decouple the loading logic from the `server` crate. This currently
+ // adapts the old server segmented log into the new `partitions` log.
+ for (segment_index, (segment, storage)) in loaded_log
+ .segments()
+ .iter()
+ .zip(loaded_log.storages().iter().cloned())
+ .enumerate()
+ {
+ let max_timestamp = match loaded_log
+ .indexes()
+ .get(segment_index)
+ .and_then(|indexes| indexes.as_ref())
+ {
+ Some(indexes) => indexes_max_timestamp(indexes),
+ None => load_segment_max_timestamp(&storage, stream_id, topic_id, partition_id).await?,
+ };
+ partition.log.add_persisted_segment(
+ convert_segment(segment, max_timestamp),
+ storage,
+ None,
+ None,
+ );
+ }
+
+ if let Some(active_index) = partition.log.segments().len().checked_sub(1) {
+ let storage = &partition.log.storages()[active_index];
+ if let (Some(messages_reader), Some(index_reader)) = (
+ storage.messages_reader.as_ref(),
+ storage.index_reader.as_ref(),
+ ) {
+ let index_path = index_reader.path();
+ let index_size = std::fs::metadata(&index_path).map_or(0, |metadata| metadata.len());
Review Comment:
`std::fs::metadata(&index_path)` is a blocking syscall called inside `async fn hydrate_partition_log` running on the compio reactor. bootstrap-only so it's a brief reactor stall, but trivial to fix with `compio::fs::metadata`.
##########
core/server-ng/src/server_error.rs:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.
+ */
+
+use metadata::impls::recovery::RecoveryError;
+// TODO: decouple logging errors from the `server` crate.
+use server::server_error::LogError;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum ServerNgError {
+ #[error("failed to load server-ng config")]
+ Config(#[source] configs::ConfigurationError),
+ #[error("failed to prepare server-ng directories")]
+ CreateDirectories(#[source] iggy_common::IggyError),
+ #[error("failed to serialize current server-ng config")]
+ CurrentConfigSerialize(#[source] toml::ser::Error),
+ #[error("failed to write current server-ng config at {path}")]
+ CurrentConfigWrite {
+ path: String,
+ #[source]
+ source: std::io::Error,
+ },
+ #[error("failed to initialize server-ng logging")]
+ Logging(#[source] LogError),
+ #[error("failed to recover metadata snapshot and journal")]
+ MetadataRecovery(#[source] RecoveryError),
+ #[error("failed to parse {context} socket address '{address}'")]
Review Comment:
also `transport: &'static str` at lines 79-84. `context` and `transport` are stringly-typed enums in disguise. call sites use literal strings `"tcp.tls"`, `"quic"`, `"tcp.address"`, `"websocket.address"`. typo-prone, no exhaustiveness. replace with small `enum ListenerKind { Tcp, TcpTls, Quic, Ws }` and `enum AddressContext`.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
+ topic_stats.zero_out_all();
+ }
+
+ for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
+ let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
+ let partition = load_partition(
+ config,
+ namespace,
+ topic_stats,
+ &partition_metadata,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?;
+ let local_idx = partitions.insert(namespace, partition);
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ );
+ }
+
+ let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
+ let senders = vec![sender];
+ let shard_handle = Rc::new(RefCell::new(None));
+ let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
+ let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let built = IggyShardBuilder::new(
+ ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ Rc::clone(&bus),
+ on_replica_message,
+ on_client_request,
+ metadata,
+ partitions,
+ senders,
+ inbox,
+ shards_table,
+ PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
+ CoordinatorConfig::default(),
+ bus.token(),
+ )
+ .build();
+ if let Some(refresh_task) = built.refresh_task {
+ bus.track_background(refresh_task);
+ }
+
+ let shard = Rc::new(built.shard);
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ Ok(shard)
+}
+
+const fn validate_recovered_namespace(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ let namespace = &config.extra.namespace;
+ if stream_id < namespace.max_streams
+ && topic_id < namespace.max_topics
+ && partition_id < namespace.max_partitions
+ {
+ return Ok(());
+ }
+
+ Err(ServerNgError::RecoveredNamespaceOutOfBounds {
+ stream_id,
+ topic_id,
+ partition_id,
+ max_streams: namespace.max_streams,
+ max_topics: namespace.max_topics,
+ max_partitions: namespace.max_partitions,
+ })
+}
+
+async fn load_partition(
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ topic_stats: Arc<TopicStats>,
+ partition_metadata: &Partition,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerNgError> {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let stats = Arc::new(PartitionStats::new(topic_stats));
+ let consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ namespace.inner(),
+ bus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+
+ // TODO: decouple the loading logic from the `server` crate and load directly
+ // into the new `partitions` log/runtime types.
+ let loaded_log = server::bootstrap::load_segments(
+ &config.system,
+ stream_id,
+ topic_id,
+ partition_id,
+ config
+ .system
+ .get_partition_path(stream_id, topic_id, partition_id),
+ stats.clone(),
+ )
+ .await
+ .map_err(|source| ServerNgError::PartitionLogLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+
+ let mut partition = IggyPartition::new(stats.clone(), consensus);
+ hydrate_partition_log(
+ &mut partition,
+ config,
+ stream_id,
+ topic_id,
+ partition_id,
+ loaded_log,
+ )
+ .await?;
+
+ let current_offset = partition
+ .log
+ .segments()
+ .iter()
+ .filter(|segment| segment.size > IggyByteSize::default())
+ .map(|segment| segment.end_offset)
+ .max()
+ .unwrap_or(0);
+ partition.created_at = partition_metadata.created_at;
+ partition.offset.store(current_offset, Ordering::Release);
+ partition
+ .dirty_offset
+ .store(current_offset, Ordering::Relaxed);
+ partition.should_increment_offset = partition
+ .log
+ .segments()
+ .iter()
+ .any(|segment| segment.size > IggyByteSize::default());
+ partition.stats.set_current_offset(current_offset);
+
+ configure_consumer_offsets(&mut partition, config, namespace, current_offset);
+ ensure_initial_segment(&mut partition, config, stream_id, topic_id, partition_id).await?;
+
+ Ok(partition)
+}
+
+async fn hydrate_partition_log(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ loaded_log: server::streaming::partitions::log::SegmentedLog<
+ server::streaming::partitions::journal::MemoryMessageJournal,
+ >,
+) -> Result<(), ServerNgError> {
+ // TODO: decouple the loading logic from the `server` crate. This currently
+ // adapts the old server segmented log into the new `partitions` log.
+ for (segment_index, (segment, storage)) in loaded_log
+ .segments()
+ .iter()
+ .zip(loaded_log.storages().iter().cloned())
+ .enumerate()
+ {
+ let max_timestamp = match loaded_log
+ .indexes()
+ .get(segment_index)
+ .and_then(|indexes| indexes.as_ref())
+ {
+ Some(indexes) => indexes_max_timestamp(indexes),
+ None => load_segment_max_timestamp(&storage, stream_id, topic_id, partition_id).await?,
+ };
+ partition.log.add_persisted_segment(
+ convert_segment(segment, max_timestamp),
+ storage,
+ None,
+ None,
+ );
+ }
+
+ if let Some(active_index) = partition.log.segments().len().checked_sub(1) {
+ let storage = &partition.log.storages()[active_index];
+ if let (Some(messages_reader), Some(index_reader)) = (
+ storage.messages_reader.as_ref(),
+ storage.index_reader.as_ref(),
+ ) {
+ let index_path = index_reader.path();
+ let index_size = std::fs::metadata(&index_path).map_or(0, |metadata| metadata.len());
+ partition.log.messages_writers_mut()[active_index] = Some(Rc::new(
+ MessagesWriter::new(
+ &messages_reader.path(),
+ Rc::new(AtomicU64::new(u64::from(messages_reader.file_size()))),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::MessagesWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ partition.log.index_writers_mut()[active_index] = Some(Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(AtomicU64::new(index_size)),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::IndexWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ }
+ }
+
+ Ok(())
+}
+
+fn convert_segment(segment: &iggy_common::Segment, max_timestamp: u64) -> Segment {
+ Segment {
+ sealed: segment.sealed,
+ start_timestamp: segment.start_timestamp,
+ end_timestamp: segment.end_timestamp,
+ max_timestamp,
+ current_position: u64::from(segment.current_position),
+ start_offset: segment.start_offset,
+ end_offset: segment.end_offset,
+ size: segment.size,
+ max_size: segment.max_size,
+ }
+}
+
+fn indexes_max_timestamp(indexes: &server::streaming::segments::IggyIndexesMut) -> u64 {
+ let mut max_timestamp = 0;
+ for index in 0..indexes.count() {
+ if let Some(index_view) = indexes.get(index) {
+ max_timestamp = max_timestamp.max(index_view.timestamp());
+ }
+ }
+
+ max_timestamp
+}
+
+async fn load_segment_max_timestamp(
+ storage: &iggy_common::SegmentStorage,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<u64, ServerNgError> {
+ let Some(index_reader) = storage.index_reader.as_ref() else {
+ return Ok(0);
+ };
+
+ let indexes = index_reader
+ .load_all_indexes_from_disk()
+ .await
+ .map_err(|source| ServerNgError::SegmentIndexesLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+ Ok(indexes_max_timestamp(&indexes))
+}
+
+fn configure_consumer_offsets(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ current_offset: u64,
+) {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let consumer_offsets_path =
+ config
+ .system
+ .get_consumer_offsets_path(stream_id, topic_id, partition_id);
+ let consumer_group_offsets_path =
+ config
+ .system
+ .get_consumer_group_offsets_path(stream_id, topic_id, partition_id);
+
+ // TODO: decouple consumer offset loading from the `server` crate.
+ let loaded_consumer_offsets = load_consumer_offsets(&consumer_offsets_path).unwrap_or_default();
+ let consumer_offsets = ConsumerOffsets::with_capacity(loaded_consumer_offsets.len());
+ {
+ let guard = consumer_offsets.pin();
+ for offset in loaded_consumer_offsets {
+ if offset.offset.load(Ordering::Relaxed) > current_offset {
+ offset.offset.store(current_offset, Ordering::Relaxed);
+ }
+ guard.insert(offset.consumer_id as usize, offset);
+ }
+ }
+
+ // TODO: decouple consumer group offset loading from the `server` crate.
+ let loaded_group_offsets =
+ load_consumer_group_offsets(&consumer_group_offsets_path).unwrap_or_default();
+ let consumer_group_offsets = ConsumerGroupOffsets::with_capacity(loaded_group_offsets.len());
+ {
+ let guard = consumer_group_offsets.pin();
+ for (group_id, offset) in loaded_group_offsets {
+ if offset.offset.load(Ordering::Relaxed) > current_offset {
+ offset.offset.store(current_offset, Ordering::Relaxed);
+ }
+ guard.insert(group_id, offset);
+ }
+ }
+
+ partition.configure_consumer_offset_storage(
+ consumer_offsets_path,
+ consumer_group_offsets_path,
+ consumer_offsets,
+ consumer_group_offsets,
+ config.system.partition.enforce_fsync,
+ );
+}
+
+async fn ensure_initial_segment(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ if partition.log.has_segments() {
+ return Ok(());
+ }
+
+ // TODO: decouple segment storage creation from the `server` crate.
+ let storage =
+ create_segment_storage(&config.system, stream_id, topic_id, partition_id, 0, 0, 0)
+ .await
+ .map_err(|source| ServerNgError::InitialSegmentStorage {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+ let messages_path = config
+ .system
+ .get_messages_file_path(stream_id, topic_id, partition_id, 0);
+ let index_path = config
+ .system
+ .get_index_path(stream_id, topic_id, partition_id, 0);
+ partition.log.add_persisted_segment(
+ Segment::new(0, config.system.segment.size),
+ storage,
+ Some(Rc::new(
+ MessagesWriter::new(
+ &messages_path,
+ Rc::new(AtomicU64::new(0)),
+ config.system.partition.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|source| ServerNgError::MessagesWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ )),
+ Some(Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(AtomicU64::new(0)),
+ config.system.partition.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|source| ServerNgError::IndexWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ )),
+ );
+ partition.stats.increment_segments_count(1);
+
+ Ok(())
+}
+
+fn resolve_tcp_topology(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<TcpTopology, ServerNgError> {
+ let default_client_addr = parse_socket_addr("tcp.address", &config.tcp.address)?;
+ let default_ws_addr = resolve_optional_listener_addr(
+ config.websocket.enabled,
+ "websocket.address",
+ &config.websocket.address,
+ )?;
+ let default_quic_addr =
+ resolve_optional_listener_addr(config.quic.enabled, "quic.address", &config.quic.address)?;
+ if !config.cluster.enabled {
+ return Ok(TcpTopology {
+ // Keep parity with the current server binary and the integration
+ // harness: `--replica-id` may be passed unconditionally, but in
+ // single-node mode there is only replica 0.
+ self_replica_id: SHARD_REPLICA_ID,
+ replica_count: 1,
+ client_listen_addr: default_client_addr,
+ replica_listen_addr: Some(SocketAddr::new(default_client_addr.ip(), 0)),
+ ws_listen_addr: default_ws_addr,
+ quic_listen_addr: default_quic_addr,
+ tcp_tls_listen_addr: config.tcp.tls.enabled.then_some(default_client_addr),
+ peers: Vec::new(),
+ });
+ }
+
+ let self_replica_id = current_replica_id.ok_or(ServerNgError::MissingReplicaId)?;
+
+ let self_node = config
+ .cluster
+ .nodes
+ .iter()
+ .find(|node| node.replica_id == self_replica_id)
+ .ok_or(ServerNgError::ClusterNodeNotFound {
+ replica_id: self_replica_id,
+ })?;
+ let replica_count = u8::try_from(config.cluster.nodes.len()).map_err(|_| {
+ ServerNgError::ClusterReplicaCountTooLarge {
+ count: config.cluster.nodes.len(),
+ }
+ })?;
+ let (client_listen_addr, ws_listen_addr, quic_listen_addr) = resolve_cluster_client_addrs(
+ self_node,
+ default_client_addr,
+ default_ws_addr,
+ default_quic_addr,
+ )?;
+ let replica_port =
+ self_node
+ .ports
+ .tcp_replica
+ .ok_or(ServerNgError::ClusterReplicaPortMissing {
+ replica_id: self_node.replica_id,
+ })?;
+ let replica_listen_addr = Some(socket_addr_from_parts(
+ "cluster.nodes[*].ports.tcp_replica",
+ &self_node.ip,
+ replica_port,
+ )?);
+ let peers = resolve_cluster_replica_peers(&config.cluster.nodes, self_replica_id)?;
+
+ Ok(TcpTopology {
+ self_replica_id,
+ replica_count,
+ client_listen_addr,
+ replica_listen_addr,
+ ws_listen_addr,
+ quic_listen_addr,
+ tcp_tls_listen_addr: config.tcp.tls.enabled.then_some(client_listen_addr),
+ peers,
+ })
+}
+
+fn resolve_optional_listener_addr(
+ enabled: bool,
+ context: &'static str,
+ address: &str,
+) -> Result<Option<SocketAddr>, ServerNgError> {
+ if enabled {
+ return Ok(Some(parse_socket_addr(context, address)?));
+ }
+ Ok(None)
+}
+
+fn resolve_cluster_client_addrs(
+ self_node: &configs::cluster::ClusterNodeConfig,
+ default_client_addr: SocketAddr,
+ default_ws_addr: Option<SocketAddr>,
+ default_quic_addr: Option<SocketAddr>,
+) -> Result<(SocketAddr, Option<SocketAddr>, Option<SocketAddr>), ServerNgError> {
+ let client_port = self_node
+ .ports
+ .tcp
+ .unwrap_or_else(|| default_client_addr.port());
+ let client_listen_addr =
+ socket_addr_from_parts("cluster.nodes[*].ports.tcp", &self_node.ip, client_port)?;
+ let ws_listen_addr = resolve_cluster_optional_addr(
+ self_node,
+ "cluster.nodes[*].ports.websocket",
+ default_ws_addr,
+ |ports| ports.websocket,
+ )?;
+ let quic_listen_addr = resolve_cluster_optional_addr(
+ self_node,
+ "cluster.nodes[*].ports.quic",
+ default_quic_addr,
+ |ports| ports.quic,
+ )?;
+ Ok((client_listen_addr, ws_listen_addr, quic_listen_addr))
+}
+
+fn resolve_cluster_optional_addr(
+ self_node: &configs::cluster::ClusterNodeConfig,
+ context: &'static str,
+ default_addr: Option<SocketAddr>,
+ port_selector: impl Fn(&configs::cluster::TransportPorts) -> Option<u16>,
+) -> Result<Option<SocketAddr>, ServerNgError> {
+ let Some(default_addr) = default_addr else {
+ return Ok(None);
+ };
+ let port = port_selector(&self_node.ports).unwrap_or_else(|| default_addr.port());
+ socket_addr_from_parts(context, &self_node.ip, port).map(Some)
+}
+
+fn resolve_cluster_replica_peers(
+ nodes: &[configs::cluster::ClusterNodeConfig],
+ self_replica_id: u8,
+) -> Result<Vec<(u8, SocketAddr)>, ServerNgError> {
+ let mut peers = Vec::with_capacity(nodes.len().saturating_sub(1));
+ for node in nodes {
+ if node.replica_id == self_replica_id {
+ continue;
+ }
+ let replica_port =
+ node.ports
+ .tcp_replica
+ .ok_or(ServerNgError::ClusterReplicaPortMissing {
+ replica_id: node.replica_id,
+ })?;
+ peers.push((
+ node.replica_id,
+ socket_addr_from_parts("cluster.nodes[*].ports.tcp_replica", &node.ip, replica_port)?,
+ ));
+ }
+ Ok(peers)
+}
+
+async fn start_tcp_runtime(
+ shard: &Rc<ServerNgShard>,
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ accepted_replica: AcceptedReplicaFn,
+ accepted_clients: LocalClientAcceptFns,
+) -> Result<(), ServerNgError> {
+ if config.tcp.enabled && !config.tcp.tls.enabled {
+ return start_via_replica_io(shard, config, topology, accepted_replica, accepted_clients)
+ .await;
+ }
+
+ start_manual_runtime(shard, config, topology, accepted_replica, accepted_clients).await
+}
+
+async fn start_via_replica_io(
+ shard: &Rc<ServerNgShard>,
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ accepted_replica: AcceptedReplicaFn,
+ accepted_clients: LocalClientAcceptFns,
+) -> Result<(), ServerNgError> {
+ let replica_addr = topology
+ .replica_listen_addr
+ .expect("topology must include replica listener address");
+ let quic_credentials = topology
+ .quic_listen_addr
+ .is_some()
+ .then(|| load_quic_server_credentials(config))
+ .transpose()?;
+ let tcp_tls_credentials = topology
+ .tcp_tls_listen_addr
+ .is_some()
+ .then(|| load_tcp_tls_server_credentials(config))
+ .transpose()?;
+
+ let LocalClientAcceptFns {
+ tcp,
+ ws,
+ quic,
+ tcp_tls,
+ } = accepted_clients;
+
+ let bound = replica_io::start_on_shard_zero(
+ &shard.bus,
+ replica_addr,
+ topology.client_listen_addr,
+ topology.ws_listen_addr,
+ topology.quic_listen_addr,
+ quic_credentials,
+ topology.tcp_tls_listen_addr,
+ tcp_tls_credentials,
+ None,
+ None,
+ CLUSTER_ID,
+ topology.self_replica_id,
+ topology.replica_count,
+ topology.peers.clone(),
+ accepted_replica,
+ tcp,
+ topology.ws_listen_addr.map(|_| ws),
+ topology.quic_listen_addr.map(|_| quic),
+ topology.tcp_tls_listen_addr.map(|_| tcp_tls),
+ None,
+ shard.bus.config().reconnect_period,
+ )
+ .await
+ .map_err(ServerNgError::StartListeners)?;
+ let Some(bound) = bound else {
Review Comment:
`let Some(bound) = bound else { return Ok(()); };` is unreachable. `replica_io::start_on_shard_zero` returns `None` only when `bus.shard_id() != 0` (`core/message_bus/src/replica/io.rs:191-194`), but `SHARD_ID` is hardcoded to 0 at `core/server-ng/src/bootstrap.rs:78` and `IggyMessageBus::with_config(SHARD_ID, ...)` is called at line 214. dead branch hides future SHARD_ID parameterization regressions.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
+ topic_stats.zero_out_all();
+ }
+
+ for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
+ let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
+ let partition = load_partition(
+ config,
+ namespace,
+ topic_stats,
+ &partition_metadata,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?;
+ let local_idx = partitions.insert(namespace, partition);
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ );
+ }
+
+ let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
+ let senders = vec![sender];
+ let shard_handle = Rc::new(RefCell::new(None));
+ let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
+ let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let built = IggyShardBuilder::new(
+ ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ Rc::clone(&bus),
+ on_replica_message,
+ on_client_request,
+ metadata,
+ partitions,
+ senders,
+ inbox,
+ shards_table,
+ PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
+ CoordinatorConfig::default(),
+ bus.token(),
+ )
+ .build();
+ if let Some(refresh_task) = built.refresh_task {
+ bus.track_background(refresh_task);
+ }
+
+ let shard = Rc::new(built.shard);
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ Ok(shard)
+}
+
+const fn validate_recovered_namespace(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ let namespace = &config.extra.namespace;
+ if stream_id < namespace.max_streams
+ && topic_id < namespace.max_topics
+ && partition_id < namespace.max_partitions
+ {
+ return Ok(());
+ }
+
+ Err(ServerNgError::RecoveredNamespaceOutOfBounds {
+ stream_id,
+ topic_id,
+ partition_id,
+ max_streams: namespace.max_streams,
+ max_topics: namespace.max_topics,
+ max_partitions: namespace.max_partitions,
+ })
+}
+
+async fn load_partition(
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ topic_stats: Arc<TopicStats>,
+ partition_metadata: &Partition,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerNgError> {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let stats = Arc::new(PartitionStats::new(topic_stats));
+ let consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ namespace.inner(),
+ bus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+
+ // TODO: decouple the loading logic from the `server` crate and load directly
+ // into the new `partitions` log/runtime types.
+ let loaded_log = server::bootstrap::load_segments(
+ &config.system,
+ stream_id,
+ topic_id,
+ partition_id,
+ config
+ .system
+ .get_partition_path(stream_id, topic_id, partition_id),
+ stats.clone(),
+ )
+ .await
+ .map_err(|source| ServerNgError::PartitionLogLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+
+ let mut partition = IggyPartition::new(stats.clone(), consensus);
+ hydrate_partition_log(
+ &mut partition,
+ config,
+ stream_id,
+ topic_id,
+ partition_id,
+ loaded_log,
+ )
+ .await?;
+
+ let current_offset = partition
+ .log
+ .segments()
+ .iter()
+ .filter(|segment| segment.size > IggyByteSize::default())
+ .map(|segment| segment.end_offset)
+ .max()
+ .unwrap_or(0);
+ partition.created_at = partition_metadata.created_at;
+ partition.offset.store(current_offset, Ordering::Release);
+ partition
+ .dirty_offset
+ .store(current_offset, Ordering::Relaxed);
+ partition.should_increment_offset = partition
+ .log
+ .segments()
+ .iter()
+ .any(|segment| segment.size > IggyByteSize::default());
+ partition.stats.set_current_offset(current_offset);
+
+ configure_consumer_offsets(&mut partition, config, namespace, current_offset);
+ ensure_initial_segment(&mut partition, config, stream_id, topic_id, partition_id).await?;
+
+ Ok(partition)
+}
+
+async fn hydrate_partition_log(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ loaded_log: server::streaming::partitions::log::SegmentedLog<
+ server::streaming::partitions::journal::MemoryMessageJournal,
+ >,
+) -> Result<(), ServerNgError> {
+ // TODO: decouple the loading logic from the `server` crate. This currently
+ // adapts the old server segmented log into the new `partitions` log.
+ for (segment_index, (segment, storage)) in loaded_log
+ .segments()
+ .iter()
+ .zip(loaded_log.storages().iter().cloned())
+ .enumerate()
+ {
+ let max_timestamp = match loaded_log
+ .indexes()
+ .get(segment_index)
+ .and_then(|indexes| indexes.as_ref())
+ {
+ Some(indexes) => indexes_max_timestamp(indexes),
+ None => load_segment_max_timestamp(&storage, stream_id, topic_id, partition_id).await?,
+ };
+ partition.log.add_persisted_segment(
+ convert_segment(segment, max_timestamp),
+ storage,
+ None,
+ None,
+ );
+ }
+
+ if let Some(active_index) = partition.log.segments().len().checked_sub(1) {
+ let storage = &partition.log.storages()[active_index];
+ if let (Some(messages_reader), Some(index_reader)) = (
+ storage.messages_reader.as_ref(),
+ storage.index_reader.as_ref(),
+ ) {
+ let index_path = index_reader.path();
+ let index_size = std::fs::metadata(&index_path).map_or(0, |metadata| metadata.len());
+ partition.log.messages_writers_mut()[active_index] = Some(Rc::new(
+ MessagesWriter::new(
+ &messages_reader.path(),
+ Rc::new(AtomicU64::new(u64::from(messages_reader.file_size()))),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::MessagesWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ partition.log.index_writers_mut()[active_index] = Some(Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(AtomicU64::new(index_size)),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::IndexWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ }
+ }
+
+ Ok(())
+}
+
+fn convert_segment(segment: &iggy_common::Segment, max_timestamp: u64) -> Segment {
+ Segment {
+ sealed: segment.sealed,
+ start_timestamp: segment.start_timestamp,
+ end_timestamp: segment.end_timestamp,
+ max_timestamp,
+ current_position: u64::from(segment.current_position),
+ start_offset: segment.start_offset,
+ end_offset: segment.end_offset,
+ size: segment.size,
+ max_size: segment.max_size,
+ }
+}
+
+fn indexes_max_timestamp(indexes: &server::streaming::segments::IggyIndexesMut) -> u64 {
+ let mut max_timestamp = 0;
+ for index in 0..indexes.count() {
+ if let Some(index_view) = indexes.get(index) {
+ max_timestamp = max_timestamp.max(index_view.timestamp());
+ }
+ }
+
+ max_timestamp
+}
+
+async fn load_segment_max_timestamp(
+ storage: &iggy_common::SegmentStorage,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<u64, ServerNgError> {
+ let Some(index_reader) = storage.index_reader.as_ref() else {
+ return Ok(0);
+ };
+
+ let indexes = index_reader
+ .load_all_indexes_from_disk()
+ .await
+ .map_err(|source| ServerNgError::SegmentIndexesLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+ Ok(indexes_max_timestamp(&indexes))
+}
+
+fn configure_consumer_offsets(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ current_offset: u64,
+) {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let consumer_offsets_path =
+ config
+ .system
+ .get_consumer_offsets_path(stream_id, topic_id, partition_id);
+ let consumer_group_offsets_path =
+ config
+ .system
+ .get_consumer_group_offsets_path(stream_id, topic_id, partition_id);
+
+ // TODO: decouple consumer offset loading from the `server` crate.
+ let loaded_consumer_offsets = load_consumer_offsets(&consumer_offsets_path).unwrap_or_default();
+ let consumer_offsets = ConsumerOffsets::with_capacity(loaded_consumer_offsets.len());
+ {
+ let guard = consumer_offsets.pin();
+ for offset in loaded_consumer_offsets {
+ if offset.offset.load(Ordering::Relaxed) > current_offset {
+ offset.offset.store(current_offset, Ordering::Relaxed);
+ }
+ guard.insert(offset.consumer_id as usize, offset);
+ }
+ }
+
+ // TODO: decouple consumer group offset loading from the `server` crate.
+ let loaded_group_offsets =
+ load_consumer_group_offsets(&consumer_group_offsets_path).unwrap_or_default();
+ let consumer_group_offsets = ConsumerGroupOffsets::with_capacity(loaded_group_offsets.len());
+ {
+ let guard = consumer_group_offsets.pin();
+ for (group_id, offset) in loaded_group_offsets {
+ if offset.offset.load(Ordering::Relaxed) > current_offset {
+ offset.offset.store(current_offset, Ordering::Relaxed);
+ }
+ guard.insert(group_id, offset);
+ }
+ }
+
+ partition.configure_consumer_offset_storage(
+ consumer_offsets_path,
+ consumer_group_offsets_path,
+ consumer_offsets,
+ consumer_group_offsets,
+ config.system.partition.enforce_fsync,
+ );
+}
+
+async fn ensure_initial_segment(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ if partition.log.has_segments() {
+ return Ok(());
+ }
+
+ // TODO: decouple segment storage creation from the `server` crate.
+ let storage =
+ create_segment_storage(&config.system, stream_id, topic_id, partition_id, 0, 0, 0)
+ .await
+ .map_err(|source| ServerNgError::InitialSegmentStorage {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+ let messages_path = config
+ .system
+ .get_messages_file_path(stream_id, topic_id, partition_id, 0);
+ let index_path = config
+ .system
+ .get_index_path(stream_id, topic_id, partition_id, 0);
+ partition.log.add_persisted_segment(
+ Segment::new(0, config.system.segment.size),
+ storage,
+ Some(Rc::new(
+ MessagesWriter::new(
+ &messages_path,
+ Rc::new(AtomicU64::new(0)),
+ config.system.partition.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|source| ServerNgError::MessagesWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ )),
+ Some(Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(AtomicU64::new(0)),
+ config.system.partition.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|source| ServerNgError::IndexWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ )),
+ );
+ partition.stats.increment_segments_count(1);
+
+ Ok(())
+}
+
+fn resolve_tcp_topology(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<TcpTopology, ServerNgError> {
+ let default_client_addr = parse_socket_addr("tcp.address", &config.tcp.address)?;
+ let default_ws_addr = resolve_optional_listener_addr(
+ config.websocket.enabled,
+ "websocket.address",
+ &config.websocket.address,
+ )?;
+ let default_quic_addr =
+ resolve_optional_listener_addr(config.quic.enabled, "quic.address", &config.quic.address)?;
+ if !config.cluster.enabled {
+ return Ok(TcpTopology {
Review Comment:
when `cluster.enabled = false`, `current_replica_id` is silently ignored - `self_replica_id` is hardcoded to `SHARD_REPLICA_ID = 0`. an operator passing `--replica-id 5` accidentally on a single-node deployment gets zero signal. either warn-log when cluster is disabled but the flag was passed, or reject.
##########
core/configs/src/server_ng_config/server_ng.rs:
##########
@@ -60,6 +63,8 @@ const DEFAULT_CONFIG_PATH: &str = "core/server-ng/config.toml";
pub struct ServerNgConfig {
pub consumer_group: ConsumerGroupConfig,
pub data_maintenance: DataMaintenanceConfig,
+ #[serde(default)]
+ pub extra: ExtraConfig,
Review Comment:
`[extra]` TOML section name is unprecedented in this codebase. zero matches in legacy `config.toml`, no `pub extra` field anywhere else. the only field inside is `namespace`, so `ExtraConfig` is a single-field wrapper. operator-facing schema decision worth getting right before this lands. rename to `[system.namespace]` (sibling of `system.sharding`, which already lives there) or top-level `[namespace]`; drop the `ExtraConfig` wrapper.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -0,0 +1,1410 @@
+/*
+ * 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.
+ */
+
+use crate::config_writer::write_current_config;
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+use iggy_binary_protocol::RequestHeader;
+use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use iggy_common::{
+ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, PartitionStats, TopicStats,
+ sharding::LocalIdx, variadic,
+};
+use journal::Journal;
+use journal::prepare_journal::PrepareJournal;
+use message_bus::client_listener::{self, RequestHandler};
+use message_bus::fd_transfer;
+use message_bus::installer;
+use message_bus::installer::ConnectionInstaller;
+use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
+use message_bus::replica::io as replica_io;
+use message_bus::replica::listener::{self as replica_listener, MessageHandler};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem, self_signed_for_loopback,
+};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
+ AcceptedWsClientFn, IggyMessageBus, connector,
+};
+use metadata::IggyMetadata;
+use metadata::MuxStateMachine;
+use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
+use metadata::impls::recovery::recover;
+use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::snapshot::Snapshot;
+use metadata::stm::stream::{Partition, Streams};
+use metadata::stm::user::Users;
+use partitions::{
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+};
+// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
+use server::bootstrap::create_directories;
+use server::log::logger::Logging;
+use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
+use server::streaming::segments::storage::create_segment_storage;
+use shard::builder::IggyShardBuilder;
+use shard::shards_table::PapayaShardsTable;
+use shard::{
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+};
+use std::cell::{Cell, RefCell};
+use std::future::Future;
+use std::net::{IpAddr, SocketAddr};
+use std::path::{Path, PathBuf};
+use std::rc::{Rc, Weak};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tracing::{info, warn};
+
+const CLUSTER_ID: u128 = 1;
+const SHARD_ID: u16 = 0;
+const SHARD_REPLICA_ID: u8 = 0;
+const SHARD_NAME: &str = "server-ng-shard-0";
+const SHARD_INBOX_CAPACITY: usize = 1024;
+
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMetadata = IggyMetadata<
+ VsrConsensus<Rc<IggyMessageBus>>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+>;
+type ServerNgShard = IggyShard<
+ Rc<IggyMessageBus>,
+ PrepareJournal,
+ IggySnapshot,
+ ServerNgMuxStateMachine,
+ PapayaShardsTable,
+>;
+
+type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+
+struct TcpTopology {
+ self_replica_id: u8,
+ replica_count: u8,
+ client_listen_addr: SocketAddr,
+ replica_listen_addr: Option<SocketAddr>,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ tcp_tls_listen_addr: Option<SocketAddr>,
+ peers: Vec<(u8, SocketAddr)>,
+}
+
+struct LocalClientAcceptFns {
+ tcp: AcceptedClientFn,
+ ws: AcceptedWsClientFn,
+ quic: AcceptedQuicClientFn,
+ tcp_tls: AcceptedTlsClientFn,
+}
+
+#[derive(Default)]
+struct BoundClientListeners {
+ tcp: Option<SocketAddr>,
+ tcp_tls: Option<SocketAddr>,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+}
+
+pub trait RunServerNg {
+ fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> impl Future<Output = Result<(), ServerNgError>>;
+}
+
+impl RunServerNg for Rc<ServerNgShard> {
+ /// Run the fully bootstrapped `server-ng` shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if TCP listener bootstrap fails or cluster TCP
+ /// addresses cannot be resolved from config.
+ async fn run(
+ &self,
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+ ) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let (stop_tx, stop_rx) = channel(1);
+ let message_pump_shard = Self::clone(self);
+ let message_pump_handle = compio::runtime::spawn(async move {
+ message_pump_shard.run_message_pump(stop_rx).await;
+ });
+ self.bus.track_background(message_pump_handle);
+
+ let on_replica_message = make_replica_message_handler(self);
+ let on_client_request = make_client_request_handler(self);
+ let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
+ let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
+
+ info!(
+ shard = self.id,
+ partitions = self.plane.partitions().len(),
+ "server-ng shard initialized"
+ );
+
+ if let Err(error) =
+ start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
+
+ self.bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+ Ok(())
+ }
+}
+
+/// Load config, prepare directories, and complete late logging init.
+///
+/// # Errors
+///
+/// Returns an error if config loading, directory preparation, or logging
+/// setup fails.
+pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, ServerNgError> {
+ let config = ServerNgConfig::load()
+ .await
+ .map_err(ServerNgError::Config)?;
+ // TODO: decouple directory bootstrap from the `server` crate.
+ create_directories(&config.system)
+ .await
+ .map_err(ServerNgError::CreateDirectories)?;
+ logging
+ .late_init(
+ config.system.get_system_path(),
+ &config.system.logging,
+ &config.telemetry,
+ )
+ .map_err(ServerNgError::Logging)?;
+
+ Ok(config)
+}
+
+/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+///
+/// # Errors
+///
+/// Returns an error if metadata recovery, consensus restoration, or
+/// partition hydration fails.
+pub async fn bootstrap(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let topology = resolve_tcp_topology(config, current_replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
+ let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
+ recovered
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number)
+ });
+
+ let metadata = ServerNgMetadata::new(
+ Some(restore_metadata_consensus(
+ &recovered.journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )),
+ Some(recovered.journal),
+ recovered.snapshot,
+ recovered.mux_stm,
+ Some(PathBuf::from(&config.system.path)),
+ );
+ let shard = build_single_shard(config, &topology, metadata, bus).await?;
+ info!(shard = shard.id, "server-ng bootstrap complete");
+
+ Ok(shard)
+}
+
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ 0,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.set_log_view(header.view);
+ }
+
+ consensus
+}
+
+async fn build_single_shard(
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ metadata: ServerNgMetadata,
+ bus: Rc<IggyMessageBus>,
+) -> Result<Rc<ServerNgShard>, ServerNgError> {
+ let shard_id = ShardId::new(SHARD_ID);
+ let partition_count = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .map(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .map(|(_, topic)| topic.partitions.len())
+ .sum::<usize>()
+ })
+ .sum()
+ });
+ let mut partitions = IggyPartitions::with_capacity(
+ shard_id,
+ PartitionsConfig {
+ messages_required_to_save: config.system.partition.messages_required_to_save,
+ size_of_messages_required_to_save: config
+ .system
+ .partition
+ .size_of_messages_required_to_save,
+ enforce_fsync: config.system.partition.enforce_fsync,
+ segment_size: config.system.segment.size,
+ },
+ partition_count,
+ );
+ let shards_table = PapayaShardsTable::with_capacity(partition_count);
+
+ let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
+ let mut topic_stats = Vec::new();
+ let mut namespaces = Vec::with_capacity(partition_count);
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ topic_stats.push(topic.stats.clone());
+ for partition in &topic.partitions {
+ namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ }
+ }
+ }
+ (topic_stats, namespaces)
+ });
+
+ for topic_stats in topic_stats {
+ topic_stats.zero_out_all();
+ }
+
+ for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
+ let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
+ let partition = load_partition(
+ config,
+ namespace,
+ topic_stats,
+ &partition_metadata,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?;
+ let local_idx = partitions.insert(namespace, partition);
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ );
+ }
+
+ let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
+ let senders = vec![sender];
+ let shard_handle = Rc::new(RefCell::new(None));
+ let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
+ let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let built = IggyShardBuilder::new(
+ ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ Rc::clone(&bus),
+ on_replica_message,
+ on_client_request,
+ metadata,
+ partitions,
+ senders,
+ inbox,
+ shards_table,
+ PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
+ CoordinatorConfig::default(),
+ bus.token(),
+ )
+ .build();
+ if let Some(refresh_task) = built.refresh_task {
+ bus.track_background(refresh_task);
+ }
+
+ let shard = Rc::new(built.shard);
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ Ok(shard)
+}
+
+const fn validate_recovered_namespace(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ let namespace = &config.extra.namespace;
+ if stream_id < namespace.max_streams
+ && topic_id < namespace.max_topics
+ && partition_id < namespace.max_partitions
+ {
+ return Ok(());
+ }
+
+ Err(ServerNgError::RecoveredNamespaceOutOfBounds {
+ stream_id,
+ topic_id,
+ partition_id,
+ max_streams: namespace.max_streams,
+ max_topics: namespace.max_topics,
+ max_partitions: namespace.max_partitions,
+ })
+}
+
+async fn load_partition(
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ topic_stats: Arc<TopicStats>,
+ partition_metadata: &Partition,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerNgError> {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let stats = Arc::new(PartitionStats::new(topic_stats));
+ let consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ namespace.inner(),
+ bus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+
+ // TODO: decouple the loading logic from the `server` crate and load directly
+ // into the new `partitions` log/runtime types.
+ let loaded_log = server::bootstrap::load_segments(
+ &config.system,
+ stream_id,
+ topic_id,
+ partition_id,
+ config
+ .system
+ .get_partition_path(stream_id, topic_id, partition_id),
+ stats.clone(),
+ )
+ .await
+ .map_err(|source| ServerNgError::PartitionLogLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+
+ let mut partition = IggyPartition::new(stats.clone(), consensus);
+ hydrate_partition_log(
+ &mut partition,
+ config,
+ stream_id,
+ topic_id,
+ partition_id,
+ loaded_log,
+ )
+ .await?;
+
+ let current_offset = partition
+ .log
+ .segments()
+ .iter()
+ .filter(|segment| segment.size > IggyByteSize::default())
+ .map(|segment| segment.end_offset)
+ .max()
+ .unwrap_or(0);
+ partition.created_at = partition_metadata.created_at;
+ partition.offset.store(current_offset, Ordering::Release);
+ partition
+ .dirty_offset
+ .store(current_offset, Ordering::Relaxed);
+ partition.should_increment_offset = partition
+ .log
+ .segments()
+ .iter()
+ .any(|segment| segment.size > IggyByteSize::default());
+ partition.stats.set_current_offset(current_offset);
+
+ configure_consumer_offsets(&mut partition, config, namespace, current_offset);
+ ensure_initial_segment(&mut partition, config, stream_id, topic_id, partition_id).await?;
+
+ Ok(partition)
+}
+
+async fn hydrate_partition_log(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ loaded_log: server::streaming::partitions::log::SegmentedLog<
+ server::streaming::partitions::journal::MemoryMessageJournal,
+ >,
+) -> Result<(), ServerNgError> {
+ // TODO: decouple the loading logic from the `server` crate. This currently
+ // adapts the old server segmented log into the new `partitions` log.
+ for (segment_index, (segment, storage)) in loaded_log
+ .segments()
+ .iter()
+ .zip(loaded_log.storages().iter().cloned())
+ .enumerate()
+ {
+ let max_timestamp = match loaded_log
+ .indexes()
+ .get(segment_index)
+ .and_then(|indexes| indexes.as_ref())
+ {
+ Some(indexes) => indexes_max_timestamp(indexes),
+ None => load_segment_max_timestamp(&storage, stream_id, topic_id, partition_id).await?,
+ };
+ partition.log.add_persisted_segment(
+ convert_segment(segment, max_timestamp),
+ storage,
+ None,
+ None,
+ );
+ }
+
+ if let Some(active_index) = partition.log.segments().len().checked_sub(1) {
+ let storage = &partition.log.storages()[active_index];
+ if let (Some(messages_reader), Some(index_reader)) = (
+ storage.messages_reader.as_ref(),
+ storage.index_reader.as_ref(),
+ ) {
+ let index_path = index_reader.path();
+ let index_size = std::fs::metadata(&index_path).map_or(0, |metadata| metadata.len());
+ partition.log.messages_writers_mut()[active_index] = Some(Rc::new(
+ MessagesWriter::new(
+ &messages_reader.path(),
+ Rc::new(AtomicU64::new(u64::from(messages_reader.file_size()))),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::MessagesWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ partition.log.index_writers_mut()[active_index] = Some(Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(AtomicU64::new(index_size)),
+ config.system.partition.enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| ServerNgError::IndexWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ ));
+ }
+ }
+
+ Ok(())
+}
+
+fn convert_segment(segment: &iggy_common::Segment, max_timestamp: u64) -> Segment {
+ Segment {
+ sealed: segment.sealed,
+ start_timestamp: segment.start_timestamp,
+ end_timestamp: segment.end_timestamp,
+ max_timestamp,
+ current_position: u64::from(segment.current_position),
+ start_offset: segment.start_offset,
+ end_offset: segment.end_offset,
+ size: segment.size,
+ max_size: segment.max_size,
+ }
+}
+
+fn indexes_max_timestamp(indexes: &server::streaming::segments::IggyIndexesMut) -> u64 {
+ let mut max_timestamp = 0;
+ for index in 0..indexes.count() {
+ if let Some(index_view) = indexes.get(index) {
+ max_timestamp = max_timestamp.max(index_view.timestamp());
+ }
+ }
+
+ max_timestamp
+}
+
+async fn load_segment_max_timestamp(
+ storage: &iggy_common::SegmentStorage,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<u64, ServerNgError> {
+ let Some(index_reader) = storage.index_reader.as_ref() else {
+ return Ok(0);
+ };
+
+ let indexes = index_reader
+ .load_all_indexes_from_disk()
+ .await
+ .map_err(|source| ServerNgError::SegmentIndexesLoad {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+ Ok(indexes_max_timestamp(&indexes))
+}
+
+fn configure_consumer_offsets(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ namespace: IggyNamespace,
+ current_offset: u64,
+) {
+ let stream_id = namespace.stream_id();
+ let topic_id = namespace.topic_id();
+ let partition_id = namespace.partition_id();
+ let consumer_offsets_path =
+ config
+ .system
+ .get_consumer_offsets_path(stream_id, topic_id, partition_id);
+ let consumer_group_offsets_path =
+ config
+ .system
+ .get_consumer_group_offsets_path(stream_id, topic_id, partition_id);
+
+ // TODO: decouple consumer offset loading from the `server` crate.
+ let loaded_consumer_offsets = load_consumer_offsets(&consumer_offsets_path).unwrap_or_default();
+ let consumer_offsets = ConsumerOffsets::with_capacity(loaded_consumer_offsets.len());
+ {
+ let guard = consumer_offsets.pin();
+ for offset in loaded_consumer_offsets {
+ if offset.offset.load(Ordering::Relaxed) > current_offset {
+ offset.offset.store(current_offset, Ordering::Relaxed);
+ }
+ guard.insert(offset.consumer_id as usize, offset);
+ }
+ }
+
+ // TODO: decouple consumer group offset loading from the `server` crate.
+ let loaded_group_offsets =
+ load_consumer_group_offsets(&consumer_group_offsets_path).unwrap_or_default();
+ let consumer_group_offsets = ConsumerGroupOffsets::with_capacity(loaded_group_offsets.len());
+ {
+ let guard = consumer_group_offsets.pin();
+ for (group_id, offset) in loaded_group_offsets {
+ if offset.offset.load(Ordering::Relaxed) > current_offset {
+ offset.offset.store(current_offset, Ordering::Relaxed);
+ }
+ guard.insert(group_id, offset);
+ }
+ }
+
+ partition.configure_consumer_offset_storage(
+ consumer_offsets_path,
+ consumer_group_offsets_path,
+ consumer_offsets,
+ consumer_group_offsets,
+ config.system.partition.enforce_fsync,
+ );
+}
+
+async fn ensure_initial_segment(
+ partition: &mut IggyPartition<Rc<IggyMessageBus>>,
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<(), ServerNgError> {
+ if partition.log.has_segments() {
+ return Ok(());
+ }
+
+ // TODO: decouple segment storage creation from the `server` crate.
+ let storage =
+ create_segment_storage(&config.system, stream_id, topic_id, partition_id, 0, 0, 0)
+ .await
+ .map_err(|source| ServerNgError::InitialSegmentStorage {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?;
+ let messages_path = config
+ .system
+ .get_messages_file_path(stream_id, topic_id, partition_id, 0);
+ let index_path = config
+ .system
+ .get_index_path(stream_id, topic_id, partition_id, 0);
+ partition.log.add_persisted_segment(
+ Segment::new(0, config.system.segment.size),
+ storage,
+ Some(Rc::new(
+ MessagesWriter::new(
+ &messages_path,
+ Rc::new(AtomicU64::new(0)),
+ config.system.partition.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|source| ServerNgError::MessagesWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ )),
+ Some(Rc::new(
+ IggyIndexWriter::new(
+ &index_path,
+ Rc::new(AtomicU64::new(0)),
+ config.system.partition.enforce_fsync,
+ false,
+ )
+ .await
+ .map_err(|source| ServerNgError::IndexWriterInit {
+ stream_id,
+ topic_id,
+ partition_id,
+ source,
+ })?,
+ )),
+ );
+ partition.stats.increment_segments_count(1);
+
+ Ok(())
+}
+
+fn resolve_tcp_topology(
+ config: &ServerNgConfig,
+ current_replica_id: Option<u8>,
+) -> Result<TcpTopology, ServerNgError> {
+ let default_client_addr = parse_socket_addr("tcp.address", &config.tcp.address)?;
+ let default_ws_addr = resolve_optional_listener_addr(
+ config.websocket.enabled,
+ "websocket.address",
+ &config.websocket.address,
+ )?;
+ let default_quic_addr =
+ resolve_optional_listener_addr(config.quic.enabled, "quic.address", &config.quic.address)?;
+ if !config.cluster.enabled {
+ return Ok(TcpTopology {
+ // Keep parity with the current server binary and the integration
+ // harness: `--replica-id` may be passed unconditionally, but in
+ // single-node mode there is only replica 0.
+ self_replica_id: SHARD_REPLICA_ID,
+ replica_count: 1,
+ client_listen_addr: default_client_addr,
+ replica_listen_addr: Some(SocketAddr::new(default_client_addr.ip(), 0)),
+ ws_listen_addr: default_ws_addr,
+ quic_listen_addr: default_quic_addr,
+ tcp_tls_listen_addr: config.tcp.tls.enabled.then_some(default_client_addr),
+ peers: Vec::new(),
+ });
+ }
+
+ let self_replica_id = current_replica_id.ok_or(ServerNgError::MissingReplicaId)?;
+
+ let self_node = config
+ .cluster
+ .nodes
+ .iter()
+ .find(|node| node.replica_id == self_replica_id)
+ .ok_or(ServerNgError::ClusterNodeNotFound {
+ replica_id: self_replica_id,
+ })?;
+ let replica_count = u8::try_from(config.cluster.nodes.len()).map_err(|_| {
+ ServerNgError::ClusterReplicaCountTooLarge {
+ count: config.cluster.nodes.len(),
+ }
+ })?;
+ let (client_listen_addr, ws_listen_addr, quic_listen_addr) = resolve_cluster_client_addrs(
+ self_node,
+ default_client_addr,
+ default_ws_addr,
+ default_quic_addr,
+ )?;
+ let replica_port =
+ self_node
+ .ports
+ .tcp_replica
+ .ok_or(ServerNgError::ClusterReplicaPortMissing {
+ replica_id: self_node.replica_id,
+ })?;
+ let replica_listen_addr = Some(socket_addr_from_parts(
+ "cluster.nodes[*].ports.tcp_replica",
+ &self_node.ip,
+ replica_port,
+ )?);
+ let peers = resolve_cluster_replica_peers(&config.cluster.nodes, self_replica_id)?;
+
+ Ok(TcpTopology {
+ self_replica_id,
+ replica_count,
+ client_listen_addr,
+ replica_listen_addr,
+ ws_listen_addr,
+ quic_listen_addr,
+ tcp_tls_listen_addr: config.tcp.tls.enabled.then_some(client_listen_addr),
+ peers,
+ })
+}
+
+fn resolve_optional_listener_addr(
+ enabled: bool,
+ context: &'static str,
+ address: &str,
+) -> Result<Option<SocketAddr>, ServerNgError> {
+ if enabled {
+ return Ok(Some(parse_socket_addr(context, address)?));
+ }
+ Ok(None)
+}
+
+fn resolve_cluster_client_addrs(
+ self_node: &configs::cluster::ClusterNodeConfig,
+ default_client_addr: SocketAddr,
+ default_ws_addr: Option<SocketAddr>,
+ default_quic_addr: Option<SocketAddr>,
+) -> Result<(SocketAddr, Option<SocketAddr>, Option<SocketAddr>), ServerNgError> {
+ let client_port = self_node
+ .ports
+ .tcp
+ .unwrap_or_else(|| default_client_addr.port());
+ let client_listen_addr =
+ socket_addr_from_parts("cluster.nodes[*].ports.tcp", &self_node.ip, client_port)?;
+ let ws_listen_addr = resolve_cluster_optional_addr(
+ self_node,
+ "cluster.nodes[*].ports.websocket",
+ default_ws_addr,
+ |ports| ports.websocket,
+ )?;
+ let quic_listen_addr = resolve_cluster_optional_addr(
+ self_node,
+ "cluster.nodes[*].ports.quic",
+ default_quic_addr,
+ |ports| ports.quic,
+ )?;
+ Ok((client_listen_addr, ws_listen_addr, quic_listen_addr))
+}
+
+fn resolve_cluster_optional_addr(
+ self_node: &configs::cluster::ClusterNodeConfig,
+ context: &'static str,
+ default_addr: Option<SocketAddr>,
+ port_selector: impl Fn(&configs::cluster::TransportPorts) -> Option<u16>,
+) -> Result<Option<SocketAddr>, ServerNgError> {
+ let Some(default_addr) = default_addr else {
+ return Ok(None);
+ };
+ let port = port_selector(&self_node.ports).unwrap_or_else(|| default_addr.port());
+ socket_addr_from_parts(context, &self_node.ip, port).map(Some)
+}
+
+fn resolve_cluster_replica_peers(
+ nodes: &[configs::cluster::ClusterNodeConfig],
+ self_replica_id: u8,
+) -> Result<Vec<(u8, SocketAddr)>, ServerNgError> {
+ let mut peers = Vec::with_capacity(nodes.len().saturating_sub(1));
+ for node in nodes {
+ if node.replica_id == self_replica_id {
+ continue;
+ }
+ let replica_port =
+ node.ports
+ .tcp_replica
+ .ok_or(ServerNgError::ClusterReplicaPortMissing {
+ replica_id: node.replica_id,
+ })?;
+ peers.push((
+ node.replica_id,
+ socket_addr_from_parts("cluster.nodes[*].ports.tcp_replica", &node.ip, replica_port)?,
+ ));
+ }
+ Ok(peers)
+}
+
+async fn start_tcp_runtime(
+ shard: &Rc<ServerNgShard>,
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ accepted_replica: AcceptedReplicaFn,
+ accepted_clients: LocalClientAcceptFns,
+) -> Result<(), ServerNgError> {
+ if config.tcp.enabled && !config.tcp.tls.enabled {
+ return start_via_replica_io(shard, config, topology, accepted_replica, accepted_clients)
+ .await;
+ }
+
+ start_manual_runtime(shard, config, topology, accepted_replica, accepted_clients).await
+}
+
+async fn start_via_replica_io(
+ shard: &Rc<ServerNgShard>,
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ accepted_replica: AcceptedReplicaFn,
+ accepted_clients: LocalClientAcceptFns,
+) -> Result<(), ServerNgError> {
+ let replica_addr = topology
+ .replica_listen_addr
+ .expect("topology must include replica listener address");
+ let quic_credentials = topology
+ .quic_listen_addr
+ .is_some()
+ .then(|| load_quic_server_credentials(config))
+ .transpose()?;
+ let tcp_tls_credentials = topology
+ .tcp_tls_listen_addr
+ .is_some()
+ .then(|| load_tcp_tls_server_credentials(config))
+ .transpose()?;
+
+ let LocalClientAcceptFns {
+ tcp,
+ ws,
+ quic,
+ tcp_tls,
+ } = accepted_clients;
+
+ let bound = replica_io::start_on_shard_zero(
+ &shard.bus,
+ replica_addr,
+ topology.client_listen_addr,
+ topology.ws_listen_addr,
+ topology.quic_listen_addr,
+ quic_credentials,
+ topology.tcp_tls_listen_addr,
+ tcp_tls_credentials,
+ None,
+ None,
+ CLUSTER_ID,
+ topology.self_replica_id,
+ topology.replica_count,
+ topology.peers.clone(),
+ accepted_replica,
+ tcp,
+ topology.ws_listen_addr.map(|_| ws),
+ topology.quic_listen_addr.map(|_| quic),
+ topology.tcp_tls_listen_addr.map(|_| tcp_tls),
+ None,
+ shard.bus.config().reconnect_period,
+ )
+ .await
+ .map_err(ServerNgError::StartListeners)?;
+ let Some(bound) = bound else {
+ return Ok(());
+ };
+
+ write_current_config(
+ config,
+ Some(topology.self_replica_id),
+ Some(bound.client),
+ config.cluster.enabled.then_some(bound.replica),
+ bound.tcp_tls,
+ bound.quic,
+ bound.ws,
+ )
+ .await?;
+ if config.cluster.enabled {
+ info!(
+ shard = shard.id,
+ replica = %bound.replica,
+ tcp = %bound.client,
+ tcp_tls = ?bound.tcp_tls,
+ ws = ?bound.ws,
+ quic = ?bound.quic,
+ "server-ng listeners started"
+ );
+ } else {
+ info!(
+ shard = shard.id,
+ tcp = %bound.client,
+ tcp_tls = ?bound.tcp_tls,
+ ws = ?bound.ws,
+ quic = ?bound.quic,
+ "server-ng client listeners started"
+ );
+ }
+
+ Ok(())
+}
+
+async fn start_manual_runtime(
+ shard: &Rc<ServerNgShard>,
+ config: &ServerNgConfig,
+ topology: &TcpTopology,
+ accepted_replica: AcceptedReplicaFn,
+ accepted_clients: LocalClientAcceptFns,
+) -> Result<(), ServerNgError> {
+ let bound_replica = if config.cluster.enabled {
+ let replica_addr = topology
+ .replica_listen_addr
+ .expect("cluster-enabled topology must include replica listener address");
+ let (replica_listener, bound_addr) = replica_listener::bind(replica_addr)
+ .await
+ .map_err(ServerNgError::StartListeners)?;
+ let token = shard.bus.token();
+ let max_message_size = shard.bus.config().max_message_size;
+ let handshake_grace = shard.bus.config().handshake_grace;
+ let self_replica_id = topology.self_replica_id;
+ let replica_count = topology.replica_count;
+ let accepted_replica_for_listener = accepted_replica.clone();
+ let replica_handle = compio::runtime::spawn(async move {
+ replica_listener::run(
+ replica_listener,
+ token,
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ accepted_replica_for_listener,
+ max_message_size,
+ handshake_grace,
+ )
+ .await;
+ });
+ shard.bus.track_background(replica_handle);
+ connector::start(
+ &shard.bus,
+ CLUSTER_ID,
+ topology.self_replica_id,
+ topology.peers.clone(),
+ accepted_replica,
+ shard.bus.config().reconnect_period,
+ )
+ .await;
+ Some(bound_addr)
+ } else {
+ None
+ };
+
+ let bound_clients = start_client_listeners(shard, config, topology, &accepted_clients).await?;
+ write_current_config(
+ config,
+ Some(topology.self_replica_id),
+ bound_clients.tcp,
+ bound_replica,
+ bound_clients.tcp_tls,
+ bound_clients.quic,
+ bound_clients.ws,
+ )
+ .await?;
+
+ if config.cluster.enabled {
+ info!(
+ shard = shard.id,
+ replica = ?bound_replica,
+ tcp = ?bound_clients.tcp,
+ tcp_tls = ?bound_clients.tcp_tls,
+ ws = ?bound_clients.ws,
+ quic = ?bound_clients.quic,
+ "server-ng listeners started"
+ );
+ } else {
+ info!(
+ shard = shard.id,
+ tcp = ?bound_clients.tcp,
+ tcp_tls = ?bound_clients.tcp_tls,
+ ws = ?bound_clients.ws,
+ quic = ?bound_clients.quic,
+ "server-ng client listeners started"
+ );
+ }
+
+ Ok(())
+}
+
+fn make_replica_message_handler(shard: &Rc<ServerNgShard>) -> MessageHandler {
+ let shard = Rc::clone(shard);
+ Rc::new(move |_replica_id, message| {
+ shard.dispatch(message);
+ })
+}
+
+fn make_client_request_handler(shard: &Rc<ServerNgShard>) -> RequestHandler {
+ let shard = Rc::clone(shard);
+ Rc::new(move |client_id, message| {
+ let request = match message.try_into_typed::<RequestHeader>() {
+ Ok(request) => request,
+ Err(error) => {
+ warn!(client_id, error = %error, "dropping client request with invalid header");
+ return;
+ }
+ };
+ let request = request.transmute_header(|header, new_header: &mut RequestHeader| {
+ *new_header = header;
+ new_header.client = client_id;
+ });
+ shard.dispatch(request.into_generic());
+ })
+}
+
+fn make_deferred_replica_message_handler(shard_handle: &ServerNgShardHandle) -> MessageHandler {
Review Comment:
also `make_deferred_client_request_handler` at line 1103. each call into the deferred handler does `RefCell::borrow()` + `Weak::upgrade()` (Rc clone + drop) per replica/client message on the fd-delegation path. the cycle-break is correct, but the per-message indirection is unnecessary once init is done. swap for `Rc<OnceCell<Rc<ServerNgShard>>>` set once at the end of `build_single_shard`; the closure becomes `cell.get().unwrap()` with no borrow check and no Weak upgrade.
##########
core/server-ng/src/server_error.rs:
##########
@@ -0,0 +1,127 @@
+/*
+ * 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.
+ */
+
+use metadata::impls::recovery::RecoveryError;
+// TODO: decouple logging errors from the `server` crate.
+use server::server_error::LogError;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum ServerNgError {
+ #[error("failed to load server-ng config")]
+ Config(#[source] configs::ConfigurationError),
+ #[error("failed to prepare server-ng directories")]
+ CreateDirectories(#[source] iggy_common::IggyError),
+ #[error("failed to serialize current server-ng config")]
+ CurrentConfigSerialize(#[source] toml::ser::Error),
+ #[error("failed to write current server-ng config at {path}")]
+ CurrentConfigWrite {
+ path: String,
+ #[source]
+ source: std::io::Error,
+ },
+ #[error("failed to initialize server-ng logging")]
+ Logging(#[source] LogError),
+ #[error("failed to recover metadata snapshot and journal")]
+ MetadataRecovery(#[source] RecoveryError),
+ #[error("failed to parse {context} socket address '{address}'")]
+ SocketAddressParse {
+ context: &'static str,
+ address: String,
+ #[source]
+ source: std::net::AddrParseError,
+ },
+ #[error("cluster enabled but no node is configured for replica {replica_id}")]
+ ClusterNodeNotFound { replica_id: u8 },
+ #[error("cluster node count {count} exceeds supported u8 replica count")]
+ ClusterReplicaCountTooLarge { count: usize },
+ #[error("cluster mode requires --replica-id to identify the current node")]
+ MissingReplicaId,
+ #[error("cluster node for replica {replica_id} is missing tcp_replica port")]
+ ClusterReplicaPortMissing { replica_id: u8 },
+ #[error(
+ "recovered namespace stream {stream_id}, topic {topic_id}, partition {partition_id} exceeds configured limits (max_streams={max_streams}, max_topics={max_topics}, max_partitions={max_partitions})"
+ )]
+ RecoveredNamespaceOutOfBounds {
Review Comment:
five variants - `PartitionLogLoad` (61-68), `MessagesWriterInit` (87-96), `IndexWriterInit` (97-106), `SegmentIndexesLoad` (107-116), `InitialSegmentStorage` (117-126) - all carry the same `{stream_id, topic_id, partition_id, source}` triplet. factor a `PartitionScope { stream_id, topic_id, partition_id }` field with a single source per variant.
##########
core/consensus/src/impls.rs:
##########
@@ -662,6 +662,33 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
assert!(self.commit_max.get() >= self.commit_min.get());
}
+ /// Restore local commit progress from already-applied state during bootstrap.
+ ///
+ /// Unlike `advance_commit_min`, this is intended for recovery paths where the
+ /// state machine has already been restored up to the supplied commit point.
+ ///
+ /// # Panics
+ /// - If `commit_min > commit_max`.
+ /// - If commit progress has already been initialized on this consensus instance.
+ pub fn restore_commit_state(&self, commit_min: u64, commit_max: u64) {
+ assert!(
+ commit_min <= commit_max,
+ "commit_min ({commit_min}) must be <= commit_max ({commit_max})"
+ );
+ assert_eq!(
Review Comment:
`restore_commit_state` asserts `commit_min == 0 && commit_max == 0` on entry. today fine, but if `VsrConsensus::new` ever changes its initial commit defaults (e.g., for recovery hooks), the asserts break silently in tests but mask corruption in prod. replace with a Builder pattern, or a `restored: Cell<bool>` guard set on first restore so the contract is enforced by construction.
##########
core/configs/src/server_config/server.rs:
##########
@@ -183,9 +183,21 @@ impl ServerConfig {
///
/// Uses compile-time generated env var mappings for unambiguous resolution.
pub async fn load() -> Result<ServerConfig, ConfigurationError> {
+ Self::load_with_path(
+ DEFAULT_CONFIG_PATH,
+ include_str!("../../../server/config.toml"),
+ )
+ .await
+ }
+
+ pub async fn load_with_path(
Review Comment:
also `config_provider_with_default` at line 222. new public API with no external callers. `ServerNgConfig::load` reimplements path/include logic at `core/configs/src/server_ng_config/server_ng.rs:116-131` instead of delegating to these helpers. all 5 grep hits are internal to this file. either route `ServerNgConfig::load` through the new helpers, or revert these methods - right now they pay public-API tax for zero callers.
##########
core/metadata/src/impls/recovery.rs:
##########
@@ -104,28 +104,36 @@ pub struct RecoveredMetadata<M> {
pub async fn recover<M>(data_dir: &Path) -> Result<RecoveredMetadata<M>, RecoveryError>
where
M: StateMachine<Input = Message<PrepareHeader>, Error = IggyError>
- + RestoreSnapshot<MetadataSnapshot>,
+ + RestoreSnapshot<MetadataSnapshot>
+ + Default,
{
let metadata_dir = data_dir.join(super::METADATA_DIR);
std::fs::create_dir_all(&metadata_dir)?;
- // 1. Load snapshot (or empty default if missing)
+ // 1. Load snapshot if present.
let snapshot_path = metadata_dir.join("snapshot.bin");
- let (snapshot, replay_from) = if snapshot_path.exists() {
- let s = IggySnapshot::load(&snapshot_path)?;
- let from = s.sequence_number() + 1;
- (s, from)
+ let snapshot = if snapshot_path.exists() {
+ Some(IggySnapshot::load(&snapshot_path)?)
} else {
- // No snapshot, replay from op 0.
- (IggySnapshot::new(0), 0)
+ None
+ };
+ let replay_from = snapshot
+ .as_ref()
+ .map_or(0, |snapshot| snapshot.sequence_number() + 1);
+
+ // 2. Restore state machine from snapshot, or start from a fresh empty state.
+ let mux_stm = match snapshot.as_ref() {
+ Some(snapshot) => M::restore_snapshot(snapshot.snapshot())?,
+ None => M::default(),
};
-
- // 2. Restore state machine from snapshot
- let mux_stm = M::restore_snapshot(snapshot.snapshot())?;
// 3. Open journal, scan the WAL and build index
let journal_path = metadata_dir.join("journal.wal");
- let journal = PrepareJournal::open(&journal_path, snapshot.sequence_number()).await?;
+ let journal = PrepareJournal::open(
+ &journal_path,
+ snapshot.as_ref().map_or(0, IggySnapshot::sequence_number),
+ )
+ .await?;
// 4. Replay journal entries after snapshot
let headers_to_replay = journal.iter_headers_from(replay_from);
Review Comment:
replay loop `mux_stm.update(entry)?` aborts on first error and `recover()` returns `Err`, bootstrap fails. the journal still contains those entries, so an operator restart hits the same error - node is permanently un-bootable until the journal is manually truncated. there's no skip-and-repair-via-VSR path and no contiguity check on the WAL. compounds with `PrepareJournal` `SLOT_COUNT = 1024` silent eviction. confirm intentional fail-fast for now and document the repair path; longer-term, walk the hash chain and stop at the first break.
##########
core/common/src/sharding/namespace.rs:
##########
@@ -89,4 +130,81 @@ impl IggyNamespace {
| ((partition as u64) & PARTITION_MASK) << PARTITION_SHIFT;
Self(value)
}
+
+ pub fn validate_capacity(
+ max_streams: usize,
+ max_topics: usize,
+ max_partitions: usize,
+ ) -> Result<(), NamespaceCapacityError> {
+ let stream_bits = if max_streams == 0 {
+ return Err(NamespaceCapacityError::ZeroStreams);
+ } else {
+ bits_required((max_streams - 1) as u64)
+ };
+ let topic_bits = if max_topics == 0 {
+ return Err(NamespaceCapacityError::ZeroTopics);
+ } else {
+ bits_required((max_topics - 1) as u64)
+ };
+ let partition_bits = if max_partitions == 0 {
+ return Err(NamespaceCapacityError::ZeroPartitions);
+ } else {
+ bits_required((max_partitions - 1) as u64)
+ };
+
+ let required_bits = stream_bits
Review Comment:
`ExceedsU64 { required_bits: u32::MAX }` placeholder is unreachable. `bits_required` returns `u32 ≤ 64` (loop bounded by `u64::BITS` at lines 28-38); `checked_add` of three values each ≤ 64 has max sum 192, cannot overflow `u32`. drop the `checked_add` branch and sum directly - the `required_bits > u64::BITS` check below is the real guard.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org