13335910144

Committed 14 Feb 2025 07:17PM UTC coverage: 74.72% (+0.03%) from 74.693%

Build # 13335910144

Build Type

push

github

Committed by

web-flow

Commit Message

fix(server): don't panic during segment deletion, improve max topic size (#1532)

This commit addresses a panic issue that occurred during segment
deletion and closure by ensuring proper shutdown of reading and writing tasks.

It also improves the handling of maximum topic size by allowing unlimited
size and ensuring that the oldest segments are removed by a background
job when necessary.

Additionally, the logging level for certain operations has been adjusted 
to provide more detailed trace information.

These changes enhance the stability and flexibility of the system
in managing topic segments and their lifecycle.

Run Details

58 of 72 new or added lines in 5 files covered. (80.56%)

4 existing lines in 3 files now uncovered.

25247 of 33789 relevant lines covered (74.72%)

10277.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.86

/server/src/streaming/segments/segment.rs

use super::indexes::*;
use super::logs::*;
use crate::configs::system::SystemConfig;
use crate::streaming::batching::batch_accumulator::BatchAccumulator;
use crate::streaming::segments::*;
use error_set::ErrContext;
use iggy::error::IggyError;
use iggy::utils::byte_size::IggyByteSize;
use iggy::utils::expiry::IggyExpiry;
use iggy::utils::timestamp::IggyTimestamp;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tokio::fs::remove_file;
use tracing::{info, warn};

#[derive(Debug)]
pub struct Segment {
    pub stream_id: u32,
    pub topic_id: u32,
    pub partition_id: u32,
    pub start_offset: u64,
    pub end_offset: u64,
    pub current_offset: u64,
    pub index_path: String,
    pub log_path: String,
    pub size_bytes: IggyByteSize,
    pub last_index_position: u32,
    pub max_size_bytes: IggyByteSize,
    pub size_of_parent_stream: Arc<AtomicU64>,
    pub size_of_parent_topic: Arc<AtomicU64>,
    pub size_of_parent_partition: Arc<AtomicU64>,
    pub messages_count_of_parent_stream: Arc<AtomicU64>,
    pub messages_count_of_parent_topic: Arc<AtomicU64>,
    pub messages_count_of_parent_partition: Arc<AtomicU64>,
    pub is_closed: bool,
    pub log_writer: Option<SegmentLogWriter>,
    pub log_reader: Option<SegmentLogReader>,
    pub index_writer: Option<SegmentIndexWriter>,
    pub index_reader: Option<SegmentIndexReader>,
    pub message_expiry: IggyExpiry,
    pub unsaved_messages: Option<BatchAccumulator>,
    pub config: Arc<SystemConfig>,
    pub indexes: Option<Vec<Index>>,
    pub log_size_bytes: Arc<AtomicU64>,
    pub index_size_bytes: Arc<AtomicU64>,
}

impl Segment {
    #[allow(clippy::too_many_arguments)]
    pub fn create(
        stream_id: u32,
        topic_id: u32,
        partition_id: u32,
        start_offset: u64,
        config: Arc<SystemConfig>,
        message_expiry: IggyExpiry,
        size_of_parent_stream: Arc<AtomicU64>,
        size_of_parent_topic: Arc<AtomicU64>,
        size_of_parent_partition: Arc<AtomicU64>,
        messages_count_of_parent_stream: Arc<AtomicU64>,
        messages_count_of_parent_topic: Arc<AtomicU64>,
        messages_count_of_parent_partition: Arc<AtomicU64>,
    ) -> Segment {
        let path = config.get_segment_path(stream_id, topic_id, partition_id, start_offset);
        let log_path = Self::get_log_path(&path);
        let index_path = Self::get_index_path(&path);
        let message_expiry = match message_expiry {
            IggyExpiry::ServerDefault => config.segment.message_expiry,
            _ => message_expiry,
        };
        let indexes = match config.segment.cache_indexes {
            true => Some(Vec::new()),
            false => None,
        };

        Segment {
            stream_id,
            topic_id,
            partition_id,
            start_offset,
            end_offset: 0,
            current_offset: start_offset,
            log_path,
            index_path,
            size_bytes: IggyByteSize::from(0),
            last_index_position: 0,
            max_size_bytes: config.segment.size,
            message_expiry,
            indexes,
            unsaved_messages: None,
            is_closed: false,
            log_writer: None,
            log_reader: None,
            index_writer: None,
            index_reader: None,
            size_of_parent_stream,
            size_of_parent_partition,
            size_of_parent_topic,
            messages_count_of_parent_stream,
            messages_count_of_parent_topic,
            messages_count_of_parent_partition,
            config,
            log_size_bytes: Arc::new(AtomicU64::new(0)),
            index_size_bytes: Arc::new(AtomicU64::new(0)),
        }
    }

