nautilus_persistence/backend/
feather.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::{
17    any::Any,
18    cell::RefCell,
19    collections::{HashMap, HashSet},
20    rc::Rc,
21    sync::Arc,
22};
23
24use ahash::AHashMap;
25use chrono_tz::Tz;
26use datafusion::arrow::{
27    datatypes::Schema, error::ArrowError, ipc::writer::StreamWriter, record_batch::RecordBatch,
28};
29use nautilus_common::{
30    cache::fifo::FifoCache,
31    clock::Clock,
32    msgbus::{mstr::MStr, subscribe_any, typed_handler::ShareableMessageHandler, unsubscribe_any},
33};
34use nautilus_core::{UUID4, UnixNanos, datetime::NANOSECONDS_IN_SECOND};
35use nautilus_model::{
36    data::{
37        Bar, CatalogPathPrefix, CustomData, CustomDataTrait, Data, FundingRateUpdate,
38        IndexPriceUpdate, InstrumentStatus, MarkPriceUpdate, OptionGreeks, OrderBookDelta,
39        OrderBookDeltas, OrderBookDepth10, QuoteTick, TradeTick, close::InstrumentClose,
40        encode_custom_to_arrow, get_arrow_schema,
41    },
42    events::{
43        AccountState, OrderAccepted, OrderCancelRejected, OrderCanceled, OrderDenied,
44        OrderEmulated, OrderExpired, OrderFilled, OrderInitialized, OrderModifyRejected,
45        OrderPendingCancel, OrderPendingUpdate, OrderRejected, OrderReleased, OrderSnapshot,
46        OrderSubmitted, OrderTriggered, OrderUpdated, PositionAdjusted, PositionChanged,
47        PositionClosed, PositionOpened, PositionSnapshot,
48    },
49    instruments::InstrumentAny,
50    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
51};
52use nautilus_serialization::arrow::{EncodeToRecordBatch, KEY_INSTRUMENT_ID};
53use object_store::{ObjectStore, ObjectStoreExt, path::Path};
54
55use super::catalog::urisafe_instrument_id;
56use crate::backend::{
57    catalog::safe_directory_identifier,
58    custom::{augment_batch_with_data_type_column, schema_with_data_type_column},
59};
60
61#[derive(Debug, Default, PartialEq, PartialOrd, Hash, Eq, Clone)]
62pub struct FileWriterPath {
63    path: Path,
64    type_str: String,
65    instrument_id: Option<String>,
66}
67
68/// A `FeatherBuffer` encodes data via an Arrow `StreamWriter`.
69///
70/// It flushes the internal byte buffer according to rotation policy.
71pub struct FeatherBuffer {
72    /// Arrow `StreamWriter` that writes to an in-memory `Vec<u8>`.
73    writer: StreamWriter<Vec<u8>>,
74    /// Current size in bytes.
75    size: u64,
76    /// TODO: Optional next rotation timestamp.
77    // next_rotation: Option<UnixNanos>,
78    /// Schema of the data being written.
79    schema: Schema,
80    /// Maximum buffer size in bytes.
81    max_buffer_size: u64,
82    /// Rotation config
83    rotation_config: RotationConfig,
84}
85
86impl FeatherBuffer {
87    /// Creates a new [`FeatherBuffer`] using the given path, schema and maximum buffer size.
88    ///
89    /// # Errors
90    ///
91    /// Returns an error if the Arrow stream writer cannot be created.
92    pub fn new(schema: &Schema, rotation_config: RotationConfig) -> Result<Self, ArrowError> {
93        let writer = StreamWriter::try_new(Vec::new(), schema)?;
94        let mut max_buffer_size = 1_000_000_000_000; // 1 GB
95
96        if let RotationConfig::Size { max_size } = &rotation_config {
97            max_buffer_size = *max_size;
98        }
99
100        Ok(Self {
101            writer,
102            size: 0,
103            // next_rotation: None,
104            max_buffer_size,
105            schema: schema.clone(),
106            rotation_config,
107        })
108    }
109
110    /// Writes the given `RecordBatch` to the internal buffer.
111    ///
112    /// Returns true if it should be rotated according rotation policy
113    ///
114    /// # Errors
115    ///
116    /// Returns an error if Arrow IPC writing fails.
117    pub fn write_record_batch(&mut self, batch: &RecordBatch) -> Result<bool, ArrowError> {
118        self.writer.write(batch)?;
119        self.size += batch.get_array_memory_size() as u64;
120        Ok(self.size >= self.max_buffer_size)
121    }
122
123    /// Consumes the writer and returns the buffer of bytes from the `StreamWriter`
124    ///
125    /// # Errors
126    ///
127    /// Returns an error if the replacement writer cannot be created or the previous
128    /// writer cannot be finalized.
129    pub fn take_buffer(&mut self) -> Result<Vec<u8>, ArrowError> {
130        let mut writer = StreamWriter::try_new(Vec::new(), &self.schema)?;
131        std::mem::swap(&mut self.writer, &mut writer);
132        let buffer = writer.into_inner()?;
133        // TODO: Handle rotation config here
134        self.size = 0;
135        Ok(buffer)
136    }
137
138    /// Should rotate
139    #[must_use]
140    pub const fn should_rotate(&self) -> bool {
141        match &self.rotation_config {
142            RotationConfig::Size { max_size } => self.size >= *max_size,
143            _ => false,
144        }
145    }
146}
147
148/// Configuration for file rotation.
149#[derive(Debug, Clone)]
150pub enum RotationConfig {
151    /// Rotate based on file size.
152    Size {
153        /// Maximum buffer size in bytes before rotation.
154        max_size: u64,
155    },
156    /// Rotate based on a time interval.
157    Interval {
158        /// Interval in nanoseconds.
159        interval_ns: u64,
160    },
161    /// Rotate based on scheduled dates.
162    ScheduledDates {
163        /// Interval in nanoseconds.
164        interval_ns: u64,
165        /// Time of day for rotation (nanoseconds since midnight).
166        rotation_time: UnixNanos,
167        /// Timezone for rotation calculations.
168        rotation_timezone: Tz,
169    },
170    /// No automatic rotation.
171    NoRotation,
172}
173
174/// Manages multiple `FeatherBuffers` and handles encoding, rotation, and flushing to the object store.
175///
176/// The `write()` method is the single entry point for clients: they supply a data value (of generic type T)
177/// and the manager encodes it (using T's metadata via `EncodeToRecordBatch`), routes it by `CatalogPathPrefix`,
178/// and writes it to the appropriate `FileWriter`. When a writer's buffer is full or rotation criteria are met,
179/// its contents are flushed to the object store and it is replaced.
180pub struct FeatherWriter {
181    /// Base directory for writing files.
182    base_path: String,
183    /// Object store for persistence.
184    store: Arc<dyn ObjectStore>,
185    /// Clock for timestamps and rotation.
186    clock: Rc<RefCell<dyn Clock>>,
187    /// Rotation configuration.
188    rotation_config: RotationConfig,
189    /// Optional set of type names to include.
190    included_types: Option<HashSet<String>>,
191    /// Set of types that should be split by instrument.
192    per_instrument_types: HashSet<String>,
193    /// Map of active `FeatherBuffers` keyed by their path.
194    writers: HashMap<FileWriterPath, FeatherBuffer>,
195    /// Map of next rotation times keyed by their path.
196    next_rotation_times: HashMap<FileWriterPath, UnixNanos>,
197    /// Runtime handle for async operations.
198    runtime: tokio::runtime::Handle,
199    /// Flush interval in milliseconds (0 = no automatic flushing).
200    flush_interval_ms: u64,
201    /// Last flush timestamp in nanoseconds.
202    last_flush_ns: UnixNanos,
203    /// Bounded cache of recently seen event IDs for deduplication.
204    seen_event_ids: Box<FifoCache<UUID4, 10_000>>,
205}
206
207impl FeatherWriter {
208    /// Creates a new [`FeatherWriter`] instance.
209    pub fn new(
210        base_path: String,
211        store: Arc<dyn ObjectStore>,
212        clock: Rc<RefCell<dyn Clock>>,
213        rotation_config: RotationConfig,
214        included_types: Option<HashSet<String>>,
215        per_instrument_types: Option<HashSet<String>>,
216        flush_interval_ms: Option<u64>,
217    ) -> Self {
218        // Get the runtime handle for async operations
219        let runtime = nautilus_common::live::get_runtime().handle().clone();
220        let flush_interval_ms = flush_interval_ms.unwrap_or(1000); // Default 1 second
221        let last_flush_ns = clock.borrow().timestamp_ns();
222
223        Self {
224            base_path,
225            store,
226            clock,
227            rotation_config,
228            included_types,
229            per_instrument_types: per_instrument_types.unwrap_or_default(),
230            writers: HashMap::new(),
231            next_rotation_times: HashMap::new(),
232            runtime,
233            flush_interval_ms,
234            last_flush_ns,
235            seen_event_ids: Box::new(FifoCache::new()),
236        }
237    }
238
239    /// Writes a single data value.
240    /// This is the user entry point. The data is encoded into a `RecordBatch` and written to the appropriate `FileWriter`.
241    /// If the writer's buffer reaches capacity or meets rotation criteria (based on the rotation configuration),
242    /// the `FileWriter` is flushed to the object store and replaced.
243    ///
244    /// # Errors
245    ///
246    /// Returns an error if path selection, Arrow encoding, writer creation, buffering,
247    /// rotation, or flushing fails.
248    pub async fn write<T>(&mut self, data: T) -> Result<(), Box<dyn std::error::Error>>
249    where
250        T: EncodeToRecordBatch + CatalogPathPrefix + 'static,
251    {
252        if !self.should_write::<T>() {
253            return Ok(());
254        }
255
256        let path = self.get_writer_path(&data)?;
257
258        // Create a new FileWriter if one does not exist.
259        if !self.writers.contains_key(&path) {
260            self.create_writer::<T>(path.clone(), &data)?;
261        }
262
263        // Encode the data into a RecordBatch using T's encoding logic.
264        let batch = T::encode_batch(&T::metadata(&data), &[data])?;
265
266        // Write the RecordBatch to the appropriate FileWriter.
267        if let Some(writer) = self.writers.get_mut(&path) {
268            let should_rotate = writer.write_record_batch(&batch)?;
269            if should_rotate || self.check_scheduled_rotation(&path) {
270                self.rotate_writer(&path).await?;
271            }
272        }
273
274        // Check if we should auto-flush based on time interval
275        self.check_flush().await?;
276
277        Ok(())
278    }
279
280    /// Writes a batch of data values as one or more `RecordBatch`es.
281    ///
282    /// Uses `T::chunk_metadata` to derive the file schema metadata. This protects
283    /// types like `OrderBookDelta` from having their file metadata poisoned by a
284    /// leading sentinel row (e.g. `BookAction::Clear`, which carries
285    /// `price_precision=0, size_precision=0`).
286    ///
287    /// Per-instrument types are partitioned by instrument so a mixed-instrument
288    /// batch lands in the correct file for each instrument.
289    ///
290    /// # Errors
291    ///
292    /// Returns an error if path selection, Arrow encoding, writer creation, buffering,
293    /// rotation, or flushing fails.
294    pub async fn write_batch<T>(&mut self, data: Vec<T>) -> Result<(), Box<dyn std::error::Error>>
295    where
296        T: EncodeToRecordBatch + CatalogPathPrefix + 'static,
297    {
298        if data.is_empty() || !self.should_write::<T>() {
299            return Ok(());
300        }
301
302        // Group by logical writer identity (instrument_id for per-instrument types).
303        // Grouping on FileWriterPath would split same-instrument rows across distinct
304        // timestamped paths when the writer does not yet exist under a LiveClock.
305        let type_str = T::path_prefix();
306        let needs_instrument =
307            self.per_instrument_types.contains(type_str) || type_str.starts_with("custom_");
308
309        let mut groups: AHashMap<Option<String>, Vec<T>> = AHashMap::new();
310
311        for item in data {
312            let instrument_id = if needs_instrument {
313                T::metadata(&item).get(KEY_INSTRUMENT_ID).cloned()
314            } else {
315                None
316            };
317            groups.entry(instrument_id).or_default().push(item);
318        }
319
320        for group in groups.into_values() {
321            let path = self.get_writer_path(&group[0])?;
322            let metadata = T::chunk_metadata(&group);
323
324            if !self.writers.contains_key(&path) {
325                self.create_writer_with_metadata::<T>(path.clone(), metadata.clone())?;
326            }
327
328            let batch = T::encode_batch(&metadata, &group)?;
329
330            if let Some(writer) = self.writers.get_mut(&path) {
331                let should_rotate = writer.write_record_batch(&batch)?;
332                if should_rotate || self.check_scheduled_rotation(&path) {
333                    self.rotate_writer(&path).await?;
334                }
335            }
336        }
337
338        self.check_flush().await?;
339
340        Ok(())
341    }
342
343    /// Checks if enough time has passed since last flush and flushes if needed.
344    async fn check_flush(&mut self) -> Result<(), Box<dyn std::error::Error>> {
345        if self.flush_interval_ms == 0 {
346            return Ok(()); // Auto-flush disabled
347        }
348
349        let now_ns = self.clock.borrow().timestamp_ns();
350        let elapsed_ms = (now_ns.as_u64() - self.last_flush_ns.as_u64()) / 1_000_000;
351
352        if elapsed_ms >= self.flush_interval_ms {
353            self.flush().await?;
354            self.last_flush_ns = now_ns;
355        }
356
357        Ok(())
358    }
359
360    fn check_scheduled_rotation(&mut self, path: &FileWriterPath) -> bool {
361        match self.rotation_config {
362            RotationConfig::Interval { interval_ns } => {
363                let now = self.clock.borrow().timestamp_ns();
364                let next_rotation = self.next_rotation_times.get(path).copied();
365
366                match next_rotation {
367                    None => {
368                        self.next_rotation_times
369                            .insert(path.clone(), now + interval_ns);
370                        false
371                    }
372                    Some(next) if now >= next => {
373                        self.next_rotation_times
374                            .insert(path.clone(), now + interval_ns);
375                        true
376                    }
377                    _ => false,
378                }
379            }
380            RotationConfig::ScheduledDates {
381                interval_ns,
382                rotation_time,
383                rotation_timezone,
384            } => {
385                let now = self.clock.borrow().timestamp_ns();
386                let next_rotation = self.next_rotation_times.get(path).copied();
387
388                match next_rotation {
389                    None => {
390                        let next = self.calculate_next_scheduled_rotation(
391                            rotation_time,
392                            rotation_timezone,
393                            interval_ns,
394                        );
395                        self.next_rotation_times.insert(path.clone(), next);
396                        false
397                    }
398                    Some(next) if now >= next => {
399                        self.next_rotation_times
400                            .insert(path.clone(), now + interval_ns);
401                        true
402                    }
403                    _ => false,
404                }
405            }
406            _ => false,
407        }
408    }
409
410    fn calculate_next_scheduled_rotation(
411        &self,
412        rotation_time: UnixNanos,
413        rotation_timezone: Tz,
414        interval_ns: u64,
415    ) -> UnixNanos {
416        use chrono::TimeZone;
417        let now_utc = self.clock.borrow().utc_now();
418        let now_tz = now_utc.with_timezone(&rotation_timezone);
419
420        let rotation_time_secs = u32::try_from(*rotation_time / NANOSECONDS_IN_SECOND).unwrap_or(0);
421        let rotation_time_nanos =
422            u32::try_from(*rotation_time % NANOSECONDS_IN_SECOND).unwrap_or(0);
423        let rotation_time_naive = chrono::NaiveTime::from_num_seconds_from_midnight_opt(
424            rotation_time_secs,
425            rotation_time_nanos,
426        )
427        .unwrap_or_else(|| chrono::NaiveTime::from_hms_opt(0, 0, 0).unwrap());
428
429        let mut next_rotation_tz = rotation_timezone
430            .from_local_datetime(&now_tz.date_naive().and_time(rotation_time_naive))
431            .earliest()
432            .unwrap_or(now_tz);
433
434        if next_rotation_tz <= now_tz {
435            // If the time has already passed today, we would usually add the interval
436            // But let's align exactly with how Python does it:
437            while next_rotation_tz <= now_tz {
438                // Add interval_ns to next_rotation_tz
439                // Since chrono::Duration doesn't take u64 nanos directly comfortably for large values,
440                // we'll convert to seconds and nanos.
441                let secs = i64::try_from(interval_ns / NANOSECONDS_IN_SECOND).unwrap_or(i64::MAX);
442                let nanos = u32::try_from(interval_ns % NANOSECONDS_IN_SECOND).unwrap_or(0);
443                next_rotation_tz = next_rotation_tz
444                    + chrono::Duration::seconds(secs)
445                    + chrono::Duration::nanoseconds(i64::from(nanos));
446            }
447        }
448
449        let timestamp_ns = next_rotation_tz
450            .with_timezone(&chrono::Utc)
451            .timestamp_nanos_opt()
452            .unwrap_or(0);
453        UnixNanos::from(u64::try_from(timestamp_ns.max(0)).unwrap_or(0))
454    }
455
456    /// Flushes and rotates `FileWriter` associated with `key`.
457    /// TODO: Fix error type to handle arrow error and object store error
458    async fn rotate_writer(
459        &mut self,
460        path: &FileWriterPath,
461    ) -> Result<(), Box<dyn std::error::Error>> {
462        let mut writer = self.writers.remove(path).unwrap();
463        let bytes = writer.take_buffer()?;
464        self.store.put(&path.path, bytes.into()).await?;
465        let new_path = self.regen_writer_path(path);
466        self.writers.insert(new_path, writer);
467        Ok(())
468    }
469
470    /// Creates (and inserts) a new `FileWriter` for type T.
471    fn create_writer<T>(&mut self, path: FileWriterPath, data: &T) -> Result<(), ArrowError>
472    where
473        T: EncodeToRecordBatch + CatalogPathPrefix + 'static,
474    {
475        self.create_writer_with_metadata::<T>(path, T::metadata(data))
476    }
477
478    /// Creates (and inserts) a new `FileWriter` for type T with pre-computed metadata.
479    ///
480    /// Use this variant when the caller has selected metadata from a chunk
481    /// (e.g. via `T::chunk_metadata`) to avoid schema poisoning by sentinel rows.
482    fn create_writer_with_metadata<T>(
483        &mut self,
484        path: FileWriterPath,
485        metadata: HashMap<String, String>,
486    ) -> Result<(), ArrowError>
487    where
488        T: EncodeToRecordBatch + CatalogPathPrefix + 'static,
489    {
490        let schema = if self.per_instrument_types.contains(T::path_prefix()) {
491            T::get_schema(Some(metadata))
492        } else {
493            T::get_schema(None)
494        };
495
496        let writer = FeatherBuffer::new(&schema, self.rotation_config.clone())?;
497        self.writers.insert(path, writer);
498        Ok(())
499    }
500
501    /// Creates (and inserts) a new `FeatherBuffer` for custom data at the given path.
502    fn create_custom_writer(
503        &mut self,
504        path: FileWriterPath,
505        type_name: &str,
506    ) -> Result<(), Box<dyn std::error::Error>> {
507        if self.writers.contains_key(&path) {
508            return Ok(());
509        }
510        let base_schema = get_arrow_schema(type_name).ok_or_else(|| {
511            format!("Custom data type \"{type_name}\" is not registered for Arrow encoding")
512        })?;
513        let schema = schema_with_data_type_column(base_schema.as_ref(), type_name);
514        let writer = FeatherBuffer::new(&schema, self.rotation_config.clone())
515            .map_err(|e| format!("Failed to create feather buffer for custom {type_name}: {e}"))?;
516        self.writers.insert(path, writer);
517        Ok(())
518    }
519
520    /// Encodes a single `CustomData` into a `RecordBatch` with `data_type` column (catalog-compatible).
521    fn encode_custom_to_batch(
522        custom: &CustomData,
523    ) -> Result<RecordBatch, Box<dyn std::error::Error>> {
524        let type_name = custom.data.type_name();
525        let data_type_json = custom
526            .data_type
527            .to_persistence_json()
528            .map_err(|e| format!("Failed to serialize data_type for persistence: {e}"))?;
529        let dt_meta = custom.data_type.metadata_string_map();
530        let items: [Arc<dyn CustomDataTrait>; 1] = [Arc::clone(&custom.data)];
531        let batch = encode_custom_to_arrow(type_name, &items)
532            .map_err(|e| format!("Failed to encode custom data: {e}"))?
533            .ok_or_else(|| {
534                format!("Custom data type \"{type_name}\" is not registered for Arrow")
535            })?;
536        let batch = augment_batch_with_data_type_column(
537            &batch,
538            &data_type_json,
539            type_name,
540            dt_meta.as_ref(),
541        )
542        .map_err(|e| e.to_string())?;
543        Ok(batch)
544    }
545
546    /// Flushes all active `FeatherBuffers` by writing any remaining buffered bytes to the object store.
547    ///
548    /// This is called automatically based on `flush_interval_ms` if configured, but can also
549    /// be called manually by the client.
550    ///
551    /// Note: In Rust, we use in-memory buffers. Flushing writes the current buffer to the
552    /// object store and creates a new buffer for continued writing. This is different from
553    /// Python which just flushes OS buffers.
554    ///
555    /// # Errors
556    ///
557    /// Returns an error if buffer finalization or object store writes fail.
558    pub async fn flush(&mut self) -> Result<(), Box<dyn std::error::Error>> {
559        // Collect paths and their current buffers before flushing
560        let paths_to_flush: Vec<FileWriterPath> = self.writers.keys().cloned().collect();
561
562        // Flush each writer and recreate it
563        for path in paths_to_flush {
564            if let Some(mut writer) = self.writers.remove(&path) {
565                let bytes = writer.take_buffer()?;
566                if !bytes.is_empty() {
567                    // Write to the object store
568                    self.store.put(&path.path, bytes.into()).await?;
569                }
570
571                // Recreate writer with same schema for continued writing
572                // We need the schema and type info - for now, we'll recreate on next write
573                // The writer will be recreated automatically when write() is called again
574            }
575        }
576
577        self.last_flush_ns = self.clock.borrow().timestamp_ns();
578        Ok(())
579    }
580
581    /// Closes all writers by flushing and removing them.
582    ///
583    /// After calling this, no further writes should be performed.
584    ///
585    /// # Errors
586    ///
587    /// Returns an error if flushing buffered data fails.
588    pub async fn close(&mut self) -> Result<(), Box<dyn std::error::Error>> {
589        self.flush().await?;
590        self.writers.clear();
591        Ok(())
592    }
593
594    /// Returns whether the writer has been closed (all writers cleared).
595    #[must_use]
596    pub fn is_closed(&self) -> bool {
597        self.writers.is_empty()
598    }
599
600    /// Returns information about the current files being written.
601    ///
602    /// Each entry maps a writer key (`type_str` and optional `instrument_id`) to
603    /// its current buffer size and file path.
604    #[must_use]
605    pub fn get_current_file_info(&self) -> HashMap<String, (u64, String)> {
606        let mut info = HashMap::new();
607
608        for (path, buffer) in &self.writers {
609            let key = match &path.instrument_id {
610                Some(id) => format!("{}:{}", path.type_str, id),
611                None => path.type_str.clone(),
612            };
613            info.insert(key, (buffer.size, path.path.to_string()));
614        }
615        info
616    }
617
618    /// Returns the next rotation time for a specific writer key, if set.
619    #[must_use]
620    pub fn get_next_rotation_time(
621        &self,
622        type_str: &str,
623        instrument_id: Option<&str>,
624    ) -> Option<UnixNanos> {
625        self.next_rotation_times
626            .iter()
627            .find(|(k, _)| k.type_str == type_str && k.instrument_id.as_deref() == instrument_id)
628            .map(|(_, &v)| v)
629    }
630
631    /// Determines whether type T should be written, based on the inclusion filter.
632    fn should_write<T: CatalogPathPrefix>(&self) -> bool {
633        self.included_types.as_ref().is_none_or(|included| {
634            let path = T::path_prefix();
635            included.contains(path)
636        })
637    }
638
639    /// Returns whether the given event ID has already been seen,
640    /// adding it to the cache if new.
641    pub fn is_duplicate_event_id(&mut self, event_id: &UUID4) -> bool {
642        if self.seen_event_ids.contains(event_id) {
643            return true;
644        }
645
646        self.seen_event_ids.add(*event_id);
647
648        false
649    }
650
651    fn regen_writer_path(&self, path: &FileWriterPath) -> FileWriterPath {
652        let type_str = path.type_str.clone();
653        let instrument_id = path.instrument_id.clone();
654        let timestamp = self.clock.borrow().timestamp_ns();
655        // Note: Path removes prefixing slashes
656        let mut path = Path::from(self.base_path.clone());
657
658        if type_str.starts_with("data/custom/") {
659            // Custom data: data/custom/{type_name}/[{identifier_segments}/]{file_stem}_{ts}.feather
660            let type_name = type_str.strip_prefix("data/custom/").unwrap_or(&type_str);
661            path = path.join("data").join("custom").join(type_name.to_string());
662
663            if let Some(ref id) = instrument_id {
664                let safe = safe_directory_identifier(id);
665                if !safe.is_empty() {
666                    for segment in safe.split('/') {
667                        path = path.join(segment.to_string());
668                    }
669                }
670            }
671            let file_stem = instrument_id.as_deref().unwrap_or(type_name);
672            path = path.join(format!("{file_stem}_{timestamp}.feather"));
673        } else if let Some(ref instrument_id) = instrument_id {
674            let safe_id = urisafe_instrument_id(instrument_id);
675            path = path.join(type_str.clone());
676            path = path.join(safe_id.clone());
677            path = path.join(format!("{safe_id}_{timestamp}.feather"));
678        } else {
679            path = path.join(format!("{type_str}_{timestamp}.feather"));
680        }
681
682        FileWriterPath {
683            path,
684            type_str,
685            instrument_id,
686        }
687    }
688
689    /// Builds `FileWriterPath` for custom data using `DataType` identifier as folder partition (catalog layout).
690    fn get_writer_path_custom(&self, type_name: &str, identifier: Option<&str>) -> FileWriterPath {
691        let timestamp = self.clock.borrow().timestamp_ns();
692        let type_str = format!("data/custom/{type_name}");
693        let instrument_id = identifier.map(String::from);
694
695        let mut path = Path::from(self.base_path.clone());
696        path = path.join("data").join("custom").join(type_name.to_string());
697
698        if let Some(id) = &identifier {
699            let safe = safe_directory_identifier(id);
700            if !safe.is_empty() {
701                for segment in safe.split('/') {
702                    path = path.join(segment.to_string());
703                }
704            }
705        }
706        let file_stem = identifier.unwrap_or(type_name);
707        path = path.join(format!("{file_stem}_{timestamp}.feather"));
708
709        FileWriterPath {
710            path,
711            type_str,
712            instrument_id,
713        }
714    }
715
716    /// Generates a key for a `FileWriter` based on type T and optional instrument ID.
717    /// Reuses an existing writer key (same `type_str` and `instrument_id`) if present, so we
718    /// buffer multiple items in the same file until rotation; otherwise creates a new path with current timestamp.
719    fn get_writer_path<T>(&self, data: &T) -> Result<FileWriterPath, Box<dyn std::error::Error>>
720    where
721        T: EncodeToRecordBatch + CatalogPathPrefix,
722    {
723        let type_str = T::path_prefix();
724        let metadata = T::metadata(data);
725
726        let instrument_id = if self.per_instrument_types.contains(type_str)
727            || (type_str.starts_with("custom_") && metadata.contains_key(KEY_INSTRUMENT_ID))
728        {
729            Some(metadata.get(KEY_INSTRUMENT_ID).cloned().ok_or_else(|| {
730                format!("Data {type_str} expected instrument_id metadata for per instrument writer")
731            })?)
732        } else {
733            None
734        };
735
736        // Reuse existing writer for same (type_str, instrument_id) so we buffer in one file until rotation
737        if let Some(existing) = self
738            .writers
739            .keys()
740            .find(|k| k.type_str == type_str && k.instrument_id == instrument_id)
741        {
742            return Ok(existing.clone());
743        }
744
745        let timestamp = self.clock.borrow().timestamp_ns();
746        let mut path = Path::from(self.base_path.clone());
747
748        if let Some(ref instrument_id) = instrument_id {
749            let safe_id = urisafe_instrument_id(instrument_id);
750            path = path.join(type_str);
751            path = path.join(safe_id.clone());
752            path = path.join(format!("{safe_id}_{timestamp}.feather"));
753        } else {
754            path = path.join(format!("{type_str}_{timestamp}.feather"));
755        }
756
757        Ok(FileWriterPath {
758            path,
759            type_str: type_str.to_string(),
760            instrument_id,
761        })
762    }
763
764    /// Writes a Data enum value to the appropriate writer.
765    ///
766    /// This is a convenience method that routes the Data enum to the appropriate
767    /// typed write method.
768    ///
769    /// # Errors
770    ///
771    /// Returns an error if the routed typed or custom data write fails.
772    #[allow(
773        clippy::match_wildcard_for_single_variants,
774        reason = "Data::Defi appears through nautilus-model feature unification"
775    )]
776    pub async fn write_data(&mut self, data: Data) -> Result<(), Box<dyn std::error::Error>> {
777        match data {
778            Data::Quote(quote) => self.write(quote).await,
779            Data::Trade(trade) => self.write(trade).await,
780            Data::Bar(bar) => self.write(bar).await,
781            Data::Delta(delta) => self.write(delta).await,
782            Data::Depth10(depth) => self.write(*depth).await,
783            Data::IndexPriceUpdate(price) => self.write(price).await,
784            Data::MarkPriceUpdate(price) => self.write(price).await,
785            Data::FundingRateUpdate(funding) => self.write(funding).await,
786            Data::OptionGreeks(greeks) => self.write(greeks).await,
787            Data::InstrumentStatus(status) => self.write(status).await,
788            Data::InstrumentClose(close) => self.write(close).await,
789            Data::Custom(custom) => self.write_custom_data(&custom).await,
790            Data::Deltas(deltas_api) => {
791                // Batch write so chunk_metadata can skip a leading BookAction::Clear sentinel
792                self.write_batch(deltas_api.deltas.clone()).await
793            }
794            #[cfg(feature = "defi")]
795            Data::Defi(_) => Err("Unsupported Data::Defi variant for feather writes".into()),
796            #[allow(unreachable_patterns)]
797            _ => Err("Unsupported Data variant for feather writes".into()),
798        }
799    }
800
801    /// Writes a single custom data value (catalog layout: `data/custom/{type_name}/[{identifier}/]`).
802    async fn write_custom_data(
803        &mut self,
804        custom: &CustomData,
805    ) -> Result<(), Box<dyn std::error::Error>> {
806        let type_name = custom.data.type_name();
807        let identifier = custom.data_type.identifier().map(String::from);
808
809        if !self.should_write_custom(type_name) {
810            return Ok(());
811        }
812
813        let path = self.get_writer_path_custom(type_name, identifier.as_deref());
814        if !self.writers.contains_key(&path) {
815            self.create_custom_writer(path.clone(), type_name)?;
816        }
817
818        let batch = Self::encode_custom_to_batch(custom)?;
819
820        if let Some(writer) = self.writers.get_mut(&path) {
821            let should_rotate = writer.write_record_batch(&batch)?;
822            if should_rotate || self.check_scheduled_rotation(&path) {
823                self.rotate_writer(&path).await?;
824            }
825        }
826
827        self.check_flush().await?;
828        Ok(())
829    }
830
831    fn should_write_custom(&self, type_name: &str) -> bool {
832        self.included_types.as_ref().is_none_or(|included| {
833            included.contains(type_name)
834                || included.contains("custom")
835                || included.contains(&format!("custom/{type_name}"))
836        })
837    }
838
839    /// Writes an instrument to the appropriate writer.
840    ///
841    /// Instruments are written to feather files and organized by instrument ID.
842    /// This method supports writing instruments that implement `EncodeToRecordBatch` and `CatalogPathPrefix`.
843    ///
844    /// # Errors
845    ///
846    /// Returns an error if the instrument write fails.
847    pub async fn write_instrument(
848        &mut self,
849        instrument: InstrumentAny,
850    ) -> Result<(), Box<dyn std::error::Error>> {
851        self.write(instrument).await
852    }
853
854    /// Subscribes to all messages on the message bus (pattern "*").
855    ///
856    /// This will automatically write all supported data types that are published
857    /// on the message bus to the feather files.
858    ///
859    /// The writer must be wrapped in `Rc<RefCell<>>` to be shareable with the message bus handler.
860    ///
861    /// Note: The handler spawns async tasks to write data, so writes happen asynchronously
862    /// and won't block the message bus.
863    ///
864    /// # Errors
865    ///
866    /// Returns an error if subscription setup fails.
867    pub fn subscribe_to_message_bus(
868        writer: Rc<RefCell<Self>>,
869    ) -> Result<ShareableMessageHandler, Box<dyn std::error::Error>> {
870        let runtime = writer.borrow().runtime.clone();
871
872        // Create handler that downcasts messages and writes them
873        // Note: We use Handle::enter() to allow blocking in the handler context
874        // This works when the handler is called from outside an async runtime
875        let handler = ShareableMessageHandler::from_any(move |message: &dyn Any| {
876            // Enter the runtime context to allow blocking
877            let _guard = runtime.enter();
878
879            // Try to downcast to various data types and write them
880            macro_rules! try_write {
881                ($message:expr, $type:ty, $name:literal) => {
882                    if let Some(value) = $message.downcast_ref::<$type>() {
883                        let mut writer = writer.borrow_mut();
884                        if let Err(e) = runtime.block_on(writer.write(value.clone())) {
885                            log::warn!("Failed to write {}: {e}", $name);
886                        }
887                        return;
888                    }
889                };
890            }
891
892            try_write!(message, QuoteTick, "QuoteTick");
893            try_write!(message, TradeTick, "TradeTick");
894            try_write!(message, Bar, "Bar");
895            try_write!(message, OrderBookDelta, "OrderBookDelta");
896            try_write!(message, OrderBookDepth10, "OrderBookDepth10");
897            try_write!(message, IndexPriceUpdate, "IndexPriceUpdate");
898            try_write!(message, MarkPriceUpdate, "MarkPriceUpdate");
899            try_write!(message, FundingRateUpdate, "FundingRateUpdate");
900            try_write!(message, OptionGreeks, "OptionGreeks");
901            try_write!(message, InstrumentStatus, "InstrumentStatus");
902            try_write!(message, InstrumentClose, "InstrumentClose");
903            try_write!(message, AccountState, "AccountState");
904            try_write!(message, OrderInitialized, "OrderInitialized");
905            try_write!(message, OrderDenied, "OrderDenied");
906            try_write!(message, OrderEmulated, "OrderEmulated");
907            try_write!(message, OrderSubmitted, "OrderSubmitted");
908            try_write!(message, OrderAccepted, "OrderAccepted");
909            try_write!(message, OrderRejected, "OrderRejected");
910            try_write!(message, OrderPendingCancel, "OrderPendingCancel");
911            try_write!(message, OrderCanceled, "OrderCanceled");
912            try_write!(message, OrderCancelRejected, "OrderCancelRejected");
913            try_write!(message, OrderExpired, "OrderExpired");
914            try_write!(message, OrderTriggered, "OrderTriggered");
915            try_write!(message, OrderPendingUpdate, "OrderPendingUpdate");
916            try_write!(message, OrderReleased, "OrderReleased");
917            try_write!(message, OrderModifyRejected, "OrderModifyRejected");
918            try_write!(message, OrderUpdated, "OrderUpdated");
919            try_write!(message, OrderFilled, "OrderFilled");
920            try_write!(message, PositionOpened, "PositionOpened");
921            try_write!(message, PositionChanged, "PositionChanged");
922            try_write!(message, PositionClosed, "PositionClosed");
923            try_write!(message, PositionAdjusted, "PositionAdjusted");
924            try_write!(message, OrderSnapshot, "OrderSnapshot");
925            try_write!(message, PositionSnapshot, "PositionSnapshot");
926            try_write!(message, OrderStatusReport, "OrderStatusReport");
927            try_write!(message, FillReport, "FillReport");
928            try_write!(message, PositionStatusReport, "PositionStatusReport");
929            try_write!(message, ExecutionMassStatus, "ExecutionMassStatus");
930
931            if let Some(deltas) = message.downcast_ref::<OrderBookDeltas>() {
932                // Batch write so chunk_metadata can skip a leading BookAction::Clear sentinel
933                let mut writer = writer.borrow_mut();
934                if let Err(e) = runtime.block_on(writer.write_batch(deltas.deltas.clone())) {
935                    log::warn!("Failed to write OrderBookDeltas: {e}");
936                }
937            } else if let Some(custom) = message.downcast_ref::<CustomData>() {
938                let mut writer = writer.borrow_mut();
939                if let Err(e) = runtime.block_on(writer.write_data(Data::Custom(custom.clone()))) {
940                    log::warn!("Failed to write CustomData: {e}");
941                }
942            } else if let Some(instrument) = message.downcast_ref::<InstrumentAny>() {
943                let mut writer = writer.borrow_mut();
944                if let Err(e) = runtime.block_on(writer.write_instrument(instrument.clone())) {
945                    log::warn!("Failed to write InstrumentAny: {e}");
946                }
947            }
948            // Silently ignore unsupported message types.
949        });
950
951        // Subscribe to all messages using wildcard pattern
952        subscribe_any(
953            MStr::pattern("*"),
954            handler.clone(),
955            None, // No priority
956        );
957
958        Ok(handler)
959    }
960
961    /// Unsubscribes from the message bus.
962    pub fn unsubscribe_from_message_bus(handler: &ShareableMessageHandler) {
963        unsubscribe_any(MStr::pattern("*"), handler);
964    }
965}
966
967#[cfg(test)]
968mod tests {
969    use std::{io::Cursor, sync::Arc};
970
971    use datafusion::arrow::ipc::reader::StreamReader;
972    use nautilus_common::clock::TestClock;
973    use nautilus_model::{
974        data::{Data, OrderBookDeltas_API, QuoteTick, TradeTick},
975        enums::AggressorSide,
976        identifiers::{InstrumentId, TradeId},
977        types::{Price, Quantity},
978    };
979    use nautilus_serialization::arrow::{
980        ArrowSchemaProvider, DecodeDataFromRecordBatch, EncodeToRecordBatch,
981    };
982    use object_store::{ObjectStore, local::LocalFileSystem};
983    use rstest::rstest;
984    use tempfile::TempDir;
985
986    use super::*;
987
988    #[tokio::test]
989    async fn test_writer_manager_keys() {
990        // Create a temporary directory for base path
991        let temp_dir = TempDir::new().unwrap();
992        let base_path = temp_dir.path().to_str().unwrap().to_string();
993
994        // Create a LocalFileSystem based object store using the temp directory
995        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
996        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
997
998        // Create a test clock
999        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1000        let timestamp = clock.borrow().timestamp_ns();
1001
1002        let quote_type_str = QuoteTick::path_prefix();
1003
1004        let mut per_instrument = HashSet::new();
1005        per_instrument.insert(quote_type_str.to_string());
1006
1007        let mut manager = FeatherWriter::new(
1008            base_path.clone(),
1009            store,
1010            clock,
1011            RotationConfig::NoRotation,
1012            None,
1013            Some(per_instrument),
1014            None, // flush_interval_ms
1015        );
1016
1017        let instrument_id = "AAPL.AAPL";
1018        // Write a dummy value
1019        let quote = QuoteTick::new(
1020            InstrumentId::from(instrument_id),
1021            Price::from("100.0"),
1022            Price::from("100.0"),
1023            Quantity::from("100.0"),
1024            Quantity::from("100.0"),
1025            UnixNanos::from(1_000_000_000_000_000_000),
1026            UnixNanos::from(1_000_000_000_000_000_000),
1027        );
1028
1029        let trade = TradeTick::new(
1030            InstrumentId::from(instrument_id),
1031            Price::from("100.0"),
1032            Quantity::from("100.0"),
1033            AggressorSide::Buyer,
1034            TradeId::from("1"),
1035            UnixNanos::from(1_000_000_000_000_000_000),
1036            UnixNanos::from(1_000_000_000_000_000_000),
1037        );
1038
1039        manager.write(quote).await.unwrap();
1040        manager.write(trade).await.unwrap();
1041
1042        // Check keys and paths for quotes and trades
1043        let path = manager.get_writer_path(&quote).unwrap();
1044        let safe_id = instrument_id.replace('/', "");
1045        let expected_path = Path::from(format!(
1046            "{base_path}/quotes/{safe_id}/{safe_id}_{timestamp}.feather"
1047        ));
1048        assert_eq!(path.path, expected_path);
1049        assert!(manager.writers.contains_key(&path));
1050        let writer = manager.writers.get(&path).unwrap();
1051        assert!(writer.size > 0);
1052
1053        let path = manager.get_writer_path(&trade).unwrap();
1054        let expected_path = Path::from(format!("{base_path}/trades_{timestamp}.feather"));
1055        assert_eq!(path.path, expected_path);
1056        assert!(manager.writers.contains_key(&path));
1057        let writer = manager.writers.get(&path).unwrap();
1058        assert!(writer.size > 0);
1059    }
1060
1061    #[rstest]
1062    fn test_file_writer_round_trip() {
1063        let instrument_id = "AAPL.AAPL";
1064        // Write a dummy value.
1065        let quote = QuoteTick::new(
1066            InstrumentId::from(instrument_id),
1067            Price::from("100.0"),
1068            Price::from("100.0"),
1069            Quantity::from("100.0"),
1070            Quantity::from("100.0"),
1071            UnixNanos::from(100),
1072            UnixNanos::from(100),
1073        );
1074        let metadata = QuoteTick::metadata(&quote);
1075        let schema = QuoteTick::get_schema(Some(metadata.clone()));
1076        let batch = QuoteTick::encode_batch(&QuoteTick::metadata(&quote), &[quote]).unwrap();
1077
1078        let mut writer = FeatherBuffer::new(&schema, RotationConfig::NoRotation).unwrap();
1079        writer.write_record_batch(&batch).unwrap();
1080
1081        let buffer = writer.take_buffer().unwrap();
1082        let mut reader = StreamReader::try_new(Cursor::new(buffer.as_slice()), None).unwrap();
1083
1084        let read_metadata = reader.schema().metadata().clone();
1085        assert_eq!(read_metadata, metadata);
1086
1087        let read_batch = reader.next().unwrap().unwrap();
1088        assert_eq!(read_batch.column(0), batch.column(0));
1089
1090        let decoded = QuoteTick::decode_data_batch(&metadata, batch).unwrap();
1091        assert_eq!(decoded[0], Data::from(quote));
1092    }
1093
1094    #[tokio::test]
1095    async fn test_round_trip() {
1096        // Create a temporary directory for base path
1097        let temp_dir = TempDir::new_in(".").unwrap();
1098        let base_path = temp_dir.path().to_str().unwrap().to_string();
1099
1100        // Create a LocalFileSystem based object store using the temp directory
1101        let local_fs = LocalFileSystem::new_with_prefix(&base_path).unwrap();
1102        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1103
1104        // Create a test clock
1105        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1106
1107        let quote_type_str = QuoteTick::path_prefix();
1108        let trade_type_str = TradeTick::path_prefix();
1109
1110        let mut per_instrument = HashSet::new();
1111        per_instrument.insert(quote_type_str.to_string());
1112        per_instrument.insert(trade_type_str.to_string());
1113
1114        let mut manager = FeatherWriter::new(
1115            base_path.clone(),
1116            store,
1117            clock,
1118            RotationConfig::NoRotation,
1119            None,
1120            Some(per_instrument),
1121            None, // flush_interval_ms
1122        );
1123
1124        let instrument_id = "AAPL.AAPL";
1125        // Write a dummy value.
1126        let quote = QuoteTick::new(
1127            InstrumentId::from(instrument_id),
1128            Price::from("100.0"),
1129            Price::from("100.0"),
1130            Quantity::from("100.0"),
1131            Quantity::from("100.0"),
1132            UnixNanos::from(100),
1133            UnixNanos::from(100),
1134        );
1135
1136        let trade = TradeTick::new(
1137            InstrumentId::from(instrument_id),
1138            Price::from("100.0"),
1139            Quantity::from("100.0"),
1140            AggressorSide::Buyer,
1141            TradeId::from("1"),
1142            UnixNanos::from(100),
1143            UnixNanos::from(100),
1144        );
1145
1146        manager.write(quote).await.unwrap();
1147        manager.write(trade).await.unwrap();
1148
1149        let paths = manager.writers.keys().cloned().collect::<Vec<_>>();
1150        assert_eq!(paths.len(), 2);
1151
1152        // Flush data
1153        manager.flush().await.unwrap();
1154
1155        // Read files from the temporary directory
1156        let mut recovered_quotes = Vec::new();
1157        let mut recovered_trades = Vec::new();
1158        let local_fs = LocalFileSystem::new_with_prefix(&base_path).unwrap();
1159        for path in paths {
1160            let path_str = local_fs.path_to_filesystem(&path.path).unwrap();
1161            let buffer = std::fs::File::open(&path_str).unwrap();
1162            let reader = StreamReader::try_new(buffer, None).unwrap();
1163            let metadata = reader.schema().metadata().clone();
1164            for batch in reader {
1165                let batch = batch.unwrap();
1166                if path_str.to_str().unwrap().contains("quotes") {
1167                    let decoded = QuoteTick::decode_data_batch(&metadata, batch).unwrap();
1168                    recovered_quotes.extend(decoded);
1169                } else if path_str.to_str().unwrap().contains("trades") {
1170                    let decoded = TradeTick::decode_data_batch(&metadata, batch).unwrap();
1171                    recovered_trades.extend(decoded);
1172                }
1173            }
1174        }
1175
1176        // Assert that the recovered data matches the written data
1177        assert_eq!(recovered_quotes.len(), 1, "Expected one QuoteTick record");
1178        assert_eq!(recovered_trades.len(), 1, "Expected one TradeTick record");
1179
1180        // Check key fields to ensure the data round-tripped correctly
1181        assert_eq!(recovered_quotes[0], Data::from(quote));
1182        assert_eq!(recovered_trades[0], Data::from(trade));
1183    }
1184
1185    #[tokio::test]
1186    async fn test_write_data_enum() {
1187        let temp_dir = TempDir::new().unwrap();
1188        let base_path = temp_dir.path().to_str().unwrap().to_string();
1189        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1190        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1191        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1192
1193        let mut writer = FeatherWriter::new(
1194            base_path,
1195            store,
1196            clock,
1197            RotationConfig::NoRotation,
1198            None,
1199            None,
1200            None,
1201        );
1202
1203        let quote = QuoteTick::new(
1204            InstrumentId::from("AUD/USD.SIM"),
1205            Price::from("1.0"),
1206            Price::from("1.0"),
1207            Quantity::from("1000"),
1208            Quantity::from("1000"),
1209            UnixNanos::from(1000),
1210            UnixNanos::from(1000),
1211        );
1212
1213        // Test writing via write_data
1214        writer.write_data(Data::Quote(quote)).await.unwrap();
1215        writer.flush().await.unwrap();
1216
1217        // Verify file was created
1218        assert!(!writer.writers.is_empty() || temp_dir.path().read_dir().unwrap().count() > 0);
1219    }
1220
1221    #[tokio::test]
1222    async fn test_write_data_all_types() {
1223        let temp_dir = TempDir::new().unwrap();
1224        let base_path = temp_dir.path().to_str().unwrap().to_string();
1225        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1226        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1227        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1228
1229        let mut writer = FeatherWriter::new(
1230            base_path,
1231            store,
1232            clock,
1233            RotationConfig::NoRotation,
1234            None,
1235            None,
1236            None,
1237        );
1238
1239        let instrument_id = InstrumentId::from("AUD/USD.SIM");
1240
1241        // Test all data types
1242        let quote = QuoteTick::new(
1243            instrument_id,
1244            Price::from("1.0"),
1245            Price::from("1.0"),
1246            Quantity::from("1000"),
1247            Quantity::from("1000"),
1248            UnixNanos::from(1000),
1249            UnixNanos::from(1000),
1250        );
1251        writer.write_data(Data::Quote(quote)).await.unwrap();
1252
1253        let trade = TradeTick::new(
1254            instrument_id,
1255            Price::from("1.0"),
1256            Quantity::from("1000"),
1257            AggressorSide::Buyer,
1258            TradeId::from("1"),
1259            UnixNanos::from(2000),
1260            UnixNanos::from(2000),
1261        );
1262        writer.write_data(Data::Trade(trade)).await.unwrap();
1263
1264        let delta = OrderBookDelta::clear(
1265            instrument_id,
1266            0,
1267            UnixNanos::from(3000),
1268            UnixNanos::from(3000),
1269        );
1270        writer.write_data(Data::Delta(delta)).await.unwrap();
1271
1272        writer.flush().await.unwrap();
1273    }
1274
1275    #[tokio::test]
1276    async fn test_auto_flush() {
1277        let temp_dir = TempDir::new().unwrap();
1278        let base_path = temp_dir.path().to_str().unwrap().to_string();
1279        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1280        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1281        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1282
1283        let mut writer = FeatherWriter::new(
1284            base_path,
1285            store,
1286            clock.clone(),
1287            RotationConfig::NoRotation,
1288            None,
1289            None,
1290            Some(100), // 100ms flush interval
1291        );
1292
1293        let quote = QuoteTick::new(
1294            InstrumentId::from("AUD/USD.SIM"),
1295            Price::from("1.0"),
1296            Price::from("1.0"),
1297            Quantity::from("1000"),
1298            Quantity::from("1000"),
1299            UnixNanos::from(1000),
1300            UnixNanos::from(1000),
1301        );
1302
1303        // Write first quote
1304        writer.write(quote).await.unwrap();
1305
1306        // Note: TestClock doesn't have set_time_ns, so we can't easily test auto-flush
1307        // with time advancement. Instead, we test that check_flush is called during write.
1308        // For a proper test, we'd need a mock clock or use LiveClock with time advancement.
1309
1310        // Write second quote - check_flush will be called but won't flush if time hasn't advanced
1311        let quote2 = QuoteTick::new(
1312            InstrumentId::from("AUD/USD.SIM"),
1313            Price::from("1.1"),
1314            Price::from("1.1"),
1315            Quantity::from("1000"),
1316            Quantity::from("1000"),
1317            UnixNanos::from(2000),
1318            UnixNanos::from(2000),
1319        );
1320        writer.write(quote2).await.unwrap();
1321
1322        // Verify that writes succeeded (check_flush was called, even if it didn't flush)
1323        // The flush_interval_ms is set, so check_flush runs but won't flush without time advancement
1324    }
1325
1326    #[tokio::test]
1327    async fn test_close() {
1328        let temp_dir = TempDir::new().unwrap();
1329        let base_path = temp_dir.path().to_str().unwrap().to_string();
1330        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1331        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1332        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1333
1334        let mut writer = FeatherWriter::new(
1335            base_path,
1336            store,
1337            clock,
1338            RotationConfig::NoRotation,
1339            None,
1340            None,
1341            None,
1342        );
1343
1344        let quote = QuoteTick::new(
1345            InstrumentId::from("AUD/USD.SIM"),
1346            Price::from("1.0"),
1347            Price::from("1.0"),
1348            Quantity::from("1000"),
1349            Quantity::from("1000"),
1350            UnixNanos::from(1000),
1351            UnixNanos::from(1000),
1352        );
1353
1354        writer.write(quote).await.unwrap();
1355        assert!(!writer.writers.is_empty());
1356
1357        writer.close().await.unwrap();
1358        assert!(writer.writers.is_empty());
1359    }
1360
1361    // Note: Message bus subscription test is skipped due to async/sync boundary complexity.
1362    // The handler uses block_on which can't be used from within an async runtime.
1363    // This functionality is better tested via Python integration tests where the message bus
1364    // is used in a non-async context or via proper async task spawning.
1365
1366    #[tokio::test]
1367    async fn test_write_data_orderbook_deltas() {
1368        let temp_dir = TempDir::new().unwrap();
1369        let base_path = temp_dir.path().to_str().unwrap().to_string();
1370        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1371        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1372        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1373
1374        let mut writer = FeatherWriter::new(
1375            base_path,
1376            store,
1377            clock,
1378            RotationConfig::NoRotation,
1379            None,
1380            None,
1381            None,
1382        );
1383
1384        let instrument_id = InstrumentId::from("AUD/USD.SIM");
1385        let delta1 = OrderBookDelta::clear(
1386            instrument_id,
1387            0,
1388            UnixNanos::from(1000),
1389            UnixNanos::from(1000),
1390        );
1391        let delta2 = OrderBookDelta::clear(
1392            instrument_id,
1393            0,
1394            UnixNanos::from(2000),
1395            UnixNanos::from(2000),
1396        );
1397
1398        let book_deltas = OrderBookDeltas::new(instrument_id, vec![delta1, delta2]);
1399        let deltas_api = OrderBookDeltas_API::new(book_deltas);
1400
1401        // Test writing OrderBookDeltas via write_data
1402        writer.write_data(Data::Deltas(deltas_api)).await.unwrap();
1403        writer.flush().await.unwrap();
1404    }
1405
1406    #[tokio::test]
1407    #[cfg(feature = "python")]
1408    async fn test_write_custom_data_round_trip() {
1409        use std::sync::Arc;
1410
1411        use futures::StreamExt;
1412        use nautilus_model::{
1413            data::{CustomData, Data, DataType},
1414            identifiers::InstrumentId,
1415        };
1416        use nautilus_serialization::{
1417            arrow::custom::CustomDataDecoder, ensure_custom_data_registered,
1418        };
1419
1420        use crate::test_data::RustTestCustomData;
1421
1422        ensure_custom_data_registered::<RustTestCustomData>();
1423
1424        let temp_dir = TempDir::new().unwrap();
1425        let base_path = temp_dir.path().to_str().unwrap().to_string();
1426        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1427        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1428        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1429
1430        let mut writer = FeatherWriter::new(
1431            base_path.clone(),
1432            store.clone(),
1433            clock,
1434            RotationConfig::NoRotation,
1435            None,
1436            None,
1437            None,
1438        );
1439
1440        let instrument_id = InstrumentId::from("RUST.TEST");
1441        let data_type = DataType::new("RustTestCustomData", None, Some(instrument_id.to_string()));
1442        let original = RustTestCustomData {
1443            instrument_id,
1444            value: 1.23,
1445            flag: true,
1446            ts_event: UnixNanos::from(1000),
1447            ts_init: UnixNanos::from(1000),
1448        };
1449        let custom = CustomData::new(Arc::new(original.clone()), data_type);
1450
1451        writer
1452            .write_data(Data::Custom(custom))
1453            .await
1454            .expect("write_data CustomData");
1455        writer.flush().await.expect("flush");
1456
1457        let prefix = Path::from(format!("{base_path}/data/custom/RustTestCustomData"));
1458        let mut list_stream = store.list(Some(&prefix));
1459        let first = list_stream.next().await.expect("at least one object");
1460        let meta = first.expect("list item");
1461        let bytes = store
1462            .get(&meta.location)
1463            .await
1464            .expect("get")
1465            .bytes()
1466            .await
1467            .expect("bytes");
1468        let mut reader =
1469            StreamReader::try_new(Cursor::new(bytes.as_ref()), None).expect("StreamReader");
1470        let schema = reader.schema();
1471        let metadata: std::collections::HashMap<String, String> = schema
1472            .metadata()
1473            .iter()
1474            .map(|(k, v)| (k.clone(), v.clone()))
1475            .collect();
1476        let batch = reader.next().expect("batch").expect("batch ok");
1477        let decoded =
1478            CustomDataDecoder::decode_data_batch(&metadata, batch).expect("decode_data_batch");
1479        assert_eq!(decoded.len(), 1);
1480        if let Data::Custom(decoded_custom) = &decoded[0] {
1481            assert_eq!(decoded_custom.data_type.type_name(), "RustTestCustomData");
1482            let rust: &RustTestCustomData = decoded_custom
1483                .data
1484                .as_any()
1485                .downcast_ref::<RustTestCustomData>()
1486                .expect("RustTestCustomData");
1487            assert_eq!(rust, &original);
1488        } else {
1489            panic!("Expected Data::Custom");
1490        }
1491    }
1492}
nautilus_persistence/backend/feather.rs

nautilus_persistence/backend/
feather.rs
