fix: Move cursor only after side-effect

src/model.rs

@@ -351,7 +351,7 @@ pub enum EventData {
     },
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct Event {
     pub context: EventContext,
 

src/sinks/assert/run.rs

@@ -70,9 +70,6 @@ pub fn assertion_loop(
     let mut state = State::default();
 
     for event in input.iter() {
-        // notify pipeline about the progress
-        utils.track_sink_progress(&event);
-
         log::info!("starting assertions for event: {:?}", event.fingerprint);
 
         state = reduce_state(state, event);
@@ -83,6 +80,11 @@ pub fn assertion_loop(
         run_check!(config, &state, event_timestamp_increases);
         run_check!(config, &state, tx_records_matches_block_count);
         run_check!(config, &state, tx_has_input_and_output);
+
+        if let Some(event) = &state.current_event {
+            // notify pipeline about the progress
+            utils.track_sink_progress(event);
+        }
     }
 
     Ok(())

src/sinks/aws_lambda/run.rs

@@ -37,16 +37,19 @@ pub fn writer_loop(
         .build()?;
 
     for event in input.iter() {
-        // notify the pipeline where we are
-        utils.track_sink_progress(&event);
-
         let client = client.clone();
 
         let result = rt.block_on(invoke_lambda_function(client, function_name, &event));
 
-        if let Err(err) = result {
-            log::error!("unrecoverable error invoking lambda function: {:?}", err);
-            return Err(err);
+        match result {
+            Ok(_) => {
+                // notify the pipeline where we are
+                utils.track_sink_progress(&event);
+            }
+            Err(err) => {
+                log::error!("unrecoverable error invoking lambda function: {:?}", err);
+                return Err(err);
+            }
         }
     }
 

src/sinks/aws_s3/run.rs

@@ -106,12 +106,9 @@ pub fn writer_loop(
         .build()?;
 
     for event in input.iter() {
-        // notify the pipeline where we are
-        utils.track_sink_progress(&event);
-
-        if let EventData::Block(record) = event.data {
-            let key = define_obj_key(prefix, &naming, &record);
-            let content = define_content(&content_type, &record);
+        if let EventData::Block(record) = &event.data {
+            let key = define_obj_key(prefix, &naming, record);
+            let content = define_content(&content_type, record);
 
             let client = client.clone();
 
@@ -121,12 +118,18 @@ pub fn writer_loop(
                 &key,
                 content,
                 &content_type,
-                &record,
+                record,
             ));
 
-            if let Err(err) = result {
-                log::error!("unrecoverable error sending block to S3: {:?}", err);
-                return Err(err);
+            match result {
+                Ok(_) => {
+                    // notify the pipeline where we are
+                    utils.track_sink_progress(&event);
+                }
+                Err(err) => {
+                    log::error!("unrecoverable error sending block to S3: {:?}", err);
+                    return Err(err);
+                }
             }
         }
     }

src/sinks/aws_sqs/run.rs

@@ -49,16 +49,19 @@ pub fn writer_loop(
         .build()?;
 
     for event in input.iter() {
-        // notify the pipeline where we are
-        utils.track_sink_progress(&event);
-
         let client = client.clone();
 
         let result = rt.block_on(send_sqs_msg(client, queue_url, group_id, fifo, &event));
 
-        if let Err(err) = result {
-            log::error!("unrecoverable error sending message to SQS: {:?}", err);
-            return Err(err);
+        match result {
+            Ok(_) => {
+                // notify the pipeline where we are
+                utils.track_sink_progress(&event);
+            }
+            Err(err) => {
+                log::error!("unrecoverable error sending message to SQS: {:?}", err);
+                return Err(err);
+            }
         }
     }
 

src/sinks/common/web.rs

@@ -88,18 +88,18 @@ pub(crate) fn request_loop(
     utils: Arc<Utils>,
 ) -> Result<(), Error> {
     for event in input.iter() {
-        // notify progress to the pipeline
-        utils.track_sink_progress(&event);
-
-        let body = RequestBody::from(event);
+        let body = RequestBody::from(event.clone());
 
         let result = retry::retry_operation(
             || execute_fallible_request(client, url, &body),
             retry_policy,
         );
 
         match result {
-            Ok(()) => (),
+            Ok(_) => {
+                // notify progress to the pipeline
+                utils.track_sink_progress(&event);
+            }
             Err(err) => match error_policy {
                 ErrorPolicy::Exit => return Err(err),
                 ErrorPolicy::Continue => {

src/sinks/elastic/run.rs

@@ -101,23 +101,32 @@ pub fn writer_loop(
         .build()?;
 
     for event in input.iter() {
-        // notify the pipeline where we are
-        utils.track_sink_progress(&event);
-
         let index = index.to_owned();
         let client = client.clone();
 
+        let event2 = event.clone();
         let result = rt.block_on(async move {
             match idempotency {
-                true => index_event_with_id(client, &index, event).await,
-                false => index_event_without_id(client, &index, event).await,
+                true => index_event_with_id(client, &index, event2).await,
+                false => index_event_without_id(client, &index, event2).await,
             }
         });
 
         if let Err(err) = result {
             log::error!("error indexing record in Elasticsearch: {}", err);
             return Err(err);
         }
+
+        match result {
+            Ok(_) => {
+                // notify progress to the pipeline
+                utils.track_sink_progress(&event);
+            }
+            Err(err) => {
+                log::error!("error indexing record in Elasticsearch: {}", err);
+                return Err(err);
+            }
+        }
     }
 
     Ok(())

src/sinks/gcp_pubsub/run.rs

@@ -35,16 +35,16 @@ pub fn writer_loop(
     let topic = publisher.topic(topic_name);
 
     for event in input.iter() {
-        // notify the pipeline where we are
-        utils.track_sink_progress(&event);
-
         let result = retry::retry_operation(
             || rt.block_on(send_pubsub_msg(&topic, &event)),
             retry_policy,
         );
 
         match result {
-            Ok(()) => (),
+            Ok(_) => {
+                // notify the pipeline where we are
+                utils.track_sink_progress(&event);
+            }
             Err(err) => match error_policy {
                 ErrorPolicy::Exit => return Err(Box::new(err)),
                 ErrorPolicy::Continue => {

src/sinks/kafka/run.rs

@@ -1,7 +1,6 @@
 use std::sync::Arc;
 
 use kafka::producer::{Producer, Record};
-use log::debug;
 
 use crate::{model::Event, pipelining::StageReceiver, utils::Utils, Error};
 
@@ -22,24 +21,31 @@ pub fn producer_loop(
     utils: Arc<Utils>,
 ) -> Result<(), Error> {
     for event in input.iter() {
-        // notify the pipeline where we are
-        utils.track_sink_progress(&event);
-
         let json = serde_json::to_vec(&event)?;
         let key = define_event_key(&event, &partitioning);
 
-        match key {
+        let result = match key {
             Some(key) => {
                 let r = Record::from_key_value(&topic, &key[..], json);
-                producer.send(&r)?;
+                producer.send(&r)
             }
             None => {
                 let r = Record::from_value(&topic, json);
-                producer.send(&r)?;
+                producer.send(&r)
             }
         };
 
-        debug!("pushed event to kafka: {:?}", &event);
+        match result {
+            Ok(_) => {
+                log::debug!("pushed event to kafka: {:?}", &event);
+                // notify the pipeline where we are
+                utils.track_sink_progress(&event);
+            }
+            Err(err) => {
+                log::error!("error sending kafka message: {}", err);
+                return Err(Box::new(err));
+            }
+        }
     }
 
     Ok(())

src/sinks/logs/run.rs

@@ -10,12 +10,22 @@ pub fn jsonl_writer_loop(
     utils: Arc<Utils>,
 ) -> Result<(), Error> {
     for evt in input.iter() {
-        // notify pipeline about the progress
-        utils.track_sink_progress(&evt);
-
         let buf = json!(evt).to_string();
-        output.write_all(buf.as_bytes())?;
-        output.write_all(b"\n")?;
+
+        let result = output
+            .write_all(buf.as_bytes())
+            .and_then(|_| output.write_all(b"\n"));
+
+        match result {
+            Ok(_) => {
+                // notify pipeline about the progress
+                utils.track_sink_progress(&evt);
+            }
+            Err(err) => {
+                log::error!("error writing to log output: {}", err);
+                return Err(Box::new(err));
+            }
+        }
     }
 
     Ok(())

src/sinks/redis/run.rs

@@ -1,23 +1,8 @@
-#![allow(unused_variables)]
 use super::StreamStrategy;
 use crate::{model::Event, pipelining::StageReceiver, utils::Utils, Error};
-use serde::Serialize;
 use serde_json::json;
 use std::sync::Arc;
 
-#[derive(Serialize)]
-pub struct RedisRecord {
-    pub event: Event,
-    pub key: String,
-}
-
-impl From<Event> for RedisRecord {
-    fn from(event: Event) -> Self {
-        let key = key(&event);
-        RedisRecord { event, key }
-    }
-}
-
 fn key(event: &Event) -> String {
     if let Some(fingerprint) = &event.fingerprint {
         fingerprint.clone()
@@ -34,26 +19,35 @@ pub fn producer_loop(
     redis_stream: String,
 ) -> Result<(), Error> {
     for event in input.iter() {
-        utils.track_sink_progress(&event);
-        let payload = RedisRecord::from(event);
+        let key = key(&event);
 
         let stream = match stream_strategy {
-            StreamStrategy::ByEventType => payload.event.data.clone().to_string().to_lowercase(),
+            StreamStrategy::ByEventType => event.data.clone().to_string().to_lowercase(),
             _ => redis_stream.clone(),
         };
 
         log::debug!(
             "Stream: {:?}, Key: {:?}, Event: {:?}",
-            stream,
-            payload.key,
-            payload.event
+            &stream,
+            &key,
+            &event
         );
 
-        redis::cmd("XADD")
+        let result: Result<(), _> = redis::cmd("XADD")
             .arg(stream)
             .arg("*")
-            .arg(&[(payload.key, json!(payload.event).to_string())])
-            .query(conn)?;
+            .arg(&[(key, json!(event).to_string())])
+            .query(conn);
+
+        match result {
+            Ok(_) => {
+                utils.track_sink_progress(&event);
+            }
+            Err(err) => {
+                log::error!("error sending message to redis: {}", err);
+                return Err(Box::new(err));
+            }
+        }
     }
 
     Ok(())

src/sinks/stdout/run.rs

@@ -10,12 +10,19 @@ pub fn jsonl_writer_loop(
     utils: Arc<Utils>,
 ) -> Result<(), Error> {
     for evt in input.iter() {
-        // notify pipeline about the progress
-        utils.track_sink_progress(&evt);
-
         let buf = json!(evt).to_string();
-        output.write_all(buf.as_bytes())?;
-        output.write_all(b"\n")?;
+
+        let result = output
+            .write_all(buf.as_bytes())
+            .and_then(|_| output.write_all(b"\n"));
+
+        match result {
+            Ok(_) => {
+                // notify pipeline about the progress
+                utils.track_sink_progress(&evt);
+            }
+            Err(err) => return Err(Box::new(err)),
+        }
     }
 
     Ok(())