    /// Load the segment state from disk.
    pub async fn load_from_disk(&mut self) -> Result<(), IggyError> {
        info!(
            "Loading segment from disk: log_path: {}, index_path: {}",
            self.log_path, self.index_path
        );

        if self.log_reader.is_none() || self.index_reader.is_none() {
            self.initialize_writing().await?;
            self.initialize_reading().await?;
        }

        let log_size_bytes = self.log_size_bytes.load(Ordering::Acquire);
        info!("Log file size: {}", IggyByteSize::from(log_size_bytes));

        // TODO(hubcio): in future, remove size_bytes and use only atomic log_size_bytes everywhere
        self.size_bytes = IggyByteSize::from(log_size_bytes);
        self.last_index_position = log_size_bytes as _;

        if self.config.segment.cache_indexes {
            self.indexes = Some(
                self.index_reader
                    .as_ref()
                    .unwrap()
                    .load_all_indexes_impl()
                    .await
                    .with_error_context(|e| {
                        format!("Failed to load indexes, error: {e} for {}", self)
                    })
                    .map_err(|_| IggyError::CannotReadFile)?,
            );

            let last_index_offset = if self.indexes.as_ref().unwrap().is_empty() {
                0_u64
            } else {
                self.indexes.as_ref().unwrap().last().unwrap().offset as u64
            };

            self.current_offset = self.start_offset + last_index_offset;

            info!(
                "Loaded {} indexes for segment with start offset: {} and partition with ID: {} for topic with ID: {} and stream with ID: {}.",
                self.indexes.as_ref().unwrap().len(),
                self.start_offset,
                self.partition_id,
                self.topic_id,
                self.stream_id
            );
        }

        if self.is_full().await {
            self.is_closed = true;
        }

        let messages_count = self.get_messages_count();

        info!(
            "Loaded segment with log file of size {} ({} messages) for start offset {}, current offset: {}, and partition with ID: {} for topic with ID: {} and stream with ID: {}.",
            IggyByteSize::from(log_size_bytes),
            messages_count,
            self.start_offset,
            self.current_offset,
            self.partition_id,
            self.topic_id,
            self.stream_id
        );

        self.size_of_parent_stream
            .fetch_add(log_size_bytes, Ordering::SeqCst);
        self.size_of_parent_topic
            .fetch_add(log_size_bytes, Ordering::SeqCst);
        self.size_of_parent_partition
            .fetch_add(log_size_bytes, Ordering::SeqCst);
        self.messages_count_of_parent_stream
            .fetch_add(messages_count, Ordering::SeqCst);
        self.messages_count_of_parent_topic
            .fetch_add(messages_count, Ordering::SeqCst);
        self.messages_count_of_parent_partition
            .fetch_add(messages_count, Ordering::SeqCst);

        Ok(())
    }

    /// Save the segment state to disk.
    pub async fn persist(&mut self) -> Result<(), IggyError> {
        info!("Saving segment with start offset: {} for partition with ID: {} for topic with ID: {} and stream with ID: {}",
            self.start_offset, self.partition_id, self.topic_id, self.stream_id);
        self.initialize_writing().await?;
        self.initialize_reading().await?;
        info!("Saved segment log file with start offset: {} for partition with ID: {} for topic with ID: {} and stream with ID: {}",
            self.start_offset, self.partition_id, self.topic_id, self.stream_id);
        Ok(())
    }

    pub async fn initialize_writing(&mut self) -> Result<(), IggyError> {
        // TODO(hubcio): consider splitting enforce_fsync for index/log to separate entries in config
        let log_fsync = self.config.partition.enforce_fsync;
        let index_fsync = self.config.partition.enforce_fsync;

        let server_confirmation = self.config.segment.server_confirmation;
        let max_file_operation_retries = self.config.state.max_file_operation_retries;
        let retry_delay = self.config.state.retry_delay;

        let log_writer = SegmentLogWriter::new(
            &self.log_path,
            self.log_size_bytes.clone(),
            log_fsync,
            server_confirmation,
            max_file_operation_retries,
            retry_delay,
        )
        .await?;

        let index_writer =
            SegmentIndexWriter::new(&self.index_path, self.index_size_bytes.clone(), index_fsync)
                .await?;

        self.log_writer = Some(log_writer);
        self.index_writer = Some(index_writer);
        Ok(())
    }

    pub async fn initialize_reading(&mut self) -> Result<(), IggyError> {
        let log_reader = SegmentLogReader::new(&self.log_path, self.log_size_bytes.clone()).await?;
        // TODO(hubcio): there is no need to store open fd for reader if we have index cache enabled
        let index_reader =
            SegmentIndexReader::new(&self.index_path, self.index_size_bytes.clone()).await?;

        self.log_reader = Some(log_reader);
        self.index_reader = Some(index_reader);
        Ok(())
    }

    pub async fn is_full(&self) -> bool {
        if self.size_bytes >= self.max_size_bytes {
            return true;
        }

        self.is_expired(IggyTimestamp::now()).await
    }

