This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch server_perf_cut_extra_work
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/server_perf_cut_extra_work by
this push:
new e52aeaad0 fix
e52aeaad0 is described below
commit e52aeaad0f3fd0acb11daa62277519c0336ea8a6
Author: spetz <[email protected]>
AuthorDate: Mon Sep 14 10:40:08 2026 +0200
fix
---
core/journal/src/partition_journal.rs | 86 ++++--
core/partitions/src/iggy_partition.rs | 134 ++++++++-
core/partitions/src/persistence.rs | 25 +-
core/partitions/src/poll_plan.rs | 241 ++++++++++++---
core/server/src/http.rs | 13 +-
core/server/src/http/forward.rs | 322 +++++++++++++++++----
core/server/src/http/state.rs | 2 +-
core/simulator/src/storage/tests.rs | 6 +-
.../org/apache/iggy/serde/BytesSerializer.java | 42 ++-
.../iggy/serde/MessagesBatchWireFormatTest.java | 30 +-
10 files changed, 736 insertions(+), 165 deletions(-)
diff --git a/core/journal/src/partition_journal.rs
b/core/journal/src/partition_journal.rs
index 47f18fb3f..9f03259fb 100644
--- a/core/journal/src/partition_journal.rs
+++ b/core/journal/src/partition_journal.rs
@@ -863,7 +863,7 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
}
async fn retain_segment_inodes(
- &self,
+ &mut self,
records: &[(u64, StoredPrepare, usize)],
) -> io::Result<()> {
if self.state.segment_storage.is_some() {
@@ -876,6 +876,11 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
&& retained_segments.insert((reference.generation,
reference.start_offset))
{
let retained = reference.path(&self.directory);
+ // A failed unlink leaves its path queued for a retry. This
+ // append is adopting that very link, so the retry would delete
+ // a segment the new generation references. Claiming it back is
+ // what makes reusing the existing inode safe.
+ self.obsolete.retain(|queued| queued != &retained);
if !self.storage.exists(&retained).await? {
let parent = self
.directory
@@ -1074,25 +1079,17 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
Ok(())
}
- /// Whether [`Self::reclaim_obsolete`] has anything left to remove.
- #[must_use]
- pub fn has_obsolete(&self) -> bool {
- !self.obsolete.is_empty() || self.cleanup_directory_dirty
- }
-
- /// Remove a bounded batch of the files no published generation retains.
+ /// Remove a bounded batch of the files no published generation retains,
+ /// and retry a directory barrier a previous batch could not complete.
+ /// Does nothing when the queue is empty and no barrier is owed.
///
- /// Separate from the append path so an acknowledgement never waits on
- /// unlinks and a directory barrier for history it does not depend on. Only
- /// a checkpoint and the boot scan ever queue work here, and both do so
- /// after the publication that excludes those files, so the queue holds
- /// nothing a reader or a recovery could still need. The writer owns the
- /// journal, so this runs between mutations and never beside one.
- pub async fn reclaim_obsolete(&mut self) {
- self.cleanup_obsolete().await;
- }
-
- async fn cleanup_obsolete(&mut self) {
+ /// Off the append path, so the unlinks and the barrier do not sit inside a
+ /// group's durability. A generation reaches the queue from the boot scan,
+ /// or from the rewrite behind a checkpoint, a truncate, or a reset, and
+ /// always after the publication that stops naming it, so nothing here is
+ /// still reachable by a reader or a recovery. The writer owns the journal,
+ /// so this runs between mutations and never beside one.
+ pub async fn cleanup_obsolete(&mut self) {
let count = self.obsolete.len().min(16);
for _ in 0..count {
let Some(path) = self.obsolete.pop_front() else {
@@ -1865,18 +1862,55 @@ mod tests {
.unwrap();
journal.sync().await.unwrap();
- assert!(journal.has_obsolete());
assert!(
stale.exists(),
"the acknowledgement must not have waited on the unlink"
);
- journal.reclaim_obsolete().await;
+ journal.cleanup_obsolete().await;
- assert!(!journal.has_obsolete());
assert!(!stale.exists());
}
+ /// A failed unlink leaves its path queued for a retry. When a later append
+ /// references that same segment again it adopts the link already on disk,
+ /// so the retry would delete an inode the live generation depends on.
+ #[compio::test]
+ async fn an_append_reclaims_the_retained_link_it_adopts() {
+ let partition = tempdir().unwrap();
+ let directory = partition.path().join("prepares-7");
+ let mut journal = PartitionPrepareJournal::open(&directory, 42, 7)
+ .await
+ .unwrap();
+
+ let prepare = sized_prepare(1, 0, size_of::<PrepareHeader>() + 4096);
+ let parent = prepare.header().checksum;
+ let reference = write_segment(partition.path(), 0, 0, &prepare).await;
+ let retained = reference.path(&directory);
+ journal
+ .append_batch_referenced_buffered(&[prepare.into_frozen()],
&[Some(reference)])
+ .await
+ .unwrap();
+ journal.sync().await.unwrap();
+ assert!(retained.exists(), "the body link must exist to be adopted");
+
+ // Stand in for a removal that failed and was queued again.
+ journal.obsolete.push_back(retained.clone());
+
+ let second = sized_prepare(2, parent, size_of::<PrepareHeader>() +
4096);
+ journal
+ .append_batch_referenced_buffered(&[second.into_frozen()],
&[Some(reference)])
+ .await
+ .unwrap();
+ journal.sync().await.unwrap();
+ journal.cleanup_obsolete().await;
+
+ assert!(
+ retained.exists(),
+ "reclamation must not delete a link the newest generation
references"
+ );
+ }
+
#[compio::test]
async fn
referenced_bodies_survive_retention_and_checkpoint_without_wal_copies() {
let partition = tempdir().unwrap();
@@ -1919,7 +1953,7 @@ mod tests {
assert_eq!(recovered[1].as_slice(), second.as_slice());
journal.checkpoint(2).await.unwrap();
assert_eq!(journal.size_bytes(), PARTITION_WAL_BLOCK_SIZE as u64);
- journal.reclaim_obsolete().await;
+ journal.cleanup_obsolete().await;
assert!(!first_reference.path(&directory).exists());
assert!(second_reference.path(&directory).exists());
drop(journal);
@@ -2078,7 +2112,7 @@ mod tests {
assert_eq!(journal.purge_marker(), (1, 2));
journal.truncate_from(3).await.unwrap();
assert!(first_reference.path(&directory).exists());
- journal.reclaim_obsolete().await;
+ journal.cleanup_obsolete().await;
assert!(!second_reference.path(&directory).exists());
drop(journal);
let journal = PartitionPrepareJournal::open(&directory, 42, 7)
@@ -2653,7 +2687,7 @@ mod tests {
"purge must not fabricate a committed frontier"
);
assert_eq!(journal.retained_bytes(), retained_bytes);
- journal.reclaim_obsolete().await;
+ journal.cleanup_obsolete().await;
for reference in references {
assert!(!reference.path(&directory).exists());
std::fs::remove_file(
@@ -3278,7 +3312,7 @@ mod tests {
journal.prepares().await.unwrap()[0].as_slice(),
next.as_slice()
);
- journal.reclaim_obsolete().await;
+ journal.cleanup_obsolete().await;
assert_eq!(checkpoint_reference.path(&directory).exists(),
materialized);
let expected = SegmentPosition {
length: initial.length + BODY_BYTES as u64,
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 8f0183ffc..8300980bc 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -29,8 +29,8 @@ use crate::offset_storage::{
};
use crate::persistence::{PartitionPersistence, PersistenceCompletion,
PersistenceNotifier};
use crate::poll_plan::{
- DiskReadPlan, DiskSegment, PartitionDirResolution, PollContext, PollPlan,
PollReadResult,
- PollTier, ResidentTailSnapshot,
+ DISK_POLL_CHUNK_MAX, DiskReadPlan, DiskSegment, PartitionDirResolution,
PollContext, PollPlan,
+ PollReadResult, PollTier, ResidentTailSnapshot,
};
use crate::segment::Segment;
use crate::state_transfer::{PartitionTransferSession, PendingTransferRearm};
@@ -140,6 +140,9 @@ where
/// persisted): a fresh server treats a group as never-polled.
pub last_polled_offsets: Arc<ConsumerGroupOffsets>,
pub stats: Arc<PartitionStats>,
+ /// Widest batch committed here, the floor for a disk poll's first read.
+ /// See [`Self::widest_committed_batch`].
+ widest_batch_bytes: Cell<u64>,
pub created_at: IggyTimestamp,
pub revision_id: u64,
pub(crate) offset_space: OffsetSpace,
@@ -587,6 +590,7 @@ where
consumer_group_offsets:
Arc::new(ConsumerGroupOffsets::with_capacity(1)),
last_polled_offsets:
Arc::new(ConsumerGroupOffsets::with_capacity(1)),
stats,
+ widest_batch_bytes: Cell::new(0),
created_at: IggyTimestamp::now(),
revision_id: 0,
offset_space: OffsetSpace::default(),
@@ -2992,6 +2996,17 @@ where
}
self.check_local_poll_key(kind, consumer_id)
.map_err(|error| self.poll_capacity_error(error))?;
+ // Already durable: the commit this poll would make has happened and
+ // replicated, so applying it locally syncs this replica to something
+ // the group agreed. Safe wherever the read was served, and checked
+ // before the role so a caught-up backup is not refused for a
+ // commit nobody needs.
+ if self
+ .durable_consumer_offsets
+ .covers(kind, consumer_id, offset)
+ {
+ return Ok(None);
+ }
let consensus = self.consensus();
// A replica that cannot originate the prepare cannot record this
// progress anywhere a peer will ever see. `Ok(None)` would leave
@@ -3002,12 +3017,6 @@ where
if !consensus.is_primary() || !consensus.is_normal() ||
consensus.is_transferring() {
return Err(IggyError::TransientNotAccepted);
}
- if self
- .durable_consumer_offsets
- .covers(kind, consumer_id, offset)
- {
- return Ok(None);
- }
let reservation = self
.consumer_offset_capacity_for(kind)
@@ -3890,7 +3899,7 @@ where
};
}
- let (start_segment, start_position) = self.disk_poll_start(&query);
+ let (start_segment, start_position, start_index_offset) =
self.disk_poll_start(&query);
// Cap resident sealed read handles: touch this poll's start segment so
// the LRU keeps the hot set and drops the least-recently-used fd +
// index (a no-op for the active segment, whose slot is bounded by
@@ -3915,9 +3924,11 @@ where
partition_dir: self.partition_dir_resolution(),
segments,
start_position,
+ start_index_offset,
namespace_raw: self.namespace().inner(),
validate_checksum,
bytes_per_message: self.mean_encoded_message_size(),
+ widest_batch_bytes: self.widest_committed_batch(),
};
// Snapshot the resident journal tail now (on the pump, under the
// borrow) so the straddle splice runs off-task on owned data with no
@@ -4146,10 +4157,27 @@ where
(messages > 0).then(|| u32::try_from(bytes /
messages).unwrap_or(u32::MAX))
}
+ /// Widest batch this partition has committed, for the disk walk's chunk
+ /// floor. A batch is the unit that walk can consume, so a read below the
+ /// widest one risks decoding nothing and paying a re-read.
+ ///
+ /// A high-water, never lowered: retention cannot make an older batch
+ /// narrower, and the read path clamps it to the chunk ceiling anyway, so
+ /// the worst a stale value costs is the fixed-size read polls did before
+ /// they were sized at all.
+ fn widest_committed_batch(&self) -> u64 {
+ let widest = self.widest_batch_bytes.get();
+ if widest == 0 || self.recovered_durable_offset.is_some() {
+ widest.max(DISK_POLL_CHUNK_MAX)
+ } else {
+ widest
+ }
+ }
+
/// Starting `(segment index, byte position)` for a disk poll, resolved
/// via each segment's sparse index cache. An index miss starts at the
/// segment's first byte (the walk filters precisely).
- fn disk_poll_start(&self, query: &MessageLookup) -> (usize, u64) {
+ fn disk_poll_start(&self, query: &MessageLookup) -> (usize, u64,
Option<u64>) {
let segments = self.log.segments();
match query {
MessageLookup::Offset { offset, .. } => {
@@ -4157,12 +4185,17 @@ where
.iter()
.rposition(|segment| segment.start_offset <= *offset)
.unwrap_or(0);
- let position = self
+ let entry = self
.log
.segment_indexes(segment_index)
- .and_then(|cache| cache.offset_lower_bound(*offset))
- .map_or(0, |index| index.position);
- (segment_index, position)
+ .and_then(|cache| cache.offset_lower_bound(*offset));
+ let position = entry.map_or(0, |index| index.position);
+ let entry_offset = entry.map(|index| index.offset).or_else(|| {
+ segments
+ .get(segment_index)
+ .map(|segment| segment.start_offset)
+ });
+ (segment_index, position, entry_offset)
}
MessageLookup::Timestamp { timestamp, .. } => {
// Resolve the starting SEGMENT from segment metadata, not from
@@ -4182,7 +4215,7 @@ where
.segment_indexes(segment_index)
.and_then(|cache| cache.timestamp_lower_bound(*timestamp))
.map_or(0, |index| index.position);
- (segment_index, position)
+ (segment_index, position, None)
}
}
}
@@ -6506,6 +6539,8 @@ where
}
if let Some(batch_stats) = batch_stats {
+ self.widest_batch_bytes
+
.set(self.widest_batch_bytes.get().max(batch_stats.size_bytes));
let end_offset = batch_stats.end_offset();
// The committed counter now names data, which is what
makes
// it pollable and persistable. Outside the
recovered-offset
@@ -11016,6 +11051,31 @@ mod tests {
pub(super) type SentFrames = Rc<RefCell<Vec<(u128,
Frozen<MESSAGE_ALIGN>)>>>;
+ #[test]
+ fn recovered_history_keeps_a_conservative_batch_read_floor() {
+ let (mut partition, _) = recording_partition();
+ assert_eq!(partition.widest_committed_batch(), DISK_POLL_CHUNK_MAX);
+ partition.widest_batch_bytes.set(4096);
+ assert_eq!(partition.widest_committed_batch(), 4096);
+ partition.recovered_durable_offset = Some(100);
+ assert_eq!(partition.widest_committed_batch(), DISK_POLL_CHUNK_MAX);
+ }
+
+ #[test]
+ fn active_disk_poll_keeps_the_sparse_index_offset() {
+ let (mut partition, _) = recording_partition();
+ partition.log.ensure_indexes();
+ let index = partition.log.active_indexes_mut().unwrap();
+ index.insert(0, 0, 0);
+ index.insert(100, 100, 6400);
+ let query = MessageLookup::Offset {
+ offset: 150,
+ count: 10,
+ ceiling: 200,
+ };
+ assert_eq!(partition.disk_poll_start(&query), (0, 6400, Some(100)));
+ }
+
fn recording_partition() -> (IggyPartition<RecordingBus>, SentFrames) {
recording_partition_at(0, 1)
}
@@ -12185,6 +12245,24 @@ mod tests {
assert_eq!(partition.consensus.pipeline_len(), 0);
}
+ /// A backup whose durable table already covers the offset has nothing to
+ /// replicate, so there is no divergence to prevent and the refusal must
+ /// not reach it. Otherwise a caught-up follower would fail reads over a
+ /// commit the group already agreed.
+ #[test]
+ fn
given_backup_when_auto_commit_offset_is_already_durable_should_be_accepted() {
+ let (mut partition, _) = recording_partition_at(1, 3);
+ let consumer = PollingConsumer::Consumer(7, 0);
+ partition.seed_recovered_consumer_offset(ConsumerKind::Consumer, 7, 9,
9);
+ let read_result = poll_read_result(&partition, consumer, true,
Some(9));
+
+ let completion = partition
+ .complete_poll(read_result)
+ .expect("an already-durable offset admits no commit");
+ assert!(completion.replication.is_none());
+ assert_eq!(partition.consensus.pipeline_len(), 0);
+ }
+
/// An empty read never reaches automatic-commit admission and mutates no
/// progress, so the refusal above must not spread to it: a backup has to
/// keep answering the tail of a partition it is caught up on.
@@ -13902,6 +13980,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir: PartitionDirResolution::Resolved(partition_dir),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![
DiskSegment {
@@ -13918,6 +13997,7 @@ mod tests {
},
],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
@@ -13976,6 +14056,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir: PartitionDirResolution::Resolved(partition_dir),
bytes_per_message: Some(1),
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![DiskSegment {
start_offset: 0,
@@ -13984,6 +14065,7 @@ mod tests {
sealed: false,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
@@ -14062,6 +14144,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir: PartitionDirResolution::Resolved(partition_dir),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![
DiskSegment {
@@ -14078,6 +14161,7 @@ mod tests {
},
],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
@@ -14138,6 +14222,7 @@ mod tests {
let plan = |validate_checksum| DiskReadPlan {
partition_dir:
PartitionDirResolution::Resolved(partition_dir.clone()),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum,
segments: vec![DiskSegment {
start_offset: 0,
@@ -14146,6 +14231,7 @@ mod tests {
sealed: false,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
let query = MessageLookup::Offset {
@@ -14178,6 +14264,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir: PartitionDirResolution::NoFiles,
bytes_per_message: None,
+ widest_batch_bytes: 0,
segments: vec![DiskSegment {
start_offset: 0,
persisted: 512,
@@ -14185,6 +14272,7 @@ mod tests {
sealed: false,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: IggyNamespace::new(1, 1, 0).inner(),
validate_checksum: true,
};
@@ -14211,6 +14299,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir: PartitionDirResolution::Unresolvable,
bytes_per_message: None,
+ widest_batch_bytes: 0,
segments: vec![DiskSegment {
start_offset: 0,
persisted: 512,
@@ -14218,6 +14307,7 @@ mod tests {
sealed: false,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: IggyNamespace::new(1, 1, 0).inner(),
validate_checksum: true,
};
@@ -14279,6 +14369,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir:
PartitionDirResolution::Resolved(partition_dir.clone()),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![DiskSegment {
start_offset: 0,
@@ -14287,6 +14378,7 @@ mod tests {
sealed: true,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
let first = plan
@@ -14312,6 +14404,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir:
PartitionDirResolution::Resolved(partition_dir.clone()),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![DiskSegment {
start_offset: 0,
@@ -14320,6 +14413,7 @@ mod tests {
sealed: true,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
let second = plan
@@ -14374,6 +14468,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir:
PartitionDirResolution::Resolved(partition_dir.clone()),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![DiskSegment {
start_offset: 0,
@@ -14382,6 +14477,7 @@ mod tests {
sealed: true,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
let outcome = plan
@@ -14460,6 +14556,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir:
PartitionDirResolution::Resolved(partition_dir.clone()),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![DiskSegment {
start_offset: 0,
@@ -14470,6 +14567,7 @@ mod tests {
// Byte 0, exactly what disk_poll_start returns for a sealed
segment
// whose resident index was dropped.
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
let outcome = plan
@@ -14561,6 +14659,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir:
PartitionDirResolution::Resolved(partition_dir.clone()),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![DiskSegment {
start_offset: 0,
@@ -14569,6 +14668,7 @@ mod tests {
sealed: true,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
let outcome = plan
@@ -14640,6 +14740,7 @@ mod tests {
let plan = DiskReadPlan {
partition_dir:
PartitionDirResolution::Resolved(partition_dir.clone()),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![DiskSegment {
start_offset: 0,
@@ -14648,6 +14749,7 @@ mod tests {
sealed: true,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
let before_purge = plan
@@ -14690,6 +14792,7 @@ mod tests {
let resumed = DiskReadPlan {
partition_dir:
PartitionDirResolution::Resolved(partition_dir.clone()),
bytes_per_message: None,
+ widest_batch_bytes: 0,
validate_checksum: true,
segments: vec![DiskSegment {
start_offset: 0,
@@ -14698,6 +14801,7 @@ mod tests {
sealed: true,
}],
start_position: 0,
+ start_index_offset: None,
namespace_raw: namespace.inner(),
};
let after_purge = resumed
diff --git a/core/partitions/src/persistence.rs
b/core/partitions/src/persistence.rs
index 4c9255b03..026872152 100644
--- a/core/partitions/src/persistence.rs
+++ b/core/partitions/src/persistence.rs
@@ -45,6 +45,12 @@ use nix::sys::resource::{Resource, getrlimit};
const APPEND_BATCH_BYTES_MAX: u64 = 8 * 1024 * 1024;
const APPEND_BATCH_OPS_MAX: usize = 256;
const CHECKPOINT_DIRTY_FILES_MAX: usize = 1024;
+/// Mutations a partition may apply before its obsolete files are reclaimed
+/// whether or not the queue has drained. Reclaiming only on an idle queue
+/// keeps the unlinks off every acknowledgement, but a partition under
+/// continuous load never goes idle and would hold its old generations until
+/// it did.
+const RECLAIM_MUTATIONS_MAX: u32 = 64;
#[cfg(unix)]
const OFFSET_FILES_TOTAL_MAX: usize = 1024;
const OFFSET_FILES_PER_PARTITION_MAX: usize = 64;
@@ -1112,6 +1118,7 @@ impl<S: DurableStorage> PartitionPersistence<S> {
guard.complete = true;
return;
};
+ let mut mutations_since_reclaim = 0u32;
loop {
if self.retired.get() {
break;
@@ -1150,14 +1157,16 @@ impl<S: DurableStorage> PartitionPersistence<S> {
if epoch == self.epoch.get() && !self.retired.get() {
self.publish_mutation(journal, rebuild_references);
}
- // After the notification, never before the barrier it would delay.
- // A checkpoint queues the generation it replaced, so reclaiming it
- // here keeps the unlinks and the directory barrier out of the
- // acknowledgement the next append is waiting on, while still
- // running once per mutation so a busy partition reclaims as
- // promptly as an idle one.
- if journal.has_obsolete() {
- journal.reclaim_obsolete().await;
+ // Reclaim when nothing is queued behind this mutation, so no
+ // acknowledgement pays for the unlinks and the directory barrier.
+ // A partition that never drains would then never reclaim, so force
+ // a pass every `RECLAIM_MUTATIONS_MAX` mutations and accept that
+ // one group's latency.
+ mutations_since_reclaim += 1;
+ let idle = self.queue.borrow().is_empty();
+ if idle || mutations_since_reclaim >= RECLAIM_MUTATIONS_MAX {
+ mutations_since_reclaim = 0;
+ journal.cleanup_obsolete().await;
}
}
guard.complete = true;
diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs
index b0906d54b..64cdfdcbc 100644
--- a/core/partitions/src/poll_plan.rs
+++ b/core/partitions/src/poll_plan.rs
@@ -28,10 +28,11 @@ use crate::journal::{
};
use crate::{PollFragments, PollingConsumer};
use compio::io::AsyncReadAtExt;
+use iggy_binary_protocol::{WireError, batch};
use iggy_common::{ConsumerKind, IggyError};
use server_common::iobuf::{Frozen, Owned};
use server_common::poll::PollHistoryId;
-use server_common::send_messages::{BatchIntegrity, COMMAND_HEADER_SIZE,
decode_batch_slice_with};
+use server_common::send_messages::{BatchIntegrity, COMMAND_HEADER_SIZE};
use std::cell::{Cell, RefCell};
use std::rc::Rc;
use tracing::{error, warn};
@@ -135,14 +136,21 @@ pub struct DiskReadPlan {
/// first one.
pub(crate) segments: Vec<DiskSegment>,
pub(crate) start_position: u64,
+ pub(crate) start_index_offset: Option<u64>,
pub(crate) namespace_raw: u64,
/// Whether to verify each batch's `batch_checksum` against the bytes read.
/// Detection only; a mismatch fails the poll closed and repairs nothing.
pub(crate) validate_checksum: bool,
/// Mean encoded bytes per message on this partition, or `None` before it
- /// has committed anything. Sizes the disk walk's reads; never a bound on
- /// what a read may return, since messages vary in size within a partition.
+ /// has committed anything. Sizes the disk walk's reads, and is never a
+ /// bound on what a read may return, since messages vary in size within a
+ /// partition.
pub(crate) bytes_per_message: Option<u32>,
+ /// Widest batch this partition has committed, which is the smallest read
+ /// that is guaranteed to contain a whole one. The walk cannot advance on a
+ /// chunk holding no complete batch, so a count-derived estimate below this
+ /// buys nothing and costs the re-read it triggers.
+ pub(crate) widest_batch_bytes: u64,
}
pub struct DiskSegment {
@@ -390,7 +398,7 @@ pub enum DiskReadOutcome {
/// Largest first read of a disk poll, and the size every poll used to read
/// whatever it asked for. A batch wider than this still grows past it through
/// the re-read path below; this bounds only where a walk starts.
-const DISK_POLL_CHUNK_MAX: u64 = 1 << 20;
+pub const DISK_POLL_CHUNK_MAX: u64 = 1 << 20;
/// Smallest first read of a disk poll. Below this the syscall and the segment
/// walk cost more than the bytes the smaller read saves, and a poll for a
@@ -411,6 +419,10 @@ enum SegmentWalk {
struct DiskWalk {
/// Byte offset into the segment being walked; reset at each boundary.
position: u64,
+ /// Messages between the resolved index entry and the requested offset,
+ /// which the first read has to cover on top of what the poll asked for.
+ /// Cleared once anything matches, since the walk is then at the target.
+ skipped: u32,
matched: u32,
fragments: PollFragments<4096>,
last_matching_offset: Option<u64>,
@@ -421,9 +433,16 @@ struct DiskWalk {
}
impl DiskWalk {
- fn starting_at(position: u64) -> Self {
+ /// Messages the next read has to cover: what the poll still wants, plus
+ /// the run the sparse index left in front of the first match.
+ const fn remaining_to_read(&self, count: u32) -> u32 {
+ (count - self.matched).saturating_add(self.skipped)
+ }
+
+ fn starting_at(position: u64, skipped: u32) -> Self {
Self {
position,
+ skipped,
matched: 0,
fragments: PollFragments::new(),
last_matching_offset: None,
@@ -438,15 +457,25 @@ impl DiskWalk {
impl DiskReadPlan {
/// Bytes to read for the next `remaining` messages.
///
- /// A poll asks for a message count, and the walk reads bytes, so the two
- /// are bridged by the partition's own mean encoded size. Reading a fixed
+ /// A poll asks for a message count and the walk reads bytes, so the two
are
+ /// bridged by the partition's own mean encoded size. Reading a fixed
/// megabyte instead costs a poll for a thousand hundred-byte messages ten
/// times the bytes it returns, and the sparse-selection copy that follows
/// scales with the chunk rather than with the selection.
///
- /// The estimate is deliberately not a bound. Messages vary in size, the
- /// starting offset can sit inside a batch the index resolved before it,
- /// and an underestimate only costs another read of the next chunk.
+ /// The count-derived estimate alone is not enough, because a batch is the
+ /// unit the walk can consume. A poll for fewer messages than a producer
put
+ /// in one batch estimates below that batch, decodes nothing, and pays the
+ /// quadrupling re-read below: at one message short of a full batch that is
+ /// five times the bytes a flat megabyte read would have taken. So the
+ /// estimate is floored at the widest batch this partition has committed,
+ /// which is the smallest read guaranteed to hold a whole one.
+ ///
+ /// The result is still not a bound. Messages vary in size, the starting
+ /// offset can sit inside a batch the index resolved before it, and a batch
+ /// wider than the ceiling still grows through the re-read path. The
ceiling
+ /// is what every poll read before it was sized at all, so no poll reads
+ /// more than it used to.
fn chunk_len(&self, remaining: u32) -> u64 {
let Some(bytes_per_message) = self.bytes_per_message else {
return DISK_POLL_CHUNK_MAX;
@@ -454,6 +483,7 @@ impl DiskReadPlan {
u64::from(bytes_per_message)
.saturating_mul(u64::from(remaining))
.saturating_add(COMMAND_HEADER_SIZE as u64)
+ .max(self.widest_batch_bytes)
.clamp(DISK_POLL_CHUNK_MIN, DISK_POLL_CHUNK_MAX)
}
@@ -500,14 +530,25 @@ impl DiskReadPlan {
// miss or load failure keeps `start_position` (the pre-existing
// full-scan fallback). An active first segment keeps its
// resident-index-resolved `start_position` untouched.
- let position = match self.segments.first() {
- Some(first) => self
- .resolve_sealed_start(first, query, partition_dir)
- .await
- .unwrap_or(self.start_position),
- None => self.start_position,
+ let resolved = match self.segments.first() {
+ Some(first) => self.resolve_sealed_start(first, query,
partition_dir).await,
+ None => None,
+ };
+ let position = resolved.map_or(self.start_position, |(position, _)|
position);
+ // The index is sparse, so the entry it resolved can sit a whole flush
+ // group before the requested offset. The walk has to read that run to
+ // reach the first match, and sizing the read from the requested count
+ // alone would cross it in floor-sized reads.
+ let entry_offset = resolved
+ .map(|(_, offset)| offset)
+ .or(self.start_index_offset);
+ let skipped = match (entry_offset, query) {
+ (Some(entry_offset), MessageLookup::Offset { offset, .. }) => {
+
u32::try_from(offset.saturating_sub(entry_offset)).unwrap_or(u32::MAX)
+ }
+ _ => 0,
};
- let mut walk = DiskWalk::starting_at(position);
+ let mut walk = DiskWalk::starting_at(position, skipped);
// Set when an open/read retry exhausts. The walk breaks immediately so
// later segments are never read into the result (which would leave a
// gap at the faulted segment). Pre-fault matches are still served.
@@ -577,8 +618,10 @@ impl DiskReadPlan {
/// Read one segment from `walk.position` until the count is filled, the
/// segment is exhausted, or the walk must fail closed.
///
- /// The chunk length is already bounded by `DISK_POLL_CHUNK_MAX` and by the
- /// segment's persisted bytes, so narrowing it to a `usize` cannot
truncate.
+ /// The read length is the chunk clipped to the segment's persisted bytes,
+ /// so narrowing it to a `usize` cannot truncate. The chunk itself is not
+ /// bounded by `DISK_POLL_CHUNK_MAX`: a batch wider than the ceiling grows
+ /// past it below.
#[allow(clippy::cast_possible_truncation)]
async fn walk_segment(
&self,
@@ -588,7 +631,7 @@ impl DiskReadPlan {
persisted: u64,
walk: &mut DiskWalk,
) -> SegmentWalk {
- let mut chunk_len = self.chunk_len(count - walk.matched);
+ let mut chunk_len = self.chunk_len(walk.remaining_to_read(count));
while walk.matched < count && walk.position < persisted {
let len = (persisted - walk.position).min(chunk_len) as usize;
let Some(chunk) = self.read_chunk_with_retry(file, len,
walk).await else {
@@ -597,7 +640,11 @@ impl DiskReadPlan {
return SegmentWalk::Faulted;
};
let fragments_before_chunk = walk.fragments.len();
- let ChunkWalk { consumed, corrupt } = walk_disk_chunk(
+ let ChunkWalk {
+ consumed,
+ needed,
+ corrupt,
+ } = walk_disk_chunk(
&chunk,
query,
count,
@@ -632,11 +679,23 @@ impl DiskReadPlan {
// run, which would punch a silent gap into the poll.
return SegmentWalk::Faulted;
}
- // A single batch larger than the chunk: grow and re-read.
- chunk_len = chunk_len.saturating_mul(4);
+ // A single batch larger than the chunk. Its own header says
+ // how wide it is, so re-read exactly that; only a header this
+ // read could not reach leaves the old quadrupling.
+ if needed as u64 > persisted - walk.position {
+ return SegmentWalk::Faulted;
+ }
+ chunk_len = if needed > len {
+ needed as u64
+ } else {
+ chunk_len.saturating_mul(4)
+ };
continue;
}
- chunk_len = self.chunk_len(count - walk.matched);
+ if walk.matched > 0 {
+ walk.skipped = 0;
+ }
+ chunk_len = self.chunk_len(walk.remaining_to_read(count));
walk.position += consumed as u64;
}
SegmentWalk::Done
@@ -682,7 +741,7 @@ impl DiskReadPlan {
segment: &DiskSegment,
query: MessageLookup,
partition_dir: &str,
- ) -> Option<u64> {
+ ) -> Option<(u64, u64)> {
// The active segment grows under the reader, so neither the shared
// sparse index nor the offset memo can describe it; its own resident
// index already resolved `start_position`.
@@ -707,7 +766,7 @@ impl DiskReadPlan {
&& offset >= cursor.offset
&& offset < cursor.valid_until
{
- return Some(cursor.position);
+ return Some((cursor.position, cursor.offset));
}
let path = format!("{partition_dir}/{:0>20}.index",
segment.start_offset);
let reader = match IggyIndexReader::new(&path).await {
@@ -758,7 +817,7 @@ impl DiskReadPlan {
}
};
match looked_up {
- Ok(entry) => entry.map(|entry| entry.position),
+ Ok(entry) => entry.map(|entry| (entry.position, entry.offset)),
Err(error) => {
self.warn_sparse_index_fallback(&path, "lower_bound", &error);
None
@@ -848,12 +907,15 @@ impl DiskReadPlan {
/// timestamp, or `None` when the query is below the first indexed entry (the
/// caller then scans from the segment start). Mirrors `disk_poll_start`'s
/// resident-index resolution for the sealed, off-pump path.
-fn resolve_index_position(index: &IggyIndexCache, query: MessageLookup) ->
Option<u64> {
+/// Start byte for `query`, and the offset of the index entry it resolved to.
+/// The entry sits at or before the requested offset, so the difference is the
+/// run the walk has to skip before it can match anything.
+fn resolve_index_position(index: &IggyIndexCache, query: MessageLookup) ->
Option<(u64, u64)> {
match query {
MessageLookup::Offset { offset, .. } =>
index.offset_lower_bound(offset),
MessageLookup::Timestamp { timestamp, .. } =>
index.timestamp_lower_bound(timestamp),
}
- .map(|entry| entry.position)
+ .map(|entry| (entry.position, entry.offset))
}
/// Walk stamped `[256B BatchHeader][blob]` batches in one disk
@@ -873,11 +935,16 @@ fn walk_disk_chunk(
) -> ChunkWalk {
let bytes: &[u8] = chunk;
let mut cursor = 0usize;
+ let mut needed = 0usize;
while *matched < count && cursor + COMMAND_HEADER_SIZE <= bytes.len() {
- let batch = match decode_batch_slice_with(&bytes[cursor..], integrity)
{
+ let batch = match batch::decode_batch_slice_with(&bytes[cursor..],
integrity) {
Ok(batch) => batch,
- Err(IggyError::InvalidBatchChecksum(found, expected, base_offset))
=> {
+ Err(WireError::InvalidBatchChecksum {
+ stored: found,
+ computed: expected,
+ base_offset,
+ }) => {
// Distinguished from the incomplete-tail case below: this
batch is
// entirely present and fails its own checksum, so it is
damaged at rest.
error!(
@@ -892,13 +959,24 @@ fn walk_disk_chunk(
);
return ChunkWalk {
consumed: cursor.min(bytes.len()),
+ needed: 0,
corrupt: true,
};
}
- Err(_) => {
- // Incomplete tail batch: hand the position back to re-read or
bail.
+ Err(WireError::UnexpectedEof { need, .. })
+ if need <= journal::partition_journal::PREPARE_BYTES_MAX =>
+ {
+ needed = need;
break;
}
+ Err(error) => {
+ error!(namespace_raw, position = cursor, %error, "invalid disk
batch");
+ return ChunkWalk {
+ consumed: cursor,
+ needed: 0,
+ corrupt: true,
+ };
+ }
};
let total_size = batch.header.total_size();
@@ -921,6 +999,7 @@ fn walk_disk_chunk(
ChunkWalk {
consumed: cursor.min(bytes.len()),
+ needed,
corrupt: false,
}
}
@@ -929,6 +1008,10 @@ fn walk_disk_chunk(
/// than on a batch that simply did not fit in the chunk.
struct ChunkWalk {
consumed: usize,
+ /// Bytes the batch that did not fit needs in full, from its own header, or
+ /// zero when that header could not be read. Lets the caller re-read
+ /// exactly the batch instead of doubling its way up to it.
+ needed: usize,
corrupt: bool,
}
@@ -936,8 +1019,13 @@ struct ChunkWalk {
mod tests {
use super::*;
use crate::iggy_index::IggyIndex;
+ use bytes::Bytes;
use compio::io::AsyncWriteAtExt;
use server_common::iobuf::Owned;
+ use server_common::send_messages::{
+ IggyMessage, IggyMessageHeader, IggyMessages, SendMessagesOwned,
+ };
+ use server_common::sharding::IggyNamespace;
/// Write a sealed-segment index file too large to materialize
/// (`entry_count * IGGY_INDEX_SIZE > SEALED_INDEX_RESIDENT_MAX_BYTES`), so
@@ -962,16 +1050,47 @@ mod tests {
entry_count
}
- #[test]
- fn chunk_len_sizes_the_first_read_from_the_requested_count() {
- let plan = |bytes_per_message| DiskReadPlan {
+ fn sizing_plan(bytes_per_message: Option<u32>, widest_batch_bytes: u64) ->
DiskReadPlan {
+ DiskReadPlan {
partition_dir: PartitionDirResolution::NoFiles,
bytes_per_message,
+ widest_batch_bytes,
segments: Vec::new(),
start_position: 0,
+ start_index_offset: None,
namespace_raw: 0,
validate_checksum: false,
- };
+ }
+ }
+
+ /// A batch is the unit the walk can consume, so a count-derived estimate
+ /// that lands under one decodes nothing and pays the quadrupling re-read.
+ /// A poll one message short of a producer's batch was the worst case,
+ /// reading about five times what a flat megabyte would have.
+ #[test]
+ fn chunk_len_never_lands_under_a_whole_batch() {
+ let batch = COMMAND_HEADER_SIZE as u64 + 1000 * 1000;
+ let mean = u32::try_from(batch / 1000).expect("mean fits");
+ let plan = sizing_plan(Some(mean), batch);
+
+ assert!(plan.chunk_len(999) >= batch);
+ assert!(plan.chunk_len(500) >= batch);
+ assert!(plan.chunk_len(1) >= batch);
+ // The floor never pushes a read above what an unsized poll would take.
+ assert_eq!(
+ sizing_plan(Some(mean), 4 << 20).chunk_len(1),
+ DISK_POLL_CHUNK_MAX
+ );
+ // A count wide enough to matter still wins over the floor.
+ assert_eq!(
+ sizing_plan(Some(10), 2048).chunk_len(1000),
+ DISK_POLL_CHUNK_MIN
+ );
+ }
+
+ #[test]
+ fn chunk_len_sizes_the_first_read_from_the_requested_count() {
+ let plan = |bytes_per_message| sizing_plan(bytes_per_message, 0);
// Nothing committed yet, so nothing bridges a count to bytes.
assert_eq!(plan(None).chunk_len(1000), DISK_POLL_CHUNK_MAX);
@@ -1001,12 +1120,14 @@ mod tests {
let plan = DiskReadPlan {
partition_dir: PartitionDirResolution::NoFiles,
bytes_per_message: None,
+ widest_batch_bytes: 0,
segments: Vec::new(),
start_position: 0,
+ start_index_offset: None,
namespace_raw: 0,
validate_checksum: true,
};
- let mut walk = DiskWalk::starting_at(0);
+ let mut walk = DiskWalk::starting_at(0, 0);
assert!(
plan.read_chunk_with_retry(&file, 64, &mut walk)
.await
@@ -1017,6 +1138,42 @@ mod tests {
assert_eq!(walk.matched, 0);
}
+ #[cfg(feature = "poll-diagnostics")]
+ #[compio::test]
+ async fn incomplete_batch_reread_uses_its_exact_length() {
+ let directory = tempfile::tempdir().unwrap();
+ let mut messages = IggyMessages::with_capacity(1);
+ messages.push(IggyMessage {
+ header: IggyMessageHeader {
+ payload_length: 128 << 10,
+ ..Default::default()
+ },
+ payload: Bytes::from(vec![1; 128 << 10]),
+ user_headers: None,
+ });
+ let batch =
+ SendMessagesOwned::from_messages(IggyNamespace::new(1, 1, 0),
&messages).unwrap();
+ let length = batch.header.total_size();
+ let mut record = vec![0; length];
+ batch.header.encode_into(&mut record);
+ record[COMMAND_HEADER_SIZE..].copy_from_slice(&batch.blob);
+ let mut file =
compio::fs::File::create(directory.path().join("batches.log"))
+ .await
+ .unwrap();
+ let (written, _) = file.write_all_at(record.repeat(4), 0).await.into();
+ written.unwrap();
+ let plan = sizing_plan(Some(1), 0);
+ let mut walk = DiskWalk::starting_at(0, 0);
+ assert!(matches!(
+ plan.walk_segment(&file, offset_query(0), 1, (length * 4) as u64,
&mut walk)
+ .await,
+ SegmentWalk::Done
+ ));
+ assert_eq!(walk.matched, 1);
+ assert_eq!(walk.chunk_reads, 2);
+ assert_eq!(walk.requested_bytes, DISK_POLL_CHUNK_MIN + length as u64);
+ }
+
fn offset_query(offset: u64) -> MessageLookup {
MessageLookup::Offset {
offset,
@@ -1048,8 +1205,10 @@ mod tests {
let plan = DiskReadPlan {
partition_dir:
PartitionDirResolution::Resolved(dir.display().to_string()),
bytes_per_message: None,
+ widest_batch_bytes: 0,
segments: Vec::new(),
start_position: 0,
+ start_index_offset: None,
namespace_raw: 0,
validate_checksum: false,
};
@@ -1060,7 +1219,9 @@ mod tests {
let first = plan
.resolve_sealed_start(&segment, offset_query(25), &partition_dir)
.await;
- assert_eq!(first, Some(200));
+ // The entry offset rides along so the caller can size its first read
+ // to cover the run between that entry and the requested offset.
+ assert_eq!(first, Some((200, 20)));
let cursor = handle.offset_cursor.get().expect("cursor memoized");
assert_eq!(
(cursor.offset, cursor.valid_until, cursor.position),
@@ -1073,7 +1234,7 @@ mod tests {
let in_interval = plan
.resolve_sealed_start(&segment, offset_query(29), &partition_dir)
.await;
- assert_eq!(in_interval, Some(200));
+ assert_eq!(in_interval, Some((200, 20)));
// ...while an offset past the interval misses the cursor, reaches for
// the (now gone) file, and falls back to the byte-0 scan.
diff --git a/core/server/src/http.rs b/core/server/src/http.rs
index e7ecdc1ff..9d04339f5 100644
--- a/core/server/src/http.rs
+++ b/core/server/src/http.rs
@@ -350,10 +350,6 @@ fn router(
"/personal-access-tokens/login",
post(login_with_personal_access_token),
)
- .route(
- "/streams/{stream_id}/topics/{topic_id}/messages",
- get(poll_messages),
- )
.route(
"/streams/{stream_id}/topics/{topic_id}/consumer-offsets",
get(get_consumer_offset),
@@ -418,11 +414,18 @@ fn router(
/// Acknowledged partition writes use a bounded HTTP roster fallback. This is
/// a correctness path for the existing stateless HTTP transport. Direct
/// partition-primary routing remains the scalable long-term design.
+///
+/// A poll rides the same fallback, because an automatic commit is a partition
+/// write in every sense but the verb: only a replica that can originate the
+/// offset operation may serve one, and the refusal it would otherwise return
+/// is terminal for an HTTP caller, which has no roster of its own to walk.
+/// [`forward::forward_partition_write`] leaves a poll without automatic commit
+/// entirely local, so an ordinary read still pays nothing for this.
fn partition_write_routes(state: HttpState) -> Router<HttpState> {
Router::new()
.route(
"/streams/{stream_id}/topics/{topic_id}/messages",
- post(send_messages),
+ post(send_messages).get(poll_messages),
)
.route(
"/streams/{stream_id}/topics/{topic_id}/consumer-offsets",
diff --git a/core/server/src/http/forward.rs b/core/server/src/http/forward.rs
index 423f073a4..d8e513675 100644
--- a/core/server/src/http/forward.rs
+++ b/core/server/src/http/forward.rs
@@ -54,11 +54,12 @@
use std::cell::Cell;
use std::net::SocketAddr;
use std::path::Path;
+use std::rc::Rc;
use std::sync::Arc;
use std::time::{Duration, Instant};
use axum::body::{Body, to_bytes};
-use axum::extract::{Request, State};
+use axum::extract::{Query, Request, State};
use axum::http::header::{AUTHORIZATION, CONTENT_TYPE, RETRY_AFTER};
use axum::http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
use axum::middleware::Next;
@@ -66,7 +67,7 @@ use axum::response::{IntoResponse, Response};
use bytes::Bytes;
use configs::http::HttpTlsConfig;
use consensus::MetadataHandle;
-use futures::StreamExt;
+use futures::{Stream, StreamExt};
use iggy_common::IggyError;
use message_bus::transports::tls::load_pem;
use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified,
ServerCertVerifier};
@@ -118,9 +119,9 @@ const MAX_IN_FLIGHT_FORWARDS: u32 = 128;
/// Bound on a relayed response body, enforced twice: against a declared
/// `content-length` before the read, and as a running cap on the streamed
-/// bytes so a length-less reply is bounded too. Forwarded routes answer
-/// entity JSON, not message batches (poll and snapshot are served locally),
-/// so this is a backstop, not a working limit.
+/// bytes so a length-less reply is bounded too. Successful poll responses
+/// stream separately because their automatic commit may already have advanced
+/// progress, so rejecting their total size would discard acknowledged data.
const RESPONSE_BODY_LIMIT: usize = 64 * 1024 * 1024;
/// Pre-allocation hint cap for a relayed body: honest control-plane replies
@@ -220,7 +221,7 @@ pub(in crate::http) fn build_forward_state(
client,
scheme,
body_limit,
- in_flight: Cell::new(0),
+ in_flight: Rc::new(Cell::new(0)),
})
}
@@ -334,6 +335,14 @@ async fn forward_partition_or_pass(state: HttpState,
request: Request, next: Nex
if !state.forward.active ||
request.headers().contains_key(FORWARDED_HEADER) {
return next.run(request).await;
}
+ // A read that does not move consumer progress is answered by whichever
+ // replica the caller reached, which is the point of reading from one. Only
+ // an automatic commit needs a replica that can originate the offset
+ // operation, so only that poll pays for the buffering and the credential
+ // resolution below.
+ if request.method() == Method::GET &&
!wants_auto_commit(request.uri().query()) {
+ return next.run(request).await;
+ }
let bearer = match bearer_token(request.headers()) {
Ok(bearer) => bearer,
Err(error) => return CustomError::from(error).into_response(),
@@ -364,7 +373,7 @@ async fn forward_partition_or_pass(state: HttpState,
request: Request, next: Nex
return response;
}
- let Some(_guard) = ForwardGuard::admit(&state.forward.in_flight) else {
+ let Some(guard) = ForwardGuard::admit(&state.forward.in_flight) else {
return with_retry_after(error_response(
StatusCode::SERVICE_UNAVAILABLE,
"forward_busy",
@@ -385,7 +394,7 @@ async fn forward_partition_or_pass(state: HttpState,
request: Request, next: Nex
}
let url = format!("{}://{socket}{path_and_query}",
state.forward.scheme);
match attempt(&state, &method, &request_headers, &body, &url,
false).await {
- AttemptOutcome::Relay(response) => return response,
+ AttemptOutcome::Relay(response) => return
retain_forward_guard(response, guard),
AttemptOutcome::Retry => {}
}
}
@@ -476,51 +485,7 @@ async fn attempt(
Ok(response) => response,
Err(error) => return classify_transport_error(&error),
};
- let status = response.status();
- // Only the relayed subset survives; the response is consumed by the
- // body stream below, so the values are pulled out first.
- let relayed_headers: Vec<(HeaderName, HeaderValue)> =
RELAYED_RESPONSE_HEADERS
- .into_iter()
- .filter_map(|name| {
- let value = response.headers().get(&name)?.clone();
- Some((name, value))
- })
- .collect();
- let declared = response.content_length();
- if declared.is_some_and(|length| length > RESPONSE_BODY_LIMIT as u64) {
- warn!(?declared, "relayed response exceeds the body bound");
- return AttemptOutcome::Relay(bad_gateway());
- }
- // Streamed with a running cap so a length-less reply is bounded by
- // the limit, not merely by the attempt timeout. The capacity hint is
- // clamped to RESPONSE_CAPACITY_HINT so a mis-declared content-length
- // cannot pre-reserve the full bound. The running cap still bounds the
- // real total.
- let mut body = Vec::with_capacity(
- declared
- .and_then(|length| usize::try_from(length).ok())
- .unwrap_or(0)
- .min(RESPONSE_CAPACITY_HINT),
- );
- let mut stream = response.bytes_stream();
- while let Some(chunk) = stream.next().await {
- let chunk = match chunk {
- Ok(chunk) => chunk,
- Err(error) => {
- warn!(%error, "forward response body read failed; outcome
unknown");
- return AttemptOutcome::Relay(bad_gateway());
- }
- };
- if body.len() + chunk.len() > RESPONSE_BODY_LIMIT {
- warn!(
- received = body.len() + chunk.len(),
- "relayed response exceeds the body bound"
- );
- return AttemptOutcome::Relay(bad_gateway());
- }
- body.extend_from_slice(&chunk);
- }
- classify_reply(status, relayed_headers, Bytes::from(body),
retry_redirect)
+ classify_forwarded_reply(response, method, retry_redirect).await
};
match compio::time::timeout(FORWARD_ATTEMPT_TIMEOUT, attempt).await {
// Elapsed: the request may be mid-commit on the primary. Outcome
@@ -534,6 +499,90 @@ async fn attempt(
}
}
+async fn classify_forwarded_reply(
+ response: cyper::Response,
+ method: &Method,
+ retry_redirect: bool,
+) -> AttemptOutcome {
+ let status = response.status();
+ // Only the relayed subset survives; the response is consumed by the
+ // body stream below, so the values are pulled out first.
+ let relayed_headers: Vec<(HeaderName, HeaderValue)> =
RELAYED_RESPONSE_HEADERS
+ .into_iter()
+ .filter_map(|name| {
+ let value = response.headers().get(&name)?.clone();
+ Some((name, value))
+ })
+ .collect();
+ if method == Method::GET && !retry_redirect && status.is_success() {
+ let mut response =
Response::new(stream_poll_body(response.bytes_stream()));
+ *response.status_mut() = status;
+ for (name, value) in relayed_headers {
+ response.headers_mut().insert(name, value);
+ }
+ return AttemptOutcome::Relay(response);
+ }
+ let declared = response.content_length();
+ if declared.is_some_and(|length| length > RESPONSE_BODY_LIMIT as u64) {
+ warn!(?declared, "relayed response exceeds the body bound");
+ return AttemptOutcome::Relay(bad_gateway());
+ }
+ // Streamed with a running cap so a length-less reply is bounded by
+ // the limit, not merely by the attempt timeout. The capacity hint is
+ // clamped to RESPONSE_CAPACITY_HINT so a mis-declared content-length
+ // cannot pre-reserve the full bound. The running cap still bounds the
+ // real total.
+ let mut body = Vec::with_capacity(
+ declared
+ .and_then(|length| usize::try_from(length).ok())
+ .unwrap_or(0)
+ .min(RESPONSE_CAPACITY_HINT),
+ );
+ let mut stream = response.bytes_stream();
+ while let Some(chunk) = stream.next().await {
+ let chunk = match chunk {
+ Ok(chunk) => chunk,
+ Err(error) => {
+ warn!(%error, "forward response body read failed; outcome
unknown");
+ return AttemptOutcome::Relay(bad_gateway());
+ }
+ };
+ if body.len() + chunk.len() > RESPONSE_BODY_LIMIT {
+ warn!(
+ received = body.len() + chunk.len(),
+ "relayed response exceeds the body bound"
+ );
+ return AttemptOutcome::Relay(bad_gateway());
+ }
+ body.extend_from_slice(&chunk);
+ }
+ classify_reply(status, relayed_headers, Bytes::from(body), retry_redirect)
+}
+
+fn stream_poll_body(stream: impl Stream<Item = Result<Bytes, cyper::Error>> +
'static) -> Body {
+ let stream = futures::stream::try_unfold(Box::pin(stream), |mut stream|
async move {
+ match compio::time::timeout(FORWARD_ATTEMPT_TIMEOUT,
stream.next()).await {
+ Ok(Some(Ok(bytes))) => Ok(Some((bytes, stream))),
+ Ok(None) => Ok(None),
+ Ok(Some(Err(error))) =>
Err(std::io::Error::other(error.to_string())),
+ Err(_) => Err(std::io::Error::new(
+ std::io::ErrorKind::TimedOut,
+ "forwarded poll response stalled",
+ )),
+ }
+ });
+ Body::from_stream(SendWrapper::new(stream))
+}
+
+fn retain_forward_guard(response: Response, guard: ForwardGuard) -> Response {
+ let (parts, body) = response.into_parts();
+ let stream = futures::stream::unfold(
+ (body.into_data_stream(), guard),
+ |(mut stream, guard)| async move { stream.next().await.map(|chunk|
(chunk, (stream, guard))) },
+ );
+ Response::from_parts(parts, Body::from_stream(SendWrapper::new(stream)))
+}
+
/// Copy the forwardable request headers: the bearer (the primary
/// re-authenticates it) and the content type. Everything else - including any
/// client-supplied forward marker, which `forward_or_pass` already bounced -
@@ -599,6 +648,9 @@ fn classify_reply(
/// Inspect a response produced on this node without changing any terminal
/// response. Only the typed never-admitted denial opens the roster fallback.
async fn classify_local_partition_reply(response: Response) -> AttemptOutcome {
+ if response.status() != StatusCode::SERVICE_UNAVAILABLE {
+ return AttemptOutcome::Relay(response);
+ }
let (parts, body) = response.into_parts();
let body = match to_bytes(body, RESPONSE_BODY_LIMIT).await {
Ok(body) => body,
@@ -654,6 +706,24 @@ fn partition_http_sockets(
.collect()
}
+/// Whether a poll asks the server to store its offset after serving it.
+///
+/// Match the handler's URL decoding and boolean parsing. An invalid query
+/// reaches the handler through the forwarding layer for its normal rejection.
+fn wants_auto_commit(query: Option<&str>) -> bool {
+ #[derive(Default, Deserialize)]
+ struct AutoCommitQuery {
+ #[serde(default)]
+ auto_commit: bool,
+ }
+ query.is_some_and(|query| {
+ let Ok(uri) = format!("/?{query}").parse() else {
+ return true;
+ };
+ Query::<AutoCommitQuery>::try_from_uri(&uri).map_or(true,
|Query(query)| query.auto_commit)
+ })
+}
+
fn wants_linearizable(query: Option<&str>) -> bool {
query.is_some_and(|query| {
query
@@ -672,21 +742,23 @@ fn bad_gateway() -> Response {
/// RAII admission against [`MAX_IN_FLIGHT_FORWARDS`]; releases on drop, so a
/// client disconnect mid-forward frees the slot.
-struct ForwardGuard<'a> {
- in_flight: &'a Cell<u32>,
+struct ForwardGuard {
+ in_flight: Rc<Cell<u32>>,
}
-impl<'a> ForwardGuard<'a> {
- fn admit(in_flight: &'a Cell<u32>) -> Option<Self> {
+impl ForwardGuard {
+ fn admit(in_flight: &Rc<Cell<u32>>) -> Option<Self> {
if in_flight.get() >= MAX_IN_FLIGHT_FORWARDS {
return None;
}
in_flight.set(in_flight.get() + 1);
- Some(Self { in_flight })
+ Some(Self {
+ in_flight: Rc::clone(in_flight),
+ })
}
}
-impl Drop for ForwardGuard<'_> {
+impl Drop for ForwardGuard {
fn drop(&mut self) {
self.in_flight.set(self.in_flight.get() - 1);
}
@@ -748,6 +820,8 @@ impl ServerCertVerifier for PinnedCertVerifier {
mod tests {
use super::*;
+ use std::io::{Read, Write};
+
use configs::cluster::{ClusterNodeConfig, ResolvedClusterNode,
TransportPorts};
fn node(replica_id: u8, ip: &str, http: Option<u16>) -> ClusterNodeConfig {
@@ -793,6 +867,21 @@ mod tests {
assert!(!wants_linearizable(None));
}
+ #[test]
+ fn auto_commit_query_matches_handler_decoding() {
+ assert!(wants_auto_commit(Some("auto_commit=true")));
+ assert!(wants_auto_commit(Some("auto_commit=1")));
+ assert!(wants_auto_commit(Some("count=10&auto_commit=true")));
+ assert!(wants_auto_commit(Some("auto_commit=yes")));
+ assert!(!wants_auto_commit(Some("auto_commit=false")));
+ assert!(wants_auto_commit(Some("auto_commit=0")));
+ assert!(!wants_auto_commit(Some("count=10")));
+ assert!(!wants_auto_commit(None));
+ assert!(wants_auto_commit(Some("%61uto_commit=true")));
+ assert!(wants_auto_commit(Some("auto_commit=%74rue")));
+ assert!(!wants_auto_commit(Some("%61uto_commit=%66alse")));
+ }
+
#[test]
fn transient_not_accepted_body_matches_only_its_code() {
let accepted = format!(
@@ -811,7 +900,7 @@ mod tests {
#[test]
fn forward_guard_caps_and_releases() {
- let in_flight = Cell::new(0);
+ let in_flight = Rc::new(Cell::new(0));
let guards: Vec<_> = (0..MAX_IN_FLIGHT_FORWARDS)
.map(|_| ForwardGuard::admit(&in_flight).expect("under cap"))
.collect();
@@ -852,4 +941,117 @@ mod tests {
};
assert_eq!(terminal.status(), StatusCode::SERVICE_UNAVAILABLE);
}
+
+ #[compio::test]
+ async fn successful_local_poll_body_is_not_collected_or_size_limited() {
+ let polled = Rc::new(Cell::new(0));
+ let observed = Rc::clone(&polled);
+ let chunk = Bytes::from(vec![0; 1024 * 1024]);
+ let stream = futures::stream::iter((0..65).map(move |_| {
+ observed.set(observed.get() + 1);
+ Ok::<_, std::io::Error>(chunk.clone())
+ }));
+ let response =
Response::new(Body::from_stream(SendWrapper::new(stream)));
+ let AttemptOutcome::Relay(response) =
classify_local_partition_reply(response).await else {
+ panic!("successful response must not be retried")
+ };
+ assert_eq!(response.status(), StatusCode::OK);
+ assert_eq!(polled.get(), 0);
+ let mut stream = response.into_body().into_data_stream();
+ let mut received = 0;
+ while let Some(chunk) = stream.next().await {
+ received += chunk.expect("successful chunk").len();
+ }
+ assert_eq!(received, 65 * 1024 * 1024);
+ }
+
+ #[compio::test]
+ async fn forwarded_poll_streams_large_body_and_retains_admission() {
+ let in_flight = Rc::new(Cell::new(0));
+ let guard = ForwardGuard::admit(&in_flight).expect("forward admitted");
+ let polled = Rc::new(Cell::new(0));
+ let observed = Rc::clone(&polled);
+ let chunk = Bytes::from(vec![0; 1024 * 1024]);
+ let stream = futures::stream::iter((0..65).map(move |_| {
+ observed.set(observed.get() + 1);
+ Ok::<_, cyper::Error>(chunk.clone())
+ }));
+ let response =
retain_forward_guard(Response::new(stream_poll_body(stream)), guard);
+ assert_eq!(polled.get(), 0);
+ assert_eq!(in_flight.get(), 1);
+ let mut stream = response.into_body().into_data_stream();
+ let mut received = 0;
+ while let Some(chunk) = stream.next().await {
+ received += chunk.expect("successful chunk").len();
+ assert_eq!(in_flight.get(), 1);
+ }
+ assert_eq!(received, 65 * 1024 * 1024);
+ assert_eq!(in_flight.get(), 0);
+ }
+
+ #[compio::test]
+ async fn dropping_forwarded_poll_body_releases_admission() {
+ let in_flight = Rc::new(Cell::new(0));
+ let guard = ForwardGuard::admit(&in_flight).expect("forward admitted");
+ let response = retain_forward_guard(
+ Response::new(stream_poll_body(futures::stream::pending())),
+ guard,
+ );
+ assert_eq!(in_flight.get(), 1);
+ drop(response);
+ assert_eq!(in_flight.get(), 0);
+ }
+
+ #[compio::test]
+ async fn forwarded_poll_accepts_large_http_content_length() {
+ let _ = rustls::crypto::ring::default_provider().install_default();
+ let forward =
+ build_forward_state(&HttpTlsConfig::default(), 1024,
true).expect("forward client");
+ let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("HTTP
listener");
+ let address = listener.local_addr().expect("listener address");
+ let server = std::thread::spawn(move || {
+ let (mut socket, _) = listener.accept().expect("client
connection");
+ socket
+ .set_read_timeout(Some(Duration::from_secs(10)))
+ .expect("read timeout");
+ socket
+ .set_write_timeout(Some(Duration::from_secs(10)))
+ .expect("write timeout");
+ let mut request = Vec::new();
+ let mut bytes = [0; 1024];
+ while !request.windows(4).any(|window| window == b"\r\n\r\n") {
+ let read = socket.read(&mut bytes).expect("request bytes");
+ assert_ne!(read, 0);
+ request.extend_from_slice(&bytes[..read]);
+ }
+ write!(socket, "HTTP/1.1 200 OK\r\nContent-Length:
{}\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n", 65 * 1024 *
1024)
+ .expect("response headers");
+ let chunk = vec![0; 1024 * 1024];
+ for _ in 0..65 {
+ socket.write_all(&chunk).expect("response chunk");
+ }
+ });
+ let response = forward
+ .client
+ .get(format!("http://{address}/messages";))
+ .expect("poll request")
+ .send()
+ .await
+ .expect("poll response");
+ assert_eq!(response.content_length(), Some(65 * 1024 * 1024));
+ let AttemptOutcome::Relay(response) =
+ classify_forwarded_reply(response, &Method::GET, false).await
+ else {
+ panic!("successful poll must not be retried")
+ };
+ assert_eq!(response.status(), StatusCode::OK);
+ assert_eq!(response.headers()[CONTENT_TYPE], "application/json");
+ let mut stream = response.into_body().into_data_stream();
+ let mut received = 0;
+ while let Some(chunk) = stream.next().await {
+ received += chunk.expect("response chunk").len();
+ }
+ assert_eq!(received, 65 * 1024 * 1024);
+ server.join().expect("HTTP server finished");
+ }
}
diff --git a/core/server/src/http/state.rs b/core/server/src/http/state.rs
index 5c64f4c21..f3ef51700 100644
--- a/core/server/src/http/state.rs
+++ b/core/server/src/http/state.rs
@@ -146,7 +146,7 @@ pub(in crate::http) struct ForwardState {
/// the same scheme as this node (uniform cluster HTTP config).
pub(in crate::http) scheme: &'static str,
pub(in crate::http) body_limit: usize,
- pub(in crate::http) in_flight: Cell<u32>,
+ pub(in crate::http) in_flight: Rc<Cell<u32>>,
}
/// Shared shard-0 HTTP state.
diff --git a/core/simulator/src/storage/tests.rs
b/core/simulator/src/storage/tests.rs
index 7a63f85f8..51f58cc50 100644
--- a/core/simulator/src/storage/tests.rs
+++ b/core/simulator/src/storage/tests.rs
@@ -1210,7 +1210,7 @@ fn
obsolete_wal_generations_are_reclaimed_after_restart_and_failed_unlink() {
let (storage, mut journal) = baseline().await;
storage.clear_trace();
journal.checkpoint(2).await.unwrap();
- journal.reclaim_obsolete().await;
+ journal.cleanup_obsolete().await;
let unlink = storage
.trace()
.iter()
@@ -1220,7 +1220,7 @@ fn
obsolete_wal_generations_are_reclaimed_after_restart_and_failed_unlink() {
let (storage, mut journal) = baseline().await;
storage.fail_at(unlink, FaultMode::Before);
journal.checkpoint(2).await.unwrap();
- journal.reclaim_obsolete().await;
+ journal.cleanup_obsolete().await;
storage.clear_trace();
let obsolete = Path::new("/partition/wal/prepares-0.wal");
assert!(storage.exists(obsolete).await.unwrap());
@@ -1256,7 +1256,7 @@ fn
obsolete_wal_generations_are_reclaimed_after_restart_and_failed_unlink() {
// have waited on the retry, and the writer's own maintenance
// pass is what must still take it.
assert!(storage.exists(obsolete).await.unwrap());
- journal.reclaim_obsolete().await;
+ journal.cleanup_obsolete().await;
}
assert!(!storage.exists(obsolete).await.unwrap());
assert_eq!(journal.checkpoint_op(), 2);
diff --git
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java
index 7166861e7..db89596bd 100644
---
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java
+++
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java
@@ -67,6 +67,9 @@ public final class BytesSerializer {
/** The timestamp delta is a u32 microsecond offset from the batch origin
timestamp. */
private static final BigInteger MAX_TIMESTAMP_DELTA_MICROS =
BigInteger.valueOf(0xFFFF_FFFFL);
+ /** Encoded user headers of a message that carries none. */
+ private static final byte[] EMPTY_USER_HEADERS = new byte[0];
+
/** Batch checksum input: five u64 header fields plus the u32 message
count. */
private static final int BATCH_CHECKSUM_FIXED_INPUT_BYTES = 5 * Long.BYTES
+ Integer.BYTES;
@@ -281,7 +284,7 @@ public final class BytesSerializer {
encodedMessageId(message.header().id()),
message.header().originTimestamp(),
message.payload(),
- readAllBytes(toBytes(message.userHeaders()))));
+ encodedUserHeaders(message.userHeaders())));
}
return rawMessages;
}
@@ -300,21 +303,33 @@ public final class BytesSerializer {
private static BatchExtent measureBatch(List<RawMessage> messages, long
capacityAllowance) {
var originTimestamp = messages.get(0).originTimestamp();
var latestTimestamp = originTimestamp;
+ var latestIndex = 0;
long length = BATCH_HEADER_SIZE;
- for (RawMessage message : messages) {
+ for (int index = 0; index < messages.size(); index++) {
+ RawMessage message = messages.get(index);
var timestamp = message.originTimestamp();
if (timestamp.signum() < 0 || timestamp.bitLength() > Long.SIZE) {
- throw new IggyInvalidArgumentException("Message origin
timestamp is outside unsigned 64-bit range");
+ throw new IggyInvalidArgumentException("Message " + index + "
origin timestamp " + timestamp
+ + " is outside the unsigned 64-bit range");
}
originTimestamp = originTimestamp.min(timestamp);
- latestTimestamp = latestTimestamp.max(timestamp);
+ if (timestamp.compareTo(latestTimestamp) > 0) {
+ latestTimestamp = timestamp;
+ latestIndex = index;
+ }
length += (long) MessageHeader.SIZE + message.payload().length +
message.userHeaders().length;
if (length > capacityAllowance) {
throw new IggyInvalidArgumentException("Message batch exceeds
the output buffer capacity");
}
}
- if
(latestTimestamp.subtract(originTimestamp).compareTo(MAX_TIMESTAMP_DELTA_MICROS)
> 0) {
- throw new IggyInvalidArgumentException("Message origin timestamp
delta exceeds unsigned 32-bit range");
+ // Name the offending message and its delta, the way the server's own
+ // InvalidMessageTimestampDelta does: the batch origin is whichever
+ // message is oldest, so neither is obvious from the caller's input.
+ var delta = latestTimestamp.subtract(originTimestamp);
+ if (delta.compareTo(MAX_TIMESTAMP_DELTA_MICROS) > 0) {
+ throw new IggyInvalidArgumentException("Message " + latestIndex
+ + " origin timestamp exceeds the batch origin by " + delta
+ + " microseconds, more than the timestamp delta field can
hold");
}
return new BatchExtent(originTimestamp, length);
}
@@ -411,6 +426,21 @@ public final class BytesSerializer {
return Hashing.xxh3_64().hashBytesToLong(bytes);
}
+ /**
+ * Encoded user headers, or an empty array when there are none.
+ *
+ * <p>Returns before a buffer exists for the empty case. {@link
#toBytes(Map)} answers that case
+ * with the shared {@link Unpooled#EMPTY_BUFFER}, which {@link
#readAllBytes(ByteBuf)} would then
+ * release. That release happens to be a no-op on the singleton, which is
the only reason the
+ * previous shape was safe.
+ */
+ private static byte[] encodedUserHeaders(Map<HeaderKey, HeaderValue>
userHeaders) {
+ if (userHeaders == null || userHeaders.isEmpty()) {
+ return EMPTY_USER_HEADERS;
+ }
+ return readAllBytes(toBytes(userHeaders));
+ }
+
private static byte[] readAllBytes(ByteBuf buffer) {
try {
var bytes = new byte[buffer.readableBytes()];
diff --git
a/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/MessagesBatchWireFormatTest.java
b/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/MessagesBatchWireFormatTest.java
index 5351f3c0f..23a5c114a 100644
---
a/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/MessagesBatchWireFormatTest.java
+++
b/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/MessagesBatchWireFormatTest.java
@@ -243,7 +243,35 @@ class MessagesBatchWireFormatTest {
var messages = List.of(message(1, 0, "a", Map.of()), message(2,
0x1_0000_0000L, "b", Map.of()));
assertThatThrownBy(() -> BytesSerializer.toMessagesBatch(messages))
- .isInstanceOf(IggyInvalidArgumentException.class);
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("Message 1")
+ .hasMessageContaining("4294967296 microseconds");
+ }
+
+ @Test
+ void shouldReportTimestampDeltaAgainstTheOldestMessageRegardlessOfOrder() {
+ var messages = List.of(
+ message(1, 0x1_0000_0000L + 7, "latest", Map.of()),
+ message(2, 7, "oldest", Map.of()),
+ message(3, 8, "middle", Map.of()));
+
+ assertThatThrownBy(() -> BytesSerializer.toMessagesBatch(messages))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("Message 0")
+ .hasMessageContaining("4294967296 microseconds");
+ }
+
+ @Test
+ void shouldEncodeNullAndEmptyUserHeadersIdentically() {
+ var absent = BytesSerializer.toMessagesBatch(List.of(message(1, 7,
"payload", null)));
+ var empty = BytesSerializer.toMessagesBatch(List.of(message(1, 7,
"payload", Map.of())));
+ try {
+
assertThat(ByteBufUtil.hexDump(absent)).isEqualTo(ByteBufUtil.hexDump(empty));
+ assertThat(absent.getIntLE(256 + 32)).isZero();
+ } finally {
+ absent.release();
+ empty.release();
+ }
}
@Test
