Merge pull request #1784 from j5ik2o/remote-reliable-deathwatch

feat(remote): add reliable remote DeathWatch delivery

Author: j5ik2o

docs/gap-analysis/remote-gap-analysis.md

@@ -16,4 +16,8 @@ impl RemoteWatchHook for NoopRemoteWatchHook {
   fn handle_unwatch(&mut self, _target: Pid, _watcher: Pid) -> bool {
     false
   }
+
+  fn handle_deathwatch_notification(&mut self, _watcher: Pid, _terminated: Pid) -> bool {
+    false
+  }
 }

modules/actor-core-kernel/src/system/remote/noop_remote_watch_hook.rs

@@ -12,4 +12,8 @@ pub trait RemoteWatchHook: Send + 'static {
 
   /// Handles an unwatch request. Returns `true` when the provider consumed the message.
   fn handle_unwatch(&mut self, target: Pid, watcher: Pid) -> bool;
+
+  /// Handles a remote-bound termination notification. Returns `true` when the provider consumed the
+  /// message.
+  fn handle_deathwatch_notification(&mut self, watcher: Pid, terminated: Pid) -> bool;
 }

modules/actor-core-kernel/src/system/remote/remote_watch_hook.rs

@@ -55,6 +55,11 @@ impl RemoteWatchHookDynShared {
   pub(crate) fn handle_unwatch(&self, target: Pid, watcher: Pid) -> bool {
     self.with_write(|inner| inner.handle_unwatch(target, watcher))
   }
+
+  /// Handles a remote-bound termination notification by delegating to the inner hook.
+  pub(crate) fn handle_deathwatch_notification(&self, watcher: Pid, terminated: Pid) -> bool {
+    self.with_write(|inner| inner.handle_deathwatch_notification(watcher, terminated))
+  }
 }
 
 impl SharedAccess<Box<dyn RemoteWatchHook>> for RemoteWatchHookDynShared {

modules/actor-core-kernel/src/system/remote/remote_watch_hook_dyn_shared.rs

@@ -753,7 +753,12 @@ impl SystemStateShared {
           }
           Ok(())
         },
-        | SystemMessage::DeathWatchNotification(_) => Ok(()),
+        | SystemMessage::DeathWatchNotification(terminated) => {
+          if self.remote_watch_hook.handle_deathwatch_notification(pid, terminated) {
+            return Ok(());
+          }
+          Ok(())
+        },
         | SystemMessage::PipeTask(_) => Ok(()),
         | other => Err(SendError::closed(AnyMessage::new(other))),
       }

modules/actor-core-kernel/src/system/state/system_state_shared.rs

@@ -687,6 +687,29 @@ fn remote_watch_hook_consumes_unwatch_is_invoked() {
   assert_eq!(calls.last_unwatch, Some((target_pid, watcher_pid)));
 }
 
+#[test]
+fn remote_watch_hook_consumes_deathwatch_notification_is_invoked() {
+  let state = build_shared_state();
+  let watcher_pid = state.allocate_pid();
+  let target_pid = state.allocate_pid();
+
+  let calls = ArcShared::new(SpinSyncMutex::new(RemoteWatchHookCalls::default()));
+  state.register_remote_watch_hook(Box::new(RecordingRemoteWatchHook::with_notification(
+    calls.clone(),
+    false,
+    false,
+    true,
+  )));
+
+  state
+    .send_system_message(watcher_pid, SystemMessage::DeathWatchNotification(target_pid))
+    .expect("death-watch notification send ok");
+
+  let calls = calls.lock();
+  assert_eq!(calls.notification_calls, 1);
+  assert_eq!(calls.last_notification, Some((watcher_pid, target_pid)));
+}
+
 #[test]
 fn remote_watch_hook_replaces_previous_registration() {
   let state = build_shared_state_with_noop_dispatcher();
@@ -759,21 +782,33 @@ struct RestartProbeActor;
 
 #[derive(Default)]
 struct RemoteWatchHookCalls {
-  watch_calls:   usize,
-  unwatch_calls: usize,
-  last_watch:    Option<(Pid, Pid)>,
-  last_unwatch:  Option<(Pid, Pid)>,
+  watch_calls:        usize,
+  unwatch_calls:      usize,
+  notification_calls: usize,
+  last_watch:         Option<(Pid, Pid)>,
+  last_unwatch:       Option<(Pid, Pid)>,
+  last_notification:  Option<(Pid, Pid)>,
 }
 
 struct RecordingRemoteWatchHook {
-  calls:           ArcShared<SpinSyncMutex<RemoteWatchHookCalls>>,
-  consume_watch:   bool,
-  consume_unwatch: bool,
+  calls:                ArcShared<SpinSyncMutex<RemoteWatchHookCalls>>,
+  consume_watch:        bool,
+  consume_unwatch:      bool,
+  consume_notification: bool,
 }
 
 impl RecordingRemoteWatchHook {
   fn new(calls: ArcShared<SpinSyncMutex<RemoteWatchHookCalls>>, consume_watch: bool, consume_unwatch: bool) -> Self {
-    Self { calls, consume_watch, consume_unwatch }
+    Self::with_notification(calls, consume_watch, consume_unwatch, false)
+  }
+
+  fn with_notification(
+    calls: ArcShared<SpinSyncMutex<RemoteWatchHookCalls>>,
+    consume_watch: bool,
+    consume_unwatch: bool,
+    consume_notification: bool,
+  ) -> Self {
+    Self { calls, consume_watch, consume_unwatch, consume_notification }
   }
 }
 
@@ -791,6 +826,13 @@ impl crate::system::remote::RemoteWatchHook for RecordingRemoteWatchHook {
     calls.last_unwatch = Some((target, watcher));
     self.consume_unwatch
   }
+
+  fn handle_deathwatch_notification(&mut self, watcher: Pid, terminated: Pid) -> bool {
+    let mut calls = self.calls.lock();
+    calls.notification_calls += 1;
+    calls.last_notification = Some((watcher, terminated));
+    self.consume_notification
+  }
 }
 
 struct RemoteEventRecorder {