    pub async fn is_expired(&self, now: IggyTimestamp) -> bool {
        if !self.is_closed {
            return false;
        }

        match self.message_expiry {
            IggyExpiry::NeverExpire => false,
            IggyExpiry::ServerDefault => false,
            IggyExpiry::ExpireDuration(expiry) => {
                let last_messages = self.get_messages(self.current_offset, 1).await;
                if last_messages.is_err() {
                    return false;
                }

                let last_messages = last_messages.unwrap();
                if last_messages.is_empty() {
                    return false;
                }

                let last_message = &last_messages[0];
                let last_message_timestamp = last_message.timestamp;
                last_message_timestamp + expiry.as_micros() <= now.as_micros()
            }
        }
    }

    pub async fn shutdown_reading(&mut self) {
        if let Some(log_reader) = self.log_reader.take() {
            drop(log_reader);
        }
        if let Some(index_reader) = self.index_reader.take() {
            drop(index_reader);
        }
    }

    pub async fn shutdown_writing(&mut self) {
        if let Some(log_writer) = self.log_writer.take() {
            tokio::spawn(async move {
                let _ = log_writer.fsync().await;
                log_writer.shutdown_persister_task().await;
            });
        } else {
            warn!(
                "Log writer already closed when calling close() for {}",
                self
            );
        }

        if let Some(index_writer) = self.index_writer.take() {
            tokio::spawn(async move {
                let _ = index_writer.fsync().await;
                drop(index_writer)
            });
        } else {
            warn!("Index writer already closed when calling close()");
        }
    }

    pub async fn delete(&mut self) -> Result<(), IggyError> {
        let segment_size = self.size_bytes;
        let segment_count_of_messages = self.get_messages_count();
        info!(
            "Deleting segment of size {segment_size} with start offset: {} for partition with ID: {} for stream with ID: {} and topic with ID: {}...",
            self.start_offset, self.partition_id, self.stream_id, self.topic_id,
        );

        self.shutdown_reading().await;

        if !self.is_closed {
            self.shutdown_writing().await;
        }

        let _ = remove_file(&self.log_path).await.with_error_context(|e| {
            format!("Failed to delete log file: {}, error: {e}", self.log_path)
        });
        let _ = remove_file(&self.index_path).await.with_error_context(|e| {
            format!(
                "Failed to delete index file: {}, error: {e}",
                self.index_path
            )
        });

        let segment_size_bytes = self.size_bytes.as_bytes_u64();
        self.size_of_parent_stream
            .fetch_sub(segment_size_bytes, Ordering::SeqCst);
        self.size_of_parent_topic
            .fetch_sub(segment_size_bytes, Ordering::SeqCst);
        self.size_of_parent_partition
            .fetch_sub(segment_size_bytes, Ordering::SeqCst);
        self.messages_count_of_parent_stream
            .fetch_sub(segment_count_of_messages, Ordering::SeqCst);
        self.messages_count_of_parent_topic
            .fetch_sub(segment_count_of_messages, Ordering::SeqCst);
        self.messages_count_of_parent_partition
            .fetch_sub(segment_count_of_messages, Ordering::SeqCst);

        info!(
            "Deleted segment of size {segment_size} with start offset: {} for partition with ID: {} for stream with ID: {} and topic with ID: {}.",
            self.start_offset, self.partition_id, self.stream_id, self.topic_id,
        );

        Ok(())
    }

    fn get_log_path(path: &str) -> String {
        format!("{}.{}", path, LOG_EXTENSION)
    }

    fn get_index_path(path: &str) -> String {
        format!("{}.{}", path, INDEX_EXTENSION)
    }
}

impl std::fmt::Display for Segment {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "Segment {{ stream_id: {}, topic_id: {}, partition_id: {}, start_offset: {}, end_offset: {}, current_offset: {}, size_bytes: {}, last_index_position: {}, max_size_bytes: {}, closed: {} }}",
            self.stream_id,
            self.topic_id,
            self.partition_id,
            self.start_offset,
            self.end_offset,
            self.current_offset,
            self.size_bytes,
            self.last_index_position,
            self.max_size_bytes,
            self.is_closed
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::configs::system::SegmentConfig;
    use iggy::utils::duration::IggyDuration;

