Google GitGoogle Git
Sign in
apache/iggy/refs/heads/cpp-high-level-client-4/./core/server/src/dispatch/partition.rs
blob: a32d8e9bc1593c9a7f8d03d66add2612b4da3992 [file]
// 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.
//! Partition requests and replies through the shard mesh.
//!
//! Writes enter through [`dispatch_partition_request`]. Reads enter through
//! [`handle_poll_messages`] and [`handle_get_consumer_offset`], then execute
//! on the owning shard. The partition engine sends replies for committed writes;
//! this module sends read replies and admission errors through the connection's
//! home shard.
use crate::consumer_group::maybe_rewrite_consumer_offset_request;
use crate::dispatch::authz::{authorize_partition_op, authorize_partition_read};
use crate::dispatch::failure::{
FrameChannel, send_deny_reply, send_host_frame, send_non_replicated_bytes,
send_non_replicated_deny, send_result_rejection,
};
use crate::dispatch::submit::submit_client_request_on_owner;
use crate::namespace::{resolve_partition_namespace, resolve_partition_request_namespace};
use crate::reply_frame::{
build_consumer_offset_body, build_polled_messages_reply, current_metadata_commit,
};
use crate::shell::{ShellBus, ShellShard};
use crate::wire::request_body;
use bytes::Bytes;
use consensus::client_table::SessionAttachment;
use consensus::{MetadataHandle, PartitionsHandle};
use iggy_binary_protocol::PrepareHeader;
use iggy_binary_protocol::primitives::consumer::WireConsumer;
use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy;
use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
use iggy_binary_protocol::requests::messages::PollMessagesRequest;
use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
use iggy_binary_protocol::{
KIND_CONSUMER, KIND_CONSUMER_GROUP, KIND_EXTERNAL_GROUP, Operation, RoutedRequestHeader,
WireDecode, WireIdentifier,
};
use iggy_common::{ConsumerKind, IggyError, PollingStrategy, RESYNC_REQUIRED_PARTITION_SENTINEL};
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
use message_bus::BusMessage;
use metadata::impls::metadata::{
StreamsFrontend, build_truncate_partition_client_message,
build_truncate_partition_client_message_with_identifiers,
};
use metadata::stm::stream::TruncatePartitionRequest;
use partitions::{PollingArgs, PollingConsumer};
use server_common::Message;
use server_common::sharding::IggyNamespace;
use shard::shards_table::ShardsTable;
use shard::{ConsumerAttachment, PartitionRead, PartitionReadReply};
use std::rc::Rc;
use tracing::{debug, warn};
/// Route a partition data-plane op (`SendMessages` / consumer-offset writes)
/// through the shard mesh by namespace: the op belongs to the partition's
/// own consensus group, not the metadata group. The owning shard's
/// partitions plane dedups the request against its group's slice, so
/// `header.client` carries the VSR consensus id (the dedup key) rather than
/// the transport id.
///
/// Attached requests retain their session fence through the owning shard's
/// materialization queue and relay the committed reply through an in-process
/// channel. HTTP registers its session id on the bus, which delivers that
/// reply to its installed request slot, including detached `?ack=none` writes.
/// Binary transports relay to the socket's transport id. An unattached request
/// whose consensus id is directly routable may dispatch without a relay.
///
/// Callers must have authenticated the transport already: `vsr_client_id` /
/// `bound_session` come from its bound VSR session. Every failure before
/// dispatch replies with a nonzero status -- unresolvable namespace,
/// authorization denial, exhausted routable wait -- so the client fails fast
/// instead of wedging on a silent drop or reading a status-0 frame as a
/// committed write.
///
/// `vsr_client_id` keys the consumer-group offset fence (the member id),
/// not the transport id stamped into the partition-op header.
#[allow(
clippy::future_not_send,
clippy::too_many_arguments,
clippy::too_many_lines
)]
pub async fn dispatch_partition_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
request: Message<RoutedRequestHeader>,
vsr_client_id: u128,
bound_session: u64,
transport_client_id: u128,
acting_user_id: Option<u32>,
consumer_session: Option<(u128, SessionAttachment)>,
resolved_namespace: Option<u64>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
let header = *request.header();
let metadata_watermark = shard.plane.metadata().applied_frontier().get();
if consumer_session
.as_ref()
.is_some_and(|(_, attachment)| !attachment.is_valid())
{
send_deny_reply(
shard,
transport_client_id,
&header,
IggyError::StaleClient.as_code(),
)
.await;
return;
}
let namespace = match resolved_namespace.map_or_else(
|| {
resolve_partition_request_namespace(
shard,
header.operation,
request_body(&request),
consumer_session
.as_ref()
.map_or(vsr_client_id, |(parent, _)| *parent),
)
},
Ok,
) {
Ok(namespace) => namespace,
Err(error) => {
// A partition op against a stream/topic that no longer resolves
// (e.g. a consumer's trailing auto-commit racing a `delete_stream`,
// or an explicit partition id that skipped the client-side
// resolve). The op never reached the partition plane, so a status-0
// reply would read as a committed ack for work that never happened.
// A silent drop is no better: the SDK connection processes replies
// in lockstep and would wedge forever.
warn!(
transport_client_id,
error = %error,
operation = ?header.operation,
"partition request with unresolved namespace; replying denied"
);
let status = if matches!(
header.operation,
Operation::StoreConsumerOffset | Operation::DeleteConsumerOffset
) {
error.as_code()
} else {
IggyError::ResourceNotFound(String::new()).as_code()
};
send_deny_reply(shard, transport_client_id, &header, status).await;
return;
}
};
// Dispatch-time RBAC. The partition plane is not replicated through the
// metadata STM, so the in-apply gate cannot cover it; authorize here, on
// the connection's own shard, before burning the routable wait or touching
// the plane. The namespace resolved above, so its stream/topic are the
// committed slab ids the permissioner keys on directly. A denial replies
// the op's frame with an empty body and a nonzero `status` the SDK peeks.
//
// Consistency: this reads THIS shard's local committed permissioner. On a
// peer shard that is a replicated read-mirror, so a permission revocation
// takes effect on the partition plane only once this shard applies the
// revoking commit -- an apply-lag window bounded by replication lag.
// Control-plane ops are exact (gated in-apply, in the same committed order
// on every replica); this local-read relaxation on the data plane is the
// accepted trade for keeping partition ops off the metadata consensus.
let scope = IggyNamespace::from_raw(namespace);
if let Some(status) = authorize_partition_op(
shard,
header.operation,
acting_user_id,
scope.stream_id(),
scope.topic_id(),
) {
warn!(
transport_client_id,
status,
operation = ?header.operation,
"partition request denied by authorization; replying with status"
);
send_deny_reply(shard, transport_client_id, &header, status).await;
return;
}
let attachment = consumer_session
.map(|(parent, session)| {
capture_offset_attachment(
shard.plane.metadata().mux_stm.streams(),
scope,
request_body(&request),
parent,
session,
header.operation,
)
})
.transpose();
let attachment = match attachment {
Ok(attachment) => attachment,
Err(error) => {
send_deny_reply(shard, transport_client_id, &header, error.as_code()).await;
return;
}
};
// Convergence wait: a CreateTopic commit returns to the client before the
// per-shard reconcilers seed routing rows and materialise the partition
// (next wake/periodic tick). An op arriving inside that window is not lost
// if it skips this wait -- `router::route_typed` falls back to the hash
// assignment, and the owning shard parks it -- so this is an admission
// courtesy that keeps the steady state off that park buffer, not a
// correctness gate. See `wait_for_partition_routable`, which spells out why
// there is no owner-readiness probe here any more.
if !wait_for_partition_routable(shard, IggyNamespace::from_raw(namespace)).await {
// The op never reached the partition plane, so it is safe to re-issue
// anywhere -- the same contract the plane itself answers for a
// non-primary routing artifact. A status-0 empty reply here would
// fabricate a success ack for a write that hit no partition at all.
warn!(
transport_client_id,
namespace,
operation = ?header.operation,
"partition request not routable within budget; replying transient"
);
send_deny_reply(
shard,
transport_client_id,
&header,
IggyError::TransientNotAccepted.as_code(),
)
.await;
return;
}
// A group consumer-offset op carries the group NAME on the wire; the
// partition plane keys the offset by the group's monotonic id (the same
// key the poll path auto-commits under and the read path resolves), so
// rewrite the consumer id before replication -- the apply layer has no
// metadata access to resolve it.
let request = match maybe_rewrite_consumer_offset_request(shard, request) {
Ok(rewritten) => rewritten,
// Metadata can change during the routing wait, so a previously valid
// group may be missing now. Preserve the typed failure before submit.
Err(error) => {
warn!(
transport_client_id,
error = %error,
operation = ?header.operation,
"failed to rewrite consumer-offset request; replying denied"
);
send_deny_reply(shard, transport_client_id, &header, error.as_code()).await;
return;
}
};
let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
*new_header = header;
new_header.group = namespace;
// The VSR consensus id, exactly as metadata ops are stamped: it is the
// dedup key every replica keys its slice by, and unlike the transport
// id it stays valid across nodes. Replies therefore cannot be routed
// by this field -- they ride the submit's channel back to this shard,
// which owns the socket.
new_header.client = vsr_client_id;
// Header validation requires `session > 0` for non-register ops. The
// slices mint no epoch of their own, so the bound VSR session only
// satisfies validation here.
new_header.session = bound_session;
// The session's owner as this shard resolved it, never the
// client-supplied value: the dedup slice keys on it to tell a re-minted
// id's next holder apart from its previous one, so it has to be
// trustworthy on every replica. Same rule as the metadata plane. The
// gate above failed closed on `None`, so `0` (unattributed) is only a
// type-level fallback here.
new_header.user_id = acting_user_id.unwrap_or(0);
new_header.metadata_watermark = metadata_watermark;
});
if attachment.is_none() && vsr_client_id == transport_client_id {
shard.dispatch(request.into_generic());
return;
}
relay_partition_reply(
shard,
IggyNamespace::from_raw(namespace),
request,
transport_client_id,
&header,
attachment,
)
.await;
}
/// Submit a partition write whose reply the bus cannot route (the routed
/// header's `client` is the VSR consensus id, whose bits encode no home shard)
/// and relay the committed reply to the socket this shard holds.
///
/// Only the submit runs on the caller's drain loop: it is what fixes the order
/// two writes from one connection reach the owning shard in. The wait for the
/// commit is spawned, so a connection keeps draining its queued polls and
/// metadata ops instead of holding them behind one replication round trip.
#[allow(clippy::future_not_send)]
async fn relay_partition_reply<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
namespace: IggyNamespace,
request: Message<RoutedRequestHeader>,
transport_client_id: u128,
header: &RoutedRequestHeader,
attachment: Option<ConsumerAttachment>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
let consumer_kind = consumer_offset_kind(&request);
let Ok(ticket) = shard.partition_submit_attached(namespace, request, attachment) else {
// `PartitionSubmitRefused`: the frame never reached the owning shard,
// so this is a known outcome and the client can be told now rather
// than after its read-timeout. Same transient the plane itself answers
// for a request it could not admit.
send_deny_reply(
shard,
transport_client_id,
header,
IggyError::TransientNotAccepted.as_code(),
)
.await;
return;
};
let operation = header.operation;
let shard = Rc::clone(shard);
// Through the bus, not the runtime directly: the simulator supplies its
// own executor and virtual clock.
shard.bus.clone().spawn(async move {
let Some(reply) = shard.await_partition_submit(ticket).await else {
// Abandoned or expired. Deliberately silent: the outcome is
// unknown, and a synthesized failure could contradict a write that
// commits moments later. The client's read-timeout is the recovery.
return;
};
if let Some(kind) = consumer_kind
&& reply
.as_slice()
.get(..size_of::<iggy_binary_protocol::ReplyHeader>())
.and_then(|bytes| {
bytemuck::checked::try_from_bytes::<iggy_binary_protocol::ReplyHeader>(bytes)
.ok()
})
.is_some_and(|header| header.status == IggyError::TooManyConsumerOffsets.as_code())
{
shard.metrics().record_consumer_offset_denied(kind);
}
if let Err(error) = shard
.bus
.send_to_client(transport_client_id, reply.into_frozen())
.await
{
if matches!(error, message_bus::SendError::ClientNotFound(_)) {
tracing::debug!(transport_client_id, ?operation, %error, "partition reply target is gone");
} else {
warn!(transport_client_id, ?operation, %error, "failed to forward committed partition reply to its socket");
}
}
});
}
fn capture_offset_attachment(
streams: &metadata::stm::stream::Streams,
namespace: IggyNamespace,
body: &[u8],
parent: u128,
session: SessionAttachment,
operation: Operation,
) -> Result<ConsumerAttachment, IggyError> {
if !session.is_valid() {
return Err(IggyError::StaleClient);
}
let wire = if operation == Operation::SendMessages {
None
} else {
Some(
GetConsumerOffsetRequest::decode(body)
.map_err(|_| IggyError::InvalidCommand)?
.0,
)
};
let group = if let Some(wire) = wire.filter(|wire| wire.consumer.kind == KIND_CONSUMER_GROUP) {
Some(crate::namespace::resolve_offset_group_id(
streams,
&wire.stream_id,
&wire.topic_id,
&wire.consumer.id,
)?)
} else {
None
};
let metadata = streams
.consumer_offset_metadata(namespace, group, parent)
.ok_or(IggyError::TransientNotAccepted)?;
Ok(ConsumerAttachment { session, metadata })
}
fn consumer_offset_kind(request: &Message<RoutedRequestHeader>) -> Option<ConsumerKind> {
if request.header().operation != Operation::StoreConsumerOffset {
return None;
}
// WireConsumer starts with its kind byte. The dispatch path has already
// decoded and validated the complete request.
ConsumerKind::from_code(*request_body(request).first()?).ok()
}
/// Serve `poll_messages`: resolve the partition namespace, run the read on
/// the owning shard ([`shard::IggyShard::partition_read`]), and re-encode
/// the stored batches into the legacy wire `PolledMessages` body.
///
/// Owner rejections and missing replies return a nonzero status. A missing
/// reply leaves acceptance unknown: the owner may still advance progress, so
/// it cannot carry a retry hint that promises the poll was not accepted.
/// A consumer group poll fenced by the coordinator carries the resync sentinel.
/// Other unexpected owner replies and encoding failures retain the empty poll
/// fallback; these do not establish that the partition has no messages.
#[allow(clippy::future_not_send)]
pub(in crate::dispatch) async fn handle_poll_messages<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
request: &Message<RoutedRequestHeader>,
user_id: Option<u32>,
consumer_client_id: u128,
attachment: Option<SessionAttachment>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
let Ok(wire) = PollMessagesRequest::decode_from(request_body(request)) else {
// A permanent client error, so it must not borrow the empty-poll
// shape: that body decodes as a successful 0-message poll, and a
// consumer looping on it never learns why. The server already sends
// nonzero-status poll replies (authz, unresolved target).
send_non_replicated_deny(
shard,
request,
transport_client_id,
IggyError::InvalidCommand.as_code(),
)
.await;
return;
};
if attachment
.as_ref()
.is_some_and(|attachment| !attachment.is_valid())
{
send_non_replicated_deny(
shard,
request,
transport_client_id,
IggyError::StaleClient.as_code(),
)
.await;
return;
}
// Gate on (stream, topic) before touching the partition plane. A resolution
// miss denies typed on the resolve path below; a denial here replies
// status!=0 with an empty body, distinct from the empty-poll "0 messages"
// shape.
if let Some(status) = authorize_partition_read(
shard,
&wire.stream_id,
&wire.topic_id,
user_id,
|permissioner, uid, stream_id, topic_id| {
permissioner.poll_messages(uid, stream_id, topic_id)
},
) {
send_non_replicated_deny(shard, request, transport_client_id, status).await;
return;
}
let (body, channel) = match resolve_poll_request(shard, &wire, consumer_client_id) {
Ok(resolved) => {
match read_polled_messages(
shard,
transport_client_id,
request,
resolved,
attachment,
consumer_client_id,
)
.await
{
Ok(reply) => {
send_host_frame(
&shard.bus,
transport_client_id,
reply,
FrameChannel::Reply,
"poll_messages",
)
.await;
return;
}
Err(ReadPolledMessagesError::Fallback(fallback)) => fallback,
Err(ReadPolledMessagesError::Rejected(error)) => {
send_non_replicated_deny(shard, request, transport_client_id, error.as_code())
.await;
return;
}
}
}
// A generation fence: the client's cached assignment went stale after a
// rebalance. The empty poll carries the re-sync sentinel so the SDK
// re-syncs and retries instead of reading end-of-partition.
Err(error @ IggyError::ConsumerGroupPartitionNotOwned(..)) => {
warn!(
transport_client_id,
error = %error,
"poll_messages fenced; replying re-sync sentinel"
);
empty_poll_fallback(RESYNC_REQUIRED_PARTITION_SENTINEL)
}
// Everything else the resolve rejects is a permanent client error: an
// unresolved stream, topic, or partition, a polling strategy or
// consumer kind outside the wire enum, or a group poll that omitted
// its partition id. All must surface typed -- the empty poll is a
// successful 0-message read, and a consumer looping on it never learns
// why.
Err(error) => {
warn!(
transport_client_id,
error = %error,
"poll_messages rejected; replying denied"
);
send_non_replicated_deny(shard, request, transport_client_id, error.as_code()).await;
return;
}
};
send_non_replicated_bytes(
shard,
request,
transport_client_id,
body,
channel,
"poll_messages",
)
.await;
}
/// Run the resolved poll on the owning shard and re-encode the stored
/// batches into the wire `PolledMessages` reply. Owner rejections preserve
/// their error; a missing reply reports a communication error without claiming
/// that the owner rejected the read. Unexpected replies and encoding failures
/// retain the empty poll fallback.
#[allow(clippy::future_not_send)]
async fn read_polled_messages<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
request: &Message<RoutedRequestHeader>,
(namespace, partition_id, consumer, args): DecodedPollRequest,
attachment: Option<SessionAttachment>,
consumer_client_id: u128,
) -> Result<BusMessage, ReadPolledMessagesError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
let read = if let Some(attachment) = attachment {
let group_id = match consumer {
PollingConsumer::ConsumerGroup(group, _) => Some(group as u64),
PollingConsumer::Consumer(..) => None,
};
let metadata = shard
.plane
.metadata()
.mux_stm
.streams()
.poll_metadata(namespace, group_id, consumer_client_id)
.ok_or(ReadPolledMessagesError::Rejected(
IggyError::TransientNotAccepted,
))?;
PartitionRead::PollOnPrimary {
consumer,
args,
attachment: ConsumerAttachment {
session: attachment,
metadata,
},
}
} else {
PartitionRead::Poll { consumer, args }
};
match shard.partition_read(namespace, read).await {
Some(PartitionReadReply::Poll {
fragments,
current_offset,
}) => build_polled_messages_reply(
request.header(),
current_metadata_commit(shard),
partition_id,
current_offset,
fragments,
shard.plane.partitions().config().encryptor.as_deref(),
)
.map_err(|error| {
warn!(
transport_client_id,
error = %error,
"failed to re-encode polled batches; replying empty poll"
);
ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id))
}),
Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)),
None => {
// The owner rejects a disconnected receiver before poll admission,
// but timeout can race with that check. Missing replies therefore
// do not prove rejection and must not be reported as empty success.
warn!(
transport_client_id,
namespace = namespace.inner(),
"partition read reply unavailable; acceptance unknown"
);
Err(ReadPolledMessagesError::Rejected(
IggyError::ShardCommunicationError,
))
}
Some(other) => {
warn!(
transport_client_id,
namespace = namespace.inner(),
reply = ?other,
"unexpected partition poll reply; replying empty poll"
);
Err(ReadPolledMessagesError::Fallback(empty_poll_fallback(
partition_id,
)))
}
}
}
enum ReadPolledMessagesError {
Fallback((Bytes, FrameChannel)),
Rejected(IggyError),
}
/// The fail-fast poll reply for a partition that could not answer: the
/// 16-byte empty poll for `partition_id`, riding the re-sync sentinel
/// channel when the id is the sentinel and the empty-frame channel
/// otherwise.
fn empty_poll_fallback(partition_id: u32) -> (Bytes, FrameChannel) {
let channel = if partition_id == RESYNC_REQUIRED_PARTITION_SENTINEL {
FrameChannel::ResyncSentinel
} else {
FrameChannel::EmptyFrame
};
(empty_polled_messages_body(partition_id), channel)
}
/// Serve `get_consumer_offset`. An empty status-0 body decodes as `None` on
/// the SDK side, so it is reserved for "this consumer has no stored offset" --
/// an unresolved consumer group included, since a deleted group has no offset
/// to report. A malformed request or an unresolved stream, topic, or partition
/// denies with a nonzero status instead, so an addressing typo cannot read
/// back as a fresh consumer. A read that its owner does not answer in time, or
/// whose owner does not hold the partition, denies with `TransientNotAccepted`.
// TODO(hubcio): plain local partition_read with no primary gate, so a
// follower answers from its own (possibly lagging) offset state. Needs the
// same is-caught-up-primary gate the auto-commit path has, or an explicit
// read-from-follower contract.
#[allow(clippy::future_not_send)]
pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
request: &Message<RoutedRequestHeader>,
user_id: Option<u32>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
let Ok(wire) = GetConsumerOffsetRequest::decode_from(request_body(request)) else {
// Same rule as the poll above: an empty body is the legitimate
// "no offset stored" answer, so a malformed request must not send
// one. Byte-identical frames on the same channel with the same
// context would also be indistinguishable in the send-failure log.
send_non_replicated_deny(
shard,
request,
transport_client_id,
IggyError::InvalidCommand.as_code(),
)
.await;
return;
};
if let Some(status) = authorize_partition_read(
shard,
&wire.stream_id,
&wire.topic_id,
user_id,
|permissioner, uid, stream_id, topic_id| {
permissioner.get_consumer_offset(uid, stream_id, topic_id)
},
) {
send_non_replicated_deny(shard, request, transport_client_id, status).await;
return;
}
let body = match resolve_consumer_offset_request(shard, &wire) {
Ok(Some((namespace, partition_id, read))) => {
match shard.partition_read(namespace, read).await {
Some(PartitionReadReply::ConsumerOffset {
stored: Some(stored_offset),
current_offset,
}) => build_consumer_offset_body(partition_id, current_offset, stored_offset),
Some(PartitionReadReply::ConsumerOffset { stored: None, .. }) => Bytes::new(),
Some(PartitionReadReply::Rejected(error)) => {
send_non_replicated_deny(shard, request, transport_client_id, error.as_code())
.await;
return;
}
// No reply, or an owner that does not hold the partition yet,
// says nothing about the offset, so it must not read as "no
// stored offset". A read moves no progress, so the client may
// retry it. The SDK retries every 50 ms, so the not-yet-held
// case logs at debug.
reply => {
if matches!(reply, Some(PartitionReadReply::NotFound)) {
debug!(
transport_client_id,
namespace = namespace.inner(),
"consumer offset partition not held by owner; replying not accepted"
);
} else {
warn!(
transport_client_id,
namespace = namespace.inner(),
?reply,
"consumer offset read unanswered; replying not accepted"
);
}
send_non_replicated_deny(
shard,
request,
transport_client_id,
IggyError::TransientNotAccepted.as_code(),
)
.await;
return;
}
}
}
// An unresolved group has no offset to report, the one thing the
// empty body may mean.
Ok(None) => Bytes::new(),
// Everything the resolve rejects is a client error: an unresolved
// stream, topic, or partition, or `resolve_partition_namespace`'s
// generic rejection. All must surface typed -- an empty body decodes as
// `None`, indistinguishable from "no stored offset yet", so a typo'd or
// deleted target would read back as a fresh consumer and resume from
// its configured default with no error anywhere. The same read over
// REST 404s.
Err(error) => {
warn!(
transport_client_id,
error = %error,
"get_consumer_offset rejected; replying denied"
);
send_non_replicated_deny(shard, request, transport_client_id, error.as_code()).await;
return;
}
};
send_non_replicated_bytes(
shard,
request,
transport_client_id,
body,
FrameChannel::Reply,
"get_consumer_offset",
)
.await;
}
/// Wait (bounded) until this shard holds a routing row for `namespace`. Fast
/// path: row already present -> no wait.
///
/// Covers the post-`CreateTopic` convergence window where the metadata commit
/// has returned to the client but the per-shard reconcilers have not yet seeded
/// routing rows. This is an admission courtesy, not a correctness gate: the row
/// is a cache of the deterministic hash assignment and may exist before the
/// owner has materialised anything, so its presence proves only where the
/// partition belongs. What makes an early arrival safe is the owning shard
/// itself - `park_if_unmaterialised` holds the frame until its partition lands,
/// and `serves_committed_incarnation` refuses to serve a mismatched
/// incarnation. Waiting here simply keeps the steady state off that park
/// buffer, whose overflow is the one path that still sheds a request without
/// replying (`frame_drops_total{variant=partition,reason=park_overflow}`).
///
/// Deliberately no owner-readiness probe. One used to run here, on the theory
/// that the table could not be trusted; it could not close the window either,
/// because the fast path above skipped it in exactly the case it was meant to
/// cover - a row seeded from the hash by a shard that owns nothing. Readiness
/// belongs to the owner, which is where it is now enforced.
#[allow(clippy::future_not_send)]
async fn wait_for_partition_routable<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
namespace: IggyNamespace,
) -> bool
where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
const ATTEMPT_DELAY: std::time::Duration = std::time::Duration::from_millis(50);
// 3s budget at 50ms per attempt. Counting attempts, not reading a
// wall-clock deadline, keeps the wait virtual under the simulator: the
// bus sleep advances virtual time, whereas `Instant::now` would not.
const MAX_ATTEMPTS: u32 = 60;
let mut attempts = 0u32;
while shard.shards_table().shard_for(namespace).is_none() {
if attempts >= MAX_ATTEMPTS {
return false;
}
attempts += 1;
shard.bus.sleep(ATTEMPT_DELAY).await;
}
true
}
/// The 16-byte `PolledMessages` body with zero messages
/// (`[partition_id:4][current_offset:8][count:4]`). The SDK decoder
/// requires at least this header, so failure paths must never reply a
/// zero-byte body.
fn empty_polled_messages_body(partition_id: u32) -> Bytes {
let mut body = Vec::with_capacity(16);
body.extend_from_slice(&partition_id.to_le_bytes());
body.extend_from_slice(&0u64.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
Bytes::from(body)
}
type DecodedPollRequest = (IggyNamespace, u32, PollingConsumer, PollingArgs);
/// Resolve a decoded poll request into its owning-shard read: namespace,
/// partition, polling consumer, and args. Shared by the TCP dispatch (client
/// id = the connection's bound VSR client) and the HTTP route (client id 0,
/// which fences group polls closed).
#[allow(clippy::cast_possible_truncation)]
pub fn resolve_poll_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
wire: &PollMessagesRequest,
client_id: u128,
) -> Result<DecodedPollRequest, IggyError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
// An external group only holds offsets. Nothing in Iggy may poll with it.
if wire.consumer.kind == KIND_EXTERNAL_GROUP {
return Err(IggyError::FeatureUnavailable);
}
let strategy = polling_strategy_from_wire(&wire.strategy)?;
let args = PollingArgs::new(strategy, wire.count, wire.auto_commit);
// Consumer-group poll: the client selects which of its assigned partitions
// to read and sends it explicitly. The coordinator FENCES ownership (a stale
// client whose partition was reassigned is rejected with
// `ConsumerGroupPartitionNotOwned`, prompting a re-sync) and resolves the
// group's monotonic id -- the offset key the store rewrite and read path
// both use, so `next()` reads back the offset it just committed.
if wire.consumer.kind == KIND_CONSUMER_GROUP {
let partition_id = wire.partition_id.ok_or(IggyError::InvalidIdentifier)?;
let group_id = shard
.plane
.metadata()
.mux_stm
.streams()
.consumer_group_fence(
&wire.stream_id,
&wire.topic_id,
&wire.consumer.id,
client_id,
partition_id,
// Poll fence: reject a pending-revoked partition so the source
// re-syncs and skips it (it still commits it via the offset fence).
true,
)
.ok_or(IggyError::ConsumerGroupPartitionNotOwned(
client_id as u32,
partition_id,
))?;
let namespace = resolve_partition_namespace(
shard,
&wire.stream_id,
&wire.topic_id,
Some(partition_id),
)?;
#[allow(clippy::cast_possible_truncation)]
let consumer = PollingConsumer::ConsumerGroup(group_id as usize, partition_id as usize);
return Ok((namespace, partition_id, consumer, args));
}
// Plain-consumer poll: an omitted partition selects partition 0, matching
// the legacy resolver (`resolve_consumer_with_partition_id` uses
// `unwrap_or(0)` for `ConsumerKind::Consumer`).
let partition_id = wire.partition_id.unwrap_or(0);
let namespace =
resolve_partition_namespace(shard, &wire.stream_id, &wire.topic_id, Some(partition_id))?;
let consumer = polling_consumer_from_wire(&wire.consumer, partition_id)?;
Ok((namespace, partition_id, consumer, args))
}
/// Resolve a decoded consumer-offset read into its owning-shard read:
/// namespace, partition, and the read to run there. Shared by the TCP dispatch
/// and the HTTP route; needs no client id because offset reads are not fenced
/// (any client may read a group's offset, member or not).
///
/// `Ok(None)` is the one outcome that means "no offset exists to report" (an
/// unresolved group). It is a separate outcome rather than an error code
/// because [`resolve_partition_namespace`] also rejects an addressing miss
/// with `InvalidIdentifier`, and the caller must deny that one typed.
pub fn resolve_consumer_offset_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
wire: &GetConsumerOffsetRequest,
) -> Result<Option<(IggyNamespace, u32, PartitionRead)>, IggyError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
// Omitted partition reads partition 0, matching the legacy resolver for
// both consumer kinds (`unwrap_or(0)`).
let partition_id = wire.partition_id.unwrap_or(0);
let namespace =
resolve_partition_namespace(shard, &wire.stream_id, &wire.topic_id, Some(partition_id))?;
if wire.consumer.kind != KIND_CONSUMER_GROUP && wire.consumer.kind != KIND_EXTERNAL_GROUP {
let consumer = polling_consumer_from_wire(&wire.consumer, partition_id)?;
return Ok(Some((
namespace,
partition_id,
PartitionRead::ConsumerOffset { consumer },
)));
}
// A group offset is keyed by the group's monotonic id (any client may read
// it, member or not), the same key the write path is rewritten to. An
// unresolved group (e.g. deleted) has no offset, so the read reports None.
let Some(group_id) = shard
.plane
.metadata()
.mux_stm
.streams()
.resolve_consumer_group_id(&wire.stream_id, &wire.topic_id, &wire.consumer.id)
else {
return Ok(None);
};
#[allow(clippy::cast_possible_truncation)]
let read = if wire.consumer.kind == KIND_EXTERNAL_GROUP {
PartitionRead::ExternalGroupOffset {
group_id: u32::try_from(group_id).unwrap_or(u32::MAX),
}
} else {
PartitionRead::ConsumerOffset {
consumer: PollingConsumer::ConsumerGroup(group_id as usize, partition_id as usize),
}
};
Ok(Some((namespace, partition_id, read)))
}
fn polling_consumer_from_wire(
consumer: &WireConsumer,
partition_id: u32,
) -> Result<PollingConsumer, IggyError> {
// Mirrors the legacy server's `PollingConsumer::resolve_consumer_id`:
// numeric ids pass through, named consumers hash to a stable u32 so
// reads derive the same offset-table key the write path stores under.
let consumer_id = match &consumer.id {
iggy_binary_protocol::WireIdentifier::Numeric(id) => *id,
iggy_binary_protocol::WireIdentifier::String(name) => {
iggy_common::calculate_32(name.as_str().as_bytes())
}
} as usize;
match consumer.kind {
KIND_CONSUMER => Ok(PollingConsumer::Consumer(
consumer_id,
partition_id as usize,
)),
KIND_CONSUMER_GROUP => Ok(PollingConsumer::ConsumerGroup(
consumer_id,
partition_id as usize,
)),
_ => Err(IggyError::InvalidCommand),
}
}
fn polling_strategy_from_wire(
strategy: &WirePollingStrategy,
) -> Result<PollingStrategy, IggyError> {
let mut mapped = match strategy.kind {
1 => PollingStrategy::offset(0),
2 => PollingStrategy::timestamp(iggy_common::IggyTimestamp::from(strategy.value)),
3 => PollingStrategy::first(),
4 => PollingStrategy::last(),
5 => PollingStrategy::next(),
_ => return Err(IggyError::InvalidCommand),
};
mapped.set_value(strategy.value);
Ok(mapped)
}
/// Handle a client `DeleteSegments`: resolve the requested count to an offset
/// on the owning shard, replicate a `TruncatePartition` through metadata so
/// every replica trims to the same watermark, then ack the client. The local
/// deletion happens later, when each replica's reconciler observes the commit.
///
/// The consensus reply is forwarded verbatim: nothing-to-delete commits a
/// no-op `TruncatePartition(0)` and acks, while a not-primary rejection
/// reaches the client as `TransientNotCommitted` so the SDK replays instead
/// of mistaking a dropped delete for success. Nothing is acked empty: a
/// malformed body denies typed, and an unresolvable namespace commits a typed
/// rejection against the client's raw identifiers so its retry dedups.
#[allow(clippy::future_not_send)]
pub(in crate::dispatch) async fn handle_delete_segments_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
bound: Option<(u128, u64)>,
request: &Message<RoutedRequestHeader>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
let header = *request.header();
let body = request_body(request);
// An unbound transport cannot be attributed a VSR request sequence; the
// outer handler already short-circuits these, so this is defensive.
let Some((vsr_client_id, session)) = bound else {
return;
};
// The client numbers DeleteSegments in the same monotonic request sequence
// as every other metadata op. So resolve the requested count to a concrete
// offset on the owning shard, then replicate a `TruncatePartition(offset)`
// AS the client's own request through the standard owner path: the commit
// records (client, session, request) in the `ClientTable` on every replica,
// advancing the watermark. Skipping the commit (or attributing it to an
// internal id) leaves this request id unrecorded, so the SDK's own retry
// of it would re-execute instead of deduping. A no-op delete still
// commits `up_to_offset = 0` (monotonic apply) for the same reason.
let truncate = match resolve_delete_segments_truncate(
shard,
&header,
vsr_client_id,
session,
body,
)
.await
{
Ok(truncate) => truncate,
// The owning partition has not converged on the committed log yet, so
// the delete cannot be resolved to a watermark. Reply with the
// result-framed transient rejection (under the TruncatePartition
// operation, which the SDK decodes) so the client replays the same
// request once the partition catches up. Nothing was submitted, hence
// the re-issuable-anywhere flavor.
Err(IggyError::TransientNotAccepted) => {
let template = build_truncate_partition_client_message(
&header,
vsr_client_id,
session,
&TruncatePartitionRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
partition_id: 0,
up_to_offset: 0,
expected_history: None,
},
);
send_result_rejection(
shard,
transport_client_id,
template.header(),
&IggyError::TransientNotAccepted,
"delete_segments_transient_rejection",
)
.await;
return;
}
// Undecodable body (never produced by the SDK): deny typed rather
// than ack empty. Both keep the lockstep stream framed, but a
// status-0 ack reads as a completed trim. Unresolvable-but-well-formed
// targets commit a typed rejection instead (see the resolve).
Err(error) => {
send_deny_reply(shard, transport_client_id, &header, error.as_code()).await;
return;
}
};
// Forward the consensus reply verbatim, exactly like the generic metadata
// path: a committed success acks the delete, and a result-framed
// `TransientNotCommitted` rejection makes the SDK replay the request.
// Acking unconditionally here would swallow a not-primary rejection and
// drop the delete on the floor while the client believes it succeeded.
let Some(reply) = submit_client_request_on_owner(shard, truncate).await else {
// Transient submit failure (not primary / view change). Stay silent;
// the SDK read-timeout replays the same request id, which re-resolves
// and commits. Acking here would advance the client past an
// unrecorded request and gap the next metadata op.
warn!(
transport_client_id,
"delete_segments: transient submit; client will replay"
);
return;
};
send_host_frame(
&shard.bus,
transport_client_id,
reply.into_frozen(),
FrameChannel::Reply,
"delete_segments_reply",
)
.await;
}
/// Resolve a client `DeleteSegments` to the `TruncatePartition` that commits the
/// trim. Shared by the TCP dispatch and the HTTP listener so both resolve the
/// requested segment count to a concrete watermark identically.
///
/// `template` supplies the wire `cluster` / `view` / `release` and the client's
/// `request` number; `client_id` / `session` are the bound VSR identity the
/// truncate commits under. A resolvable namespace with nothing sealed to delete
/// still yields a `TruncatePartition(up_to_offset = 0)` so the metadata request
/// sequence stays contiguous. Resolved no-ops retain their history guard; a
/// missing owner reply or local partition yields a zero watermark without one.
/// An UNRESOLVABLE namespace yields the truncate
/// against the client's raw identifiers (the apply rejects it as a committed
/// result). `Err` is only `InvalidCommand` for a malformed body and
/// `TransientNotAccepted` for a partition behind the commit frontier or committed
/// history: both transports surface that transient rejection.
#[allow(clippy::future_not_send)]
#[allow(clippy::cast_possible_truncation)]
pub async fn resolve_delete_segments_truncate<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
template: &RoutedRequestHeader,
client_id: u128,
session: u64,
body: &[u8],
) -> Result<Message<RoutedRequestHeader>, IggyError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
S: 'static,
SB: SuperblockStore + 'static,
{
let parsed = DeleteSegmentsRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?;
let namespace_raw = match resolve_partition_request_namespace(
shard,
Operation::DeleteSegments,
body,
client_id,
) {
Ok(namespace_raw) => namespace_raw,
// Unresolvable stream/topic: still commit the truncate, against the
// client's raw identifiers -- the apply rejects it as a committed
// result, so the failure is recorded against the client's request id
// and its retry dedups, while the client gets the typed error an
// empty ack would swallow.
Err(error) => {
debug!(
client_id,
%error,
"delete_segments: unresolved target; committing typed rejection"
);
return Ok(build_truncate_partition_client_message_with_identifiers(
template,
client_id,
session,
parsed.stream_id,
parsed.topic_id,
parsed.partition_id,
0,
));
}
};
let namespace = IggyNamespace::from_raw(namespace_raw);
// A purge after offset resolution must keep its committed history-change result.
let committed_history = shard
.plane
.metadata()
.mux_stm
.streams()
.consumer_offset_metadata(namespace, None, client_id)
.ok_or(IggyError::TransientNotAccepted)?;
let (up_to_offset, expected_history) = match shard
.partition_read(
namespace,
PartitionRead::ResolveSegmentDeleteOffset {
count: parsed.segments_count,
},
)
.await
{
Some(PartitionReadReply::SegmentDeleteOffset {
created_revision,
purge_generation,
..
}) if !committed_history.matches_partition(Some(created_revision), purge_generation) => {
return Err(IggyError::TransientNotAccepted);
}
Some(PartitionReadReply::SegmentDeleteOffset {
up_to_offset: Some(offset),
created_revision,
purge_generation,
..
}) => (offset, Some((created_revision, purge_generation))),
// Nothing sealed to delete on a replica that has not converged on the
// replicated log (a backup behind the commit frontier may be missing
// whole sealed segments). Answering now would commit a no-op truncate
// and silently drop the delete, so surface a transient and let the
// client replay once the partition catches up. A converged primary
// whose resident tail is merely unflushed settles as a no-op below.
Some(PartitionReadReply::SegmentDeleteOffset {
up_to_offset: None,
lagging: true,
..
}) => {
debug!(
client_id,
namespace_raw, "delete_segments: partition not converged; transient"
);
return Err(IggyError::TransientNotAccepted);
}
Some(PartitionReadReply::Rejected(error)) => return Err(error),
Some(PartitionReadReply::SegmentDeleteOffset {
up_to_offset: None,
created_revision,
purge_generation,
..
}) => (0, Some((created_revision, purge_generation))),
other => {
debug!(
client_id,
namespace_raw,
reply = ?other,
"delete_segments: nothing to delete; committing no-op truncate"
);
(0, None)
}
};
Ok(build_truncate_partition_client_message(
template,
client_id,
session,
&TruncatePartitionRequest {
stream_id: WireIdentifier::numeric(namespace.stream_id() as u32),
topic_id: WireIdentifier::numeric(namespace.topic_id() as u32),
partition_id: namespace.partition_id() as u32,
up_to_offset,
expected_history,
},
))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dispatch::test_support::{
SpyBus, TestMux, TestShard, prepare_message, request_message, test_shard,
};
#[cfg(target_os = "linux")]
use consensus::Sequencer;
use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment;
use iggy_binary_protocol::requests::consumer_offsets::DeleteConsumerOffsetRequest;
use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest;
use iggy_binary_protocol::requests::messages::SendMessagesHeader;
use iggy_binary_protocol::requests::streams::CreateStreamRequest;
use iggy_binary_protocol::requests::topics::{
CreateTopicRequest, CreateTopicWithAssignmentsRequest, PurgeTopicRequest,
};
use iggy_binary_protocol::{
Command, PrepareOkHeader, ReplyHeader, WireEncode, WireIdentifier, WireName, WireOptions,
WirePartitioning,
};
use iggy_common::Identifier;
use iggy_common::defaults::DEFAULT_ROOT_USER_ID;
use metadata::IggyMetadata;
use metadata::stm::StateMachine as _;
use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig};
use server_common::MessageBag;
use server_common::sharding::{PartitionLocation, ShardId};
use shard::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant};
use shard::shards_table::PapayaShardsTable;
use shard::{
LifecycleFrame, NoopHost, PartitionConsensusConfig, ReconcileOp, ReplicaTopology,
ShardFrame, ShardIdentity, shard_channel,
};
#[compio::test]
async fn given_unapplied_history_when_resolving_segment_deletes_should_retry() {
const CLIENT: u128 = 1;
let bus = SpyBus::default();
let mut shard = test_shard(&bus, 0, 1, 1);
let (namespace, created_revision) = create_segment_delete_topic(&shard);
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
shard
.plane
.metadata()
.mux_stm
.update(prepare_message(
Operation::PurgeTopic,
CLIENT,
1,
&PurgeTopicRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
}
.to_bytes(),
))
.unwrap();
let body = DeleteSegmentsRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
partition_id: 0,
segments_count: 1,
}
.to_bytes();
let request = request_message(Operation::DeleteSegments, CLIENT, 1, 1, &body);
for history in [(created_revision, 0), (created_revision.wrapping_sub(1), 1)] {
for up_to_offset in [Some(7), None] {
let owner = async {
let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead {
namespace: resolved,
read: PartitionRead::ResolveSegmentDeleteOffset { count: 1 },
reply,
}) = owner_inbox.recv().await.unwrap()
else {
panic!("segment deletion must read its owner");
};
assert_eq!(resolved, namespace);
reply
.try_send(PartitionReadReply::SegmentDeleteOffset {
up_to_offset,
lagging: false,
created_revision: history.0,
purge_generation: history.1,
})
.unwrap();
};
let (result, ()) = futures::join!(
resolve_delete_segments_truncate(&shard, request.header(), CLIENT, 1, &body),
owner,
);
assert!(
matches!(result, Err(IggyError::TransientNotAccepted)),
"offset {up_to_offset:?} from history {history:?} must retry before submission"
);
}
}
}
#[compio::test]
async fn given_purge_after_segment_resolution_when_committing_should_report_history_changed() {
const CLIENT: u128 = 1;
let bus = SpyBus::default();
let mut shard = test_shard(&bus, 0, 1, 1);
let (_, created_revision) = create_segment_delete_topic(&shard);
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
let body = DeleteSegmentsRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
partition_id: 0,
segments_count: 1,
}
.to_bytes();
let request = request_message(Operation::DeleteSegments, CLIENT, 1, 1, &body);
for up_to_offset in [Some(7), None] {
let purge_generation = shard
.plane
.metadata()
.mux_stm
.streams()
.partition_purge_generation(0, 0, 0);
let owner = async {
let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead { reply, .. }) =
owner_inbox.recv().await.unwrap()
else {
panic!("segment deletion must read its owner");
};
reply
.try_send(PartitionReadReply::SegmentDeleteOffset {
up_to_offset,
lagging: false,
created_revision,
purge_generation,
})
.unwrap();
shard
.plane
.metadata()
.mux_stm
.update(prepare_message(
Operation::PurgeTopic,
CLIENT,
1,
&PurgeTopicRequest {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
}
.to_bytes(),
))
.unwrap();
};
let (resolved, ()) = futures::join!(
resolve_delete_segments_truncate(&shard, request.header(), CLIENT, 1, &body),
owner,
);
let resolved = resolved.unwrap();
let truncate = TruncatePartitionRequest::decode(request_body(&resolved))
.unwrap()
.0;
assert_eq!(truncate.up_to_offset, up_to_offset.unwrap_or(0));
assert_eq!(
truncate.expected_history,
Some((created_revision, purge_generation))
);
let reply = shard
.plane
.metadata()
.mux_stm
.update(prepare_message(
Operation::TruncatePartition,
CLIENT,
1,
&truncate.to_bytes(),
))
.unwrap();
assert_eq!(reply.code, IggyError::PartitionHistoryChanged.as_code());
assert_eq!(
shard
.plane
.metadata()
.mux_stm
.streams()
.partition_delete_watermark(0, 0, 0),
0
);
}
}
#[cfg(target_os = "linux")]
#[compio::test]
async fn checkpoint_index_failure_is_returned_by_the_same_partition_tick() {
let root = tempfile::tempdir().unwrap();
let bus = SpyBus::default();
let shard = test_shard(&bus, 0, 3, 1);
let namespace = IggyNamespace::new(1, 1, 0);
let consensus = consensus::VsrConsensus::new(
1,
0,
3,
namespace.inner(),
bus,
consensus::LocalPipeline::new(),
);
consensus.init();
let mut partition = partitions::IggyPartition::with_in_memory_storage(
std::sync::Arc::new(iggy_common::PartitionStats::default()),
consensus,
shard.plane.partitions().config().segment_size,
);
partition.set_runtime_options(iggy_common::TopicRuntimeOptions {
durability: iggy_common::Durability::Persisted,
preallocate_segments: Some(false),
..Default::default()
});
partition.set_partition_dir(root.path().to_string_lossy().into_owned());
let capacity = journal::partition_journal::PARTITION_WAL_BYTES_MAX;
let (persistence, prepares) = partitions::PartitionPersistence::open_with_capacity(
&root.path().join("prepares-0"),
namespace.inner(),
0,
journal::durable_storage::DiskStorage,
capacity,
false,
)
.await
.unwrap();
partition
.open_persistence_with_recovered(
capacity,
std::time::Duration::ZERO,
Some((Rc::clone(&persistence), prepares)),
)
.await
.unwrap();
let mut messages = server_common::send_messages::IggyMessages::with_capacity(1);
messages.push(server_common::send_messages::IggyMessage {
header: server_common::send_messages::IggyMessageHeader::default(),
payload: Bytes::from_static(b"checkpoint"),
user_headers: None,
});
let batch =
server_common::send_messages::SendMessagesOwned::from_messages(namespace, &messages)
.unwrap();
let mut body = vec![0; batch.header.total_size()];
batch.header.encode_into(&mut body);
body[iggy_binary_protocol::batch::BATCH_HEADER_SIZE..].copy_from_slice(&batch.blob);
let prepare = prepare_message(Operation::SendMessages, 1, 1, &body).transmute_header(
|original, header: &mut PrepareHeader| {
*header = original;
header.cluster = 1;
header.group = namespace.inner();
header.checksum = header.identity_checksum();
},
);
let checksum = prepare.header().checksum;
persistence
.append(prepare.clone().into_frozen(), true)
.unwrap();
assert!(persistence.start());
Rc::clone(&persistence).run().await;
partition
.log
.journal()
.inner
.append(prepare.into_frozen())
.await
.unwrap();
partition.log.journal_mut().info.messages_count = 1;
partition.log.journal_mut().info.size = iggy_common::IggyByteSize::from(body.len() as u64);
partition.consensus().sequencer().set_sequence(1);
partition.consensus().set_last_prepare_checksum(checksum);
partition.consensus().restore_commit_state(1, 1);
partition.log.index_writers_mut()[0] = Some(Rc::new(
partitions::IggyIndexWriter::new(
"/dev/full",
Rc::new(std::sync::atomic::AtomicU64::new(0)),
false,
false,
)
.await
.unwrap(),
));
persistence.request_checkpoint();
assert!(partition.needs_persistence_checkpoint());
assert!(partition.fatal().is_none());
shard.plane.partitions().insert(namespace, partition);
let fault = shard
.tick_partitions(&mut Vec::new())
.await
.expect("the checkpoint fault must be returned in its originating sweep");
assert_eq!(fault.namespace_raw, namespace.inner());
assert_eq!(fault.op, 1);
assert_eq!(fault.operation, None);
assert_eq!(persistence.checkpoint_op(), 0);
}
#[compio::test]
async fn pre_resolved_balanced_write_keeps_its_namespace_and_admission_frontier() {
const CLIENT: u128 = 1;
const ADMISSION_FRONTIER: u64 = 10;
const LATER_FRONTIER: u64 = 20;
let bus = SpyBus::default();
let mut shard = test_shard(&bus, 0, 1, 1);
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
let metadata = shard.plane.metadata();
metadata.mux_stm.users().ensure_root_user("iggy", "hash");
metadata
.mux_stm
.update(prepare_message(
Operation::CreateStream,
CLIENT,
1,
&CreateStreamRequest {
name: WireName::new("stream").unwrap(),
options: WireOptions::empty(),
}
.to_bytes(),
))
.unwrap();
metadata
.mux_stm
.update(prepare_message(
Operation::CreateTopicWithAssignments,
CLIENT,
2,
&CreateTopicWithAssignmentsRequest {
request: CreateTopicRequest {
stream_id: WireIdentifier::numeric(0),
partitions_count: 2,
name: WireName::new("topic").unwrap(),
options: WireOptions::empty(),
},
derived_options: WireOptions::empty(),
partitions: vec![
CreatedPartitionAssignment {
partition_id: 0,
consensus_group_id: 1,
},
CreatedPartitionAssignment {
partition_id: 1,
consensus_group_id: 2,
},
],
created_view: 0,
}
.to_bytes(),
))
.unwrap();
metadata.applied_frontier().advance(ADMISSION_FRONTIER);
let wire_header = SendMessagesHeader {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
partitioning: WirePartitioning::Balanced,
messages_count: 1,
}
.to_bytes();
let mut body = Vec::new();
body.extend_from_slice(&u32::try_from(wire_header.len()).unwrap().to_le_bytes());
body.extend_from_slice(&wire_header);
let namespace =
resolve_partition_request_namespace(&shard, Operation::SendMessages, &body, CLIENT)
.unwrap();
let delayed_shard = Rc::clone(&shard);
let updater = compio::runtime::spawn(async move {
compio::time::sleep(std::time::Duration::from_millis(1)).await;
delayed_shard
.plane
.metadata()
.applied_frontier()
.advance(LATER_FRONTIER);
for partition_id in 0..2 {
let namespace = delayed_shard
.plane
.metadata()
.mux_stm
.streams()
.namespace_from_partition(
&WireIdentifier::numeric(0),
&WireIdentifier::numeric(0),
partition_id,
)
.unwrap();
delayed_shard
.shards_table()
.insert(namespace, PartitionLocation::new(ShardId::new(0), 0));
}
});
dispatch_partition_request(
&shard,
request_message(Operation::SendMessages, CLIENT, 1, 1, &body),
CLIENT,
1,
CLIENT,
Some(DEFAULT_ROOT_USER_ID),
None,
Some(namespace),
)
.await;
let _ = updater.await;
let ShardFrame::Consensus {
message: MessageBag::Request(request),
..
} = owner_inbox.try_recv().unwrap()
else {
panic!("the dispatcher must submit one partition write");
};
assert_eq!(
request.header().group,
namespace,
"Balanced must advance once for the gate and write"
);
assert_eq!(
request.header().metadata_watermark,
ADMISSION_FRONTIER,
"routing waits cannot promote an old binding past Logout"
);
assert_eq!(metadata.applied_frontier().get(), LATER_FRONTIER);
}
#[compio::test]
async fn given_invalid_partition_writes_when_resolving_should_preserve_offset_error_codes() {
const VSR_CLIENT: u128 = 1;
let bus = SpyBus::default();
let shard = Rc::new(test_shard(&bus, 0, 1, 1));
// Stream 0 / topic 0 / partition 0 committed straight into the STM, so a
// group op resolves its namespace and reaches the group fence, which
// runs before the routable wait.
let md = shard.plane.metadata();
md.mux_stm.users().ensure_root_user("iggy", "hash");
let create_stream = CreateStreamRequest {
name: WireName::new("stream").unwrap(),
options: WireOptions::empty(),
};
md.mux_stm
.update(prepare_message(
Operation::CreateStream,
VSR_CLIENT,
1,
&create_stream.to_bytes(),
))
.unwrap();
let create_topic = CreateTopicWithAssignmentsRequest {
request: CreateTopicRequest {
stream_id: WireIdentifier::numeric(0),
partitions_count: 1,
name: WireName::new("topic").unwrap(),
options: WireOptions::empty(),
},
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
partition_id: 0,
consensus_group_id: 1,
}],
created_view: 0,
};
md.mux_stm
.update(prepare_message(
Operation::CreateTopicWithAssignments,
VSR_CLIENT,
2,
&create_topic.to_bytes(),
))
.unwrap();
let offset_body = |operation: Operation,
consumer: WireConsumer,
stream_id: WireIdentifier,
topic_id: WireIdentifier| {
if operation == Operation::StoreConsumerOffset {
StoreConsumerOffsetRequest {
consumer,
stream_id,
topic_id,
partition_id: Some(0),
offset: 0,
ack: iggy_binary_protocol::AckLevel::Quorum,
}
.to_bytes()
} else {
DeleteConsumerOffsetRequest {
consumer,
stream_id,
topic_id,
partition_id: Some(0),
ack: iggy_binary_protocol::AckLevel::Quorum,
}
.to_bytes()
}
};
let mut cases = Vec::new();
for operation in [
Operation::StoreConsumerOffset,
Operation::DeleteConsumerOffset,
] {
cases.push((operation, vec![1], IggyError::InvalidCommand));
cases.push((
operation,
offset_body(
operation,
WireConsumer::consumer(WireIdentifier::Numeric(1)),
WireIdentifier::Numeric(404),
WireIdentifier::Numeric(1),
)
.to_vec(),
IggyError::StreamIdNotFound(Identifier::numeric(404).unwrap()),
));
// The group fence: an unknown group on a known topic answers the
// group's own not-found code, numeric and named, not a bare
// ResourceNotFound.
cases.push((
operation,
offset_body(
operation,
WireConsumer::consumer_group(WireIdentifier::Numeric(999)),
WireIdentifier::Numeric(0),
WireIdentifier::Numeric(0),
)
.to_vec(),
IggyError::ConsumerGroupIdNotFound(
Identifier::numeric(999).unwrap(),
Identifier::numeric(0).unwrap(),
),
));
cases.push((
operation,
offset_body(
operation,
WireConsumer::consumer_group(WireIdentifier::named("ghost").unwrap()),
WireIdentifier::Numeric(0),
WireIdentifier::Numeric(0),
)
.to_vec(),
IggyError::ConsumerGroupNameNotFound(
"ghost".to_owned(),
Identifier::numeric(0).unwrap(),
),
));
}
cases.push((
Operation::SendMessages,
vec![1],
IggyError::ResourceNotFound(String::new()),
));
for (index, (operation, body, expected)) in cases.into_iter().enumerate() {
let request = request_message(operation, 1, 1, index as u64 + 1, &body);
dispatch_partition_request(
&shard,
request,
1,
1,
91,
Some(DEFAULT_ROOT_USER_ID),
None,
None,
)
.await;
let replies = bus.client_replies.borrow();
assert_eq!(
replies.len(),
index + 1,
"{operation:?} must reply exactly once"
);
let (_, frame) = replies.last().unwrap();
let start = std::mem::offset_of!(ReplyHeader, status);
let status = u32::from_le_bytes(frame[start..start + 4].try_into().unwrap());
assert_eq!(status, expected.as_code(), "{operation:?}");
}
}
/// An undecodable request body is a PERMANENT client error, so every read
/// on this path must answer a nonzero status. The fail-fast shapes these
/// used to borrow all decode as success: the 16-byte empty poll reads as a
/// 0-message poll, an empty consumer-offset body as "no offset stored",
/// and a status-0 `delete_segments` ack as a completed trim. A client on a
/// skewed protocol would loop on those forever with no error to surface.
#[compio::test]
async fn undecodable_read_bodies_must_deny_typed_not_fabricate_success() {
const TRANSPORT: u128 = 91;
const VSR_CLIENT: u128 = 1;
const SESSION: u64 = 1;
const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
// Shorter than any of the three request encodings, so each decoder
// fails on length before it can interpret a field.
const TRUNCATED_BODY: &[u8] = &[0x01];
let bus = SpyBus::default();
let shard = Rc::new(test_shard(&bus, 0, 1, 1));
let poll = request_message(
Operation::NonReplicated,
VSR_CLIENT,
SESSION,
1,
TRUNCATED_BODY,
);
handle_poll_messages(
&shard,
TRANSPORT,
&poll,
Some(DEFAULT_ROOT_USER_ID),
poll.header().client,
None,
)
.await;
let offset = request_message(
Operation::NonReplicated,
VSR_CLIENT,
SESSION,
2,
TRUNCATED_BODY,
);
handle_get_consumer_offset(&shard, TRANSPORT, &offset, Some(DEFAULT_ROOT_USER_ID)).await;
let delete = request_message(
Operation::DeleteSegments,
VSR_CLIENT,
SESSION,
3,
TRUNCATED_BODY,
);
handle_delete_segments_request(&shard, TRANSPORT, Some((VSR_CLIENT, SESSION)), &delete)
.await;
let replies = bus.client_replies.borrow();
assert_eq!(
replies.len(),
3,
"one deny frame per undecodable request, none of them silent"
);
for (label, (client, frame)) in ["poll_messages", "get_consumer_offset", "delete_segments"]
.into_iter()
.zip(replies.iter())
{
assert_eq!(*client, TRANSPORT, "{label} deny must target the transport");
let status =
u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap());
assert_eq!(
status,
IggyError::InvalidCommand.as_code(),
"{label} with an undecodable body must carry a nonzero status"
);
}
}
/// A WELL-FORMED read against a stream that does not exist is a permanent
/// addressing error, and it reaches the server through a public SDK method
/// (a typo'd or deleted stream). Both fail-fast shapes decode as success:
/// the 16-byte empty poll as a 0-message read, and an empty
/// consumer-offset body as `None`, which is indistinguishable from a fresh
/// consumer -- so the caller resumes from its configured default and
/// silently reprocesses. The same lookup 404s over REST.
#[compio::test]
async fn unresolved_read_targets_must_deny_typed_not_read_as_empty() {
const TRANSPORT: u128 = 91;
const VSR_CLIENT: u128 = 1;
const SESSION: u64 = 1;
const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
const MISSING_STREAM: u32 = 404;
let bus = SpyBus::default();
let shard = Rc::new(test_shard(&bus, 0, 1, 1));
let poll_body = PollMessagesRequest {
consumer: WireConsumer::consumer(WireIdentifier::Numeric(1)),
stream_id: WireIdentifier::Numeric(MISSING_STREAM),
topic_id: WireIdentifier::Numeric(1),
partition_id: Some(1),
strategy: WirePollingStrategy::offset(0),
count: 10,
auto_commit: false,
}
.to_bytes();
let poll = request_message(Operation::NonReplicated, VSR_CLIENT, SESSION, 1, &poll_body);
handle_poll_messages(
&shard,
TRANSPORT,
&poll,
Some(DEFAULT_ROOT_USER_ID),
poll.header().client,
None,
)
.await;
let offset_body = GetConsumerOffsetRequest {
consumer: WireConsumer::consumer(WireIdentifier::Numeric(1)),
stream_id: WireIdentifier::Numeric(MISSING_STREAM),
topic_id: WireIdentifier::Numeric(1),
partition_id: Some(1),
}
.to_bytes();
let offset = request_message(
Operation::NonReplicated,
VSR_CLIENT,
SESSION,
2,
&offset_body,
);
handle_get_consumer_offset(&shard, TRANSPORT, &offset, Some(DEFAULT_ROOT_USER_ID)).await;
let replies = bus.client_replies.borrow();
assert_eq!(
replies.len(),
2,
"one deny frame per unresolved read, none of them silent"
);
for (label, (client, frame)) in ["poll_messages", "get_consumer_offset"]
.into_iter()
.zip(replies.iter())
{
assert_eq!(*client, TRANSPORT, "{label} deny must target the transport");
let status =
u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap());
assert_eq!(
status,
IggyError::StreamIdNotFound(
Identifier::numeric(MISSING_STREAM).expect("a nonzero numeric identifier")
)
.as_code(),
"{label} against a missing stream must carry the not-found status"
);
}
}
#[compio::test]
async fn given_unroutable_partition_when_polling_should_report_not_accepted() {
let bus = SpyBus::default();
let mut shard = test_shard(&bus, 0, 1, 1);
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
let namespace = IggyNamespace::new(1, 1, 0);
// The owner has a live inbox, but no route identifies it for this partition.
assert_eq!(
poll_and_expect_error(&shard, namespace).await,
IggyError::TransientNotAccepted
);
assert!(
owner_inbox.try_recv().is_err(),
"a poll rejected during routing must never reach the owner"
);
}
#[compio::test]
async fn given_missing_owner_sender_when_polling_should_report_not_accepted() {
let bus = SpyBus::default();
let shard = Rc::new(test_shard(&bus, 0, 1, 1));
let namespace = IggyNamespace::new(1, 1, 0);
shard
.shards_table()
.insert(namespace, PartitionLocation::new(ShardId::new(0), 0));
// A route exists, but no sender was attached to submit work to its owner.
assert_eq!(
poll_and_expect_error(&shard, namespace).await,
IggyError::TransientNotAccepted
);
}
#[compio::test]
async fn given_full_owner_inbox_when_polling_should_report_not_accepted() {
let bus = SpyBus::default();
let mut shard = test_shard(&bus, 0, 1, 1);
let inbox_capacity = 1;
let (sender, owner_inbox, _owner_replies) = shard_channel(0, inbox_capacity, 1);
// Occupy the only slot before polling; no task drains this inbox.
sender
.try_send(ShardFrame::lifecycle(LifecycleFrame::MetadataCommitTick))
.unwrap();
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
let namespace = IggyNamespace::new(1, 1, 0);
shard
.shards_table()
.insert(namespace, PartitionLocation::new(ShardId::new(0), 0));
assert_eq!(
poll_and_expect_error(&shard, namespace).await,
IggyError::TransientNotAccepted
);
assert!(matches!(
owner_inbox.try_recv().unwrap(),
ShardFrame::Lifecycle(LifecycleFrame::MetadataCommitTick)
));
assert!(
owner_inbox.try_recv().is_err(),
"only the filler frame may be queued; the poll was never submitted"
);
}
#[compio::test]
async fn given_closed_owner_inbox_when_polling_should_report_not_accepted() {
let bus = SpyBus::default();
let mut shard = test_shard(&bus, 0, 1, 1);
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
// Closing the destination leaves its sender attached, so submission fails.
drop(owner_inbox);
let shard = Rc::new(shard);
let namespace = IggyNamespace::new(1, 1, 0);
shard
.shards_table()
.insert(namespace, PartitionLocation::new(ShardId::new(0), 0));
assert_eq!(
poll_and_expect_error(&shard, namespace).await,
IggyError::TransientNotAccepted
);
}
#[compio::test]
async fn given_submitted_poll_when_owner_drops_reply_should_report_unknown_acceptance() {
let bus = SpyBus::default();
let mut shard = test_shard(&bus, 0, 1, 1);
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
let namespace = IggyNamespace::new(1, 1, 0);
shard
.shards_table()
.insert(namespace, PartitionLocation::new(ShardId::new(0), 0));
// Receive the request before dropping its reply. Submission succeeded,
// so the caller cannot infer whether the owner accepted any progress.
let owner_task = compio::runtime::spawn(async move {
let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead { reply, .. }) =
owner_inbox.recv().await.unwrap()
else {
panic!("the poll must have reached the owner");
};
drop(reply);
});
assert_eq!(
poll_and_expect_error(&shard, namespace).await,
IggyError::ShardCommunicationError
);
owner_task.await.expect("the owner task must finish");
}
#[compio::test]
async fn given_pending_poll_when_owner_reply_times_out_should_report_unknown_acceptance() {
const TRANSPORT_CLIENT_ID: u128 = 91;
const VSR_CLIENT_ID: u128 = 1;
let bus = SpyBus::default();
// Expire the reply timeout deterministically, without a wall clock wait.
bus.instant_timers.set(true);
let mut shard = test_shard(&bus, 0, 1, 1);
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
route_one_partition(&shard);
let poll_body = PollMessagesRequest {
consumer: WireConsumer::consumer(WireIdentifier::numeric(1)),
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
partition_id: Some(0),
strategy: WirePollingStrategy::offset(0),
count: 10,
auto_commit: true,
}
.to_bytes();
let request = request_message(Operation::NonReplicated, VSR_CLIENT_ID, 1, 1, &poll_body);
// Keep the owner request queued so its live reply sender forces the
// timeout path. The owner can still receive the request after that timeout.
handle_poll_messages(
&shard,
TRANSPORT_CLIENT_ID,
&request,
Some(DEFAULT_ROOT_USER_ID),
request.header().client,
None,
)
.await;
// The client receives one error header, with no successful empty poll body.
let replies = bus.client_replies.borrow();
assert_eq!(replies.len(), 1);
let (client_id, frame) = &replies[0];
assert_eq!(*client_id, TRANSPORT_CLIENT_ID);
assert_eq!(frame.len(), std::mem::size_of::<ReplyHeader>());
let status_start = std::mem::offset_of!(ReplyHeader, status);
let status = u32::from_le_bytes(frame[status_start..status_start + 4].try_into().unwrap());
assert_eq!(status, IggyError::ShardCommunicationError.as_code());
drop(replies);
// Timing out abandons the reply receiver; a late owner reply cannot
// replace the communication error or produce a second client response.
let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead { reply, .. }) =
owner_inbox.try_recv().unwrap()
else {
panic!("the timed out poll must remain queued for its owner");
};
assert!(
reply
.try_send(PartitionReadReply::Rejected(
IggyError::TransientNotAccepted
))
.is_err(),
"the timeout must have dropped the caller's reply receiver"
);
assert_eq!(bus.client_replies.borrow().len(), 1);
}
/// An empty body reads as "no stored offset", and a consumer that reads it
/// resets its position. A read with no reply must deny instead.
#[compio::test]
async fn given_pending_offset_read_when_owner_reply_times_out_should_deny_not_accepted() {
const TRANSPORT_CLIENT_ID: u128 = 91;
let bus = SpyBus::default();
bus.instant_timers.set(true);
let mut shard = test_shard(&bus, 0, 1, 1);
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
route_one_partition(&shard);
let offset_body = GetConsumerOffsetRequest {
consumer: WireConsumer::consumer(WireIdentifier::numeric(1)),
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
partition_id: Some(0),
}
.to_bytes();
let request = request_message(Operation::NonReplicated, 1, 1, 1, &offset_body);
handle_get_consumer_offset(
&shard,
TRANSPORT_CLIENT_ID,
&request,
Some(DEFAULT_ROOT_USER_ID),
)
.await;
let replies = bus.client_replies.borrow();
assert_eq!(replies.len(), 1);
let (client_id, frame) = &replies[0];
assert_eq!(*client_id, TRANSPORT_CLIENT_ID);
assert_eq!(frame.len(), std::mem::size_of::<ReplyHeader>());
let status_start = std::mem::offset_of!(ReplyHeader, status);
let status = u32::from_le_bytes(frame[status_start..status_start + 4].try_into().unwrap());
assert_eq!(status, IggyError::TransientNotAccepted.as_code());
assert!(
owner_inbox.try_recv().is_ok(),
"the read must have reached its owner before the timeout"
);
}
/// Only the owner's own "no stored offset" may go back as the empty body.
/// An owner that does not hold the partition says nothing about the offset.
#[compio::test]
async fn given_owner_reply_when_reading_offset_should_send_empty_body_only_for_no_offset() {
const TRANSPORT_CLIENT_ID: u128 = 91;
for (owner_reply, expected_status) in [
(
PartitionReadReply::NotFound,
IggyError::TransientNotAccepted.as_code(),
),
(
PartitionReadReply::ConsumerOffset {
stored: None,
current_offset: 0,
},
0,
),
] {
let bus = SpyBus::default();
let mut shard = test_shard(&bus, 0, 1, 1);
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
route_one_partition(&shard);
let owner_task = compio::runtime::spawn(async move {
let ShardFrame::Lifecycle(LifecycleFrame::PartitionRead { reply, .. }) =
owner_inbox.recv().await.unwrap()
else {
panic!("the read must have reached the owner");
};
reply.try_send(owner_reply).unwrap();
});
let offset_body = GetConsumerOffsetRequest {
consumer: WireConsumer::consumer(WireIdentifier::numeric(1)),
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
partition_id: Some(0),
}
.to_bytes();
let request = request_message(Operation::NonReplicated, 1, 1, 1, &offset_body);
handle_get_consumer_offset(
&shard,
TRANSPORT_CLIENT_ID,
&request,
Some(DEFAULT_ROOT_USER_ID),
)
.await;
owner_task.await.expect("the owner task must finish");
let replies = bus.client_replies.borrow();
assert_eq!(replies.len(), 1);
let (_, frame) = &replies[0];
assert_eq!(frame.len(), std::mem::size_of::<ReplyHeader>());
let status_start = std::mem::offset_of!(ReplyHeader, status);
let status =
u32::from_le_bytes(frame[status_start..status_start + 4].try_into().unwrap());
assert_eq!(status, expected_status);
}
}
/// Creates one stream with one single-partition topic and routes the
/// partition to shard 0, so authorization and resolution let a read reach
/// the owner inbox.
fn route_one_partition(shard: &TestShard) {
const VSR_CLIENT_ID: u128 = 1;
let metadata = shard.plane.metadata();
metadata.mux_stm.users().ensure_root_user("iggy", "hash");
metadata
.mux_stm
.update(prepare_message(
Operation::CreateStream,
VSR_CLIENT_ID,
1,
&CreateStreamRequest {
name: WireName::new("stream").unwrap(),
options: WireOptions::empty(),
}
.to_bytes(),
))
.unwrap();
metadata
.mux_stm
.update(prepare_message(
Operation::CreateTopicWithAssignments,
VSR_CLIENT_ID,
2,
&CreateTopicWithAssignmentsRequest {
request: CreateTopicRequest {
stream_id: WireIdentifier::numeric(0),
partitions_count: 1,
name: WireName::new("topic").unwrap(),
options: WireOptions::empty(),
},
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
partition_id: 0,
consensus_group_id: 1,
}],
created_view: 0,
}
.to_bytes(),
))
.unwrap();
let namespace = metadata
.mux_stm
.streams()
.namespace_from_partition(&WireIdentifier::numeric(0), &WireIdentifier::numeric(0), 0)
.unwrap();
shard
.shards_table()
.insert(namespace, PartitionLocation::new(ShardId::new(0), 0));
}
/// A partition write whose routable wait exhausts (namespace committed,
/// but no reconciler ever seeds this shard's routing row -- the state a
/// teardown/rematerialise churn leaves behind) must answer a nonzero
/// retriable status. A status-0 empty reply is a fabricated success: the
/// SDK grades the send as acknowledged while zero bytes reached any
/// partition.
#[compio::test]
async fn unroutable_partition_send_must_reply_transient_error_not_success() {
const VSR_CLIENT: u128 = 1;
const SESSION: u64 = 1;
const TRANSPORT: u128 = 91;
const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
let bus = SpyBus::default();
let metadata = IggyMetadata::new(None, None, None, None, TestMux::default(), None);
let partitions = IggyPartitions::new(
ShardId::new(0),
PartitionsConfig {
messages_required_to_save: 1,
size_of_messages_required_to_save: iggy_common::IggyByteSize::from(1024_u64),
validate_checksum: true,
segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
preallocate_segments: false,
encryptor: None,
path_layout: PartitionPathLayout::default(),
},
);
let shard = Rc::new(TestShard::without_inbox(
ShardIdentity::new(0, "unroutable-send-test".to_string()),
bus.clone(),
metadata,
partitions,
PapayaShardsTable::new(),
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1), bus.clone()),
));
let md = shard.plane.metadata();
// Committed stream 0 / topic 0 / partition 0, applied straight into
// the STM: the namespace resolves and root authorizes, but no
// reconciler runs, so the shards table never gains a routing row and
// the routable wait exhausts its budget.
md.mux_stm.users().ensure_root_user("iggy", "hash");
let create_stream = CreateStreamRequest {
name: WireName::new("stream").unwrap(),
options: WireOptions::empty(),
};
md.mux_stm
.update(prepare_message(
Operation::CreateStream,
VSR_CLIENT,
1,
&create_stream.to_bytes(),
))
.unwrap();
let create_topic = CreateTopicWithAssignmentsRequest {
request: CreateTopicRequest {
stream_id: WireIdentifier::numeric(0),
partitions_count: 1,
name: WireName::new("topic").unwrap(),
options: WireOptions::empty(),
},
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
partition_id: 0,
consensus_group_id: 1,
}],
created_view: 0,
};
md.mux_stm
.update(prepare_message(
Operation::CreateTopicWithAssignments,
VSR_CLIENT,
2,
&create_topic.to_bytes(),
))
.unwrap();
assert!(
md.mux_stm
.streams()
.namespace_from_partition(
&WireIdentifier::numeric(0),
&WireIdentifier::numeric(0),
0
)
.is_some(),
"seeded namespace must resolve, or the unresolved-namespace path \
would reply instead of the exhausted routable wait"
);
let send_header = SendMessagesHeader {
stream_id: WireIdentifier::numeric(0),
topic_id: WireIdentifier::numeric(0),
partitioning: WirePartitioning::PartitionId(0),
messages_count: 1,
};
let send_metadata = send_header.to_bytes();
let mut send_body = Vec::with_capacity(4 + send_metadata.len());
send_body.extend_from_slice(&u32::try_from(send_metadata.len()).unwrap().to_le_bytes());
send_body.extend_from_slice(&send_metadata);
let request = request_message(Operation::SendMessages, VSR_CLIENT, SESSION, 1, &send_body);
dispatch_partition_request(
&shard,
request,
VSR_CLIENT,
SESSION,
TRANSPORT,
Some(DEFAULT_ROOT_USER_ID),
None,
None,
)
.await;
let replies = bus.client_replies.borrow();
assert_eq!(replies.len(), 1, "one reply frame for the failed send");
let (client, frame) = &replies[0];
assert_eq!(*client, TRANSPORT, "reply must target the transport id");
let status =
u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap());
assert_eq!(
status,
IggyError::TransientNotAccepted.as_code(),
"an unroutable partition write must surface the retriable \
transient status; status 0 with an empty body grades as a \
successfully acknowledged send"
);
}
/// A send that reaches the owning shard while its namespace is
/// tombstoned (the teardown fence a delete/recreate churn sets before
/// the disk delete) must answer the retriable transient status. The
/// partition plane's own tombstone guard drops the frame without any
/// reply; the transports decode replies in lockstep, so that silence
/// wedges the connection until the SDK's response read-timeout.
#[compio::test]
async fn tombstoned_partition_send_must_reply_transient_error_not_silence() {
const TRANSPORT: u128 = 91;
const SESSION: u64 = 1;
const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
let bus = SpyBus::default();
let metadata = IggyMetadata::new(None, None, None, None, TestMux::default(), None);
let partitions = IggyPartitions::new(
ShardId::new(0),
PartitionsConfig {
messages_required_to_save: 1,
size_of_messages_required_to_save: iggy_common::IggyByteSize::from(1024_u64),
validate_checksum: true,
segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
preallocate_segments: false,
encryptor: None,
path_layout: PartitionPathLayout::default(),
},
);
let shard = Rc::new(TestShard::without_inbox(
ShardIdentity::new(0, "tombstoned-send-test".to_string()),
bus.clone(),
metadata,
partitions,
PapayaShardsTable::new(),
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1), bus.clone()),
));
let namespace = IggyNamespace::new(0, 0, 0);
shard.plane.partitions().tombstone(namespace);
let request = request_message(Operation::SendMessages, TRANSPORT, SESSION, 1, &[])
.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
*new_header = header;
new_header.group = namespace.inner();
});
shard.on_message(MessageBag::Request(request)).await;
let replies = bus.client_replies.borrow();
assert_eq!(
replies.len(),
1,
"a send into a tombstoned namespace must produce a reply frame; \
silence wedges the connection's lockstep decode"
);
let (client, frame) = &replies[0];
assert_eq!(*client, TRANSPORT, "reply must target the request's client");
let status =
u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap());
assert_eq!(
status,
IggyError::TransientNotAccepted.as_code(),
"a tombstoned-namespace send must surface the retriable transient \
status so the SDK replays it after the partition rematerialises"
);
}
/// A request no plane claims must be denied at the shard boundary, not left
/// to the chain terminator and the client's read-timeout. `DeleteSegments`
/// is the live example: dispatch resolves it to `TruncatePartition`, so it
/// satisfies neither plane predicate. The deny is permanent -- a transient
/// code is what the SDK replays.
#[compio::test]
async fn unroutable_operation_must_reply_denied_not_silence() {
const TRANSPORT: u128 = 91;
const SESSION: u64 = 1;
const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
let bus = SpyBus::default();
let metadata = IggyMetadata::new(None, None, None, None, TestMux::default(), None);
let partitions = IggyPartitions::new(
ShardId::new(0),
PartitionsConfig {
messages_required_to_save: 1,
size_of_messages_required_to_save: iggy_common::IggyByteSize::from(1024_u64),
validate_checksum: true,
segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
preallocate_segments: false,
encryptor: None,
path_layout: PartitionPathLayout::default(),
},
);
let shard = Rc::new(TestShard::without_inbox(
ShardIdentity::new(0, "unroutable-operation-test".to_string()),
bus.clone(),
metadata,
partitions,
PapayaShardsTable::new(),
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1), bus.clone()),
));
let request = request_message(Operation::DeleteSegments, TRANSPORT, SESSION, 1, &[]);
shard.on_message(MessageBag::Request(request)).await;
let replies = bus.client_replies.borrow();
assert_eq!(
replies.len(),
1,
"a request no plane claims must be answered, not absorbed in silence"
);
let (client, frame) = &replies[0];
assert_eq!(*client, TRANSPORT, "reply must target the request's client");
let status =
u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap());
assert_eq!(
status,
IggyError::InvalidCommand.as_code(),
"the deny must be permanent; a transient status has the SDK replay it"
);
}
/// The replicated halves of the same hole. A prepare or an ack whose
/// operation no plane claims reaches the same terminator, and there is no
/// client to answer, so the counter and the log are the only record. Dropping
/// it is the only option: nothing journals or acks an operation no plane
/// owns.
#[compio::test]
async fn unroutable_replicated_frames_must_be_dropped_and_counted() {
const TRANSPORT: u128 = 91;
let bus = SpyBus::default();
let shard = Rc::new(test_shard(&bus, 0, 1, 1));
let unroutable_drops = || {
shard
.metrics()
.frame_drop_count(frame_drop_variant::CONSENSUS, frame_drop_reason::UNROUTABLE)
};
let prepare = prepare_message(Operation::DeleteSegments, TRANSPORT, 1, &[]);
shard.on_message(MessageBag::Prepare(prepare)).await;
assert_eq!(
unroutable_drops(),
1,
"a prepare no plane claims must be counted, not absorbed by the chain"
);
shard
.on_message(MessageBag::PrepareOk(prepare_ok_message(
Operation::DeleteSegments,
1,
)))
.await;
assert_eq!(
unroutable_drops(),
2,
"an ack no plane claims must be counted, not absorbed by the chain"
);
assert!(
bus.client_replies.borrow().is_empty(),
"replicated frames answer nobody on this node"
);
}
/// Bare `PrepareOk` for the routing guard: only `operation`, `group` and `op`
/// are read before the plane chain, so the rest stays zeroed.
fn prepare_ok_message(operation: Operation, op: u64) -> Message<PrepareOkHeader> {
let header_size = size_of::<PrepareOkHeader>();
let mut msg = Message::<PrepareOkHeader>::new(header_size);
let header = bytemuck::checked::try_from_bytes_mut::<PrepareOkHeader>(
&mut msg.as_mut_slice()[..header_size],
)
.expect("zeroed bytes form a valid PrepareOkHeader");
header.command = Command::PrepareOk;
header.size = u32::try_from(header_size).expect("ack size fits u32");
header.operation = operation;
header.op = op;
header.group = server_common::sharding::METADATA_GROUP;
msg
}
/// A send parked for a namespace that is torn down before materialising
/// (create -> delete before the reconciler's `InsertOwned`) is discarded
/// on `ConfirmRemove`. The discard must stage the same retriable
/// transient deny toward the client -- through the shard's own pump as a
/// `ForwardClientSend` -- instead of dropping the request without any
/// reply.
#[compio::test]
async fn discarded_parked_partition_send_must_reply_transient_error_not_silence() {
const TRANSPORT: u128 = 91;
const SESSION: u64 = 1;
const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
// This test does not dispatch disk polls, so keep that lane minimal.
const POLL_COMPLETION_CAPACITY: usize = 1;
let bus = SpyBus::default();
let metadata = IggyMetadata::new(None, None, None, None, TestMux::default(), None);
let partitions = IggyPartitions::new(
ShardId::new(0),
PartitionsConfig {
messages_required_to_save: 1,
size_of_messages_required_to_save: iggy_common::IggyByteSize::from(1024_u64),
validate_checksum: true,
segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
preallocate_segments: false,
encryptor: None,
path_layout: PartitionPathLayout::default(),
},
);
// Real sender ring so the staged deny is observable: the test holds
// the receiving ends of this shard's own lanes. The deny is a client
// Reply forward, so it lands on the REPLY lane.
let (sender, _pump_rx, reply_rx) = shard_channel(0, 16, 16);
let (_inbox_tx, inbox_rx, reply_inbox_rx) = shard_channel(0, 1, 1);
let shard = TestShard::new(
ShardIdentity::new(0, "discarded-parked-send-test".to_string()),
bus.clone(),
Rc::new(NoopHost),
metadata,
partitions,
vec![sender],
inbox_rx,
reply_inbox_rx,
POLL_COMPLETION_CAPACITY,
None,
PapayaShardsTable::new(),
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1), bus.clone()),
None,
ShardMetrics::for_shard(),
)
.expect("single-sender ring is canonically ordered");
let namespace = IggyNamespace::new(0, 0, 0);
let request = request_message(Operation::SendMessages, TRANSPORT, SESSION, 1, &[])
.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
*new_header = header;
new_header.group = namespace.inner();
});
// Namespace neither materialised nor tombstoned: the frame parks.
shard.on_message(MessageBag::Request(request)).await;
shard.enqueue_reconcile_op(ReconcileOp::ConfirmRemove { namespace });
shard.apply_reconcile_ops();
let mut denies = Vec::new();
while let Ok(frame) = reply_rx.try_recv() {
if let ShardFrame::Lifecycle(LifecycleFrame::ForwardClientSend { client_id, msg }) =
frame
{
denies.push((client_id, msg.into_contiguous().as_slice().to_vec()));
}
}
assert_eq!(
denies.len(),
1,
"discarding a parked client request must stage exactly one deny \
reply; silence wedges the connection's lockstep decode"
);
let (client, frame) = &denies[0];
assert_eq!(*client, TRANSPORT, "deny must target the request's client");
let status =
u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap());
assert_eq!(
status,
IggyError::TransientNotAccepted.as_code(),
"a discarded parked send must surface the retriable transient \
status so the SDK replays it instead of timing out"
);
}
fn create_segment_delete_topic(shard: &TestShard) -> (IggyNamespace, u64) {
let metadata = shard.plane.metadata();
metadata.mux_stm.users().ensure_root_user("iggy", "hash");
for prepare in [
prepare_message(
Operation::CreateStream,
1,
1,
&CreateStreamRequest {
name: WireName::new("stream").unwrap(),
options: WireOptions::empty(),
}
.to_bytes(),
),
prepare_message(
Operation::CreateTopicWithAssignments,
1,
2,
&CreateTopicWithAssignmentsRequest {
request: CreateTopicRequest {
stream_id: WireIdentifier::numeric(0),
partitions_count: 1,
name: WireName::new("topic").unwrap(),
options: WireOptions::empty(),
},
derived_options: WireOptions::empty(),
partitions: vec![CreatedPartitionAssignment {
partition_id: 0,
consensus_group_id: 1,
}],
created_view: 0,
}
.to_bytes(),
),
] {
assert_eq!(metadata.mux_stm.update(prepare).unwrap().code, 0);
}
let namespace = IggyNamespace::new(0, 0, 0);
shard
.shards_table()
.insert(namespace, PartitionLocation::new(ShardId::new(0), 0));
let created_revision = metadata
.mux_stm
.streams()
.created_revision_for_namespace(namespace)
.unwrap();
(namespace, created_revision)
}
/// Attempt an already resolved poll and require an error, including when no
/// owner reply arrives. Bypassing metadata resolution isolates owner routing
/// and reply handling from authentication and request decoding.
async fn poll_and_expect_error(shard: &Rc<TestShard>, namespace: IggyNamespace) -> IggyError {
let vsr_client_id = 1;
let transport_client_id = 91;
let consumer_id = 1;
let partition_id = 0;
let request = request_message(Operation::NonReplicated, vsr_client_id, 1, 1, &[]);
let resolved_poll = (
namespace,
partition_id,
PollingConsumer::Consumer(consumer_id, usize::try_from(partition_id).unwrap()),
PollingArgs {
strategy: PollingStrategy::offset(0),
count: 1,
auto_commit: true,
},
);
match read_polled_messages(
shard,
transport_client_id,
&request,
resolved_poll,
None,
transport_client_id,
)
.await
{
Err(ReadPolledMessagesError::Rejected(error)) => error,
Err(ReadPolledMessagesError::Fallback(_)) => {
panic!("poll must not return an empty reply")
}
Ok(_) => panic!("poll must not succeed without an owner reply"),
}
}
}