Unit test for panick, and handling a post-mortem observation.

Author: fulmicoton

quickwit-actors/src/actor_handle.rs

@@ -235,10 +235,7 @@ mod tests {
             _ctx: &ActorContext<Self>,
         ) -> Result<(), ActorExitStatus> {
             self.count += 1;
-            if self.count == 2 {
-                panic!("Oops");
-            }
-            Ok(())
+            panic!("Oops");
         }
     }
 
@@ -250,10 +247,7 @@ mod tests {
             _ctx: &ActorContext<Self>,
         ) -> Result<(), ActorExitStatus> {
             self.count += 1;
-            if self.count == 2 {
-                panic!("Oops");
-            }
-            Ok(())
+            panic!("Oops");
         }
     }
 
@@ -277,10 +271,7 @@ mod tests {
             _ctx: &ActorContext<Self>,
         ) -> Result<(), ActorExitStatus> {
             self.count += 1;
-            if self.count == 2 {
-                return Err(ActorExitStatus::DownstreamClosed);
-            }
-            Ok(())
+            Err(ActorExitStatus::DownstreamClosed)
         }
     }
 
@@ -292,10 +283,7 @@ mod tests {
             _ctx: &ActorContext<Self>,
         ) -> Result<(), ActorExitStatus> {
             self.count += 1;
-            if self.count == 2 {
-                return Err(ActorExitStatus::DownstreamClosed);
-            }
-            Ok(())
+            Err(ActorExitStatus::DownstreamClosed)
         }
     }
 
@@ -304,8 +292,6 @@ mod tests {
         let universe = Universe::new();
         let (mailbox, handle) = universe.spawn(PanickingActor::default());
         universe.send_message(&mailbox, ()).await?;
-        assert_eq!(handle.process_pending_and_observe().await.state, 1);
-        universe.send_message(&mailbox, ()).await?;
         let (exit_status, count) = handle.join().await;
         assert!(matches!(exit_status, ActorExitStatus::Panicked));
         assert!(matches!(count, 1)); //< Upon panick we cannot get a post mortem state.
@@ -317,11 +303,9 @@ mod tests {
         let universe = Universe::new();
         let (mailbox, handle) = universe.spawn_sync_actor(PanickingActor::default());
         universe.send_message(&mailbox, ()).await?;
-        assert_eq!(handle.process_pending_and_observe().await.state, 1);
-        universe.send_message(&mailbox, ()).await?;
         let (exit_status, count) = handle.join().await;
         assert!(matches!(exit_status, ActorExitStatus::Panicked));
-        assert!(matches!(count, 1)); //< Upon panick we cannot get a post mortem state.
+        assert!(matches!(count, 1));
         Ok(())
     }
 
@@ -330,11 +314,9 @@ mod tests {
         let universe = Universe::new();
         let (mailbox, handle) = universe.spawn(ExitActor::default());
         universe.send_message(&mailbox, ()).await?;
-        assert_eq!(handle.process_pending_and_observe().await.state, 1);
-        universe.send_message(&mailbox, ()).await?;
         let (exit_status, count) = handle.join().await;
         assert!(matches!(exit_status, ActorExitStatus::DownstreamClosed));
-        assert!(matches!(count, 2)); //< Upon panick we cannot get a post mortem state.
+        assert!(matches!(count, 1)); //< Upon panick we cannot get a post mortem state.
         Ok(())
     }
 
@@ -343,11 +325,9 @@ mod tests {
         let universe = Universe::new();
         let (mailbox, handle) = universe.spawn_sync_actor(ExitActor::default());
         universe.send_message(&mailbox, ()).await?;
-        assert_eq!(handle.process_pending_and_observe().await.state, 1);
-        universe.send_message(&mailbox, ()).await?;
         let (exit_status, count) = handle.join().await;
         assert!(matches!(exit_status, ActorExitStatus::DownstreamClosed));
-        assert!(matches!(count, 2));
+        assert!(matches!(count, 1));
         Ok(())
     }
 }

quickwit-actors/src/actor_with_state_tx.rs

@@ -0,0 +1,36 @@
+use tokio::sync::watch::Sender;
+
+use crate::Actor;
+
+// Quickwit
+//  Copyright (C) 2021 Quickwit Inc.
+//
+//  Quickwit is offered under the AGPL v3.0 and as commercial software.
+//  For commercial licensing, contact us at [email protected].
+//
+//  AGPL:
+//  This program is free software: you can redistribute it and/or modify
+//  it under the terms of the GNU Affero General Public License as
+//  published by the Free Software Foundation, either version 3 of the
+//  License, or (at your option) any later version.
+//
+//  This program is distributed in the hope that it will be useful,
+//  but WITHOUT ANY WARRANTY; without even the implied warranty of
+//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+//  GNU Affero General Public License for more details.
+//
+//  You should have received a copy of the GNU Affero General Public License
+//  along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+pub struct ActorWithStateTx<A: Actor> {
+    pub actor: A,
+    pub state_tx: Sender<A::ObservableState>,
+}
+
+impl<A: Actor> Drop for ActorWithStateTx<A> {
+    fn drop(&mut self) {
+        let final_state = self.actor.observable_state();
+        // We ignore the result here. An error only marks the absence of an observer.
+        let _ = self.state_tx.send(final_state);
+    }
+}

