4062405381

Build Type

github

Committed by GitHub

Commit Message

Merge c50dcefa1 into a81d03c98

Pull Request Pull Request #776: fix: `ApiEndpoint` is not serializing all fields

Run Details

5 of 5 new or added lines in 1 file covered. (100.0%)

24316 of 37369 relevant lines covered (65.07%)

35801.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.57

/dozer-core/src/dag/executor/source_node.rs

use std::{
    collections::HashMap,
    path::Path,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
    time::Duration,
};

use crossbeam::channel::{Receiver, RecvTimeoutError, Sender};
use dozer_types::log::debug;
use dozer_types::{
    internal_err,
    parking_lot::RwLock,
    types::{Operation, Schema},
};

use crate::dag::{
    channels::SourceChannelForwarder,
    dag::Edge,
    epoch::{EpochManager, OpIdentifier},
    errors::ExecutionError::{self, InternalError},
    executor_utils::{create_ports_databases_and_fill_downstream_record_readers, init_component},
    forwarder::{SourceChannelManager, StateWriter},
    node::{NodeHandle, OutputPortDef, PortHandle, Source, SourceFactory},
    record_store::RecordReader,
};

use super::{node::Node, ExecutorOperation};

#[derive(Debug)]
struct InternalChannelSourceForwarder {
    sender: Sender<(PortHandle, u64, u64, Operation)>,
}

impl InternalChannelSourceForwarder {
    pub fn new(sender: Sender<(PortHandle, u64, u64, Operation)>) -> Self {
        Self { sender }
    }
}

impl SourceChannelForwarder for InternalChannelSourceForwarder {
    fn send(
        &mut self,
        txid: u64,
        seq_in_tx: u64,
        op: Operation,
        port: PortHandle,
    ) -> Result<(), ExecutionError> {
        internal_err!(self.sender.send((port, txid, seq_in_tx, op)))
    }
}

/// The sender half of a source in the execution DAG.
#[derive(Debug)]
pub struct SourceSenderNode {
    /// Node handle in description DAG.
    node_handle: NodeHandle,
    /// The source.
    source: Box<dyn Source>,
    /// Last checkpointed output data sequence number.
    last_checkpoint: Option<OpIdentifier>,
    /// The forwarder that will be passed to the source for outputig data.
    forwarder: InternalChannelSourceForwarder,
    /// If the execution DAG should be running. Used for terminating the execution DAG.
    running: Arc<AtomicBool>,
}

impl SourceSenderNode {
    /// # Arguments
    ///
    /// - `node_handle`: Node handle in description DAG.
    /// - `source_factory`: Source factory in description DAG.
    /// - `output_schemas`: Output data schemas.
    /// - `last_checkpoint`: Last checkpointed output of this source.
    /// - `sender`: Channel to send data to.
    /// - `running`: If the execution DAG should still be running.
    pub fn new<T: Clone>(
        node_handle: NodeHandle,
        source_factory: &dyn SourceFactory<T>,
        output_schemas: HashMap<PortHandle, Schema>,
        last_checkpoint: Option<OpIdentifier>,
        sender: Sender<(PortHandle, u64, u64, Operation)>,
        running: Arc<AtomicBool>,
    ) -> Result<Self, ExecutionError> {
        let source = source_factory.build(output_schemas)?;
        let forwarder = InternalChannelSourceForwarder::new(sender);
        Ok(Self {
            node_handle,
            source,
            last_checkpoint,
            forwarder,
            running,
        })
    }
}

impl Node for SourceSenderNode {
    fn run(mut self) -> Result<(), ExecutionError> {
        let result = self.source.start(
            &mut self.forwarder,
            self.last_checkpoint
                .map(|op_id| (op_id.txid, op_id.seq_in_tx)),
        );
        self.running.store(false, Ordering::SeqCst);
        debug!("[{}-sender] Quit", self.node_handle);
        result
    }
}

/// The listener part of a source in the execution DAG.
#[derive(Debug)]
pub struct SourceListenerNode {
    /// Node handle in description DAG.
    node_handle: NodeHandle,
    /// Output from corresponding source sender.
    receiver: Receiver<(PortHandle, u64, u64, Operation)>,
    /// Receiving timeout.
    timeout: Duration,
    /// If the execution DAG should be running. Used for determining if a `terminate` message should be sent.
    running: Arc<AtomicBool>,
    /// This node's output channel manager, for communicating to other sources to coordinate terminate and commit, forwarding data, writing metadata and writing port state.
    channel_manager: SourceChannelManager,
}

