Improve handling of closed channels.

Author: tmpfs

crates/database/src/audit_provider.rs

@@ -100,23 +100,25 @@ where
                     let mut rows = stmt.query([])?;
 
                     while let Some(row) = rows.next()? {
+                        if (tx.is_closed()) {
+                            break;
+                        }
                         let row: AuditRow = row.try_into()?;
                         let record: AuditRecord = row.try_into()?;
                         let inner_tx = tx.clone();
-                        futures::executor::block_on(async move {
-                            if let Err(e) =
-                                inner_tx.send(Ok(record.event)).await
-                            {
-                                tracing::error!(error = %e);
-                            }
+                        let res = futures::executor::block_on(async move {
+                            inner_tx.send(Ok(record.event)).await
                         });
+                        if let Err(e) = res {
+                            tracing::error!(error = %e);
+                            break;
+                        }
                     }
 
                     Ok::<_, Error>(())
                 })
                 .await
                 .map_err(Error::from)?;
-
             Ok::<_, Self::Error>(())
         });
 

crates/database/src/event_log.rs

@@ -393,14 +393,19 @@ where
                     })?;
 
                     for row in rows {
+                        if tx.is_closed() {
+                            break;
+                        }
                         let row = row?;
                         let record: EventRecord = row.try_into()?;
-                        let sender = tx.clone();
-                        futures::executor::block_on(async move {
-                            if let Err(err) = sender.send(Ok(record)).await {
-                                tracing::error!(error = %err);
-                            }
+                        let inner_tx = tx.clone();
+                        let res = futures::executor::block_on(async move {
+                            inner_tx.send(Ok(record)).await
                         });
+                        if let Err(e) = res {
+                            tracing::error!(error = %e);
+                            break;
+                        }
                     }
 
                     Ok::<_, Error>(())

crates/filesystem/src/audit_provider.rs

@@ -222,6 +222,9 @@ where
         let it_file = self.file.clone();
         tokio::task::spawn(async move {
             while let Some(record) = it.next().await? {
+                if tx.is_closed() {
+                    break;
+                }
                 let mut inner = it_file.lock().await;
                 let event = inner.read_event(&record).await?;
                 if let Err(e) = tx.send(Ok(event)).await {

crates/filesystem/src/event_log.rs

@@ -142,6 +142,9 @@ where
         let file_path = self.data.clone();
         tokio::task::spawn(async move {
             while let Some(record) = it.next().await? {
+                if tx.is_closed() {
+                    break;
+                }
                 let event_buffer =
                     read_event_buffer(file_path.clone(), &record).await?;
                 let event_record = record.into_event_record(event_buffer);