    #[tokio::test]
    async fn should_be_created_given_valid_parameters() {
        let stream_id = 1;
        let topic_id = 2;
        let partition_id = 3;
        let start_offset = 0;
        let config = Arc::new(SystemConfig::default());
        let path = config.get_segment_path(stream_id, topic_id, partition_id, start_offset);
        let log_path = Segment::get_log_path(&path);
        let index_path = Segment::get_index_path(&path);
        let message_expiry = IggyExpiry::ExpireDuration(IggyDuration::from(10));
        let size_of_parent_stream = Arc::new(AtomicU64::new(0));
        let size_of_parent_topic = Arc::new(AtomicU64::new(0));
        let size_of_parent_partition = Arc::new(AtomicU64::new(0));
        let messages_count_of_parent_stream = Arc::new(AtomicU64::new(0));
        let messages_count_of_parent_topic = Arc::new(AtomicU64::new(0));
        let messages_count_of_parent_partition = Arc::new(AtomicU64::new(0));

        let segment = Segment::create(
            stream_id,
            topic_id,
            partition_id,
            start_offset,
            config,
            message_expiry,
            size_of_parent_stream,
            size_of_parent_topic,
            size_of_parent_partition,
            messages_count_of_parent_stream,
            messages_count_of_parent_topic,
            messages_count_of_parent_partition,
        );

        assert_eq!(segment.stream_id, stream_id);
        assert_eq!(segment.topic_id, topic_id);
        assert_eq!(segment.partition_id, partition_id);
        assert_eq!(segment.start_offset, start_offset);
        assert_eq!(segment.current_offset, 0);
        assert_eq!(segment.end_offset, 0);
        assert_eq!(segment.size_bytes, 0);
        assert_eq!(segment.log_path, log_path);
        assert_eq!(segment.index_path, index_path);
        assert_eq!(segment.message_expiry, message_expiry);
        assert!(segment.unsaved_messages.is_none());
        assert!(segment.indexes.is_some());
        assert!(!segment.is_closed);
        assert!(!segment.is_full().await);
    }

    #[tokio::test]
    async fn should_not_initialize_indexes_cache_when_disabled() {
        let stream_id = 1;
        let topic_id = 2;
        let partition_id = 3;
        let start_offset = 0;
        let config = Arc::new(SystemConfig {
            segment: SegmentConfig {
                cache_indexes: false,
                ..Default::default()
            },
            ..Default::default()
        });
        let message_expiry = IggyExpiry::NeverExpire;
        let size_of_parent_stream = Arc::new(AtomicU64::new(0));
        let size_of_parent_topic = Arc::new(AtomicU64::new(0));
        let size_of_parent_partition = Arc::new(AtomicU64::new(0));
        let messages_count_of_parent_stream = Arc::new(AtomicU64::new(0));
        let messages_count_of_parent_topic = Arc::new(AtomicU64::new(0));
        let messages_count_of_parent_partition = Arc::new(AtomicU64::new(0));

        let segment = Segment::create(
            stream_id,
            topic_id,
            partition_id,
            start_offset,
            config,
            message_expiry,
            size_of_parent_stream,
            size_of_parent_topic,
            size_of_parent_partition,
            messages_count_of_parent_stream,
            messages_count_of_parent_topic,
            messages_count_of_parent_partition,
        );

        assert!(segment.indexes.is_none());
    }
}