impl SourceListenerNode {
    /// # Arguments
    ///
    /// - `node_handle`: Node handle in description DAG.
    /// - `receiver`: Channel that the data comes in.
    /// - `timeout`: `Listener timeout. After this timeout, listener will check if commit or terminate need to happen.
    /// - `base_path`: Base path of persisted data for the last execution of the description DAG.
    /// - `output_ports`: Output port definition of the source in description DAG.
    /// - `record_readers`: Record readers of all stateful ports.
    /// - `senders`: Output channels from this processor.
    /// - `edges`: All edges in the description DAG, used for creating record readers for input ports which is connected to this processor's stateful output ports.
    /// - `running`: If the execution DAG should still be running.
    /// - `epoch_manager`: Used for coordinating commit and terminate between sources. Shared by all sources.
    /// - `output_schemas`: Output data schemas.
    /// - `retention_queue_size`: Size of retention queue (used by RecordWriter)
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        node_handle: NodeHandle,
        receiver: Receiver<(PortHandle, u64, u64, Operation)>,
        timeout: Duration,
        base_path: &Path,
        output_ports: &[OutputPortDef],
        record_readers: Arc<
            RwLock<HashMap<NodeHandle, HashMap<PortHandle, Box<dyn RecordReader>>>>,
        >,
        senders: HashMap<PortHandle, Vec<Sender<ExecutorOperation>>>,
        edges: &[Edge],
        running: Arc<AtomicBool>,
        commit_sz: u32,
        max_duration_between_commits: Duration,
        epoch_manager: Arc<EpochManager>,
        output_schemas: HashMap<PortHandle, Schema>,
        retention_queue_size: usize,
    ) -> Result<Self, ExecutionError> {
        let state_meta = init_component(&node_handle, base_path, |_| Ok(()))?;
        let (master_tx, port_databases) =
            create_ports_databases_and_fill_downstream_record_readers(
                &node_handle,
                edges,
                state_meta.env,
                output_ports,
                &mut record_readers.write(),
            )?;
        let channel_manager = SourceChannelManager::new(
            node_handle.clone(),
            senders,
            StateWriter::new(
                state_meta.meta_db,
                port_databases,
                master_tx,
                output_schemas,
                retention_queue_size,
            )?,
            true,
            commit_sz,
            max_duration_between_commits,
            epoch_manager,
        );
        Ok(Self {
            node_handle,
            receiver,
            timeout,
            running,
            channel_manager,
        })
    }
}

impl SourceListenerNode {
    /// Returns if the node should terminate.
    fn send_and_trigger_commit_if_needed(
        &mut self,
        data: Option<(PortHandle, u64, u64, Operation)>,
    ) -> Result<bool, ExecutionError> {
        // First check if termination was requested.
        let terminating = !self.running.load(Ordering::SeqCst);
        // If this commit was not requested with termination at the start, we shouldn't terminate either.
        let terminating = match data {
            Some((port, txid, seq_in_tx, op)) => self
                .channel_manager
                .send_and_trigger_commit_if_needed(txid, seq_in_tx, op, port, terminating)?,
            None => self.channel_manager.trigger_commit_if_needed(terminating)?,
        };
        if terminating {
            self.channel_manager.terminate()?;
            debug!("[{}-listener] Quitting", &self.node_handle);
        }
        Ok(terminating)
    }
}

impl Node for SourceListenerNode {
    fn run(mut self) -> Result<(), ExecutionError> {
        loop {
            match self.receiver.recv_timeout(self.timeout) {
                Ok(data) => {
                    if self.send_and_trigger_commit_if_needed(Some(data))? {
                        return Ok(());
                    }
                }
                Err(e) => {
                    if self.send_and_trigger_commit_if_needed(None)? {
                        return Ok(());
                    }
                    // Channel disconnected but running flag not set to false, the source sender must have panicked.
                    if self.running.load(Ordering::SeqCst) && e == RecvTimeoutError::Disconnected {
                        return Err(ExecutionError::ChannelDisconnected);
                    }
                }
            }
        }
    }
}