quickwit-actors/src/async_actor.rs

@@ -21,6 +21,7 @@
 use crate::actor::{process_command, ActorExitStatus};
 use crate::actor_handle::ActorHandle;
 use crate::actor_state::ActorState;
+use crate::actor_with_state_tx::ActorWithStateTx;
 use crate::mailbox::{create_mailbox, CommandOrMessage, Inbox};
 use crate::scheduler::SchedulerMessage;
 use crate::{Actor, ActorContext, KillSwitch, Mailbox, RecvError};
@@ -135,29 +136,40 @@ async fn process_msg<A: Actor + AsyncActor>(
 }
 
 async fn async_actor_loop<A: AsyncActor>(
-    mut actor: A,
+    actor: A,
     mut inbox: Inbox<A::Message>,
     mut ctx: ActorContext<A>,
     state_tx: Sender<A::ObservableState>,
 ) -> ActorExitStatus {
-    let mut exit_status_opt: Option<ActorExitStatus> = actor.initialize(&ctx).await.err();
+    // We rely on this object internally to fetch a post-mortem state,
+    // even in case of a panic.
+    let mut actor_with_state_tx = ActorWithStateTx { actor, state_tx };
+
+    let mut exit_status_opt: Option<ActorExitStatus> =
+        actor_with_state_tx.actor.initialize(&ctx).await.err();
+
     let exit_status: ActorExitStatus = loop {
         tokio::task::yield_now().await;
         if let Some(exit_status) = exit_status_opt {
             break exit_status;
         }
-        exit_status_opt = process_msg(&mut actor, &mut inbox, &mut ctx, &state_tx).await;
+        exit_status_opt = process_msg(
+            &mut actor_with_state_tx.actor,
+            &mut inbox,
+            &mut ctx,
+            &actor_with_state_tx.state_tx,
+        )
+        .await;
     };
     ctx.exit(&exit_status);
 
-    if let Err(finalize_error) = actor
+    if let Err(finalize_error) = actor_with_state_tx
+        .actor
         .finalize(&exit_status, &ctx)
         .await
-        .with_context(|| format!("Finalization of actor {}", actor.name()))
+        .with_context(|| format!("Finalization of actor {}", actor_with_state_tx.actor.name()))
     {
         error!(err=?finalize_error, "finalize_error");
     }
-    let final_state = actor.observable_state();
-    let _ = state_tx.send(final_state);
     exit_status
 }

quickwit-actors/src/lib.rs

@@ -33,6 +33,7 @@ use tokio::time::Duration;
 mod actor;
 mod actor_handle;
 mod actor_state;
+mod actor_with_state_tx;
 mod async_actor;
 pub(crate) mod channel_with_priority;
 mod kill_switch;

quickwit-actors/src/sync_actor.rs

@@ -24,6 +24,7 @@ use tracing::{debug, error, info};
 
 use crate::actor::{process_command, ActorExitStatus};
 use crate::actor_state::ActorState;
+use crate::actor_with_state_tx::ActorWithStateTx;
 use crate::mailbox::{create_mailbox, CommandOrMessage, Inbox};
 use crate::scheduler::SchedulerMessage;
 use crate::{Actor, ActorContext, ActorHandle, KillSwitch, Mailbox, RecvError};
@@ -132,23 +133,32 @@ fn process_msg<A: Actor + SyncActor>(
 }
 
 fn sync_actor_loop<A: SyncActor>(
-    mut actor: A,
+    actor: A,
     mut inbox: Inbox<A::Message>,
     mut ctx: ActorContext<A>,
     state_tx: Sender<A::ObservableState>,
 ) -> ActorExitStatus {
-    let mut exit_status_opt: Option<ActorExitStatus> = actor.initialize(&ctx).err();
+    // We rely on this object internally to fetch a post-mortem state,
+    // even in case of a panic.
+    let mut actor_with_state_tx = ActorWithStateTx { actor, state_tx };
+
+    let mut exit_status_opt: Option<ActorExitStatus> =
+        actor_with_state_tx.actor.initialize(&ctx).err();
+
     let exit_status: ActorExitStatus = loop {
         if let Some(exit_status) = exit_status_opt {
             break exit_status;
         }
-        exit_status_opt = process_msg(&mut actor, &mut inbox, &mut ctx, &state_tx);
+        exit_status_opt = process_msg(
+            &mut actor_with_state_tx.actor,
+            &mut inbox,
+            &mut ctx,
+            &actor_with_state_tx.state_tx,
+        );
     };
     ctx.exit(&exit_status);
-    if let Err(error) = actor.finalize(&exit_status, &ctx) {
+    if let Err(error) = actor_with_state_tx.actor.finalize(&exit_status, &ctx) {
         error!(error=?error, "Finalizing failed");
     }
-    let final_state = actor.observable_state();
-    let _ = state_tx.send(final_state);
     exit_status
 }