1	use super::indexes::*;
2	use super::logs::*;
3	use crate::configs::system::SystemConfig;
4	use crate::streaming::batching::batch_accumulator::BatchAccumulator;
5	use crate::streaming::segments::*;
6	use error_set::ErrContext;
7	use iggy::error::IggyError;
8	use iggy::utils::byte_size::IggyByteSize;
9	use iggy::utils::expiry::IggyExpiry;
10	use iggy::utils::timestamp::IggyTimestamp;
11	use std::sync::atomic::{AtomicU64, Ordering};
12	use std::sync::Arc;
13	use tokio::fs::remove_file;
14	use tracing::{info, warn};
15
16	#[derive(Debug)]
17	pub struct Segment {
18	pub stream_id: u32,
19	pub topic_id: u32,
20	pub partition_id: u32,
21	pub start_offset: u64,
22	pub end_offset: u64,
23	pub current_offset: u64,
24	pub index_path: String,
25	pub log_path: String,
26	pub size_bytes: IggyByteSize,
27	pub last_index_position: u32,
28	pub max_size_bytes: IggyByteSize,
29	pub size_of_parent_stream: Arc<AtomicU64>,
30	pub size_of_parent_topic: Arc<AtomicU64>,
31	pub size_of_parent_partition: Arc<AtomicU64>,
32	pub messages_count_of_parent_stream: Arc<AtomicU64>,
33	pub messages_count_of_parent_topic: Arc<AtomicU64>,
34	pub messages_count_of_parent_partition: Arc<AtomicU64>,
35	pub is_closed: bool,
36	pub log_writer: Option<SegmentLogWriter>,
37	pub log_reader: Option<SegmentLogReader>,
38	pub index_writer: Option<SegmentIndexWriter>,
39	pub index_reader: Option<SegmentIndexReader>,
40	pub message_expiry: IggyExpiry,
41	pub unsaved_messages: Option<BatchAccumulator>,
42	pub config: Arc<SystemConfig>,
43	pub indexes: Option<Vec<Index>>,
44	pub log_size_bytes: Arc<AtomicU64>,
45	pub index_size_bytes: Arc<AtomicU64>,
46	}
47
48	impl Segment {
49	#[allow(clippy::too_many_arguments)]
50	pub fn create(	876✔
51	stream_id: u32,	876✔
52	topic_id: u32,	876✔
53	partition_id: u32,	876✔
54	start_offset: u64,	876✔
55	config: Arc<SystemConfig>,	876✔
56	message_expiry: IggyExpiry,	876✔
57	size_of_parent_stream: Arc<AtomicU64>,	876✔
58	size_of_parent_topic: Arc<AtomicU64>,	876✔
59	size_of_parent_partition: Arc<AtomicU64>,	876✔
60	messages_count_of_parent_stream: Arc<AtomicU64>,	876✔
61	messages_count_of_parent_topic: Arc<AtomicU64>,	876✔
62	messages_count_of_parent_partition: Arc<AtomicU64>,	876✔
63	) -> Segment {	876✔
64	let path = config.get_segment_path(stream_id, topic_id, partition_id, start_offset);	876✔
65	let log_path = Self::get_log_path(&path);	876✔
66	let index_path = Self::get_index_path(&path);	876✔
67	let message_expiry = match message_expiry {	876✔
68	IggyExpiry::ServerDefault => config.segment.message_expiry,	×
69	_ => message_expiry,	876✔
70	};
71	let indexes = match config.segment.cache_indexes {	876✔
72	true => Some(Vec::new()),	875✔
73	false => None,	1✔
74	};
75
76	Segment {	876✔
77	stream_id,	876✔
78	topic_id,	876✔
79	partition_id,	876✔
80	start_offset,	876✔
81	end_offset: 0,	876✔
82	current_offset: start_offset,	876✔
83	log_path,	876✔
84	index_path,	876✔
85	size_bytes: IggyByteSize::from(0),	876✔
86	last_index_position: 0,	876✔
87	max_size_bytes: config.segment.size,	876✔
88	message_expiry,	876✔
89	indexes,	876✔
90	unsaved_messages: None,	876✔
91	is_closed: false,	876✔
92	log_writer: None,	876✔
93	log_reader: None,	876✔
94	index_writer: None,	876✔
95	index_reader: None,	876✔
96	size_of_parent_stream,	876✔
97	size_of_parent_partition,	876✔
98	size_of_parent_topic,	876✔
99	messages_count_of_parent_stream,	876✔
100	messages_count_of_parent_topic,	876✔
101	messages_count_of_parent_partition,	876✔
102	config,	876✔
103	log_size_bytes: Arc::new(AtomicU64::new(0)),	876✔
104	index_size_bytes: Arc::new(AtomicU64::new(0)),	876✔
105	}	876✔
106	}	876✔
107
108	/// Load the segment state from disk.
109	pub async fn load_from_disk(&mut self) -> Result<(), IggyError> {	57✔
110	info!(	57✔
111	"Loading segment from disk: log_path: {}, index_path: {}",	×
112	self.log_path, self.index_path
113	);
114
115	if self.log_reader.is_none() \|\| self.index_reader.is_none() {	57✔
116	self.initialize_writing().await?;	57✔
117	self.initialize_reading().await?;	57✔
118	}	×
119
120	let log_size_bytes = self.log_size_bytes.load(Ordering::Acquire);	57✔
121	info!("Log file size: {}", IggyByteSize::from(log_size_bytes));	57✔
122
123	// TODO(hubcio): in future, remove size_bytes and use only atomic log_size_bytes everywhere
124	self.size_bytes = IggyByteSize::from(log_size_bytes);	57✔
125	self.last_index_position = log_size_bytes as _;	57✔
126		57✔
127	if self.config.segment.cache_indexes {	57✔
128	self.indexes = Some(	57✔
129	self.index_reader	57✔
130	.as_ref()	57✔
131	.unwrap()	57✔
132	.load_all_indexes_impl()	57✔
133	.await	57✔
134	.with_error_context(\|e\| {	57✔
135	format!("Failed to load indexes, error: {e} for {}", self)	×
136	})	57✔
137	.map_err(\|_\| IggyError::CannotReadFile)?,	57✔
138	);
139
140	let last_index_offset = if self.indexes.as_ref().unwrap().is_empty() {	57✔
141	0_u64	46✔
142	} else {
143	self.indexes.as_ref().unwrap().last().unwrap().offset as u64	11✔
144	};
145
146	self.current_offset = self.start_offset + last_index_offset;	57✔
147		57✔
148	info!(	57✔
149	"Loaded {} indexes for segment with start offset: {} and partition with ID: {} for topic with ID: {} and stream with ID: {}.",	×
150	self.indexes.as_ref().unwrap().len(),	×
151	self.start_offset,
152	self.partition_id,
153	self.topic_id,
154	self.stream_id
155	);
156	}	×
157
158	if self.is_full().await {	57✔
159	self.is_closed = true;	×
160	}	57✔
161
162	let messages_count = self.get_messages_count();	57✔
163		57✔
164	info!(	57✔
165	"Loaded segment with log file of size {} ({} messages) for start offset {}, current offset: {}, and partition with ID: {} for topic with ID: {} and stream with ID: {}.",	×
166	IggyByteSize::from(log_size_bytes),	×
167	messages_count,
168	self.start_offset,
169	self.current_offset,
170	self.partition_id,
171	self.topic_id,
172	self.stream_id
173	);
174
175	self.size_of_parent_stream	57✔
176	.fetch_add(log_size_bytes, Ordering::SeqCst);	57✔
177	self.size_of_parent_topic	57✔
178	.fetch_add(log_size_bytes, Ordering::SeqCst);	57✔
179	self.size_of_parent_partition	57✔
180	.fetch_add(log_size_bytes, Ordering::SeqCst);	57✔
181	self.messages_count_of_parent_stream	57✔
182	.fetch_add(messages_count, Ordering::SeqCst);	57✔
183	self.messages_count_of_parent_topic	57✔
184	.fetch_add(messages_count, Ordering::SeqCst);	57✔
185	self.messages_count_of_parent_partition	57✔
186	.fetch_add(messages_count, Ordering::SeqCst);	57✔
187		57✔
188	Ok(())	57✔
189	}	57✔
190
191	/// Save the segment state to disk.
192	pub async fn persist(&mut self) -> Result<(), IggyError> {	800✔
193	info!("Saving segment with start offset: {} for partition with ID: {} for topic with ID: {} and stream with ID: {}",	800✔
194	self.start_offset, self.partition_id, self.topic_id, self.stream_id);
195	self.initialize_writing().await?;	800✔
196	self.initialize_reading().await?;	800✔
197	info!("Saved segment log file with start offset: {} for partition with ID: {} for topic with ID: {} and stream with ID: {}",	800✔
198	self.start_offset, self.partition_id, self.topic_id, self.stream_id);
199	Ok(())	800✔
200	}	800✔
201
202	pub async fn initialize_writing(&mut self) -> Result<(), IggyError> {	857✔
203	// TODO(hubcio): consider splitting enforce_fsync for index/log to separate entries in config	857✔
204	let log_fsync = self.config.partition.enforce_fsync;	857✔
205	let index_fsync = self.config.partition.enforce_fsync;	857✔
206		857✔
207	let server_confirmation = self.config.segment.server_confirmation;	857✔
208	let max_file_operation_retries = self.config.state.max_file_operation_retries;	857✔
209	let retry_delay = self.config.state.retry_delay;	857✔
210
211	let log_writer = SegmentLogWriter::new(	857✔
212	&self.log_path,	857✔
213	self.log_size_bytes.clone(),	857✔
214	log_fsync,	857✔
215	server_confirmation,	857✔
216	max_file_operation_retries,	857✔
217	retry_delay,	857✔
218	)	857✔
219	.await?;	857✔
220
221	let index_writer =	857✔
222	SegmentIndexWriter::new(&self.index_path, self.index_size_bytes.clone(), index_fsync)	857✔
223	.await?;	857✔
224
225	self.log_writer = Some(log_writer);	857✔
226	self.index_writer = Some(index_writer);	857✔
227	Ok(())	857✔
228	}	857✔
229
230	pub async fn initialize_reading(&mut self) -> Result<(), IggyError> {	857✔
231	let log_reader = SegmentLogReader::new(&self.log_path, self.log_size_bytes.clone()).await?;	857✔
232	// TODO(hubcio): there is no need to store open fd for reader if we have index cache enabled
233	let index_reader =	857✔
234	SegmentIndexReader::new(&self.index_path, self.index_size_bytes.clone()).await?;	857✔
235
236	self.log_reader = Some(log_reader);	857✔
237	self.index_reader = Some(index_reader);	857✔
238	Ok(())	857✔
239	}	857✔
240
241	pub async fn is_full(&self) -> bool {	23,895✔
242	if self.size_bytes >= self.max_size_bytes {	23,895✔
243	return true;	×
244	}	23,895✔
245		23,895✔
246	self.is_expired(IggyTimestamp::now()).await	23,895✔
247	}	23,895✔
248
249	pub async fn is_expired(&self, now: IggyTimestamp) -> bool {	23,897✔
250	if !self.is_closed {	23,897✔
251	return false;	23,896✔
252	}	1✔
253		1✔
254	match self.message_expiry {	1✔
255	IggyExpiry::NeverExpire => false,	×
256	IggyExpiry::ServerDefault => false,	×
257	IggyExpiry::ExpireDuration(expiry) => {	1✔
258	let last_messages = self.get_messages(self.current_offset, 1).await;	1✔
259	if last_messages.is_err() {	1✔
260	return false;	×
261	}	1✔
262		1✔
263	let last_messages = last_messages.unwrap();	1✔
264	if last_messages.is_empty() {	1✔
265	return false;	×
266	}	1✔
267		1✔
268	let last_message = &last_messages[0];	1✔
269	let last_message_timestamp = last_message.timestamp;	1✔
270	last_message_timestamp + expiry.as_micros() <= now.as_micros()	1✔
271	}
272	}
273	}	23,897✔
274
275	pub async fn shutdown_reading(&mut self) {	561✔
276	if let Some(log_reader) = self.log_reader.take() {	561✔
277	drop(log_reader);	561✔
278	}	561✔
279	if let Some(index_reader) = self.index_reader.take() {	561✔
280	drop(index_reader);	561✔
281	}	561✔
282	}	561✔
283
284	pub async fn shutdown_writing(&mut self) {	561✔
285	if let Some(log_writer) = self.log_writer.take() {	561✔
286	tokio::spawn(async move {	561✔
287	let _ = log_writer.fsync().await;	561✔
288	log_writer.shutdown_persister_task().await;	560✔
289	});	561✔
290	} else {	561✔
NEW 291	warn!(	×
NEW 292	"Log writer already closed when calling close() for {}",	×
293	self
294	);
295	}
296
297	if let Some(index_writer) = self.index_writer.take() {	561✔
298	tokio::spawn(async move {	561✔
299	let _ = index_writer.fsync().await;	561✔
300	drop(index_writer)	561✔
301	});	561✔
302	} else {	561✔
NEW 303	warn!("Index writer already closed when calling close()");	×
304	}
305	}	561✔
306
307	pub async fn delete(&mut self) -> Result<(), IggyError> {	561✔
308	let segment_size = self.size_bytes;	561✔
309	let segment_count_of_messages = self.get_messages_count();	561✔
310	info!(	561✔
NEW 311	"Deleting segment of size {segment_size} with start offset: {} for partition with ID: {} for stream with ID: {} and topic with ID: {}...",	×
312	self.start_offset, self.partition_id, self.stream_id, self.topic_id,
313	);
314
315	self.shutdown_reading().await;	561✔
316
317	if !self.is_closed {	561✔
318	self.shutdown_writing().await;	561✔
NEW 319	}	×
320
321	let _ = remove_file(&self.log_path).await.with_error_context(\|e\| {	561✔
NEW 322	format!("Failed to delete log file: {}, error: {e}", self.log_path)	×
323	});	561✔
324	let _ = remove_file(&self.index_path).await.with_error_context(\|e\| {	561✔
NEW 325	format!(	×
NEW 326	"Failed to delete index file: {}, error: {e}",	×
NEW 327	self.index_path	×
NEW 328	)	×
329	});	561✔
330		561✔
331	let segment_size_bytes = self.size_bytes.as_bytes_u64();	561✔
332	self.size_of_parent_stream	561✔
333	.fetch_sub(segment_size_bytes, Ordering::SeqCst);	561✔
334	self.size_of_parent_topic	561✔
335	.fetch_sub(segment_size_bytes, Ordering::SeqCst);	561✔
336	self.size_of_parent_partition	561✔
337	.fetch_sub(segment_size_bytes, Ordering::SeqCst);	561✔
338	self.messages_count_of_parent_stream	561✔
339	.fetch_sub(segment_count_of_messages, Ordering::SeqCst);	561✔
340	self.messages_count_of_parent_topic	561✔
341	.fetch_sub(segment_count_of_messages, Ordering::SeqCst);	561✔
342	self.messages_count_of_parent_partition	561✔
343	.fetch_sub(segment_count_of_messages, Ordering::SeqCst);	561✔
344		561✔
345	info!(	561✔
NEW 346	"Deleted segment of size {segment_size} with start offset: {} for partition with ID: {} for stream with ID: {} and topic with ID: {}.",	×
347	self.start_offset, self.partition_id, self.stream_id, self.topic_id,
348	);
349
350	Ok(())	561✔
351	}	561✔
352
353	fn get_log_path(path: &str) -> String {	877✔
354	format!("{}.{}", path, LOG_EXTENSION)	877✔
355	}	877✔
356
357	fn get_index_path(path: &str) -> String {	877✔
358	format!("{}.{}", path, INDEX_EXTENSION)	877✔
359	}	877✔
360	}
361
362	impl std::fmt::Display for Segment {
363	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {	×
364	write!(	×
365	f,	×
366	"Segment {{ stream_id: {}, topic_id: {}, partition_id: {}, start_offset: {}, end_offset: {}, current_offset: {}, size_bytes: {}, last_index_position: {}, max_size_bytes: {}, closed: {} }}",	×
367	self.stream_id,	×
368	self.topic_id,	×
369	self.partition_id,	×
370	self.start_offset,	×
371	self.end_offset,	×
372	self.current_offset,	×
373	self.size_bytes,	×
374	self.last_index_position,	×
375	self.max_size_bytes,	×
376	self.is_closed	×
377	)	×
378	}	×
379	}
380
381	#[cfg(test)]
382	mod tests {
383	use super::*;
384	use crate::configs::system::SegmentConfig;
385	use iggy::utils::duration::IggyDuration;
386
387	#[tokio::test]
388	async fn should_be_created_given_valid_parameters() {	1✔
389	let stream_id = 1;	1✔
390	let topic_id = 2;	1✔
391	let partition_id = 3;	1✔
392	let start_offset = 0;	1✔
393	let config = Arc::new(SystemConfig::default());	1✔
394	let path = config.get_segment_path(stream_id, topic_id, partition_id, start_offset);	1✔
395	let log_path = Segment::get_log_path(&path);	1✔
396	let index_path = Segment::get_index_path(&path);	1✔
397	let message_expiry = IggyExpiry::ExpireDuration(IggyDuration::from(10));	1✔
398	let size_of_parent_stream = Arc::new(AtomicU64::new(0));	1✔
399	let size_of_parent_topic = Arc::new(AtomicU64::new(0));	1✔
400	let size_of_parent_partition = Arc::new(AtomicU64::new(0));	1✔
401	let messages_count_of_parent_stream = Arc::new(AtomicU64::new(0));	1✔
402	let messages_count_of_parent_topic = Arc::new(AtomicU64::new(0));	1✔
403	let messages_count_of_parent_partition = Arc::new(AtomicU64::new(0));	1✔
404		1✔
405	let segment = Segment::create(	1✔
406	stream_id,	1✔
407	topic_id,	1✔
408	partition_id,	1✔
409	start_offset,	1✔
410	config,	1✔
411	message_expiry,	1✔
412	size_of_parent_stream,	1✔
413	size_of_parent_topic,	1✔
414	size_of_parent_partition,	1✔
415	messages_count_of_parent_stream,	1✔
416	messages_count_of_parent_topic,	1✔
417	messages_count_of_parent_partition,	1✔
418	);	1✔
419		1✔
420	assert_eq!(segment.stream_id, stream_id);	1✔
421	assert_eq!(segment.topic_id, topic_id);	1✔
422	assert_eq!(segment.partition_id, partition_id);	1✔
423	assert_eq!(segment.start_offset, start_offset);	1✔
424	assert_eq!(segment.current_offset, 0);	1✔
425	assert_eq!(segment.end_offset, 0);	1✔
426	assert_eq!(segment.size_bytes, 0);	1✔
427	assert_eq!(segment.log_path, log_path);	1✔
428	assert_eq!(segment.index_path, index_path);	1✔
429	assert_eq!(segment.message_expiry, message_expiry);	1✔
430	assert!(segment.unsaved_messages.is_none());	1✔
431	assert!(segment.indexes.is_some());	1✔
432	assert!(!segment.is_closed);	1✔
433	assert!(!segment.is_full().await);	1✔
434	}	1✔
435
436	#[tokio::test]
437	async fn should_not_initialize_indexes_cache_when_disabled() {	1✔
438	let stream_id = 1;	1✔
439	let topic_id = 2;	1✔
440	let partition_id = 3;	1✔
441	let start_offset = 0;	1✔
442	let config = Arc::new(SystemConfig {	1✔
443	segment: SegmentConfig {	1✔
444	cache_indexes: false,	1✔
445	..Default::default()	1✔
446	},	1✔
447	..Default::default()	1✔
448	});	1✔
449	let message_expiry = IggyExpiry::NeverExpire;	1✔
450	let size_of_parent_stream = Arc::new(AtomicU64::new(0));	1✔
451	let size_of_parent_topic = Arc::new(AtomicU64::new(0));	1✔
452	let size_of_parent_partition = Arc::new(AtomicU64::new(0));	1✔
453	let messages_count_of_parent_stream = Arc::new(AtomicU64::new(0));	1✔
454	let messages_count_of_parent_topic = Arc::new(AtomicU64::new(0));	1✔
455	let messages_count_of_parent_partition = Arc::new(AtomicU64::new(0));	1✔
456		1✔
457	let segment = Segment::create(	1✔
458	stream_id,	1✔
459	topic_id,	1✔
460	partition_id,	1✔
461	start_offset,	1✔
462	config,	1✔
463	message_expiry,	1✔
464	size_of_parent_stream,	1✔
465	size_of_parent_topic,	1✔
466	size_of_parent_partition,	1✔
467	messages_count_of_parent_stream,	1✔
468	messages_count_of_parent_topic,	1✔
469	messages_count_of_parent_partition,	1✔
470	);	1✔
471		1✔
472	assert!(segment.indexes.is_none());	1✔
473	}	1✔
474	}

iggy-rs / iggy / 13335910144

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous