You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/10/13 19:11:04 UTC
(iggy) branch improve_errors created (now 3f8dc8682)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch improve_errors
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 3f8dc8682 feat(io_uring): improve error handling
This branch includes the following new commits:
new 3f8dc8682 feat(io_uring): improve error handling
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(io_uring): improve error handling
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch improve_errors
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 3f8dc868247db2a5440ca7f13cd28bbe0f883a6f
Author: numminex <g....@gmail.com>
AuthorDate: Mon Oct 13 21:10:51 2025 +0200
feat(io_uring): improve error handling
---
core/server/src/shard/system/consumer_groups.rs | 1 +
core/server/src/shard/system/messages.rs | 17 ++--
core/server/src/shard/system/topics.rs | 1 +
core/server/src/shard/system/utils.rs | 35 +++++++-
core/server/src/slab/traits_ext.rs | 113 +-----------------------
5 files changed, 45 insertions(+), 122 deletions(-)
diff --git a/core/server/src/shard/system/consumer_groups.rs b/core/server/src/shard/system/consumer_groups.rs
index 66b16bb97..6fac6c8fa 100644
--- a/core/server/src/shard/system/consumer_groups.rs
+++ b/core/server/src/shard/system/consumer_groups.rs
@@ -239,6 +239,7 @@ impl IggyShard {
group_id: &Identifier,
) -> Result<(), IggyError> {
self.ensure_authenticated(session)?;
+ self.ensure_consumer_group_exists(stream_id, topic_id, group_id)?;
{
let topic_id = self.streams2.with_topic_by_id(
stream_id,
diff --git a/core/server/src/shard/system/messages.rs b/core/server/src/shard/system/messages.rs
index 0ce7fed78..94e1aaaca 100644
--- a/core/server/src/shard/system/messages.rs
+++ b/core/server/src/shard/system/messages.rs
@@ -47,6 +47,8 @@ impl IggyShard {
partitioning: &Partitioning,
batch: IggyMessagesBatchMut,
) -> Result<(), IggyError> {
+ self.ensure_topic_exists(&stream_id, &topic_id)?;
+
let numeric_stream_id = self
.streams2
.with_stream_by_id(&stream_id, streams::helpers::get_stream_id());
@@ -99,6 +101,7 @@ impl IggyShard {
},
)?;
+ self.ensure_partition_exists(&stream_id, &topic_id, partition_id)?;
let namespace = IggyNamespace::new(numeric_stream_id, numeric_topic_id, partition_id);
let payload = ShardRequestPayload::SendMessages { batch };
let request = ShardRequest::new(stream_id.clone(), topic_id.clone(), partition_id, payload);
@@ -160,6 +163,8 @@ impl IggyShard {
maybe_partition_id: Option<u32>,
args: PollingArgs,
) -> Result<(IggyPollMetadata, IggyMessagesBatchSet), IggyError> {
+ self.ensure_topic_exists(&stream_id, &topic_id)?;
+
let numeric_stream_id = self
.streams2
.with_stream_by_id(&stream_id, streams::helpers::get_stream_id());
@@ -189,17 +194,7 @@ impl IggyShard {
return Ok((IggyPollMetadata::new(0, 0), IggyMessagesBatchSet::empty()));
};
- let has_partition = self
- .streams2
- .with_topic_by_id(&stream_id, &topic_id, |(root, ..)| {
- root.partitions().exists(partition_id)
- });
- if !has_partition {
- return Err(IggyError::NoPartitions(
- numeric_topic_id as u32,
- numeric_stream_id as u32,
- ));
- }
+ self.ensure_partition_exists(&stream_id, &topic_id, partition_id)?;
let current_offset = self.streams2.with_partition_by_id(
&stream_id,
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index 3cc90c941..32490badd 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -279,6 +279,7 @@ impl IggyShard {
topic_id: &Identifier,
) -> Result<(), IggyError> {
self.ensure_authenticated(session)?;
+ self.ensure_topic_exists(stream_id, topic_id)?;
{
let topic_id = self.streams2.with_topic_by_id(
stream_id,
diff --git a/core/server/src/shard/system/utils.rs b/core/server/src/shard/system/utils.rs
index a0a753137..4de0e7747 100644
--- a/core/server/src/shard/system/utils.rs
+++ b/core/server/src/shard/system/utils.rs
@@ -3,8 +3,7 @@ use iggy_common::{Consumer, ConsumerKind, Identifier, IggyError};
use crate::{
shard::IggyShard,
streaming::{
- polling_consumer::PollingConsumer,
- topics::{self},
+ polling_consumer::PollingConsumer, streams::helpers::get_stream_id, topics::{self}
},
};
@@ -68,6 +67,38 @@ impl IggyShard {
Ok(())
}
+ pub fn ensure_partition_exists(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partition_id: usize,
+ ) -> Result<(), IggyError> {
+ self.ensure_topic_exists(stream_id, topic_id)?;
+ let partition_exists = self.streams2.with_topic_by_id(
+ stream_id,
+ topic_id,
+ |(root, ..)| root.partitions().exists(partition_id),
+ );
+
+ if !partition_exists {
+ let numeric_stream_id = self
+ .streams2
+ .with_stream_by_id(stream_id, get_stream_id());
+ let numeric_topic_id = self.streams2.with_topic_by_id(
+ stream_id,
+ topic_id,
+ topics::helpers::get_topic_id(),
+ );
+ return Err(IggyError::PartitionNotFound(
+ partition_id as u32,
+ numeric_topic_id as u32,
+ numeric_stream_id as u32,
+ ));
+ }
+
+ Ok(())
+ }
+
pub fn resolve_consumer_with_partition_id(
&self,
stream_id: &Identifier,
diff --git a/core/server/src/slab/traits_ext.rs b/core/server/src/slab/traits_ext.rs
index d69dfd85b..6a23d30e9 100644
--- a/core/server/src/slab/traits_ext.rs
+++ b/core/server/src/slab/traits_ext.rs
@@ -53,102 +53,6 @@ mod private {
pub trait Sealed {}
}
-//TODO: Maybe two seperate traits for Ref and RefMut.
-pub trait ComponentsMapping<T>: private::Sealed {
- type Ref<'a>;
- type RefMut<'a>;
-}
-
-pub trait ComponentsByIdMapping<T>: private::Sealed {
- type Ref<'a>;
- type RefMut<'a>;
-}
-
-// TODO: This should be a proc macro.
-macro_rules! impl_components_mapping_for_slab {
- ($T:ident) => {
- impl<$T> private::Sealed for ($T,) {}
-
- impl<$T> ComponentsMapping<Borrow> for ($T,)
- where for<'a> $T :'a
- {
- type Ref<'a> = (&'a ::slab::Slab<$T>,);
- type RefMut<'a> = (&'a mut ::slab::Slab<$T>,);
- }
-
- impl<$T> ComponentsMapping<InteriorMutability> for ($T,)
- where for<'a> $T :'a
- {
- type Ref<'a> = (::std::cell::Ref<'a, ::slab::Slab<$T>>,);
- type RefMut<'a> = (::std::cell::RefMut<'a, ::slab::Slab<$T>>,);
- }
-
- impl<$T> ComponentsByIdMapping<Borrow> for ($T,)
- where for<'a> $T :'a
- {
- type Ref<'a> = (&'a $T,);
- type RefMut<'a> = (&'a mut $T,);
- }
-
- impl<$T> ComponentsByIdMapping<InteriorMutability> for ($T,)
- where for<'a> $T :'a
- {
- type Ref<'a> = (::std::cell::Ref<'a, $T>,);
- type RefMut<'a> = (::std::cell::RefMut<'a, $T>,);
- }
- };
-
- ($T:ident, $($rest:ident),+) => {
- impl<$T, $($rest),+> private::Sealed for ($T, $($rest),+) {}
-
- impl<$T, $($rest),+> ComponentsMapping<Borrow> for ($T, $($rest),+)
- where
- for<'a> $T :'a,
- $(for<'a> $rest: 'a),+
- {
- type Ref<'a> = (&'a ::slab::Slab<$T>, $(&'a ::slab::Slab<$rest>),+);
- type RefMut<'a> = (&'a mut ::slab::Slab<$T>, $(&'a mut ::slab::Slab<$rest>),+);
- }
-
- impl<$T, $($rest),+> ComponentsMapping<InteriorMutability> for ($T, $($rest),+)
- where
- for<'a> $T :'a,
- $(for<'a> $rest: 'a),+
- {
- type Ref<'a> = (std::cell::Ref<'a, ::slab::Slab<$T>>, $(::std::cell::Ref<'a, ::slab::Slab<$rest>>),+);
- type RefMut<'a> = (std::cell::RefMut<'a, ::slab::Slab<$T>>, $(::std::cell::RefMut<'a, ::slab::Slab<$rest>>),+);
- }
-
- impl<$T, $($rest),+> ComponentsByIdMapping<Borrow> for ($T, $($rest),+)
- where
- for<'a> $T :'a,
- $(for<'a> $rest: 'a),+
- {
- type Ref<'a> = (&'a $T, $(&'a $rest),+);
- type RefMut<'a> = (&'a mut $T, $(&'a mut $rest),+);
- }
-
- impl<$T, $($rest),+> ComponentsByIdMapping<InteriorMutability> for ($T, $($rest),+)
- where
- for<'a> $T :'a,
- $(for<'a> $rest: 'a),+
- {
- type Ref<'a> = (std::cell::Ref<'a, $T>, $(::std::cell::Ref<'a, $rest>),+);
- type RefMut<'a> = (std::cell::RefMut<'a, $T>, $(::std::cell::RefMut<'a, $rest>),+);
- }
- impl_components_mapping_for_slab!($($rest),+);
- };
-}
-impl_components_mapping_for_slab!(T1, T2, T3, T4, T5, T6, T7, T8);
-
-type Mapping<'a, E, T> = <<E as IntoComponents>::Components as ComponentsMapping<T>>::Ref<'a>;
-type MappingMut<'a, E, T> = <<E as IntoComponents>::Components as ComponentsMapping<T>>::RefMut<'a>;
-
-type MappingById<'a, E, T> =
- <<E as IntoComponents>::Components as ComponentsByIdMapping<T>>::Ref<'a>;
-type MappingByIdMut<'a, E, T> =
- <<E as IntoComponents>::Components as ComponentsByIdMapping<T>>::RefMut<'a>;
-
// I think it's better to *NOT* use `Components` directly on the `with` methods.
// Instead use the `Self::EntityRef` type directly.
// This way we can auto implement the `with_by_id` method.
@@ -175,14 +79,10 @@ pub type ComponentsById<'a, T> = <T as IntoComponentsById>::Output;
// and everytime we add a new component to an entity, we need to update the tuple type everywhere.
// Better idea would be to use the `EntityRef` type directly inside of the `with_components_by_id` closure
// -- f(components.into_components_by_id(id)) -> components.into_components_by_id(id) would return `EntityRef`, rather than the tuple.
-pub trait EntityComponentSystem<T>
-where
- <Self::Entity as IntoComponents>::Components: ComponentsMapping<T> + ComponentsByIdMapping<T>,
-{
+pub trait EntityComponentSystem<T> {
type Idx;
type Entity: IntoComponents + EntityMarker;
- type EntityComponents<'a>: IntoComponents<Components = Mapping<'a, Self::Entity, T>>
- + IntoComponentsById<Idx = Self::Idx, Output = MappingById<'a, Self::Entity, T>>;
+ type EntityComponents<'a>: IntoComponents + IntoComponentsById<Idx = Self::Idx>;
fn with_components<O, F>(&self, f: F) -> O
where
@@ -197,8 +97,7 @@ where
}
pub trait EntityComponentSystemMut: EntityComponentSystem<Borrow> {
- type EntityComponentsMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, Borrow>>
- + IntoComponentsById<Idx = Self::Idx, Output = MappingByIdMut<'a, Self::Entity, Borrow>>;
+ type EntityComponentsMut<'a>: IntoComponents + IntoComponentsById<Idx = Self::Idx>;
fn with_components_mut<O, F>(&mut self, f: F) -> O
where
@@ -213,11 +112,7 @@ pub trait EntityComponentSystemMut: EntityComponentSystem<Borrow> {
}
pub trait EntityComponentSystemMutCell: EntityComponentSystem<InteriorMutability> {
- type EntityComponentsMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, InteriorMutability>>
- + IntoComponentsById<
- Idx = Self::Idx,
- Output = MappingByIdMut<'a, Self::Entity, InteriorMutability>,
- >;
+ type EntityComponentsMut<'a>: IntoComponents + IntoComponentsById<Idx = Self::Idx>;
fn with_components_mut<O, F>(&self, f: F) -> O
where