1	use std::{
2	collections::HashMap,
3	path::Path,
4	sync::{
5	atomic::{AtomicBool, Ordering},
6	Arc,
7	},
8	time::Duration,
9	};
10
11	use crossbeam::channel::{Receiver, RecvTimeoutError, Sender};
12	use dozer_types::log::debug;
13	use dozer_types::{
14	internal_err,
15	parking_lot::RwLock,
16	types::{Operation, Schema},
17	};
18
19	use crate::dag::{
20	channels::SourceChannelForwarder,
21	dag::Edge,
22	epoch::{EpochManager, OpIdentifier},
23	errors::ExecutionError::{self, InternalError},
24	executor_utils::{create_ports_databases_and_fill_downstream_record_readers, init_component},
25	forwarder::{SourceChannelManager, StateWriter},
26	node::{NodeHandle, OutputPortDef, PortHandle, Source, SourceFactory},
27	record_store::RecordReader,
28	};
29
30	use super::{node::Node, ExecutorOperation};
31
32	#[derive(Debug)]	×
33	struct InternalChannelSourceForwarder {
34	sender: Sender<(PortHandle, u64, u64, Operation)>,
35	}
36
37	impl InternalChannelSourceForwarder {
38	pub fn new(sender: Sender<(PortHandle, u64, u64, Operation)>) -> Self {	180✔
39	Self { sender }	180✔
40	}	180✔
41	}
42
43	impl SourceChannelForwarder for InternalChannelSourceForwarder {
44	fn send(	3,688,756✔
45	&mut self,	3,688,756✔
46	txid: u64,	3,688,756✔
47	seq_in_tx: u64,	3,688,756✔
48	op: Operation,	3,688,756✔
49	port: PortHandle,	3,688,756✔
50	) -> Result<(), ExecutionError> {	3,688,756✔
51	internal_err!(self.sender.send((port, txid, seq_in_tx, op)))	10✔
52	}	3,688,756✔
53	}
54
55	/// The sender half of a source in the execution DAG.
56	#[derive(Debug)]	×
57	pub struct SourceSenderNode {
58	/// Node handle in description DAG.
59	node_handle: NodeHandle,
60	/// The source.
61	source: Box<dyn Source>,
62	/// Last checkpointed output data sequence number.
63	last_checkpoint: Option<OpIdentifier>,
64	/// The forwarder that will be passed to the source for outputig data.
65	forwarder: InternalChannelSourceForwarder,
66	/// If the execution DAG should be running. Used for terminating the execution DAG.
67	running: Arc<AtomicBool>,
68	}
69
70	impl SourceSenderNode {
71	/// # Arguments
72	///
73	/// - `node_handle`: Node handle in description DAG.
74	/// - `source_factory`: Source factory in description DAG.
75	/// - `output_schemas`: Output data schemas.
76	/// - `last_checkpoint`: Last checkpointed output of this source.
77	/// - `sender`: Channel to send data to.
78	/// - `running`: If the execution DAG should still be running.
79	pub fn new<T: Clone>(	71✔
80	node_handle: NodeHandle,	71✔
81	source_factory: &dyn SourceFactory<T>,	71✔
82	output_schemas: HashMap<PortHandle, Schema>,	71✔
83	last_checkpoint: Option<OpIdentifier>,	71✔
84	sender: Sender<(PortHandle, u64, u64, Operation)>,	71✔
85	running: Arc<AtomicBool>,	71✔
86	) -> Result<Self, ExecutionError> {	71✔
87	let source = source_factory.build(output_schemas)?;	71✔
88	let forwarder = InternalChannelSourceForwarder::new(sender);	70✔
89	Ok(Self {	70✔
90	node_handle,	70✔
91	source,	70✔
92	last_checkpoint,	70✔
93	forwarder,	70✔
94	running,	70✔
95	})	70✔
96	}	71✔
97	}
98
99	impl Node for SourceSenderNode {
100	fn run(mut self) -> Result<(), ExecutionError> {	180✔
101	let result = self.source.start(	180✔
102	&mut self.forwarder,	180✔
103	self.last_checkpoint	180✔
104	.map(\|op_id\| (op_id.txid, op_id.seq_in_tx)),	180✔
105	);	180✔
106	self.running.store(false, Ordering::SeqCst);	180✔
107	debug!("[{}-sender] Quit", self.node_handle);	180✔
108	result	180✔
109	}	180✔
110	}
111		×
112	/// The listener part of a source in the execution DAG.
113	#[derive(Debug)]	×
114	pub struct SourceListenerNode {
115	/// Node handle in description DAG.
116	node_handle: NodeHandle,
117	/// Output from corresponding source sender.
118	receiver: Receiver<(PortHandle, u64, u64, Operation)>,
119	/// Receiving timeout.
120	timeout: Duration,
121	/// If the execution DAG should be running. Used for determining if a `terminate` message should be sent.
122	running: Arc<AtomicBool>,
123	/// This node's output channel manager, for communicating to other sources to coordinate terminate and commit, forwarding data, writing metadata and writing port state.
124	channel_manager: SourceChannelManager,
125	}
126
127	impl SourceListenerNode {
128	/// # Arguments
129	///
130	/// - `node_handle`: Node handle in description DAG.
131	/// - `receiver`: Channel that the data comes in.
132	/// - `timeout`: `Listener timeout. After this timeout, listener will check if commit or terminate need to happen.
133	/// - `base_path`: Base path of persisted data for the last execution of the description DAG.
134	/// - `output_ports`: Output port definition of the source in description DAG.
135	/// - `record_readers`: Record readers of all stateful ports.
136	/// - `senders`: Output channels from this processor.
137	/// - `edges`: All edges in the description DAG, used for creating record readers for input ports which is connected to this processor's stateful output ports.
138	/// - `running`: If the execution DAG should still be running.
139	/// - `epoch_manager`: Used for coordinating commit and terminate between sources. Shared by all sources.
140	/// - `output_schemas`: Output data schemas.
141	/// - `retention_queue_size`: Size of retention queue (used by RecordWriter)	×
142	#[allow(clippy::too_many_arguments)]	×
143	pub(crate) fn new(	182✔
144	node_handle: NodeHandle,	182✔
145	receiver: Receiver<(PortHandle, u64, u64, Operation)>,	182✔
146	timeout: Duration,	182✔
147	base_path: &Path,	182✔
148	output_ports: &[OutputPortDef],	182✔
149	record_readers: Arc<	182✔
150	RwLock<HashMap<NodeHandle, HashMap<PortHandle, Box<dyn RecordReader>>>>,	182✔
151	>,	182✔
152	senders: HashMap<PortHandle, Vec<Sender<ExecutorOperation>>>,	182✔
153	edges: &[Edge],	182✔
154	running: Arc<AtomicBool>,	182✔
155	commit_sz: u32,	182✔
156	max_duration_between_commits: Duration,	182✔
157	epoch_manager: Arc<EpochManager>,	182✔
158	output_schemas: HashMap<PortHandle, Schema>,	182✔
159	retention_queue_size: usize,	182✔
160	) -> Result<Self, ExecutionError> {	182✔
161	let state_meta = init_component(&node_handle, base_path, \|_\| Ok(()))?;	182✔
162	let (master_tx, port_databases) =	181✔
163	create_ports_databases_and_fill_downstream_record_readers(	181✔
164	&node_handle,	181✔
165	edges,	181✔
166	state_meta.env,	181✔
167	output_ports,	181✔
168	&mut record_readers.write(),	181✔
169	)?;	181✔
170	let channel_manager = SourceChannelManager::new(	181✔
171	node_handle.clone(),	181✔
172	senders,	181✔
173	StateWriter::new(	181✔
174	state_meta.meta_db,	181✔
175	port_databases,	181✔
176	master_tx,	181✔
177	output_schemas,	181✔
178	retention_queue_size,	181✔
179	)?,	181✔
180	true,	×
181	commit_sz,	181✔
182	max_duration_between_commits,	181✔
183	epoch_manager,	181✔
184	);	181✔
185	Ok(Self {	181✔
186	node_handle,	181✔
187	receiver,	181✔
188	timeout,	181✔
189	running,	181✔
190	channel_manager,	181✔
191	})	181✔
192	}	182✔
193	}
194
195	impl SourceListenerNode {
196	/// Returns if the node should terminate.
197	fn send_and_trigger_commit_if_needed(	3,689,263✔
198	&mut self,	3,689,263✔
199	data: Option<(PortHandle, u64, u64, Operation)>,	3,689,263✔
200	) -> Result<bool, ExecutionError> {	3,689,263✔
201	// First check if termination was requested.	3,689,263✔
202	let terminating = !self.running.load(Ordering::SeqCst);	3,689,263✔
203	// If this commit was not requested with termination at the start, we shouldn't terminate either.
204	let terminating = match data {	3,689,263✔
205	Some((port, txid, seq_in_tx, op)) => self	3,688,347✔
206	.channel_manager	3,688,347✔
207	.send_and_trigger_commit_if_needed(txid, seq_in_tx, op, port, terminating)?,	3,688,347✔
208	None => self.channel_manager.trigger_commit_if_needed(terminating)?,	916✔
209	};
210	if terminating {	3,689,256✔
211	self.channel_manager.terminate()?;	179✔
212	debug!("[{}-listener] Quitting", &self.node_handle);	179✔
213	}	3,689,077✔
214	Ok(terminating)	3,689,249✔
215	}	3,689,256✔
216	}
217
218	impl Node for SourceListenerNode {
219	fn run(mut self) -> Result<(), ExecutionError> {	181✔
220	loop {	3,689,248✔
221	match self.receiver.recv_timeout(self.timeout) {	3,689,248✔
222	Ok(data) => {	3,688,449✔
223	if self.send_and_trigger_commit_if_needed(Some(data))? {	3,688,449✔
224	return Ok(());	6✔
225	}	3,688,436✔
226	}
227	Err(e) => {	799✔
228	if self.send_and_trigger_commit_if_needed(None)? {	799✔
229	return Ok(());	166✔
230	}	633✔
231	// Channel disconnected but running flag not set to false, the source sender must have panicked.	633✔
232	if self.running.load(Ordering::SeqCst) && e == RecvTimeoutError::Disconnected {	633✔
233	return Err(ExecutionError::ChannelDisconnected);	2✔
234	}	631✔
235	}
236	}
237	}
238	}	181✔
239	}

getdozer / dozer / 4062405381

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous