tenshift-core 0.1.2

Thread-safe, backpressure-aware data loading pipeline for iterative processing
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

//! Error types for tenshift.
//!
//! Every error is actionable and carries context. No generic "something went wrong."
//!
//! All error variants include a "Fix:" suggestion that provides immediate
//! actionable guidance for resolving the issue.

use std::path::PathBuf;

/// All errors that can occur in the pipeline.
///
/// Note: This enum is `#[non_exhaustive]` because the pipeline architecture
/// is designed to be extensible. We anticipate adding new variants in the future
/// as we introduce new hardware integrations and data processing stages.
#[non_exhaustive]
#[derive(Debug, Clone, thiserror::Error)]
pub enum Error {
    /// A source failed to initialize or advance safely.
    #[error("source '{source_name}' failed: {reason}. Fix: inspect the source implementation and input boundary for panics, stalls, or invalid reads.")]
    SourceFailed {
        /// The source name.
        source_name: String,
        /// What went wrong.
        reason: String,
    },

    /// A source file could not be read.
    #[error("failed to read {path}: {reason}. Fix: verify the path exists and is readable.")]
    ReadFailed {
        /// The path that failed.
        path: PathBuf,
        /// Why it failed.
        reason: String,
    },

    /// A source file was corrupt or unparseable.
    #[error("corrupt data in {path}: {reason}. Fix: repair or remove the malformed input.")]
    CorruptData {
        /// The path containing corrupt data.
        path: PathBuf,
        /// What went wrong.
        reason: String,
    },

    /// No files matched the source pattern.
    #[error("no files matched pattern: {pattern}. Fix: point the source at at least one existing input file.")]
    EmptySource {
        /// The glob pattern that matched nothing.
        pattern: String,
    },

    /// The pipeline was shut down (Ctrl+C or explicit stop).
    #[error("pipeline shut down")]
    Shutdown,

    /// A user-provided transform function failed.
    #[error("transform failed on item {index}: {reason}. Fix: inspect the transform input and handle that case explicitly.")]
    TransformFailed {
        /// Which item in the stream caused the failure.
        index: u64,
        /// What went wrong.
        reason: String,
    },

    /// Collation of a batch failed.
    #[error("collation failed: {reason}. Fix: ensure each sample in the batch has the same required fields, dtype, and shape.")]
    CollateFailed {
        /// What went wrong.
        reason: String,
    },

    /// An I/O error occurred.
    #[error("i/o error: {0}. Fix: resolve the underlying operating system error and retry.")]
    Io(std::sync::Arc<std::io::Error>),

    /// A glob pattern was invalid.
    #[error("invalid glob pattern: {0}. Fix: provide a valid glob expression.")]
    InvalidPattern(String),

    /// Pipeline configuration was invalid.
    #[error(
        "invalid config: {reason}. Fix: update the pipeline configuration to a supported value."
    )]
    InvalidConfig {
        /// What was wrong.
        reason: String,
    },
}

impl From<std::io::Error> for Error {
    fn from(e: std::io::Error) -> Self {
        Error::Io(std::sync::Arc::new(e))
    }
}

impl From<glob::PatternError> for Error {
    fn from(e: glob::PatternError) -> Self {
        Error::InvalidPattern(e.to_string())
    }
}

/// Convenience result type.
pub type Result<T> = std::result::Result<T, Error>;