API-evolution proof the DowningStrategyDirective

Author: ktoso

Sources/ActorSingletonPlugin/ActorSingletonManager.swift

@@ -71,7 +71,7 @@ internal class ActorSingletonManager<Message: Codable> {
         try context.stop(child: singleton)
     }
 
-    internal enum Directive: NotActuallyCodableMessage {
+    internal enum Directive: _NotActuallyCodableMessage {
         case takeOver(from: UniqueNode?, replyTo: _ActorRef<_ActorRef<Message>?>)
         case handOver(to: UniqueNode?)
         case stop

Sources/DistributedActors/ActorMessages.swift

@@ -16,7 +16,7 @@ import struct Foundation.Data
 import NIO
 
 /// A `Never` can never be sent as message, even more so over the wire.
-extension Never: NotActuallyCodableMessage {}
+extension Never: _NotActuallyCodableMessage {}
 
 // ==== ----------------------------------------------------------------------------------------------------------------
 // MARK: Common utility messages
@@ -128,7 +128,7 @@ public struct BestEffortStringError: Error, Codable, Equatable, CustomStringConv
 }
 
 /// Useful error wrapper which performs an best effort Error serialization as configured by the actor system.
-public struct NonTransportableAnyError: Error, NotActuallyCodableMessage {
+public struct NonTransportableAnyError: Error, _NotActuallyCodableMessage {
     public let failure: Error
 
     public init<Failure: Error>(_ failure: Failure) {
@@ -150,22 +150,22 @@ public struct NonTransportableAnyError: Error, NotActuallyCodableMessage {
 /// No serializer is expected to be registered for such types.
 ///
 /// - Warning: Attempting to send such message over the network will fail at runtime (and log an error or warning).
-public protocol NotActuallyCodableMessage: Codable {}
+public protocol _NotActuallyCodableMessage: Codable {}
 
-extension NotActuallyCodableMessage {
+extension _NotActuallyCodableMessage {
     public init(from decoder: Swift.Decoder) throws {
-        fatalError("Attempted to decode NotActuallyCodableMessage message: \(Self.self)! This should never happen.")
+        fatalError("Attempted to decode _NotActuallyCodableMessage message: \(Self.self)! This should never happen.")
     }
 
     public func encode(to encoder: Swift.Encoder) throws {
-        fatalError("Attempted to encode NotActuallyCodableMessage message: \(Self.self)! This should never happen.")
+        fatalError("Attempted to encode _NotActuallyCodableMessage message: \(Self.self)! This should never happen.")
     }
 
     public init(context: Serialization.Context, from buffer: inout ByteBuffer, using manifest: Serialization.Manifest) throws {
-        fatalError("Attempted to deserialize NotActuallyCodableMessage message: \(Self.self)! This should never happen.")
+        fatalError("Attempted to deserialize _NotActuallyCodableMessage message: \(Self.self)! This should never happen.")
     }
 
     public func serialize(context: Serialization.Context, to bytes: inout ByteBuffer) throws {
-        fatalError("Attempted to serialize NotActuallyCodableMessage message: \(Self.self)! This should never happen.")
+        fatalError("Attempted to serialize _NotActuallyCodableMessage message: \(Self.self)! This should never happen.")
     }
 }

Sources/DistributedActors/ActorRef+Ask.swift

@@ -238,7 +238,7 @@ extension AskResponse {
 ///
 // TODO: replace with a special minimal `_ActorRef` that does not require spawning or scheduling.
 internal enum AskActor {
-    enum Event: NotActuallyCodableMessage {
+    enum Event: _NotActuallyCodableMessage {
         case timeout
     }
 

Sources/DistributedActors/Cluster/ClusterShell.swift

@@ -324,7 +324,7 @@ internal class ClusterShell {
     }
 
     // this is basically our API internally for this system
-    enum CommandMessage: NotActuallyCodableMessage, SilentDeadLetter {
+    enum CommandMessage: _NotActuallyCodableMessage, SilentDeadLetter {
         /// Connect and handshake with remote `Node`, obtaining an `UniqueNode` in the process.
         /// Once the handshake is completed, reply to `replyTo` with the handshake result, and also mark the unique node as `.joining`.
         ///
@@ -343,7 +343,7 @@ internal class ClusterShell {
         case cleanUpAssociationTombstones
     }
 
-    enum QueryMessage: NotActuallyCodableMessage {
+    enum QueryMessage: _NotActuallyCodableMessage {
         case associatedNodes(_ActorRef<Set<UniqueNode>>) // TODO: better type here
         case currentMembership(_ActorRef<Cluster.Membership>)
     }
@@ -362,7 +362,7 @@ internal class ClusterShell {
     }
 
     // TODO: reformulate as Wire.accept / reject?
-    internal enum HandshakeResult: Equatable, NotActuallyCodableMessage {
+    internal enum HandshakeResult: Equatable, _NotActuallyCodableMessage {
         case success(UniqueNode)
         case failure(HandshakeStateMachine.HandshakeConnectionError)
     }

Sources/DistributedActors/Cluster/DiscoveryShell.swift

@@ -16,7 +16,7 @@ import Logging
 import ServiceDiscovery
 
 final class DiscoveryShell {
-    enum Message: NotActuallyCodableMessage {
+    enum Message: _NotActuallyCodableMessage {
         case listing(Set<Node>)
         case stop(CompletionReason?)
     }

Sources/DistributedActors/Cluster/Downing/DowningStrategy.swift

@@ -21,22 +21,44 @@ import Logging
 public protocol DowningStrategy {
     /// Invoked whenever the cluster emits an event.
     ///
-    /// - Parameter event: cluster event that just ocurred
+    /// - Parameter event: cluster event that just occurred
     /// - Returns: directive, instructing the cluster to take some specific action.
     /// - Throws: If unable to handle the event for some reason; the failure will be logged and ignored.
     func onClusterEvent(event: Cluster.Event) throws -> DowningStrategyDirective
 
     func onTimeout(_ member: Cluster.Member) -> DowningStrategyDirective
 }
 
-public enum DowningStrategyDirective {
-    case none
-    case markAsDown(Set<Cluster.Member>)
-    case startTimer(key: TimerKey, member: Cluster.Member, delay: TimeAmount)
-    case cancelTimer(key: TimerKey)
+/// Return to instruct the downing shell how to react.
+public struct DowningStrategyDirective {
+    internal let underlying: Repr
+    internal enum Repr {
+        case none
+        case markAsDown(Set<Cluster.Member>)
+        case startTimer(key: TimerKey, member: Cluster.Member, delay: TimeAmount)
+        case cancelTimer(key: TimerKey)
+    }
+
+    internal init(_ underlying: Repr) {
+        self.underlying = underlying
+    }
+
+    public static var none: Self {
+        .init(.none)
+    }
 
-    static func markAsDown(_ member: Cluster.Member) -> Self {
-        Self.markAsDown([member])
+    public static func startTimer(key: TimerKey, member: Cluster.Member, delay: TimeAmount) -> Self {
+        .init(.startTimer(key: key, member: member, delay: delay))
+    }
+    public static func cancelTimer(key: TimerKey) -> Self {
+        .init(.cancelTimer(key: key))
+    }
+
+    public static func markAsDown(members: Set<Cluster.Member>) -> Self {
+        .init(.markAsDown(members))
+    }
+    public static func markAsDown(_ member: Cluster.Member) -> Self {
+        .init(.markAsDown([member]))
     }
 }
 
@@ -85,7 +107,7 @@ internal distributed actor DowningStrategyShell {
     }
 
     func interpret(directive: DowningStrategyDirective) {
-        switch directive {
+        switch directive.underlying {
         case .markAsDown(let members):
             self.markAsDown(members: members)
 

Sources/DistributedActors/Cluster/Downing/TimeoutBasedDowningStrategy.swift

@@ -60,7 +60,7 @@ public final class TimeoutBasedDowningStrategy: DowningStrategy {
             } else if let replaced = change.replaced {
                 _ = self._markAsDown.remove(replaced)
                 _ = self._unreachable.remove(replaced)
-                return .markAsDown([replaced])
+                return .markAsDown(members: [replaced])
             }
 
             return .none
@@ -82,7 +82,7 @@ public final class TimeoutBasedDowningStrategy: DowningStrategy {
             return .none // perhaps we removed it already for other reasons (e.g. node replacement)
         }
         if self.isLeader {
-            return .markAsDown([nodeToDown])
+            return .markAsDown(members: [nodeToDown])
         } else {
             self._markAsDown.insert(nodeToDown)
             return .none
@@ -119,7 +119,7 @@ public final class TimeoutBasedDowningStrategy: DowningStrategy {
 
         if self.isLeader, !self._markAsDown.isEmpty {
             defer { self._markAsDown = [] }
-            return .markAsDown(self._markAsDown)
+            return .markAsDown(members: self._markAsDown)
         } else {
             return .none
         }

Sources/DistributedActors/Cluster/NodeDeathWatcher.swift

@@ -171,7 +171,7 @@ enum NodeDeathWatcherShell {
     /// Message protocol for interacting with the failure detector.
     /// By default, the `FailureDetectorShell` handles these messages by interpreting them with an underlying `FailureDetector`,
     /// it would be possible however to allow implementing the raw protocol by user actors if we ever see the need for it.
-    internal enum Message: NotActuallyCodableMessage {
+    internal enum Message: _NotActuallyCodableMessage {
         case remoteActorWatched(watcher: _AddressableActorRef, remoteNode: UniqueNode)
         case remoteDistributedActorWatched(remoteNode: UniqueNode, watcherID: ClusterSystem.ActorID, nodeTerminated: @Sendable (UniqueNode) async -> Void)
         case removeWatcher(watcherID: ClusterSystem.ActorID)

Sources/DistributedActors/Cluster/Reception/OperationLogDistributedReceptionist.swift

@@ -85,7 +85,6 @@ import Logging
 /// number observations.
 ///
 /// #### Optimization: "Blip" Registration Replication Avoidance
-// TODO: This is done "automatically" once we do log compaction
 /// We define a "blip registration" as a registration of an actor, which immediately (or very quickly) after registering
 /// terminates. It can be argued it is NOT useful to replicate the very existence of such short lived actor to other peers,
 /// as even if they'd act on the `register`, it'd be immediately followed by `remove` and/or a termination signal.
@@ -139,15 +138,16 @@ import Logging
 ///   (Note that we simply always `ack(latest)` and if in the meantime the pusher got more updates, it'll push those to us as well.
 ///
 /// - SeeAlso: [Wikipedia: Atomic broadcast](https://en.wikipedia.org/wiki/Atomic_broadcast)
-// TODO: compact the log whenever we know all members of the cluster have seen
-// TODO: Optimization: gap collapsing: [+a,+b,+c,-c] -> [+a,+b,gap(until:4)]
-// TODO: Optimization: head collapsing: [+a,+b,+c,-b,-a] -> [gap(until:2),+c,-b]
-//
-// TODO: slow/fast ticks: When we know there's nothing new to share with others, we use the slow tick (which should be increased to 5 seconds or less)
-//       when we received a register() or observed an "ahead" receptionist, we should schedule a "fast tick" in order to more quickly spread this information
-//       This should still be done on a delay, e.g. if we are receiving many registrations, we want to get the benefit of batching them up before sending after all
-//       The fast tick could be 1s or 0.5s for example as a default.
 public distributed actor OpLogDistributedReceptionist: DistributedReceptionist, CustomStringConvertible {
+    // TODO: compact the log whenever we know all members of the cluster have seen
+    // TODO: Optimization: gap collapsing: [+a,+b,+c,-c] -> [+a,+b,gap(until:4)]
+    // TODO: Optimization: head collapsing: [+a,+b,+c,-b,-a] -> [gap(until:2),+c,-b]
+    //
+    // TODO: slow/fast ticks: When we know there's nothing new to share with others, we use the slow tick (which should be increased to 5 seconds or less)
+    //       when we received a register() or observed an "ahead" receptionist, we should schedule a "fast tick" in order to more quickly spread this information
+    //       This should still be done on a delay, e.g. if we are receiving many registrations, we want to get the benefit of batching them up before sending after all
+    //       The fast tick could be 1s or 0.5s for example as a default.
+
     public typealias ActorSystem = ClusterSystem
 
     // TODO: remove this
@@ -1028,7 +1028,7 @@ extension OpLogDistributedReceptionist {
         }
     }
 
-    final class PublishLocalListingsTrigger: Receptionist.Message, NotActuallyCodableMessage, CustomStringConvertible {
+    final class PublishLocalListingsTrigger: Receptionist.Message, _NotActuallyCodableMessage, CustomStringConvertible {
         override init() {
             super.init()
         }

Sources/DistributedActors/Cluster/Reception/_OperationLogClusterReceptionistBehavior.swift

@@ -774,7 +774,7 @@ extension _OperationLogClusterReceptionist {
         }
     }
 
-    class PeriodicAckTick: Receptionist.Message, NotActuallyCodableMessage, CustomStringConvertible {
+    class PeriodicAckTick: Receptionist.Message, _NotActuallyCodableMessage, CustomStringConvertible {
         override init() {
             super.init()
         }
@@ -788,7 +788,7 @@ extension _OperationLogClusterReceptionist {
         }
     }
 
-    class PublishLocalListingsTrigger: Receptionist.Message, NotActuallyCodableMessage, CustomStringConvertible {
+    class PublishLocalListingsTrigger: Receptionist.Message, _NotActuallyCodableMessage, CustomStringConvertible {
         override init() {
             super.init()
         }