Introduce terminate feature for actors, including handling of termination state, tests, and enhanced Actor lifecycle management.

Author: smyrgeorge

actor4k-cluster/src/commonMain/kotlin/io/github/smyrgeorge/actor4k/cluster/ClusterActorRef.kt

@@ -125,6 +125,26 @@ class ClusterActorRef(
         }
     }
 
+    /**
+     * Terminates the associated actor by sending a termination request to its service.
+     *
+     * This method interacts with the actor's service to initiate a termination operation.
+     * It processes the service's response to determine whether the termination succeeded,
+     * failed, or resulted in an unexpected outcome.
+     *
+     * @return A [Result] wrapping [Unit] if the termination operation is successful.
+     *         If the response indicates a failure, an exception is wrapped in the result.
+     *         An [IllegalStateException] is returned for unexpected response types.
+     */
+    override suspend fun terminate(): Result<Unit> {
+        val res = service.terminate(address).getOrElse { return Result.failure(it) }
+        return when (res) {
+            is RpcEnvelope.Response.Empty -> Result.success(Unit)
+            is RpcEnvelope.Response.Failure -> Result.failure(res.exception())
+            else -> Result.failure(IllegalStateException("Unexpected response $res for ask command."))
+        }
+    }
+
     /**
      * Provides a string representation of the `ClusterActorRef` instance.
      *

actor4k-cluster/src/commonMain/kotlin/io/github/smyrgeorge/actor4k/cluster/rpc/RpcEnvelope.kt

@@ -34,6 +34,9 @@ interface RpcEnvelope {
 
         @Serializable
         data class Shutdown(override val id: Long, val addr: Address) : Request
+
+        @Serializable
+        data class Terminate(override val id: Long, val addr: Address) : Request
     }
 
     /**

actor4k-cluster/src/commonMain/kotlin/io/github/smyrgeorge/actor4k/cluster/rpc/RpcReceiveService.kt

@@ -5,7 +5,9 @@ import io.github.smyrgeorge.actor4k.cluster.rpc.RpcEnvelope.Request
 import io.github.smyrgeorge.actor4k.cluster.rpc.RpcEnvelope.Response
 import io.github.smyrgeorge.actor4k.util.Logger
 import io.github.smyrgeorge.actor4k.util.extentions.launch
-import io.ktor.websocket.*
+import io.ktor.websocket.Frame
+import io.ktor.websocket.WebSocketSession
+import io.ktor.websocket.send
 import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.protobuf.ProtoBuf
 
@@ -42,14 +44,15 @@ class RpcReceiveService(
                 log.warn("Received non-binary frame: $frame")
                 return@launch
             }
-            val msg = protoBuf.decodeFromByteArray(Request.serializer(), frame.data)
-            val res = when (msg) {
+            val res = when (val msg = protoBuf.decodeFromByteArray(Request.serializer(), frame.data)) {
                 is Request.Echo -> echo(msg)
                 is Request.Tell -> tell(msg)
                 is Request.Ask -> ask(msg)
                 is Request.Status -> status(msg)
                 is Request.Stats -> stats(msg)
                 is Request.Shutdown -> shutdown(msg)
+                is Request.Terminate -> terminate(msg)
+
             }
             session.send(protoBuf.encodeToByteArray(Response.serializer(), res))
         }
@@ -140,6 +143,22 @@ class RpcReceiveService(
         return Response.Empty(msg.id)
     }
 
+    /**
+     * Processes a Terminate request by retrieving the actor associated with the specified address
+     * and attempting to terminate it. Constructs a Response based on the outcome of the operation.
+     *
+     * @param msg The Terminate request containing the unique message ID and the address of the actor to terminate.
+     * @return A `Response.Empty` if the termination is successfully completed, or a `Response.Failure`
+     *         if an error occurs during the operation.
+     */
+    private suspend fun terminate(msg: Request.Terminate): Response {
+        registry.get(msg.addr)
+            .getOrElse { return it.failure(msg.id) }
+            .terminate()
+            .getOrElse { return it.failure(msg.id) }
+        return Response.Empty(msg.id)
+    }
+
     companion object {
         private fun Throwable.failure(id: Long): Response.Failure =
             Response.Failure(id, message, cause?.message)

actor4k-cluster/src/commonMain/kotlin/io/github/smyrgeorge/actor4k/cluster/rpc/RpcSendService.kt

@@ -159,6 +159,24 @@ class RpcSendService(
         res
     }
 
+    /**
+     * Sends a termination request to the specified actor's address and waits for the corresponding response.
+     * This method ensures that the response ID matches the request ID to maintain consistency.
+     *
+     * @param addr The address of the actor to be terminated.
+     * @return A [Result] containing the [Response] received in response to the termination request,
+     * or an error if the operation fails.
+     * @throws IllegalStateException if the response ID does not match the request ID.
+     */
+    suspend fun terminate(addr: Address): Result<Response> = runCatching {
+        val req = Request.Terminate(nextId(), addr)
+        val res = rpc.request<Response>(req.id) {
+            session.send(req.serialize())
+        }
+        if (res.id != req.id) error("Sanity check failed :: req.id != res.id.")
+        res
+    }
+
     /**
      * Closes the current session associated with the `RpcSendService`.
      *

actor4k/src/commonMain/kotlin/io/github/smyrgeorge/actor4k/actor/Actor.kt

@@ -47,8 +47,9 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
     protected val log: Logger = ActorSystem.loggerFactory.getLogger(this::class)
 
     private val stats: Stats = Stats()
-    private var status = Status.CREATED
-    private var initializationFailed: Exception? = null
+    private var status: Status = Status.CREATED
+    private var initializationError: Throwable? = null
+    private var terminationError: Throwable? = null
     private val address: Address = Address.of(this::class, key)
     private val ref: LocalRef = LocalRef(address = address, actor = this)
 
@@ -164,10 +165,10 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
                 onBeforeActivate()
                 // Set 'ACTIVATING' status.
                 status = Status.ACTIVATING
-            } catch (e: Exception) {
+            } catch (e: Throwable) {
                 // In case of an error, we need to close the [Actor] immediately.
                 log.error("[$address::onBeforeActivate] Failed to activate, will shutdown (${e.message ?: ""})")
-                initializationFailed = e
+                initializationError = e
                 shutdown()
             }
 
@@ -178,11 +179,18 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
                 // Case that activation flow failed and we still have messages to consume.
                 // If we get a shutdown event and the actor never initialized successfully,
                 // we need to reply with an error and to drop all the messages.
-                if (initializationFailed != null) {
+                if (initializationError != null) {
                     replyActivationError(pattern)
                     return@consumeEach
                 }
 
+                // Case that termination flow was triggered.
+                // In this case, we need to reply with an error and to drop all the messages.
+                if (terminationError != null) {
+                    replyTerminationError(pattern)
+                    return@consumeEach
+                }
+
                 val msg: Req = pattern.msg.apply {
                     id = stats.receivedMessages
                     kind = pattern.messageKind
@@ -195,10 +203,10 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
                         // Set 'READY' status.
                         status = Status.READY
                         stats.initializedAt = Clock.System.now().toEpochMilliseconds()
-                    } catch (e: Exception) {
+                    } catch (e: Throwable) {
                         // In case of an error, we need to close the [Actor] immediately.
                         log.error("[$address::activate] Failed to activate, will shutdown (${e.message ?: ""})")
-                        initializationFailed = e
+                        initializationError = e
                         replyActivationError(pattern)
                         shutdown()
                         return@consumeEach
@@ -234,7 +242,7 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
 
                 else -> r
             }
-        } catch (e: Exception) {
+        } catch (e: Throwable) {
             Behavior.Error(e)
         }
 
@@ -262,7 +270,7 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
                 // Handle afterReceive hook.
                 try {
                     afterReceive(msg, result)
-                } catch (e: Exception) {
+                } catch (e: Throwable) {
                     log.warn("[$address::afterReceive] Failed to process afterReceive hook (${e.message ?: ""})")
                 }
             }
@@ -272,12 +280,13 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
                 // This allows Router workers (FIRST_AVAILABLE) to re-register availability, avoiding deadlocks.
                 try {
                     afterReceive(msg)
-                } catch (e: Exception) {
+                } catch (e: Throwable) {
                     log.warn("[$address::afterReceive] Failed to process afterReceive hook for Behavior.None (${e.message ?: ""})")
                 }
             }
 
             is Behavior.Shutdown -> shutdown()
+            is Behavior.Terminate -> terminate()
         }
     }
 
@@ -323,7 +332,7 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
                 mail.send(ask)
                 ask.replyTo.receive()
             }
-        } catch (e: Exception) {
+        } catch (e: Throwable) {
             Result.failure(e)
         } finally {
             ask.replyTo.close()
@@ -391,6 +400,8 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
         // Drain the stash channel.
         @OptIn(ExperimentalCoroutinesApi::class)
         while (!stash.isEmpty) {
+            // Skip processing if actor cannot handle messages (drain the stash).
+            if (status == Status.SHUT_DOWN || status == Status.TERMINATED) continue
             val pattern = stash.receive()
             stats.stashedMessages -= 1
             process(pattern)
@@ -415,6 +426,24 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
         stash.close()
     }
 
+    /**
+     * Initiates the termination process for the system or component.
+     *
+     * This method checks whether termination can be triggered based on the current status.
+     * If termination is allowed, it updates the termination timestamp, changes the status
+     * to `TERMINATING`, and closes associated resources such as mail and stash.
+     *
+     * The method has no effect if termination is not permitted.
+     */
+    fun terminate(error: Throwable? = null) {
+        if (!status.canTriggerTermination) return
+        stats.triggeredTerminationAt = Clock.System.now().toEpochMilliseconds()
+        terminationError = error ?: Exception("Actor terminated by user.")
+        status = Status.TERMINATING
+        mail.close()
+        stash.close()
+    }
+
     /**
      * Represents message patterns used by the `Actor` for communication and message handling.
      *
@@ -479,17 +508,22 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
      * - `READY`: The actor is fully initialized and ready to process messages. Messages can be accepted during this state.
      * - `SHUTTING_DOWN`: The actor is in the process of shutting down. Messages cannot be accepted during this state.
      * - `SHUT_DOWN`: The actor has completed the shutdown process. Messages cannot be accepted during this state.
+     * - `TERMINATING`: The actor is in the process of terminating. Messages cannot be accepted during this state.
+     * - `TERMINATED`: The actor has terminated and is no longer available for use. Messages cannot be accepted during this state.
      */
     @Serializable
     enum class Status(
         val canAcceptMessages: Boolean,
         val canTriggerShutdown: Boolean,
+        val canTriggerTermination: Boolean,
     ) {
-        CREATED(true, true),
-        ACTIVATING(true, true),
-        READY(true, true),
-        SHUTTING_DOWN(false, false),
-        SHUT_DOWN(false, false)
+        CREATED(true, true, true),
+        ACTIVATING(true, true, true),
+        READY(true, true, true),
+        SHUTTING_DOWN(false, false, false),
+        SHUT_DOWN(false, false, false),
+        TERMINATING(false, false, false),
+        TERMINATED(false, false, false);
     }
 
     /**
@@ -513,6 +547,8 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
         var initializedAt: Long? = null,
         var triggeredShutDownAt: Long? = null,
         var shutDownAt: Long? = null,
+        var triggeredTerminationAt: Long? = null,
+        var terminatedAt: Long? = null,
         var lastMessageAt: Long = Clock.System.now().toEpochMilliseconds(),
         var receivedMessages: Long = 0,
         var stashedMessages: Long = 0
@@ -523,6 +559,8 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
             append("initializedAt=${instantFromEpochMilliseconds(initializedAt)}, ")
             append("triggeredShutDownAt=${instantFromEpochMilliseconds(triggeredShutDownAt)}, ")
             append("shutDownAt=${instantFromEpochMilliseconds(shutDownAt)}, ")
+            append("triggeredTerminationAt=${instantFromEpochMilliseconds(triggeredTerminationAt)}, ")
+            append("terminatedAt=${instantFromEpochMilliseconds(terminatedAt)}, ")
             append("lastMessageAt=${instantFromEpochMilliseconds(lastMessageAt)}, ")
             append("receivedMessages=${instantFromEpochMilliseconds(receivedMessages)}, ")
             append("stashedMessages=$stashedMessages")
@@ -544,7 +582,7 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
             for (e in this) {
                 try {
                     action(e)
-                } catch (e: Exception) {
+                } catch (e: Throwable) {
                     log.warn("[$address::consume] An error occurred while processing. ${e.message ?: ""}")
                 }
             }
@@ -561,12 +599,16 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
                 }
             } catch (_: TimeoutCancellationException) {
                 log.error("[$address::onShutdown] Shutdown hook timed out after ${ActorSystem.conf.actorShutdownHookTimeout}. Forcing shutdown.")
-            } catch (e: Exception) {
+            } catch (e: Throwable) {
                 log.error("[$address::onShutdown] Error during shutdown hook: ${e.message ?: "Unknown error"}")
             } finally {
-                // Unregister the actor even if the shutdown hook fails or times out
-                status = Status.SHUT_DOWN
-                stats.shutDownAt = Clock.System.now().toEpochMilliseconds()
+                if (terminationError != null) {
+                    status = Status.TERMINATED
+                    stats.terminatedAt = Clock.System.now().toEpochMilliseconds()
+                } else {
+                    status = Status.SHUT_DOWN
+                    stats.shutDownAt = Clock.System.now().toEpochMilliseconds()
+                }
                 ActorSystem.registry.unregister(this@Actor.ref)
             }
         }
@@ -591,7 +633,7 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
             log.warn("[$address::$operation] Could not reply in time (timeout after ${ActorSystem.conf.actorReplyTimeout}) (the message was processed successfully).")
         } catch (_: ClosedSendChannelException) {
             log.warn("[$address::$operation] Could not reply, the channel is closed (the message was processed successfully).")
-        } catch (e: Exception) {
+        } catch (e: Throwable) {
             val error = e.message ?: "Unknown error."
             log.warn("[$address::$operation] Could not reply (the message was processed successfully). {}", error)
         }
@@ -607,14 +649,32 @@ abstract class Actor<Req : ActorProtocol, Res : ActorProtocol.Response>(
         when (pattern) {
             is Patterns.Tell -> Unit
             is Patterns.Ask -> {
-                val e = initializationFailed
+                val e = initializationError
                     ?: IllegalStateException("Actor is prematurely closed (could not be initialized).")
                 val r: Result<Res> = Result.failure(e)
                 reply(operation = "activate", pattern = pattern, reply = r)
             }
         }
     }
 
+    /**
+     * Handles the termination error scenario for the provided message pattern.
+     *
+     * @param pattern The message pattern that specifies the type of interaction (e.g., Tell or Ask)
+     * and determines how the termination error is managed.
+     */
+    private suspend fun replyTerminationError(pattern: Patterns<Req, Res>) {
+        when (pattern) {
+            is Patterns.Tell -> Unit
+            is Patterns.Ask -> {
+                val e = terminationError
+                    ?: IllegalStateException("Actor is terminated, all pending messages are replied with errors.")
+                val r: Result<Res> = Result.failure(e)
+                reply(operation = "terminate", pattern = pattern, reply = r)
+            }
+        }
+    }
+
     companion object {
         /**
          * Generates a unique random key as a string, which includes a "key-" prefix followed by a hash code

actor4k/src/commonMain/kotlin/io/github/smyrgeorge/actor4k/actor/Behavior.kt

@@ -58,4 +58,15 @@ sealed interface Behavior<Res : Response> {
      * @param Res The type of `Response` this behavior operates on.
      */
     class Shutdown<Res : Response> : Behavior<Res>
+
+    /**
+     * Represents a termination behavior that extends the capabilities of a `Behavior`.
+     *
+     * The `Terminate` class models the end of a communication or processing process, providing
+     * specialized features tailored for handling scenarios where a response of type `Res` is required.
+     *
+     * @param Res The type of the response that extends the `Response` class. This ensures the termination
+     * behavior conforms to the expected structure and properties of a response defined by the protocol.
+     */
+    class Terminate<Res : Response> : Behavior<Res>
 }