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
commit 73ad56cfbe62e59330cfcefb64e959e30e485705 Author: spetz <[email protected]> AuthorDate: Sun Sep 13 22:38:18 2026 +0200 perf(server): reduce produce and poll overhead --- core/journal/src/partition_journal.rs | 59 +++- core/metadata/src/stm/stream.rs | 66 +++-- core/partitions/Cargo.toml | 7 + core/partitions/src/iggy_partition.rs | 202 ++++++++++++-- core/partitions/src/journal.rs | 11 + core/partitions/src/persistence.rs | 103 ++++--- core/partitions/src/poll_plan.rs | 306 ++++++++++++++++----- core/server/config.toml | 7 + core/server_common/src/send_messages.rs | 78 +++++- core/shard/Cargo.toml | 2 +- core/shard/src/lib.rs | 8 + core/shard/src/metrics.rs | 35 +++ core/simulator/src/storage/tests.rs | 37 +++ .../iggy/client/async/tcp/AsyncTcpConnection.java | 16 ++ .../iggy/client/async/tcp/MessagesTcpClient.java | 40 ++- .../org/apache/iggy/serde/BytesSerializer.java | 166 +++++++---- .../tcp/AsyncTcpConnectionConcurrencyTest.java | 55 ++++ .../iggy/serde/MessagesBatchWireFormatTest.java | 93 +++++++ 18 files changed, 1074 insertions(+), 217 deletions(-) diff --git a/core/journal/src/partition_journal.rs b/core/journal/src/partition_journal.rs index 9610ad37d..47f18fb3f 100644 --- a/core/journal/src/partition_journal.rs +++ b/core/journal/src/partition_journal.rs @@ -750,7 +750,6 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { references: Option<&[Option<SegmentReference>]>, ) -> io::Result<()> { self.ensure_healthy()?; - self.cleanup_obsolete().await; self.recovered_prepares.clear(); let mut state = self.state; let mut retained_bytes = self.retained_bytes; @@ -1075,6 +1074,24 @@ 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. + /// + /// 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) { let count = self.obsolete.len().min(16); for _ in 0..count { @@ -1508,7 +1525,6 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { self.retain_active_segment_file(); self.poisoned = false; self.obsolete.push_back(obsolete); - self.cleanup_obsolete().await; Ok(()) } } @@ -1826,6 +1842,41 @@ mod tests { } } + /// Reclamation used to run at the head of every append, so an unlink and a + /// directory barrier for a generation the append does not touch landed + /// inside the acknowledgement it was waiting on. The append must leave the + /// queue alone and the writer must drain it between mutations instead. + #[compio::test] + async fn append_leaves_obsolete_files_for_the_writer_to_reclaim() { + let partition = tempdir().unwrap(); + let directory = partition.path().join("prepares-7"); + let mut journal = PartitionPrepareJournal::open(&directory, 42, 7) + .await + .unwrap(); + + let stale = directory.join("prepares-6.wal"); + std::fs::write(&stale, b"a generation a checkpoint replaced").unwrap(); + journal.obsolete.push_back(stale.clone()); + + let prepare = prepare(1, 0); + journal + .append_batch_buffered(&[prepare.into_frozen()]) + .await + .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; + + assert!(!journal.has_obsolete()); + assert!(!stale.exists()); + } + #[compio::test] async fn referenced_bodies_survive_retention_and_checkpoint_without_wal_copies() { let partition = tempdir().unwrap(); @@ -1868,6 +1919,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; assert!(!first_reference.path(&directory).exists()); assert!(second_reference.path(&directory).exists()); drop(journal); @@ -2026,6 +2078,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; assert!(!second_reference.path(&directory).exists()); drop(journal); let journal = PartitionPrepareJournal::open(&directory, 42, 7) @@ -2600,6 +2653,7 @@ mod tests { "purge must not fabricate a committed frontier" ); assert_eq!(journal.retained_bytes(), retained_bytes); + journal.reclaim_obsolete().await; for reference in references { assert!(!reference.path(&directory).exists()); std::fs::remove_file( @@ -3224,6 +3278,7 @@ mod tests { journal.prepares().await.unwrap()[0].as_slice(), next.as_slice() ); + journal.reclaim_obsolete().await; assert_eq!(checkpoint_reference.path(&directory).exists(), materialized); let expected = SegmentPosition { length: initial.length + BODY_BYTES as u64, diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs index ae03bade5..2ee4e5db8 100644 --- a/core/metadata/src/stm/stream.rs +++ b/core/metadata/src/stm/stream.rs @@ -1586,11 +1586,8 @@ impl Streams { let stream = inner.items.get(stream_id)?; let topic = stream.topics.get(topic_id)?; let partition_id = usize::try_from(partition_id).ok()?; - topic - .partitions - .iter() - .any(|partition| partition.id == partition_id) - .then(|| IggyNamespace::new(stream_id, topic_id, partition_id)) + find_partition(&topic.partitions, partition_id) + .map(|_| IggyNamespace::new(stream_id, topic_id, partition_id)) }) } @@ -1632,17 +1629,7 @@ impl Streams { self.inner.read(|inner| { let stream = inner.items.get(namespace.stream_id())?; let topic = stream.topics.get(namespace.topic_id())?; - let partition_id = namespace.partition_id(); - if let Some(partition) = topic.partitions.get(partition_id) - && partition.id == partition_id - { - return Some(read(partition)); - } - topic - .partitions - .iter() - .find(|partition| partition.id == partition_id) - .map(read) + find_partition(&topic.partitions, namespace.partition_id()).map(read) }) } @@ -1840,6 +1827,22 @@ const fn admits_slab_key(vacant_key: usize, ceiling: usize) -> bool { vacant_key < ceiling } +/// The partition carrying `partition_id`, or `None` when the topic has none. +/// +/// A primary mints dense 0-based ids, so the vector index IS the id and the +/// direct hit answers every ordinary topic. The scan exists only for a vector +/// left sparse by a partition delete, where an index no longer names its id. +fn find_partition(partitions: &[Partition], partition_id: usize) -> Option<&Partition> { + if let Some(partition) = partitions.get(partition_id) + && partition.id == partition_id + { + return Some(partition); + } + partitions + .iter() + .find(|partition| partition.id == partition_id) +} + /// Range and distinctness for the ABSOLUTE partition ids on a topic create, /// `None` when the vector is well formed. /// @@ -4228,6 +4231,37 @@ mod tests { .clone() } + /// A delete leaves every surviving id above the hole naming an index that + /// is no longer its own, so the direct hit has to fall through to the scan + /// or the survivor resolves to nothing and its namespace stops routing. + #[test] + fn given_sparse_partition_ids_when_resolving_should_fall_back_to_the_scan() { + let inner = inner_with_registered_partition(); + let template = committed_partition(&inner, 0, 0, 0); + let partition = |id| Partition { + id, + ..template.clone() + }; + let dense = vec![partition(0), partition(1)]; + let sparse = vec![partition(3), partition(7)]; + + assert_eq!( + find_partition(&dense, 1).map(|partition| partition.id), + 1.into() + ); + assert_eq!( + find_partition(&sparse, 3).map(|partition| partition.id), + 3.into() + ); + assert_eq!( + find_partition(&sparse, 7).map(|partition| partition.id), + 7.into() + ); + assert!(find_partition(&sparse, 0).is_none()); + assert!(find_partition(&sparse, 1).is_none()); + assert!(find_partition(&dense, 2).is_none()); + } + /// A checkpoint reads a stream's total and each of its topics' as separate /// loads while the partition plane keeps counting, so the two can disagree /// in either direction. The boot restore adopts neither level, and journal diff --git a/core/partitions/Cargo.toml b/core/partitions/Cargo.toml index 41524d685..8b1efd778 100644 --- a/core/partitions/Cargo.toml +++ b/core/partitions/Cargo.toml @@ -29,6 +29,13 @@ readme = "../../README.md" publish = false [features] +# Per-poll read accounting on the disk walk: bytes requested from the file +# API against the encoded bytes served, and the chunk reads it took. Off by +# default, and the server forwards `shard/poll-diagnostics` to it, so the +# ratio is available on a short diagnostic run without a permanent log line +# on the read path. +poll-diagnostics = [] + # Simulator-only detector hook (`IggyPartitions::hold_borrow_across_await`): # deliberately holds a `with_partition` borrow across an `.await` so the # dispatch shell can prove its borrow-across-await detector. A diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 0f5fac0af..8f0183ffc 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -1107,6 +1107,11 @@ where .map(|persistence| persistence.take_metrics()) } + /// Entries and bytes this partition's repair ring pins right now. + pub fn repair_ring_occupancy(&self) -> (usize, u64) { + self.log.journal().inner.evicted_ring_occupancy() + } + fn persistence_checkpoint_pending(&self) -> bool { self.persistence .as_ref() @@ -2972,9 +2977,10 @@ where } /// Admit an automatic commit without advancing this read's progress. - /// `Some` carries an assigned prepare. `None` means the request is queued, - /// the durable offset already covers it, or the current consensus role or - /// state cannot originate a prepare. Errors release any provisional guard. + /// `Some` carries an assigned prepare. `None` means the request is queued + /// or the durable offset already covers it, both of which leave the + /// caller free to apply local progress. Errors release any provisional + /// guard. fn admit_poll_auto_commit( &self, kind: ConsumerKind, @@ -2987,12 +2993,18 @@ where self.check_local_poll_key(kind, consumer_id) .map_err(|error| self.poll_capacity_error(error))?; let consensus = self.consensus(); - if !consensus.is_primary() - || !consensus.is_normal() - || consensus.is_transferring() - || self - .durable_consumer_offsets - .covers(kind, consumer_id, offset) + // A replica that cannot originate the prepare cannot record this + // progress anywhere a peer will ever see. `Ok(None)` would leave + // `complete_poll` applying the offset to local state alone, so the + // poll would report progress the group never agreed, and a later + // read on the primary would hand the same messages out again. + // Refusing keeps the outcome retriable on a replica that can commit. + 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); } @@ -3905,6 +3917,7 @@ where start_position, namespace_raw: self.namespace().inner(), validate_checksum, + bytes_per_message: self.mean_encoded_message_size(), }; // Snapshot the resident journal tail now (on the pump, under the // borrow) so the straddle splice runs off-task on owned data with no @@ -4121,6 +4134,18 @@ where .any(|segment| segment.size.as_bytes_u64() > 0) } + /// Mean encoded bytes per committed message, including its share of the + /// batch headers, or `None` while the partition has committed nothing. + /// + /// Both counters are relaxed loads that retention also decrements, so this + /// is a hint and nothing reads it as a bound. Its one consumer sizes the + /// first read of a disk poll, where being wrong costs an extra read. + fn mean_encoded_message_size(&self) -> Option<u32> { + let messages = self.stats.messages_count_inconsistent(); + let bytes = self.stats.size_bytes_inconsistent(); + (messages > 0).then(|| u32::try_from(bytes / messages).unwrap_or(u32::MAX)) + } + /// 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). @@ -12143,6 +12168,60 @@ mod tests { assert_eq!(partition.consensus.pipeline_len(), 0); } + /// A backup answering a read cannot originate the offset prepare, so + /// admitting the commit would advance local progress alone. The primary + /// would still hold the old offset and hand the same messages out again. + #[test] + fn given_backup_when_auto_commit_poll_completes_should_reject_without_progress() { + let (mut partition, _) = recording_partition_at(1, 3); + let consumer = PollingConsumer::Consumer(7, 0); + let read_result = poll_read_result(&partition, consumer, true, Some(9)); + + assert!(matches!( + partition.complete_poll(read_result), + Err(IggyError::TransientNotAccepted) + )); + assert_eq!(partition.get_consumer_offset(consumer), 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. + #[test] + fn given_backup_when_empty_auto_commit_poll_completes_should_be_accepted() { + let (mut partition, _) = recording_partition_at(1, 3); + let consumer = PollingConsumer::Consumer(7, 0); + let read_result = poll_read_result(&partition, consumer, true, None); + + let completion = partition + .complete_poll(read_result) + .expect("an empty read admits no commit"); + assert!(completion.replication.is_none()); + assert_eq!(partition.get_consumer_offset(consumer), None); + } + + #[test] + fn given_primary_when_role_changes_before_auto_commit_completion_should_reject_without_progress() + { + for transferring in [false, true] { + let (mut partition, _) = recording_partition_at(0, 3); + let consumer = PollingConsumer::ConsumerGroup(7, 0); + let read_result = poll_read_result(&partition, consumer, true, Some(9)); + if transferring { + partition.consensus.begin_state_transfer_await(); + } else { + partition.consensus.begin_view_probe(); + } + assert!(matches!( + partition.complete_poll(read_result), + Err(IggyError::TransientNotAccepted) + )); + assert_eq!(partition.group_offset_state(7), (None, None)); + assert_eq!(partition.consensus.pipeline_len(), 0); + } + } + #[test] fn given_group_read_without_auto_commit_when_history_changes_should_not_record_last_polled() { let (mut partition, _) = recording_partition(); @@ -12340,12 +12419,12 @@ mod tests { )); } + /// Disk reads of one group can finish out of order, and each carries its + /// own automatic commit. Both are admitted, so ordering has to be settled + /// where progress is recorded rather than by the order they complete in. #[test] fn given_group_reads_completing_in_reverse_order_should_keep_progress_monotone() { - // A backup accepts local poll progress without assigning replication. - let backup_replica = 1; - let replica_count = 3; - let (mut partition, _) = recording_partition_at(backup_replica, replica_count); + let (mut partition, _) = recording_partition_at(0, 3); let group_id = 7; let member_id = 1; let auto_commit = true; @@ -12359,11 +12438,15 @@ mod tests { .complete_poll(later_result) .expect("accept later poll") .replication - .is_none() + .is_some() + ); + assert!( + partition + .complete_poll(earlier_result) + .expect("accept earlier poll") + .replication + .is_some() ); - partition - .complete_poll(earlier_result) - .expect("accept earlier poll"); let (last_polled, committed) = partition.group_offset_state(group_id as u64); assert_eq!( last_polled, @@ -12375,7 +12458,7 @@ mod tests { Some(9), "the slower read must not rewind its offset" ); - assert_eq!(partition.consensus.pipeline_len(), 0); + assert_eq!(partition.consensus.pipeline_len(), 2); } #[test] @@ -13818,6 +13901,7 @@ mod tests { // open exhausts retries -> the walk must fault-close before segment two. let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir), + bytes_per_message: None, validate_checksum: true, segments: vec![ DiskSegment { @@ -13853,6 +13937,77 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// Sizing the first read from the requested count makes it routinely + /// narrower than one batch, which the walk answers by re-reading the same + /// position four times as wide. A partition whose mean message is small + /// and whose next batch is not must still serve that batch. + #[compio::test] + async fn read_disk_serves_a_batch_wider_than_the_sized_chunk() { + let namespace = IggyNamespace::new(1, 1, 0); + + let dir = std::env::temp_dir().join(format!( + "iggy-read-disk-wide-batch-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock after epoch") + .as_nanos(), + )); + compio::fs::create_dir_all(&dir) + .await + .expect("create temp partition dir"); + let partition_dir = dir.to_string_lossy().into_owned(); + + let payload = Bytes::from(vec![0x5Au8; 256 << 10]); + let record = build_segment_record_with_payload(namespace, 0, payload.clone()); + let record_len = record.len() as u64; + let path = format!("{partition_dir}/{:0>20}.log", 0u64); + { + let mut file = compio::fs::File::create(&path) + .await + .expect("create segment file"); + let (written, _) = file.write_all_at(record, 0).await.into(); + written.expect("write segment record"); + file.sync_all().await.expect("flush segment file"); + } + + // One byte per message floors the first read at 64 KiB, a quarter of + // the batch waiting at offset 0. + let plan = DiskReadPlan { + partition_dir: PartitionDirResolution::Resolved(partition_dir), + bytes_per_message: Some(1), + validate_checksum: true, + segments: vec![DiskSegment { + start_offset: 0, + persisted: record_len, + read_state: SealedSegmentHandle::default(), + sealed: false, + }], + start_position: 0, + namespace_raw: namespace.inner(), + }; + + let outcome = plan + .read_disk(MessageLookup::Offset { + offset: 0, + count: 1, + ceiling: u64::MAX, + }) + .await; + + let DiskReadOutcome::Matched { + fragments, matched, .. + } = outcome + else { + panic!("a batch wider than the first read must still be served"); + }; + assert_eq!(matched, 1); + let served: u64 = fragments.iter().map(|fragment| fragment.len() as u64).sum(); + assert_eq!(served, record_len); + + let _ = std::fs::remove_dir_all(&dir); + } + /// Fail-closed disk read on a CORRUPT (present-but-undecodable) batch in an /// EARLIER segment: like a missing/unreadable segment, the walk must stop /// (`Faulted`) rather than skip past the garbage and serve a LATER @@ -13906,6 +14061,7 @@ mod tests { let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir), + bytes_per_message: None, validate_checksum: true, segments: vec![ DiskSegment { @@ -13981,6 +14137,7 @@ mod tests { let plan = |validate_checksum| DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + bytes_per_message: None, validate_checksum, segments: vec![DiskSegment { start_offset: 0, @@ -14020,6 +14177,7 @@ mod tests { async fn read_disk_serves_journal_when_partition_has_no_files() { let plan = DiskReadPlan { partition_dir: PartitionDirResolution::NoFiles, + bytes_per_message: None, segments: vec![DiskSegment { start_offset: 0, persisted: 512, @@ -14052,6 +14210,7 @@ mod tests { async fn read_disk_faults_closed_when_partition_dir_unresolvable() { let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Unresolvable, + bytes_per_message: None, segments: vec![DiskSegment { start_offset: 0, persisted: 512, @@ -14119,6 +14278,7 @@ mod tests { let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + bytes_per_message: None, validate_checksum: true, segments: vec![DiskSegment { start_offset: 0, @@ -14151,6 +14311,7 @@ mod tests { let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + bytes_per_message: None, validate_checksum: true, segments: vec![DiskSegment { start_offset: 0, @@ -14212,6 +14373,7 @@ mod tests { let handle = SealedSegmentHandle::default(); let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + bytes_per_message: None, validate_checksum: true, segments: vec![DiskSegment { start_offset: 0, @@ -14297,6 +14459,7 @@ mod tests { let handle = SealedSegmentHandle::default(); let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + bytes_per_message: None, validate_checksum: true, segments: vec![DiskSegment { start_offset: 0, @@ -14397,6 +14560,7 @@ mod tests { let handle = SealedSegmentHandle::default(); let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + bytes_per_message: None, validate_checksum: true, segments: vec![DiskSegment { start_offset: 0, @@ -14475,6 +14639,7 @@ mod tests { let handle = Rc::clone(&partition.log.sealed_read_state()[0]); let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + bytes_per_message: None, validate_checksum: true, segments: vec![DiskSegment { start_offset: 0, @@ -14524,6 +14689,7 @@ mod tests { // unlinked pre-purge inode. let resumed = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(partition_dir.clone()), + bytes_per_message: None, validate_checksum: true, segments: vec![DiskSegment { start_offset: 0, diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index 0afb99442..71491f6e7 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -340,6 +340,17 @@ impl PartitionJournal<PartitionJournalMemStorage> { self.evicted_ring_bytes_max.set(bytes_max); } + /// Entries and bytes the repair ring currently pins, for the shard sweep. + /// + /// Both ceilings are per partition, so an operator raising them is really + /// raising `partitions * bytes_max` of anonymous memory on every replica, + /// held beside the same bytes in page cache. The configured ceiling says + /// nothing about what is actually retained; this does. + pub fn evicted_ring_occupancy(&self) -> (usize, u64) { + let ring = unsafe { &*self.evicted_ring.get() }; + (ring.len(), self.evicted_ring_bytes.get()) + } + /// Resident (un-evicted) entry count; diagnostics only. pub fn resident_count(&self) -> usize { let op_to_storage_offset = unsafe { &*self.op_to_storage_offset.get() }; diff --git a/core/partitions/src/persistence.rs b/core/partitions/src/persistence.rs index 3f1bb8870..4c9255b03 100644 --- a/core/partitions/src/persistence.rs +++ b/core/partitions/src/persistence.rs @@ -108,6 +108,11 @@ pub struct PersistenceMetrics { pub checkpoints_pending: u64, pub completed_batches: u64, pub batched_prepares: u64, + /// Durable groups that took the optional pre-barrier wait. Zero while the + /// delay is disabled, and zero with it enabled means the arrival gap never + /// cleared the guard, so a group-size change measured against the delay + /// alone would be attributing something else. + pub group_commit_waits: u64, pub completed_checkpoints: u64, pub failed_writes: u64, } @@ -160,6 +165,7 @@ pub struct PartitionPersistence<S: DurableStorage = DiskStorage> { last_append: Cell<Option<Instant>>, completed_batches: Cell<u64>, batched_prepares: Cell<u64>, + group_commit_waits: Cell<u64>, completed_checkpoints: Cell<u64>, failed_writes: Cell<u64>, } @@ -543,6 +549,7 @@ impl<S: DurableStorage> PartitionPersistence<S> { last_append: Cell::new(None), completed_batches: Cell::new(0), batched_prepares: Cell::new(0), + group_commit_waits: Cell::new(0), completed_checkpoints: Cell::new(0), failed_writes: Cell::new(0), }); @@ -981,6 +988,7 @@ impl<S: DurableStorage> PartitionPersistence<S> { checkpoints_pending: u64::from(self.checkpoint_pending()), completed_batches: self.completed_batches.replace(0), batched_prepares: self.batched_prepares.replace(0), + group_commit_waits: self.group_commit_waits.replace(0), completed_checkpoints: self.completed_checkpoints.replace(0), failed_writes: self.failed_writes.replace(0), } @@ -1140,49 +1148,66 @@ impl<S: DurableStorage> PartitionPersistence<S> { break; } if epoch == self.epoch.get() && !self.retired.get() { - let mut references = self.segment_references.borrow_mut(); - if rebuild_references { - references.clear(); - references.extend(journal.written_segment_references(0)); - } else if let Some(from_op) = self.written_head.get().checked_add(1) { - references.extend(journal.written_segment_references(from_op)); - } - drop(references); - self.disk_bytes.set(journal.size_bytes()); - self.retained_bytes.set(journal.retained_bytes()); - self.segment_checkpoint.set(journal.segment_checkpoint()); - let advanced = journal.durable_op() != self.durable_head.get() - || journal.checkpoint_op() != self.checkpoint.get() - || journal.certified_log_view() != self.certified_log_view.get() - || (journal.segment_checkpoint().is_some() - && journal.head() != self.written_head.get()); - self.certified_log_view.set(journal.certified_log_view()); - if self - .requested_log_view - .get() - .is_some_and(|(view, _, _)| Some(view) == self.certified_log_view.get()) - { - self.requested_log_view.set(None); - } - self.written_head.set(journal.head()); - self.durable_head.set(journal.durable_op()); - if journal.checkpoint_op() > self.checkpoint.get() { - self.accepted - .borrow_mut() - .checkpoint(journal.checkpoint_op()); - } - self.checkpoint.set(journal.checkpoint_op()); - self.checkpoint_checksum.set(journal.checkpoint_checksum()); - self.purge_generation.set(journal.purge_marker().0); - self.purge_floor.set(journal.purge_marker().1); - if advanced { - self.notify(); - } + 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; } } guard.complete = true; } + /// Republish what the completed mutation moved, and wake the partition when + /// any of it advanced. Runs only while the writer still owns this epoch: a + /// retired or re-epoched writer must not overwrite the state its successor + /// published. + fn publish_mutation(&self, journal: &PartitionPrepareJournal<S>, rebuild_references: bool) { + let mut references = self.segment_references.borrow_mut(); + if rebuild_references { + references.clear(); + references.extend(journal.written_segment_references(0)); + } else if let Some(from_op) = self.written_head.get().checked_add(1) { + references.extend(journal.written_segment_references(from_op)); + } + drop(references); + self.disk_bytes.set(journal.size_bytes()); + self.retained_bytes.set(journal.retained_bytes()); + self.segment_checkpoint.set(journal.segment_checkpoint()); + let advanced = journal.durable_op() != self.durable_head.get() + || journal.checkpoint_op() != self.checkpoint.get() + || journal.certified_log_view() != self.certified_log_view.get() + || (journal.segment_checkpoint().is_some() + && journal.head() != self.written_head.get()); + self.certified_log_view.set(journal.certified_log_view()); + if self + .requested_log_view + .get() + .is_some_and(|(view, _, _)| Some(view) == self.certified_log_view.get()) + { + self.requested_log_view.set(None); + } + self.written_head.set(journal.head()); + self.durable_head.set(journal.durable_op()); + if journal.checkpoint_op() > self.checkpoint.get() { + self.accepted + .borrow_mut() + .checkpoint(journal.checkpoint_op()); + } + self.checkpoint.set(journal.checkpoint_op()); + self.checkpoint_checksum.set(journal.checkpoint_checksum()); + self.purge_generation.set(journal.purge_marker().0); + self.purge_floor.set(journal.purge_marker().1); + if advanced { + self.notify(); + } + } + async fn apply_mutation( &self, journal: &mut PartitionPrepareJournal<S>, @@ -1274,6 +1299,8 @@ impl<S: DurableStorage> PartitionPersistence<S> { // interval between arrivals groups nothing and every prepare pays its // own writes. This wait puts that grouping back under operator control. if durable && let Some(delay) = self.group_commit_wait(&batch, bytes) { + self.group_commit_waits + .set(self.group_commit_waits.get() + 1); compio::runtime::time::sleep(delay).await; self.collect_queued(&mut batch, &mut bytes, &mut durable, epoch); self.in_flight_bytes.set(bytes); diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs index 74943432e..b0906d54b 100644 --- a/core/partitions/src/poll_plan.rs +++ b/core/partitions/src/poll_plan.rs @@ -139,6 +139,10 @@ pub struct DiskReadPlan { /// 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. + pub(crate) bytes_per_message: Option<u32>, } pub struct DiskSegment { @@ -383,16 +387,82 @@ pub enum DiskReadOutcome { Faulted, } +/// 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; + +/// 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 +/// handful of messages would issue a read per batch. +const DISK_POLL_CHUNK_MIN: u64 = 64 << 10; + +/// How the chunk loop over one segment ended. +enum SegmentWalk { + /// The segment is exhausted or the requested count is filled. The walk + /// may continue into the next segment. + Done, + /// Fail-closed: the segment may hold present-but-unreadable or corrupt + /// data, so no later segment may be served over it. + Faulted, +} + +/// The state one disk walk carries across its segments. +struct DiskWalk { + /// Byte offset into the segment being walked; reset at each boundary. + position: u64, + matched: u32, + fragments: PollFragments<4096>, + last_matching_offset: Option<u64>, + #[cfg(feature = "poll-diagnostics")] + requested_bytes: u64, + #[cfg(feature = "poll-diagnostics")] + chunk_reads: u32, +} + +impl DiskWalk { + fn starting_at(position: u64) -> Self { + Self { + position, + matched: 0, + fragments: PollFragments::new(), + last_matching_offset: None, + #[cfg(feature = "poll-diagnostics")] + requested_bytes: 0, + #[cfg(feature = "poll-diagnostics")] + chunk_reads: 0, + } + } +} + 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 + /// 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. + fn chunk_len(&self, remaining: u32) -> u64 { + let Some(bytes_per_message) = self.bytes_per_message else { + return DISK_POLL_CHUNK_MAX; + }; + u64::from(bytes_per_message) + .saturating_mul(u64::from(remaining)) + .saturating_add(COMMAND_HEADER_SIZE as u64) + .clamp(DISK_POLL_CHUNK_MIN, DISK_POLL_CHUNK_MAX) + } + /// Serve a poll from the on-disk segment files, off the partition borrow. /// Reads from owned descriptors so no partition reference is held across /// the file IO. Walks stamped `[256B BatchHeader][blob]` batches in /// chunked reads, re-reading a batch split across a chunk boundary in the /// next chunk. - #[allow(clippy::cast_possible_truncation)] pub(crate) async fn read_disk(self, query: MessageLookup) -> DiskReadOutcome { - const DISK_POLL_CHUNK: u64 = 1 << 20; - let count = query.count(); if count == 0 || self.segments.is_empty() { return DiskReadOutcome::Empty; @@ -430,30 +500,28 @@ 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 mut position = match self.segments.first() { + 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 mut fragments = PollFragments::new(); - let mut last_matching_offset = None; - let mut matched: u32 = 0; + let mut walk = DiskWalk::starting_at(position); // 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. let mut faulted = false; - 'walk: for segment in &self.segments { - if matched >= count { + for segment in &self.segments { + if walk.matched >= count { break; } let persisted = segment.persisted; - if persisted == 0 || position >= persisted { + if persisted == 0 || walk.position >= persisted { // Benign skip: nothing persisted for this segment yet, or the // start position is already past it. Not a fault. - position = 0; + walk.position = 0; continue; } let path = format!("{partition_dir}/{:0>20}.log", segment.start_offset); @@ -461,69 +529,43 @@ impl DiskReadPlan { // Open exhausted retries: the segment may hold present-but- // unreadable data. Stop here rather than walking past it. faulted = true; - break 'walk; + break; }; - let mut chunk_len = DISK_POLL_CHUNK; - while matched < count && position < persisted { - let len = (persisted - position).min(chunk_len) as usize; - let Some(chunk) = self.read_chunk_with_retry(&file, position, len).await else { - // Chunk read exhausted retries: same fail-closed reason as - // a failed open. - faulted = true; - break 'walk; - }; - let fragments_before_chunk = fragments.len(); - let ChunkWalk { consumed, corrupt } = walk_disk_chunk( - &chunk, - query, - count, - &mut matched, - &mut fragments, - &mut last_matching_offset, - if self.validate_checksum { - BatchIntegrity::Verify - } else { - BatchIntegrity::LayoutOnly - }, - self.namespace_raw, - ); - // Detached from the pump, so the ratio alone bounds the copy. - unpin_sparse_source(&mut fragments, fragments_before_chunk, &chunk, usize::MAX); - if corrupt { - // A batch that does not match its own checksum. Fail closed like - // an IO fault: serving it hands a consumer data provably not what - // was written, and skipping ahead punches a silent gap. - faulted = true; - break 'walk; - } - if consumed == 0 { - if (len as u64) >= persisted - position { - // The whole remainder fit yet no complete batch - // decoded: a corrupt batch in this segment. Fail-closed - // like an IO fault (set `faulted`, stop the walk) so a - // later segment is never served over the corrupt run, - // which would punch a silent gap into the poll. - faulted = true; - break 'walk; - } - // A single batch larger than the chunk: grow and re-read. - chunk_len = chunk_len.saturating_mul(4); - continue; - } - chunk_len = DISK_POLL_CHUNK; - position += consumed as u64; + if matches!( + self.walk_segment(&file, query, count, persisted, &mut walk) + .await, + SegmentWalk::Faulted + ) { + faulted = true; + break; } - position = 0; + walk.position = 0; } - if matched > 0 { + // The three ratios a read-sizing change is judged on: bytes asked of + // the file API, bytes actually served, and the reads it took to get + // them. Per poll, so a short run answers whether a sized first read + // pays for itself before anything becomes a permanent counter. + #[cfg(feature = "poll-diagnostics")] + tracing::debug!( + target: "iggy.partitions.poll_diagnostics", + namespace_raw = self.namespace_raw, + requested_bytes = walk.requested_bytes, + served_bytes = walk.fragments.iter().map(|fragment| fragment.len() as u64).sum::<u64>(), + chunk_reads = walk.chunk_reads, + requested_count = count, + matched = walk.matched, + "disk poll read accounting" + ); + + if walk.matched > 0 { // Pre-fault matches are always a contiguous prefix (the walk stops // at the first fault), so a partial result carries no gap. DiskReadOutcome::Matched { - fragments, - last_matching_offset, - matched, + fragments: walk.fragments, + last_matching_offset: walk.last_matching_offset, + matched: walk.matched, } } else if faulted { DiskReadOutcome::Faulted @@ -532,6 +574,74 @@ 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. + #[allow(clippy::cast_possible_truncation)] + async fn walk_segment( + &self, + file: &compio::fs::File, + query: MessageLookup, + count: u32, + persisted: u64, + walk: &mut DiskWalk, + ) -> SegmentWalk { + let mut chunk_len = self.chunk_len(count - walk.matched); + 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 { + // Chunk read exhausted retries: same fail-closed reason as + // a failed open. + return SegmentWalk::Faulted; + }; + let fragments_before_chunk = walk.fragments.len(); + let ChunkWalk { consumed, corrupt } = walk_disk_chunk( + &chunk, + query, + count, + &mut walk.matched, + &mut walk.fragments, + &mut walk.last_matching_offset, + if self.validate_checksum { + BatchIntegrity::Verify + } else { + BatchIntegrity::LayoutOnly + }, + self.namespace_raw, + ); + // Detached from the pump, so the ratio alone bounds the copy. + unpin_sparse_source( + &mut walk.fragments, + fragments_before_chunk, + &chunk, + usize::MAX, + ); + if corrupt { + // A batch that does not match its own checksum. Fail closed like + // an IO fault: serving it hands a consumer data provably not what + // was written, and skipping ahead punches a silent gap. + return SegmentWalk::Faulted; + } + if consumed == 0 { + if (len as u64) >= persisted - walk.position { + // The whole remainder fit yet no complete batch decoded: a + // corrupt batch in this segment. Fail-closed like an IO + // fault so a later segment is never served over the corrupt + // 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); + continue; + } + chunk_len = self.chunk_len(count - walk.matched); + walk.position += consumed as u64; + } + SegmentWalk::Done + } + /// Resolve the read-only descriptor for `segment`'s file. A hit clones the /// cached fd (sharing the kernel fd, no syscall); a miss opens by path and /// stores the fd back so later polls skip the `openat`. Returns `None` only @@ -698,10 +808,16 @@ impl DiskReadPlan { async fn read_chunk_with_retry( &self, file: &compio::fs::File, - position: u64, len: usize, + walk: &mut DiskWalk, ) -> Option<Frozen<4096>> { + let position = walk.position; for attempt in 0..3u8 { + #[cfg(feature = "poll-diagnostics")] + { + walk.requested_bytes += len as u64; + walk.chunk_reads += 1; + } // `with_capacity` (len == 0, capacity == len) instead of `zeroed`: // `read_exact_at` fills the whole capacity in place and advances the // length via `SetLen`, so the `zeroed` memset of up to 1MiB per @@ -846,6 +962,61 @@ mod tests { entry_count } + #[test] + fn chunk_len_sizes_the_first_read_from_the_requested_count() { + let plan = |bytes_per_message| DiskReadPlan { + partition_dir: PartitionDirResolution::NoFiles, + bytes_per_message, + segments: Vec::new(), + start_position: 0, + namespace_raw: 0, + validate_checksum: false, + }; + + // Nothing committed yet, so nothing bridges a count to bytes. + assert_eq!(plan(None).chunk_len(1000), DISK_POLL_CHUNK_MAX); + // A thousand small messages used to read a megabyte to return 150 KB. + assert_eq!( + plan(Some(150)).chunk_len(1000), + 150 * 1000 + COMMAND_HEADER_SIZE as u64 + ); + // The floor keeps a poll for a few messages off a read per batch, the + // ceiling is what every poll read before it was sized at all. + assert_eq!(plan(Some(150)).chunk_len(1), DISK_POLL_CHUNK_MIN); + assert_eq!(plan(Some(64 << 10)).chunk_len(1000), DISK_POLL_CHUNK_MAX); + // Wide messages and a wide count must clamp, never wrap. + assert_eq!( + plan(Some(u32::MAX)).chunk_len(u32::MAX), + DISK_POLL_CHUNK_MAX + ); + } + + #[cfg(feature = "poll-diagnostics")] + #[compio::test] + async fn read_accounting_includes_failed_retry_attempts() { + let directory = tempfile::tempdir().unwrap(); + let file = compio::fs::File::create(directory.path().join("empty.log")) + .await + .unwrap(); + let plan = DiskReadPlan { + partition_dir: PartitionDirResolution::NoFiles, + bytes_per_message: None, + segments: Vec::new(), + start_position: 0, + namespace_raw: 0, + validate_checksum: true, + }; + let mut walk = DiskWalk::starting_at(0); + assert!( + plan.read_chunk_with_retry(&file, 64, &mut walk) + .await + .is_none() + ); + assert_eq!(walk.chunk_reads, 3); + assert_eq!(walk.requested_bytes, 192); + assert_eq!(walk.matched, 0); + } + fn offset_query(offset: u64) -> MessageLookup { MessageLookup::Offset { offset, @@ -876,6 +1047,7 @@ mod tests { }; let plan = DiskReadPlan { partition_dir: PartitionDirResolution::Resolved(dir.display().to_string()), + bytes_per_message: None, segments: Vec::new(), start_position: 0, namespace_raw: 0, diff --git a/core/server/config.toml b/core/server/config.toml index e5009ce26..9ceae5651 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -912,6 +912,9 @@ clients_table_max = 8192 # completed batch, prepare, checkpoint and error counters. # partition_wal_retained_bytes plus queued_bytes and in_flight_bytes measures budget usage; # partition_wal_disk_bytes measures WAL file length, excluding referenced segment bodies. +# partition_wal_prepares over partition_wal_batches is the prepares per durable group, and +# partition_wal_group_commit_waits counts the groups that actually took the optional wait +# below, so a delay that never fires is distinguishable from one that fires and buys nothing. wal_bytes_max = "256 MiB" # Bounded wait for more prepares before a persisted partition's WAL writer @@ -998,6 +1001,10 @@ evicted_ring_capacity = 4096 # Byte ceiling for the evicted ring per partition; whichever ring cap (this or # evicted_ring_capacity) trips first evicts. Bounds the ring memory a burst of # large batches can pin. Must be > 0 and <= "256 MiB". +# PER PARTITION and per replica, so the node-wide ceiling is this times the partitions it +# hosts, pinned as anonymous memory beside the same bytes in page cache. Raising it on a +# node with many partitions is a large memory decision; partition_repair_ring_bytes on the +# metrics endpoint reports what the rings on each shard actually hold. evicted_ring_bytes_max = "16 MiB" # Byte budget for segment payloads a SERVING shard keeps resident to answer diff --git a/core/server_common/src/send_messages.rs b/core/server_common/src/send_messages.rs index 539fbf40a..dca3cd989 100644 --- a/core/server_common/src/send_messages.rs +++ b/core/server_common/src/send_messages.rs @@ -475,17 +475,19 @@ pub fn convert_request_message( return Ok(message); } - admit_wire_request(namespace, body, request_header, checksum) + let admitted = admit_wire_request(body)?; + compact_wire_request(namespace, message, admitted, checksum) } -/// Validate a producer's `[metadata][batch]` body and rebuild it as the -/// pipeline form with the partition stamped. -fn admit_wire_request( - namespace: IggyNamespace, - body: &[u8], - mut request_header: RoutedRequestHeader, - checksum: ChecksumMode, -) -> Result<Message<RoutedRequestHeader>, IggyError> { +/// A validated producer body: where its batch starts inside the body, and the +/// batch header the pipeline form carries once the partition is stamped in. +struct AdmittedBatch { + body_offset: usize, + header: BatchHeader, +} + +/// Validate a producer's `[metadata][batch]` body and locate its batch. +fn admit_wire_request(body: &[u8]) -> Result<AdmittedBatch, IggyError> { if body.len() < 4 { return Err(IggyError::InvalidCommand); } @@ -516,17 +518,45 @@ fn admit_wire_request( return Err(IggyError::InvalidCommand); } + Ok(AdmittedBatch { + body_offset: batch_start, + header: batch.header, + }) +} + +/// Rewrite a validated `[header][metadata][batch]` frame as the pipeline form +/// `[header][batch]`, with the resolved partition stamped in. +/// +/// The metadata prefix is dropped by sliding the batch down over it rather +/// than by filling a second buffer: the frame is already owned and already +/// aligned, and the batch is all but a few dozen bytes of it, so a second +/// allocation copies the same bytes and then frees the original, once per +/// produce. +fn compact_wire_request( + namespace: IggyNamespace, + message: Message<RoutedRequestHeader>, + admitted: AdmittedBatch, + checksum: ChecksumMode, +) -> Result<Message<RoutedRequestHeader>, IggyError> { let header_size = std::mem::size_of::<RoutedRequestHeader>(); - let total_size = header_size + batch.header.total_size(); + let total_size = header_size + admitted.header.total_size(); + let mut request_header = *message.header(); request_header.size = u32::try_from(total_size).map_err(|_| IggyError::InvalidCommand)?; - let mut buffer = Owned::<MESSAGE_ALIGN>::with_capacity(total_size); - buffer.extend_from_slice(bytemuck::bytes_of(&request_header)); - buffer.extend_from_slice(batch_bytes); + + let batch_start = header_size + admitted.body_offset; + let batch_end = batch_start + admitted.header.total_size(); + let mut buffer = message.into_owned(); + buffer + .as_mut_slice() + .copy_within(batch_start..batch_end, header_size); + buffer.truncate(total_size); + let bytes = buffer.as_mut_slice(); + bytes[..header_size].copy_from_slice(bytemuck::bytes_of(&request_header)); // The producer hashed `partition_id = 0`; stamp the resolved partition // and restamp (or clear, for the stamp-fills-it path) the batch checksum. - let mut stamped = batch.header; + let mut stamped = admitted.header; stamped.partition_id = namespace.partition_id() as u64; stamped.batch_checksum = match checksum { ChecksumMode::Compute => { @@ -1089,6 +1119,26 @@ mod tests { ); } + /// Admission slides the batch over the metadata prefix inside the buffer + /// the request arrived in. A second buffer would copy the same bytes and + /// free the original once per produce, and the frame's own length has to + /// follow the prefix it just dropped. + #[test] + fn convert_request_message_admits_wire_body_in_place() { + let namespace = IggyNamespace::new(1, 1, 3); + let wire = wire_request_message(&wire_send_messages_body(&sample_messages())); + let buffer = wire.as_slice().as_ptr(); + let wire_size = wire.header().size as usize; + + let converted = convert_request_message(namespace, wire, ChecksumMode::Compute) + .expect("wire body admits"); + + assert_eq!(converted.as_slice().as_ptr(), buffer); + let size = converted.header().size as usize; + assert!(size < wire_size, "the metadata prefix must be gone"); + assert_eq!(converted.as_slice().len(), size); + } + #[test] fn convert_request_message_rejects_tampered_wire_body() { // A flipped payload byte invalidates the producer's per-message diff --git a/core/shard/Cargo.toml b/core/shard/Cargo.toml index 00cc3d642..2dfbeb77b 100644 --- a/core/shard/Cargo.toml +++ b/core/shard/Cargo.toml @@ -23,7 +23,7 @@ license = "Apache-2.0" publish = false [features] -poll-diagnostics = [] +poll-diagnostics = ["partitions/poll-diagnostics"] # Simulator-only test hook (`IggyShard::init_partition`): bypasses the # reconciler's `ReconcileOp::InsertOwned` funnel, mutating `IggyPartitions` # off the pump task. A `-p iggy-server` build excludes it; `cargo build diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 1fcb8536c..5747214c9 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -7565,9 +7565,14 @@ where } let mut persistence_metrics = partitions::PersistenceMetrics::default(); + let mut repair_ring_entries = 0usize; + let mut repair_ring_bytes = 0u64; for namespace in namespace_scratch.iter() { if let Some(partition) = partitions.get_mut_by_ns(namespace) { partition.drive_persistence().await; + let (entries, bytes) = partition.repair_ring_occupancy(); + repair_ring_entries += entries; + repair_ring_bytes += bytes; if let Some(metrics) = partition.take_persistence_metrics() { persistence_metrics.disk_bytes += metrics.disk_bytes; persistence_metrics.retained_bytes += metrics.retained_bytes; @@ -7576,12 +7581,15 @@ where persistence_metrics.checkpoints_pending += metrics.checkpoints_pending; persistence_metrics.completed_batches += metrics.completed_batches; persistence_metrics.batched_prepares += metrics.batched_prepares; + persistence_metrics.group_commit_waits += metrics.group_commit_waits; persistence_metrics.completed_checkpoints += metrics.completed_checkpoints; persistence_metrics.failed_writes += metrics.failed_writes; } } } self.metrics.record_persistence(&persistence_metrics); + self.metrics + .set_repair_ring(repair_ring_entries, repair_ring_bytes); // Counted at most ONCE per sweep and only if a re-arm actually fires, // then tracked locally as arms land. Counting per namespace is a full diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index c5ee4f29a..a58c01ef1 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -227,6 +227,9 @@ pub struct ShardMetrics { partition_wal_checkpoints_pending: Gauge, partition_wal_batches: Counter, partition_wal_prepares: Counter, + partition_wal_group_commit_waits: Counter, + partition_repair_ring_entries: Gauge, + partition_repair_ring_bytes: Gauge, partition_wal_checkpoints: Counter, partition_wal_errors: Counter, frame_drops: FrameDropMetrics, @@ -295,6 +298,9 @@ impl ShardMetrics { partition_wal_checkpoints_pending: Gauge::default(), partition_wal_batches: Counter::default(), partition_wal_prepares: Counter::default(), + partition_wal_group_commit_waits: Counter::default(), + partition_repair_ring_entries: Gauge::default(), + partition_repair_ring_bytes: Gauge::default(), partition_wal_checkpoints: Counter::default(), partition_wal_errors: Counter::default(), frame_drops: FrameDropMetrics { @@ -334,6 +340,8 @@ impl ShardMetrics { .set(i64::try_from(metrics.checkpoints_pending).unwrap_or(i64::MAX)); self.partition_wal_batches.inc_by(metrics.completed_batches); self.partition_wal_prepares.inc_by(metrics.batched_prepares); + self.partition_wal_group_commit_waits + .inc_by(metrics.group_commit_waits); self.partition_wal_checkpoints .inc_by(metrics.completed_checkpoints); self.partition_wal_errors.inc_by(metrics.failed_writes); @@ -375,6 +383,21 @@ impl ShardMetrics { "prepares covered by completed partition WAL batches", self.partition_wal_prepares.clone(), ); + registry.register( + "partition_wal_group_commit_waits", + "durable groups that took the optional pre-barrier wait", + self.partition_wal_group_commit_waits.clone(), + ); + registry.register( + "partition_repair_ring_entries", + "committed entries this shard's partitions retain for peer repair", + self.partition_repair_ring_entries.clone(), + ); + registry.register( + "partition_repair_ring_bytes", + "bytes this shard's partitions retain for peer repair", + self.partition_repair_ring_bytes.clone(), + ); registry.register( "partition_wal_checkpoints", "completed partition WAL checkpoints", @@ -387,6 +410,18 @@ impl ShardMetrics { ); } + /// Republished by every partition sweep: what the repair rings on this + /// shard actually hold, which the configured per-partition ceilings do not + /// say. The ceilings multiply by the partition count on every replica, so + /// this is the only place an operator can see the real cost of raising + /// them. + pub fn set_repair_ring(&self, entries: usize, bytes: u64) { + self.partition_repair_ring_entries + .set(i64::try_from(entries).unwrap_or(i64::MAX)); + self.partition_repair_ring_bytes + .set(i64::try_from(bytes).unwrap_or(i64::MAX)); + } + /// Count consumer offset capacity denials from explicit client requests /// and automatic commit admission during poll completion. pub fn record_consumer_offset_denied(&self, kind: ConsumerKind) { diff --git a/core/simulator/src/storage/tests.rs b/core/simulator/src/storage/tests.rs index 20a0b77d1..7a63f85f8 100644 --- a/core/simulator/src/storage/tests.rs +++ b/core/simulator/src/storage/tests.rs @@ -36,6 +36,7 @@ use server_common::{ Message, iobuf::{IOV_MAX, Owned}, }; +use std::cell::Cell; use std::collections::BTreeSet; use std::io; use std::path::Path; @@ -1209,6 +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; let unlink = storage .trace() .iter() @@ -1218,6 +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; storage.clear_trace(); let obsolete = Path::new("/partition/wal/prepares-0.wal"); assert!(storage.exists(obsolete).await.unwrap()); @@ -1249,6 +1252,11 @@ fn obsolete_wal_generations_are_reclaimed_after_restart_and_failed_unlink() { .append(prepare(4, parent).into_frozen()) .await .unwrap(); + // Reclamation is not on the append path: the append must not + // 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; } assert!(!storage.exists(obsolete).await.unwrap()); assert_eq!(journal.checkpoint_op(), 2); @@ -1262,6 +1270,35 @@ fn obsolete_wal_generations_are_reclaimed_after_restart_and_failed_unlink() { }); } +#[test] +fn checkpoint_notifies_before_reclaiming_its_obsolete_generation() { + block_on(async { + let (storage, persistence) = queued_batch(4).await; + assert!(persistence.start()); + Rc::clone(&persistence).run().await; + storage.clear_trace(); + let notified = Rc::new(Cell::new(false)); + let observed = Rc::clone(¬ified); + let observed_storage = storage.clone(); + persistence.set_notifier(Rc::new(move |_| { + assert!(!observed_storage.trace().contains(&StorageOperation::Unlink)); + observed.set(true); + })); + persistence.checkpoint(2); + assert!(persistence.start()); + Rc::clone(&persistence).run().await; + assert!(notified.get()); + assert!(storage.trace().contains(&StorageOperation::Unlink)); + assert_eq!(persistence.checkpoint_op(), 2); + storage.crash(Crash::PowerLoss); + let recovered = PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage) + .await + .unwrap(); + assert_eq!(recovered.head(), 4); + assert_eq!(recovered.checkpoint_op(), 2); + }); +} + #[test] fn dropping_a_stalled_writer_restores_ownership_and_releases_drain_waiters() { block_on(async { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java index 656103416..df2378c69 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java @@ -37,6 +37,7 @@ import io.netty.channel.pool.AbstractChannelPoolHandler; import io.netty.channel.pool.ChannelHealthChecker; import io.netty.channel.pool.FixedChannelPool; import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.flush.FlushConsolidationHandler; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; import io.netty.handler.ssl.SslHandler; @@ -99,6 +100,13 @@ public class AsyncTcpConnection { static final int TRANSIENT_NOT_ACCEPTED = 58; // The pool holds one channel, and one channel lives on one loop. static final int DEFAULT_IO_THREADS = 1; + /** + * Writes consolidated into one flush inside a read loop. Netty's own + * default, which is tuned for exactly this shape: several requests written + * while one batch of replies is being dispatched. + */ + private static final int FLUSH_CONSOLIDATION_MAX = 256; + private static final Logger log = LoggerFactory.getLogger(AsyncTcpConnection.class); private static final Duration DEFAULT_CONNECTION_TIMEOUT = Duration.ofMillis(3000); // A missing reply must not hold the single VSR-pinned channel forever. @@ -1031,6 +1039,14 @@ public class AsyncTcpConnection { ssl.setHandshakeTimeoutMillis(dialTimeoutMillis); pipeline.addLast("ssl", ssl); } + // A pipelining producer's next request is usually written from the + // completion of the previous reply, so its flush lands inside the + // read loop that delivered it and several requests leave in one + // syscall instead of one each. Consolidation is confined to that + // read loop: with no read in progress every flush passes straight + // through, so a request on an otherwise idle connection is never + // waiting on later traffic to push it out. + pipeline.addLast("flushConsolidation", new FlushConsolidationHandler(FLUSH_CONSOLIDATION_MAX, false)); pipeline.addLast("frameDecoder", new VsrFrameDecoder(maxVsrFrameSize)); pipeline.addLast("responseHandler", new VsrResponseHandler(consensusSession, onEviction)); } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java index bee4cb825..ba6aeb553 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java @@ -19,6 +19,7 @@ package org.apache.iggy.client.async.tcp; +import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import org.apache.iggy.client.async.ConsumerGroupsClient; import org.apache.iggy.client.async.MessagesClient; @@ -50,8 +51,8 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.function.Supplier; +import static org.apache.iggy.serde.BytesSerializer.encodeMessagesBatchInto; import static org.apache.iggy.serde.BytesSerializer.toBytes; -import static org.apache.iggy.serde.BytesSerializer.toMessagesBatch; /** * Async TCP implementation of MessagesClient using Netty for non-blocking I/O. @@ -183,17 +184,24 @@ public class MessagesTcpClient implements MessagesClient { StreamId streamId, TopicId topicId, Partitioning partitioning, List<Message> messages) { var metadataLength = streamId.getSize() + topicId.getSize() + partitioning.getSize() + 4; - var batch = toMessagesBatch(messages); - var payload = Unpooled.buffer(4 + metadataLength + batch.readableBytes()); - - payload.writeIntLE(metadataLength); - payload.writeBytes(toBytes(streamId)); - payload.writeBytes(toBytes(topicId)); - payload.writeBytes(toBytes(partitioning)); - payload.writeIntLE(messages.size()); - payload.writeBytes(batch); - - return connection().send(CommandCode.Messages.SEND.getValue(), payload).thenApply(response -> { + // The batch is encoded straight after the metadata rather than into its + // own buffer and copied over, which is the whole payload once per send. + var payload = Unpooled.buffer(4 + metadataLength); + + CompletableFuture<ByteBuf> sent; + try { + payload.writeIntLE(metadataLength); + writeAndRelease(payload, toBytes(streamId)); + writeAndRelease(payload, toBytes(topicId)); + writeAndRelease(payload, toBytes(partitioning)); + payload.writeIntLE(messages.size()); + encodeMessagesBatchInto(payload, messages); + sent = connection().send(CommandCode.Messages.SEND.getValue(), payload); + } catch (RuntimeException | Error error) { + payload.release(); + throw error; + } + return sent.thenApply(response -> { try { return BytesDeserializer.readSendMessagesResponse(response); } catch (RuntimeException e) { @@ -208,6 +216,14 @@ public class MessagesTcpClient implements MessagesClient { }); } + private static void writeAndRelease(ByteBuf destination, ByteBuf source) { + try { + destination.writeBytes(source); + } finally { + source.release(); + } + } + /** * One group-poll attempt: sync the assignment when missing or stale, pick * the next assigned partition round-robin, poll it explicitly, and on a 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 51dcaed93..7166861e7 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 @@ -257,6 +257,21 @@ public final class BytesSerializer { * so they are encoded as zero here. */ public static ByteBuf toMessagesBatch(List<Message> messages) { + var rawMessages = toRawMessages(messages); + return encodeBatch(rawMessages); + } + + /** + * Appends the same batch record to {@code out} at its current writer index. A caller that + * has already written the bytes preceding the batch encodes it straight into their buffer + * instead of filling a second one and copying it over, which is the whole payload once per + * request. + */ + public static void encodeMessagesBatchInto(ByteBuf out, List<Message> messages) { + encodeBatchInto(out, toRawMessages(messages)); + } + + private static List<RawMessage> toRawMessages(List<Message> messages) { if (messages.isEmpty()) { throw new IggyInvalidArgumentException("Cannot encode an empty message batch"); } @@ -268,48 +283,83 @@ public final class BytesSerializer { message.payload(), readAllBytes(toBytes(message.userHeaders())))); } - return encodeBatch(rawMessages); + return rawMessages; } static ByteBuf encodeBatch(List<RawMessage> messages) { - var batchOriginTimestamp = messages.stream() - .map(RawMessage::originTimestamp) - .min(BigInteger::compareTo) - .orElseThrow(() -> new IggyInvalidArgumentException("Cannot encode an empty message batch")); - var blobLength = 0; - for (RawMessage message : messages) { - blobLength += MessageHeader.SIZE + message.payload().length + message.userHeaders().length; + var batch = Unpooled.buffer(BATCH_HEADER_SIZE); + try { + encodeBatchInto(batch, messages); + return batch; + } catch (RuntimeException | Error error) { + batch.release(); + throw error; } + } - var batch = Unpooled.buffer(BATCH_HEADER_SIZE + blobLength); - batch.writeZero(BATCH_HEADER_SIZE); - for (int index = 0; index < messages.size(); index++) { - RawMessage message = messages.get(index); - var timestampDelta = message.originTimestamp().subtract(batchOriginTimestamp); - if (timestampDelta.compareTo(MAX_TIMESTAMP_DELTA_MICROS) > 0) { - throw new IggyInvalidArgumentException("Message origin timestamp exceeds the batch origin by " - + timestampDelta + " microseconds, more than the timestamp delta field can hold"); + private static BatchExtent measureBatch(List<RawMessage> messages, long capacityAllowance) { + var originTimestamp = messages.get(0).originTimestamp(); + var latestTimestamp = originTimestamp; + long length = BATCH_HEADER_SIZE; + for (RawMessage message : messages) { + var timestamp = message.originTimestamp(); + if (timestamp.signum() < 0 || timestamp.bitLength() > Long.SIZE) { + throw new IggyInvalidArgumentException("Message origin timestamp is outside unsigned 64-bit range"); + } + originTimestamp = originTimestamp.min(timestamp); + latestTimestamp = latestTimestamp.max(timestamp); + length += (long) MessageHeader.SIZE + message.payload().length + message.userHeaders().length; + if (length > capacityAllowance) { + throw new IggyInvalidArgumentException("Message batch exceeds the output buffer capacity"); } - var frameStart = batch.writerIndex(); - batch.writeLongLE(0); // checksum, backpatched below - batch.writeBytes(message.id()); - batch.writeIntLE(index); // offset_delta - batch.writeIntLE(timestampDelta.intValue()); - batch.writeIntLE(message.userHeaders().length); - batch.writeIntLE(message.payload().length); - batch.writeLongLE(0); // reserved - batch.writeBytes(message.payload()); - batch.writeBytes(message.userHeaders()); - batch.setLongLE( - frameStart, xxHash3(batch, frameStart + Long.BYTES, batch.writerIndex() - frameStart - Long.BYTES)); } + if (latestTimestamp.subtract(originTimestamp).compareTo(MAX_TIMESTAMP_DELTA_MICROS) > 0) { + throw new IggyInvalidArgumentException("Message origin timestamp delta exceeds unsigned 32-bit range"); + } + return new BatchExtent(originTimestamp, length); + } - long batchLength = BATCH_HEADER_SIZE + blobLength; - batch.setBytes(24, toBytesAsU64(batchOriginTimestamp)); - batch.setLongLE(32, batchLength); - batch.setLongLE(40, batchChecksum(batch, batchOriginTimestamp, batchLength, messages)); - batch.setIntLE(48, messages.size()); - return batch; + static void encodeBatchInto(ByteBuf out, List<RawMessage> messages) { + if (messages.isEmpty()) { + throw new IggyInvalidArgumentException("Cannot encode an empty message batch"); + } + var batchStart = out.writerIndex(); + var extent = measureBatch(messages, (long) out.maxCapacity() - batchStart); + var batchOriginTimestamp = extent.originTimestamp(); + var batchLength = extent.length(); + // Size to the exact total before the first batch byte. Letting the + // writes grow the buffer instead rounds up to the next power of two, + // which on a batch just over a megabyte reserves two. + var required = batchStart + (int) batchLength; + if (out.capacity() < required) { + out.capacity(required); + } + try { + out.writeZero(BATCH_HEADER_SIZE); + for (int index = 0; index < messages.size(); index++) { + RawMessage message = messages.get(index); + var timestampDelta = message.originTimestamp().subtract(batchOriginTimestamp); + var frameStart = out.writerIndex(); + out.writeLongLE(0); + out.writeBytes(message.id()); + out.writeIntLE(index); + out.writeIntLE(timestampDelta.intValue()); + out.writeIntLE(message.userHeaders().length); + out.writeIntLE(message.payload().length); + out.writeLongLE(0); + out.writeBytes(message.payload()); + out.writeBytes(message.userHeaders()); + out.setLongLE( + frameStart, xxHash3(out, frameStart + Long.BYTES, out.writerIndex() - frameStart - Long.BYTES)); + } + out.setLongLE(batchStart + 24, batchOriginTimestamp.longValue()); + out.setLongLE(batchStart + 32, batchLength); + out.setLongLE(batchStart + 40, batchChecksum(out, batchStart, batchOriginTimestamp, batchLength, messages)); + out.setIntLE(batchStart + 48, messages.size()); + } catch (RuntimeException | Error error) { + out.writerIndex(batchStart); + throw error; + } } /** @@ -317,20 +367,28 @@ public final class BytesSerializer { * message bodies; bodies are bound through the per-frame checksums. */ private static long batchChecksum( - ByteBuf batch, BigInteger batchOriginTimestamp, long batchLength, List<RawMessage> messages) { + ByteBuf batch, + int batchStart, + BigInteger batchOriginTimestamp, + long batchLength, + List<RawMessage> messages) { var input = Unpooled.buffer(BATCH_CHECKSUM_FIXED_INPUT_BYTES + Long.BYTES * messages.size()); - input.writeLongLE(0); // partition_id - input.writeLongLE(0); // base_offset - input.writeLongLE(0); // base_timestamp - input.writeBytes(toBytesAsU64(batchOriginTimestamp)); - input.writeLongLE(batchLength); - input.writeIntLE(messages.size()); - var frameStart = BATCH_HEADER_SIZE; - for (RawMessage message : messages) { - input.writeLongLE(batch.getLongLE(frameStart)); - frameStart += MessageHeader.SIZE + message.payload().length + message.userHeaders().length; + try { + input.writeLongLE(0); + input.writeLongLE(0); + input.writeLongLE(0); + input.writeLongLE(batchOriginTimestamp.longValue()); + input.writeLongLE(batchLength); + input.writeIntLE(messages.size()); + var frameStart = batchStart + BATCH_HEADER_SIZE; + for (RawMessage message : messages) { + input.writeLongLE(batch.getLongLE(frameStart)); + frameStart += MessageHeader.SIZE + message.payload().length + message.userHeaders().length; + } + return xxHash3(input, 0, input.readableBytes()); + } finally { + input.release(); } - return xxHash3(input, 0, input.readableBytes()); } /** @@ -354,9 +412,13 @@ public final class BytesSerializer { } private static byte[] readAllBytes(ByteBuf buffer) { - var bytes = new byte[buffer.readableBytes()]; - buffer.readBytes(bytes); - return bytes; + try { + var bytes = new byte[buffer.readableBytes()]; + buffer.readBytes(bytes); + return bytes; + } finally { + buffer.release(); + } } /** @@ -378,6 +440,12 @@ public final class BytesSerializer { * One message as it enters the batch encoder: the id already encoded to its 16 wire bytes * and the user headers already encoded to their opaque bytes. */ + /** + * The batch-header values a set of messages implies. Measured before a byte is written so an + * input the wire cannot carry is refused with the output buffer untouched. + */ + private record BatchExtent(BigInteger originTimestamp, long length) {} + record RawMessage(byte[] id, BigInteger originTimestamp, byte[] payload, byte[] userHeaders) { RawMessage { if (id.length != 16) { diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java index 07812d853..d42cfa604 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java @@ -66,6 +66,61 @@ class AsyncTcpConnectionConcurrencyTest { private static final int SEND_MESSAGES_CODE = 101; private static final int TRANSIENT_NOT_COMMITTED = 57; + @Test + void shouldFlushRequestFromReplyCallbackWithoutFurtherInboundTraffic() throws Exception { + InetAddress loopback = InetAddress.getLoopbackAddress(); + try (ServerSocket serverSocket = new ServerSocket(0, 1, loopback)) { + CompletableFuture<Void> server = CompletableFuture.runAsync(() -> { + try (Socket socket = serverSocket.accept()) { + socket.setSoTimeout((int) TimeUnit.SECONDS.toMillis(2)); + InputStream input = socket.getInputStream(); + OutputStream output = socket.getOutputStream(); + Request register = readRequest(input); + writeResponse(output, register, registerBody()); + Request first = readRequest(input); + writeResponse(output, first, new byte[0]); + Request next = readRequest(input); + assertThat(next.requestId()).isNotEqualTo(first.requestId()); + writeResponse(output, next, "done".getBytes(StandardCharsets.UTF_8)); + } catch (IOException error) { + throw new IllegalStateException("Mock VSR server failed", error); + } + }); + AsyncTcpConnection connection = new AsyncTcpConnection( + loopback.getHostAddress(), + serverSocket.getLocalPort(), + false, + Optional.empty(), + new AsyncTcpConnection.TcpConnectionPoolConfig(1000, 50), + Optional.empty(), + 1, + Optional.of(Duration.ofSeconds(1)), + Optional.of(Duration.ofSeconds(2)), + Duration.ofHours(1), + 1024 * 1024, + null, + errorCode -> {}, + ignored -> {}); + try { + connection.connect().get(5, TimeUnit.SECONDS); + connection + .send(LOGIN_CODE, loginPayload()) + .get(5, TimeUnit.SECONDS) + .release(); + var sent = connection + .send(SEND_MESSAGES_CODE, sendMessagesPayload()) + .thenCompose(response -> { + response.release(); + return connection.send(SEND_MESSAGES_CODE, sendMessagesPayload()); + }); + assertResponse(sent, "done"); + server.get(5, TimeUnit.SECONDS); + } finally { + connection.close().get(5, TimeUnit.SECONDS); + } + } + } + @Test void shouldCorrelateConcurrentPartitionResponsesInReverseOrder() throws Exception { InetAddress loopback = InetAddress.getLoopbackAddress(); 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 056e563fd..5351f3c0f 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 @@ -33,6 +33,7 @@ import org.junit.jupiter.api.Test; import java.math.BigInteger; import java.nio.charset.StandardCharsets; +import java.util.Collections; import java.util.List; import java.util.Map; @@ -70,6 +71,98 @@ class MessagesBatchWireFormatTest { assertThat(ByteBufUtil.hexDump(batch)).isEqualTo(PRODUCE_BATCH_ONLY); } + /** + * The send path encodes the batch straight after the request metadata + * instead of into its own buffer, so every absolute offset the encoder + * back-patches has to be relative to where the batch starts rather than to + * the buffer. A prefix of any length must leave the same bytes behind it. + */ + @Test + void shouldEncodeTheSameBatchAfterAPrefix() { + var messages = List.of(message(100, 2000, "first", Map.of()), message(200, 2500, "second", Map.of())); + var standalone = BytesSerializer.toMessagesBatch(messages); + + var prefixed = Unpooled.buffer(); + prefixed.writeBytes("request-metadata".getBytes(StandardCharsets.UTF_8)); + var batchStart = prefixed.writerIndex(); + BytesSerializer.encodeMessagesBatchInto(prefixed, messages); + + assertThat(ByteBufUtil.hexDump(prefixed, batchStart, prefixed.writerIndex() - batchStart)) + .isEqualTo(ByteBufUtil.hexDump(standalone)); + assertThat(prefixed.writerIndex()).isEqualTo(batchStart + standalone.readableBytes()); + } + + @Test + void shouldLeavePrefixAndIndexesUnchangedForInvalidTimestamps() { + var output = Unpooled.buffer(); + try { + output.writeIntLE(0x12345678); + var invalid = List.of(message(1, 0, "a", Map.of()), message(2, 0x1_0000_0000L, "b", Map.of())); + assertThatThrownBy(() -> BytesSerializer.encodeMessagesBatchInto(output, invalid)) + .isInstanceOf(IggyInvalidArgumentException.class); + assertThat(output.writerIndex()).isEqualTo(Integer.BYTES); + assertThat(output.readerIndex()).isZero(); + assertThat(output.getIntLE(0)).isEqualTo(0x12345678); + BytesSerializer.encodeMessagesBatchInto(output, List.of(message(1, 0, "valid", Map.of()))); + assertThat(output.writerIndex()).isGreaterThan(Integer.BYTES); + } finally { + output.release(); + } + } + + @Test + void shouldRejectBatchSizeOverflowBeforeGrowingTheBuffer() { + var output = Unpooled.buffer(4); + try { + output.writeIntLE(0x12345678); + var message = new BytesSerializer.RawMessage(idBytes(1), BigInteger.ZERO, new byte[1 << 20], new byte[0]); + assertThatThrownBy(() -> BytesSerializer.encodeBatchInto(output, Collections.nCopies(2048, message))) + .isInstanceOf(IggyInvalidArgumentException.class); + assertThat(output.capacity()).isEqualTo(4); + assertThat(output.writerIndex()).isEqualTo(4); + assertThat(output.getIntLE(0)).isEqualTo(0x12345678); + } finally { + output.release(); + } + } + + @Test + void shouldRejectInvalidUnsignedTimestampsBeforeWriting() { + for (var timestamp : List.of(BigInteger.valueOf(-1), BigInteger.ONE.shiftLeft(64))) { + var output = Unpooled.buffer(); + try { + output.writeByte(42); + var messages = List.of(new BytesSerializer.RawMessage(idBytes(1), timestamp, new byte[0], new byte[0])); + assertThatThrownBy(() -> BytesSerializer.encodeBatchInto(output, messages)) + .isInstanceOf(IggyInvalidArgumentException.class); + assertThat(output.writerIndex()).isEqualTo(1); + assertThat(output.getByte(0)).isEqualTo((byte) 42); + } finally { + output.release(); + } + } + } + + @Test + void shouldEncodeUnsignedTimestampBoundaryIntoHeapAndDirectBuffers() { + var timestamp = BigInteger.ONE.shiftLeft(64).subtract(BigInteger.ONE); + var messages = + List.of(new BytesSerializer.RawMessage(idBytes(1), timestamp, new byte[] {1, 2, 3}, new byte[0])); + var heap = Unpooled.buffer(); + var direct = Unpooled.directBuffer(); + try { + heap.writeByte(42); + direct.writeByte(42); + BytesSerializer.encodeBatchInto(heap, messages); + BytesSerializer.encodeBatchInto(direct, messages); + assertThat(heap.getLongLE(1 + 24)).isEqualTo(-1L); + assertThat(ByteBufUtil.hexDump(direct)).isEqualTo(ByteBufUtil.hexDump(heap)); + } finally { + heap.release(); + direct.release(); + } + } + @Test void shouldDecodeGoldenPollBody() { var polled = BytesDeserializer.readPolledMessages(Unpooled.wrappedBuffer(ByteBufUtil.decodeHexDump(POLL_BODY)));
