nautilus_persistence/writer/
promotion.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
16//! Shared background execution for Feather-to-catalog promotion.
17
18use std::{
19    any::Any,
20    panic::{self, AssertUnwindSafe},
21    path::PathBuf,
22    sync::{
23        Arc, Mutex,
24        atomic::{AtomicUsize, Ordering},
25        mpsc::{self, Receiver, SyncSender, TryRecvError},
26    },
27    thread::JoinHandle,
28    time::Duration,
29};
30
31use ahash::AHashSet;
32use nautilus_common::live::block_on_nautilus_with;
33use nautilus_core::UnixNanos;
34use nautilus_model::data::Data;
35
36use crate::{
37    common::{
38        conversion::FeatherConversionSummary, paths::normalize_path_separators,
39        storage::StorageBackend,
40    },
41    writer::{
42        run::{FeatherSessionSource, RunStatus},
43        traits::StreamingSink,
44    },
45};
46
47pub(crate) fn list_session_feather_files(
48    storage: &StorageBackend,
49    kind: &str,
50    instance_id: &str,
51) -> anyhow::Result<Vec<String>> {
52    let run_directory = format!("{kind}/{instance_id}");
53    let mut files =
54        block_on_nautilus_with(|| storage.list_files(&run_directory, Some(".feather")))?;
55    files.sort();
56    Ok(files)
57}
58
59#[derive(Clone, Debug, Eq, PartialEq)]
60pub(crate) struct PromotionSession {
61    pub(crate) catalog_uri: String,
62    pub(crate) kind: String,
63    pub(crate) instance_id: String,
64}
65
66impl PromotionSession {
67    pub(crate) fn from_uri(uri: &str) -> Option<Self> {
68        if let Ok(url) = url::Url::parse(uri)
69            && let Some(session) = Self::from_url(url)
70        {
71            return Some(session);
72        }
73        Self::from_local_path(uri)
74    }
75
76    fn from_url(mut url: url::Url) -> Option<Self> {
77        let components = url
78            .path()
79            .trim_matches('/')
80            .split('/')
81            .filter(|component| !component.is_empty())
82            .map(str::to_string)
83            .collect::<Vec<_>>();
84        let instance_id = components.last()?.clone();
85        let kind = components.get(components.len().checked_sub(2)?)?.clone();
86        if !is_run_kind(&kind) {
87            return None;
88        }
89
90        let catalog_segments = &components[..components.len().saturating_sub(2)];
91        let catalog_path = if catalog_segments.is_empty() {
92            "/".to_string()
93        } else {
94            format!("/{}", catalog_segments.join("/"))
95        };
96        url.set_path(&catalog_path);
97        url.set_query(None);
98        url.set_fragment(None);
99
100        Some(Self {
101            catalog_uri: url.to_string(),
102            kind,
103            instance_id,
104        })
105    }
106
107    fn from_local_path(path: &str) -> Option<Self> {
108        let normalized = normalize_path_separators(path);
109        let path = PathBuf::from(&normalized);
110        let instance_id = path.file_name()?.to_string_lossy().to_string();
111        let kind = path.parent()?.file_name()?.to_string_lossy().to_string();
112        if !is_run_kind(&kind) {
113            return None;
114        }
115        let catalog_uri = path.parent()?.parent()?.to_string_lossy().to_string();
116        Some(Self {
117            catalog_uri,
118            kind,
119            instance_id,
120        })
121    }
122}
123
124fn is_run_kind(kind: &str) -> bool {
125    matches!(kind, "backtest" | "live" | "sandbox")
126}
127
128pub(crate) trait PromotionSink: Send {
129    fn stage_data(&mut self, data: Data) -> anyhow::Result<()>;
130
131    fn stage_batch(&mut self, data: Vec<Data>) -> anyhow::Result<()>;
132
133    fn stage_any(&mut self, message: &dyn Any) -> anyhow::Result<bool>;
134
135    fn flush_staging(&mut self) -> anyhow::Result<()>;
136
137    fn close_staging(&mut self) -> anyhow::Result<()>;
138
139    fn mark_run_non_empty(&mut self) -> anyhow::Result<()>;
140
141    fn maybe_promote_by_period(&mut self) -> anyhow::Result<()>;
142
143    fn wait_for_background_promotions(&mut self) -> anyhow::Result<Vec<FeatherConversionSummary>>;
144
145    fn stop_periodic_promotion(&mut self);
146
147    fn take_pending_error(&mut self) -> anyhow::Result<()>;
148
149    fn should_promote_on_flush(&self) -> bool;
150
151    fn should_promote_on_close(&self) -> bool;
152
153    fn promote(&mut self) -> anyhow::Result<Vec<FeatherConversionSummary>>;
154
155    fn record_completed(&mut self);
156}
157
158impl<T> StreamingSink for T
159where
160    T: PromotionSink + std::fmt::Debug,
161{
162    fn write_data(&mut self, data: Data) -> anyhow::Result<()> {
163        self.stage_data(data)?;
164        self.mark_run_non_empty()?;
165        self.maybe_promote_by_period()
166    }
167
168    fn write_batch(&mut self, data: Vec<Data>) -> anyhow::Result<()> {
169        if data.is_empty() {
170            return Ok(());
171        }
172        self.stage_batch(data)?;
173        self.mark_run_non_empty()?;
174        self.maybe_promote_by_period()
175    }
176
177    fn write_any(&mut self, message: &dyn Any) -> anyhow::Result<bool> {
178        let handled = self.stage_any(message)?;
179
180        if handled {
181            self.mark_run_non_empty()?;
182            self.maybe_promote_by_period()?;
183        }
184        Ok(handled)
185    }
186
187    fn flush(&mut self) -> anyhow::Result<()> {
188        self.flush_staging()?;
189        let background_result = self.wait_for_background_promotions();
190
191        let promotion_result = if self.should_promote_on_flush() {
192            self.promote().map(|_| ())
193        } else {
194            Ok(())
195        };
196
197        let pending_result = self.take_pending_error();
198
199        background_result?;
200        promotion_result?;
201        pending_result
202    }
203
204    fn close(&mut self) -> anyhow::Result<()> {
205        self.stop_periodic_promotion();
206        self.close_staging()?;
207        let background_result = self.wait_for_background_promotions();
208        let pending_result = self.take_pending_error();
209
210        let promotion_result = if self.should_promote_on_close() {
211            self.promote().map(|_| ())
212        } else if background_result.is_ok() && pending_result.is_ok() {
213            self.record_completed();
214            Ok(())
215        } else {
216            Ok(())
217        };
218
219        background_result?;
220        promotion_result?;
221        pending_result
222    }
223}
224
225pub(crate) trait PromotionBackend: Send + 'static {
226    type Source: Send + 'static;
227
228    const NAME: &'static str;
229
230    fn convert_file(
231        &mut self,
232        source: &Self::Source,
233        file: &str,
234        use_ts_event_for_ts_init: bool,
235        record_promoted: bool,
236    ) -> anyhow::Result<Option<FeatherConversionSummary>>;
237
238    fn delete_file(&mut self, source: &Self::Source, file: &str) -> anyhow::Result<()>;
239}
240
241pub(crate) struct PromotionScope {
242    pub(crate) source: FeatherSessionSource,
243    pub(crate) staging_uri: String,
244    pub(crate) files: Vec<String>,
245    pub(crate) use_ts_event_for_ts_init: bool,
246    pub(crate) delete_feather_after_commit: bool,
247}
248
249pub(crate) trait StagedPromotionBackend:
250    PromotionBackend<Source = FeatherSessionSource> + Sized
251{
252    const REQUIRES_SESSION: bool = false;
253    const RECORDS_PROMOTED_ATOMICALLY: bool = false;
254
255    fn into_work(self, scope: PromotionScope) -> PromotionWork<Self> {
256        PromotionWork::new(
257            self,
258            scope.source,
259            scope.staging_uri,
260            scope.files,
261            scope.use_ts_event_for_ts_init,
262            scope.delete_feather_after_commit,
263        )
264    }
265
266    fn record_run_state(
267        &mut self,
268        _source: &FeatherSessionSource,
269        _staging_uri: &str,
270        _status: RunStatus,
271        _empty: bool,
272        _error: Option<&str>,
273    ) -> anyhow::Result<()> {
274        Ok(())
275    }
276
277    /// Deletes staged files whose promotion this backend has already durably recorded
278    /// with a deletion intent, returning the deleted paths.
279    ///
280    /// The hook owns both the predicate and the deletion, so a file whose retention
281    /// was intended is never reported and never removed. The work loop runs it before
282    /// converting and skips the returned paths.
283    fn delete_recorded_leftovers(
284        &mut self,
285        _source: &FeatherSessionSource,
286        _files: &[String],
287    ) -> anyhow::Result<Vec<String>> {
288        Ok(Vec::new())
289    }
290}
291
292pub(crate) struct PromotionWork<B>
293where
294    B: StagedPromotionBackend,
295{
296    backend: B,
297    source: B::Source,
298    staging_uri: String,
299    files: Vec<String>,
300    use_ts_event_for_ts_init: bool,
301    delete_feather_after_commit: bool,
302}
303
304impl<B> PromotionWork<B>
305where
306    B: StagedPromotionBackend,
307{
308    pub(crate) fn new(
309        backend: B,
310        source: B::Source,
311        staging_uri: String,
312        files: Vec<String>,
313        use_ts_event_for_ts_init: bool,
314        delete_feather_after_commit: bool,
315    ) -> Self {
316        Self {
317            backend,
318            source,
319            staging_uri,
320            files,
321            use_ts_event_for_ts_init,
322            delete_feather_after_commit,
323        }
324    }
325
326    pub(crate) fn execute(self) -> PromotionResult {
327        let files = self.files.clone();
328
329        match panic::catch_unwind(AssertUnwindSafe(|| self.execute_inner())) {
330            Ok(result) => result,
331            Err(payload) => PromotionResult {
332                files,
333                committed_paths: Vec::new(),
334                deleted_paths: Vec::new(),
335                partial_converted: Vec::new(),
336                run_state_recorded: false,
337                converted: Err(anyhow::anyhow!(
338                    "{} promotion worker panicked: {}",
339                    B::NAME,
340                    panic_payload_message(payload.as_ref()),
341                )),
342            },
343        }
344    }
345
346    pub(crate) fn files(&self) -> &[String] {
347        &self.files
348    }
349
350    #[expect(
351        clippy::too_many_lines,
352        reason = "the promotion loop keeps conversion, promoted-record fallback, and deletion accounting together"
353    )]
354    fn execute_inner(mut self) -> PromotionResult {
355        let mut converted = Vec::new();
356        let mut committed_paths = Vec::new();
357        let mut deleted_paths = Vec::new();
358        let mut errors = Vec::new();
359        let mut promoted_recorded = false;
360        let mut record_error = None;
361
362        // Recorded leftovers are promotions the backend already recorded durably with
363        // a deletion intent; a failed scan only logs so promotion still proceeds.
364        let leftovers = match self
365            .backend
366            .delete_recorded_leftovers(&self.source, &self.files)
367        {
368            Ok(leftovers) => leftovers,
369            Err(e) => {
370                log::warn!(
371                    "{} recorded-leftover cleanup failed; continuing with promotion: {e:#}",
372                    B::NAME,
373                );
374                Vec::new()
375            }
376        };
377        deleted_paths.extend(leftovers.iter().cloned());
378        let convert_files = self
379            .files
380            .iter()
381            .filter(|file| !leftovers.contains(file))
382            .cloned()
383            .collect::<Vec<_>>();
384
385        for (index, file) in convert_files.iter().enumerate() {
386            let record_promoted = index + 1 == convert_files.len() && errors.is_empty();
387
388            match self.backend.convert_file(
389                &self.source,
390                file,
391                self.use_ts_event_for_ts_init,
392                record_promoted,
393            ) {
394                Ok(Some(summary)) => {
395                    if B::RECORDS_PROMOTED_ATOMICALLY && record_promoted {
396                        if summary.native_version.is_some() {
397                            promoted_recorded = true;
398                        } else {
399                            // A no-op conversion (replay dedup, fully covered rows)
400                            // commits nothing to carry the promoted record, so write it
401                            // before deleting the staged file: a crash then leaves a
402                            // promoted run with leftover files, which recovery deletes,
403                            // never a promotion recorded nowhere.
404                            match self.backend.record_run_state(
405                                &self.source,
406                                &self.staging_uri,
407                                RunStatus::Promoted,
408                                false,
409                                None,
410                            ) {
411                                Ok(()) => promoted_recorded = true,
412                                Err(e) => record_error = Some(e),
413                            }
414                        }
415                    }
416                    converted.push(summary);
417                    committed_paths.push(file.clone());
418                    if record_error.is_some() {
419                        // Keep the staged file so the retried promotion can re-run the
420                        // record write; the replay dedup makes the re-run a no-op.
421                    } else if self.delete_feather_after_commit
422                        && let Err(e) = self.backend.delete_file(&self.source, file)
423                    {
424                        errors.push(format!(
425                            "{:#}",
426                            e.context(format!(
427                                "{} source deletion failed for Feather file '{file}'",
428                                B::NAME,
429                            )),
430                        ));
431                    } else if self.delete_feather_after_commit {
432                        deleted_paths.push(file.clone());
433                    }
434                }
435                Ok(None) => {}
436                Err(e) => {
437                    let e = e.context(format!(
438                        "{} promotion failed for Feather file '{file}'",
439                        B::NAME,
440                    ));
441                    errors.push(format!("{e:#}"));
442                }
443            }
444        }
445
446        if let Some(e) = record_error {
447            return PromotionResult {
448                files: self.files,
449                committed_paths,
450                deleted_paths,
451                partial_converted: converted,
452                run_state_recorded: false,
453                converted: Err(
454                    e.context(format!("{} failed to record promoted run state", B::NAME))
455                ),
456            };
457        }
458
459        if !errors.is_empty() {
460            let error = errors.join("; ");
461            let run_state_recorded = match self.backend.record_run_state(
462                &self.source,
463                &self.staging_uri,
464                RunStatus::Failed,
465                false,
466                Some(&error),
467            ) {
468                Ok(()) => true,
469                Err(state_error) => {
470                    errors.push(format!(
471                        "{} failed to record failed run state: {state_error}",
472                        B::NAME,
473                    ));
474                    false
475                }
476            };
477            return PromotionResult {
478                files: self.files,
479                committed_paths,
480                deleted_paths,
481                partial_converted: converted,
482                run_state_recorded,
483                converted: Err(anyhow::anyhow!(errors.join("; "))),
484            };
485        }
486
487        if !converted.is_empty()
488            && !B::RECORDS_PROMOTED_ATOMICALLY
489            && let Err(e) = self.backend.record_run_state(
490                &self.source,
491                &self.staging_uri,
492                RunStatus::Promoted,
493                false,
494                None,
495            )
496        {
497            return PromotionResult {
498                files: self.files,
499                committed_paths,
500                deleted_paths,
501                partial_converted: converted,
502                run_state_recorded: false,
503                converted: Err(
504                    e.context(format!("{} failed to record promoted run state", B::NAME))
505                ),
506            };
507        }
508        let run_state_recorded = if B::RECORDS_PROMOTED_ATOMICALLY {
509            promoted_recorded
510        } else {
511            !converted.is_empty()
512        };
513        PromotionResult {
514            files: self.files,
515            committed_paths,
516            deleted_paths,
517            partial_converted: Vec::new(),
518            run_state_recorded,
519            converted: Ok(converted),
520        }
521    }
522}
523
524/// Opaque outcome returned by a prepared catalog promotion.
525///
526/// Callers pass this value back to the writer that prepared the promotion so it
527/// can update scheduling and run state.
528pub struct PromotionResult {
529    pub(crate) files: Vec<String>,
530    pub(crate) committed_paths: Vec<String>,
531    pub(crate) deleted_paths: Vec<String>,
532    pub(crate) partial_converted: Vec<FeatherConversionSummary>,
533    pub(crate) run_state_recorded: bool,
534    pub(crate) converted: anyhow::Result<Vec<FeatherConversionSummary>>,
535}
536
537fn panic_payload_message(payload: &(dyn std::any::Any + Send)) -> String {
538    if let Some(message) = payload.downcast_ref::<&'static str>() {
539        return (*message).to_string();
540    }
541
542    if let Some(message) = payload.downcast_ref::<String>() {
543        return message.clone();
544    }
545    "unknown panic payload".to_string()
546}
547
548pub(crate) trait PromotionTask: Send + 'static {
549    type Output: Send + 'static;
550
551    fn execute(self) -> Self::Output;
552}
553
554pub(crate) struct PromotionTimer {
555    stop_tx: SyncSender<()>,
556    handle: Option<JoinHandle<()>>,
557}
558
559impl PromotionTimer {
560    pub(crate) fn spawn<F>(
561        thread_name: &str,
562        interval_ms: u64,
563        mut callback: F,
564    ) -> anyhow::Result<Self>
565    where
566        F: FnMut() + Send + 'static,
567    {
568        anyhow::ensure!(interval_ms > 0, "Promotion timer interval must be positive");
569        let (stop_tx, stop_rx) = mpsc::sync_channel(1);
570        let interval = Duration::from_millis(interval_ms);
571        let handle = std::thread::Builder::new() // dst-ok: timer performs blocking catalog I/O
572            .name(thread_name.to_string())
573            .spawn(move || {
574                loop {
575                    match stop_rx.recv_timeout(interval) {
576                        Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => break,
577                        Err(mpsc::RecvTimeoutError::Timeout) => callback(),
578                    }
579                }
580            })?;
581        Ok(Self {
582            stop_tx,
583            handle: Some(handle),
584        })
585    }
586
587    pub(crate) fn stop(&mut self) {
588        let _ = self.stop_tx.try_send(());
589
590        if let Some(handle) = self.handle.take()
591            && let Err(e) = handle.join()
592        {
593            log::warn!("Promotion timer panicked: {e:?}");
594        }
595    }
596}
597
598impl Drop for PromotionTimer {
599    fn drop(&mut self) {
600        self.stop();
601    }
602}
603
604enum PromotionMessage<T> {
605    Work(Box<T>),
606    Close,
607}
608
609pub(crate) struct PromotionSubmitter<T>
610where
611    T: PromotionTask,
612{
613    tx: SyncSender<PromotionMessage<T>>,
614    pending: Arc<AtomicUsize>,
615}
616
617impl<T> Clone for PromotionSubmitter<T>
618where
619    T: PromotionTask,
620{
621    fn clone(&self) -> Self {
622        Self {
623            tx: self.tx.clone(),
624            pending: Arc::clone(&self.pending),
625        }
626    }
627}
628
629impl<T> PromotionSubmitter<T>
630where
631    T: PromotionTask,
632{
633    pub(crate) fn submit(&self, work: T) -> anyhow::Result<()> {
634        self.pending.fetch_add(1, Ordering::AcqRel);
635        if let Err(e) = self.tx.send(PromotionMessage::Work(Box::new(work))) {
636            self.pending.fetch_sub(1, Ordering::AcqRel);
637            anyhow::bail!("Promotion worker disconnected: {e}");
638        }
639        Ok(())
640    }
641}
642
643pub(crate) struct PromotionWorker<T>
644where
645    T: PromotionTask,
646{
647    tx: SyncSender<PromotionMessage<T>>,
648    rx: Receiver<T::Output>,
649    handle: Option<JoinHandle<()>>,
650    pending: Arc<AtomicUsize>,
651}
652
653impl<T> PromotionWorker<T>
654where
655    T: PromotionTask,
656{
657    pub(crate) fn spawn(thread_name: &str) -> anyhow::Result<Self> {
658        let (tx, work_rx) = mpsc::sync_channel::<PromotionMessage<T>>(1);
659        let (result_tx, rx) = mpsc::channel::<T::Output>();
660        let handle = std::thread::Builder::new() // dst-ok: catalog promotion performs blocking I/O
661            .name(thread_name.to_string())
662            .spawn(move || {
663                while let Ok(message) = work_rx.recv() {
664                    match message {
665                        PromotionMessage::Work(work) => {
666                            let _ = result_tx.send((*work).execute());
667                        }
668                        PromotionMessage::Close => break,
669                    }
670                }
671            })?;
672
673        Ok(Self {
674            tx,
675            rx,
676            handle: Some(handle),
677            pending: Arc::new(AtomicUsize::new(0)),
678        })
679    }
680
681    pub(crate) fn submit(&self, work: T) -> anyhow::Result<()> {
682        self.submitter().submit(work)
683    }
684
685    pub(crate) fn submitter(&self) -> PromotionSubmitter<T> {
686        PromotionSubmitter {
687            tx: self.tx.clone(),
688            pending: Arc::clone(&self.pending),
689        }
690    }
691
692    pub(crate) fn has_pending_work(&self) -> bool {
693        self.pending.load(Ordering::Acquire) > 0
694    }
695
696    pub(crate) fn try_recv_result(&self) -> anyhow::Result<Option<T::Output>> {
697        match self.rx.try_recv() {
698            Ok(result) => {
699                self.pending.fetch_sub(1, Ordering::AcqRel);
700                Ok(Some(result))
701            }
702            Err(TryRecvError::Empty) => Ok(None),
703            Err(TryRecvError::Disconnected) => anyhow::bail!("Promotion worker disconnected"),
704        }
705    }
706
707    pub(crate) fn recv_result(&self) -> anyhow::Result<T::Output> {
708        let result = self
709            .rx
710            .recv()
711            .map_err(|e| anyhow::anyhow!("Promotion worker disconnected: {e}"))?;
712        self.pending.fetch_sub(1, Ordering::AcqRel);
713        Ok(result)
714    }
715}
716
717impl<T> Drop for PromotionWorker<T>
718where
719    T: PromotionTask,
720{
721    fn drop(&mut self) {
722        while self.pending.load(Ordering::Acquire) > 0 {
723            if self.recv_result().is_err() {
724                break;
725            }
726        }
727
728        let _ = self.tx.send(PromotionMessage::Close);
729
730        if let Some(handle) = self.handle.take()
731            && let Err(e) = handle.join()
732        {
733            log::warn!("Promotion worker panicked: {e:?}");
734        }
735    }
736}
737
738pub(crate) struct PromotionDriver<B>
739where
740    B: StagedPromotionBackend,
741{
742    worker: Option<PromotionWorker<PromotionWork<B>>>,
743    scheduled_paths: Arc<Mutex<AHashSet<String>>>,
744    last_commit_time_ns: UnixNanos,
745}
746
747impl<B> PromotionDriver<B>
748where
749    B: StagedPromotionBackend,
750{
751    pub(crate) fn new(last_commit_time_ns: UnixNanos) -> Self {
752        Self {
753            worker: None,
754            scheduled_paths: Arc::new(Mutex::new(AHashSet::new())),
755            last_commit_time_ns,
756        }
757    }
758
759    pub(crate) fn is_due(&self, now: UnixNanos, interval_ns: u64) -> bool {
760        now.as_u64()
761            .saturating_sub(self.last_commit_time_ns.as_u64())
762            >= interval_ns
763    }
764
765    pub(crate) fn mark_committed_at(&mut self, now: UnixNanos) {
766        self.last_commit_time_ns = now;
767    }
768
769    pub(crate) fn schedule_new(&self, files: Vec<String>) -> anyhow::Result<Vec<String>> {
770        schedule_new_paths(&self.scheduled_paths, files)
771    }
772
773    pub(crate) fn submit(
774        &mut self,
775        work: PromotionWork<B>,
776        worker_name: &str,
777    ) -> anyhow::Result<()> {
778        let files = work.files().to_vec();
779
780        if self.worker.is_none() {
781            self.worker = Some(PromotionWorker::spawn(worker_name)?);
782        }
783
784        if let Err(e) = self.worker.as_mut().unwrap().submit(work) {
785            self.unschedule(&files);
786            return Err(e);
787        }
788        Ok(())
789    }
790
791    pub(crate) fn submitter(
792        &mut self,
793        worker_name: &str,
794    ) -> anyhow::Result<PromotionSubmitter<PromotionWork<B>>> {
795        if self.worker.is_none() {
796            self.worker = Some(PromotionWorker::spawn(worker_name)?);
797        }
798        Ok(self.worker.as_ref().unwrap().submitter())
799    }
800
801    pub(crate) fn scheduled_paths(&self) -> Arc<Mutex<AHashSet<String>>> {
802        Arc::clone(&self.scheduled_paths)
803    }
804
805    pub(crate) fn drain_completed(&mut self) -> anyhow::Result<Vec<PromotionResult>> {
806        let mut results = Vec::new();
807
808        loop {
809            let Some(worker) = self.worker.as_mut() else {
810                return Ok(results);
811            };
812
813            match worker.try_recv_result()? {
814                Some(result) => {
815                    results.push(result);
816                }
817                None => return Ok(results),
818            }
819        }
820    }
821
822    pub(crate) fn wait(&mut self) -> anyhow::Result<Vec<PromotionResult>> {
823        let mut results = self.drain_completed()?;
824
825        loop {
826            let Some(worker) = self.worker.as_mut() else {
827                return Ok(results);
828            };
829
830            if !worker.has_pending_work() {
831                return Ok(results);
832            }
833
834            results.push(worker.recv_result()?);
835        }
836    }
837
838    pub(crate) fn unschedule(&self, files: &[String]) {
839        let mut scheduled_paths = self
840            .scheduled_paths
841            .lock()
842            .expect("promotion schedule lock poisoned");
843
844        for file in files {
845            scheduled_paths.remove(file);
846        }
847    }
848
849    pub(crate) fn unschedule_uncommitted(&self, files: &[String], committed_paths: &[String]) {
850        let committed_paths = committed_paths
851            .iter()
852            .map(String::as_str)
853            .collect::<AHashSet<_>>();
854
855        let mut scheduled_paths = self
856            .scheduled_paths
857            .lock()
858            .expect("promotion schedule lock poisoned");
859
860        for file in files {
861            if !committed_paths.contains(file.as_str()) {
862                scheduled_paths.remove(file);
863            }
864        }
865    }
866}
867
868pub(crate) fn schedule_new_paths(
869    scheduled_paths: &Arc<Mutex<AHashSet<String>>>,
870    files: Vec<String>,
871) -> anyhow::Result<Vec<String>> {
872    let mut scheduled = scheduled_paths
873        .lock()
874        .map_err(|e| anyhow::anyhow!("Promotion schedule lock poisoned: {e}"))?;
875    Ok(files
876        .into_iter()
877        .filter(|file| scheduled.insert(file.clone()))
878        .collect())
879}
880
881impl<B> PromotionTask for PromotionWork<B>
882where
883    B: StagedPromotionBackend,
884{
885    type Output = PromotionResult;
886
887    fn execute(self) -> Self::Output {
888        Self::execute(self)
889    }
890}
891
892#[cfg(test)]
893mod tests {
894    use std::{
895        any::Any,
896        sync::{Arc, Mutex},
897    };
898
899    use nautilus_model::data::Data;
900    use rstest::rstest;
901
902    use super::{
903        PromotionBackend, PromotionSession, PromotionSink, PromotionWork, StagedPromotionBackend,
904    };
905    use crate::{
906        common::{conversion::FeatherConversionSummary, storage::create_storage_backend_from_path},
907        writer::{
908            run::{FeatherSessionSource, RunStatus},
909            traits::StreamingSink,
910        },
911    };
912
913    #[derive(Debug, Default)]
914    struct RecordingState {
915        converted: Vec<(String, bool)>,
916        deleted: Vec<String>,
917        recorded_states: Vec<RunStatus>,
918    }
919
920    #[derive(Debug)]
921    struct RecordingBackend {
922        leftovers: Vec<String>,
923        state: Arc<Mutex<RecordingState>>,
924    }
925
926    impl PromotionBackend for RecordingBackend {
927        type Source = FeatherSessionSource;
928
929        const NAME: &'static str = "Recording";
930
931        fn convert_file(
932            &mut self,
933            _source: &FeatherSessionSource,
934            file: &str,
935            _use_ts_event_for_ts_init: bool,
936            record_promoted: bool,
937        ) -> anyhow::Result<Option<FeatherConversionSummary>> {
938            self.state
939                .lock()
940                .unwrap()
941                .converted
942                .push((file.to_string(), record_promoted));
943            Ok(Some(FeatherConversionSummary {
944                type_name: "quotes".to_string(),
945                identifier: None,
946                feather_path: file.to_string(),
947                native_version: Some(1),
948                unmatched_identifiers: None,
949            }))
950        }
951
952        fn delete_file(
953            &mut self,
954            _source: &FeatherSessionSource,
955            file: &str,
956        ) -> anyhow::Result<()> {
957            self.state.lock().unwrap().deleted.push(file.to_string());
958            Ok(())
959        }
960    }
961
962    impl StagedPromotionBackend for RecordingBackend {
963        fn record_run_state(
964            &mut self,
965            _source: &FeatherSessionSource,
966            _staging_uri: &str,
967            status: RunStatus,
968            _empty: bool,
969            _error: Option<&str>,
970        ) -> anyhow::Result<()> {
971            self.state.lock().unwrap().recorded_states.push(status);
972            Ok(())
973        }
974
975        fn delete_recorded_leftovers(
976            &mut self,
977            _source: &FeatherSessionSource,
978            files: &[String],
979        ) -> anyhow::Result<Vec<String>> {
980            Ok(files
981                .iter()
982                .filter(|file| self.leftovers.contains(file))
983                .cloned()
984                .collect())
985        }
986    }
987
988    #[rstest]
989    fn recorded_leftovers_are_deleted_and_skipped_before_conversion() {
990        let temp = tempfile::TempDir::new().unwrap();
991        let storage =
992            create_storage_backend_from_path(temp.path().to_str().unwrap(), None).unwrap();
993        let source = FeatherSessionSource::new(storage, "backtest", "run-leftovers");
994        let state = Arc::new(Mutex::new(RecordingState::default()));
995        let backend = RecordingBackend {
996            leftovers: vec!["backtest/run-leftovers/b.feather".to_string()],
997            state: Arc::clone(&state),
998        };
999        let files = vec![
1000            "backtest/run-leftovers/a.feather".to_string(),
1001            "backtest/run-leftovers/b.feather".to_string(),
1002            "backtest/run-leftovers/c.feather".to_string(),
1003        ];
1004
1005        let result = PromotionWork::new(backend, source, "staging".to_string(), files, false, true)
1006            .execute();
1007
1008        let converted = result.converted.unwrap();
1009        let state = state.lock().unwrap();
1010        assert_eq!(
1011            converted
1012                .iter()
1013                .map(|summary| summary.feather_path.as_str())
1014                .collect::<Vec<_>>(),
1015            vec![
1016                "backtest/run-leftovers/a.feather",
1017                "backtest/run-leftovers/c.feather",
1018            ],
1019        );
1020        assert_eq!(
1021            result.committed_paths,
1022            vec![
1023                "backtest/run-leftovers/a.feather",
1024                "backtest/run-leftovers/c.feather",
1025            ],
1026        );
1027        // The leftover is deleted without conversion; the converted files are deleted
1028        // by the configured post-commit deletion.
1029        assert_eq!(
1030            result.deleted_paths,
1031            vec![
1032                "backtest/run-leftovers/b.feather",
1033                "backtest/run-leftovers/a.feather",
1034                "backtest/run-leftovers/c.feather",
1035            ],
1036        );
1037        assert_eq!(
1038            state.converted,
1039            vec![
1040                ("backtest/run-leftovers/a.feather".to_string(), false),
1041                ("backtest/run-leftovers/c.feather".to_string(), true),
1042            ],
1043        );
1044        assert_eq!(
1045            state.deleted,
1046            vec![
1047                "backtest/run-leftovers/a.feather",
1048                "backtest/run-leftovers/c.feather",
1049            ],
1050        );
1051        assert_eq!(state.recorded_states, vec![RunStatus::Promoted]);
1052        assert!(result.run_state_recorded);
1053    }
1054
1055    #[derive(Debug)]
1056    struct FailingBackgroundSink {
1057        calls: Vec<&'static str>,
1058        promote_on_flush: bool,
1059        promote_on_close: bool,
1060    }
1061
1062    impl FailingBackgroundSink {
1063        const fn new(promote_on_flush: bool, promote_on_close: bool) -> Self {
1064            Self {
1065                calls: Vec::new(),
1066                promote_on_flush,
1067                promote_on_close,
1068            }
1069        }
1070    }
1071
1072    impl PromotionSink for FailingBackgroundSink {
1073        fn stage_data(&mut self, _data: Data) -> anyhow::Result<()> {
1074            unreachable!()
1075        }
1076
1077        fn stage_batch(&mut self, _data: Vec<Data>) -> anyhow::Result<()> {
1078            unreachable!()
1079        }
1080
1081        fn stage_any(&mut self, _message: &dyn Any) -> anyhow::Result<bool> {
1082            unreachable!()
1083        }
1084
1085        fn flush_staging(&mut self) -> anyhow::Result<()> {
1086            self.calls.push("flush_staging");
1087            Ok(())
1088        }
1089
1090        fn close_staging(&mut self) -> anyhow::Result<()> {
1091            self.calls.push("close_staging");
1092            Ok(())
1093        }
1094
1095        fn mark_run_non_empty(&mut self) -> anyhow::Result<()> {
1096            unreachable!()
1097        }
1098
1099        fn maybe_promote_by_period(&mut self) -> anyhow::Result<()> {
1100            unreachable!()
1101        }
1102
1103        fn wait_for_background_promotions(
1104            &mut self,
1105        ) -> anyhow::Result<Vec<FeatherConversionSummary>> {
1106            self.calls.push("wait_for_background_promotions");
1107            anyhow::bail!("background promotion failed")
1108        }
1109
1110        fn stop_periodic_promotion(&mut self) {
1111            self.calls.push("stop_periodic_promotion");
1112        }
1113
1114        fn take_pending_error(&mut self) -> anyhow::Result<()> {
1115            self.calls.push("take_pending_error");
1116            Ok(())
1117        }
1118
1119        fn should_promote_on_flush(&self) -> bool {
1120            self.promote_on_flush
1121        }
1122
1123        fn should_promote_on_close(&self) -> bool {
1124            self.promote_on_close
1125        }
1126
1127        fn promote(&mut self) -> anyhow::Result<Vec<FeatherConversionSummary>> {
1128            self.calls.push("promote");
1129            Ok(Vec::new())
1130        }
1131
1132        fn record_completed(&mut self) {
1133            self.calls.push("record_completed");
1134        }
1135    }
1136
1137    #[rstest]
1138    fn promotion_session_parses_windows_drive_path() {
1139        let session =
1140            PromotionSession::from_uri(r"C:\catalog\backtest\run-1").expect("valid run path");
1141
1142        assert_eq!(session.catalog_uri, "C:/catalog");
1143        assert_eq!(session.kind, "backtest");
1144        assert_eq!(session.instance_id, "run-1");
1145    }
1146
1147    #[rstest]
1148    fn promotion_session_parses_windows_unc_path() {
1149        let session =
1150            PromotionSession::from_uri(r"\\server\share\live\run-2").expect("valid UNC run path");
1151
1152        // catalog_uri is platform-dependent here (Windows retains a trailing
1153        // separator at the UNC prefix+root floor), so only kind/instance_id are asserted.
1154        assert_eq!(session.kind, "live");
1155        assert_eq!(session.instance_id, "run-2");
1156    }
1157
1158    #[rstest]
1159    fn promotion_session_parses_posix_path() {
1160        let session =
1161            PromotionSession::from_uri("/tmp/catalog/sandbox/run-3").expect("valid run path");
1162
1163        assert_eq!(session.catalog_uri, "/tmp/catalog");
1164        assert_eq!(session.kind, "sandbox");
1165        assert_eq!(session.instance_id, "run-3");
1166    }
1167
1168    #[rstest]
1169    fn flush_finalizes_and_promotes_before_returning_background_error() {
1170        let mut sink = FailingBackgroundSink::new(true, false);
1171
1172        let e = StreamingSink::flush(&mut sink).unwrap_err();
1173
1174        assert_eq!(e.to_string(), "background promotion failed");
1175        assert_eq!(
1176            sink.calls,
1177            [
1178                "flush_staging",
1179                "wait_for_background_promotions",
1180                "promote",
1181                "take_pending_error",
1182            ]
1183        );
1184    }
1185
1186    #[rstest]
1187    fn close_finalizes_and_promotes_before_returning_background_error() {
1188        let mut sink = FailingBackgroundSink::new(false, true);
1189
1190        let e = StreamingSink::close(&mut sink).unwrap_err();
1191
1192        assert_eq!(e.to_string(), "background promotion failed");
1193        assert_eq!(
1194            sink.calls,
1195            [
1196                "stop_periodic_promotion",
1197                "close_staging",
1198                "wait_for_background_promotions",
1199                "take_pending_error",
1200                "promote",
1201            ]
1202        );
1203    }
1204}
nautilus_persistence/writer/promotion.rs

nautilus_persistence/writer/
promotion.rs
