=cluster,membership compressed gossip messages by node ID in seen tables

Author: ktoso

Protos/Clocks/VersionVector.proto

@@ -27,6 +27,7 @@ message VersionReplicaID {
     oneof value {
         ActorAddress actorAddress = 1;
         UniqueNode uniqueNode = 2;
+        uint32 uniqueNodeID = 3;
     }
 }
 

Protos/Cluster/Membership.proto

@@ -19,6 +19,7 @@ option optimize_for = SPEED;
 option swift_prefix = "Proto";
 
 import "ActorAddress.proto";
+import "Clocks/VersionVector.proto";
 
 message ClusterMembership {
     repeated ClusterMember members = 1;
@@ -47,8 +48,24 @@ enum ClusterMemberStatus {
     CLUSTER_MEMBER_STATUS_REMOVED = 5;
 }
 
+// ==== Membership Gossip ----------------------------------------------------------------------------------------------
+
 message ClusterMembershipGossip {
-    // origin of the gossip
-    UniqueNode from = 1;
-    repeated ClusterMember members = 2; // TODO: Something else, "membership diff"?
+    // Membership contains full UniqueNode renderings, and the owner and seen table refer to them by UniqueNode.ID
+    // this saves us space (by avoiding to render the unique node explicitly many times for each member/seen-entry).
+    ClusterMembership membership = 1;
+
+    // The following fields will use compressed UniqueNode encoding and ONLY serialize them as their uniqueNodeID.
+    // During deserialization the fields can be resolved against the membership to obtain full UniqueNode values if necessary.
+    uint32 ownerUniqueNodeID = 2;
+    ClusterMembershipSeenTable seenTable = 3;
 }
+
+message ClusterMembershipSeenTable {
+    repeated ClusterMembershipSeenTableRow rows = 1;
+}
+
+message ClusterMembershipSeenTableRow {
+    uint32 uniqueNodeID = 1;
+    VersionVector version = 2;
+}

Sources/DistributedActors/ActorAddress.swift

@@ -612,20 +612,22 @@ extension Node: Comparable {
 /// Once the remote node accepts our handshake, it offers the other node its unique address.
 /// Only once this address has been obtained can a node communicate with actors located on the remote node.
 public struct UniqueNode: Hashable {
+    public typealias ID = UniqueNodeID
+
     public var node: Node
-    public let nid: NodeID
+    public let nid: UniqueNodeID
 
-    public init(node: Node, nid: NodeID) {
+    public init(node: Node, nid: UniqueNodeID) {
         precondition(node.port > 0, "port MUST be > 0")
         self.node = node
         self.nid = nid
     }
 
-    public init(protocol: String, systemName: String, host: String, port: Int, nid: NodeID) {
+    public init(protocol: String, systemName: String, host: String, port: Int, nid: UniqueNodeID) {
         self.init(node: Node(protocol: `protocol`, systemName: systemName, host: host, port: port), nid: nid)
     }
 
-    public init(systemName: String, host: String, port: Int, nid: NodeID) {
+    public init(systemName: String, host: String, port: Int, nid: UniqueNodeID) {
         self.init(protocol: "sact", systemName: systemName, host: host, port: port, nid: nid)
     }
 
@@ -676,29 +678,29 @@ extension UniqueNode: Comparable {
     }
 }
 
-public struct NodeID: Hashable {
+public struct UniqueNodeID: Hashable {
     let value: UInt32 // TODO: redesign / reconsider exact size
 
     public init(_ value: UInt32) {
         self.value = value
     }
 }
 
-extension NodeID: Comparable {
-    public static func < (lhs: NodeID, rhs: NodeID) -> Bool {
+extension UniqueNodeID: Comparable {
+    public static func < (lhs: UniqueNodeID, rhs: UniqueNodeID) -> Bool {
         lhs.value < rhs.value
     }
 }
 
-extension NodeID: CustomStringConvertible {
+extension UniqueNodeID: CustomStringConvertible {
     public var description: String {
         "\(self.value)"
     }
 }
 
-public extension NodeID {
-    static func random() -> NodeID {
-        NodeID(UInt32.random(in: 1 ... .max))
+public extension UniqueNodeID {
+    static func random() -> UniqueNodeID {
+        UniqueNodeID(UInt32.random(in: 1 ... .max))
     }
 }
 

Sources/DistributedActors/Clocks/Protobuf/VersionVector+Serialization.swift

@@ -25,6 +25,8 @@ extension ReplicaID: ProtobufRepresentable {
             proto.actorAddress = try actorAddress.toProto(context: context)
         case .uniqueNode(let node):
             proto.uniqueNode = try node.toProto(context: context)
+        case .uniqueNodeID(let nid):
+            proto.uniqueNodeID = nid.value
         }
         return proto
     }
@@ -41,6 +43,8 @@ extension ReplicaID: ProtobufRepresentable {
         case .uniqueNode(let protoNode):
             let node = try UniqueNode(fromProto: protoNode, context: context)
             self = .uniqueNode(node)
+        case .uniqueNodeID(let nid):
+            self = .uniqueNodeID(nid)
         }
     }
 }
@@ -65,6 +69,29 @@ extension VersionVector: ProtobufRepresentable {
         return proto
     }
 
+    /// Serialize using uniqueNodeID specifically (or crash);
+    /// Used in situations where an enclosing message already has the unique nodes serialized and we can save space by avoiding to serialize them again.
+    public func toCompactReplicaNodeIDProto(context: Serialization.Context) throws -> ProtoVersionVector {
+        var proto = ProtoVersionVector()
+
+        let replicaVersions: [ProtoReplicaVersion] = try self.state.map { replicaID, version in
+            var replicaVersion = ProtoReplicaVersion()
+            switch replicaID.storage {
+            case .uniqueNode(let node):
+                replicaVersion.replicaID.uniqueNodeID = node.nid.value
+            case .uniqueNodeID(let nid):
+                replicaVersion.replicaID.uniqueNodeID = nid.value
+            case .actorAddress:
+                throw SerializationError.unableToSerialize(hint: "Can't serialize using actor address as replica id! Was: \(replicaID)")
+            }
+            replicaVersion.version = UInt64(version)
+            return replicaVersion
+        }
+        proto.state = replicaVersions
+
+        return proto
+    }
+
     public init(fromProto proto: ProtoVersionVector, context: Serialization.Context) throws {
         // `state` defaults to [:]
         self.state.reserveCapacity(proto.state.count)

Sources/DistributedActors/Clocks/Protobuf/VersionVector.pb.swift

@@ -59,17 +59,27 @@ public struct ProtoVersionReplicaID {
     set {_uniqueStorage()._value = .uniqueNode(newValue)}
   }
 
+  public var uniqueNodeID: UInt32 {
+    get {
+      if case .uniqueNodeID(let v)? = _storage._value {return v}
+      return 0
+    }
+    set {_uniqueStorage()._value = .uniqueNodeID(newValue)}
+  }
+
   public var unknownFields = SwiftProtobuf.UnknownStorage()
 
   public enum OneOf_Value: Equatable {
     case actorAddress(ProtoActorAddress)
     case uniqueNode(ProtoUniqueNode)
+    case uniqueNodeID(UInt32)
 
   #if !swift(>=4.1)
     public static func ==(lhs: ProtoVersionReplicaID.OneOf_Value, rhs: ProtoVersionReplicaID.OneOf_Value) -> Bool {
       switch (lhs, rhs) {
       case (.actorAddress(let l), .actorAddress(let r)): return l == r
       case (.uniqueNode(let l), .uniqueNode(let r)): return l == r
+      case (.uniqueNodeID(let l), .uniqueNodeID(let r)): return l == r
       default: return false
       }
     }
@@ -190,6 +200,7 @@ extension ProtoVersionReplicaID: SwiftProtobuf.Message, SwiftProtobuf._MessageIm
   public static let _protobuf_nameMap: SwiftProtobuf._NameMap = [
     1: .same(proto: "actorAddress"),
     2: .same(proto: "uniqueNode"),
+    3: .same(proto: "uniqueNodeID"),
   ]
 
   fileprivate class _StorageClass {
@@ -232,6 +243,11 @@ extension ProtoVersionReplicaID: SwiftProtobuf.Message, SwiftProtobuf._MessageIm
           }
           try decoder.decodeSingularMessageField(value: &v)
           if let v = v {_storage._value = .uniqueNode(v)}
+        case 3:
+          if _storage._value != nil {try decoder.handleConflictingOneOf()}
+          var v: UInt32?
+          try decoder.decodeSingularUInt32Field(value: &v)
+          if let v = v {_storage._value = .uniqueNodeID(v)}
         default: break
         }
       }
@@ -245,6 +261,8 @@ extension ProtoVersionReplicaID: SwiftProtobuf.Message, SwiftProtobuf._MessageIm
         try visitor.visitSingularMessageField(value: v, fieldNumber: 1)
       case .uniqueNode(let v)?:
         try visitor.visitSingularMessageField(value: v, fieldNumber: 2)
+      case .uniqueNodeID(let v)?:
+        try visitor.visitSingularUInt32Field(value: v, fieldNumber: 3)
       case nil: break
       }
     }

Sources/DistributedActors/Clocks/VersionVector.swift

@@ -241,6 +241,7 @@ public struct ReplicaID: Hashable {
     internal enum Storage: Hashable {
         case actorAddress(ActorAddress)
         case uniqueNode(UniqueNode)
+        case uniqueNodeID(UniqueNode.ID)
     }
 
     internal let storage: Storage
@@ -262,12 +263,24 @@ public struct ReplicaID: Hashable {
         .init(.uniqueNode(uniqueNode))
     }
 
+    public static func uniqueNodeID(_ uniqueNode: UniqueNode) -> ReplicaID {
+        .init(.uniqueNodeID(uniqueNode.nid))
+    }
+
+    internal static func uniqueNodeID(_ uniqueNodeID: UInt32) -> ReplicaID {
+        .init(.uniqueNodeID(.init(uniqueNodeID)))
+    }
+
     func ensuringNode(_ node: UniqueNode) -> ReplicaID {
         switch self.storage {
-        case .uniqueNode:
-            return self
         case .actorAddress(let address):
             return .actorAddress(address.ensuringNode(node))
+        case .uniqueNode(let existingNode):
+            assert(existingNode.nid == node.nid, "Attempted to ensureNode with non-matching node identifier, was: \(existingNode)], attempted: \(node)")
+            return self
+        case .uniqueNodeID(let nid): // drops the nid
+            assert(nid == node.nid, "Attempted to ensureNode with non-matching node identifier, was: \(nid)], attempted: \(node)")
+            return .uniqueNode(node)
         }
     }
 }
@@ -279,6 +292,8 @@ extension ReplicaID: CustomStringConvertible {
             return "actor:\(address)"
         case .uniqueNode(let node):
             return "uniqueNode:\(node)"
+        case .uniqueNodeID(let nid):
+            return "uniqueNodeID:\(nid)"
         }
     }
 }
@@ -290,43 +305,56 @@ extension ReplicaID: Comparable {
             return l < r
         case (.uniqueNode(let l), .uniqueNode(let r)):
             return l < r
-        case (.uniqueNode, _), (.actorAddress, _):
-            return false // TODO: should we even disallow comparing them?
+        case (.uniqueNodeID(let l), .uniqueNodeID(let r)):
+            return l < r
+        case (.uniqueNode, _), (.uniqueNodeID, _), (.actorAddress, _):
+            return false
         }
     }
 
     public static func == (lhs: ReplicaID, rhs: ReplicaID) -> Bool {
         switch (lhs.storage, rhs.storage) {
         case (.actorAddress(let l), .actorAddress(let r)):
             return l == r
+
         case (.uniqueNode(let l), .uniqueNode(let r)):
             return l == r
-        case (.uniqueNode, _), (.actorAddress, _):
-            return false // TODO: should we even disallow comparing them?
+
+        case (.uniqueNodeID(let l), .uniqueNodeID(let r)):
+            return l == r
+        case (.uniqueNode(let l), .uniqueNodeID(let r)):
+            return l.nid == r
+        case (.uniqueNodeID(let l), .uniqueNode(let r)):
+            return l == r.nid
+
+        case (.uniqueNode, _), (.uniqueNodeID, _), (.actorAddress, _):
+            return false
         }
     }
 }
 
 extension ReplicaID: Codable {
     public enum DiscriminatorKeys: String, Codable {
-        case actorAddress
-        case uniqueNode
+        case actorAddress = "a"
+        case uniqueNode = "N"
+        case uniqueNodeID = "n"
     }
 
     public enum CodingKeys: CodingKey {
         case _case
 
-        case actorAddress_value
-        case uniqueNode_value
+        case value
     }
 
     public init(from decoder: Decoder) throws {
         let container = try decoder.container(keyedBy: CodingKeys.self)
         switch try container.decode(DiscriminatorKeys.self, forKey: ._case) {
         case .actorAddress:
-            self = try .actorAddress(container.decode(ActorAddress.self, forKey: .actorAddress_value))
+            self = try .actorAddress(container.decode(ActorAddress.self, forKey: .value))
         case .uniqueNode:
-            self = try .uniqueNode(container.decode(UniqueNode.self, forKey: .uniqueNode_value))
+            self = try .uniqueNode(container.decode(UniqueNode.self, forKey: .value))
+        case .uniqueNodeID:
+            self = try .uniqueNodeID(container.decode(UInt32.self, forKey: .value))
         }
     }
 
@@ -335,10 +363,13 @@ extension ReplicaID: Codable {
         switch self.storage {
         case .actorAddress(let address):
             try container.encode(DiscriminatorKeys.actorAddress, forKey: ._case)
-            try container.encode(address, forKey: .actorAddress_value)
+            try container.encode(address, forKey: .value)
         case .uniqueNode(let node):
             try container.encode(DiscriminatorKeys.uniqueNode, forKey: ._case)
-            try container.encode(node, forKey: .uniqueNode_value)
+            try container.encode(node, forKey: .value)
+        case .uniqueNodeID(let nid):
+            try container.encode(DiscriminatorKeys.uniqueNodeID, forKey: ._case)
+            try container.encode(nid.value, forKey: .value)
         }
     }
 }