You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "numinnex (via GitHub)" <gi...@apache.org> on 2026/02/13 14:04:40 UTC
[PR] feat(partitions): impl first version of partitions module [iggy]
numinnex opened a new pull request, #2735:
URL: https://github.com/apache/iggy/pull/2735
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): impl first version of partitions module [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #2735:
URL: https://github.com/apache/iggy/pull/2735#issuecomment-3897468229
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/2735?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `37.87393%` with `1163 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 67.56%. Comparing base ([`153cb37`](https://app.codecov.io/gh/apache/iggy/commit/153cb378c49eaca0af03a1040fe2b921dfc54427?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`361555e`](https://app.codecov.io/gh/apache/iggy/commit/361555e963ad01b13d1b2d57bcb80b6ab3e0bb55?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/2735?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [core/partitions/src/iggy\_partitions.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Figgy_partitions.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9pZ2d5X3BhcnRpdGlvbnMucnM=) | 0.00% | [571 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/partitions/src/log.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Flog.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9sb2cucnM=) | 0.00% | [128 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/common/src/types/consensus/header.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Ftypes%2Fconsensus%2Fheader.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3R5cGVzL2NvbnNlbnN1cy9oZWFkZXIucnM=) | 0.00% | [74 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/simulator/src/client.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fsimulator%2Fsrc%2Fclient.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaW11bGF0b3Ivc3JjL2NsaWVudC5ycw==) | 0.00% | [58 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...e/common/src/types/segment\_storage/index\_reader.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Ftypes%2Fsegment_storage%2Findex_reader.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3R5cGVzL3NlZ21lbnRfc3RvcmFnZS9pbmRleF9yZWFkZXIucnM=) | 73.23% | [47 Missing and 10 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/partitions/src/journal.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Fjournal.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9qb3VybmFsLnJz) | 0.00% | [50 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/common/src/types/message/in\_flight.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Ftypes%2Fmessage%2Fin_flight.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3R5cGVzL21lc3NhZ2UvaW5fZmxpZ2h0LnJz) | 38.18% | [34 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [core/simulator/src/replica.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fsimulator%2Fsrc%2Freplica.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaW11bGF0b3Ivc3JjL3JlcGxpY2EucnM=) | 0.00% | [22 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...e/common/src/types/segment\_storage/index\_writer.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Ftypes%2Fsegment_storage%2Findex_writer.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3R5cGVzL3NlZ21lbnRfc3RvcmFnZS9pbmRleF93cml0ZXIucnM=) | 72.85% | [10 Missing and 9 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ommon/src/types/segment\_storage/messages\_writer.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Ftypes%2Fsegment_storage%2Fmessages_writer.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3R5cGVzL3NlZ21lbnRfc3RvcmFnZS9tZXNzYWdlc193cml0ZXIucnM=) | 72.85% | [11 Missing and 8 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [15 more](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
:x: Your patch check has failed because the patch coverage (37.87%) is below the target coverage (50.00%). You can increase the patch coverage or adjust the [target](https://docs.codecov.com/docs/commit-status#target) coverage.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #2735 +/- ##
============================================
- Coverage 68.58% 67.56% -1.03%
Complexity 585 585
============================================
Files 735 738 +3
Lines 60370 61275 +905
Branches 56783 57688 +905
============================================
- Hits 41403 41398 -5
- Misses 16941 17853 +912
+ Partials 2026 2024 -2
```
| [Flag](https://app.codecov.io/gh/apache/iggy/pull/2735/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [rust](https://app.codecov.io/gh/apache/iggy/pull/2735/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `69.00% <37.87%> (-1.12%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/2735?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/common/src/types/consumer/consumer\_offset.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Ftypes%2Fconsumer%2Fconsumer_offset.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3R5cGVzL2NvbnN1bWVyL2NvbnN1bWVyX29mZnNldC5ycw==) | `100.00% <100.00%> (ø)` | |
| [core/server/src/streaming/polling\_consumer.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fstreaming%2Fpolling_consumer.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL3N0cmVhbWluZy9wb2xsaW5nX2NvbnN1bWVyLnJz) | `79.31% <ø> (+3.90%)` | :arrow_up: |
| [core/server/src/streaming/segments/storage.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fstreaming%2Fsegments%2Fstorage.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL3N0cmVhbWluZy9zZWdtZW50cy9zdG9yYWdlLnJz) | `100.00% <ø> (+9.37%)` | :arrow_up: |
| [core/consensus/src/impls.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fconsensus%2Fsrc%2Fimpls.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25zZW5zdXMvc3JjL2ltcGxzLnJz) | `0.00% <0.00%> (ø)` | |
| [core/simulator/src/deps.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fsimulator%2Fsrc%2Fdeps.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaW11bGF0b3Ivc3JjL2RlcHMucnM=) | `0.00% <0.00%> (ø)` | |
| [core/common/src/sharding/namespace.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Fsharding%2Fnamespace.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3NoYXJkaW5nL25hbWVzcGFjZS5ycw==) | `56.25% <0.00%> (-5.82%)` | :arrow_down: |
| [...ore/common/src/types/consumer/consumer\_group\_id.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Ftypes%2Fconsumer%2Fconsumer_group_id.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3R5cGVzL2NvbnN1bWVyL2NvbnN1bWVyX2dyb3VwX2lkLnJz) | `0.00% <0.00%> (ø)` | |
| [core/metadata/src/impls/metadata.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fmetadata%2Fsrc%2Fimpls%2Fmetadata.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXRhZGF0YS9zcmMvaW1wbHMvbWV0YWRhdGEucnM=) | `8.30% <0.00%> (+0.12%)` | :arrow_up: |
| [...ommon/src/types/consumer/consumer\_group\_offsets.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Ftypes%2Fconsumer%2Fconsumer_group_offsets.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3R5cGVzL2NvbnN1bWVyL2NvbnN1bWVyX2dyb3VwX29mZnNldHMucnM=) | `50.00% <50.00%> (ø)` | |
| [core/common/src/types/consumer/consumer\_offsets.rs](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree&filepath=core%2Fcommon%2Fsrc%2Ftypes%2Fconsumer%2Fconsumer_offsets.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb21tb24vc3JjL3R5cGVzL2NvbnN1bWVyL2NvbnN1bWVyX29mZnNldHMucnM=) | `50.00% <50.00%> (ø)` | |
| ... and [18 more](https://app.codecov.io/gh/apache/iggy/pull/2735?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [5 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/2735/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): impl first version of partitions module [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #2735:
URL: https://github.com/apache/iggy/pull/2735
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(partitions): impl first version of partitions module [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #2735:
URL: https://github.com/apache/iggy/pull/2735#discussion_r2812069101
##########
core/partitions/src/iggy_partitions.rs:
##########
@@ -116,4 +180,873 @@ impl IggyPartitions {
Some(partition)
}
+
+ /// Remove multiple partitions at once.
+ pub fn remove_many(&mut self, namespaces: &[IggyNamespace]) -> Vec<IggyPartition> {
+ namespaces.iter().filter_map(|ns| self.remove(ns)).collect()
+ }
+
+ /// Iterate over all namespaces owned by this shard.
+ pub fn namespaces(&self) -> impl Iterator<Item = &IggyNamespace> {
+ self.namespace_to_local.keys()
+ }
+
+ /// Iterate over all (namespace, partition) pairs.
+ pub fn iter(&self) -> impl Iterator<Item = (&IggyNamespace, &IggyPartition)> {
+ self.namespace_to_local
+ .iter()
+ .filter_map(|(ns, idx)| self.partitions().get(**idx).map(|p| (ns, p)))
+ }
+
+ /// Iterate over all (namespace, partition) pairs mutably.
+ pub fn iter_mut(&self) -> impl Iterator<Item = (&IggyNamespace, &mut IggyPartition)> {
+ let partitions = self.partitions_mut();
+ let partitions_ptr = partitions.as_mut_ptr();
+ let partitions_len = partitions.len();
+ self.namespace_to_local.iter().filter_map(move |(ns, idx)| {
+ let i = **idx;
+ if i < partitions_len {
+ // Safety: each LocalIdx is unique, so no two iterations alias the same element.
+ Some((ns, unsafe { &mut *partitions_ptr.add(i) }))
+ } else {
+ None
+ }
+ })
+ }
+
+ /// Get partition by namespace, initializing if not present.
+ pub fn get_or_init<F>(&mut self, namespace: IggyNamespace, init: F) -> &mut IggyPartition
+ where
+ F: FnOnce() -> IggyPartition,
+ {
+ // TODO: get_or_insert
+ if !self.namespace_to_local.contains_key(&namespace) {
+ let partition = init();
+ self.insert(namespace, partition);
+ }
+ let idx = *self.namespace_to_local[&namespace];
+ &mut self.partitions_mut()[idx]
+ }
+
+ /// Initialize a new partition with in-memory storage (for testing/simulation).
+ ///
+ /// This is a simplified version that doesn't create file-backed storage.
+ /// Use `init_partition()` for production use with real files.
+ ///
+ /// TODO: Make the log generic over its storage backend to support both
+ /// in-memory (for testing) and file-backed (for production) storage without
+ /// needing separate initialization methods.
+ pub fn init_partition_in_memory(&mut self, namespace: IggyNamespace) -> LocalIdx {
+ // Check if already initialized
+ if let Some(idx) = self.local_idx(&namespace) {
+ return idx;
+ }
+
+ // Create initial segment with default (in-memory) storage
+ let start_offset = 0;
+ let segment = Segment::new(start_offset, self.config.segment_size);
+ let storage = SegmentStorage::default();
+
+ // Create partition with initialized log
+ let stats = Arc::new(PartitionStats::default());
+ let mut partition = IggyPartition::new(stats.clone());
+ partition.log.add_persisted_segment(segment, storage);
+ partition.offset.store(start_offset, Ordering::Relaxed);
+ partition
+ .dirty_offset
+ .store(start_offset, Ordering::Relaxed);
+ partition.should_increment_offset = false;
+ partition.stats.increment_segments_count(1);
+
+ // Insert and return local index
+ self.insert(namespace, partition)
+ }
+
+ /// Initialize a new partition with file-backed storage.
+ ///
+ /// This is the data plane initialization - creates the partition structure,
+ /// initial segment, and storage. Skips the control plane metadata broadcasting.
+ ///
+ /// Corresponds to the "INITIATE PARTITION" phase in the server's flow:
+ /// 1. Control plane: create PartitionMeta (SKIPPED in this method)
+ /// 2. Control plane: broadcast to shards (SKIPPED in this method)
+ /// 3. Data plane: INITIATE PARTITION (THIS METHOD)
+ ///
+ /// Idempotent - returns existing LocalIdx if partition already exists.
+ pub async fn init_partition(&mut self, namespace: IggyNamespace) -> LocalIdx {
+ // Check if already initialized
+ if let Some(idx) = self.local_idx(&namespace) {
+ return idx;
+ }
+
+ // Create initial segment with storage
+ let start_offset = 0;
+ let segment = Segment::new(start_offset, self.config.segment_size);
+
+ // TODO: Waiting for issue to move server config to shared module.
+ // Once complete, paths will come from proper base_path/streams_path/etc config fields.
+ let messages_path = self.config.get_messages_path(
+ namespace.stream_id(),
+ namespace.topic_id(),
+ namespace.partition_id(),
+ start_offset,
+ );
+ let index_path = self.config.get_index_path(
+ namespace.stream_id(),
+ namespace.topic_id(),
+ namespace.partition_id(),
+ start_offset,
+ );
+
+ let storage = SegmentStorage::new(
+ &messages_path,
+ &index_path,
+ 0, // messages_size (new segment)
+ 0, // indexes_size (new segment)
+ self.config.enforce_fsync,
+ self.config.enforce_fsync,
+ false, // file_exists (new segment)
+ )
+ .await
+ .expect("Failed to create segment storage");
+
+ // Create partition with initialized log
+ let stats = Arc::new(PartitionStats::default());
+ let mut partition = IggyPartition::new(stats.clone());
+ partition.log.add_persisted_segment(segment, storage);
+ partition.offset.store(start_offset, Ordering::Relaxed);
+ partition
+ .dirty_offset
+ .store(start_offset, Ordering::Relaxed);
+ partition.should_increment_offset = false;
+ partition.stats.increment_segments_count(1);
+
+ // Insert and return local index
+ self.insert(namespace, partition)
+ }
+}
+
+impl<B> Partitions<VsrConsensus<B>> for IggyPartitions<VsrConsensus<B>>
+where
+ B: MessageBus<Replica = u8, Data = Message<GenericHeader>, Client = u128>,
+{
+ async fn on_request(&self, message: <VsrConsensus<B> as Consensus>::RequestMessage) {
+ let consensus = self.consensus.as_ref().unwrap();
+
+ debug!("handling partition request");
+ let prepare = message.project(consensus);
+ self.pipeline_prepare(prepare).await;
+ }
+
+ async fn on_replicate(&self, message: <VsrConsensus<B> as Consensus>::ReplicateMessage) {
+ let consensus = self.consensus.as_ref().unwrap();
+
+ let header = message.header();
+
+ assert_eq!(header.command, Command2::Prepare);
+
+ if !self.fence_old_prepare(&message) {
+ self.replicate(message.clone()).await;
+ } else {
+ warn!("received old prepare, not replicating");
+ }
+
+ // If syncing, ignore the replicate message.
+ if consensus.is_syncing() {
+ warn!(
+ replica = consensus.replica(),
+ "on_replicate: ignoring (sync)"
+ );
+ return;
+ }
+
+ let current_op = consensus.sequencer().current_sequence();
+
+ // If status is not normal, ignore the replicate.
+ if consensus.status() != Status::Normal {
+ warn!(
+ replica = consensus.replica(),
+ "on_replicate: ignoring (not normal state)"
+ );
+ return;
+ }
+
+ // If message from future view, we ignore the replicate.
+ if header.view > consensus.view() {
+ warn!(
+ replica = consensus.replica(),
+ "on_replicate: ignoring (newer view)"
+ );
+ return;
+ }
+
+ // If we are a follower, we advance the commit number.
+ if consensus.is_follower() {
+ consensus.advance_commit_number(message.header().commit);
+ }
+
+ //
+ assert_eq!(header.op, current_op + 1);
Review Comment:
i think this should be debug_assert, not actual assert in production
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org