Periodically prune association tombstones

Author: drexin

Sources/DistributedActors/Cluster/Association.swift

@@ -244,7 +244,7 @@ extension Association {
 
         /// Determines when the Tombstone should be removed from kept tombstones in the ClusterShell.
         /// End of life of the tombstone is calculated as `now + settings.associationTombstoneTTL`.
-        let removalDeadline: Deadline // TODO: cluster should have timer to try to remove those periodically
+        let removalDeadline: Deadline
 
         init(_ node: UniqueNode, settings: ClusterSystemSettings) {
             // TODO: if we made system carry system.time we could always count from that point in time with a TimeAmount; require Clock and settings then

Sources/DistributedActors/Cluster/ClusterShell.swift

@@ -333,6 +333,7 @@ internal class ClusterShell {
         /// Used to signal a "down was issued" either by the user, or another part of the system.
         case downCommandMember(Cluster.Member)
         case shutdown(BlockingReceptacle<Void>) // TODO: could be NIO future
+        case cleanUpAssociationTombstones
     }
 
     enum QueryMessage: NonTransportableActorMessage {
@@ -461,6 +462,17 @@ extension ClusterShell {
         }
     }
 
+    private func cleanUpAssociationTombstones() -> _Behavior<Message> {
+        self._associationsLock.withLockVoid {
+            for (id, tombstone) in self._associationTombstones {
+                if tombstone.removalDeadline.isOverdue() {
+                    self._associationTombstones.removeValue(forKey: id)
+                }
+            }
+        }
+        return .same
+    }
+
     /// Ready and interpreting commands and incoming messages.
     ///
     /// Serves as main "driver" for handshake and association state machines.
@@ -498,6 +510,8 @@ extension ClusterShell {
                 }
             case .downCommandMember(let member):
                 return self.ready(state: self.onDownCommand(context, state: state, member: member))
+            case .cleanUpAssociationTombstones:
+                return self.cleanUpAssociationTombstones()
             }
         }
 

Sources/DistributedActors/ClusterSystem.swift

@@ -57,6 +57,8 @@ public class ClusterSystem: DistributedActorSystem, @unchecked Sendable {
     internal let lifecycleWatchLock = Lock()
     internal var _lifecycleWatches: [ActorAddress: LifecycleWatchContainer] = [:]
 
+    private var _associationTombstoneCleanupTask: RepeatedTask?
+
     private let dispatcher: InternalMessageDispatcher
 
     // Access MUST be protected with `namingLock`.
@@ -300,6 +302,13 @@ public class ClusterSystem: DistributedActorSystem, @unchecked Sendable {
         let clusterRef = try! cluster.start(system: self, clusterEvents: clusterEvents) // only spawns when cluster is initialized
         _ = self._clusterControlStore.storeIfNilThenLoad(Box(ClusterControl(settings, clusterRef: clusterRef, eventStream: clusterEvents)))
 
+        self._associationTombstoneCleanupTask = eventLoopGroup.next().scheduleRepeatedTask(
+            initialDelay: settings.associationTombstoneCleanupInterval.toNIO,
+            delay: settings.associationTombstoneCleanupInterval.toNIO
+        ) { _ in
+            clusterRef.tell(.command(.cleanUpAssociationTombstones))
+        }
+
         // node watcher MUST be prepared before receptionist (or any other actor) because it (and all actors) need it if we're running clustered
         // Node watcher MUST be started AFTER cluster and clusterEvents
         let lazyNodeDeathWatcher = try! self._prepareSystemActor(
@@ -458,6 +467,8 @@ public class ClusterSystem: DistributedActorSystem, @unchecked Sendable {
                  // self._serialization = nil // FIXME: need to release serialization
              }
              */
+            self._associationTombstoneCleanupTask?.cancel()
+            self._associationTombstoneCleanupTask = nil
             _ = self._clusterStore.storeIfNilThenLoad(Box(nil))
 
             self.shutdownReceptacle.offerOnce(nil)

Sources/DistributedActors/ClusterSystemSettings.swift

@@ -115,6 +115,9 @@ public struct ClusterSystemSettings {
     /// communicated with again. Tombstones are used to ensure this, even if the downed ("zombie") node, attempts to reconnect.
     public var associationTombstoneTTL: TimeAmount = .hours(24) * 1
 
+    /// Defines the interval with which the list of associated tombstones is freed from expired tombstones.
+    public var associationTombstoneCleanupInterval: TimeAmount = .minutes(10)
+
     // ==== ------------------------------------------------------------------------------------------------------------
     // MARK: Cluster protocol versioning