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

nautilus_persistence/writer/
promotion.rs
