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 datafusion::arrow::{
26    datatypes::Schema, error::ArrowError, ipc::writer::StreamWriter, record_batch::RecordBatch,
27};
28use jiff::{
29    SignedDuration,
30    civil::Time,
31    tz::{AmbiguousOffset, TimeZone},
32};
33use nautilus_common::{
34    cache::fifo::FifoCache,
35    clock::Clock,
36    msgbus::{mstr::MStr, subscribe_any, typed_handler::ShareableMessageHandler, unsubscribe_any},
37};
38use nautilus_core::{UUID4, UnixNanos, datetime::NANOSECONDS_IN_SECOND};
39use nautilus_model::{
40    data::{
41        Bar, CatalogPathPrefix, CustomData, CustomDataTrait, Data, FundingRateUpdate,
42        IndexPriceUpdate, InstrumentStatus, MarkPriceUpdate, OptionGreeks, OrderBookDelta,
43        OrderBookDeltas, OrderBookDepth10, QuoteTick, TradeTick, close::InstrumentClose,
44        encode_custom_to_arrow, get_arrow_schema,
45    },
46    events::{
47        AccountState, OrderAccepted, OrderCancelRejected, OrderCanceled, OrderDenied,
48        OrderEmulated, OrderExpired, OrderFillVoided, OrderFilled, OrderInitialized,
49        OrderModifyRejected, OrderPendingCancel, OrderPendingUpdate, OrderRejected, OrderReleased,
50        OrderSnapshot, OrderSubmitted, OrderTriggered, OrderUpdated, PositionAdjusted,
51        PositionChanged, PositionClosed, PositionOpened, PositionSnapshot,
52    },
53    instruments::InstrumentAny,
54    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
55};
56use nautilus_serialization::arrow::{EncodeToRecordBatch, KEY_INSTRUMENT_ID};
57use object_store::{ObjectStore, ObjectStoreExt, path::Path};
58
59use super::catalog::urisafe_instrument_id;
60use crate::backend::{
61    catalog::safe_directory_identifier,
62    custom::{augment_batch_with_data_type_column, schema_with_data_type_column},
63};
64
65#[derive(Debug, Default, PartialEq, PartialOrd, Hash, Eq, Clone)]
66pub struct FileWriterPath {
67    path: Path,
68    type_str: String,
69    instrument_id: Option<String>,
70}
71
72/// A `FeatherBuffer` encodes data via an Arrow `StreamWriter`.
73///
74/// It flushes the internal byte buffer according to rotation policy.
75pub struct FeatherBuffer {
76    /// Arrow `StreamWriter` that writes to an in-memory `Vec<u8>`.
77    writer: StreamWriter<Vec<u8>>,
78    /// Current size in bytes.
79    size: u64,
80    /// TODO: Optional next rotation timestamp.
81    // next_rotation: Option<UnixNanos>,
82    /// Schema of the data being written.
83    schema: Schema,
84    /// Maximum buffer size in bytes.
85    max_buffer_size: u64,
86    /// Rotation config
87    rotation_config: RotationConfig,
88}
89
90impl FeatherBuffer {
91    /// Creates a new [`FeatherBuffer`] using the given path, schema, and maximum buffer size.
92    ///
93    /// # Errors
94    ///
95    /// Returns an error if the Arrow stream writer cannot be created.
96    pub fn new(schema: &Schema, rotation_config: RotationConfig) -> Result<Self, ArrowError> {
97        let writer = StreamWriter::try_new(Vec::new(), schema)?;
98        let mut max_buffer_size = 1_000_000_000_000; // 1 GB
99
100        if let RotationConfig::Size { max_size } = &rotation_config {
101            max_buffer_size = *max_size;
102        }
103
104        Ok(Self {
105            writer,
106            size: 0,
107            // next_rotation: None,
108            max_buffer_size,
109            schema: schema.clone(),
110            rotation_config,
111        })
112    }
113
114    /// Writes the given `RecordBatch` to the internal buffer.
115    ///
116    /// Returns true if it should be rotated according rotation policy
117    ///
118    /// # Errors
119    ///
120    /// Returns an error if Arrow IPC writing fails.
121    pub fn write_record_batch(&mut self, batch: &RecordBatch) -> Result<bool, ArrowError> {
122        self.writer.write(batch)?;
123        self.size += batch.get_array_memory_size() as u64;
124        Ok(self.size >= self.max_buffer_size)
125    }
126
127    /// Consumes the writer and returns the buffer of bytes from the `StreamWriter`
128    ///
129    /// # Errors
130    ///
131    /// Returns an error if the replacement writer cannot be created or the previous
132    /// writer cannot be finalized.
133    pub fn take_buffer(&mut self) -> Result<Vec<u8>, ArrowError> {
134        let mut writer = StreamWriter::try_new(Vec::new(), &self.schema)?;
135        std::mem::swap(&mut self.writer, &mut writer);
136        let buffer = writer.into_inner()?;
137        // TODO: Handle rotation config here
138        self.size = 0;
139        Ok(buffer)
140    }
141
142    /// Should rotate
143    #[must_use]
144    pub const fn should_rotate(&self) -> bool {
145        match &self.rotation_config {
146            RotationConfig::Size { max_size } => self.size >= *max_size,
147            _ => false,
148        }
149    }
150}
151
152/// Configuration for file rotation.
153#[derive(Debug, Clone)]
154pub enum RotationConfig {
155    /// Rotate based on file size.
156    Size {
157        /// Maximum buffer size in bytes before rotation.
158        max_size: u64,
159    },
160    /// Rotate based on a time interval.
161    Interval {
162        /// Interval in nanoseconds.
163        interval_ns: u64,
164    },
165    /// Rotate based on scheduled dates.
166    ScheduledDates {
167        /// Interval in nanoseconds.
168        interval_ns: u64,
169        /// Time of day for rotation (nanoseconds since midnight).
170        rotation_time: UnixNanos,
171        /// Timezone for rotation calculations.
172        rotation_timezone: TimeZone,
173    },
174    /// No automatic rotation.
175    NoRotation,
176}
177
178/// Manages multiple `FeatherBuffers` and handles encoding, rotation, and flushing to the object store.
179///
180/// The `write()` method is the single entry point for clients: they supply a data value (of generic type T)
181/// and the manager encodes it (using T's metadata via `EncodeToRecordBatch`), routes it by `CatalogPathPrefix`,
182/// and writes it to the appropriate `FileWriter`. When a writer's buffer is full or rotation criteria are met,
183/// its contents are flushed to the object store and it is replaced.
184pub struct FeatherWriter {
185    /// Base directory for writing files.
186    base_path: String,
187    /// Object store for persistence.
188    store: Arc<dyn ObjectStore>,
189    /// Clock for timestamps and rotation.
190    clock: Rc<RefCell<dyn Clock>>,
191    /// Rotation configuration.
192    rotation_config: RotationConfig,
193    /// Optional set of type names to include.
194    included_types: Option<HashSet<String>>,
195    /// Set of types that should be split by instrument.
196    per_instrument_types: HashSet<String>,
197    /// Map of active `FeatherBuffers` keyed by their path.
198    writers: HashMap<FileWriterPath, FeatherBuffer>,
199    /// Map of next rotation times keyed by their path.
200    next_rotation_times: HashMap<FileWriterPath, UnixNanos>,
201    /// Runtime handle for async operations.
202    runtime: tokio::runtime::Handle,
203    /// Flush interval in milliseconds (0 = no automatic flushing).
204    flush_interval_ms: u64,
205    /// Last flush timestamp in nanoseconds.
206    last_flush_ns: UnixNanos,
207    /// Bounded cache of recently seen event IDs for deduplication.
208    seen_event_ids: Box<FifoCache<UUID4, 10_000>>,
209}
210
211impl FeatherWriter {
212    /// Creates a new [`FeatherWriter`] instance.
213    pub fn new(
214        base_path: String,
215        store: Arc<dyn ObjectStore>,
216        clock: Rc<RefCell<dyn Clock>>,
217        rotation_config: RotationConfig,
218        included_types: Option<HashSet<String>>,
219        per_instrument_types: Option<HashSet<String>>,
220        flush_interval_ms: Option<u64>,
221    ) -> Self {
222        // Get the runtime handle for async operations
223        let runtime = nautilus_common::live::get_runtime().handle().clone();
224        let flush_interval_ms = flush_interval_ms.unwrap_or(1000); // Default 1 second
225        let last_flush_ns = clock.borrow().timestamp_ns();
226
227        Self {
228            base_path,
229            store,
230            clock,
231            rotation_config,
232            included_types,
233            per_instrument_types: per_instrument_types.unwrap_or_default(),
234            writers: HashMap::new(),
235            next_rotation_times: HashMap::new(),
236            runtime,
237            flush_interval_ms,
238            last_flush_ns,
239            seen_event_ids: Box::new(FifoCache::new()),
240        }
241    }
242
243    /// Writes a single data value.
244    /// This is the user entry point. The data is encoded into a `RecordBatch` and written to the appropriate `FileWriter`.
245    /// If the writer's buffer reaches capacity or meets rotation criteria (based on the rotation configuration),
246    /// the `FileWriter` is flushed to the object store and replaced.
247    ///
248    /// # Errors
249    ///
250    /// Returns an error if path selection, Arrow encoding, writer creation, buffering,
251    /// rotation, or flushing fails.
252    pub async fn write<T>(&mut self, data: T) -> Result<(), Box<dyn std::error::Error>>
253    where
254        T: EncodeToRecordBatch + CatalogPathPrefix + 'static,
255    {
256        if !self.should_write::<T>() {
257            return Ok(());
258        }
259
260        let path = self.get_writer_path(&data)?;
261
262        // Create a new FileWriter if one does not exist.
263        if !self.writers.contains_key(&path) {
264            self.create_writer::<T>(path.clone(), &data)?;
265        }
266
267        // Encode the data into a RecordBatch using T's encoding logic.
268        let batch = T::encode_batch(&T::metadata(&data), &[data])?;
269
270        // Write the RecordBatch to the appropriate FileWriter.
271        if let Some(writer) = self.writers.get_mut(&path) {
272            let should_rotate = writer.write_record_batch(&batch)?;
273            if should_rotate || self.check_scheduled_rotation(&path) {
274                self.rotate_writer(&path).await?;
275            }
276        }
277
278        // Check if we should auto-flush based on time interval
279        self.check_flush().await?;
280
281        Ok(())
282    }
283
284    /// Writes a batch of data values as one or more `RecordBatch`es.
285    ///
286    /// Uses `T::chunk_metadata` to derive the file schema metadata. This protects
287    /// types like `OrderBookDelta` from having their file metadata poisoned by a
288    /// leading sentinel row (e.g. `BookAction::Clear`, which carries
289    /// `price_precision=0, size_precision=0`).
290    ///
291    /// Per-instrument types are partitioned by instrument so a mixed-instrument
292    /// batch lands in the correct file for each instrument.
293    ///
294    /// # Errors
295    ///
296    /// Returns an error if path selection, Arrow encoding, writer creation, buffering,
297    /// rotation, or flushing fails.
298    pub async fn write_batch<T>(&mut self, data: Vec<T>) -> Result<(), Box<dyn std::error::Error>>
299    where
300        T: EncodeToRecordBatch + CatalogPathPrefix + 'static,
301    {
302        if data.is_empty() || !self.should_write::<T>() {
303            return Ok(());
304        }
305
306        // Group by logical writer identity (instrument_id for per-instrument types).
307        // Grouping on FileWriterPath would split same-instrument rows across distinct
308        // timestamped paths when the writer does not yet exist under a LiveClock.
309        let type_str = T::path_prefix();
310        let needs_instrument =
311            self.per_instrument_types.contains(type_str) || type_str.starts_with("custom_");
312
313        let mut groups: AHashMap<Option<String>, Vec<T>> = AHashMap::new();
314
315        for item in data {
316            let instrument_id = if needs_instrument {
317                T::metadata(&item).get(KEY_INSTRUMENT_ID).cloned()
318            } else {
319                None
320            };
321            groups.entry(instrument_id).or_default().push(item);
322        }
323
324        for group in groups.into_values() {
325            let path = self.get_writer_path(&group[0])?;
326            let metadata = T::chunk_metadata(&group);
327
328            if !self.writers.contains_key(&path) {
329                self.create_writer_with_metadata::<T>(path.clone(), metadata.clone())?;
330            }
331
332            let batch = T::encode_batch(&metadata, &group)?;
333
334            if let Some(writer) = self.writers.get_mut(&path) {
335                let should_rotate = writer.write_record_batch(&batch)?;
336                if should_rotate || self.check_scheduled_rotation(&path) {
337                    self.rotate_writer(&path).await?;
338                }
339            }
340        }
341
342        self.check_flush().await?;
343
344        Ok(())
345    }
346
347    /// Checks if enough time has passed since last flush and flushes if needed.
348    async fn check_flush(&mut self) -> Result<(), Box<dyn std::error::Error>> {
349        if self.flush_interval_ms == 0 {
350            return Ok(()); // Auto-flush disabled
351        }
352
353        let now_ns = self.clock.borrow().timestamp_ns();
354        let elapsed_ms = (now_ns.as_u64() - self.last_flush_ns.as_u64()) / 1_000_000;
355
356        if elapsed_ms >= self.flush_interval_ms {
357            self.flush().await?;
358            self.last_flush_ns = now_ns;
359        }
360
361        Ok(())
362    }
363
364    fn check_scheduled_rotation(&mut self, path: &FileWriterPath) -> bool {
365        match &self.rotation_config {
366            RotationConfig::Interval { interval_ns } => {
367                let now = self.clock.borrow().timestamp_ns();
368                let next_rotation = self.next_rotation_times.get(path).copied();
369
370                match next_rotation {
371                    None => {
372                        self.next_rotation_times
373                            .insert(path.clone(), now + *interval_ns);
374                        false
375                    }
376                    Some(next) if now >= next => {
377                        self.next_rotation_times
378                            .insert(path.clone(), now + *interval_ns);
379                        true
380                    }
381                    _ => false,
382                }
383            }
384            RotationConfig::ScheduledDates {
385                interval_ns,
386                rotation_time,
387                rotation_timezone,
388            } => {
389                let now = self.clock.borrow().timestamp_ns();
390                let next_rotation = self.next_rotation_times.get(path).copied();
391
392                match next_rotation {
393                    None => {
394                        let next = self.calculate_next_scheduled_rotation(
395                            *rotation_time,
396                            rotation_timezone,
397                            *interval_ns,
398                        );
399                        self.next_rotation_times.insert(path.clone(), next);
400                        false
401                    }
402                    Some(next) if now >= next => {
403                        self.next_rotation_times
404                            .insert(path.clone(), now + *interval_ns);
405                        true
406                    }
407                    _ => false,
408                }
409            }
410            _ => false,
411        }
412    }
413
414    fn calculate_next_scheduled_rotation(
415        &self,
416        rotation_time: UnixNanos,
417        rotation_timezone: &TimeZone,
418        interval_ns: u64,
419    ) -> UnixNanos {
420        let now_utc = self.clock.borrow().utc_now();
421        let now_local = rotation_timezone.to_datetime(now_utc);
422
423        let rotation_time_secs = u32::try_from(*rotation_time / NANOSECONDS_IN_SECOND).unwrap_or(0);
424        let rotation_time_nanos =
425            i32::try_from(*rotation_time % NANOSECONDS_IN_SECOND).unwrap_or(0);
426        let rotation_time = if rotation_time_secs < 86_400 {
427            Time::new(
428                i8::try_from(rotation_time_secs / 3_600).unwrap_or(0),
429                i8::try_from(rotation_time_secs % 3_600 / 60).unwrap_or(0),
430                i8::try_from(rotation_time_secs % 60).unwrap_or(0),
431                rotation_time_nanos,
432            )
433            .unwrap_or(Time::MIN)
434        } else {
435            Time::MIN
436        };
437
438        let local_rotation = now_local.date().to_datetime(rotation_time);
439        let ambiguous = rotation_timezone.to_ambiguous_timestamp(local_rotation);
440        let mut next_rotation = match ambiguous.offset() {
441            AmbiguousOffset::Gap { .. } => now_utc,
442            _ => ambiguous.earlier().unwrap_or(now_utc),
443        };
444
445        if next_rotation <= now_utc {
446            // If the time has already passed today, we would usually add the interval
447            // But let's align exactly with how Python does it:
448            while next_rotation <= now_utc {
449                next_rotation += SignedDuration::from_nanos_i128(i128::from(interval_ns));
450            }
451        }
452
453        UnixNanos::from(u64::try_from(next_rotation.as_nanosecond()).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::IndexPrice(price) => self.write(price).await,
784            Data::MarkPrice(price) => self.write(price).await,
785            Data::FundingRate(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) => {
791                // Batch write so chunk_metadata can skip a leading BookAction::Clear sentinel
792                self.write_batch(deltas.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, OrderFillVoided, "OrderFillVoided");
921            try_write!(message, PositionOpened, "PositionOpened");
922            try_write!(message, PositionChanged, "PositionChanged");
923            try_write!(message, PositionClosed, "PositionClosed");
924            try_write!(message, PositionAdjusted, "PositionAdjusted");
925            try_write!(message, OrderSnapshot, "OrderSnapshot");
926            try_write!(message, PositionSnapshot, "PositionSnapshot");
927            try_write!(message, OrderStatusReport, "OrderStatusReport");
928            try_write!(message, FillReport, "FillReport");
929            try_write!(message, PositionStatusReport, "PositionStatusReport");
930            try_write!(message, ExecutionMassStatus, "ExecutionMassStatus");
931
932            if let Some(deltas) = message.downcast_ref::<OrderBookDeltas>() {
933                // Batch write so chunk_metadata can skip a leading BookAction::Clear sentinel
934                let mut writer = writer.borrow_mut();
935                if let Err(e) = runtime.block_on(writer.write_batch(deltas.deltas.clone())) {
936                    log::warn!("Failed to write OrderBookDeltas: {e}");
937                }
938            } else if let Some(custom) = message.downcast_ref::<CustomData>() {
939                let mut writer = writer.borrow_mut();
940                if let Err(e) = runtime.block_on(writer.write_data(Data::Custom(custom.clone()))) {
941                    log::warn!("Failed to write CustomData: {e}");
942                }
943            } else if let Some(instrument) = message.downcast_ref::<InstrumentAny>() {
944                let mut writer = writer.borrow_mut();
945                if let Err(e) = runtime.block_on(writer.write_instrument(instrument.clone())) {
946                    log::warn!("Failed to write InstrumentAny: {e}");
947                }
948            }
949            // Silently ignore unsupported message types.
950        });
951
952        // Subscribe to all messages using wildcard pattern
953        subscribe_any(
954            MStr::pattern("*"),
955            handler.clone(),
956            None, // No priority
957        );
958
959        Ok(handler)
960    }
961
962    /// Unsubscribes from the message bus.
963    pub fn unsubscribe_from_message_bus(handler: &ShareableMessageHandler) {
964        unsubscribe_any(MStr::pattern("*"), handler);
965    }
966}
967
968#[cfg(test)]
969mod tests {
970    use std::{io::Cursor, sync::Arc};
971
972    use datafusion::arrow::ipc::reader::StreamReader;
973    use nautilus_common::clock::TestClock;
974    use nautilus_model::{
975        data::{Data, QuoteTick, TradeTick},
976        enums::AggressorSide,
977        identifiers::{InstrumentId, TradeId},
978        types::{Price, Quantity},
979    };
980    use nautilus_serialization::arrow::{
981        ArrowSchemaProvider, DecodeDataFromRecordBatch, EncodeToRecordBatch,
982    };
983    use object_store::{ObjectStore, local::LocalFileSystem};
984    use rstest::rstest;
985    use tempfile::TempDir;
986
987    use super::*;
988
989    #[tokio::test]
990    async fn test_writer_manager_keys() {
991        // Create a temporary directory for base path
992        let temp_dir = TempDir::new().unwrap();
993        let base_path = temp_dir.path().to_str().unwrap().to_string();
994
995        // Create a LocalFileSystem based object store using the temp directory
996        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
997        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
998
999        // Create a test clock
1000        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1001        let timestamp = clock.borrow().timestamp_ns();
1002
1003        let quote_type_str = QuoteTick::path_prefix();
1004
1005        let mut per_instrument = HashSet::new();
1006        per_instrument.insert(quote_type_str.to_string());
1007
1008        let mut manager = FeatherWriter::new(
1009            base_path.clone(),
1010            store,
1011            clock,
1012            RotationConfig::NoRotation,
1013            None,
1014            Some(per_instrument),
1015            None, // flush_interval_ms
1016        );
1017
1018        let instrument_id = "AAPL.AAPL";
1019        // Write a dummy value
1020        let quote = QuoteTick::new(
1021            InstrumentId::from(instrument_id),
1022            Price::from("100.0"),
1023            Price::from("100.0"),
1024            Quantity::from("100.0"),
1025            Quantity::from("100.0"),
1026            UnixNanos::from(1_000_000_000_000_000_000),
1027            UnixNanos::from(1_000_000_000_000_000_000),
1028        );
1029
1030        let trade = TradeTick::new(
1031            InstrumentId::from(instrument_id),
1032            Price::from("100.0"),
1033            Quantity::from("100.0"),
1034            AggressorSide::Buy,
1035            TradeId::from("1"),
1036            UnixNanos::from(1_000_000_000_000_000_000),
1037            UnixNanos::from(1_000_000_000_000_000_000),
1038        );
1039
1040        manager.write(quote).await.unwrap();
1041        manager.write(trade).await.unwrap();
1042
1043        // Check keys and paths for quotes and trades
1044        let path = manager.get_writer_path(&quote).unwrap();
1045        let safe_id = instrument_id.replace('/', "");
1046        let expected_path = Path::from(format!(
1047            "{base_path}/quotes/{safe_id}/{safe_id}_{timestamp}.feather"
1048        ));
1049        assert_eq!(path.path, expected_path);
1050        assert!(manager.writers.contains_key(&path));
1051        let writer = manager.writers.get(&path).unwrap();
1052        assert!(writer.size > 0);
1053
1054        let path = manager.get_writer_path(&trade).unwrap();
1055        let expected_path = Path::from(format!("{base_path}/trades_{timestamp}.feather"));
1056        assert_eq!(path.path, expected_path);
1057        assert!(manager.writers.contains_key(&path));
1058        let writer = manager.writers.get(&path).unwrap();
1059        assert!(writer.size > 0);
1060    }
1061
1062    #[rstest]
1063    fn test_file_writer_round_trip() {
1064        let instrument_id = "AAPL.AAPL";
1065        // Write a dummy value.
1066        let quote = QuoteTick::new(
1067            InstrumentId::from(instrument_id),
1068            Price::from("100.0"),
1069            Price::from("100.0"),
1070            Quantity::from("100.0"),
1071            Quantity::from("100.0"),
1072            UnixNanos::from(100),
1073            UnixNanos::from(100),
1074        );
1075        let metadata = QuoteTick::metadata(&quote);
1076        let schema = QuoteTick::get_schema(Some(metadata.clone()));
1077        let batch = QuoteTick::encode_batch(&QuoteTick::metadata(&quote), &[quote]).unwrap();
1078
1079        let mut writer = FeatherBuffer::new(&schema, RotationConfig::NoRotation).unwrap();
1080        writer.write_record_batch(&batch).unwrap();
1081
1082        let buffer = writer.take_buffer().unwrap();
1083        let mut reader = StreamReader::try_new(Cursor::new(buffer.as_slice()), None).unwrap();
1084
1085        let read_metadata = reader.schema().metadata().clone();
1086        assert_eq!(read_metadata, metadata);
1087
1088        let read_batch = reader.next().unwrap().unwrap();
1089        assert_eq!(read_batch.column(0), batch.column(0));
1090
1091        let decoded = QuoteTick::decode_data_batch(&metadata, batch).unwrap();
1092        assert_eq!(decoded[0], Data::from(quote));
1093    }
1094
1095    #[tokio::test]
1096    async fn test_round_trip() {
1097        // Create a temporary directory for base path
1098        let temp_dir = TempDir::new_in(".").unwrap();
1099        let base_path = temp_dir.path().to_str().unwrap().to_string();
1100
1101        // Create a LocalFileSystem based object store using the temp directory
1102        let local_fs = LocalFileSystem::new_with_prefix(&base_path).unwrap();
1103        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1104
1105        // Create a test clock
1106        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1107
1108        let quote_type_str = QuoteTick::path_prefix();
1109        let trade_type_str = TradeTick::path_prefix();
1110
1111        let mut per_instrument = HashSet::new();
1112        per_instrument.insert(quote_type_str.to_string());
1113        per_instrument.insert(trade_type_str.to_string());
1114
1115        let mut manager = FeatherWriter::new(
1116            base_path.clone(),
1117            store,
1118            clock,
1119            RotationConfig::NoRotation,
1120            None,
1121            Some(per_instrument),
1122            None, // flush_interval_ms
1123        );
1124
1125        let instrument_id = "AAPL.AAPL";
1126        // Write a dummy value.
1127        let quote = QuoteTick::new(
1128            InstrumentId::from(instrument_id),
1129            Price::from("100.0"),
1130            Price::from("100.0"),
1131            Quantity::from("100.0"),
1132            Quantity::from("100.0"),
1133            UnixNanos::from(100),
1134            UnixNanos::from(100),
1135        );
1136
1137        let trade = TradeTick::new(
1138            InstrumentId::from(instrument_id),
1139            Price::from("100.0"),
1140            Quantity::from("100.0"),
1141            AggressorSide::Buy,
1142            TradeId::from("1"),
1143            UnixNanos::from(100),
1144            UnixNanos::from(100),
1145        );
1146
1147        manager.write(quote).await.unwrap();
1148        manager.write(trade).await.unwrap();
1149
1150        let paths = manager.writers.keys().cloned().collect::<Vec<_>>();
1151        assert_eq!(paths.len(), 2);
1152
1153        // Flush data
1154        manager.flush().await.unwrap();
1155
1156        // Read files from the temporary directory
1157        let mut recovered_quotes = Vec::new();
1158        let mut recovered_trades = Vec::new();
1159        let local_fs = LocalFileSystem::new_with_prefix(&base_path).unwrap();
1160        for path in paths {
1161            let path_str = local_fs.path_to_filesystem(&path.path).unwrap();
1162            let buffer = std::fs::File::open(&path_str).unwrap();
1163            let reader = StreamReader::try_new(buffer, None).unwrap();
1164            let metadata = reader.schema().metadata().clone();
1165            for batch in reader {
1166                let batch = batch.unwrap();
1167                if path_str.to_str().unwrap().contains("quotes") {
1168                    let decoded = QuoteTick::decode_data_batch(&metadata, batch).unwrap();
1169                    recovered_quotes.extend(decoded);
1170                } else if path_str.to_str().unwrap().contains("trades") {
1171                    let decoded = TradeTick::decode_data_batch(&metadata, batch).unwrap();
1172                    recovered_trades.extend(decoded);
1173                }
1174            }
1175        }
1176
1177        // Assert that the recovered data matches the written data
1178        assert_eq!(recovered_quotes.len(), 1, "Expected one QuoteTick record");
1179        assert_eq!(recovered_trades.len(), 1, "Expected one TradeTick record");
1180
1181        // Check key fields to ensure the data round-tripped correctly
1182        assert_eq!(recovered_quotes[0], Data::from(quote));
1183        assert_eq!(recovered_trades[0], Data::from(trade));
1184    }
1185
1186    #[tokio::test]
1187    async fn test_write_data_enum() {
1188        let temp_dir = TempDir::new().unwrap();
1189        let base_path = temp_dir.path().to_str().unwrap().to_string();
1190        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1191        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1192        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1193
1194        let mut writer = FeatherWriter::new(
1195            base_path,
1196            store,
1197            clock,
1198            RotationConfig::NoRotation,
1199            None,
1200            None,
1201            None,
1202        );
1203
1204        let quote = QuoteTick::new(
1205            InstrumentId::from("AUD/USD.SIM"),
1206            Price::from("1.0"),
1207            Price::from("1.0"),
1208            Quantity::from("1000"),
1209            Quantity::from("1000"),
1210            UnixNanos::from(1000),
1211            UnixNanos::from(1000),
1212        );
1213
1214        // Test writing via write_data
1215        writer.write_data(Data::Quote(quote)).await.unwrap();
1216        writer.flush().await.unwrap();
1217
1218        // Verify file was created
1219        assert!(!writer.writers.is_empty() || temp_dir.path().read_dir().unwrap().count() > 0);
1220    }
1221
1222    #[tokio::test]
1223    async fn test_write_data_all_types() {
1224        let temp_dir = TempDir::new().unwrap();
1225        let base_path = temp_dir.path().to_str().unwrap().to_string();
1226        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1227        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1228        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1229
1230        let mut writer = FeatherWriter::new(
1231            base_path,
1232            store,
1233            clock,
1234            RotationConfig::NoRotation,
1235            None,
1236            None,
1237            None,
1238        );
1239
1240        let instrument_id = InstrumentId::from("AUD/USD.SIM");
1241
1242        // Test all data types
1243        let quote = QuoteTick::new(
1244            instrument_id,
1245            Price::from("1.0"),
1246            Price::from("1.0"),
1247            Quantity::from("1000"),
1248            Quantity::from("1000"),
1249            UnixNanos::from(1000),
1250            UnixNanos::from(1000),
1251        );
1252        writer.write_data(Data::Quote(quote)).await.unwrap();
1253
1254        let trade = TradeTick::new(
1255            instrument_id,
1256            Price::from("1.0"),
1257            Quantity::from("1000"),
1258            AggressorSide::Buy,
1259            TradeId::from("1"),
1260            UnixNanos::from(2000),
1261            UnixNanos::from(2000),
1262        );
1263        writer.write_data(Data::Trade(trade)).await.unwrap();
1264
1265        let delta = OrderBookDelta::clear(
1266            instrument_id,
1267            0,
1268            UnixNanos::from(3000),
1269            UnixNanos::from(3000),
1270        );
1271        writer.write_data(Data::Delta(delta)).await.unwrap();
1272
1273        writer.flush().await.unwrap();
1274    }
1275
1276    #[tokio::test]
1277    async fn test_auto_flush() {
1278        let temp_dir = TempDir::new().unwrap();
1279        let base_path = temp_dir.path().to_str().unwrap().to_string();
1280        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1281        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1282        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1283
1284        let mut writer = FeatherWriter::new(
1285            base_path,
1286            store,
1287            clock.clone(),
1288            RotationConfig::NoRotation,
1289            None,
1290            None,
1291            Some(100), // 100ms flush interval
1292        );
1293
1294        let quote = QuoteTick::new(
1295            InstrumentId::from("AUD/USD.SIM"),
1296            Price::from("1.0"),
1297            Price::from("1.0"),
1298            Quantity::from("1000"),
1299            Quantity::from("1000"),
1300            UnixNanos::from(1000),
1301            UnixNanos::from(1000),
1302        );
1303
1304        // Write first quote
1305        writer.write(quote).await.unwrap();
1306
1307        // Note: TestClock doesn't have set_time_ns, so we can't easily test auto-flush
1308        // with time advancement. Instead, we test that check_flush is called during write.
1309        // For a proper test, we'd need a mock clock or use LiveClock with time advancement.
1310
1311        // Write second quote - check_flush will be called but won't flush if time hasn't advanced
1312        let quote2 = QuoteTick::new(
1313            InstrumentId::from("AUD/USD.SIM"),
1314            Price::from("1.1"),
1315            Price::from("1.1"),
1316            Quantity::from("1000"),
1317            Quantity::from("1000"),
1318            UnixNanos::from(2000),
1319            UnixNanos::from(2000),
1320        );
1321        writer.write(quote2).await.unwrap();
1322
1323        // Verify that writes succeeded (check_flush was called, even if it didn't flush)
1324        // The flush_interval_ms is set, so check_flush runs but won't flush without time advancement
1325    }
1326
1327    #[tokio::test]
1328    async fn test_close() {
1329        let temp_dir = TempDir::new().unwrap();
1330        let base_path = temp_dir.path().to_str().unwrap().to_string();
1331        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1332        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1333        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1334
1335        let mut writer = FeatherWriter::new(
1336            base_path,
1337            store,
1338            clock,
1339            RotationConfig::NoRotation,
1340            None,
1341            None,
1342            None,
1343        );
1344
1345        let quote = QuoteTick::new(
1346            InstrumentId::from("AUD/USD.SIM"),
1347            Price::from("1.0"),
1348            Price::from("1.0"),
1349            Quantity::from("1000"),
1350            Quantity::from("1000"),
1351            UnixNanos::from(1000),
1352            UnixNanos::from(1000),
1353        );
1354
1355        writer.write(quote).await.unwrap();
1356        assert!(!writer.writers.is_empty());
1357
1358        writer.close().await.unwrap();
1359        assert!(writer.writers.is_empty());
1360    }
1361
1362    // Note: Message bus subscription test is skipped due to async/sync boundary complexity.
1363    // The handler uses block_on which can't be used from within an async runtime.
1364    // This functionality is better tested via Python integration tests where the message bus
1365    // is used in a non-async context or via proper async task spawning.
1366
1367    #[tokio::test]
1368    async fn test_write_data_orderbook_deltas() {
1369        let temp_dir = TempDir::new().unwrap();
1370        let base_path = temp_dir.path().to_str().unwrap().to_string();
1371        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1372        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1373        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1374
1375        let mut writer = FeatherWriter::new(
1376            base_path,
1377            store,
1378            clock,
1379            RotationConfig::NoRotation,
1380            None,
1381            None,
1382            None,
1383        );
1384
1385        let instrument_id = InstrumentId::from("AUD/USD.SIM");
1386        let delta1 = OrderBookDelta::clear(
1387            instrument_id,
1388            0,
1389            UnixNanos::from(1000),
1390            UnixNanos::from(1000),
1391        );
1392        let delta2 = OrderBookDelta::clear(
1393            instrument_id,
1394            0,
1395            UnixNanos::from(2000),
1396            UnixNanos::from(2000),
1397        );
1398
1399        let book_deltas = OrderBookDeltas::new(instrument_id, vec![delta1, delta2]);
1400
1401        // Test writing OrderBookDeltas via write_data
1402        writer
1403            .write_data(Data::Deltas(Box::new(book_deltas)))
1404            .await
1405            .unwrap();
1406        writer.flush().await.unwrap();
1407    }
1408
1409    #[tokio::test]
1410    #[cfg(feature = "python")]
1411    async fn test_write_custom_data_round_trip() {
1412        use std::sync::Arc;
1413
1414        use futures::StreamExt;
1415        use nautilus_model::{
1416            data::{CustomData, Data, DataType},
1417            identifiers::InstrumentId,
1418        };
1419        use nautilus_serialization::{
1420            arrow::custom::CustomDataDecoder, ensure_custom_data_registered,
1421        };
1422
1423        use crate::test_data::RustTestCustomData;
1424
1425        ensure_custom_data_registered::<RustTestCustomData>();
1426
1427        let temp_dir = TempDir::new().unwrap();
1428        let base_path = temp_dir.path().to_str().unwrap().to_string();
1429        let local_fs = LocalFileSystem::new_with_prefix(temp_dir.path()).unwrap();
1430        let store: Arc<dyn ObjectStore> = Arc::new(local_fs);
1431        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
1432
1433        let mut writer = FeatherWriter::new(
1434            base_path.clone(),
1435            store.clone(),
1436            clock,
1437            RotationConfig::NoRotation,
1438            None,
1439            None,
1440            None,
1441        );
1442
1443        let instrument_id = InstrumentId::from("RUST.TEST");
1444        let data_type = DataType::new("RustTestCustomData", None, Some(instrument_id.to_string()));
1445        let original = RustTestCustomData {
1446            instrument_id,
1447            value: 1.23,
1448            flag: true,
1449            ts_event: UnixNanos::from(1000),
1450            ts_init: UnixNanos::from(1000),
1451        };
1452        let custom = CustomData::new(Arc::new(original.clone()), data_type);
1453
1454        writer
1455            .write_data(Data::Custom(custom))
1456            .await
1457            .expect("write_data CustomData");
1458        writer.flush().await.expect("flush");
1459
1460        let prefix = Path::from(format!("{base_path}/data/custom/RustTestCustomData"));
1461        let mut list_stream = store.list(Some(&prefix));
1462        let first = list_stream.next().await.expect("at least one object");
1463        let meta = first.expect("list item");
1464        let bytes = store
1465            .get(&meta.location)
1466            .await
1467            .expect("get")
1468            .bytes()
1469            .await
1470            .expect("bytes");
1471        let mut reader =
1472            StreamReader::try_new(Cursor::new(bytes.as_ref()), None).expect("StreamReader");
1473        let schema = reader.schema();
1474        let metadata: std::collections::HashMap<String, String> = schema
1475            .metadata()
1476            .iter()
1477            .map(|(k, v)| (k.clone(), v.clone()))
1478            .collect();
1479        let batch = reader.next().expect("batch").expect("batch ok");
1480        let decoded =
1481            CustomDataDecoder::decode_data_batch(&metadata, batch).expect("decode_data_batch");
1482        assert_eq!(decoded.len(), 1);
1483        if let Data::Custom(decoded_custom) = &decoded[0] {
1484            assert_eq!(decoded_custom.data_type.type_name(), "RustTestCustomData");
1485            let rust: &RustTestCustomData = decoded_custom
1486                .data
1487                .as_any()
1488                .downcast_ref::<RustTestCustomData>()
1489                .expect("RustTestCustomData");
1490            assert_eq!(rust, &original);
1491        } else {
1492            panic!("Expected Data::Custom");
1493        }
1494    }
1495}
nautilus_persistence/backend/feather.rs

nautilus_persistence/backend/
feather.rs
